Spaces:
Paused
Paused
| """/replace media, /swap links, /handleswap and /undo (owner only).""" | |
| from __future__ import annotations | |
| import logging | |
| import re | |
| import secrets | |
| import time | |
| from dataclasses import dataclass, field | |
| from urllib.parse import urlparse | |
| from telethon import Button, errors | |
| from ..common import ( | |
| Ctx, | |
| UserError, | |
| apply_edit_reported, | |
| cmd, | |
| edit_callback_message, | |
| edit_report_text, | |
| guard, | |
| on_cb, | |
| purge_pending, | |
| require_owner, | |
| say, | |
| show, | |
| ) | |
| from ..linkswap import normalize_username | |
| from ..replace_engine import ApplyResult, ChannelScan, ScanOptions, apply_channel, scan_channel, undo_batch | |
| from ..richmsg import is_rich | |
| from ..richtext import message_text | |
| from ..sync_engine import SyncOptions, sync_ids | |
| from ..tgutil import classify_media, esc, explain_rpc, get_rights, pack_file_id | |
| from ..channelref import parse_channel_ref | |
| from .panel import check_lengths, markup_note | |
| log = logging.getLogger(__name__) | |
| USAGE = ( | |
| "Usage: <code>/swap @old @new</code>\n" | |
| "Options: <code>--channel @name</code> (only that channel), <code>--last 200</code> (only the newest 200 " | |
| "post ids), <code>--posts</code> (also change t.me/old/123 post links), <code>--no-typed</code> (don't touch " | |
| "links typed in the text)." | |
| ) | |
| USAGE_HANDLES = ( | |
| "Usage: <code>/handleswap @old @new</code>\n" | |
| "Finds the posts that mention the username <code>@old</code> (in their text or caption, and in button names) in all " | |
| "channels and changes it to <code>@new</code>. Only usernames change: links, hyperlinks and button links are left alone.\n" | |
| "Options: <code>--channel @name</code> (only that channel), <code>--last 200</code> (only the newest 200 post ids)." | |
| ) | |
| HANDLE_PREFIX = "hs" # ids of /handleswap runs start with this, so /undo can word its question right | |
| class Job: | |
| user: int | |
| opts: ScanOptions | |
| scans: list | |
| running: bool = False | |
| created: float = field(default_factory=time.monotonic) | |
| def parse_replace_args(raw: str, *, handles: bool = False) -> tuple: | |
| toks = (raw or "").split() | |
| flags = {"channel": None, "last": None, "posts": False, "typed": None} | |
| pos: list = [] | |
| i = 0 | |
| while i < len(toks): | |
| t = toks[i] | |
| if t == "--channel" and i + 1 < len(toks): | |
| flags["channel"] = toks[i + 1] | |
| i += 2 | |
| elif t == "--last" and i + 1 < len(toks): | |
| if not toks[i + 1].isdigit() or int(toks[i + 1]) < 1: | |
| raise ValueError("--last needs a number, for example --last 200") | |
| flags["last"] = int(toks[i + 1]) | |
| i += 2 | |
| elif t == "--posts" and not handles: | |
| flags["posts"] = True | |
| i += 1 | |
| elif t == "--no-typed" and not handles: | |
| flags["typed"] = False | |
| i += 1 | |
| elif t.startswith("--"): | |
| raise ValueError(f"Unknown option {t}") | |
| else: | |
| pos.append(t) | |
| i += 1 | |
| if len(pos) != 2: | |
| raise ValueError("Give the old and the new username.") | |
| old, new = normalize_username(pos[0]), normalize_username(pos[1]) | |
| if not old or not new: | |
| raise ValueError("Usernames look like @name (letters, digits and underscores).") | |
| if old.lower() == new.lower(): | |
| raise ValueError("The old and the new username are the same.") | |
| return old, new, flags | |
| def pick_channels(chans: list, ref) -> list: | |
| if not ref: | |
| return chans | |
| r = parse_channel_ref(ref) | |
| out = [c for c in chans if (isinstance(r, int) and c.id == r) or (isinstance(r, str) and (c.username or "").lower() == r.lower())] | |
| if not out: | |
| raise ValueError("That channel isn't registered (see /channels).") | |
| return out | |
| def parse_post_link(value: str): | |
| """Return (public username or private channel id, message id) for Telegram post links.""" | |
| try: | |
| u = urlparse(value if "://" in value else "https://" + value) | |
| bits = [x for x in u.path.strip("/").split("/") if x] | |
| if u.netloc.lower() not in {"t.me", "www.t.me", "telegram.me", "www.telegram.me"}: | |
| return None | |
| if bits and bits[0] == "c" and len(bits) == 3 and bits[1].isdigit() and bits[2].isdigit(): | |
| # Telegram's t.me/c URL uses the same positive channel ID stored by Telethon and in our database. | |
| return int(bits[1]), int(bits[2]) | |
| if bits and bits[0] == "s": | |
| bits = bits[1:] | |
| if len(bits) == 2 and bits[1].isdigit() and re.fullmatch(r"[A-Za-z][A-Za-z0-9_]{3,}", bits[0]): | |
| return bits[0].lower(), int(bits[1]) | |
| except Exception: | |
| pass | |
| return None | |
| def handles_summary(job: Job) -> str: | |
| o = job.opts | |
| lines = [f"🔎 <b>Scan finished</b> - swap the username <code>@{esc(o.old)}</code> → <code>@{esc(o.new)}</code>"] | |
| total = 0 | |
| for s in job.scans: | |
| head = f"\n<b>{esc(s.channel.title)}</b>" | |
| if s.error: | |
| lines.append(f"{head}\n⚠️ {esc(s.error)}") | |
| continue | |
| block = f"{head} - looked at post ids {s.first}-{s.scanned}" | |
| if s.total == 0: | |
| block += "\nNothing to change." | |
| else: | |
| t = sum(i.typed for i in s.infos.values()) | |
| b = sum(i.buttons for i in s.infos.values()) | |
| what = f"{t} mention(s) in the text" + (f", {b} in button names" if b else "") | |
| block += f"\n{s.total} post(s) to change: {what}" | |
| if s.foreign: | |
| block += f"\n• {s.foreign} of them were not posted by this bot" | |
| cb = sum(1 for i in s.infos.values() if i.callbacks) | |
| if cb: | |
| block += f"\n• {cb} carry other bots' buttons (e.g. reactions) - those stay as they are" | |
| if s.skipped_post_links: | |
| block += f"\n• left {s.skipped_post_links} mention(s) that are part of a link alone" | |
| if s.not_editable: | |
| block += f"\n⚠️ {s.not_editable} post(s) by others can't be edited: the bot lacks “Edit messages of others” here" | |
| total += s.total | |
| lines.append(block) | |
| lines.append(f"\n<b>Total: {total} post(s).</b>") | |
| if total: | |
| lines.append("Nothing has been changed yet. Every edit is logged, so /undo can restore the old versions.") | |
| return "\n".join(lines) | |
| def summary_text(job: Job) -> str: | |
| if job.opts.mode == "handles": | |
| return handles_summary(job) | |
| o = job.opts | |
| lines = [f"🔎 <b>Scan finished</b> - replace <code>@{esc(o.old)}</code> → <code>@{esc(o.new)}</code>"] | |
| total = 0 | |
| for s in job.scans: | |
| head = f"\n<b>{esc(s.channel.title)}</b>" | |
| if s.error: | |
| lines.append(f"{head}\n⚠️ {esc(s.error)}") | |
| continue | |
| block = f"{head} - looked at post ids {s.first}-{s.scanned}" | |
| if s.total == 0: | |
| block += "\nNothing to change." | |
| else: | |
| b = sum(i.buttons for i in s.infos.values()) | |
| h = sum(i.links for i in s.infos.values()) | |
| t = sum(i.typed for i in s.infos.values()) | |
| block += f"\n{s.total} post(s) to change: {b} button link(s), {h} hyperlink(s), {t} typed link(s)" | |
| if s.foreign: | |
| block += f"\n• {s.foreign} of them were not posted by this bot" | |
| fb = sum(1 for i in s.infos.values() if not i.mine and i.buttons) | |
| if fb: | |
| block += f"\n• {fb} are button posts made by someone else - Telegram may refuse to edit those; /repost can re-post them with the new links instead" | |
| cb = sum(1 for i in s.infos.values() if i.callbacks) | |
| if cb: | |
| block += f"\n• {cb} carry other bots' buttons (e.g. reactions) - those stay as they are" | |
| if s.skipped_post_links: | |
| block += f"\n• skipped {s.skipped_post_links} link(s) to channel posts (t.me/{o.old}/123) - use --posts to change them too" | |
| if s.not_editable: | |
| block += f"\n⚠️ {s.not_editable} post(s) by others can't be edited: the bot lacks “Edit messages of others” here" | |
| total += s.total | |
| lines.append(block) | |
| lines.append(f"\n<b>Total: {total} post(s).</b>") | |
| if total: | |
| lines.append("Nothing has been changed yet. Every edit is logged, so /undo can restore the old versions.") | |
| return "\n".join(lines) | |
| def register(ctx: Ctx) -> None: | |
| client, db, cfg = ctx.client, ctx.db, ctx.cfg | |
| lock = ctx.lock | |
| def make_updater(event): | |
| last = [0.0] | |
| async def upd(text: str, force: bool = False) -> None: | |
| now = time.monotonic() | |
| if not force and now - last[0] < 3: | |
| return | |
| last[0] = now | |
| try: | |
| await edit_callback_message(event, text) | |
| except errors.MessageNotModifiedError: | |
| pass | |
| except Exception: | |
| log.debug("progress update failed", exc_info=True) | |
| return upd | |
| # ------------------------------------------------------------------ /replace media | |
| async def h_replace(event): | |
| purge_pending(ctx) | |
| raw = (event.pattern_match.group(1) or "").strip() | |
| if not raw: | |
| await say(event, "Reply to a new photo, video, or file with <code>/replace <channel post link></code>.") | |
| return | |
| # File replacement mode: reply to a media message with /replace <link to the saved post>. | |
| link_bits = parse_post_link(raw) if len(raw.split()) == 1 else None | |
| if link_bits: | |
| reply = await event.get_reply_message() | |
| if reply is None: | |
| raise UserError("Reply to the new photo, video, or file, then send /replace <channel post link>.") | |
| kind = classify_media(reply) | |
| if not kind: | |
| raise UserError("The replied-to message must contain a photo, video, GIF, audio file, or document.") | |
| ref, message_id = link_bits | |
| channels = await db.list_channels() | |
| channel = next((c for c in channels if c.id == ref), None) if isinstance(ref, int) else next( | |
| (c for c in channels if (c.username or "").lower() == ref), None | |
| ) | |
| if channel is None: | |
| raise UserError("That post link is not from a registered channel. Check /channels.") | |
| post = await db.find_by_message(channel.id, message_id) | |
| if post is None: | |
| report = await sync_ids( | |
| ctx.client, | |
| db, | |
| channel, | |
| [message_id], | |
| SyncOptions(history=False, grace=0, delete_services=False, ignore=ctx.sync_ignore), | |
| adopt_ids={message_id}, | |
| ) | |
| post = await db.find_by_message(channel.id, message_id) | |
| if post is None: | |
| if report.error: | |
| raise UserError(f"I couldn't read that post: {report.error}") | |
| if report.restricted: | |
| raise UserError("Telegram is currently holding that post back, so I can't replace its media.") | |
| if report.unsupported: | |
| raise UserError("That post can't be saved in My posts, so its media can't be replaced here.") | |
| raise UserError("That post no longer exists in the channel, or I can't add it to My posts.") | |
| if post.status != "sent": | |
| raise UserError("That post is not published yet.") | |
| if is_rich(post): | |
| raise UserError("That post is a rich message - it has no media or caption to replace. Edit its text in My posts.") | |
| caption, entities, how = message_text(reply, cfg.markup_input) # Telegram's formatting, or HTML / Markdown | |
| check_lengths(caption, True) | |
| async def notify(text: str) -> None: | |
| await say(event, text) | |
| _, report = await apply_edit_reported( | |
| ctx, post, channel, replace_media=True, notify=notify, media_kind=kind, media_file_id=pack_file_id(reply), | |
| text=caption, entities=entities, | |
| ) | |
| extra = edit_report_text(report) | |
| await say( | |
| event, | |
| markup_note(how) | |
| + f"✅ Replaced the media in <b>{esc(channel.title)}</b> and updated My posts." | |
| + (f"\n{extra}" if extra else ""), | |
| ) | |
| return | |
| raise UserError("Use /replace with a channel post link, replying to the new media. Username swaps use /swap.") | |
| # ------------------------------------------------------------------ /swap usernames in links | |
| async def h_swap(event): | |
| purge_pending(ctx) | |
| raw = (event.pattern_match.group(1) or "").strip() | |
| if not raw: | |
| await say(event, USAGE) | |
| return | |
| try: | |
| old, new, flags = parse_replace_args(raw) | |
| chans = pick_channels(await db.list_channels(), flags["channel"]) | |
| except ValueError as e: | |
| raise UserError(f"{e}\n\n" + re.sub(r"<[^>]+>", "", USAGE)) | |
| if not chans: | |
| raise UserError("No channels registered yet. Use /addchannel.") | |
| opts = ScanOptions( | |
| old=old, | |
| new=new, | |
| include_typed=cfg.replace_typed_links if flags["typed"] is None else flags["typed"], | |
| include_posts=flags["posts"], | |
| last=flags["last"], | |
| ) | |
| await begin_scan(event, opts, chans) | |
| # ---------------------------------------------------------------- /handleswap | |
| async def h_handleswap(event): | |
| purge_pending(ctx) | |
| raw = (event.pattern_match.group(1) or "").strip() | |
| if not raw: | |
| await say(event, USAGE_HANDLES) | |
| return | |
| try: | |
| old, new, flags = parse_replace_args(raw, handles=True) | |
| chans = pick_channels(await db.list_channels(), flags["channel"]) | |
| except ValueError as e: | |
| raise UserError(f"{e}\n\n" + re.sub(r"<[^>]+>", "", USAGE_HANDLES)) | |
| if not chans: | |
| raise UserError("No channels registered yet. Use /addchannel.") | |
| await begin_scan(event, ScanOptions(old=old, new=new, last=flags["last"], mode="handles"), chans) | |
| async def begin_scan(event, opts, chans) -> None: | |
| """Read the channels, show what would change and wait for the owner's yes.""" | |
| if lock.locked(): | |
| raise UserError("Another /swap, /handleswap or /undo is still running - wait for it to finish.") | |
| async with lock: | |
| status = await say(event, f"🔍 Scanning {len(chans)} channel(s)…") | |
| last_edit = [0.0] | |
| scans = [] | |
| for ch in chans: | |
| rights = await get_rights(client, ch) | |
| if rights is not None and not rights.admin: | |
| s = await _empty_scan(ch, "The bot is not an admin here any more.") | |
| scans.append(s) | |
| continue | |
| can_edit = True if rights is None else rights.edit | |
| async def prog(scan, title=ch.title): | |
| now = time.monotonic() | |
| if now - last_edit[0] < 3: | |
| return | |
| last_edit[0] = now | |
| try: | |
| await status.edit(f"🔍 Scanning <b>{esc(title)}</b>: post id {scan.scanned} of {scan.top or '?'}…") | |
| except Exception: | |
| pass | |
| try: | |
| scan = await scan_channel( | |
| client, ch, opts, can_edit_others=can_edit, mine_ids=await db.mine_message_ids(ch.id), progress=prog | |
| ) | |
| except errors.RPCError as e: | |
| scan = await _empty_scan(ch, explain_rpc(e)) | |
| scan.rights = rights | |
| scans.append(scan) | |
| job = Job(user=event.sender_id, opts=opts, scans=scans) | |
| token = secrets.token_hex(4) | |
| ctx.pending[token] = job | |
| total = sum(s.total for s in scans) | |
| kb = None | |
| if total: | |
| kb = [[Button.inline(f"✅ Apply to {total} post(s)", f"rp:{token}:y"), Button.inline("✖️ Cancel", f"rp:{token}:n")]] | |
| try: | |
| await status.edit(summary_text(job), buttons=kb, link_preview=False) | |
| except Exception: | |
| await say(event, summary_text(job), kb) | |
| async def _empty_scan(ch, error: str): | |
| return ChannelScan(channel=ch, error=error) | |
| async def cb_replace(event, parts): | |
| require_owner(ctx, event) | |
| token, action = parts[0], parts[1] | |
| job = ctx.pending.get(token) | |
| if not isinstance(job, Job) or job.user != event.sender_id: | |
| raise UserError("That scan has expired. Run /swap again.") | |
| if action == "n": | |
| ctx.pending.pop(token, None) | |
| await show(event, "Cancelled - nothing was changed.") | |
| return | |
| if job.running or lock.locked(): | |
| raise UserError("It is already running.") | |
| job.running = True | |
| upd = make_updater(event) | |
| o = job.opts | |
| batch_id = (HANDLE_PREFIX if o.mode == "handles" else "") + secrets.token_hex(6) | |
| totals = ApplyResult() | |
| stopped: list = [] | |
| async with lock: | |
| await db.create_batch(batch_id, o.old, o.new, event.sender_id) | |
| todo = sum(s.total for s in job.scans) | |
| done_before = 0 | |
| for s in job.scans: | |
| if not s.infos: | |
| continue | |
| async def prog(res, s=s, base=done_before): | |
| await upd(f"⏳ <b>{esc(s.channel.title)}</b> - {base + res.edited}/{todo} edited…") | |
| try: | |
| res = await apply_channel( | |
| client, db, s, o, batch_id, event.sender_id, edit_delay=cfg.edit_delay, progress=prog | |
| ) | |
| except Exception as e: | |
| log.exception("apply failed in %s", s.channel.id) | |
| totals.failed.append((0, f"{s.channel.title}: {type(e).__name__}")) | |
| continue | |
| done_before += res.edited | |
| totals.edited += res.edited | |
| totals.unchanged += res.unchanged | |
| totals.missing += res.missing | |
| totals.no_perm += res.no_perm | |
| totals.failed += res.failed | |
| if res.aborted: | |
| stopped.append((s.channel.title, res.aborted)) | |
| ctx.pending.pop(token, None) | |
| what = "username swap " if o.mode == "handles" else "" | |
| lines = [f"✅ <b>Done</b> - {what}<code>@{esc(o.old)}</code> → <code>@{esc(o.new)}</code>", f"• edited: <b>{totals.edited}</b>"] | |
| if totals.unchanged: | |
| lines.append(f"• already fine / changed meanwhile: {totals.unchanged}") | |
| if totals.missing: | |
| lines.append(f"• deleted meanwhile: {totals.missing}") | |
| if totals.no_perm: | |
| lines.append(f"• skipped (no permission): {totals.no_perm}") | |
| if totals.failed: | |
| sample = ", ".join(f"#{m} {esc(n)}" for m, n in totals.failed[:8]) | |
| lines.append(f"• <b>failed: {len(totals.failed)}</b> ({sample}{'…' if len(totals.failed) > 8 else ''})") | |
| names = {n for _, n in totals.failed} | |
| if names & {"MessageAuthorRequiredError", "InlineBotRequiredError", "ChatAdminRequiredError", "MessageIdInvalidError"}: | |
| lines.append( | |
| " Telegram doesn't let this bot edit those posts." | |
| + ("" if o.mode == "handles" else " Use /repost to post them again with the new links.") | |
| ) | |
| for title, name in stopped: | |
| lines.append(f"⚠️ Stopped early in <b>{esc(title)}</b>: Telegram refused 5 edits in a row ({esc(name)}).") | |
| undo_label = "↩️ Undo this swap" if o.mode == "handles" else "↩️ Undo this replace" | |
| kb = [[Button.inline(undo_label, f"ruc:{batch_id}")]] if totals.edited else None | |
| if kb: | |
| try: | |
| await edit_callback_message(event, "\n".join(lines), kb) | |
| except errors.MessageNotModifiedError: | |
| pass | |
| else: | |
| await upd("\n".join(lines), force=True) | |
| # -------------------------------------------------------------------- /undo | |
| async def ask_undo(event, batch) -> None: | |
| n = await db.count_batch_changes(batch.id) | |
| what = "username swap" if batch.id.startswith(HANDLE_PREFIX) else "replace" | |
| await say( | |
| event, | |
| f"↩️ Undo the {what} <code>@{esc(batch.old_username)}</code> → <code>@{esc(batch.new_username)}</code>?\n" | |
| f"{n} post(s) will be put back exactly as they were before.", | |
| [[Button.inline("✅ Yes, undo", f"ru:{batch.id}"), Button.inline("✖️ No", "cx")]], | |
| ) | |
| async def h_undo(event): | |
| batch = await db.last_batch() | |
| if batch is None: | |
| raise UserError("There is nothing to undo - no /swap or /handleswap run yet.") | |
| await ask_undo(event, batch) | |
| async def cb_undo_ask(event, parts): | |
| require_owner(ctx, event) | |
| batch = await db.get_batch(parts[0]) | |
| if batch is None or batch.undone: | |
| raise UserError("That replace was already undone.") | |
| await ask_undo(event, batch) | |
| async def cb_undo(event, parts): | |
| require_owner(ctx, event) | |
| if lock.locked(): | |
| raise UserError("Another /swap, /handleswap or /undo is still running.") | |
| upd = make_updater(event) | |
| async with lock: | |
| await upd("⏳ Restoring…", force=True) | |
| async def prog(res): | |
| await upd(f"⏳ Restoring… {res.restored} done") | |
| res = await undo_batch(client, db, parts[0], edit_delay=cfg.edit_delay, progress=prog) | |
| text = f"↩️ Restored <b>{res.restored}</b> post(s)." | |
| if res.failed: | |
| text += f"\n⚠️ {len(res.failed)} could not be restored (" + ", ".join(f"#{m} {esc(n)}" for m, n in res.failed[:6]) + ")." | |
| await upd(text, force=True) | |