Automated sync from GitHub

.gitattributes

@@ -1,8 +1,8 @@
-*.otf filter=lfs diff=lfs merge=lfs -text
-*.webp filter=lfs diff=lfs merge=lfs -text
-*.mp4 filter=lfs diff=lfs merge=lfs -text
 *.mp3 filter=lfs diff=lfs merge=lfs -text
 *.jpg filter=lfs diff=lfs merge=lfs -text
 *.jpeg filter=lfs diff=lfs merge=lfs -text
 *.png filter=lfs diff=lfs merge=lfs -text
 *.ttf filter=lfs diff=lfs merge=lfs -text
+*.otf filter=lfs diff=lfs merge=lfs -text
+*.webp filter=lfs diff=lfs merge=lfs -text
+*.mp4 filter=lfs diff=lfs merge=lfs -text

app/common.py

@@ -1,6 +1,7 @@
 """Shared context, permission guard, message helpers and post send/edit operations."""
 from __future__ import annotations
 
+import asyncio
 import functools
 import logging
 from dataclasses import dataclass, field
@@ -30,6 +31,7 @@ class Ctx:
     seen_channels: dict = field(default_factory=dict)  # channel id -> Channel entity seen in updates
     text_handlers: dict = field(default_factory=dict)  # mode -> async fn(event, state)
     callbacks: dict = field(default_factory=dict)  # prefix -> async fn(event, parts)
+    lock: asyncio.Lock = field(default_factory=asyncio.Lock)  # one long job (replace / repost) at a time
 
 
 def is_cb(event) -> bool:

app/db.py

@@ -102,6 +102,37 @@ class Change(Base):
     created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)
 
 
+class Migration(Base):
+    """One /repost run: a channel copied post by post, so it can be continued, undone or finished."""
+
+    __tablename__ = "migrations"
+
+    id: Mapped[str] = mapped_column(String(32), primary_key=True)
+    channel_id: Mapped[int] = mapped_column(BigInteger, index=True)
+    old_username: Mapped[Optional[str]] = mapped_column(String(64), nullable=True)
+    new_username: Mapped[Optional[str]] = mapped_column(String(64), nullable=True)
+    include_typed: Mapped[bool] = mapped_column(Boolean, default=True)
+    include_posts: Mapped[bool] = mapped_column(Boolean, default=False)
+    first_id: Mapped[int] = mapped_column(BigInteger)
+    last_id: Mapped[int] = mapped_column(BigInteger)
+    partial: Mapped[bool] = mapped_column(Boolean, default=False)
+    # copying | stopped | incomplete | copied | old_deleted | copies_deleted
+    status: Mapped[str] = mapped_column(String(16), default="copying")
+    created_by: Mapped[int] = mapped_column(BigInteger, default=0)
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow)
+
+
+class MigrationItem(Base):
+    __tablename__ = "migration_items"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    migration_id: Mapped[str] = mapped_column(String(32), ForeignKey("migrations.id"), index=True)
+    old_id: Mapped[int] = mapped_column(BigInteger)
+    new_id: Mapped[int] = mapped_column(BigInteger)
+    old_deleted: Mapped[bool] = mapped_column(Boolean, default=False)
+    new_deleted: Mapped[bool] = mapped_column(Boolean, default=False)
+
+
 def normalize_db_url(url: str) -> tuple:
     """Turn a plain DATABASE_URL into a SQLAlchemy async URL + connect args."""
     connect_args: dict = {}
@@ -358,6 +389,85 @@ class Database:
                     b.undone = True
             await s.commit()
 
+
+    # ------------------------------------------------------------------ /repost
+    OPEN_STATES = ("copying", "stopped", "incomplete", "copied")
+
+    async def create_migration(self, mid: str, channel_id: int, *, old, new, include_typed: bool, include_posts: bool,
+                               first_id: int, last_id: int, partial: bool, user_id: int) -> None:
+        async with self.Session() as s:
+            s.add(Migration(id=mid, channel_id=channel_id, old_username=old, new_username=new,
+                            include_typed=include_typed, include_posts=include_posts, first_id=first_id,
+                            last_id=last_id, partial=partial, created_by=user_id))
+            await s.commit()
+
+    async def get_migration(self, mid: str) -> Optional[Migration]:
+        async with self.Session() as s:
+            return await s.get(Migration, mid)
+
+    async def open_migration(self, channel_id: Optional[int] = None) -> Optional[Migration]:
+        """The newest repost that is not finished yet (copies exist, old posts not deleted / copies not removed)."""
+        async with self.Session() as s:
+            q = select(Migration).where(Migration.status.in_(self.OPEN_STATES))
+            if channel_id is not None:
+                q = q.where(Migration.channel_id == channel_id)
+            q = q.order_by(desc(Migration.created_at)).limit(1)
+            return (await s.execute(q)).scalar_one_or_none()
+
+    async def set_migration_status(self, mid: str, status: str) -> None:
+        async with self.Session() as s:
+            m = await s.get(Migration, mid)
+            if m is not None:
+                m.status = status
+                await s.commit()
+
+    async def migration_done_old_ids(self, mid: str) -> set:
+        async with self.Session() as s:
+            q = select(MigrationItem.old_id).where(MigrationItem.migration_id == mid)
+            return set((await s.execute(q)).scalars())
+
+    async def record_repost(self, mid: str, channel_id: int, rows: list, user_id: int) -> None:
+        """rows: [{old_id, new_id, post: {Post fields}}] - saves the mapping and registers the new posts."""
+        if not rows:
+            return
+        async with self.Session() as s:
+            for r in rows:
+                s.add(MigrationItem(migration_id=mid, old_id=r["old_id"], new_id=r["new_id"]))
+                s.add(Post(channel_id=channel_id, message_id=r["new_id"], status="sent", source="bot",
+                           created_by=user_id, sent_at=utcnow(), **r["post"]))
+            await s.commit()
+
+    async def migration_pending_ids(self, mid: str, which: str, limit: int = 100) -> list:
+        col, flag = (MigrationItem.old_id, MigrationItem.old_deleted) if which == "old" else (MigrationItem.new_id, MigrationItem.new_deleted)
+        async with self.Session() as s:
+            q = select(col).where(MigrationItem.migration_id == mid, flag.is_(False)).order_by(col).limit(limit)
+            return list((await s.execute(q)).scalars())
+
+    async def mark_migration_deleted(self, mid: str, channel_id: int, which: str, ids: list) -> None:
+        """The messages are gone from Telegram: flag them and drop their saved post copies."""
+        if not ids:
+            return
+        col, flag = ("old_id", "old_deleted") if which == "old" else ("new_id", "new_deleted")
+        async with self.Session() as s:
+            q = select(MigrationItem).where(MigrationItem.migration_id == mid, getattr(MigrationItem, col).in_(ids))
+            for it in (await s.execute(q)).scalars():
+                setattr(it, flag, True)
+            pq = select(Post).where(Post.channel_id == channel_id, Post.message_id.in_(ids))
+            for p in (await s.execute(pq)).scalars():
+                await s.delete(p)
+            await s.commit()
+
+    async def migration_counts(self, mid: str) -> dict:
+        async with self.Session() as s:
+            total = (await s.execute(select(func.count()).select_from(MigrationItem).where(MigrationItem.migration_id == mid))).scalar_one()
+            old_left = (await s.execute(select(func.count()).select_from(MigrationItem).where(
+                MigrationItem.migration_id == mid, MigrationItem.old_deleted.is_(False)))).scalar_one()
+            new_left = (await s.execute(select(func.count()).select_from(MigrationItem).where(
+                MigrationItem.migration_id == mid, MigrationItem.new_deleted.is_(False)))).scalar_one()
+            lo = (await s.execute(select(func.min(MigrationItem.new_id)).where(MigrationItem.migration_id == mid))).scalar_one()
+            hi = (await s.execute(select(func.max(MigrationItem.new_id)).where(MigrationItem.migration_id == mid))).scalar_one()
+            return {"copied": total, "old_left": old_left, "new_left": new_left, "first_new": lo, "last_new": hi}
+
     # -------------------------------------------------------------------- export
     async def export_all(self) -> dict:
         def clean(d: dict) -> dict:

