87 lines
2.7 KiB
Python
Executable File
87 lines
2.7 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
# /// script
|
|
# requires-python = ">=3.10"
|
|
# dependencies = [
|
|
# "telethon>=1.42",
|
|
# "python-dotenv>=1.0",
|
|
# ]
|
|
# ///
|
|
"""
|
|
List broadcast channels and supergroups in a named Telegram folder (default: "Jobs").
|
|
|
|
Output: JSON array of usernames (falling back to numeric id for private channels) on stdout.
|
|
|
|
Pipe directly into the fetch script:
|
|
list_telegram_channels.py | fetch_telegram_jobs.py -
|
|
|
|
Account: uses TELEGRAM_SESSION_STRING from .env — must be the usulsu (main) account.
|
|
The "Jobs" folder lives on that account.
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from dotenv import load_dotenv
|
|
from telethon import TelegramClient
|
|
from telethon.sessions import StringSession
|
|
from telethon.tl.functions.messages import GetDialogFiltersRequest
|
|
from telethon.tl.types import Channel, InputPeerChannel
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
|
DEFAULT_FOLDER = "Jobs"
|
|
|
|
|
|
def filter_title(f):
|
|
"""DialogFilter.title is str on older Telethon, TextWithEntities on newer."""
|
|
t = getattr(f, "title", None)
|
|
if t is None:
|
|
return ""
|
|
return t if isinstance(t, str) else getattr(t, "text", "")
|
|
|
|
|
|
async def main(folder_title):
|
|
load_dotenv(PROJECT_ROOT / ".env")
|
|
try:
|
|
api_id = int(os.environ["TELEGRAM_API_ID"])
|
|
api_hash = os.environ["TELEGRAM_API_HASH"]
|
|
session = os.environ["TELEGRAM_SESSION_STRING"]
|
|
except KeyError as e:
|
|
sys.exit(f"missing env var: {e}. check .env in project root.")
|
|
|
|
async with TelegramClient(StringSession(session), api_id, api_hash) as client:
|
|
result = await client(GetDialogFiltersRequest())
|
|
|
|
target = None
|
|
for f in result.filters:
|
|
if filter_title(f) == folder_title:
|
|
target = f
|
|
break
|
|
if target is None:
|
|
sys.exit(f"folder {folder_title!r} not found")
|
|
|
|
# Combine pinned + included peers — both are part of the folder
|
|
wanted_channel_ids = set()
|
|
for peer in list(getattr(target, "pinned_peers", []) or []) + list(target.include_peers):
|
|
if isinstance(peer, InputPeerChannel):
|
|
wanted_channel_ids.add(peer.channel_id)
|
|
|
|
# Resolve channel entities to extract usernames
|
|
usernames = []
|
|
async for dialog in client.iter_dialogs():
|
|
ent = dialog.entity
|
|
if not isinstance(ent, Channel):
|
|
continue
|
|
if ent.id not in wanted_channel_ids:
|
|
continue
|
|
usernames.append(ent.username or str(-1000000000000 - ent.id))
|
|
|
|
print(json.dumps(usernames, ensure_ascii=False))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
folder = sys.argv[1] if len(sys.argv) > 1 else DEFAULT_FOLDER
|
|
asyncio.run(main(folder))
|