Spaces:
Running
Running
| """Telegram handlers: commands, inline menus and the job entry points.""" | |
| from __future__ import annotations | |
| import asyncio | |
| import base64 | |
| import html | |
| import io | |
| import json | |
| import logging | |
| import random | |
| import re | |
| import shutil | |
| import time | |
| from dataclasses import dataclass | |
| from datetime import timezone | |
| from pathlib import Path | |
| from typing import Any, Awaitable, Callable, Dict, List, Optional, Set, Tuple | |
| import telethon | |
| from telethon import errors, events, functions, types | |
| from telethon.tl.custom import Button | |
| from . import fasttransfer as transfer | |
| from . import errorlog, media, mediainfo, panels, richtext, speed, sysinfo | |
| from .config import Config | |
| from .db import MAX_FORMATS, Database | |
| from .dbflow import PREFIX as DBFLOW_PREFIX | |
| from .dbflow import DbFlow | |
| from .errors import FormatError | |
| from .formats import ( | |
| FIELD_KINDS, | |
| KIND_LABELS, | |
| NAME_MAX, | |
| FormatSettings, | |
| default_auto_format, | |
| default_metadata_changes, | |
| has_default_metadata, | |
| other_kind, | |
| validate_name, | |
| ) | |
| from .parser import compute_new_name, missing_variables, parse_caption, parse_info, split_ext | |
| from .pipeline import PROBE_EXTS, Renamer, embedded_thumb, original_name_of, source_ext | |
| from .queue import Job, JobQueue | |
| from .sequence import SeqItem, SequenceState, analyse, format_report | |
| from .settings_ops import CLEAR_WORDS, META_COMMANDS, apply_setting, parse_channel, parse_format_channel | |
| from .telegraph import Telegraph, TelegraphError | |
| from .utils import clip, esc, human_size, human_uptime | |
| log = logging.getLogger(__name__) | |
| COMMANDS = [ | |
| ("start", "Open the main menu"), | |
| ("help", "How to use the bot"), | |
| ("rename", "Rename one file with a manual format"), | |
| ("formats", "Auto and manual rename formats"), | |
| ("sequence", "Collect files to sort by episode"), | |
| ("sort", "Sort the collected files by episode"), | |
| ("mode", "Choose Filename or Caption mode"), | |
| ("filetype", "Choose how files are sent"), | |
| ("metadata", "Metadata and track settings"), | |
| ("keepaudio", "Keep audio tracks by language or number"), | |
| ("keepsubtitle", "Keep subtitle tracks by language or number"), | |
| ("tracks", "List the tracks of a file (reply to it)"), | |
| ("check", "See what I can read from a file (reply to it)"), | |
| ("mediainfo", "Full media details of a file on Telegraph (reply to it)"), | |
| ("ethumb", "Extract the thumbnail of a file or video"), | |
| ("size", "Get the dimensions of a picture (reply to it)"), | |
| ("resize", "Resize a picture to an exact size (reply to it)"), | |
| ("queue", "Show your queued files"), | |
| ("cancel", "Cancel your current job"), | |
| ("speed", "Test the bot's internet speed (Ookla)"), | |
| ("status", "Show basic bot uptime, ping, and version information."), | |
| ("stats", "Show detailed system stats such as users, CPU, RAM, and disk usage."), | |
| ] | |
| OWNER_COMMANDS = {"addadmin", "deladmin", "admins", "smedia"} | |
| BACK = { | |
| "remove": "rename", | |
| "mt_title": "meta", "mt_author": "meta", "mt_artist": "meta", | |
| "mt_audio": "meta", "mt_subtitle": "meta", "mt_video": "meta", | |
| "keep_audio": "tool", "keep_subtitle": "tool", | |
| } | |
| PANEL_NAMES = {"home", "rename", "meta", "tool", "ftype", "help", "admin", "formats", "auto"} | |
| QUEUE_SETTLE = 1.5 # seconds an idle worker waits so files sent together are queued in message order | |
| ONE_FILE_WAIT = 600 # seconds the bot waits for the file after /rename (or /ethumb ...) before forgetting about it | |
| ONE_SHOT_GRACE = 8 # seconds after a one-file command during which further files get a hint, not an auto-rename | |
| ONE_SHOT_TAKES = { # what each one-file command consumes; further files of that kind sent right after it are skipped | |
| "rename": {"doc"}, "ethumb": {"doc"}, "smedia": {"doc", "photo", "gif"}, | |
| } | |
| ETHUMB_FRAME_WIDTH = 1280 | |
| START_MEDIA_MAX_BYTES = 10 * 1024 * 1024 # biggest /start picture or GIF; a copy of it is kept in the database | |
| START_IMAGE_BOX = 1280 # pictures are shrunk to fit this box, as Telegram does with photos | |
| RESIZE_MAX_BYTES = 20 * 1024 * 1024 # biggest picture /size and /resize read | |
| RESIZE_MAX_SIDE = 4096 # the biggest width or height /resize makes | |
| PHOTO_MAX_SIDE = 1280 # Telegram shrinks a photo to this; /resize sends anything bigger as a file so its size stays exact | |
| PHOTO_MAX_BYTES = 10 * 1024 * 1024 # biggest photo Telegram takes | |
| PHOTO_MAX_RATIO = 20 # ... and the longer side of a photo can't be more than this many times the shorter one | |
| HINT_SECONDS = 5 # how long the hint of /resize stays before it is removed | |
| SEQUENCE_WAIT = 1800 # seconds /sequence keeps the collected files without a new one arriving | |
| SEQUENCE_MAX = 1000 # files one /sequence collects | |
| SEQUENCE_STATUS_DELAY = 1.2 # seconds without a new file before "N files collected" is sent | |
| SORT_GAP = 0.3 # seconds between the files /sort sends back | |
| SORT_FAST_FILES = 200 # /sort sends this many files at that pace; the rest of a long batch goes at SORT_SLOW_GAP | |
| SORT_SLOW_GAP = 0.8 # about one file a second, what Telegram lets a bot keep sending into one chat without flood waits | |
| SORT_LOOKUP = 100 # files /sort looks up per request (Telegram returns at most 100 messages at a time) | |
| SORT_MAX_WAIT = 120 # a Telegram flood wait longer than this (seconds) stops /sort instead of being waited out | |
| SORT_PROGRESS_EVERY = 10 # files between two updates of the "Sorting…" message | |
| TRACKS_PROBE_BYTES = 12 * 1024 * 1024 | |
| TRACKS_FULL_LIMIT = 700 * 1024 * 1024 | |
| SHARE_PREFIX = "fmt_" # /start fmt_<token> opens a shared format: it is the link on the card the Share format button makes | |
| SHARE_QUERY = re.compile(r"share_(\d{1,9})") # what that button types into the chosen chat: share_<format id> | |
| SHARE_TOKEN = re.compile(r"[A-Za-z0-9_-]{1,32}") | |
| INLINE_RECHECK = 60 # seconds before the bot looks again whether inline mode is on, while it was off | |
| class Pending: | |
| kind: str | |
| chat_id: int | |
| msg_id: int | |
| back: str | |
| is_photo: bool | |
| expires: float | |
| fid: int = 0 # the saved format a "fmt:*" prompt is about (0 = one that is being created) | |
| fmt_kind: str = "" # the kind (auto | manual) of the format that is being created | |
| class Mode: | |
| """What a user's next files are for: a /rename format session, /sequence, /ethumb or /smedia.""" | |
| kind: str | |
| ttl: float | |
| expires: float | |
| data: Any = None | |
| dropped: str = "" # a note about what starting this mode threw away, for the reply that announces it | |
| def cmd_pattern(name: str) -> "re.Pattern[str]": | |
| return re.compile(rf"(?is)^/{name}(?:@\w+)?(?:\s+(.*))?$") | |
| MEDIAINFO_CAPTION = cmd_pattern("mediainfo") # a file sent with this as its caption is for /mediainfo, not for renaming | |
| def visible_len(html_text: str) -> int: | |
| return len(html.unescape(re.sub(r"<[^>]+>", "", html_text))) | |
| def plain(html_text: str) -> str: | |
| return html.unescape(re.sub(r"<[^>]+>", "", html_text)) | |
| def is_renameable(message: Any) -> bool: | |
| return ( | |
| getattr(message, "document", None) is not None | |
| and not (message.sticker or message.gif or message.voice or message.video_note) | |
| ) | |
| def is_image(message: Any) -> bool: | |
| """A photo, or a file that is a still picture (not a GIF animation).""" | |
| if getattr(message, "photo", None) is not None: | |
| return True | |
| f = getattr(message, "file", None) | |
| return bool( | |
| getattr(message, "document", None) is not None and not getattr(message, "gif", None) | |
| and f is not None and (f.mime_type or "").startswith("image/") | |
| ) | |
| def fits_as_photo(width: int, height: int, size: int) -> bool: | |
| """Does Telegram keep a picture of this many pixels and bytes as a photo, exactly as it is? (It shrinks big ones.)""" | |
| longest, shortest = max(width, height), min(width, height) | |
| return longest <= PHOTO_MAX_SIDE and longest <= PHOTO_MAX_RATIO * shortest and size <= PHOTO_MAX_BYTES | |
| def sent_as(message: Any) -> str: | |
| """How Telegram labels the file: a video, an audio file, a voice message ..., or a plain document.""" | |
| for attr, label in ( | |
| ("voice", "Voice message"), ("video_note", "Video message"), ("gif", "GIF"), ("sticker", "Sticker"), | |
| ("video", "Video"), ("audio", "Audio"), | |
| ): | |
| if getattr(message, attr, None) is not None: | |
| return label | |
| return "Document" | |
| def is_playable(message: Any, name: str) -> bool: | |
| """A file that should hold audio or video streams, going by its MIME type and extension.""" | |
| mime = (getattr(message.file, "mime_type", "") or "").lower() | |
| return mime.startswith(("video/", "audio/")) or source_ext(message, name).lower() in PROBE_EXTS | |
| def telegram_facts(message: Any, caption: bool = True) -> Dict[str, str]: | |
| """What Telegram knows about the file, for the "Telegram" part of the /mediainfo page.""" | |
| date = getattr(message, "date", None) | |
| stamp = "" | |
| if date is not None: | |
| if date.tzinfo is None: | |
| date = date.replace(tzinfo=timezone.utc) | |
| stamp = date.astimezone(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC") | |
| return { | |
| "mime": getattr(message.file, "mime_type", "") or "", | |
| "sent_as": sent_as(message), | |
| "sent_on": stamp, | |
| "dc": str(getattr(message.document, "dc_id", "") or ""), | |
| "caption": clip((message.raw_text or "").strip(), 500) if caption else "", | |
| } | |
| class App: | |
| PENDING_TTL = 300 | |
| def __init__(self, client: Any, cfg: Config, db: Database): | |
| self.client = client | |
| self.cfg = cfg | |
| self.db = db | |
| self.admins: Set[int] = set(cfg.admin_ids) | |
| self.pending: Dict[int, Pending] = {} | |
| self.modes: Dict[int, Mode] = {} | |
| self._recent: Dict[int, List[Any]] = {} # uid -> [command, finished_at, hint_sent] of the last one-file command | |
| self._sorting: Dict[int, asyncio.Event] = {} # users whose /sort is sending files; set to stop it | |
| self._speed_lock = asyncio.Lock() # one speed test at a time | |
| self.telegraph = Telegraph(self._load_telegraph_token, self._save_telegraph_token) | |
| self._tasks: Set["asyncio.Future[Any]"] = set() | |
| self.renamer = Renamer(client, cfg, db) | |
| self.queue = JobQueue(self.renamer.process, workers=cfg.max_jobs, on_cancelled=self._on_job_cancelled, settle=QUEUE_SETTLE) | |
| # uploading a manually renamed episode to the Database Channel (and having it published) needs both chats | |
| self.dbflow: Optional[DbFlow] = DbFlow(client, cfg) if cfg.upload_enabled else None | |
| if self.dbflow is not None: | |
| self.renamer.on_delivered = self.dbflow.offer | |
| self._warned: Set[int] = set() | |
| self._username: Optional[str] = None # the bot's own @username, for the links on shared-format cards | |
| self._inline_on = False # is inline mode on for the bot? See _check_inline | |
| self._inline_checked: Optional[float] = None | |
| # ------------------------------------------------------------------ auth | |
| def is_owner(self, uid: Optional[int]) -> bool: | |
| return uid is not None and uid == self.cfg.owner_id | |
| def is_allowed(self, uid: Optional[int]) -> bool: | |
| return uid is not None and (uid == self.cfg.owner_id or uid in self.admins) | |
| # ----------------------------------------------------------------- setup | |
| async def setup(self) -> None: | |
| self.admins |= set(await self.db.list_admins()) | |
| self.client.parse_mode = "html" | |
| self._register() | |
| await self._set_commands() | |
| await self._check_inline(force=True) | |
| if self.cfg.log_channel: | |
| try: | |
| await self.client.get_input_entity(self.cfg.log_channel) | |
| except Exception as exc: # noqa: BLE001 | |
| log.warning("cannot access LOG_CHANNEL %s: %s (add the bot there as an admin)", self.cfg.log_channel, exc) | |
| await self._check_upload_chats() | |
| self.queue.start() | |
| async def _check_upload_chats(self) -> None: | |
| """Say in the log what the upload to the Database Channel cannot reach (or that it is switched off).""" | |
| cfg = self.cfg | |
| if self.dbflow is None: | |
| if cfg.db_channel or cfg.comm_chat: | |
| log.warning( | |
| "uploading to the Database Channel is off: it needs both DB_CHANNEL (%s) and COMMUNICATION_CHAT (%s)", | |
| cfg.db_channel or "missing", cfg.comm_chat or "missing", | |
| ) | |
| return | |
| log.info( | |
| "uploading to the Database Channel is on: Database Channel %s, communication chat %s, channel post bot %s", | |
| cfg.db_channel, cfg.comm_chat, cfg.post_bot_id, | |
| ) | |
| for name, ref in (("DB_CHANNEL", cfg.db_channel), ("COMMUNICATION_CHAT", cfg.comm_chat)): | |
| try: | |
| await self.client.get_input_entity(ref) | |
| except Exception as exc: # noqa: BLE001 | |
| log.warning("cannot access %s %s: %s (add the bot there as an admin)", name, ref, exc) | |
| async def shutdown(self) -> None: | |
| for task in list(self._tasks): | |
| task.cancel() | |
| if self.dbflow is not None: | |
| await self.dbflow.shutdown() | |
| await self.queue.stop() | |
| async def _set_commands(self) -> None: | |
| try: | |
| await self.client( | |
| functions.bots.SetBotCommandsRequest( | |
| scope=types.BotCommandScopeDefault(), | |
| lang_code="", | |
| commands=[types.BotCommand(command=c, description=d) for c, d in COMMANDS], | |
| ) | |
| ) | |
| except Exception as exc: # noqa: BLE001 | |
| log.warning("could not set the bot command menu: %s", exc) | |
| async def _log_send(self, text: str) -> None: | |
| if self.cfg.log_channel: | |
| await self.client.send_message(self.cfg.log_channel, text, parse_mode="html", link_preview=False) | |
| # --------------------------------------------------------------- plumbing | |
| def _guard(self, fn: Callable[[Any], Awaitable[None]], *, owner_only: bool = False): | |
| async def wrapper(event: Any) -> None: | |
| uid = event.sender_id | |
| allowed = self.is_owner(uid) if owner_only else self.is_allowed(uid) | |
| if not allowed: | |
| if owner_only and self.is_allowed(uid): # a known user trying an owner command | |
| try: | |
| if hasattr(event, "data"): | |
| await event.answer("Owner only.", alert=True) | |
| else: | |
| await event.respond("⛔ Only the owner can do that.") | |
| except Exception: # noqa: BLE001 | |
| pass | |
| else: | |
| await self._stranger(uid) | |
| return | |
| context = errorlog.error_context.set(lambda: errorlog.describe_update(event)) # for the log channel report | |
| try: | |
| await fn(event) | |
| except events.StopPropagation: | |
| raise | |
| except errors.MessageNotModifiedError: | |
| pass | |
| except Exception: # noqa: BLE001 | |
| log.exception("Error handling update:") | |
| try: | |
| if hasattr(event, "data"): | |
| await event.answer("⚠️ Something went wrong.", alert=True) | |
| else: | |
| await event.respond("⚠️ Something went wrong. Please try again.") | |
| except Exception: # noqa: BLE001 | |
| pass | |
| finally: | |
| errorlog.error_context.reset(context) | |
| return wrapper | |
| def _guard_inline(self, fn: Callable[[Any], Awaitable[None]]): | |
| guarded = self._guard(fn) | |
| async def wrapper(event: Any) -> None: | |
| if self.is_allowed(event.sender_id): | |
| await guarded(event) | |
| else: # anyone can type the bot's name in a chat: strangers get no results, and no notice in the log | |
| await self._answer_inline(event, []) | |
| return wrapper | |
| async def _stranger(self, uid: Optional[int]) -> None: | |
| if uid is None or uid in self._warned: | |
| return | |
| self._warned.add(uid) | |
| log.info("ignored unauthorised user %s", uid) | |
| try: | |
| await self._log_send(f"🚫 Ignored a message from <code>{uid}</code>. Use /addadmin {uid} to allow them.") | |
| except Exception: # noqa: BLE001 | |
| pass | |
| def _register(self) -> None: | |
| add = self.client.add_event_handler | |
| for name, handler in self._command_table().items(): | |
| add( | |
| self._guard(handler, owner_only=name in OWNER_COMMANDS), | |
| events.NewMessage(incoming=True, pattern=cmd_pattern(name), func=lambda e: e.is_private), | |
| ) | |
| add(self._guard(self.on_media), events.NewMessage(incoming=True, func=lambda e: e.is_private and is_renameable(e.message))) | |
| add(self._guard(self.on_photo), events.NewMessage(incoming=True, func=lambda e: e.is_private and e.message.photo is not None)) | |
| add(self._guard(self.on_gif), events.NewMessage(incoming=True, func=lambda e: e.is_private and e.message.gif is not None)) | |
| add(self._guard(self.on_text), events.NewMessage(incoming=True, func=self._wants_text)) | |
| add(self._guard(self.on_callback), events.CallbackQuery()) | |
| add(self._guard_inline(self.on_inline), events.InlineQuery()) | |
| if self.dbflow is not None: # the channel post bot's answers in the communication chat (not a user of this bot: no guard) | |
| add( | |
| self.dbflow.on_chat_message, | |
| events.NewMessage(incoming=True, chats=[self.cfg.comm_chat], from_users=[self.cfg.post_bot_id]), | |
| ) | |
| add(self._on_stranger, events.NewMessage(incoming=True, func=lambda e: e.is_private and not self.is_allowed(e.sender_id))) | |
| async def _on_stranger(self, event: Any) -> None: | |
| await self._stranger(event.sender_id) | |
| def _wants_text(self, e: Any) -> bool: | |
| text = e.raw_text or "" | |
| return bool(e.is_private and text and not text.startswith("/") and e.sender_id in self.pending and not e.message.media) | |
| def _command_table(self) -> Dict[str, Callable[[Any], Awaitable[None]]]: | |
| table: Dict[str, Callable[[Any], Awaitable[None]]] = { | |
| "start": self.cmd_start, "help": self.cmd_help, "rename": self.cmd_rename, "formats": self.cmd_formats, "format": self.cmd_formats, | |
| "sequence": self.cmd_sequence, "sort": self.cmd_sort, "mode": self.cmd_mode, "filetype": self.cmd_filetype, | |
| "metadata": self.cmd_metadata, "keepaudio": self._cmd_keep("keep_audio", "Keep Audio", "/keepaudio English Japanese"), | |
| "keepsubtitle": self._cmd_keep("keep_subtitle", "Keep Subtitle", "/keepsubtitle English"), | |
| "tracks": self.cmd_tracks, "check": self.cmd_check, "mediainfo": self.cmd_mediainfo, | |
| "ethumb": self.cmd_ethumb, "size": self.cmd_size, "resize": self.cmd_resize, | |
| "removewords": self.cmd_remove, "setchannel": self.cmd_setchannel, "queue": self.cmd_queue, | |
| "cancel": self.cmd_cancel, "speed": self.cmd_speed, "status": self.cmd_status, "stats": self.cmd_stats, | |
| "addadmin": self.cmd_addadmin, "deladmin": self.cmd_deladmin, "admins": self.cmd_admins, "smedia": self.cmd_smedia, | |
| } | |
| for command, kind in META_COMMANDS.items(): | |
| table[command] = self._cmd_meta_field(command, kind) | |
| return table | |
| def _args(event: Any) -> str: | |
| return (event.pattern_match.group(1) or "").strip() if event.pattern_match else "" | |
| def _caption_input(event: Any) -> str: | |
| """The caption the person sent. Formatting they put on it in Telegram (bold, italic, links...) becomes the HTML that is | |
| stored, so nobody has to type tags; without any formatting it is taken as typed.""" | |
| typed = event.raw_text or "" | |
| if typed.strip().lower() in CLEAR_WORDS: # "reset" stays a reset even if it was made bold | |
| return typed | |
| entities = getattr(getattr(event, "message", None), "entities", None) | |
| return richtext.from_entities(typed, entities) or typed | |
| async def _reply(self, event: Any, text: str, rows: Optional[List[panels.Row]] = None) -> Any: | |
| buttons = panels.to_buttons(rows) if rows else None | |
| try: | |
| return await event.respond(text, buttons=buttons, parse_mode="html", link_preview=False) | |
| except errors.BotInlineDisabledError: # inline mode was switched off in @BotFather: the Share button can't be shown | |
| plain_rows = panels.without_inline(rows or []) | |
| if plain_rows == rows: | |
| raise | |
| self._inline_refused() | |
| return await self._reply(event, text, plain_rows) | |
| # ----------------------------------------------------------------- panels | |
| async def panel(self, uid: int, name: str) -> panels.Panel: | |
| s = await self.db.get_settings(uid) | |
| if name == "home": | |
| return panels.home(s, owner=self.is_owner(uid), auto=await self.db.get_auto_format(uid)) | |
| if name == "rename": | |
| return panels.rename(s, await self.db.get_auto_format(uid)) | |
| if name == "meta": | |
| return panels.metadata(s) | |
| if name == "tool": | |
| return panels.tool(s) | |
| if name == "ftype": | |
| return panels.filetype(s) | |
| if name == "help": | |
| return panels.help_panel() | |
| if name == "formats": | |
| return panels.formats_home( | |
| await self.db.count_formats(uid, "auto"), await self.db.count_formats(uid, "manual"), MAX_FORMATS | |
| ) | |
| if name == "auto": | |
| active = await self.db.get_auto_format(uid) | |
| return panels.auto_menu(await self.db.list_formats(uid, "auto"), active["id"] if active else 0, MAX_FORMATS) | |
| if name == "admin": | |
| return panels.admin( | |
| self.cfg.owner_id, sorted(self.admins), self.queue.running, self.queue.waiting, self.cfg.log_channel | |
| ) | |
| raise KeyError(name) | |
| async def _remove_message(event: Any) -> None: | |
| try: | |
| await event.delete() | |
| except Exception: # noqa: BLE001 - Telegram refuses to delete messages older than 48 hours | |
| try: | |
| await event.edit("Closed.", buttons=None) | |
| except Exception: # noqa: BLE001 | |
| pass | |
| async def _edit_panel(self, event: Any, text: str, rows: List[panels.Row]) -> None: | |
| try: | |
| await event.edit(text, buttons=panels.to_buttons(rows), parse_mode="html", link_preview=False) | |
| except errors.MessageNotModifiedError: | |
| pass | |
| except errors.BotInlineDisabledError: # inline mode was switched off in @BotFather: the Share button can't be shown | |
| plain_rows = panels.without_inline(rows) | |
| if plain_rows == rows: | |
| raise | |
| self._inline_refused() | |
| await self._edit_panel(event, text, plain_rows) | |
| # ------------------------------------------------------------------ modes | |
| def _mode(self, uid: int, *kinds: str) -> Optional[Mode]: | |
| """The user's active mode (of one of ``kinds`` if given); an expired one is dropped.""" | |
| mode = self.modes.get(uid) | |
| if mode is not None and mode.expires < time.time(): | |
| self._end_mode(uid) | |
| return None | |
| if mode is None or (kinds and mode.kind not in kinds): | |
| return None | |
| return mode | |
| def _start_mode(self, uid: int, kind: str, ttl: float, data: Any = None) -> Mode: | |
| previous = self._end_mode(uid) | |
| mode = Mode(kind=kind, ttl=ttl, expires=time.time() + ttl, data=data) | |
| if previous is not None and previous.kind == "sequence" and previous.expires >= time.time() and previous.data.items: | |
| mode.dropped = f"⚠️ Your /sequence with {len(previous.data.items)} file(s) was discarded." | |
| self.modes[uid] = mode | |
| return mode | |
| def _end_mode(self, uid: int) -> Optional[Mode]: | |
| mode = self.modes.pop(uid, None) | |
| if mode is not None and isinstance(mode.data, SequenceState): | |
| mode.data.closed = True | |
| return mode | |
| def _spawn(self, coro: Awaitable[Any]) -> "asyncio.Future[Any]": | |
| task = asyncio.ensure_future(coro) | |
| self._tasks.add(task) | |
| task.add_done_callback(self._tasks.discard) | |
| return task | |
| def _consume(self, uid: int, kind: str) -> Optional[Mode]: | |
| """Take the user's one-file mode. Nothing awaits in here, so of several files that arrive together only one gets it.""" | |
| mode = self._mode(uid, kind) | |
| if mode is None: | |
| return None | |
| del self.modes[uid] | |
| self._recent[uid] = [kind, time.time(), False] | |
| return mode | |
| def _extra_file_hint(self, uid: int, via: str = "doc") -> Optional[str]: | |
| """After a one-file command: ``""`` for a further file to skip silently, a hint for the first one, ``None`` once the grace is over. | |
| ``via`` is how the file arrived (``doc``, ``photo`` or ``gif``); only files the command itself would have taken are skipped. | |
| """ | |
| rec = self._recent.get(uid) | |
| if rec is None: | |
| return None | |
| if time.time() - rec[1] > ONE_SHOT_GRACE: | |
| del self._recent[uid] | |
| return None | |
| if via not in ONE_SHOT_TAKES.get(rec[0], {"doc"}): | |
| return None | |
| if rec[2]: | |
| return "" | |
| rec[2] = True | |
| return f"ℹ️ /{rec[0]} handles one file at a time, so I skipped this one. Send /{rec[0]} again for the next file." | |
| # --------------------------------------------------------------- commands | |
| async def cmd_start(self, event: Any) -> None: | |
| self.pending.pop(event.sender_id, None) | |
| arg = self._args(event) | |
| if arg.startswith(SHARE_PREFIX): # somebody opened the link on a shared format | |
| await self._open_shared(event, arg[len(SHARE_PREFIX):]) | |
| return | |
| text, rows = await self.panel(event.sender_id, "home") | |
| if not await self._send_start_media(event.chat_id, text, rows): | |
| await self._reply(event, text, rows) | |
| async def cmd_help(self, event: Any) -> None: | |
| text, _ = panels.help_panel() | |
| await self._reply(event, text, [[("✖ Close", "x:close")]]) | |
| async def cmd_mode(self, event: Any) -> None: | |
| arg = self._args(event).lower() | |
| uid = event.sender_id | |
| if arg in ("filename", "caption"): | |
| await self.db.update_settings(uid, mode=arg) | |
| await self._reply(event, f"✅ Rename mode: <code>{arg.upper()}</code>") | |
| return | |
| text, rows = await self.panel(uid, "rename") | |
| await self._reply(event, text, rows) | |
| async def cmd_filetype(self, event: Any) -> None: | |
| arg = self._args(event).lower() | |
| uid = event.sender_id | |
| aliases = {"original": "same", "file": "document", "doc": "document"} | |
| arg = aliases.get(arg, arg) | |
| if arg in ("same", "document", "video", "audio"): | |
| await self.db.update_settings(uid, file_type=arg) | |
| label = "same as original" if arg == "same" else arg | |
| await self._reply(event, f"✅ Files will be sent as: <code>{label}</code>") | |
| return | |
| text, rows = await self.panel(uid, "ftype") | |
| await self._reply(event, text, rows) | |
| async def cmd_metadata(self, event: Any) -> None: | |
| arg = self._args(event).lower() | |
| uid = event.sender_id | |
| if arg in ("on", "off"): | |
| await self.db.update_settings(uid, metadata_on=(arg == "on")) | |
| await self._reply(event, f"✅ Metadata is now <b>{arg.upper()}</b>") | |
| return | |
| text, rows = await self.panel(uid, "meta") | |
| await self._reply(event, text, rows) | |
| async def cmd_remove(self, event: Any) -> None: | |
| arg = self._args(event) | |
| uid = event.sender_id | |
| if arg: | |
| _ok, msg = await apply_setting(self.db, uid, "remove", arg) | |
| await self._reply(event, msg) | |
| return | |
| s = await self.db.get_settings(uid) | |
| words = ", ".join(s["remove_words"]) or "none" | |
| await self._reply(event, f"<b>Words stripped from names:</b> {esc(words)}\n\nSet: <code>/removewords @Channel, [Group]</code> · clear: <code>/removewords off</code>") | |
| def _cmd_meta_field(self, command: str, kind: str) -> Callable[[Any], Awaitable[None]]: | |
| async def handler(event: Any) -> None: | |
| arg = self._args(event) | |
| if not arg: | |
| s = await self.db.get_settings(event.sender_id) | |
| field = kind[3:] | |
| current = s["meta"].get(field) or "—" | |
| await self._reply(event, f"<b>{field.title()}:</b> <code>{esc(current)}</code>\n\nSet it: <code>/{command} value</code> · clear: <code>/{command} off</code>") | |
| return | |
| _ok, msg = await apply_setting(self.db, event.sender_id, kind, arg) | |
| await self._reply(event, msg) | |
| handler.__name__ = f"cmd_{command}" | |
| return handler | |
| def _cmd_keep(self, kind: str, label: str, example: str) -> Callable[[Any], Awaitable[None]]: | |
| async def handler(event: Any) -> None: | |
| arg = self._args(event) | |
| uid = event.sender_id | |
| if not arg: | |
| s = await self.db.get_settings(uid) | |
| await self._reply( | |
| event, | |
| f"<b>{label}:</b> {esc(panels.fmt_tokens(s[kind]))}\n\n" | |
| f"Set it with language names or track numbers:\n<code>{example}</code>\n<code>{example.split()[0]} 2 1</code>\n" | |
| f"Only the listed tracks are kept, in that order. <code>{example.split()[0]} off</code> keeps everything.\n" | |
| "Reply to a file with /tracks to see its track numbers.", | |
| ) | |
| return | |
| _ok, msg = await apply_setting(self.db, uid, kind, arg) | |
| await self._reply(event, msg) | |
| handler.__name__ = f"cmd_{kind}" | |
| return handler | |
| async def cmd_setchannel(self, event: Any) -> None: | |
| arg = self._args(event) | |
| uid = event.sender_id | |
| if not arg: | |
| s = await self.db.get_settings(uid) | |
| current = f"<code>{s['target_channel']}</code>" if s["target_channel"] else "not set (files come back to this chat)" | |
| await self._reply(event, f"<b>Send renamed files to:</b> {current}\n\nSet: <code>/setchannel -1001234567890</code> (add me to the channel as an admin first) · clear: <code>/setchannel off</code>") | |
| return | |
| ok, cid, err = parse_channel(arg) | |
| if not ok: | |
| await self._reply(event, err) | |
| return | |
| if cid is not None: | |
| try: | |
| await self.client.get_input_entity(cid) | |
| except Exception: # noqa: BLE001 | |
| await self._reply(event, "❌ I can't access that chat. Add me to the channel as an admin that can post, then send the id again.") | |
| return | |
| await self.db.update_settings(uid, target_channel=cid) | |
| await self._reply(event, f"✅ Renamed files will go to <code>{cid}</code>." if cid else "✅ Renamed files will come back to this chat.") | |
| async def on_photo(self, event: Any) -> None: | |
| """A photo is taken only when a format is waiting for its thumbnail (or /smedia for its picture); any other one is ignored.""" | |
| uid = event.sender_id | |
| p = self.pending.get(uid) | |
| if not (p is not None and p.kind == "fmt:thumb" and p.expires >= time.time()): | |
| if await self._image_mode(event): | |
| return | |
| hint = self._extra_file_hint(uid, "photo") | |
| if hint: | |
| await self._reply(event, hint) | |
| return | |
| raw = await event.download_media(bytes) | |
| if not raw: | |
| await self._reply(event, "❌ I couldn't download that photo.") | |
| return | |
| try: | |
| thumb = await asyncio.to_thread(media.prepare_thumb, raw) | |
| except Exception: # noqa: BLE001 | |
| await self._reply(event, "❌ I couldn't read that image.") | |
| return | |
| await self._format_thumb_saved(event, p, thumb) | |
| async def on_gif(self, event: Any) -> None: | |
| if await self._image_mode(event): | |
| return | |
| hint = self._extra_file_hint(event.sender_id, "gif") | |
| if hint: | |
| await self._reply(event, hint) | |
| # ------------------------------------------------------------ rename flow | |
| async def cmd_formats(self, event: Any) -> None: | |
| self.pending.pop(event.sender_id, None) | |
| text, rows = await self.panel(event.sender_id, "formats") | |
| await self._reply(event, text, rows) | |
| async def cmd_rename(self, event: Any) -> None: | |
| uid = event.sender_id | |
| self.pending.pop(uid, None) | |
| text, rows = panels.rename_picker(await self.db.list_formats(uid, "manual")) | |
| await self._reply(event, text, rows) | |
| async def on_media(self, event: Any) -> None: | |
| uid = event.sender_id | |
| if MEDIAINFO_CAPTION.match(event.raw_text or ""): | |
| return # cmd_mediainfo has it | |
| mode = self._consume(uid, "rename") | |
| if mode is not None: | |
| await self._media_rename(event, mode) | |
| return | |
| mode = self._consume(uid, "ethumb") | |
| if mode is not None: | |
| await self._send_thumbnail(event, event.message) | |
| return | |
| if await self._image_mode(event): | |
| return | |
| mode = self._mode(uid, "sequence") | |
| if mode is not None: | |
| notice = self._collect(event, mode) | |
| if notice: | |
| await self._reply(event, notice) | |
| return | |
| hint = self._extra_file_hint(uid) | |
| if hint is not None: | |
| if hint: | |
| await self._reply(event, hint) | |
| return | |
| await self.enqueue(event, event.message, fmt=await self.db.get_auto_format(uid)) # None: no auto format, the defaults | |
| async def _media_rename(self, event: Any, mode: Mode) -> None: | |
| fmt = await self.db.get_format(event.sender_id, mode.data["fid"]) | |
| if fmt is None or fmt["kind"] != "manual": | |
| await self._reply(event, "⚠️ That format was deleted. Use /rename to pick another one, then send the file again.") | |
| return | |
| await self.enqueue(event, event.message, fmt=fmt) | |
| # --------------------------------------------------------- /sequence, /sort | |
| async def cmd_sequence(self, event: Any) -> None: | |
| uid = event.sender_id | |
| self.pending.pop(uid, None) | |
| mode = self._mode(uid, "sequence") | |
| if mode is not None and mode.data.items: # a second /sequence must not throw the collected files away | |
| mode.expires = time.time() + mode.ttl | |
| text, rows = panels.sequence_status(len(mode.data.items), SEQUENCE_MAX) | |
| await self._reply(event, "You're already collecting files for a sequence.\n\n" + text, rows) | |
| return | |
| mode = self._start_mode(uid, "sequence", SEQUENCE_WAIT, SequenceState()) | |
| text, rows = panels.sequence_intro(SEQUENCE_MAX) | |
| await self._reply(event, text + (f"\n\n{mode.dropped}" if mode.dropped else ""), rows) | |
| def _collect(self, event: Any, mode: Mode) -> Optional[str]: | |
| """Add a file to the user's sequence. Nothing awaits in here, so files that arrive together are all kept.""" | |
| state: SequenceState = mode.data | |
| mode.expires = time.time() + mode.ttl | |
| if len(state.items) >= SEQUENCE_MAX: | |
| if state.full_noted: | |
| return None | |
| state.full_noted = True | |
| return f"⚠️ That's the limit: {SEQUENCE_MAX} files. Send /sort now, or /cancel to start over." | |
| msg = event.message | |
| state.items.append(SeqItem(msg_id=msg.id, name=original_name_of(msg), caption=msg.raw_text or "")) | |
| state.version += 1 | |
| if state.task is None or state.task.done(): | |
| state.task = self._spawn(self._sequence_status(event.chat_id, state)) | |
| return None | |
| async def _sequence_status(self, chat_id: int, state: SequenceState) -> None: | |
| """Send "N files collected" once the files stop arriving, so it is the newest message in the chat.""" | |
| try: | |
| while not state.closed: | |
| seen = state.version | |
| await asyncio.sleep(SEQUENCE_STATUS_DELAY) | |
| if state.closed: | |
| return | |
| if state.version != seen: | |
| continue | |
| text, rows = panels.sequence_status(len(state.items), SEQUENCE_MAX) | |
| previous = state.status_id | |
| sent = await self.client.send_message(chat_id, text, buttons=panels.to_buttons(rows), parse_mode="html") | |
| if state.closed: # /sort or /cancel came while it was being sent | |
| await self._delete_quietly(chat_id, sent.id) | |
| return | |
| state.status_id = sent.id | |
| if previous is not None: | |
| await self._delete_quietly(chat_id, previous) | |
| if state.version == seen: | |
| return | |
| except asyncio.CancelledError: | |
| raise | |
| except Exception: # noqa: BLE001 | |
| log.exception("could not send the sequence status") | |
| async def _delete_quietly(self, chat_id: int, *ids: int) -> None: | |
| try: | |
| await self.client.delete_messages(chat_id, list(ids)) | |
| except Exception as exc: # noqa: BLE001 | |
| log.debug("could not delete messages %s: %s", ids, exc) | |
| async def cmd_sort(self, event: Any) -> None: | |
| await self._run_sort(event.sender_id, event.chat_id) | |
| async def _pause(seconds: float) -> None: | |
| await asyncio.sleep(seconds) | |
| async def _send_back(self, chat_id: int, msg: Any) -> None: | |
| """Send one collected file again (no re-upload), keeping its caption and formatting.""" | |
| for attempt in range(3): | |
| try: | |
| await self.client.send_file(chat_id, msg, caption=msg.raw_text or "", formatting_entities=msg.entities, parse_mode=None) | |
| return | |
| except errors.FloodWaitError as exc: | |
| if attempt == 2 or exc.seconds > SORT_MAX_WAIT: | |
| raise | |
| await self._pause(exc.seconds + 1) | |
| async def _look_up(self, chat_id: int, items: List[SeqItem]) -> List[Any]: | |
| """The messages behind ``items``, in that order (``None`` for one that is gone or could not be looked up).""" | |
| try: | |
| found = list(await self.client.get_messages(chat_id, ids=[i.msg_id for i in items])) | |
| except Exception: # noqa: BLE001 | |
| log.exception("could not look up %d files to sort", len(items)) | |
| found = [] | |
| return found + [None] * (len(items) - len(found)) | |
| async def _run_sort(self, uid: int, chat_id: int) -> None: | |
| say = lambda text, **kw: self.client.send_message(chat_id, text, parse_mode="html", link_preview=False, **kw) # noqa: E731 | |
| if uid in self._sorting: | |
| await say("⏳ I'm still sorting your last batch. /cancel to stop it.") | |
| return | |
| mode = self._mode(uid, "sequence") | |
| state: Optional[SequenceState] = mode.data if mode is not None else None | |
| if state is None or not state.items: | |
| await say("Nothing to sort yet. Send /sequence, then the files, then /sort.") | |
| return | |
| self._end_mode(uid) | |
| cancel = self._sorting[uid] = asyncio.Event() | |
| total = len(state.items) | |
| status = None | |
| try: | |
| if state.status_id is not None: | |
| await self._delete_quietly(chat_id, state.status_id) | |
| s = await self.db.get_settings(uid) | |
| plan = analyse(state.items, s["mode"]) | |
| status = await say(f"🔢 <b>Sorting {total} file{'s' if total != 1 else ''}…</b>") | |
| sent = 0 | |
| failed: List[SeqItem] = [] | |
| note = "" | |
| stopped = False | |
| found: List[Any] = [] # the messages of the files being sent, looked up SORT_LOOKUP at a time as the sort goes | |
| for index, item in enumerate(plan.ordered): | |
| if cancel.is_set(): | |
| stopped = True | |
| break | |
| if index % SORT_LOOKUP == 0: | |
| found = await self._look_up(chat_id, plan.ordered[index:index + SORT_LOOKUP]) | |
| msg = found[index % SORT_LOOKUP] | |
| if msg is None or getattr(msg, "file", None) is None: # deleted since it was sent | |
| failed.append(item) | |
| continue | |
| try: | |
| await self._send_back(chat_id, msg) | |
| except errors.FloodWaitError as exc: | |
| note = f"⚠️ Telegram asked me to wait {human_uptime(exc.seconds)} before sending more, so I stopped here." | |
| failed.extend(plan.ordered[index:]) | |
| break | |
| except Exception as exc: # noqa: BLE001 | |
| log.warning("could not send %s back: %s", item.name, exc) | |
| failed.append(item) | |
| continue | |
| sent += 1 | |
| if sent % SORT_PROGRESS_EVERY == 0 and sent < total: | |
| try: | |
| await self.client.edit_message(status, f"🔢 <b>Sorting…</b> {sent} of {total} sent", parse_mode="html") | |
| except Exception: # noqa: BLE001 | |
| pass | |
| gap = SORT_GAP if sent < SORT_FAST_FILES else SORT_SLOW_GAP | |
| if gap: | |
| await self._pause(gap) | |
| if status is not None: | |
| await self._delete_quietly(chat_id, status.id) | |
| if stopped: | |
| await say(f"⏹ Stopped after {sent} of {total} file{'s' if total != 1 else ''}.") | |
| return | |
| for chunk in format_report(plan, total, sent, failed, note): | |
| await say(chunk) | |
| finally: | |
| self._sorting.pop(uid, None) | |
| async def enqueue(self, event: Any, msg: Any, fmt: Optional[Dict[str, Any]] = None) -> None: | |
| uid = event.sender_id | |
| size = int(msg.file.size or 0) if msg.file is not None else 0 | |
| if size > self.cfg.max_file_bytes: | |
| await self._reply(event, f"❌ That file is {human_size(size)}; the limit is {human_size(self.cfg.max_file_bytes)}.") | |
| return | |
| s = await self.db.get_settings(uid) | |
| orig = original_name_of(msg) | |
| ext = source_ext(msg, orig) | |
| template = fmt["template"] if fmt else default_auto_format()["template"] | |
| new_name, info = compute_new_name( | |
| template=template, mode=s["mode"], original_name=orig, original_ext=ext, | |
| caption=msg.raw_text or "", remove_words=s["remove_words"], | |
| ) | |
| notes: List[str] = [] | |
| gaps = missing_variables(template, info) | |
| if gaps: | |
| where_from = "caption" if s["mode"] == "caption" else "file name" | |
| notes.append( | |
| "Couldn't find " + ", ".join("{%s}" % g for g in gaps) + f" in the {where_from}, so it is left empty. " | |
| "Cancel and change the template if you don't want that." | |
| ) | |
| job = Job( | |
| id=self.queue.new_id(), user_id=uid, chat_id=event.chat_id, msg_id=msg.id, | |
| file_name=orig, size=size, new_name=new_name, fmt=fmt, notes=notes, order=event.id, | |
| ) | |
| ahead = self.queue.running + self.queue.waiting | |
| where = "starting soon" if ahead == 0 else f"{ahead} ahead of it" | |
| used = f"\n🗂 {'Auto format' if fmt['kind'] == 'auto' else 'Format'}: <b>{esc(clip(fmt['name'], 30))}</b>" if fmt else "" | |
| job.status_msg = await event.reply( | |
| f"📥 <b>Queued</b> ({where}){used}\n<code>{esc(clip(new_name, 120))}</code>" + "".join(f"\n⚠️ {esc(n)}" for n in notes), | |
| buttons=panels.to_buttons([[("✖ Cancel", f"job:x:{job.id}")]]), | |
| parse_mode="html", | |
| ) | |
| self.queue.add(job) | |
| async def _on_job_cancelled(self, job: Job) -> None: | |
| if job.status_msg is not None: | |
| await self.client.edit_message( | |
| job.status_msg, f"❌ <b>Cancelled</b>\n<code>{esc(clip(job.new_name, 120))}</code>", buttons=None, parse_mode="html" | |
| ) | |
| async def cmd_queue(self, event: Any) -> None: | |
| uid = event.sender_id | |
| jobs = sorted(self.queue.jobs.values(), key=lambda j: j.id) if self.is_owner(uid) else self.queue.for_user(uid) | |
| if not jobs: | |
| await self._reply(event, "Nothing in the queue.") | |
| return | |
| lines = ["<b>Queue</b>"] | |
| for j in jobs[:20]: | |
| icon = "⚙️" if j.state == "running" else "⏳" | |
| who = f" · <code>{j.user_id}</code>" if self.is_owner(uid) else "" | |
| lines.append(f"{icon} #{j.id}{who} · <code>{esc(clip(j.new_name, 60))}</code>") | |
| if len(jobs) > 20: | |
| lines.append(f"… and {len(jobs) - 20} more") | |
| lines.append("\n/cancel — cancel yours · <code>/cancel 3</code> — cancel job #3") | |
| await self._reply(event, "\n".join(lines)) | |
| async def cmd_cancel(self, event: Any) -> None: | |
| uid = event.sender_id | |
| arg = self._args(event).lower() | |
| if not arg and self.pending.pop(uid, None) is not None: | |
| await self._reply(event, "Cancelled — nothing was changed.") | |
| return | |
| if not arg and uid in self._sorting: | |
| self._sorting[uid].set() | |
| await self._reply(event, "⏹ Stopping the sort…") | |
| return | |
| if not arg: | |
| mode = self._end_mode(uid) | |
| if mode is not None and mode.expires >= time.time(): | |
| if isinstance(mode.data, SequenceState) and mode.data.status_id is not None: | |
| await self._delete_quietly(event.chat_id, mode.data.status_id) # its buttons are of no use any more | |
| busy = " Send /cancel again to cancel your queued files." if self.queue.for_user(uid) else "" | |
| await self._reply(event, self._mode_cancelled_text(mode) + busy) | |
| return | |
| owner_scope = None if self.is_owner(uid) else uid | |
| if arg.isdigit(): | |
| ok = self.queue.cancel(int(arg), user_id=owner_scope) | |
| await self._reply(event, "🛑 Cancelling…" if ok else "I couldn't find an active job with that number.") | |
| return | |
| count = self.queue.cancel_all() if (arg == "all" and self.is_owner(uid)) else self.queue.cancel_user(uid) | |
| await self._reply(event, f"🛑 Cancelling {count} job(s)…" if count else "You have no active jobs.") | |
| def _mode_cancelled_text(mode: Mode) -> str: | |
| if mode.kind == "sequence": | |
| count = len(mode.data.items) | |
| return f"⏹ Sequence cancelled. {count} file{'s' if count != 1 else ''} discarded." if count else "⏹ Sequence cancelled." | |
| return { | |
| "rename": "⏹ Cancelled. Send /rename again to pick a format.", | |
| "smedia": "⏹ Cancelled. Send /smedia again to change the start media.", | |
| }.get(mode.kind, "Cancelled.") | |
| # ------------------------------------------------------------------ check | |
| async def cmd_check(self, event: Any) -> None: | |
| reply = await event.get_reply_message() | |
| if reply is None or not is_renameable(reply): | |
| await self._reply(event, "Reply to a file with /check to see the details I can read from its name and caption.") | |
| return | |
| s = await self.db.get_settings(event.sender_id) | |
| auto = await self.db.get_auto_format(event.sender_id) or default_auto_format() | |
| name = original_name_of(reply) | |
| caption = reply.raw_text or "" | |
| new_name, _info = compute_new_name( | |
| template=auto["template"], mode=s["mode"], original_name=name, original_ext=source_ext(reply, name), | |
| caption=caption, remove_words=s["remove_words"], | |
| ) | |
| text = panels.check_text(s["mode"], parse_info(split_ext(name)[0]), parse_caption(caption), new_name) | |
| await event.reply(text, parse_mode="html", link_preview=False) | |
| # ------------------------------------------------------------------ tracks | |
| async def _with_file_start( | |
| self, event: Any, msg: Any, tag: str, | |
| attempt: Callable[[Path], Awaitable[Any]], on_whole: Optional[Callable[[], Awaitable[None]]] = None, | |
| ) -> Any: | |
| """Run ``attempt(path)`` on the first part of the file; when it gives nothing (an index at the end of an mp4, say) on the whole file. | |
| The files live in a temporary folder that is removed afterwards. ``on_whole`` runs before the big download starts. | |
| """ | |
| name = original_name_of(msg) | |
| size = int(msg.file.size or 0) | |
| tmp_dir = Path(self.cfg.temp_dir) / f"{tag}{event.chat_id}_{event.id}" | |
| tmp_dir.mkdir(parents=True, exist_ok=True) | |
| try: | |
| path = tmp_dir / f"probe{source_ext(msg, name)}" | |
| await transfer.download(self.client, msg, path, fast=self.cfg.fast_transfer, connections=2, max_bytes=TRACKS_PROBE_BYTES) | |
| result = await attempt(path) | |
| if result is None and TRACKS_PROBE_BYTES < size <= TRACKS_FULL_LIMIT: | |
| if on_whole is not None: | |
| await on_whole() | |
| await transfer.download(self.client, msg, path, fast=self.cfg.fast_transfer, connections=self.cfg.connections) | |
| result = await attempt(path) | |
| return result | |
| finally: | |
| shutil.rmtree(tmp_dir, ignore_errors=True) | |
| async def cmd_tracks(self, event: Any) -> None: | |
| reply = await event.get_reply_message() | |
| if reply is None or not is_renameable(reply): | |
| await self._reply(event, "Reply to a video or audio file with /tracks to list its audio and subtitle tracks.") | |
| return | |
| status = await self._reply(event, "🔎 Reading the tracks…") | |
| async def read(path: Path) -> Optional[media.MediaInfo]: | |
| try: | |
| info = await media.probe(path) | |
| except media.MediaError: | |
| return None | |
| return info if info.streams else None | |
| async def whole() -> None: | |
| await self.client.edit_message(status, "🔎 Reading the tracks (this one needs the whole file)…", parse_mode="html") | |
| info = await self._with_file_start(event, reply, "tracks", read, whole) | |
| if info is None: | |
| await self.client.edit_message(status, "❌ I couldn't read any tracks from that file.", parse_mode="html") | |
| return | |
| text = f"<b>{esc(clip(original_name_of(reply), 80))}</b>\n\n{media.format_tracks(info)}" | |
| await self.client.edit_message(status, text[:3900], parse_mode="html") | |
| # --------------------------------------------------------------- /mediainfo | |
| async def _load_telegraph_token(self) -> Optional[str]: | |
| return await self.db.kv_get("telegraph_token") | |
| async def _save_telegraph_token(self, token: str) -> None: | |
| if token: | |
| await self.db.kv_set("telegraph_token", token) | |
| else: | |
| await self.db.kv_delete("telegraph_token") | |
| async def cmd_mediainfo(self, event: Any) -> None: | |
| target, own = await event.get_reply_message(), False | |
| if target is None or getattr(target, "document", None) is None: | |
| target, own = event.message, True # a file sent with /mediainfo as its caption | |
| if getattr(target, "document", None) is None: | |
| await self._reply(event, "Reply to a file or video with /mediainfo and I'll put everything MediaInfo can read from it on a Telegraph page.") | |
| return | |
| status = await self._reply(event, "🔎 Reading the file…") | |
| try: | |
| await self._mediainfo(event, target, own, status) | |
| except Exception: # noqa: BLE001 | |
| log.exception("/mediainfo failed") | |
| try: | |
| await self.client.edit_message(status, "⚠️ Something went wrong while reading that file.", parse_mode="html") | |
| except Exception: # noqa: BLE001 | |
| pass | |
| async def _mediainfo(self, event: Any, msg: Any, own: bool, status: Any) -> None: | |
| name = original_name_of(msg) | |
| size = int(msg.file.size or 0) | |
| playable = is_playable(msg, name) | |
| extra = mediainfo.telegram_section(name, size, **telegram_facts(msg, caption=not own)) | |
| last: List[mediainfo.Report] = [] | |
| async def read(path: Path) -> Optional[mediainfo.Report]: | |
| report = await mediainfo.read_file(path, name, size, extra=extra) | |
| last[:] = [report] | |
| return report if report.has_streams or not playable else None | |
| async def whole() -> None: | |
| await self.client.edit_message(status, "🔎 Reading the file (this one needs the whole file)…", parse_mode="html") | |
| try: | |
| report = await self._with_file_start(event, msg, "minfo", read, whole) or (last[0] if last else None) | |
| except Exception as exc: # noqa: BLE001 | |
| log.warning("could not read %s for /mediainfo: %s", name, exc) | |
| report = None | |
| if report is None: | |
| await self.client.edit_message(status, "❌ I couldn't read that file.", parse_mode="html") | |
| return | |
| if playable and not report.has_streams: | |
| report.notes.append( | |
| "No tracks could be read from this file, and it is too big for me to download completely." | |
| if size > TRACKS_FULL_LIMIT else "No tracks could be read from this file. It may be damaged." | |
| ) | |
| try: | |
| url = await self.telegraph.create_page(clip(f"MediaInfo – {name}", 256), mediainfo.to_nodes(report)) | |
| except Exception as exc: # noqa: BLE001 | |
| log.warning("Telegraph did not take the /mediainfo page: %s", exc) | |
| why = str(exc) if isinstance(exc, TelegraphError) else "Telegraph is not available" | |
| await self._mediainfo_file(event, status, name, report, why) | |
| return | |
| summary = report.summary() | |
| text = f"🔎 <b>MediaInfo</b> · {esc(clip(name, 80))}\n" + (f"<i>{esc(summary)}</i>\n" if summary else "") + f"\n{esc(url)}" | |
| await self.client.edit_message(status, text, buttons=[[Button.url("📄 Open MediaInfo", url)]], parse_mode="html", link_preview=False) | |
| async def _mediainfo_file(self, event: Any, status: Any, name: str, report: mediainfo.Report, why: str) -> None: | |
| """Telegraph did not work: send the same report as a text file instead.""" | |
| bio = io.BytesIO(mediainfo.render_text(report).encode("utf-8")) | |
| bio.name = re.sub(r'[\\/:*?"<>|]+', "_", clip(split_ext(name)[0], 60)) + " - MediaInfo.txt" | |
| caption = f"🔎 <b>MediaInfo</b> · {esc(clip(name, 80))}\n<i>{esc(why)}, so here it is as a text file.</i>" | |
| await self.client.send_file(event.chat_id, bio, caption=caption, parse_mode="html", force_document=True) | |
| await self._delete_quietly(event.chat_id, status.id) | |
| # ----------------------------------------------------------- /ethumb, /resize | |
| async def cmd_ethumb(self, event: Any) -> None: | |
| uid = event.sender_id | |
| self.pending.pop(uid, None) | |
| reply = await event.get_reply_message() | |
| if reply is not None and is_renameable(reply): # sent as a reply to the file: no need to ask for it | |
| await self._send_thumbnail(event, reply) | |
| return | |
| mode = self._start_mode(uid, "ethumb", ONE_FILE_WAIT) | |
| text, rows = panels.ethumb_prompt(ONE_FILE_WAIT // 60) | |
| await self._reply(event, text + (f"\n\n{mode.dropped}" if mode.dropped else ""), rows) | |
| async def _embedded_thumb(self, msg: Any, allow_stripped: bool = False) -> Optional[bytes]: | |
| return await embedded_thumb(self.client, msg, allow_stripped) | |
| async def _extract_thumbnail(self, event: Any, msg: Any) -> Optional[bytes]: | |
| data = await self._embedded_thumb(msg) | |
| if data: | |
| return data | |
| name = original_name_of(msg) | |
| playable = getattr(msg, "video", None) is not None or getattr(msg, "audio", None) is not None | |
| if playable or source_ext(msg, name).lower() in PROBE_EXTS: | |
| async def frame(path: Path) -> Optional[bytes]: | |
| out = path.with_suffix(".jpg") | |
| try: | |
| duration = (await media.probe(path)).duration | |
| except media.MediaError: | |
| duration = 0.0 # cut off before the index: ffmpeg may still find a picture | |
| ok = await media.extract_frame(path, out, media.frame_seeks(duration), width=ETHUMB_FRAME_WIDTH) | |
| return out.read_bytes() if ok else None | |
| data = await self._with_file_start(event, msg, "ethumb", frame) | |
| if data: | |
| return data | |
| return await self._embedded_thumb(msg, allow_stripped=True) | |
| async def _send_thumbnail(self, event: Any, msg: Any) -> None: | |
| status = await self._reply(event, "🖼 Extracting the thumbnail…") | |
| name = original_name_of(msg) | |
| try: | |
| data = await self._extract_thumbnail(event, msg) | |
| except Exception as exc: # noqa: BLE001 | |
| log.warning("thumbnail extraction failed for %s: %s", name, exc) | |
| data = None | |
| if not data: | |
| await self.client.edit_message(status, "❌ I couldn't find a thumbnail in that file or grab a frame from it.", parse_mode="html") | |
| return | |
| bio = io.BytesIO(data) | |
| bio.name = "thumbnail.jpg" | |
| await self.client.send_file(event.chat_id, bio, caption=f"🖼 Thumbnail of <b>{esc(clip(name, 80))}</b>", parse_mode="html") | |
| await self._delete_quietly(event.chat_id, status.id) | |
| async def _image_mode(self, event: Any) -> bool: | |
| """Give an incoming picture or GIF to /smedia when it is waiting for one. True when the message was taken care of.""" | |
| mode = self._mode(event.sender_id, "smedia") | |
| if mode is None: | |
| return False | |
| await self._take_start_media(event, mode) | |
| return True | |
| # ----------------------------------------------------------------- /size, /resize | |
| async def _picture_of(self, msg: Any) -> Tuple[Optional[bytes], str]: | |
| """``(the picture, "")`` of a message, or ``(None, why there is none)``. | |
| A photo, and a file that is a picture, are taken whole. Anything else (a video, a GIF, a sticker ...) gives the largest | |
| preview Telegram keeps with it. | |
| """ | |
| if getattr(msg, "photo", None) is not None or is_image(msg): | |
| size = int(msg.file.size or 0) if getattr(msg, "file", None) is not None else 0 | |
| if size > RESIZE_MAX_BYTES: | |
| return None, f"❌ That picture is {human_size(size)}; I can read pictures up to {human_size(RESIZE_MAX_BYTES)}." | |
| raw = await self.client.download_media(msg, file=bytes) | |
| elif getattr(msg, "document", None) is not None: | |
| raw = await self.client.download_media(msg, file=bytes, thumb=-1) | |
| else: | |
| return None, "❌ There's no picture in that." | |
| return (raw, "") if raw else (None, "❌ I couldn't get a picture from that.") | |
| async def _remove_later(self, chat_id: int, msg_id: int, seconds: float) -> None: | |
| await asyncio.sleep(seconds) | |
| await self._delete_quietly(chat_id, msg_id) | |
| async def cmd_size(self, event: Any) -> None: | |
| """Reply to a picture (or to anything with a preview) with /size: how many pixels wide and high it is.""" | |
| reply = await event.get_reply_message() | |
| if reply is None or not reply.media: | |
| await self._reply(event, "Reply to a picture or other media with /size to get its dimensions.") | |
| return | |
| status = await self._reply(event, "⏳ Processing…") | |
| raw, why = await self._picture_of(reply) | |
| if raw is None: | |
| await self.client.edit_message(status, why, parse_mode="html") | |
| return | |
| try: | |
| width, height = await asyncio.to_thread(media.image_size, raw) | |
| except Exception: # noqa: BLE001 | |
| await self.client.edit_message(status, "❌ I couldn't read that image.", parse_mode="html") | |
| return | |
| await self.client.edit_message(status, f"Dimension Of This Image Is\n<code>{width} x {height}</code>", parse_mode="html") | |
| async def cmd_resize(self, event: Any) -> None: | |
| """Reply to a picture with ``/resize 720 1080``: it comes back at exactly that many pixels wide and high (its shape is not kept).""" | |
| reply = await event.get_reply_message() | |
| if reply is None or not reply.media: | |
| await self._reply(event, "Reply to a picture or other media with /resize and a size, like <code>/resize 720 1080</code>.") | |
| return | |
| sides = self._args(event).split() | |
| if len(sides) != 2 or not all(re.fullmatch(r"[0-9]{1,6}", side) for side in sides): | |
| hint = await self._reply(event, "Give Some Size To Resize, Like <code>/resize 720 1080</code>") | |
| self._spawn(self._remove_later(event.chat_id, hint.id, HINT_SECONDS)) # it goes away by itself | |
| return | |
| width, height = int(sides[0]), int(sides[1]) | |
| if not (1 <= width <= RESIZE_MAX_SIDE and 1 <= height <= RESIZE_MAX_SIDE): | |
| await self._reply(event, f"❌ Each side can be 1 to {RESIZE_MAX_SIDE} pixels, like <code>/resize 720 1080</code>.") | |
| return | |
| status = await self._reply(event, "⏳ Processing…") | |
| raw, why = await self._picture_of(reply) | |
| if raw is None: | |
| await self.client.edit_message(status, why, parse_mode="html") | |
| return | |
| try: | |
| data = await asyncio.to_thread(media.resize_exact, raw, width, height) | |
| except media.ImageTooBig: | |
| await self.client.edit_message(status, "❌ That picture has too many pixels for me to resize.", parse_mode="html") | |
| return | |
| except Exception: # noqa: BLE001 | |
| await self.client.edit_message(status, "❌ I couldn't read that image.", parse_mode="html") | |
| return | |
| bio = io.BytesIO(data) | |
| bio.name = "resized.png" | |
| try: # as a photo when Telegram keeps it as it is, otherwise as a file, so the size stays exact | |
| await event.reply(file=bio, force_document=not fits_as_photo(width, height, len(data))) | |
| except Exception: # noqa: BLE001 | |
| log.exception("could not send the resized picture") | |
| await self.client.edit_message(status, "❌ Telegram wouldn't take the resized picture.", parse_mode="html") | |
| return | |
| await self._delete_quietly(event.chat_id, status.id) | |
| # ------------------------------------------------------- /smedia, the /start picture | |
| async def _start_media_meta(self) -> Optional[Dict[str, Any]]: | |
| """What /start shows above the menu: ``{"kind": "image" | "gif", "file_id": ..., ...}``, or None.""" | |
| raw = await self.db.kv_get("start_media") | |
| if not raw: | |
| return None | |
| try: | |
| meta = json.loads(raw) | |
| except ValueError: | |
| return None | |
| return meta if isinstance(meta, dict) and meta.get("kind") in ("image", "gif") else None | |
| async def _store_start_media(self, meta: str, data: str) -> None: | |
| """Keep the /start media (its description and a base64 copy of the file); empty strings forget it.""" | |
| for key, value in (("start_media", meta), ("start_media_data", data)): | |
| if value: | |
| await self.db.kv_set(key, value) | |
| else: | |
| await self.db.kv_delete(key) | |
| async def _send_start_media(self, chat_id: int, text: str, rows: List[panels.Row]) -> bool: | |
| """Send the menu with the owner's picture or GIF above it. False when there is none or it can't be sent.""" | |
| meta = await self._start_media_meta() | |
| if meta is None: | |
| return False | |
| kw: Dict[str, Any] = dict(caption=text, buttons=panels.to_buttons(rows), parse_mode="html") | |
| sent = None | |
| if meta.get("file_id"): | |
| try: | |
| sent = await self.client.send_file(chat_id, meta["file_id"], **kw) | |
| except Exception as exc: # noqa: BLE001 - a file reference that went stale: the stored copy is used instead | |
| log.warning("the start media could not be sent by its file id (%s); sending the stored copy", exc) | |
| if sent is None: | |
| try: | |
| data = base64.b64decode(await self.db.kv_get("start_media_data") or "", validate=True) | |
| if not data: | |
| return False | |
| sent = await self._upload_start_media(chat_id, meta, data, **kw) | |
| except Exception as exc: # noqa: BLE001 | |
| log.warning("could not send the start media (%s); /start shows the menu only", exc) | |
| return False | |
| fresh = getattr(getattr(sent, "file", None), "id", None) | |
| if fresh and fresh != meta.get("file_id"): # the next /start can reuse what was just uploaded | |
| await self.db.kv_set("start_media", json.dumps({**meta, "file_id": fresh})) | |
| return True | |
| async def _upload_start_media(self, chat_id: int, meta: Dict[str, Any], data: bytes, **kw: Any) -> Any: | |
| bio = io.BytesIO(data) | |
| if meta["kind"] == "gif": | |
| bio.name = "start.mp4" | |
| kw.update( | |
| mime_type=meta.get("mime") or "video/mp4", | |
| attributes=[ | |
| types.DocumentAttributeVideo( | |
| duration=float(meta.get("duration") or 0), w=int(meta.get("width") or 0), h=int(meta.get("height") or 0), | |
| supports_streaming=True, | |
| ), | |
| types.DocumentAttributeAnimated(), | |
| ], | |
| ) | |
| else: | |
| bio.name = "start.jpg" | |
| return await self.client.send_file(chat_id, bio, **kw) | |
| async def cmd_smedia(self, event: Any) -> None: | |
| self.pending.pop(event.sender_id, None) | |
| meta = await self._start_media_meta() | |
| text, rows = panels.smedia_menu(meta["kind"] if meta else "") | |
| await self._reply(event, text, rows) | |
| async def _cb_smedia(self, event: Any, uid: int, parts: List[str]) -> None: | |
| if not self.is_owner(uid): | |
| await event.answer("Owner only.", alert=True) | |
| return | |
| self.pending.pop(uid, None) | |
| action = parts[1] if len(parts) > 1 else "" | |
| if action in ("image", "gif"): | |
| mode = self._start_mode(uid, "smedia", ONE_FILE_WAIT, {"kind": action}) | |
| text, rows = panels.smedia_prompt(action, ONE_FILE_WAIT // 60) | |
| await self._edit_panel(event, text + (f"\n\n{mode.dropped}" if mode.dropped else ""), rows) | |
| await event.answer() | |
| elif action == "menu": | |
| if self._mode(uid, "smedia") is not None: | |
| self._end_mode(uid) | |
| meta = await self._start_media_meta() | |
| text, rows = panels.smedia_menu(meta["kind"] if meta else "") | |
| await self._edit_panel(event, text, rows) | |
| await event.answer() | |
| elif action == "rm": | |
| await self._store_start_media("", "") | |
| text, rows = panels.smedia_menu("") | |
| await self._edit_panel(event, text, rows) | |
| await event.answer("Start media removed.") | |
| elif action == "stop": | |
| if self._mode(uid, "smedia") is not None: | |
| self._end_mode(uid) | |
| await self._edit_panel(event, "⏹ Stopped. Send /smedia again to change the start media.", [[("✖ Close", "x:close")]]) | |
| await event.answer() | |
| else: | |
| await event.answer() | |
| async def _take_start_media(self, event: Any, mode: Mode) -> None: | |
| """The owner sent the picture or GIF /smedia asked for: keep it, then show /start with it.""" | |
| uid, msg, want = event.sender_id, event.message, mode.data["kind"] | |
| is_gif = getattr(msg, "gif", None) is not None | |
| if want == "gif" and not is_gif: | |
| await self._reply(event, "❌ That isn't a GIF. Send a GIF (an animation), or /cancel to stop.") | |
| return | |
| if want == "image" and (is_gif or not is_image(msg)): | |
| await self._reply(event, "❌ That isn't an image. Send a photo or an image file, or /cancel to stop.") | |
| return | |
| size = int(msg.file.size or 0) if getattr(msg, "file", None) is not None else 0 | |
| if size > START_MEDIA_MAX_BYTES: | |
| await self._reply( | |
| event, f"❌ That's {human_size(size)}; the biggest I can keep is {human_size(START_MEDIA_MAX_BYTES)}. Send a smaller one, or /cancel to stop." | |
| ) | |
| return | |
| self._consume(uid, "smedia") | |
| raw = await self.client.download_media(msg, file=bytes) | |
| if not raw: | |
| await self._reply(event, "❌ I couldn't download that. Send /smedia to try again.") | |
| return | |
| try: | |
| if want == "gif": | |
| data, meta = raw, { | |
| "kind": "gif", "mime": msg.file.mime_type or "video/mp4", "width": int(msg.file.width or 0), | |
| "height": int(msg.file.height or 0), "duration": float(msg.file.duration or 0), | |
| } | |
| elif getattr(msg, "photo", None) is not None: # Telegram already made it a JPEG that fits | |
| data, meta = raw, {"kind": "image"} | |
| else: | |
| data, meta = await asyncio.to_thread(media.fit_photo, raw, START_IMAGE_BOX), {"kind": "image"} | |
| except Exception: # noqa: BLE001 | |
| await self._reply(event, "❌ I couldn't read that image. Send /smedia to try again.") | |
| return | |
| if len(data) > START_MEDIA_MAX_BYTES: | |
| await self._reply(event, f"❌ That's {human_size(len(data))}; the biggest I can keep is {human_size(START_MEDIA_MAX_BYTES)}. Send /smedia to try another.") | |
| return | |
| before = (await self.db.kv_get("start_media") or "", await self.db.kv_get("start_media_data") or "") | |
| await self._store_start_media(json.dumps({**meta, "file_id": ""}), base64.b64encode(data).decode("ascii")) | |
| text, rows = await self.panel(uid, "home") | |
| if not await self._send_start_media(event.chat_id, text, rows): # the preview is the real /start: if it fails, so would /start | |
| await self._store_start_media(*before) | |
| await self._reply(event, "❌ Telegram wouldn't take that as the start media, so nothing was changed. Send /smedia to try another.") | |
| return | |
| await self._reply(event, f"✅ Done: /start shows this {'image' if want == 'image' else 'GIF'} from now on.") | |
| # ------------------------------------------------------------------- speed | |
| async def cmd_speed(self, event: Any) -> None: | |
| if self._speed_lock.locked(): | |
| await self._reply(event, "⏳ A speed test is already running. Try again in a minute.") | |
| return | |
| async with self._speed_lock: | |
| busy = self.queue.running > 0 | |
| status = await self._reply(event, "⚡ Running the speed test… this takes about half a minute.") | |
| try: | |
| text = speed.format_result(await speed.run_speedtest()) | |
| if busy: | |
| text += "\nℹ️ A rename job was running during the test, so the numbers may be lower than usual." | |
| except speed.SpeedError as exc: | |
| text = f"❌ I couldn't run the speed test: {esc(exc)}" | |
| except Exception: # noqa: BLE001 | |
| log.exception("speed test failed") | |
| text = "❌ The speed test failed unexpectedly." | |
| await self.client.edit_message(status, text, parse_mode="html", link_preview=False) | |
| # ----------------------------------------------------------- status/stats | |
| async def cmd_status(self, event: Any) -> None: | |
| ping_ms: Optional[float] = None | |
| try: | |
| t0 = time.perf_counter() | |
| await self.client(functions.PingRequest(ping_id=random.getrandbits(62))) | |
| ping_ms = (time.perf_counter() - t0) * 1000 | |
| except Exception as exc: # noqa: BLE001 | |
| log.debug("ping failed: %s", exc) | |
| await self._reply( | |
| event, sysinfo.format_status(ping_ms, getattr(self.client.session, "dc_id", None), telethon.__version__, self._speed_text()) | |
| ) | |
| def _speed_text(self) -> str: | |
| return sysinfo.format_speed(self.cfg.fast_transfer, self.cfg.connections, transfer.stats, self.cfg.stream_transfer) | |
| async def cmd_stats(self, event: Any) -> None: | |
| await self._reply(event, await self._stats_text()) | |
| async def _stats_text(self) -> str: | |
| users = await self.db.count_users() | |
| files, total = await self.db.totals() | |
| return await sysinfo.format_stats(users, files, total, self.queue.running, self.queue.waiting, self.cfg.temp_dir) | |
| # ------------------------------------------------------------------ admins | |
| async def cmd_addadmin(self, event: Any) -> None: | |
| arg = self._args(event) | |
| if not re.fullmatch(r"\d{4,15}", arg): | |
| await self._reply(event, "Usage: <code>/addadmin 123456789</code> (a numeric user id; they can find theirs with @userinfobot).") | |
| return | |
| uid = int(arg) | |
| if uid == self.cfg.owner_id or uid in self.admins: | |
| await self._reply(event, "They already have access.") | |
| return | |
| await self.db.add_admin(uid, event.sender_id) | |
| self.admins.add(uid) | |
| await self._reply(event, f"✅ <code>{uid}</code> can now use the bot.") | |
| async def cmd_deladmin(self, event: Any) -> None: | |
| arg = self._args(event) | |
| if not re.fullmatch(r"\d{4,15}", arg): | |
| await self._reply(event, "Usage: <code>/deladmin 123456789</code>") | |
| return | |
| uid = int(arg) | |
| removed = await self.db.del_admin(uid) | |
| if uid in self.cfg.admin_ids: | |
| await self._reply(event, "That id is listed in the ADMIN_IDS setting; remove it there.") | |
| return | |
| self.admins.discard(uid) | |
| await self._reply(event, f"✅ <code>{uid}</code> was removed." if removed else "That id wasn't in the list.") | |
| async def cmd_admins(self, event: Any) -> None: | |
| ids = ", ".join(f"<code>{a}</code>" for a in sorted(self.admins)) or "none" | |
| await self._reply(event, f"<b>Owner:</b> <code>{self.cfg.owner_id}</code>\n<b>Allowed users:</b> {ids}") | |
| # --------------------------------------------------------------- text input | |
| async def on_text(self, event: Any) -> None: | |
| uid = event.sender_id | |
| p = self.pending.get(uid) | |
| if p is None or p.expires < time.time(): | |
| self.pending.pop(uid, None) | |
| return | |
| if p.kind.startswith("fmt:"): | |
| await self._on_format_text(event, p) | |
| return | |
| ok, msg = await apply_setting(self.db, uid, p.kind, event.raw_text) | |
| if not ok: | |
| await self._reply(event, msg + "\n\nTry again, or tap Cancel.") | |
| return | |
| self.pending.pop(uid, None) | |
| try: | |
| await event.delete() | |
| except Exception: # noqa: BLE001 | |
| pass | |
| text, rows = await self.panel(uid, p.back) | |
| limit = 1000 if p.is_photo else 3900 | |
| full = f"{msg}\n\n{text}" if visible_len(msg) + visible_len(text) < limit else f"{msg}\n\n(Open the menu to see the update.)" | |
| try: | |
| await self.client.edit_message(p.chat_id, p.msg_id, full, buttons=panels.to_buttons(rows), parse_mode="html", link_preview=False) | |
| except Exception: # noqa: BLE001 | |
| await self._reply(event, full, rows) | |
| # ---------------------------------------------------------------- callbacks | |
| async def on_callback(self, event: Any) -> None: | |
| uid = event.sender_id | |
| data = event.data.decode("utf-8", "ignore") | |
| parts = data.split(":") | |
| head = parts[0] | |
| if data == "x:close": | |
| self.pending.pop(uid, None) | |
| await event.answer() | |
| await self._remove_message(event) | |
| return | |
| if head == "m" and len(parts) == 2 and parts[1] in PANEL_NAMES: | |
| self.pending.pop(uid, None) | |
| if parts[1] == "admin" and not self.is_owner(uid): | |
| await event.answer("Owner only.", alert=True) | |
| return | |
| text, rows = await self.panel(uid, parts[1]) | |
| await self._edit_panel(event, text, rows) | |
| await event.answer() | |
| return | |
| if head == "set" and len(parts) == 3: | |
| await self._cb_set(event, uid, parts[1], parts[2]) | |
| return | |
| if head == "fm": | |
| await self._cb_format(event, uid, parts) | |
| return | |
| if head == "fs": | |
| await self._cb_shared(event, uid, parts) | |
| return | |
| if head == "au": | |
| await self._cb_auto(event, uid, parts) | |
| return | |
| if head == "rn": | |
| await self._cb_rename(event, uid, parts) | |
| return | |
| if head == DBFLOW_PREFIX: | |
| if self.dbflow is None: | |
| await event.answer("Uploading to the Database Channel is switched off (DB_CHANNEL and COMMUNICATION_CHAT).", alert=True) | |
| else: | |
| await self.dbflow.on_callback(event, parts) | |
| return | |
| if head == "sq": | |
| await self._cb_sequence(event, uid, parts) | |
| return | |
| if head == "sm": | |
| await self._cb_smedia(event, uid, parts) | |
| return | |
| if data == "et:stop": | |
| if self._mode(uid, "ethumb") is not None: | |
| self._end_mode(uid) | |
| await self._edit_panel(event, "⏹ Stopped. Send /ethumb again to extract a thumbnail.", [[("✖ Close", "x:close")]]) | |
| await event.answer() | |
| return | |
| if head == "in": | |
| await self._cb_input(event, uid, parts) | |
| return | |
| if head == "clr" and len(parts) == 2 and parts[1] in ("keep_audio", "keep_subtitle"): | |
| _ok, msg = await apply_setting(self.db, uid, parts[1], "off") | |
| text, rows = await self.panel(uid, "tool") | |
| await self._edit_panel(event, text, rows) | |
| await event.answer(clip(plain(msg), 190)) | |
| return | |
| if data == "help:meta": | |
| text, rows = panels.metadata_help() | |
| await self._edit_panel(event, text, rows) | |
| await event.answer() | |
| return | |
| if head == "adm" and len(parts) == 2: | |
| if not self.is_owner(uid): | |
| await event.answer("Owner only.", alert=True) | |
| return | |
| if parts[1] == "stats": | |
| await event.answer() | |
| await self.client.send_message(event.chat_id, await self._stats_text(), parse_mode="html") | |
| elif parts[1] == "status": | |
| await event.answer() | |
| await self.client.send_message( | |
| event.chat_id, | |
| sysinfo.format_status(None, getattr(self.client.session, "dc_id", None), telethon.__version__, self._speed_text()), | |
| parse_mode="html", | |
| ) | |
| elif parts[1] == "clear": | |
| count = self.queue.cancel_all() | |
| await event.answer(f"Cancelling {count} job(s).", alert=True) | |
| return | |
| if head == "job" and len(parts) == 3 and parts[1] == "x" and parts[2].isdigit(): | |
| ok = self.queue.cancel(int(parts[2]), user_id=None if self.is_owner(uid) else uid) | |
| await event.answer("Cancelling…" if ok else "That job already finished.") | |
| return | |
| await event.answer() | |
| async def _cb_set(self, event: Any, uid: int, field: str, value: str) -> None: | |
| if field == "mode" and value in ("filename", "caption"): | |
| await self.db.update_settings(uid, mode=value) | |
| panel, toast = "rename", f"Mode: {value.title()}" | |
| elif field == "ftype" and value in ("same", "document", "video", "audio"): | |
| await self.db.update_settings(uid, file_type=value) | |
| panel, toast = "ftype", "File type: " + ("Same as original" if value == "same" else value.title()) | |
| elif field == "meta" and value in ("on", "off"): | |
| await self.db.update_settings(uid, metadata_on=(value == "on")) | |
| panel, toast = "meta", f"Metadata {value.upper()}" | |
| else: | |
| await event.answer() | |
| return | |
| text, rows = await self.panel(uid, panel) | |
| await self._edit_panel(event, text, rows) | |
| await event.answer(toast) | |
| async def _cb_input(self, event: Any, uid: int, parts: List[str]) -> None: | |
| if len(parts) == 3 and parts[1] == "cancel": | |
| self.pending.pop(uid, None) | |
| back = parts[2] if parts[2] in PANEL_NAMES else "home" | |
| text, rows = await self.panel(uid, back) | |
| await self._edit_panel(event, text, rows) | |
| await event.answer("Cancelled.") | |
| return | |
| kind = parts[1] if len(parts) == 2 else "" | |
| if kind not in panels.PROMPTS: | |
| await event.answer() | |
| return | |
| msg_obj = await event.get_message() | |
| self.pending[uid] = Pending( | |
| kind=kind, chat_id=event.chat_id, msg_id=event.message_id, back=BACK[kind], | |
| is_photo=bool(msg_obj is not None and (msg_obj.photo is not None or getattr(msg_obj, "media", None) is not None)), | |
| expires=time.time() + self.PENDING_TTL, | |
| ) | |
| text, rows = panels.input_prompt(kind, BACK[kind]) | |
| await self._edit_panel(event, text, rows) | |
| await event.answer() | |
| # ------------------------------------------------------------ saved formats | |
| async def _format_panel(self, uid: int, fmt: Dict[str, Any], share_inline: Optional[bool] = None) -> panels.Panel: | |
| """A format's screen: for an auto format it says whether it is the one in use. ``share_inline`` defaults to whether | |
| inline mode is known to be on.""" | |
| active = False | |
| if fmt["kind"] == "auto": | |
| current = await self.db.get_auto_format(uid) | |
| active = current is not None and current["id"] == fmt["id"] | |
| if share_inline is None: | |
| share_inline = await self._check_inline() | |
| return panels.format_view(fmt, share_inline=share_inline, active=active, upload_on=self.cfg.upload_enabled) | |
| async def _show_format(self, event: Any, uid: int, fid: int, note: str = "") -> bool: | |
| fmt = await self.db.get_format(uid, fid) | |
| if fmt is None: | |
| return False | |
| text, rows = await self._format_panel(uid, fmt) | |
| await self._edit_panel(event, f"{note}\n\n{text}" if note else text, rows) | |
| return True | |
| async def _show_format_list(self, event: Any, uid: int, kind: str) -> None: | |
| formats = await self.db.list_formats(uid, kind) | |
| active = await self.db.get_auto_format(uid) if kind == "auto" else None | |
| text, rows = panels.formats_menu(formats, MAX_FORMATS, kind, active["id"] if active else 0) | |
| await self._edit_panel(event, text, rows) | |
| async def _cb_auto(self, event: Any, uid: int, parts: List[str]) -> None: | |
| """A button of the /start menu's Auto Format screen: use that auto format from now on.""" | |
| self.pending.pop(uid, None) | |
| fmt = await self.db.select_auto_format(uid, int(parts[1])) if len(parts) == 2 and parts[1].isdigit() else None | |
| text, rows = await self.panel(uid, "auto") | |
| await self._edit_panel(event, text, rows) | |
| if fmt is None: | |
| await event.answer("That format no longer exists.", alert=True) | |
| else: | |
| await event.answer(f"Auto format: {clip(fmt['name'], 40)}") | |
| async def _cb_format(self, event: Any, uid: int, parts: List[str]) -> None: | |
| self.pending.pop(uid, None) # leaving a prompt open by tapping something else abandons it | |
| action = parts[1] if len(parts) > 1 else "" | |
| fid = int(parts[2]) if len(parts) > 2 and parts[2].isdigit() else 0 | |
| if action in ("l", "new"): | |
| kind = parts[2] if len(parts) > 2 else "" | |
| if kind not in ("auto", "manual"): # a button of a message from before there were two kinds: start from the choice | |
| text, rows = await self.panel(uid, "formats") | |
| await self._edit_panel(event, text, rows) | |
| await event.answer() | |
| return | |
| if action == "l": | |
| await self._show_format_list(event, uid, kind) | |
| await event.answer() | |
| elif await self.db.count_formats(uid, kind) >= MAX_FORMATS: | |
| await event.answer(f"You already have {MAX_FORMATS} {kind} formats. Delete one first.", alert=True) | |
| else: | |
| await self._format_prompt(event, uid, "new", 0, kind) | |
| return | |
| fmt = await self.db.get_format(uid, fid) if fid else None | |
| if fmt is None: | |
| text, rows = await self.panel(uid, "formats") | |
| await self._edit_panel(event, text, rows) | |
| await event.answer("That format no longer exists.", alert=True) | |
| return | |
| if action in ("v", "cx"): | |
| await self._show_format(event, uid, fid) | |
| await event.answer() | |
| elif action == "in" and len(parts) == 4 and parts[3] == "channel": | |
| if fmt["kind"] != "manual" or not self.cfg.upload_enabled: | |
| await event.answer("Channel is an option of manual formats, while uploading to the Database Channel is set up.", alert=True) | |
| return | |
| await self._format_prompt(event, uid, "channel", fid) | |
| elif action == "chc": | |
| await self.db.update_format(uid, fid, channel=None) | |
| await self._show_format(event, uid, fid) | |
| await event.answer("Channel cleared.") | |
| elif action == "in" and len(parts) == 4 and parts[3] in (*FIELD_KINDS, "name"): | |
| await self._format_prompt(event, uid, parts[3], fid) | |
| elif action == "th": | |
| await self._format_prompt(event, uid, "thumb", fid) | |
| elif action == "thd": | |
| await self.db.set_format_thumb(uid, fid, None) | |
| await self._show_format(event, uid, fid) | |
| await event.answer("Thumbnail deleted.") | |
| elif action == "cr": | |
| await apply_setting(FormatSettings(self.db, fid), uid, "caption", "reset") | |
| await self._show_format(event, uid, fid) | |
| await event.answer("Caption reset.") | |
| elif action == "use": | |
| if await self.db.select_auto_format(uid, fid) is None: | |
| await event.answer("Only an auto format can be used for auto rename.", alert=True) | |
| return | |
| await self._show_format(event, uid, fid, note=f"✅ <b>{esc(clip(fmt['name'], 30))}</b> now renames your files.") | |
| await event.answer(f"Auto format: {clip(fmt['name'], 40)}") | |
| elif action == "dm": | |
| settings = await self.db.get_settings(uid) | |
| if not has_default_metadata(settings): | |
| await event.answer("Your default metadata is empty. Set it in /metadata first, then apply it.", alert=True) | |
| return | |
| await self.db.update_format(uid, fid, **default_metadata_changes(settings)) | |
| await self._show_format(event, uid, fid, note=panels.format_metadata_applied(fmt)) | |
| await event.answer("Default metadata applied.") | |
| elif action == "cp": | |
| target = other_kind(fmt["kind"]) | |
| try: | |
| made = await self.db.copy_format(uid, uid, fid, name_max=NAME_MAX, kind=target) | |
| except FormatError: # only "limit" can be raised here | |
| await event.answer(f"You already have {MAX_FORMATS} {target} formats. Delete one first.", alert=True) | |
| return | |
| if made is None: | |
| await event.answer("That format no longer exists.", alert=True) | |
| return | |
| text, rows = panels.format_copied(fmt, made) | |
| await self._edit_panel(event, text, rows) | |
| await event.answer(f"Copied to your {target} formats.") | |
| elif action == "sh": # Share format, shown while inline mode wasn't known to be on: just try the real button | |
| text, rows = await self._format_panel(uid, fmt, share_inline=True) | |
| try: | |
| await event.edit(text, buttons=panels.to_buttons(rows), parse_mode="html", link_preview=False) | |
| except errors.MessageNotModifiedError: | |
| pass | |
| except errors.BotInlineDisabledError: # Telegram refuses it: inline mode is off | |
| self._inline_refused() | |
| await event.answer(panels.INLINE_OFF, alert=True) | |
| return | |
| self._inline_on = True | |
| await event.answer("Inline mode is on. Tap Share format again.") | |
| elif action == "del": | |
| text, rows = panels.format_delete_confirm(fmt) | |
| await self._edit_panel(event, text, rows) | |
| await event.answer() | |
| elif action == "ok": | |
| await self.db.delete_format(uid, fid) | |
| await self._show_format_list(event, uid, fmt["kind"]) | |
| await event.answer(f"Deleted {clip(fmt['name'], 40)}.") | |
| else: | |
| await event.answer() | |
| async def _format_prompt(self, event: Any, uid: int, field: str, fid: int, kind: str = "") -> None: | |
| msg_obj = await event.get_message() | |
| self.pending[uid] = Pending( | |
| kind=f"fmt:{field}", chat_id=event.chat_id, msg_id=event.message_id, back="formats", | |
| is_photo=bool(msg_obj is not None and (msg_obj.photo is not None or getattr(msg_obj, "media", None) is not None)), | |
| expires=time.time() + self.PENDING_TTL, fid=fid, fmt_kind=kind, | |
| ) | |
| text, rows = panels.format_prompt(field, fid, kind) | |
| await self._edit_panel(event, text, rows) | |
| await event.answer() | |
| async def _on_format_text(self, event: Any, p: Pending) -> None: | |
| uid = event.sender_id | |
| field = p.kind[4:] | |
| text = self._caption_input(event) if field == "caption" else event.raw_text or "" | |
| if field == "thumb": | |
| await self._reply(event, "Send a photo for the thumbnail, or tap Cancel.") | |
| return | |
| try: | |
| if field in ("new", "name"): | |
| ok, value = validate_name(text) | |
| if ok and field == "new": | |
| fmt = await self.db.create_format(uid, value, kind=p.fmt_kind or "manual") | |
| msg = ( | |
| f"✅ {KIND_LABELS[fmt['kind']]} format <b>{esc(value)}</b> created. Set its template, caption, " | |
| "thumbnail and metadata, or tap Apply default metadata." | |
| ) | |
| p.fid = fmt["id"] | |
| elif ok: | |
| if await self.db.update_format(uid, p.fid, name=value) is None: | |
| raise LookupError | |
| msg = f"✅ Renamed to <b>{esc(value)}</b>." | |
| else: | |
| msg = value | |
| elif field == "channel": | |
| ok, channel, msg = parse_format_channel(text) | |
| if ok: | |
| if await self.db.update_format(uid, p.fid, channel=channel) is None: | |
| raise LookupError | |
| msg = f"✅ Channel saved: <code>{channel}</code>" if channel else "✅ Channel cleared." | |
| else: | |
| ok, msg = await apply_setting(FormatSettings(self.db, p.fid), uid, FIELD_KINDS[field], text) | |
| except FormatError as exc: | |
| ok = False | |
| if exc.code == "limit": | |
| self.pending.pop(uid, None) | |
| await self._reply(event, f"❌ You already have {MAX_FORMATS} {p.fmt_kind or 'manual'} formats. Delete one first.") | |
| return | |
| msg = "❌ You already have a format with that name." | |
| except LookupError: | |
| self.pending.pop(uid, None) | |
| await self._reply(event, "⚠️ That format no longer exists.") | |
| return | |
| if not ok: | |
| await self._reply(event, msg + "\n\nTry again, or tap Cancel.") | |
| return | |
| self.pending.pop(uid, None) | |
| try: | |
| await event.delete() | |
| except Exception: # noqa: BLE001 | |
| pass | |
| await self._refresh_format_panel(event, p, msg) | |
| async def _format_thumb_saved(self, event: Any, p: Pending, thumb: bytes) -> None: | |
| uid = event.sender_id | |
| self.pending.pop(uid, None) | |
| if not await self.db.set_format_thumb(uid, p.fid, thumb): | |
| await self._reply(event, "⚠️ That format no longer exists.") | |
| return | |
| await self._refresh_format_panel(event, p, "✅ Thumbnail saved for this format.") | |
| async def _refresh_format_panel(self, event: Any, p: Pending, msg: str) -> None: | |
| """Put the format's panel back into the message the prompt was shown in (a new message if that fails).""" | |
| fmt = await self.db.get_format(event.sender_id, p.fid) | |
| if fmt is None: | |
| await self._reply(event, msg) | |
| return | |
| text, rows = await self._format_panel(event.sender_id, fmt) | |
| limit = 1000 if p.is_photo else 3900 | |
| full = f"{msg}\n\n{text}" if visible_len(msg) + visible_len(text) < limit else f"{msg}\n\n(Open the format to see the update.)" | |
| try: | |
| await self.client.edit_message(p.chat_id, p.msg_id, full, buttons=panels.to_buttons(rows), parse_mode="html", link_preview=False) | |
| except Exception: # noqa: BLE001 | |
| await self._reply(event, full, rows) | |
| # ------------------------------------------------------------ sharing a format | |
| async def _check_inline(self, *, force: bool = False) -> bool: | |
| """Is inline mode on for the bot (@BotFather ▸ Bot Settings ▸ Inline Mode)? | |
| The Share format button needs it: while it is off, Telegram refuses a message that carries that button. Once it has | |
| been seen on it stays on (until Telegram refuses a button); while it is off the bot looks again after | |
| ``INLINE_RECHECK`` seconds, or at once with ``force``. | |
| """ | |
| now = time.monotonic() | |
| if not force: | |
| if self._inline_on: | |
| return True | |
| if self._inline_checked is not None and now - self._inline_checked < INLINE_RECHECK: | |
| return False | |
| self._inline_checked = now | |
| try: | |
| me = await self.client.get_me() | |
| except Exception as exc: # noqa: BLE001 | |
| log.warning("could not read the bot's profile to see whether inline mode is on: %s", exc) | |
| return self._inline_on | |
| self._username = getattr(me, "username", None) or self._username | |
| # the placeholder is only there while inline mode is on ("" = on, without a placeholder text) | |
| self._inline_on = self._inline_on or getattr(me, "bot_inline_placeholder", None) is not None | |
| return self._inline_on | |
| def _inline_refused(self) -> None: | |
| """Telegram refused a Share button, so inline mode is off: the next look at the bot's profile is a minute away.""" | |
| self._inline_on = False | |
| self._inline_checked = time.monotonic() | |
| async def _bot_username(self) -> Optional[str]: | |
| if self._username is None: | |
| await self._check_inline(force=True) | |
| return self._username | |
| async def on_inline(self, event: Any) -> None: | |
| """Where the chat picker of a Share format button ends up: the owner of the format gets a card to send to that chat.""" | |
| results: List[Any] = [] | |
| match = SHARE_QUERY.fullmatch((event.text or "").strip()) | |
| if match: | |
| try: | |
| results = await self._share_cards(event.sender_id, int(match.group(1)), event.builder) | |
| except Exception: # noqa: BLE001 | |
| log.exception("Error making the card of a shared format:") | |
| await self._answer_inline(event, results) | |
| async def _share_cards(self, uid: int, fid: int, builder: Any) -> List[Any]: | |
| """The card for one of the user's formats: its name, and a button that opens it in this bot (``/start fmt_<token>``).""" | |
| fmt = await self.db.get_format(uid, fid) | |
| token = await self.db.share_token(uid, fid) if fmt is not None else None | |
| username = await self._bot_username() if token else None | |
| if fmt is None or token is None or not username: | |
| return [] | |
| title, description, text = panels.share_card(fmt) | |
| link = f"https://t.me/{username}?start={SHARE_PREFIX}{token}" | |
| card = await builder.article( | |
| title, description=description, text=text, parse_mode="html", link_preview=False, | |
| buttons=[[Button.url("👁 View format", link)]], | |
| ) | |
| return [card] | |
| async def _answer_inline(event: Any, results: List[Any]) -> None: | |
| try: | |
| await event.answer(results, cache_time=0, private=True) # not cached: the card has to show the format as it is now | |
| except Exception as exc: # noqa: BLE001 - e.g. the query is too old by now | |
| log.debug("could not answer an inline query: %s", exc) | |
| async def _open_shared(self, event: Any, token: str) -> None: | |
| """``/start fmt_<token>``: what is in a format somebody shared, with a button to add it to your own formats.""" | |
| shared = await self.db.get_shared_format(token) if SHARE_TOKEN.fullmatch(token) else None | |
| if shared is None: | |
| await self._reply(event, panels.SHARE_GONE, [[("✖ Close", "x:close")]]) | |
| return | |
| owner_id, fmt = shared | |
| text, rows = panels.shared_format(fmt, token, mine=owner_id == event.sender_id) | |
| thumb = await self.db.get_format_thumb(owner_id, fmt["id"]) if fmt["has_thumb"] else None | |
| if thumb: # the format's thumbnail on top, its details as the caption | |
| bio = io.BytesIO(thumb) | |
| bio.name = "thumbnail.jpg" | |
| try: | |
| await self.client.send_file(event.chat_id, bio, caption=text, buttons=panels.to_buttons(rows), parse_mode="html") | |
| return | |
| except Exception as exc: # noqa: BLE001 | |
| log.warning("could not send the thumbnail of a shared format (%s); showing its details only", exc) | |
| await self._reply(event, text, rows) | |
| async def _cb_shared(self, event: Any, uid: int, parts: List[str]) -> None: | |
| """The button on a shared format: put a copy of it in the user's own formats.""" | |
| token = parts[2] if len(parts) > 2 else "" | |
| shared = await self.db.get_shared_format(token) if SHARE_TOKEN.fullmatch(token) else None | |
| mine = None | |
| if shared is not None: | |
| if parts[1:2] != ["add"]: | |
| await event.answer() | |
| return | |
| if shared[0] == uid: | |
| await event.answer("That's your own format.", alert=True) | |
| return | |
| try: | |
| mine = await self.db.copy_format(uid, shared[0], shared[1]["id"], name_max=NAME_MAX) | |
| except FormatError: # only "limit" can be raised here | |
| await event.answer(f"You already have {MAX_FORMATS} {shared[1]['kind']} formats. Delete one first.", alert=True) | |
| return | |
| if mine is None: # the owner deleted it after sharing | |
| await self._edit_panel(event, panels.SHARE_GONE, [[("✖ Close", "x:close")]]) | |
| await event.answer("That shared format isn't available anymore.", alert=True) | |
| return | |
| text, rows = panels.shared_added(mine) | |
| await self._edit_panel(event, text, rows) | |
| await event.answer("Added to your formats.") | |
| # ---------------------------------------------------------- /rename picker | |
| async def _cb_rename(self, event: Any, uid: int, parts: List[str]) -> None: | |
| self.pending.pop(uid, None) | |
| action = parts[1] if len(parts) > 1 else "" | |
| if action == "p" and len(parts) == 3 and parts[2].isdigit(): | |
| fmt = await self.db.get_format(uid, int(parts[2])) | |
| if fmt is None or fmt["kind"] != "manual": | |
| text, rows = panels.rename_picker(await self.db.list_formats(uid, "manual")) | |
| await self._edit_panel(event, text, rows) | |
| await event.answer("That format no longer exists.", alert=True) | |
| return | |
| mode = self._start_mode(uid, "rename", ONE_FILE_WAIT, {"fid": fmt["id"]}) | |
| text, rows = panels.rename_ready(fmt, ONE_FILE_WAIT // 60) | |
| await self._edit_panel(event, text + (f"\n\n{mode.dropped}" if mode.dropped else ""), rows) | |
| await event.answer(f"Format: {clip(fmt['name'], 40)}") | |
| elif action == "menu": | |
| if self._mode(uid, "rename") is not None: | |
| self._end_mode(uid) | |
| text, rows = panels.rename_picker(await self.db.list_formats(uid, "manual")) | |
| await self._edit_panel(event, text, rows) | |
| await event.answer() | |
| elif action == "stop": | |
| if self._mode(uid, "rename") is not None: | |
| self._end_mode(uid) | |
| await self._edit_panel(event, "⏹ Stopped. Send /rename again to pick a format.", [[("✖ Close", "x:close")]]) | |
| await event.answer() | |
| else: | |
| await event.answer() | |
| # ------------------------------------------------------------ /sequence buttons | |
| async def _cb_sequence(self, event: Any, uid: int, parts: List[str]) -> None: | |
| action = parts[1] if len(parts) > 1 else "" | |
| mode = self._mode(uid, "sequence") | |
| if mode is None: | |
| await event.answer("That sequence is over.", alert=True) | |
| return | |
| if action == "sort": | |
| await event.answer() | |
| await self._run_sort(uid, event.chat_id) | |
| elif action == "cx": | |
| self._end_mode(uid) | |
| await event.answer("Cancelled.") | |
| await self._edit_panel(event, self._mode_cancelled_text(mode), []) | |
| else: | |
| await event.answer() | |