app/handlers/basic.py

@@ -23,7 +23,9 @@ HELP = (
     "<b>Owner tools</b>\n"
     "/replace @old @new - swap the username inside t.me links (button links and hyperlinks) in all channels\n"
     "/undo - undo the last /replace\n"
+    "/repost @old @new - copy every post of a channel in order (links swapped), then delete the old posts\n"
     "/testedit &lt;post link&gt; - check that the bot can edit the buttons of one specific post\n"
+    "/selftest - check that editing buttons works on a post made by this bot\n"
     "/export - download a JSON backup of your posts"
 )
 

app/handlers/posts.py

@@ -386,8 +386,8 @@ def register(ctx: Ctx) -> None:
         await show(
             event,
             f"Delete this post from <b>{esc(ch.title)}</b>?\n\n"
-            "Telegram only lets a bot delete posts that are less than 48 hours old. For older posts, delete them "
-            "by hand in the channel and then use “Only forget it”.",
+            "Telegram's documentation says bots can only delete posts younger than 48 hours; in channels that limit may "
+            "not apply. If Telegram refuses, delete the post by hand and then use “Only forget it in the bot”.",
             [
                 [Button.inline("🗑 Delete from channel", f"dly:{post.id}:a")],
                 [Button.inline("🧹 Only forget it in the bot", f"dly:{post.id}:b")],
@@ -402,8 +402,9 @@ def register(ctx: Ctx) -> None:
             res = await client.delete_messages(peer_of(ch), [post.message_id])
             if not sum(getattr(r, "pts_count", 0) or 0 for r in res):
                 raise UserError(
-                    "Telegram did not delete that message - bots can only delete posts younger than 48 hours. "
-                    "Delete it by hand in the channel, then use “Only forget it in the bot”."
+                    "Telegram did not delete that message (the bot may lack the Delete right, or the post is older than "
+                    "the 48 hours Telegram's documentation mentions). Delete it by hand in the channel, then use "
+                    "“Only forget it in the bot”."
                 )
         await db.delete_post(post.id)
         await show(event, "🗑 Done.", [[Button.inline("📚 My posts", "pl")]])

app/handlers/replace.py

@@ -13,7 +13,8 @@ from telethon import Button, errors, types
 from ..common import Ctx, UserError, cmd, edit_callback_message, guard, on_cb, require_owner, say, show
 from ..linkswap import normalize_username
 from ..replace_engine import ApplyResult, ChannelScan, ScanOptions, apply_channel, scan_channel, undo_batch
-from ..tgutil import button_url, edit_raw, esc, explain_rpc, get_rights, peer_of, with_url
+from ..tgutil import build_markup, button_url, edit_raw, esc, explain_rpc, get_rights, peer_of, short, with_url
+from ..ui import CANCEL
 from .channels import parse_channel_ref
 
 log = logging.getLogger(__name__)
@@ -101,6 +102,9 @@ def summary_text(job: Job) -> str:
             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 (check with /testedit)"
             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"
@@ -118,11 +122,11 @@ def summary_text(job: Job) -> str:
 
 def register(ctx: Ctx) -> None:
     client, db, cfg = ctx.client, ctx.db, ctx.cfg
-    lock = asyncio.Lock()
+    lock = ctx.lock
 
     def purge_tokens() -> None:
         now = time.monotonic()
-        for k in [k for k, v in ctx.pending.items() if now - getattr(v, "created", now) > TOKEN_TTL]:
+        for k in [k for k, v in ctx.pending.items() if not getattr(v, "running", False) and now - getattr(v, "created", now) > TOKEN_TTL]:
             ctx.pending.pop(k, None)
 
     def make_updater(event):
@@ -230,6 +234,7 @@ def register(ctx: Ctx) -> None:
         batch_id = secrets.token_hex(6)
         o = job.opts
         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)
@@ -255,6 +260,8 @@ def register(ctx: Ctx) -> None:
                 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)
         lines = [f"✅ <b>Done</b> - <code>@{esc(o.old)}</code> → <code>@{esc(o.new)}</code>", f"• edited: <b>{totals.edited}</b>"]
         if totals.unchanged:
@@ -267,8 +274,10 @@ def register(ctx: Ctx) -> None:
             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"}:
-                lines.append("  Telegram doesn't let this bot edit those posts (see /testedit).")
+            if names & {"MessageAuthorRequiredError", "InlineBotRequiredError", "ChatAdminRequiredError", "MessageIdInvalidError"}:
+                lines.append("  Telegram doesn't let this bot edit those posts (see /testedit and /selftest).")
+        for title, name in stopped:
+            lines.append(f"⚠️ Stopped early in <b>{esc(title)}</b>: Telegram refused 5 edits in a row ({esc(name)}).")
         kb = [[Button.inline("↩️ Undo this replace", f"ruc:{batch_id}")]] if totals.edited else None
         if kb:
             try:
@@ -322,6 +331,94 @@ def register(ctx: Ctx) -> None:
             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)
 
