Spaces:
Paused
Paused
| """A userbot - a normal Telegram account - that does what the bot is not allowed to do in a channel. | |
| Deleting: Telegram's documentation says bots can only delete messages younger than 48 hours. The bot always tries | |
| first; whatever is left goes to the userbot. | |
| Editing: Telegram refuses the bot an edit of a post that belongs to another bot (its buttons were added by that bot, | |
| or it was posted through it). The bot always tries first; when Telegram refuses, the userbot makes the edit (text, | |
| caption or a new photo / file). A user account can't set inline buttons, so Telegram may drop them with that edit - | |
| the bot puts them back where it can. | |
| A new photo / file goes to the userbot through the userbot's own chat with the bot: the bot sends it there (Telegram | |
| just points to the file it already has - nothing is uploaded), the userbot finds it in that chat and edits the post | |
| with that copy, again by reference. So a big file is as quick as a small one and keeps its thumbnail. Only when that | |
| chat does not work (the userbot account never started the bot) is the file downloaded and uploaded again. | |
| How it gets into the channel: | |
| 1. the bot makes a one-time invite link (or reads the channel's own link) and the userbot joins with it | |
| (public channels without a usable link: the userbot joins by name); | |
| 2. if the userbot is not allowed to do the job there yet and the bot has the "Add new admins" right, the bot | |
| makes the userbot an admin with that one right ("Delete messages" / "Edit messages of others"); | |
| 3. the userbot does the job, and afterwards the bot gives the admin rights back as they were | |
| (USERBOT_KEEP_ADMIN=true keeps the new right). | |
| The session lives in USERBOT_SESSION (made on your own computer with make_userbot_session.py). | |
| """ | |
| from __future__ import annotations | |
| import asyncio | |
| import datetime | |
| import logging | |
| import re | |
| from typing import Optional | |
| from telethon import TelegramClient, errors, functions, types, utils | |
| from telethon.sessions import StringSession | |
| from .richmsg import with_rich | |
| from .tgutil import MediaFile, edit_raw, flood_retry, get_rights, media_from_ref, media_key, peer_of, with_media | |
| log = logging.getLogger(__name__) | |
| _INVITE_RX = re.compile( | |
| r"(?i)(?:https?://)?(?:t\.me|telegram\.me|telegram\.dog)/(?:\+|joinchat/)([A-Za-z0-9_-]{8,})" | |
| r"|tg://join\?invite=([A-Za-z0-9_-]{8,})" | |
| ) | |
| class UserbotError(Exception): | |
| """Something the owner can fix; the text is meant to be shown.""" | |
| class UserbotRefused(UserbotError): | |
| """Telegram refused what the userbot tried; `name` is the Telegram error (e.g. MediaInvalidError).""" | |
| def __init__(self, text: str, name: str = ""): | |
| super().__init__(text) | |
| self.name = name | |
| def about_media(self) -> bool: | |
| """The refusal is about the photo / file the userbot used (so trying it another way may help).""" | |
| return self.name in _MEDIA_REFUSED | |
| # What Telegram answers when the photo / file the userbot points to can't be used for an edit | |
| _MEDIA_REFUSED = { | |
| "MediaInvalidError", "MediaEmptyError", "DocumentInvalidError", "PhotoInvalidError", | |
| "FileReferenceExpiredError", "FileReferenceInvalidError", "FileReferenceEmptyError", | |
| } | |
| # What Telegram answers when the bot may not write to the userbot account (it never started the bot, or blocked it) | |
| _NOT_STARTED = { | |
| "UserIsBlockedError", "PeerIdInvalidError", "InputUserDeactivatedError", "UserDeactivatedError", "UserIdInvalidError", | |
| } | |
| _DM_POLLS = 10 # how often, and how long apart, the userbot looks in its chat with the bot for the file it was sent | |
| _DM_WAIT = 0.5 | |
| def parse_invite_hash(link: str) -> Optional[str]: | |
| m = _INVITE_RX.search(link or "") | |
| return (m.group(1) or m.group(2)) if m else None | |
| def _default_factory(cfg) -> TelegramClient: | |
| return TelegramClient(StringSession(cfg.userbot_session), cfg.api_id, cfg.api_hash, receive_updates=False) | |
| class Userbot: | |
| def __init__(self, bot_client, cfg, client_factory=None): | |
| self.bot = bot_client | |
| self.cfg = cfg | |
| self._factory = client_factory or _default_factory | |
| self.client = None | |
| self.me = None | |
| self.last_error: Optional[str] = None | |
| self.lock = asyncio.Lock() # one channel job at a time (join / promote / delete or edit / demote) | |
| def enabled(self) -> bool: | |
| return bool(getattr(self.cfg, "userbot_session", "")) | |
| def ready(self) -> bool: | |
| return self.client is not None | |
| def keep_admin(self) -> bool: | |
| return bool(getattr(self.cfg, "userbot_keep_admin", False)) | |
| async def connect(self) -> bool: | |
| """Log the userbot in. True when it is usable; otherwise the reason is in `last_error`.""" | |
| if not self.enabled: | |
| self.last_error = "USERBOT_SESSION is not set." | |
| return False | |
| if self.client is not None: | |
| return True | |
| client = None | |
| try: | |
| client = self._factory(self.cfg) | |
| await client.connect() | |
| if not await client.is_user_authorized(): | |
| raise UserbotError( | |
| "The userbot session is not logged in any more (or it was revoked). " | |
| "Make a new one with make_userbot_session.py." | |
| ) | |
| me = await client.get_me() | |
| if getattr(me, "bot", False): | |
| raise UserbotError("USERBOT_SESSION belongs to a bot. The userbot has to be a normal user account.") | |
| except Exception as e: | |
| self.last_error = str(e) if isinstance(e, UserbotError) else f"{type(e).__name__}: {str(e)[:200]}" | |
| log.warning("the userbot is not available: %s", self.last_error) | |
| if client is not None: | |
| try: | |
| await client.disconnect() | |
| except Exception: | |
| pass | |
| return False | |
| self.client, self.me, self.last_error = client, me, None | |
| log.info("userbot logged in as %s (id %s)", getattr(me, "username", None) or getattr(me, "first_name", ""), me.id) | |
| return True | |
| async def close(self) -> None: | |
| client, self.client = self.client, None | |
| if client is not None: | |
| try: | |
| await client.disconnect() | |
| except Exception: | |
| pass | |
| def describe(self) -> str: | |
| if not self.enabled: | |
| return "not set up (USERBOT_SESSION is empty)" | |
| if self.ready: | |
| name = getattr(self.me, "username", None) | |
| who = f"@{name}" if name else (getattr(self.me, "first_name", "") or "the account") | |
| return f"connected as {who} (id {self.me.id})" | |
| return f"set up, but not connected: {self.last_error or 'unknown reason'}" | |
| def deleter(self, ch) -> "UserbotDeleter": | |
| return UserbotDeleter(self, ch) | |
| def editor(self, ch) -> "UserbotEditor": | |
| return UserbotEditor(self, ch) | |
| async def fallback_for(self, ch): | |
| """(delete function, deleter) to hand to the delete step, or (None, None) when no userbot is usable. | |
| Nothing happens in the channel until the bot has failed to delete something and the function is called; | |
| the caller must always `await deleter.close()` afterwards. | |
| """ | |
| if not self.enabled: | |
| return None, None | |
| if not self.ready: | |
| await self.connect() | |
| if not self.ready: | |
| return None, None | |
| d = self.deleter(ch) | |
| return d.delete, d | |
| async def editor_for(self, ch) -> Optional["UserbotEditor"]: | |
| """An editor for posts in `ch`, or None when no userbot is usable. The caller must always `await editor.close()`.""" | |
| if not self.enabled: | |
| return None | |
| if not self.ready: | |
| await self.connect() | |
| return self.editor(ch) if self.ready else None | |
| def _rights_dict(rights) -> dict: | |
| return {k: v for k, v in rights.to_dict().items() if k != "_"} if rights is not None else {} | |
| class _ChannelJob: | |
| """Everything needed to do ONE job in ONE channel with the userbot. Call close() when done, whatever happened. | |
| What the job needs is named by RIGHT (the field of ChatAdminRights) and RIGHT_NAME (what the Telegram apps call it). | |
| """ | |
| RIGHT = "delete_messages" | |
| RIGHT_NAME = "Delete messages" | |
| WHAT = "delete messages" # what the userbot may not do without the right (as the owner is told) | |
| VERB = "delete" | |
| def __init__(self, ub: Userbot, ch): | |
| self.ub = ub | |
| self.ch = ch | |
| self.peer = None # the channel as the userbot sees it | |
| self.notes: list = [] | |
| self._ready = False | |
| self._holds_lock = False | |
| self._promoted = False | |
| self._before = None # (admin rights, rank) the userbot had before the bot changed them; None = a plain member | |
| self._bot_user = None # the userbot account as the bot sees it (InputUser) | |
| self._link: Optional[str] = None # the temporary invite link we made | |
| # ------------------------------------------------------------------ public | |
| async def _begin(self) -> None: | |
| """Get the channel ready (joined, with the right). The userbot's lock is kept until close().""" | |
| if not self.ub.ready: | |
| raise UserbotError(f"The userbot is not connected: {self.ub.last_error or 'unknown reason'}") | |
| if not self._ready: | |
| if not self._holds_lock: # released again in close() | |
| await self.ub.lock.acquire() | |
| self._holds_lock = True | |
| await self._prepare() | |
| self._ready = True | |
| async def close(self) -> None: | |
| """Undo what this job changed: give the admin rights back as they were, revoke the temporary link.""" | |
| bot, ch = self.ub.bot, self.ch | |
| try: | |
| if self._promoted and self._bot_user is not None and not self.ub.keep_admin: | |
| rights, rank = self._before or (types.ChatAdminRights(), "") | |
| try: | |
| await bot( | |
| functions.channels.EditAdminRequest( | |
| channel=peer_of(ch), user_id=self._bot_user, admin_rights=rights, rank=rank | |
| ) | |
| ) | |
| self.notes.append( | |
| "took the admin right away from the userbot again" | |
| if self._before is None | |
| else "put the userbot's admin rights back as they were" | |
| ) | |
| except Exception as e: | |
| log.warning("could not demote the userbot in %s: %s", ch.id, type(e).__name__) | |
| self.notes.append("could not take the admin right away from the userbot - please do it by hand") | |
| if self._link: | |
| try: | |
| await bot( | |
| functions.messages.EditExportedChatInviteRequest(peer=peer_of(ch), link=self._link, revoked=True) | |
| ) | |
| except Exception as e: | |
| log.info("could not revoke the temporary invite link: %s", type(e).__name__) | |
| finally: | |
| self._promoted, self._link = False, None | |
| if self._holds_lock: | |
| self._holds_lock = False | |
| self.ub.lock.release() | |
| # ----------------------------------------------------------------- prepare | |
| async def _prepare(self) -> None: | |
| peer = await self._member_peer() | |
| if peer is None: | |
| peer = await self._join() | |
| self.peer = peer | |
| await self._ensure_right() | |
| async def _member_peer(self): | |
| """The channel as the userbot sees it, if the userbot is in it already.""" | |
| ub = self.ub.client | |
| try: | |
| peer = await ub.get_input_entity(types.PeerChannel(self.ch.id)) | |
| if not isinstance(peer, types.InputPeerChannel): | |
| return None | |
| await ub(functions.channels.GetParticipantRequest(peer, types.InputPeerSelf())) | |
| except (errors.RPCError, ValueError, TypeError): | |
| return None | |
| return peer | |
| async def _invite_link(self) -> Optional[str]: | |
| bot, peer = self.ub.bot, peer_of(self.ch) | |
| try: | |
| expires = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(hours=1) | |
| res = await bot( | |
| functions.messages.ExportChatInviteRequest(peer=peer, expire_date=expires, usage_limit=1, title="Cleanup helper") | |
| ) | |
| link = getattr(res, "link", None) | |
| if link: | |
| self._link = link | |
| self.notes.append("made a one-time invite link for the userbot") | |
| return link | |
| except errors.RPCError as e: | |
| log.info("could not create an invite link in %s: %s", self.ch.id, type(e).__name__) | |
| try: | |
| full = await bot(functions.channels.GetFullChannelRequest(peer)) | |
| link = getattr(getattr(full.full_chat, "exported_invite", None), "link", None) | |
| if link: | |
| self.notes.append("used the channel's own invite link") | |
| return link | |
| except errors.RPCError as e: | |
| log.info("could not read the invite link of %s: %s", self.ch.id, type(e).__name__) | |
| return None | |
| async def _join(self): | |
| link = await self._invite_link() | |
| h = parse_invite_hash(link) if link else None | |
| if h: | |
| peer = await self._import_invite(h) | |
| if peer is not None: | |
| self.notes.append("the userbot joined the channel") | |
| return peer | |
| if self.ch.username: | |
| peer = await self._join_public(self.ch.username) | |
| self.notes.append("the userbot joined the channel by its public name") | |
| return peer | |
| raise UserbotError( | |
| "I could not get the userbot into this private channel: Telegram gave me no invite link. " | |
| "Give the bot the “Invite users via link” admin right, or add the userbot to the channel yourself " | |
| f"(with the “{self.RIGHT_NAME}” admin right)." | |
| ) | |
| async def _import_invite(self, h: str): | |
| ub = self.ub.client | |
| try: | |
| res = await flood_retry(lambda: ub(functions.messages.ImportChatInviteRequest(h))) | |
| except errors.UserAlreadyParticipantError: | |
| return await self._peer_when_already_in(h) | |
| except errors.InviteRequestSentError: | |
| raise UserbotError( | |
| "This channel's invite link needs an admin to approve new members. Approve the userbot's join request " | |
| "(or add the userbot yourself) and try again." | |
| ) | |
| except (errors.InviteHashExpiredError, errors.InviteHashInvalidError, errors.InviteHashEmptyError): | |
| return None | |
| except (errors.ChannelsTooMuchError, errors.UserChannelsTooMuchError): | |
| raise UserbotError("The userbot account is in too many channels. Leave a few with that account and try again.") | |
| except errors.UserBannedInChannelError: | |
| raise UserbotError("The userbot account is banned in this channel. Unban it first.") | |
| return self._peer_from_updates(res) | |
| def _peer_from_updates(self, res): | |
| chats = [c for c in (getattr(res, "chats", None) or []) if isinstance(c, types.Channel)] | |
| for c in chats: | |
| if c.id == self.ch.id: | |
| return utils.get_input_peer(c) | |
| if len(chats) == 1: | |
| return utils.get_input_peer(chats[0]) | |
| raise UserbotError("The userbot joined, but Telegram did not tell me which channel it is. Try again.") | |
| async def _peer_when_already_in(self, h: str): | |
| ub = self.ub.client | |
| try: | |
| info = await ub(functions.messages.CheckChatInviteRequest(h)) | |
| if isinstance(info, types.ChatInviteAlready): | |
| return utils.get_input_peer(info.chat) | |
| except errors.RPCError: | |
| pass | |
| async for d in ub.iter_dialogs(): | |
| if getattr(d.entity, "id", None) == self.ch.id and isinstance(d.entity, types.Channel): | |
| return utils.get_input_peer(d.entity) | |
| raise UserbotError("The userbot is in the channel, but I could not find it in its chat list.") | |
| async def _join_public(self, username: str): | |
| ub = self.ub.client | |
| try: | |
| peer = await ub.get_input_entity(username) | |
| try: | |
| await flood_retry(lambda: ub(functions.channels.JoinChannelRequest(peer))) | |
| except errors.UserAlreadyParticipantError: | |
| pass | |
| except (errors.ChannelsTooMuchError, errors.UserChannelsTooMuchError): | |
| raise UserbotError("The userbot account is in too many channels. Leave a few with that account and try again.") | |
| except errors.UserBannedInChannelError: | |
| raise UserbotError("The userbot account is banned in this channel. Unban it first.") | |
| except (errors.UsernameNotOccupiedError, errors.UsernameInvalidError, errors.ChannelPrivateError): | |
| raise UserbotError(f"The userbot could not open @{username}. Has the channel's name changed?") | |
| return peer | |
| # ------------------------------------------------------------------ rights | |
| async def _userbot_status(self): | |
| """(is the creator, its admin rights or None, its rank) as the channel shows the userbot; None if unknown.""" | |
| try: | |
| res = await self.ub.client(functions.channels.GetParticipantRequest(self.peer, types.InputPeerSelf())) | |
| except errors.RPCError: | |
| return None | |
| p = res.participant | |
| if isinstance(p, types.ChannelParticipantCreator): | |
| return True, None, "" | |
| if isinstance(p, types.ChannelParticipantAdmin): | |
| return False, p.admin_rights, getattr(p, "rank", None) or "" | |
| return False, None, "" | |
| async def _userbot_has_right(self) -> bool: | |
| status = await self._userbot_status() | |
| if status is None: | |
| return False | |
| creator, rights, _ = status | |
| return creator or bool(getattr(rights, self.RIGHT, False)) | |
| async def _ensure_right(self) -> None: | |
| if await self._userbot_has_right(): | |
| return | |
| rights = await get_rights(self.ub.bot, self.ch) | |
| if rights is None or not (rights.admin and rights.add_admins): | |
| raise UserbotError( | |
| f"The userbot is in the channel but may not {self.WHAT} there, and the bot may not add admins. " | |
| "Give the bot the “Add new admins” right (it then promotes the userbot by itself), or make the userbot " | |
| f"an admin with “{self.RIGHT_NAME}” yourself." | |
| ) | |
| user = await self._bot_view_of_userbot() | |
| status = await self._userbot_status() | |
| had = _rights_dict(status[1]) if status else {} | |
| rank = status[2] if status else "" | |
| try: | |
| await self.ub.bot( | |
| functions.channels.EditAdminRequest( | |
| channel=peer_of(self.ch), | |
| user_id=user, | |
| admin_rights=types.ChatAdminRights(**{**had, self.RIGHT: True}), # what it had, plus this one right | |
| rank=rank, | |
| ) | |
| ) | |
| except errors.RPCError as e: | |
| name = type(e).__name__ | |
| hint = { | |
| "RightForbiddenError": "The bot may not give that right.", | |
| "UserAdminInvalidError": "The bot may not change that user's admin status.", | |
| "AdminsTooMuchError": "The channel already has the maximum number of admins.", | |
| "ChatAdminRequiredError": "The bot needs the “Add new admins” right.", | |
| }.get(name, str(e)[:150]) | |
| raise UserbotError(f"I could not make the userbot an admin ({name}). {hint}") | |
| self._promoted, self._bot_user = True, user | |
| if any(had.values()): | |
| self._before = (status[1], rank) | |
| self.notes.append(f"gave the userbot the “{self.RIGHT_NAME}” right on top of the ones it had") | |
| else: | |
| self.notes.append(f"made the userbot an admin with only “{self.RIGHT_NAME}”") | |
| for _ in range(12): # the new right can take a moment to show up for the other account | |
| if await self._userbot_has_right(): | |
| return | |
| await asyncio.sleep(0.5) | |
| raise UserbotError( | |
| f"I made the userbot an admin, but Telegram still does not let it {self.VERB} there. Check its rights." | |
| ) | |
| async def _bot_view_of_userbot(self): | |
| """The userbot account as an InputUser the BOT can use (needs the bot's own access hash for that user).""" | |
| bot, me = self.ub.bot, self.ub.me | |
| uid = me.id | |
| try: | |
| ent = await bot.get_input_entity(uid) | |
| if isinstance(ent, types.InputPeerUser): | |
| return types.InputUser(ent.user_id, ent.access_hash) | |
| except Exception: | |
| pass | |
| queries = [types.ChannelParticipantsRecent()] | |
| for q in (getattr(me, "username", None), getattr(me, "first_name", None)): | |
| if q: | |
| queries.append(types.ChannelParticipantsSearch(q)) | |
| for flt in queries: | |
| try: | |
| res = await bot( | |
| functions.channels.GetParticipantsRequest(channel=peer_of(self.ch), filter=flt, offset=0, limit=200, hash=0) | |
| ) | |
| except errors.RPCError: | |
| continue | |
| for u in getattr(res, "users", None) or []: | |
| if getattr(u, "id", None) == uid and getattr(u, "access_hash", None) is not None: | |
| return types.InputUser(u.id, u.access_hash) | |
| if getattr(me, "username", None): | |
| try: | |
| ent = await bot.get_input_entity(me.username) | |
| if isinstance(ent, types.InputPeerUser): | |
| return types.InputUser(ent.user_id, ent.access_hash) | |
| except Exception: | |
| pass | |
| raise UserbotError( | |
| "The userbot joined, but the bot cannot find that account in the channel's member list, so it cannot " | |
| f"promote it. Make the userbot an admin with “{self.RIGHT_NAME}” yourself." | |
| ) | |
| class UserbotDeleter(_ChannelJob): | |
| """Deletes messages in ONE channel with the userbot. Call close() when done, whatever happened.""" | |
| async def delete(self, ids: list) -> None: | |
| await self._begin() | |
| client = self.ub.client | |
| for i in range(0, len(ids), 100): | |
| part = list(ids[i : i + 100]) | |
| await flood_retry( | |
| lambda part=part: client(functions.channels.DeleteMessagesRequest(channel=self.peer, id=part)) | |
| ) | |
| _EDIT_HINTS = { | |
| "ChatAdminRequiredError": "It needs the “Edit messages of others” admin right in that channel.", | |
| "MessageAuthorRequiredError": "It needs the “Edit messages of others” admin right in that channel.", | |
| "MessageIdInvalidError": "The post may be gone, or Telegram does not let anyone edit it.", | |
| "InlineBotRequiredError": "That post was sent through another bot's inline mode, so only that bot can edit it.", | |
| "MessageEditTimeExpiredError": "Telegram no longer allows editing that message.", | |
| } | |
| class UserbotEditor(_ChannelJob): | |
| """Edits posts in ONE channel as the userbot. Call close() when done, whatever happened. | |
| The userbot is a user account: it can't set inline buttons, so its edit carries none and Telegram may drop the ones | |
| the post had. The caller puts them back afterwards.""" | |
| RIGHT = "edit_messages" | |
| RIGHT_NAME = "Edit messages of others" | |
| WHAT = "edit other people's posts" | |
| VERB = "edit" | |
| def __init__(self, ub: Userbot, ch): | |
| super().__init__(ub, ch) | |
| self._handed: list = [] # (the userbot account as the bot addresses it, the bot's id of the message): files put in its chat | |
| self.dm_problem: Optional[str] = None # why a file could not go through that chat (None: it did, or it was not tried) | |
| async def close(self) -> None: | |
| """Clean up the channel (rights, link) and the files the bot put into the userbot's chat with the bot.""" | |
| handed, self._handed = self._handed, [] | |
| for peer, message_id in handed: | |
| try: | |
| await self.ub.bot.delete_messages(peer, [message_id]) | |
| except Exception as e: # noqa: BLE001 - at worst a message is left in the helper account's chat | |
| log.info("could not delete the helper message in the userbot's chat: %s", type(e).__name__) | |
| await super().close() | |
| async def edit(self, message_id: int, *, text: str, entities: list, media=None, preview=None, rich=None) -> None: | |
| """Make the post show `text` (with `entities`) and, when given, `media`: what media_by_reference() returned (the | |
| file does not move at all) or a MediaFile (uploaded again). `rich` (a richmsg.Rich) makes it that rich message.""" | |
| await self._begin() | |
| client = self.ub.client | |
| try: | |
| if isinstance(media, MediaFile): | |
| media = await self._upload(media) | |
| if rich is not None: | |
| await flood_retry( | |
| lambda: with_rich(rich, lambda inp: edit_raw(client, self.peer, message_id, rich_message=inp)) | |
| ) | |
| else: | |
| await flood_retry( | |
| lambda: edit_raw( | |
| client, self.peer, message_id, text=text, entities=entities, preview=preview, media=media | |
| ) | |
| ) | |
| except errors.MessageNotModifiedError: | |
| return # it already looks like that | |
| except errors.RPCError as e: | |
| name = type(e).__name__ | |
| raise UserbotRefused( | |
| f"Telegram refused the userbot as well ({name}). {_EDIT_HINTS.get(name) or str(e)[:200]}", name | |
| ) | |
| # ----------------------------------------------------------- a file by way of the userbot's chat with the bot | |
| async def media_by_reference(self, ref: str): | |
| """The new photo / file as the userbot's own account holds it, ready to edit the post with - or None. | |
| The bot sends it to the userbot's chat with the bot (Telegram reuses the file it already has, nothing is | |
| downloaded or uploaded) and the userbot picks the message up there. The access data of a file belongs to one | |
| account, so the bot's own reference would be useless to the userbot: this is how the userbot gets its own. | |
| None when it does not work; `dm_problem` says why, and the caller uploads the file instead.""" | |
| await self._begin() | |
| self.dm_problem = None | |
| try: | |
| return await self._hand_over(ref) | |
| except (errors.RPCError, ValueError, TypeError, OSError) as e: | |
| self.dm_problem = f"it failed with {type(e).__name__}" | |
| log.info("could not hand the file to the userbot: %s", type(e).__name__, exc_info=True) | |
| return None | |
| async def _hand_over(self, ref: str): | |
| bot, client = self.ub.bot, self.ub.client | |
| want = media_key(media_from_ref(ref)) | |
| if want is None: | |
| raise ValueError("not a photo or a file") | |
| try: # both ends of the chat first: nothing is sent when one of them can't be found | |
| user = await self._bot_view_of_userbot() | |
| except UserbotError: | |
| self.dm_problem = "the bot cannot find the userbot account" | |
| return None | |
| try: | |
| bot_peer = await self._userbot_view_of_bot() | |
| except UserbotError as e: | |
| self.dm_problem = str(e) | |
| return None | |
| peer = types.InputPeerUser(user.user_id, user.access_hash) | |
| try: | |
| sent = await with_media(ref, lambda media: flood_retry(lambda: bot.send_file(peer, media, silent=True))) | |
| except errors.RPCError as e: | |
| name = type(e).__name__ | |
| self.dm_problem = ( | |
| "the userbot account has not started the bot yet (send /start to the bot from that account)" | |
| if name in _NOT_STARTED | |
| else f"Telegram would not let the bot send the file to the userbot ({name})" | |
| ) | |
| return None | |
| self._handed.append((peer, sent.id)) # deleted again in close() | |
| for i in range(_DM_POLLS): # the message has to reach the userbot's side first | |
| if i: | |
| await asyncio.sleep(_DM_WAIT) | |
| try: | |
| msgs = await flood_retry(lambda: client.get_messages(bot_peer, limit=10)) | |
| except errors.RPCError as e: | |
| self.dm_problem = f"the userbot could not read its chat with the bot ({type(e).__name__})" | |
| return None | |
| for m in msgs: # newest first; the same photo / file has the same id on both accounts | |
| if media_key(getattr(m, "media", None)) == want: | |
| return utils.get_input_media(m.media) | |
| self.dm_problem = "the file did not show up in the userbot's chat with the bot" | |
| return None | |
| async def _userbot_view_of_bot(self): | |
| """The bot as the userbot account addresses it (found by the bot's username).""" | |
| me = await self.ub.bot.get_me() | |
| name = getattr(me, "username", None) | |
| if not name: | |
| raise UserbotError("the bot has no username, so the userbot cannot find its chat with it") | |
| try: | |
| return await self.ub.client.get_input_entity(name) | |
| except (errors.RPCError, ValueError): | |
| raise UserbotError("the userbot could not find the bot by its username") | |
| # --------------------------------------------------------------------------------- the slow way | |
| async def _upload(self, f: MediaFile): | |
| """Upload a downloaded photo / file (with its thumbnail) as the new media of the post.""" | |
| client = self.ub.client | |
| handle = await flood_retry(lambda: client.upload_file(f.path, file_name=f.name)) | |
| if f.kind == "photo": | |
| return types.InputMediaUploadedPhoto(file=handle) | |
| thumb = None | |
| if f.thumb_path: | |
| try: | |
| thumb = await flood_retry(lambda: client.upload_file(f.thumb_path)) | |
| except (errors.RPCError, OSError) as e: # the file goes without it rather than not at all | |
| log.info("could not upload the thumbnail: %s", type(e).__name__) | |
| return types.InputMediaUploadedDocument( | |
| file=handle, | |
| mime_type=f.mime_type or "application/octet-stream", | |
| attributes=list(f.attributes), | |
| thumb=thumb, | |
| force_file=True if f.kind == "document" else None, # a file sent "as a file" stays one | |
| ) | |
| def delete_problem_text(ub, *, bot_error: Optional[str] = None, fallback_error: Optional[str] = None, tried: bool = False) -> str: | |
| """Why some messages could not be deleted, in words for the owner.""" | |
| parts = [] | |
| if bot_error: | |
| parts.append(f"Telegram refused the bot ({bot_error}).") | |
| if fallback_error: | |
| parts.append(f"The userbot could not help: {fallback_error}") | |
| elif tried: | |
| parts.append("The userbot tried as well, but the posts are still there.") | |
| elif ub is None or not getattr(ub, "enabled", False): | |
| parts.append("Posts older than 48 hours may need a userbot - see /userbot.") | |
| elif not ub.ready: | |
| parts.append(f"The userbot is not connected: {ub.last_error or 'unknown reason'}") | |
| return " ".join(parts) or "Telegram did not delete them (the bot's “Delete messages” right, or an age limit on old posts)." | |
| def edit_problem_text(ub, *, fallback_error: Optional[str] = None) -> str: | |
| """What to add when Telegram refused the bot's edit and the userbot could not take over (empty: nothing to add).""" | |
| if fallback_error: | |
| return f"The userbot could not do it either: {fallback_error}" | |
| if ub is None or not getattr(ub, "enabled", False): | |
| return "A userbot can edit such posts for you - see /userbot." | |
| if not ub.ready: | |
| return f"The userbot is not connected: {ub.last_error or 'unknown reason'}" | |
| return "" | |