"""/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: /swap @old @new\n" "Options: --channel @name (only that channel), --last 200 (only the newest 200 " "post ids), --posts (also change t.me/old/123 post links), --no-typed (don't touch " "links typed in the text)." ) USAGE_HANDLES = ( "Usage: /handleswap @old @new\n" "Finds the posts that mention the username @old (in their text or caption, and in button names) in all " "channels and changes it to @new. Only usernames change: links, hyperlinks and button links are left alone.\n" "Options: --channel @name (only that channel), --last 200 (only the newest 200 post ids)." ) HANDLE_PREFIX = "hs" # ids of /handleswap runs start with this, so /undo can word its question right @dataclass 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"🔎 Scan finished - swap the username @{esc(o.old)} → @{esc(o.new)}"] total = 0 for s in job.scans: head = f"\n{esc(s.channel.title)}" 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"\nTotal: {total} post(s).") 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"🔎 Scan finished - replace @{esc(o.old)} → @{esc(o.new)}"] total = 0 for s in job.scans: head = f"\n{esc(s.channel.title)}" 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"\nTotal: {total} post(s).") 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 @client.on(cmd("replace", args=True)) @guard(ctx, owner=True) 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 /replace <channel post link>.") return # File replacement mode: reply to a media message with /replace . 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 .") 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 {esc(channel.title)} 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 @client.on(cmd("swap", args=True)) @guard(ctx, owner=True) 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 @client.on(cmd("handleswap", args=True)) @guard(ctx, owner=True) 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 {esc(title)}: 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) @on_cb(ctx, "rp") 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"⏳ {esc(s.channel.title)} - {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"✅ Done - {what}@{esc(o.old)} → @{esc(o.new)}", f"• edited: {totals.edited}"] 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"• failed: {len(totals.failed)} ({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 {esc(title)}: 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} @{esc(batch.old_username)} → @{esc(batch.new_username)}?\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")]], ) @client.on(cmd("undo")) @guard(ctx, owner=True) 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) @on_cb(ctx, "ruc") 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) @on_cb(ctx, "ru") 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 {res.restored} 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)