+
+    # ---------------------------------------------------------------- /selftest
+    async def ask_selftest(event, ch) -> None:
+        await say(
+            event,
+            f"🧪 <b>Self-test in {esc(ch.title)}</b>\n"
+            "The bot posts a short test message with one button (silently), changes the button link, reads it back "
+            "and deletes the message again. It takes a few seconds and shows whether editing buttons works "
+            "for posts made by this bot.",
+            [[Button.inline("🧪 Run the test", f"st:{ch.id}"), Button.inline("✖️ Cancel", "cx")]],
+        )
+
+    @client.on(cmd("selftest", args=True))
+    @guard(ctx, owner=True)
+    async def h_selftest(event):
+        ref = (event.pattern_match.group(1) or "").strip() or None
+        try:
+            chans = pick_channels(await db.list_channels(), ref)
+        except ValueError as e:
+            raise UserError(str(e))
+        if not chans:
+            raise UserError("No channels registered yet. Use /addchannel.")
+        if len(chans) == 1:
+            await ask_selftest(event, chans[0])
+            return
+        kb = [[Button.inline(f"📢 {short(c.title, 40)}", f"stc:{c.id}")] for c in chans] + CANCEL
+        await say(event, "Run the self-test in which channel?", kb)
+
+    @on_cb(ctx, "stc")
+    async def cb_selftest_pick(event, parts):
+        require_owner(ctx, event)
+        ch = await db.get_channel(int(parts[0]))
+        if ch is None:
+            raise UserError("That channel is not registered.")
+        await ask_selftest(event, ch)
+
+    @on_cb(ctx, "st")
+    async def cb_selftest(event, parts):
+        require_owner(ctx, event)
+        ch = await db.get_channel(int(parts[0]))
+        if ch is None:
+            raise UserError("That channel is not registered.")
+        peer = peer_of(ch)
+        first = build_markup([[{"t": "Test button", "u": "https://example.com/a"}]])
+        second = build_markup([[{"t": "Test button", "u": "https://example.com/b"}]])
+        lines = [f"🧪 <b>Self-test in {esc(ch.title)}</b>"]
+        edit_ok = False
+        sent = None
+        try:
+            sent = await client.send_message(
+                peer, "🧪 Edit test - this message deletes itself in a moment.", buttons=first, link_preview=False, silent=True
+            )
+            lines.append("✅ posted a test message")
+        except errors.RPCError as e:
+            lines.append(f"❌ could not post: {esc(explain_rpc(e))}")
+        if sent is not None:
+            try:
+                await edit_raw(client, peer, sent.id, markup=second)
+                got = (await client.get_messages(peer, ids=[sent.id]))[0]
+                urls = []
+                if got is not None and isinstance(got.reply_markup, types.ReplyInlineMarkup):
+                    urls = [button_url(b) for row in got.reply_markup.rows for b in row.buttons if button_url(b)]
+                edit_ok = urls == ["https://example.com/b"]
+                lines.append(
+                    "✅ changed the button link and read it back"
+                    if edit_ok
+                    else "⚠️ Telegram accepted the edit but the link did not change"
+                )
+            except errors.RPCError as e:
+                lines.append(f"❌ could not edit the button: {esc(explain_rpc(e))}")
+            try:
+                res = await client.delete_messages(peer, [sent.id])
+                gone = sum(getattr(r, "pts_count", 0) or 0 for r in res)
+                lines.append("✅ deleted the test message" if gone else "⚠️ could not delete it - please delete it by hand")
+            except errors.RPCError as e:
+                lines.append(f"⚠️ could not delete it ({esc(type(e).__name__)}) - please delete it by hand")
+        if edit_ok:
+            lines.append(
+                "\n<b>Editing buttons works for posts made by this bot.</b> If /testedit still fails on Channel Help "
+                "posts, Telegram doesn't let this bot change buttons that another bot added."
+            )
+        else:
+            lines.append(
+                "\n<b>Editing failed even for this bot's own post</b>, so check the bot's admin rights in the channel "
+                "(Post messages, Edit messages of others, Delete messages)."
+            )
+        await show(event, "\n".join(lines))
+
     # ---------------------------------------------------------------- /testedit
     link_rx = re.compile(
         r"(?i)(?:https?://)?(?:t\.me|telegram\.me)/(?:c/(?P<cid>\d+)|(?P<user>[A-Za-z0-9_]{3,32}))/(?P<mid>\d+)"
@@ -395,9 +492,18 @@ def register(ctx: Ctx) -> None:
         try:
             await edit_raw(client, peer, job.mid, markup=types.ReplyInlineMarkup(rows))
         except errors.RPCError as e:
+            extra = ""
+            if type(e).__name__ == "MessageIdInvalidError":
+                extra = (
+                    "\n\nThe post does exist - I read it a moment ago - so this is not about a missing message. "
+                    "For edits Telegram uses this error when the bot isn't allowed to change the message, typically "
+                    "because the buttons were added by another bot.\n"
+                    "Run /selftest: it tries the same edit on a post made by this bot. If that works, your setup is fine "
+                    "and Telegram simply doesn't let this bot change those buttons."
+                )
             await show(
                 event,
-                f"❌ <b>Telegram refused the edit</b>\n{esc(explain_rpc(e))}\n\n"
+                f"❌ <b>Telegram refused the edit</b> ({esc(type(e).__name__)})\n{esc(explain_rpc(e))}{extra}\n\n"
                 "So /replace can NOT change the buttons of this post with this bot.",
             )
             return

app/handlers/repost.py

@@ -0,0 +1,434 @@
+"""/repost: copy a whole channel in order (links swapped on the way), then remove the old posts. Owner only."""
+from __future__ import annotations
+
+import logging
+import re
+import secrets
+import time
+from dataclasses import dataclass, field
+
+from telethon import Button, errors
+
+from ..common import Ctx, UserError, cmd, edit_callback_message, guard, on_cb, require_owner, say
+from ..linkswap import normalize_username
+from ..repost_engine import RepostOptions, RepostPlan, delete_copies, plan_repost, run_repost
+from ..tgutil import esc, get_rights, post_link
+from .replace import pick_channels
+
+log = logging.getLogger(__name__)
+
+USAGE = (
+    "Usage: <code>/repost @old @new</code>\n"
+    "Copies every post of the channel, in order, to the end of the channel and swaps the username inside "
+    "t.me links on the way. Leave out the usernames to copy without changing links.\n"
+    "Options: <code>--channel @name</code>, <code>--last 5</code> (trial: only the newest 5 post ids), "
+    "<code>--posts</code> (also change t.me/old/123 post links), <code>--no-typed</code>.\n\n"
+    "Nothing is deleted by /repost itself: you decide about the old posts after checking the copies."
+)
+
+
+@dataclass
+class RepostJob:
+    user: int
+    ch: object
+    opts: RepostOptions
+    plan: object = None
+    running: bool = False
+    stop: bool = False
+    mig_id: str = ""
+    created: float = field(default_factory=time.monotonic)
+
+
+def parse_repost_args(raw: str) -> 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 5")
+            flags["last"] = int(toks[i + 1])
+            i += 2
+        elif t == "--posts":
+            flags["posts"] = True
+            i += 1
+        elif t == "--no-typed":
+            flags["typed"] = False
+            i += 1
+        elif t.startswith("--"):
+            raise ValueError(f"Unknown option {t}")
+        else:
+            pos.append(t)
+            i += 1
+    old = new = None
+    if pos:
+        if len(pos) != 2:
+            raise ValueError("Give both usernames (@old @new), or none to copy without changing links.")
+        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 plan_text(p: RepostPlan, o: RepostOptions, title: str, delay: float) -> str:
+    kinds = f"{p.text} text, {p.media} media"
+    if p.polls:
+        kinds += f", {p.polls} poll(s)"
+    if p.other:
+        kinds += f", {p.other} other"
+    lines = [
+        f"🧱 <b>Repost plan - {esc(title)}</b>",
+        f"Post ids {p.first}-{p.last}: <b>{p.units} posts</b> to copy ({p.messages} messages, {p.albums} album(s)) - {kinds}",
+    ]
+    if p.service:
+        lines.append(f"• {p.service} service message(s) (like “pinned a message”) can't be copied and are skipped")
+    if p.polls:
+        lines.append("• polls are copied without their votes")
+    if o.swaps:
+        lines.append(
+            f"• <code>@{esc(o.old)}</code> → <code>@{esc(o.new)}</code> in {p.link_posts} post(s): "
+            f"{p.n_buttons} button link(s), {p.n_links} hyperlink(s), {p.n_typed} typed link(s)"
+        )
+    if p.dropped_posts:
+        lines.append(f"• {p.dropped_posts} post(s) carry other bots' buttons (reactions...) - those are not copied")
+    minutes = max(1, round(p.eta_seconds(delay) / 60))
+    lines += [
+        "",
+        "<b>How:</b> Telegram copies each post itself, so bold, links, quotes, expandable quotes, spoilers, media and "
+        "albums stay exactly as they are (if the channel blocks copying, posts are rebuilt from their parts). Copies go "
+        "to the end of the channel silently, in the original order, and are registered in My posts.",
+        "",
+        "<b>Not kept:</b> original dates, view counts, reactions, comments, poll votes, pinned state and links to the "
+        "old posts.",
+        f"Time: about {minutes} min. Please don't post in the channel meanwhile.",
+        "<b>Nothing is deleted in this step.</b>",
+    ]
+    if p.partial:
+        lines.append("Trial run: afterwards you can remove the copies again.")
+    return "\n".join(lines)
+
+
+def register(ctx: Ctx) -> None:
+    client, db, cfg = ctx.client, ctx.db, ctx.cfg
+
+    # ---------------------------------------------------------------- views
+    async def view_for(mig):
+        """Text and buttons describing where a repost stands."""
+        ch = await db.get_channel(mig.channel_id)
+        c = await db.migration_counts(mig.id)
+        name = esc(ch.title) if ch else str(mig.channel_id)
+        swap = f" (<code>@{esc(mig.old_username)}</code> → <code>@{esc(mig.new_username)}</code>)" if mig.old_username else ""
+        first_copy = ""
+        if ch and c["first_new"]:
+            first_copy = f'\n<a href="{post_link(ch, c["first_new"])}">First copy</a>'
+        undo = [Button.inline("↩️ Undo - remove the copies", f"rpu:{mig.id}")]
+        if mig.status == "copied" and not mig.partial:
+            text = (
+                f"✅ <b>{name}: all posts copied in order</b>{swap}\n{c['copied']} messages are now at the end of the channel "
+                f"and in My posts.{first_copy}\n\nLook at the end of the channel and check a few posts: formatting, "
+                "buttons, media. If it looks right, delete the old posts. If not, undo: the copies are removed and the "
+                "originals were never touched."
+            )
+            kb = [[Button.inline("🗑 Delete the old posts", f"rpd:{mig.id}")], undo]
+        elif mig.status == "copied":
+            text = (
+                f"✅ <b>{name}: trial finished</b>{swap}\n{c['copied']} message(s) copied.{first_copy}\n"
+                "Check them. Remove them again, or close this trial and keep them."
+            )
+            kb = [undo, [Button.inline("✅ Close, keep the copies", f"rpk:{mig.id}")]]
+        elif mig.status == "stopped":
+            text = f"⏹ <b>{name}: stopped</b>{swap}\n{c['copied']} message(s) copied so far.{first_copy}"
+            kb = [[Button.inline("▶️ Continue", f"rpc:{mig.id}")], undo]
+        elif mig.status == "copying":
+            text = f"⏳ <b>{name}</b>: a repost was running when the bot stopped. {c['copied']} message(s) copied.{first_copy}"
+            kb = [[Button.inline("▶️ Continue", f"rpc:{mig.id}")], undo]
+        else:  # incomplete
+            text = (
+                f"⚠️ <b>{name}: not everything was copied</b>{swap}\n{c['copied']} message(s) copied.{first_copy}\n"
+                "Old posts are only deleted once everything is copied, so nothing is lost. Try the rest again, "
+                "or undo."
+            )
+            kb = [[Button.inline("🔁 Try the rest again", f"rpc:{mig.id}")], undo]
+        return text, kb
+
+    # --------------------------------------------------------------- /repost
+    @client.on(cmd("repost", args=True))
+    @guard(ctx, owner=True)
+    async def h_repost(event):
+        raw = (event.pattern_match.group(1) or "").strip()
+        if not raw:
+            mig = await db.open_migration()
+            if mig is not None:
+                text, kb = await view_for(mig)
+                await say(event, "Unfinished repost:\n\n" + text, kb)
+            else:
+                await say(event, USAGE)
+            return
+        try:
+            old, new, flags = parse_repost_args(raw)
+            chans = await db.list_channels()
+            if flags["channel"]:
+                chans = pick_channels(chans, flags["channel"])
+            elif len(chans) > 1:
+                raise ValueError("You have several channels - add --channel @name.")
+        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.")
+        ch = chans[0]
+        mig = await db.open_migration(ch.id)
+        if mig is not None:
+            text, kb = await view_for(mig)
+            await say(event, "There is an unfinished repost for this channel. Finish or undo it first.\n\n" + text, kb)
+            return
+        rights = await get_rights(client, ch)
+        if rights is not None and not (rights.admin and rights.post):
+            raise UserError("The bot needs the “Post messages” admin right in that channel.")
+        if ctx.lock.locked():
+            raise UserError("Another long job (replace / repost) is still running.")
+        opts = RepostOptions(
+            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"],
+        )
+        async with ctx.lock:
+            status = await say(event, f"🔍 Reading <b>{esc(ch.title)}</b>…")
+            last = [0.0]
+
+            async def prog(p):
+                if time.monotonic() - last[0] < 3:
+                    return
+                last[0] = time.monotonic()
+                try:
+                    await status.edit(f"🔍 Reading <b>{esc(ch.title)}</b>: post id {p.scanned}…")
+                except Exception:
+                    pass
+
+            plan = await plan_repost(client, ch, opts, progress=prog)
+        if plan.error:
+            await status.edit(f"⚠️ {esc(plan.error)}")
+            return
+        job = RepostJob(user=event.sender_id, ch=ch, opts=opts, plan=plan)
+        token = secrets.token_hex(4)
+        ctx.pending[token] = job
+        kb = [[Button.inline("▶️ Start copying", f"rpa:{token}"), Button.inline("✖️ Cancel", f"rpn:{token}")]]
+        await status.edit(plan_text(plan, opts, ch.title, cfg.edit_delay), buttons=kb, link_preview=False)
+
+    @on_cb(ctx, "rpn")
+    async def cb_cancel(event, parts):
+        require_owner(ctx, event)
+        ctx.pending.pop(parts[0], None)
+        await edit_callback_message(event, "Cancelled - nothing was copied or deleted.")
+
+    # ------------------------------------------------------------------ copy
+    async def run_copy(event, job: RepostJob) -> None:
+        mig = await db.get_migration(job.mig_id)
+        ch = job.ch
+        job.running, job.stop = True, False
+        ctx.pending[f"job:{mig.id}"] = job
+        total = job.plan.units if job.plan else None
+        last = [0.0]
+
+        async def prog(res):
+            if time.monotonic() - last[0] < 4:
+                return
+            last[0] = time.monotonic()
+            done = res.copied_units + res.skipped_done
+            of = f"/{total}" if total else ""
+            try:
+                await edit_callback_message(
+                    event,
+                    f"⏳ Copying <b>{esc(ch.title)}</b>: {done}{of} posts…\nThe old posts stay untouched.",
+                    [[Button.inline("⏹ Stop", f"rps:{mig.id}")]],
+                )
+            except errors.MessageNotModifiedError:
+                pass
+            except Exception:
+                log.debug("progress update failed", exc_info=True)
+
+        try:
+            async with ctx.lock:
+                await db.set_migration_status(mig.id, "copying")
+                res = await run_repost(
+                    client, db, ch, mig, event.sender_id, delay=cfg.edit_delay, progress=prog, should_stop=lambda: job.stop
+                )
+        finally:
+            job.running = False
+        status = "stopped" if res.stopped else ("incomplete" if (res.failed or res.aborted) else "copied")
+        await db.set_migration_status(mig.id, status)
+        mig = await db.get_migration(mig.id)
+        text, kb = await view_for(mig)
+        notes = []
+        if res.rebuilt:
+            notes.append(f"{res.rebuilt} post(s) had to be rebuilt from their parts instead of copied by Telegram.")
+        if res.failed:
+            sample = ", ".join(f"#{i} {esc(n)}" for i, n in res.failed[:6])
+            notes.append(f"<b>{len(res.failed)} post(s) failed</b> ({sample}{'…' if len(res.failed) > 6 else ''}).")
+        if res.aborted:
+            notes.append(f"Stopped early because of {esc(res.aborted)}.")
+        await edit_callback_message(event, text + ("\n\n" + "\n".join(notes) if notes else ""), kb)
+
+    @on_cb(ctx, "rpa")
+    async def cb_start(event, parts):
+        require_owner(ctx, event)
+        job = ctx.pending.get(parts[0])
+        if not isinstance(job, RepostJob) or job.user != event.sender_id:
+            raise UserError("That plan has expired. Run /repost again.")
+        if job.running or ctx.lock.locked():
+            raise UserError("A long job is already running.")
+        if await db.open_migration(job.ch.id) is not None:
+            raise UserError("There is already an unfinished repost for this channel.")
+        p, o = job.plan, job.opts
+        job.mig_id = secrets.token_hex(6)
+        await db.create_migration(
+            job.mig_id, job.ch.id, old=o.old, new=o.new, include_typed=o.include_typed, include_posts=o.include_posts,
+            first_id=p.first, last_id=p.last, partial=p.partial, user_id=event.sender_id,
+        )
+        ctx.pending.pop(parts[0], None)
+        await run_copy(event, job)
+
+    @on_cb(ctx, "rpc")
+    async def cb_continue(event, parts):
+        require_owner(ctx, event)
+        mig = await db.get_migration(parts[0])
+        if mig is None or mig.status not in ("stopped", "incomplete", "copying"):
+            raise UserError("There is nothing to continue.")
+        if ctx.lock.locked():
+            raise UserError("A long job is already running.")
+        ch = await db.get_channel(mig.channel_id)
+        if ch is None:
+            raise UserError("That channel is not registered any more.")
+        opts = RepostOptions(old=mig.old_username, new=mig.new_username, include_typed=mig.include_typed, include_posts=mig.include_posts)
+        await run_copy(event, RepostJob(user=event.sender_id, ch=ch, opts=opts, mig_id=mig.id))
+
+    @on_cb(ctx, "rps")
+    async def cb_stop(event, parts):
+        require_owner(ctx, event)
+        job = ctx.pending.get(f"job:{parts[0]}")
+        if isinstance(job, RepostJob) and job.running:
+            job.stop = True
+            await event.respond("⏹ Stopping after the post that is being copied…")
+        else:
+            await event.respond("Nothing is running.")
+
+    # ----------------------------------------------------- delete old / undo
+    async def load(parts):
+        mig = await db.get_migration(parts[0])
+        if mig is None:
+            raise UserError("I don't know that repost any more.")
+        ch = await db.get_channel(mig.channel_id)
+        if ch is None:
+            raise UserError("That channel is not registered any more.")
+        return mig, ch
+
+    @on_cb(ctx, "rpb")
+    async def cb_back(event, parts):
+        mig, _ = await load(parts)
+        text, kb = await view_for(mig)
+        await edit_callback_message(event, text, kb)
+
+    @on_cb(ctx, "rpd")
+    async def cb_delete_ask(event, parts):
+        require_owner(ctx, event)
+        mig, ch = await load(parts)
+        if mig.status != "copied" or mig.partial:
+            raise UserError("Old posts can only be deleted after a complete, successful copy.")
+        c = await db.migration_counts(mig.id)
+        await edit_callback_message(
+            event,
+            f"🗑 <b>Delete {c['old_left']} old message(s) from {esc(ch.title)}?</b>\n"
+            "Only originals that have a copy are deleted. This can't be undone. The copies stay and become the channel.\n"
+            "If Telegram refuses (rights, or an age limit), nothing is lost: the copies are already there.",
+            [[Button.inline(f"🗑 Yes, delete {c['old_left']}", f"rpdy:{mig.id}"), Button.inline("↩️ Back", f"rpb:{mig.id}")]],
+        )
+
+    @on_cb(ctx, "rpdy")
+    async def cb_delete_old(event, parts):
+        require_owner(ctx, event)
+        mig, ch = await load(parts)
+        if mig.status not in ("copied", "old_deleted") or mig.partial:
+            raise UserError("Old posts can only be deleted after a complete, successful copy.")
+        if ctx.lock.locked():
+            raise UserError("A long job is already running.")
+        last = [0.0]
+
+        async def prog(r):
+            if time.monotonic() - last[0] < 3:
+                return
+            last[0] = time.monotonic()
+            try:
+                await edit_callback_message(event, f"🗑 Deleting old posts… {r.deleted} done")
+            except Exception:
+                pass
+
+        async with ctx.lock:
+            res = await delete_copies(client, db, ch, mig, "old", delay=0.5, progress=prog)
+        c = await db.migration_counts(mig.id)
+        if res.remaining == 0 and not res.error:
+            await db.set_migration_status(mig.id, "old_deleted")
+            await edit_callback_message(
+                event,
+                f"✅ <b>Done.</b> {res.deleted} old message(s) deleted. The channel now consists of the copies "
+                f"(from <a href=\"{post_link(ch, c['first_new'])}\">here</a>), and they are all in My posts.",
+            )
+            return
+        why = f"Telegram error: {esc(res.error)}." if res.error else (
+            "Telegram refused to delete the next batch (the bot's “Delete messages” right, or an age limit on old posts)."
+        )
+        await edit_callback_message(
+            event,
+            f"⚠️ Deleted {res.deleted}, {res.remaining} old message(s) are still there. {why}\n"
+            f"The copies are complete. You can delete the rest by hand: they are the messages with an id below "
+            f"{c['first_new']}. Then close this repost.",
+            [[Button.inline("🔁 Try again", f"rpdy:{mig.id}")], [Button.inline("✅ Close this repost", f"rpk:{mig.id}")]],
+        )
+
+    @on_cb(ctx, "rpu")
+    async def cb_undo_ask(event, parts):
+        require_owner(ctx, event)
+        mig, ch = await load(parts)
+        c = await db.migration_counts(mig.id)
+        if mig.status in ("old_deleted", "copies_deleted", "closed") or c["old_left"] < c["copied"]:
+            raise UserError("Some originals are already deleted, so removing the copies would delete your content. Undo is not possible any more.")
+        await edit_callback_message(
+            event,
+            f"↩️ <b>Remove the {c['new_left']} copied message(s) from {esc(ch.title)}?</b>\nThe original posts stay as they are.",
+            [[Button.inline("↩️ Yes, remove the copies", f"rpuy:{mig.id}"), Button.inline("Back", f"rpb:{mig.id}")]],
+        )
+
+    @on_cb(ctx, "rpuy")
+    async def cb_undo(event, parts):
+        require_owner(ctx, event)
+        mig, ch = await load(parts)
+        c = await db.migration_counts(mig.id)
+        if mig.status in ("old_deleted", "copies_deleted", "closed") or c["old_left"] < c["copied"]:
+            raise UserError("Undo is not possible any more.")
+        if ctx.lock.locked():
+            raise UserError("A long job is already running.")
+        async with ctx.lock:
+            res = await delete_copies(client, db, ch, mig, "new", delay=0.5)
+        if res.remaining == 0 and not res.error:
+            await db.set_migration_status(mig.id, "copies_deleted")
+            await edit_callback_message(event, f"↩️ Done. {res.deleted} copied message(s) removed; the original posts were not touched.")
+        else:
+            await edit_callback_message(
+                event,
+                f"⚠️ Removed {res.deleted}, {res.remaining} copy/copies are still there ({esc(res.error or 'Telegram refused')}).",
+                [[Button.inline("🔁 Try again", f"rpuy:{mig.id}")]],
+            )
+
+    @on_cb(ctx, "rpk")
+    async def cb_close(event, parts):
+        require_owner(ctx, event)
+        mig, _ = await load(parts)
+        await db.set_migration_status(mig.id, "closed")
+        await edit_callback_message(event, "✅ Closed. Nothing else will be changed in that channel by this repost.")

app/linkswap.py

@@ -209,3 +209,26 @@ def compute_message_changes(msg, old: str, new: str, *, include_typed: bool = Tr
         ),
         skipped,
     )
