"""Find a channel from what a person types: @username, t.me link, numeric id (-100...), or an invite link. Used by /shift. The bot can only talk to a channel it can see: registered channels (the database has their access details), channels it has met in updates, public channels by @username, and channels it is a member of (invite link). """ from __future__ import annotations import re from dataclasses import dataclass from typing import Optional from telethon import functions, types from .common import Ctx, UserError _REF_RX = re.compile(r"(?i)^(?:https?://)?(?:t\.me|telegram\.me)/(?:c/(\d+)|([A-Za-z0-9_]{3,32}))(?:/\d+)?/?$") _INVITE_RX = re.compile(r"(?i)^(?:https?://)?(?:t\.me|telegram\.me|telegram\.dog)/(?:\+|joinchat/)([A-Za-z0-9_-]{6,})/?$") _TG_INVITE_RX = re.compile(r"(?i)^tg://join\?invite=([A-Za-z0-9_-]{6,})$") def parse_channel_ref(text: str): """@name / t.me link / numeric id -> username (str) or channel id (int); None if unrecognised.""" t = (text or "").strip() m = _REF_RX.match(t) if m: return int(m.group(1)) if m.group(1) else m.group(2) t = t.lstrip("@") if re.fullmatch(r"-?\d+", t): s = str(abs(int(t))) return int(s[3:]) if s.startswith("100") and len(s) > 10 else int(s) if re.fullmatch(r"[A-Za-z0-9_]{3,32}", t): return t return None def classify_ref(text: str) -> tuple: """-> ("invite", hash) | ("username", name) | ("id", int) | (None, None)""" t = (text or "").strip() m = _INVITE_RX.match(t) or _TG_INVITE_RX.match(t) if m: return "invite", m.group(1) ref = parse_channel_ref(t) if isinstance(ref, int): return "id", ref if isinstance(ref, str): return "username", ref return None, None @dataclass class ChannelRef: """A channel as the bot can reach it.""" id: int access_hash: int title: str username: Optional[str] registered: bool # the bot's database knows it (it can be used with /new, /posts ...) def _from_entity(ent, registered: bool) -> ChannelRef: if not isinstance(ent, types.Channel): raise UserError("That is not a channel (it looks like a user, bot or group). Only channels can be used here.") if not ent.broadcast: raise UserError("That is a group, not a channel. Only channels can be used here.") if not ent.access_hash or getattr(ent, "min", False): raise UserError( "I can see that channel, but Telegram has not given me its access details. Use its @username, or add the " "bot to the channel as an admin and post something there first." ) return ChannelRef(ent.id, ent.access_hash, ent.title or "", ent.username, registered) async def _entity(client, what): try: return await client.get_entity(what) except Exception: return None async def resolve_channel(ctx: Ctx, text: str) -> ChannelRef: """The channel `text` points at, or a UserError that says what to do.""" kind, value = classify_ref(text) if kind is None: raise UserError( f"“{text}” is not a channel I understand. Use an @username, a t.me link, the channel id (-100…) or an invite link." ) db, client = ctx.db, ctx.client if kind == "username": for ch in await db.list_channels(active_only=False): if ch.username and ch.username.lower() == value.lower(): return ChannelRef(ch.id, ch.access_hash, ch.title, ch.username, ch.active) ent = await _entity(client, value) if ent is None: raise UserError(f"I couldn't find a channel called @{value}. Check the name, or add the bot to it as an admin.") elif kind == "id": ch = await db.get_channel(value) if ch is not None: return ChannelRef(ch.id, ch.access_hash, ch.title, ch.username, ch.active) ent = ctx.seen_channels.get(value) or await _entity(client, types.PeerChannel(value)) if ent is None: raise UserError( f"I don't know a channel with the id {value}. A bot can only use an id for channels it is part of; " "use the @username instead, or add the bot to the channel as an admin." ) else: # invite link: only tells which channel it is; the bot cannot join by itself try: info = await client(functions.messages.CheckChatInviteRequest(hash=value)) except Exception as e: raise UserError(f"That invite link doesn't work ({type(e).__name__}).") chat = getattr(info, "chat", None) if chat is None: raise UserError( "The bot is not a member of that channel, so it cannot use that invite link. Add the bot to the channel " "as an admin first (a bot cannot join by itself)." ) ent = chat known = await db.get_channel(getattr(ent, "id", 0)) return _from_entity(ent, known is not None and known.active)