+
+
+def url_only_markup(markup, old: Optional[str] = None, new: Optional[str] = None, include_posts: bool = False):
+    """Buttons for a copy of a post: link buttons only (other bots' buttons would be dead), with the username
+    swapped when `old`/`new` are given. Returns (markup or None, number_of_buttons_not_copied)."""
+    if not isinstance(markup, types.ReplyInlineMarkup):
+        return None, 0
+    rows, dropped = [], 0
+    for row in markup.rows:
+        btns = []
+        for b in row.buttons:
+            url = button_url(b)
+            if url is None:
+                dropped += 1
+                continue
+            if old and new:
+                kind = classify_url(url, old)
+                if kind == "link" or (kind == "post" and include_posts):
+                    b = with_url(b, swap_url(url, old, new))
+            btns.append(b)
+        if btns:
+            rows.append(types.KeyboardButtonRow(btns))
+    return (types.ReplyInlineMarkup(rows) if rows else None), dropped

app/replace_engine.py

@@ -25,6 +25,7 @@ log = logging.getLogger(__name__)
 
 CHUNK = 100  # Telegram returns at most 100 messages per request
 FALLBACK_EMPTY_BATCHES = 15  # used only when the newest message id can't be determined
+ABORT_AFTER = 5  # stop a channel after this many identical Telegram refusals in a row
 
 
 @dataclass
@@ -75,6 +76,7 @@ class ApplyResult:
     missing: int = 0
     no_perm: int = 0
     failed: list = field(default_factory=list)
+    aborted: Optional[str] = None  # error name that made us stop early
 
 
 @dataclass
@@ -83,7 +85,7 @@ class UndoResult:
     failed: list = field(default_factory=list)
 
 
-async def _fetch(client, peer, ids: list) -> list:
+async def fetch_messages(client, peer, ids: list) -> list:
     res = await flood_retry(lambda: client.get_messages(peer, ids=ids))
     return list(res)
 
@@ -100,7 +102,7 @@ async def get_top_id(client, peer) -> Optional[int]:
     return None
 
 
-def _skip(m) -> bool:
+def is_skippable(m) -> bool:
     return m is None or isinstance(m, (types.MessageEmpty, types.MessageService))
 
 
@@ -126,7 +128,7 @@ async def scan_channel(
         if top is not None and a > top:
             break
         hi = a + CHUNK - 1 if top is None else min(a + CHUNK - 1, top)
-        msgs = await _fetch(client, peer, list(range(a, hi + 1)))
+        msgs = await fetch_messages(client, peer, list(range(a, hi + 1)))
         found = False
         for m in msgs:
             if m is None or isinstance(m, types.MessageEmpty):
@@ -192,6 +194,7 @@ async def apply_channel(
     res = ApplyResult()
     ids = sorted(scan.infos)
     buf: list = []
+    streak, last_name = 0, None
 
     async def flush() -> None:
         nonlocal buf
@@ -202,9 +205,9 @@ async def apply_channel(
     try:
         for i in range(0, len(ids), CHUNK):
             chunk = ids[i:i + CHUNK]
-            msgs = await _fetch(client, peer, chunk)
+            msgs = await fetch_messages(client, peer, chunk)
             for mid, m in zip(chunk, msgs):
-                if _skip(m):
+                if is_skippable(m):
                     res.missing += 1
                     continue
                 # re-check against the live message: it may have been edited since the scan
@@ -226,9 +229,16 @@ async def apply_channel(
                     res.unchanged += 1
                     continue
                 except errors.RPCError as e:
-                    log.warning("edit of %s/%s failed: %s", ch.id, m.id, e)
-                    res.failed.append((m.id, type(e).__name__))
+                    name = type(e).__name__
+                    log.warning("edit of %s/%s failed: %s", ch.id, m.id, name)
+                    res.failed.append((m.id, name))
+                    streak = streak + 1 if name == last_name else 1
+                    last_name = name
+                    if streak >= ABORT_AFTER:
+                        res.aborted = name
+                        break
                     continue
+                streak, last_name = 0, None
                 res.edited += 1
                 kind, fid = media_info(m)
                 adopt = {
@@ -252,6 +262,8 @@ async def apply_channel(
                 if progress:
                     await progress(res)
                 await asyncio.sleep(edit_delay)
+            if res.aborted:
+                break
     finally:
         await flush()
     return res

app/repost_engine.py

@@ -0,0 +1,488 @@
+"""Repost a whole channel in order.
+
+Every post is copied to the end of the channel (silently), in the same order as the original:
+  1. server-side copy (messages.forwardMessages with drop_author): text, formatting, quotes, media and
+     albums are copied exactly by Telegram itself;
+  2. if the channel forbids forwarding, or a copy can't be fixed up, the post is rebuilt from its parts.
+Links can be swapped on the way, the new posts are registered in the database, and the old posts are removed
+later in a separate step. Nothing here deletes anything unless asked to (delete_copies).
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+from dataclasses import dataclass, field
+from typing import Any, AsyncIterator, Awaitable, Callable, Optional
+
+from telethon import errors, functions, types, utils
+
+from .linkswap import compute_message_changes, url_only_markup
+from .replace_engine import CHUNK, FALLBACK_EMPTY_BATCHES, fetch_messages, get_top_id, is_skippable, preview_flag
+from .tgutil import classify_media, edit_raw, flood_retry, media_ref, peer_of, ser_entities, ser_markup
+
+log = logging.getLogger(__name__)
+
+ABORT_AFTER = 8  # consecutive posts that could not be copied -> stop
+FATAL = {
+    "ChatAdminRequiredError",
+    "ChatWriteForbiddenError",
+    "ChannelPrivateError",
+    "UserBannedInChannelError",
+    "ChannelInvalidError",
+    "ChatRestrictedError",
+}
+
+
+@dataclass
+class RepostOptions:
+    old: Optional[str] = None
+    new: Optional[str] = None
+    include_typed: bool = True
+    include_posts: bool = False
+    last: Optional[int] = None  # trial run: only the newest N post ids
+
+    @property
+    def swaps(self) -> bool:
+        return bool(self.old and self.new)
+
+
+@dataclass
+class Final:
+    """What a copy of a post should look like."""
+
+    text: str
+    entities: list
+    markup: Optional[types.ReplyInlineMarkup]
+    preview: Optional[bool]
+    text_changed: bool
+    n_buttons: int = 0
+    n_links: int = 0
+    n_typed: int = 0
+    dropped: int = 0  # other bots' buttons that are not copied
+
+
+def final_state(src, opts: RepostOptions) -> Final:
+    change = None
+    if opts.swaps:
+        change, _ = compute_message_changes(
+            src, opts.old, opts.new, include_typed=opts.include_typed, include_posts=opts.include_posts
+        )
+    text = change.new_text if change else (src.message or "")
+    entities = list(change.new_entities) if change else list(src.entities or [])
+    # text mentions need an access hash to be re-sent; they are rare in channel posts
+    entities = [e for e in entities if not isinstance(e, types.MessageEntityMentionName)]
+    markup, dropped = url_only_markup(
+        src.reply_markup, opts.old if opts.swaps else None, opts.new if opts.swaps else None, opts.include_posts
+    )
+    return Final(
+        text=text,
+        entities=entities,
+        markup=markup,
+        preview=preview_flag(src),
+        text_changed=bool(change and change.entities_changed),
+        n_buttons=change.n_buttons if change else 0,
+        n_links=change.n_links if change else 0,
+        n_typed=change.n_typed if change else 0,
+        dropped=dropped,
+    )
+
+
+def new_messages_from(result) -> list:
+    """The messages created by a send / forward request, oldest first."""
+    out = []
+    for u in getattr(result, "updates", None) or []:
+        if isinstance(u, (types.UpdateNewChannelMessage, types.UpdateNewMessage)):
+            out.append(u.message)
+    out.sort(key=lambda m: m.id)
+    return out
+
+
+# ------------------------------------------------------------------------------- planning
+@dataclass
+class RepostPlan:
+    channel: Any
+    first: int = 1
+    last: int = 0  # newest existing message id when the plan was made
+    scanned: int = 0
+    units: int = 0  # posts (an album counts once)
+    messages: int = 0
+    albums: int = 0
+    text: int = 0
+    media: int = 0
+    polls: int = 0
+    other: int = 0
+    service: int = 0
+    link_posts: int = 0
+    n_buttons: int = 0
+    n_links: int = 0
+    n_typed: int = 0
+    dropped_posts: int = 0
+    partial: bool = False
+    error: Optional[str] = None
+
+    def eta_seconds(self, delay: float) -> int:
+        return int(self.units * (delay + 1.0))
+
+
+async def find_last_id(client, peer, top: int) -> Optional[int]:
+    """Newest existing message id, searching backwards from the channel counter."""
+    a = top
+    for _ in range(300):
+        lo = max(1, a - CHUNK + 1)
+        msgs = await fetch_messages(client, peer, list(range(lo, a + 1)))
+        ids = [m.id for m in msgs if m is not None and not isinstance(m, types.MessageEmpty)]
+        if ids:
+            return max(ids)
+        if lo == 1:
+            return None
+        a = lo - 1
+    return None
+
+
+async def plan_repost(
+    client, ch, opts: RepostOptions, *, progress: Optional[Callable[[RepostPlan], Awaitable[None]]] = None
+) -> RepostPlan:
+    peer = peer_of(ch)
+    plan = RepostPlan(channel=ch, partial=bool(opts.last))
+    top = await get_top_id(client, peer)
+    first = 1
+    if opts.last:
+        if top is None:
+            plan.error = "I can't tell which post is the newest, so --last can't be used here."
+            return plan
+        newest = await find_last_id(client, peer, top)
+        if newest is None:
+            plan.error = "The channel has no posts."
+            return plan
+        first, top = max(1, newest - opts.last + 1), newest
+    plan.first = first
+    a, empty_run, prev_gid, max_seen = first, 0, None, 0
+    while True:
+        if top is not None and a > top:
+            break
+        hi = a + CHUNK - 1 if top is None else min(a + CHUNK - 1, top)
+        found = False
+        for m in await fetch_messages(client, peer, list(range(a, hi + 1))):
+            if m is None or isinstance(m, types.MessageEmpty):
+                continue
+            found = True
+            max_seen = max(max_seen, m.id)
+            if isinstance(m, types.MessageService):
+                plan.service += 1
+                continue
+            gid = getattr(m, "grouped_id", None)
+            if gid is None or gid != prev_gid:
+                plan.units += 1
+                if gid is not None:
+                    plan.albums += 1
+            prev_gid = gid
+            plan.messages += 1
+            media = m.media
+            if media is None or isinstance(media, types.MessageMediaWebPage):
+                plan.text += 1
+            elif isinstance(media, (types.MessageMediaPhoto, types.MessageMediaDocument)):
+                plan.media += 1
+            elif isinstance(media, types.MessageMediaPoll):
+                plan.polls += 1
+            else:
+                plan.other += 1
+            f = final_state(m, opts)
+            if f.n_buttons or f.n_links or f.n_typed:
+                plan.link_posts += 1
+                plan.n_buttons += f.n_buttons
+                plan.n_links += f.n_links
+                plan.n_typed += f.n_typed
+            if f.dropped:
+                plan.dropped_posts += 1
+        plan.scanned = hi
+        if top is None:
+            empty_run = 0 if found else empty_run + 1
+            if empty_run >= FALLBACK_EMPTY_BATCHES:
+                break
+        if progress:
+            await progress(plan)
+        a = hi + 1
+    plan.last = max_seen
+    if not plan.units:
+        plan.error = "There is nothing to copy."
+    return plan
+
+
+# ---------------------------------------------------------------------------------- copying
+async def iter_units(client, peer, first: int, last: int) -> AsyncIterator[list]:
+    """The posts between two ids, oldest first. An album is one unit (a list of messages)."""
+    group: list = []
+    a = first
+    while a <= last:
+        hi = min(a + CHUNK - 1, last)
+        for m in await fetch_messages(client, peer, list(range(a, hi + 1))):
+            if is_skippable(m):
+                continue
+            gid = getattr(m, "grouped_id", None)
+            if group and gid != group[0].grouped_id:
+                yield group
+                group = []
+            group.append(m)
+            if gid is None:
+                yield group
+                group = []
+        a = hi + 1
+    if group:
+        yield group
+
+
+async def delete_ids(client, peer, ids: list) -> None:
+    if not ids:
+        return
+    try:
+        await flood_retry(lambda: client(functions.channels.DeleteMessagesRequest(channel=peer, id=list(ids))))
+    except errors.RPCError as e:
+        log.warning("could not remove messages %s: %s", ids, type(e).__name__)
+
+
+async def forward_unit(client, peer, unit: list) -> list:
+    ids = [m.id for m in unit]
+    result = await flood_retry(
+        lambda: client(
+            functions.messages.ForwardMessagesRequest(from_peer=peer, id=ids, to_peer=peer, silent=True, drop_author=True)
+        )
+    )
+    return new_messages_from(result)
+
+
+async def edit_copies(client, peer, unit: list, finals: list, news: list) -> None:
+    """A copy may carry the old keyboard / old links: set the final text and buttons."""
+    for src, f, new in zip(unit, finals, news):
+        if f.text_changed or src.reply_markup is not None:
+            try:
+                await flood_retry(
+                    lambda: edit_raw(client, peer, new.id, text=f.text, entities=f.entities, markup=f.markup, preview=f.preview)
+                )
+            except errors.MessageNotModifiedError:
+                pass
+
+
+def _inverted(m) -> Optional[bool]:
+    return True if getattr(m, "invert_media", False) else None
+
+
+async def resend_unit(client, peer, unit: list, finals: list) -> list:
+    """Rebuild a post from its parts (used when copying is not allowed)."""
+    if len(unit) > 1:
+        singles = [
+            types.InputSingleMedia(media=utils.get_input_media(m.media), message=f.text, entities=f.entities or None)
+            for m, f in zip(unit, finals)
+        ]
+
+        def make():
+            return functions.messages.SendMultiMediaRequest(
+                peer=peer, multi_media=singles, silent=True, invert_media=_inverted(unit[0])
+            )
+
+    else:
+        m, f = unit[0], finals[0]
+        if m.media is None or isinstance(m.media, types.MessageMediaWebPage):
+
+            def make():
+                return functions.messages.SendMessageRequest(
+                    peer=peer,
+                    message=f.text,
+                    entities=f.entities or None,
+                    reply_markup=f.markup,
+                    no_webpage=True if f.preview is False else None,
+                    invert_media=_inverted(m),
+                    silent=True,
+                )
+
+        else:
+            media = utils.get_input_media(m.media)
+
+            def make():
+                return functions.messages.SendMediaRequest(
+                    peer=peer,
+                    media=media,
+                    message=f.text,
+                    entities=f.entities or None,
+                    reply_markup=f.markup,
+                    invert_media=_inverted(m),
+                    silent=True,
+                )
+
+    result = await flood_retry(lambda: client(make()))
+    news = new_messages_from(result)
+    if len(news) != len(unit):
+        await delete_ids(client, peer, [n.id for n in news])
+        raise RuntimeError(f"Telegram created {len(news)} messages instead of {len(unit)}")
+    return news
+
+
+@dataclass
+class _State:
+    mode: str = "forward"
+    forwarded: int = 0
+    rebuilt: int = 0
+
+
+async def copy_unit(client, peer, unit: list, finals: list, st: _State) -> list:
+    if st.mode == "forward":
+        news = None
+        try:
+            news = await forward_unit(client, peer, unit)
+            if len(news) != len(unit):
+                await delete_ids(client, peer, [n.id for n in news])
+                news = None
+            else:
+                try:
+                    await edit_copies(client, peer, unit, finals, news)
+                except errors.RPCError as e:
+                    log.info("could not fix up a copy (%s) - rebuilding the post", type(e).__name__)
+                    await delete_ids(client, peer, [n.id for n in news])
+                    news = None
+        except errors.ChatForwardsRestrictedError:
+            st.mode = "resend"
+            log.info("this channel restricts forwarding - rebuilding posts from their parts instead")
+        except errors.RPCError as e:
+            if type(e).__name__ in FATAL:
+                raise
+            log.info("copy failed (%s) - rebuilding the post", type(e).__name__)
+        if news is not None:
+            st.forwarded += 1
+            return news
+    news = await resend_unit(client, peer, unit, finals)
+    st.rebuilt += 1
+    return news
+
+
+@dataclass
+class RepostResult:
+    copied_units: int = 0
+    copied_messages: int = 0
+    skipped_done: int = 0
+    failed: list = field(default_factory=list)  # [(first old id, error name)]
+    forwarded: int = 0
+    rebuilt: int = 0
+    aborted: Optional[str] = None
+    stopped: bool = False
+    first_new: Optional[int] = None
+    last_new: Optional[int] = None
+
+
+def _rows(unit: list, finals: list, news: list) -> list:
+    rows = []
+    for src, f, new in zip(unit, finals, news):
+        kind = classify_media(src)
+        ref = (media_ref(new) or media_ref(src)) if kind else None
+        rows.append(
+            {
+                "old_id": src.id,
+                "new_id": new.id,
+                "post": {
+                    "text": f.text,
+                    "entities": ser_entities(f.entities),
+                    "buttons": ser_markup(f.markup),
+                    "media_kind": kind,
+                    "media_file_id": ref,
+                    "link_preview": bool(f.preview),
+                },
+            }
+        )
+    return rows
+
+
+async def run_repost(
+    client,
+    db,
+    ch,
+    mig,
+    user_id: int,
+    *,
+    delay: float = 1.2,
+    progress: Optional[Callable[[RepostResult], Awaitable[None]]] = None,
+    should_stop: Optional[Callable[[], bool]] = None,
+) -> RepostResult:
+    """Copy the posts of migration `mig` (ids first_id..last_id) in order; safe to run again to continue."""
+    opts = RepostOptions(
+        old=mig.old_username, new=mig.new_username, include_typed=mig.include_typed, include_posts=mig.include_posts
+    )
+    peer = peer_of(ch)
+    done = await db.migration_done_old_ids(mig.id)
+    res, st, streak = RepostResult(), _State(), 0
+    async for unit in iter_units(client, peer, mig.first_id, mig.last_id):
+        if should_stop and should_stop():
+            res.stopped = True
+            break
+        ids = [m.id for m in unit]
+        if all(i in done for i in ids):
+            res.skipped_done += 1
+            continue
+        finals = [final_state(m, opts) for m in unit]
+        try:
+            news = await copy_unit(client, peer, unit, finals, st)
+        except Exception as e:  # one bad post must not stop the others
+            name = type(e).__name__
+            log.warning("copying %s failed: %s %s", ids, name, e)
+            res.failed.append((ids[0], name))
+            streak += 1
+            if name in FATAL or streak >= ABORT_AFTER:
+                res.aborted = name
+                break
+            await asyncio.sleep(delay)
+            continue
+        try:
+            await db.record_repost(mig.id, ch.id, _rows(unit, finals, news), user_id)
+        except Exception as e:  # keep the channel and the database consistent
+            log.exception("could not save the copy of %s", ids)
+            await delete_ids(client, peer, [n.id for n in news])
+            res.failed.append((ids[0], type(e).__name__))
+            res.aborted = type(e).__name__
+            break
+        streak = 0
+        done.update(ids)
+        res.copied_units += 1
+        res.copied_messages += len(unit)
+        res.first_new = res.first_new or news[0].id
+        res.last_new = news[-1].id
+        if progress:
+            await progress(res)
+        await asyncio.sleep(delay)
+    res.forwarded, res.rebuilt = st.forwarded, st.rebuilt
+    return res
+
+
+# ----------------------------------------------------------------------------- deleting
+@dataclass
+class DeleteResult:
+    deleted: int = 0
+    remaining: int = 0
+    blocked: bool = False  # Telegram removed nothing from a whole batch
+    error: Optional[str] = None
+
+
+async def delete_copies(
+    client, db, ch, mig, which: str, *, delay: float = 0.5, progress: Optional[Callable[[DeleteResult], Awaitable[None]]] = None
+) -> DeleteResult:
+    """which='old': remove the original posts; which='new': remove the copies (undo). Oldest first, 100 per request."""
+    peer = peer_of(ch)
+    res = DeleteResult()
+    while True:
+        ids = await db.migration_pending_ids(mig.id, which, 100)
+        if not ids:
+            break
+        try:
+            await flood_retry(lambda: client(functions.channels.DeleteMessagesRequest(channel=peer, id=ids)))
+        except errors.RPCError as e:
+            res.error = type(e).__name__
+            break
+        still = await fetch_messages(client, peer, ids)
+        gone = [i for i, m in zip(ids, still) if m is None or isinstance(m, types.MessageEmpty)]
+        if not gone:
+            res.blocked = True
+            break
+        await db.mark_migration_deleted(mig.id, ch.id, which, gone)
+        res.deleted += len(gone)
+        if progress:
+            await progress(res)
+        await asyncio.sleep(delay)
+    counts = await db.migration_counts(mig.id)
+    res.remaining = counts["old_left"] if which == "old" else counts["new_left"]
+    return res

bot.py

@@ -11,7 +11,7 @@ from telethon.sessions import StringSession
 from app.common import Ctx
 from app.config import Config, load_config
 from app.db import Database
-from app.handlers import basic, channels, create, posts, replace
+from app.handlers import basic, channels, create, posts, replace, repost
 from app.health import start_health_server
 
 log = logging.getLogger("bot")
@@ -22,8 +22,10 @@ COMMANDS = [
     ("addchannel", "Register a channel"),
     ("channels", "List channels"),
     ("replace", "Swap a username in links (owner)"),
+    ("repost", "Copy a whole channel in order (owner)"),
     ("undo", "Undo the last replace (owner)"),
     ("testedit", "Test editing one post (owner)"),
+    ("selftest", "Test editing the bot's own post (owner)"),
     ("export", "Backup as JSON (owner)"),
     ("cancel", "Cancel what I'm doing"),
     ("help", "Show help"),
@@ -42,7 +44,7 @@ def build_client(cfg: Config) -> TelegramClient:
 
 
 def register_all(ctx: Ctx) -> None:
-    for module in (basic, channels, create, posts, replace):
+    for module in (basic, channels, create, posts, replace, repost):
         module.register(ctx)