"""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. a post that needs no change (no buttons, no link to swap) is copied by Telegram itself (messages.forwardMessages with drop_author): text, formatting, quotes, media and albums stay exact; 2. a post with buttons, or with links to swap, is created again in ONE request that carries its final text, formatting and buttons - exactly how the bot publishes a post of its own - and is then read back to make sure the buttons really are there (a copy that is edited afterwards can end up without them); 3. if the channel forbids forwarding, every post is created like in 2. 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). What a plain copy would lose is put back as well: * a post that answers another post is posted as an answer to the COPY of that post (a reply can only be set when a post is created, so such a post is always created again instead of copied); * a link to another post of the channel (t.me/name/115) is pointed at the copy of that post - at once when the copy already exists, afterwards (finalize_repost) for posts that are copied later; * a post that was pinned has its copy pinned at the end. A post that TELEGRAM restricts (a copyright strike: the whole channel is "unavailable due to copyright infringement", or the post carries a restriction mark, or Telegram's apps show "This message couldn't be displayed on your device due to copyright infringement" instead of the post) is not copied as Telegram hands it out now. /shift rebuilds such a post from what My posts saved of it (text, formatting, buttons, media) - for a restricted CHANNEL every post that has a saved copy, however intact Telegram's version looks. A post without a saved copy is copied as Telegram hands it out when that is the post itself, and skipped (and reported) when it is only Telegram's notice. Telegram's notice is never posted, from whichever source. /repost leaves a post whose content Telegram holds back alone. Every copy of a restricted source is read back: if Telegram itself hides the new post, that is reported (copy_loop). """ from __future__ import annotations import asyncio import copy import logging from dataclasses import dataclass, field from typing import Any, AsyncIterator, Awaitable, Callable, Optional from telethon import errors, functions, types, utils from .buttons_sync import ButtonsNotShown, ensure_markup, link_urls from .crosslinks import CrossResult, relink_after_repost from .linkswap import compute_message_changes, url_only_markup from .postlinks import PostLinks, relink_message from .replace_engine import CHUNK, FALLBACK_EMPTY_BATCHES, fetch_messages, is_skippable, preview_flag from .restrictions import PLACEHOLDER_NOTE, is_placeholder_text, message_restriction, probe_channel from .richmsg import Rich, entry as rich_entry, is_rich_message, plain_of, rich_from_received, rich_of, with_rich from .tgutil import ( FATAL, build_markup, classify_media, de_entities, edit_raw, flood_retry, media_from_ref, media_ref, peer_of, ser_entities, ser_markup, ) log = logging.getLogger(__name__) ABORT_AFTER = 8 # consecutive posts that could not be copied -> stop @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 n_post_links: int = 0 # links to other posts of the same channel (whatever their new id will be) n_relinked: int = 0 # of those, how many were already pointed at a copy rich: Optional[Rich] = None # the post is a rich message: this is what the copy is made from (text / entities are empty) def rich_for(src) -> Optional[Rich]: """The rich message that `src` is - a live post, or a stand-in made from a saved post - or None for an ordinary post.""" saved = getattr(src, "rich_saved", None) if saved is not None: return saved rm = getattr(src, "rich_message", None) return rich_from_received(rm) if rm is not None else None def final_state(src, opts: RepostOptions, links: Optional[PostLinks] = None) -> Final: """What the copy of `src` should look like. `links` (optional) points links to other posts of the channel at the copies that exist so far (links.ids: old message id -> new one).""" change = None if opts.swaps: change, _ = compute_message_changes( src, opts.old, opts.new, include_typed=opts.include_typed, include_posts=opts.include_posts ) rich = rich_for(src) text = change.new_text if change else (src.message or "") entities = list(change.new_entities) if change else list(src.entities or []) if rich is not None: # a rich message's text is blocks: nothing of it is changed here, only its buttons text, entities = "", [] # 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 ) n_post_links = n_relinked = 0 text_changed = bool(change and change.entities_changed) if links is not None: n_post_links = links.count(text, entities, markup) if n_post_links and links.ids: text, entities, n_typed, n_hyper = links.remap_text(text, entities) markup, n_buttons = links.remap_markup(markup) n_relinked = n_typed + n_hyper + n_buttons text_changed = text_changed or bool(n_typed or n_hyper) # Send exactly what is saved: go through the stored form that "My posts" also uses to build its edits. markup = build_markup(ser_markup(markup)) if markup is not None else None entities = de_entities(ser_entities(entities)) return Final( text=text, entities=entities, markup=markup, preview=preview_flag(src), text_changed=text_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, n_post_links=n_post_links, n_relinked=n_relinked, rich=rich, ) # ---------------------------------------------------------------------------- replies and pins class ReplyParentMissing(RuntimeError): """A post answers another post whose copy does not exist (yet): copying it now would lose the answer.""" def reply_parent(msg) -> Optional[int]: """The message id a post answers (in the same channel), or None.""" h = getattr(msg, "reply_to", None) if h is None or getattr(h, "reply_to_peer_id", None) is not None or getattr(h, "forum_topic", None): return None return getattr(h, "reply_to_msg_id", None) or None def new_parent(msg, idmap: dict, seen: set) -> Optional[int]: """The copy of the post `msg` answers. None when it answers nothing, or the post it answered is gone / outside the range being copied. Raises ReplyParentMissing when that post was read in this run but has no copy.""" pid = reply_parent(msg) if pid is None: return None if pid in idmap: return idmap[pid] if pid in seen: raise ReplyParentMissing(f"post {msg.id} answers post {pid}, which has not been copied") return None 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 # ------------------------------------------------------ posts Telegram holds back (copyright strike ...) SavedLookup = Callable[[list], Awaitable[dict]] # message ids -> {message id: the saved post of My posts} def saved_lookup(db, channel_id: int) -> SavedLookup: """How the copy loop and the plan look up what My posts saved of the posts of one channel.""" async def lookup(ids: list) -> dict: return {p.message_id: p for p in await db.posts_at(channel_id, ids)} return lookup def usable_row(row) -> bool: """A saved post that can take the place of a restricted post: it has something to post, and what it says is not Telegram's notice (a saved copy that an older version made from a restricted post can hold the notice).""" if row is None: return False text = row.text or "" if is_placeholder_text(text): return False return bool(text.strip() or row.media_file_id) def copy_source(unit: list, rows: dict, *, restricted: Optional[str] = None, use_saved: bool = False) -> tuple: """Where the copy of this post (or album) comes from -> (source, why). "live" - as Telegram hands the post out: nothing restricts it, or Telegram restricts it but still hands out the post itself and there is no saved copy to use instead; "saved" - from what My posts saved of it: Telegram restricts the channel (`restricted`) or the post (a copyright strike) and My posts has every part of it. Only when `use_saved` (/shift) - the saved copy wins even when Telegram's version still looks intact, because that is the owner's rule for a struck channel; "skip" - not copied: what Telegram hands out is its notice - or nothing - and My posts can't take over. `why` is Telegram's own wording about the restriction, None when nothing restricts the post.""" marked = next((w for w in (message_restriction(m) for m in unit) if w), None) why = marked or restricted if why is None: return "live", None if use_saved and all(usable_row(rows.get(m.id)) for m in unit): return "saved", why for m in unit: live = m.message or "" if is_placeholder_text(live) or ( not live.strip() and getattr(m, "media", None) is None and rich_for(m) is None ): return "skip", why return "live", why def stand_in(m, row): """A copy of the live message `m` that shows what My posts saved for it - text, formatting, buttons, media. It takes the place of a post Telegram holds back (the live message keeps what Telegram still tells: id, album, reply).""" s = copy.copy(m) s.message = row.text or "" s.entities = de_entities(row.entities) s.rich_message = None # what it shows is what My posts saved ... s.rich_saved = rich_of(row) # ... which is a rich message when the saved post is one s.reply_markup = build_markup(row.buttons) s.media = None s.saved_ref = None # the saved media reference (resend_saved may need it again without its file reference) if row.media_file_id: try: s.media = media_from_ref(row.media_file_id) s.saved_ref = row.media_file_id except ValueError: # a reference that can't be read: the post goes out without media log.warning("the saved media of post %s can't be read", m.id) if s.media is None and row.link_preview: s.media = types.MessageMediaWebPage(webpage=types.WebPageEmpty(id=0)) return s def stand_ins(unit: list, rows: dict) -> Optional[list]: """The saved copies that stand in for a restricted post, or None when My posts does not have all of them (or one of them has nothing to post, or holds only Telegram's notice).""" out = [] for m in unit: row = rows.get(m.id) if not usable_row(row): return None out.append(stand_in(m, row)) return out FILE_MEDIA = (types.MessageMediaPhoto, types.MessageMediaDocument) # a photo, video, document, voice, sticker ... def has_file(m) -> bool: """The post carries a file. A link preview, a poll, a location or a contact is not a file.""" return isinstance(getattr(m, "media", None), FILE_MEDIA) def without_files(unit: list) -> tuple: """/shift --no-media: the posts of `unit` as they look without their files -> (those that still have a text, the number of files left out). A post that was only a file has nothing left to post and is dropped.""" out, lost = [], 0 for m in unit: c = m if has_file(m): c = copy.copy(m) c.media, c.saved_ref, c.invert_media = None, None, None lost += 1 if (c.message or "").strip() or rich_for(c) is not None: out.append(c) return out, lost async def single_posts(units: AsyncIterator[list]) -> AsyncIterator[list]: """/shift --no-media: an album can't be rebuilt without its pictures, so every message of it is a post of its own.""" async for unit in units: for m in unit: yield [m] def saved_message(row, channel_id: int): """A post built from its saved copy alone, for a channel that can't be read at all. Only the saved parts exist: no album, no reply, no pin.""" base = types.Message( id=row.message_id, peer_id=types.PeerChannel(channel_id), date=row.sent_at or row.created_at, message="" ) return stand_in(base, row) def saved_rows_between(rows: list, first: Optional[int], last: Optional[int]) -> list: """The saved posts (oldest first, one per message id) that can be posted - they have something to post that is not Telegram's notice - between two message ids.""" out, seen = [], set() for row in sorted(rows, key=lambda r: (r.message_id or 0, r.id)): mid = row.message_id if mid is None or mid in seen or (first and mid < first) or (last and mid > last): continue if usable_row(row): seen.add(mid) out.append(row) return out async def saved_units(db, channel_id: int, first: int, last: int) -> AsyncIterator[list]: """The saved posts of a channel between two message ids, oldest first - each one a unit of its own.""" for row in saved_rows_between(await db.sent_posts_from(channel_id, first), first, last): yield [saved_message(row, channel_id)] # ------------------------------------------------------------------------------- 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 post_link_posts: int = 0 # posts with a link to another post of this channel post_links: int = 0 pinned: int = 0 # pinned posts (their copies get pinned) replies: int = 0 # posts that answer another post held_saved: int = 0 # restricted posts that are copied from what My posts saved of them (counted in units) held_skipped: int = 0 # restricted posts that can't be copied: only Telegram's notice is left (not counted in units) live_restricted: int = 0 # restricted posts without a saved copy that are copied as Telegram hands them out no_media: bool = False # /shift --no-media: the files are left out files_left_out: int = 0 # ... posts that are copied as text without their file (counted in units) file_only: int = 0 # ... posts that are only a file: nothing to copy (not counted in units) held_why: Optional[str] = None # what Telegram says about them channel_restricted: Optional[str] = None # what Telegram says when it restricts the whole channel from_db: bool = False # the channel can't be read: the plan (and the copy) is made from My posts alone unreadable: Optional[str] = None # ... because of this partial: bool = False error: Optional[str] = None def eta_seconds(self, delay: float) -> int: return int(self.units * (delay + 1.0)) def tally(plan: RepostPlan, m, opts: RepostOptions, links: PostLinks) -> None: """Count one message (as it will be copied) into the plan.""" 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, links) 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 if f.n_post_links: plan.post_link_posts += 1 plan.post_links += f.n_post_links if getattr(m, "pinned", False): plan.pinned += 1 if reply_parent(m) is not None: plan.replies += 1 def plan_from_saved( ch, rows: list, opts: RepostOptions, *, first_id=None, last_id=None, unreadable=None, no_media: bool = False ) -> RepostPlan: """The plan of a copy made from My posts alone (the channel can't be read): one post per saved message.""" plan = RepostPlan( channel=ch, partial=bool(first_id or last_id), from_db=True, unreadable=unreadable, no_media=no_media ) links = PostLinks.for_channel(ch) chosen = saved_rows_between(rows, first_id, last_id) for row in chosen: m = saved_message(row, ch.id) if no_media: kept, lost = without_files([m]) if not kept: plan.file_only += 1 continue m, plan.files_left_out = kept[0], plan.files_left_out + lost plan.units += 1 tally(plan, m, opts, links) if chosen: plan.first, plan.last = chosen[0].message_id, chosen[-1].message_id plan.scanned = plan.last else: plan.first, plan.last = first_id or 1, last_id or 0 if not plan.units: plan.error = "There is nothing to copy." if plan.file_only: plan.error = "Every saved post in this range is only a file, so without files there is nothing to copy." return plan 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, first_id: Optional[int] = None, last_id: Optional[int] = None, saved: Optional[SavedLookup] = None, use_saved: bool = False, no_media: bool = False, ) -> RepostPlan: """Read the channel and count what a copy would do. first_id / last_id limit the range (used by /shift). `saved` looks up My posts' copies of the channel's posts; with `use_saved` a post Telegram holds back is counted as the saved post it will be copied from (/shift), without it such a post is left out (/repost). With `no_media` the files are left out of the copies (/shift --no-media): a post that is only a file has nothing left to copy.""" peer = peer_of(ch) plan = RepostPlan(channel=ch, partial=bool(opts.last), no_media=no_media) try: top, plan.channel_restricted, _ = await probe_channel(client, peer) except Exception as e: # the counter is only a help: without it the ids are read until nothing more turns up log.info("could not read channel counter: %s", e) top = None 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 if first_id: first = max(first, first_id) plan.partial = True if last_id: top = last_id if top is None else min(top, last_id) plan.partial = True plan.first = first links = PostLinks.for_channel(ch) 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 rows = await saved(list(range(a, hi + 1))) if saved else {} 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) new_unit = no_media or gid is None or gid != prev_gid # without files an album is a post per message row = rows.get(m.id) # a struck channel changes how /shift copies (My posts first); /repost copies as before, post by post source, why = copy_source( [m], rows, restricted=plan.channel_restricted if use_saved else None, use_saved=use_saved ) if why: plan.held_why = plan.held_why or why if source == "skip": plan.held_skipped += 1 if new_unit else 0 # nothing to copy it from prev_gid = gid continue if source == "saved": m = stand_in(m, row) # counted as the saved post it will be copied from if no_media: kept, lost = without_files([m]) if not kept: # nothing but a file: nothing to copy without it plan.file_only += 1 prev_gid = gid continue m, plan.files_left_out = kept[0], plan.files_left_out + lost if source == "saved": plan.held_saved += 1 if new_unit else 0 elif why: plan.live_restricted += 1 if new_unit else 0 if new_unit: plan.units += 1 if gid is not None and not no_media: plan.albums += 1 prev_gid = gid tally(plan, m, opts, links) 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." if plan.held_skipped: plan.error = ( "Telegram holds back every post in this range (it shows a notice instead of the post) and My posts has " "no saved copy of them, so there is nothing to copy." ) elif plan.file_only: plan.error = "Every post in this range is only a file, so without files 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, src_peer, unit: list, dst_peer=None) -> list: """Let Telegram copy the posts (no "forwarded from" header). `dst_peer` defaults to the same channel.""" ids = [m.id for m in unit] to_peer = dst_peer if dst_peer is not None else src_peer result = await flood_retry( lambda: client( functions.messages.ForwardMessagesRequest( from_peer=src_peer, id=ids, to_peer=to_peer, silent=True, drop_author=True ) ) ) return new_messages_from(result) def _inverted(m) -> Optional[bool]: return True if getattr(m, "invert_media", False) else None def reply_header(reply_to: Optional[int]): """The `reply_to` argument of the send requests: answer message `reply_to` of the same channel.""" return types.InputReplyToMessage(reply_to_msg_id=reply_to) if reply_to else None async def resend_unit(client, peer, unit: list, finals: list, reply_to: Optional[int] = None) -> list: """Rebuild a post from its parts (used when copying is not allowed, or the copy must differ from the original). `reply_to`: message id (in the channel `peer`) the new post answers.""" reply = reply_header(reply_to) sender = None if len(unit) == 1 and getattr(finals[0], "rich", None) is not None: rich_final = finals[0] # a rich message: Telegram builds it again from its blocks (or its HTML / Markdown) async def send_rich_final(): return await with_rich( rich_final.rich, lambda inp: client( functions.messages.SendMessageRequest( peer=peer, message="", rich_message=inp, reply_markup=rich_final.markup, silent=True, reply_to=reply ) ), ) sender = send_rich_final elif 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]), reply_to=reply ) 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, reply_to=reply, ) 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, reply_to=reply, ) result = await flood_retry(sender if sender is not None else (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" # becomes "resend" when the channel forbids forwarding forwarded: int = 0 rebuilt: int = 0 repaired: int = 0 # new posts whose buttons had to be set again after posting replies: int = 0 # new posts created as an answer to the copy of the post the original answered reply_lost: int = 0 # answers that could only be copied without the reply from_saved: int = 0 # posts Telegram holds back that were rebuilt from their saved copy in My posts media_lost: list = field(default_factory=list) # old ids of those that had to go out without their media def is_pure(src, f: Final, reply_to: Optional[int] = None) -> bool: """True when a server-side copy is already the final post: nothing to swap, no keyboard to rebuild and no reply to set (a reply can only be given when a post is created).""" return not f.text_changed and src.reply_markup is None and not reply_to def _has_media(unit: list) -> bool: return any(m.media is not None and not isinstance(m.media, types.MessageMediaWebPage) for m in unit) def _without_reference(unit: list) -> list: """The same saved posts with the file reference of their media left out (it may have expired).""" out = [] for m in unit: b = copy.copy(m) if getattr(m, "saved_ref", None): b.media = media_from_ref(m.saved_ref, blank_reference=True) out.append(b) return out async def resend_saved(client, dst_peer, unit: list, finals: list, st: _State, reply_to: Optional[int] = None) -> list: """Create the copies of posts that Telegram holds back, from the saved posts that stand in for them (stand_in). Their media is sent from the saved reference: when Telegram says the reference is out of date it is tried without it, and a single post whose text is saved goes out without its media when Telegram still refuses the file (a file taken down for copyright is refused) - counted in st.media_lost.""" failure: Optional[BaseException] = None for attempt in (unit, None): if attempt is None: if failure is None or not _has_media(unit): break attempt = _without_reference(unit) try: return await resend_unit(client, dst_peer, attempt, finals, reply_to) except (TypeError, ValueError, errors.RPCError) as e: if isinstance(e, errors.RPCError) and type(e).__name__ in FATAL: raise failure = failure or e if not _has_media(unit): raise if len(unit) == 1 and (finals[0].text or "").strip(): unit[0].media = None # what goes out now has no media, and neither has its entry in My posts news = await resend_unit(client, dst_peer, unit, finals, reply_to) st.media_lost.append(unit[0].id) return news raise failure # type: ignore[misc] async def copy_unit( client, peer, unit: list, finals: list, st: _State, reply_to: Optional[int] = None, dst_peer=None, *, from_saved: bool = False, no_forward: bool = False, ) -> list: """Create the copy of one post (or album) at the end of `dst_peer` (default: the same channel `peer`). `from_saved`: the posts in `unit` are saved copies standing in for posts Telegram restricts - never forwarded. `no_forward`: Telegram restricts these posts and refuses to copy them itself: they are rebuilt from their parts.""" dst = dst_peer if dst_peer is not None else peer if from_saved: news = await resend_saved(client, dst, unit, finals, st, reply_to) st.rebuilt += 1 st.from_saved += 1 if reply_to: st.replies += 1 return news pure = all(is_pure(m, f, reply_to) for m, f in zip(unit, finals)) if st.mode == "forward" and pure and not no_forward: try: news = await forward_unit(client, peer, unit, dst) if len(news) == len(unit): st.forwarded += 1 return news await delete_ids(client, dst, [n.id for n in news]) 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__) try: news = await resend_unit(client, dst, unit, finals, reply_to) except (TypeError, ValueError, errors.RPCError) as e: # A kind of media a post cannot be rebuilt from (or a reply Telegram refuses). If only the reply made a # rebuild necessary, a plain copy by Telegram still works - without the reply. if isinstance(e, errors.RPCError) and type(e).__name__ in FATAL: raise if reply_to and st.mode == "forward" and not no_forward and all(is_pure(m, f) for m, f in zip(unit, finals)): news = await forward_unit(client, peer, unit, dst) if len(news) != len(unit): await delete_ids(client, dst, [n.id for n in news]) raise st.reply_lost += 1 st.forwarded += 1 return news raise st.rebuilt += 1 if reply_to: st.replies += 1 return news async def settle_buttons(client, peer, finals: list, news: list, st: _State, *, pause: float) -> None: """Read the new posts back: each must show the link buttons it should. Repairs it, or removes the copies.""" for f, new in zip(finals, news): if not link_urls(f.markup): continue try: how = await ensure_markup(client, peer, new.id, f.markup, pause=pause) except errors.RPCError as e: await delete_ids(client, peer, [n.id for n in news]) if type(e).__name__ in FATAL: raise raise ButtonsNotShown(f"could not set the buttons ({type(e).__name__})") from e except (ButtonsNotShown, LookupError) as e: await delete_ids(client, peer, [n.id for n in news]) raise ButtonsNotShown(str(e)) from e if how != "fine": st.repaired += 1 @dataclass class FinalizeResult: """The finishing touches after the copies exist: links to other posts, pins.""" checked: int = 0 # copies read back relinked: int = 0 # copies whose links to other posts now point at the new copies links: int = 0 # links changed in total missing: int = 0 # copies that do not exist any more failed: list = field(default_factory=list) # [(message id, error name)] pin_wanted: int = 0 # copies of pinned posts that still had to be pinned pinned: int = 0 pin_error: Optional[str] = None aborted: Optional[str] = None stopped: bool = False @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 repaired: int = 0 replies: int = 0 # copies created as an answer to the copy of the post the original answered reply_lost: int = 0 # answers that could only be copied without the reply reply_dropped: int = 0 # answers to a post that no longer exists (nothing to answer) from_saved: int = 0 # restricted posts that were copied from their saved copy in My posts media_lost: list = field(default_factory=list) # ... of which these (old ids) went out without their media held_back: list = field(default_factory=list) # posts that were not copied: only Telegram's notice was left (old id) held_why: Optional[str] = None # what Telegram says about the restricted posts live_restricted: int = 0 # restricted posts without a saved copy, copied as Telegram hands them out no_media: bool = False # /shift --no-media: no file was copied files_left_out: int = 0 # ... posts that were copied as text, without their file file_only: list = field(default_factory=list) # ... posts that are only a file: nothing to copy (old id) used_saved: bool = False # the run could take restricted posts from My posts (/shift) from_db: bool = False # the channel can't be read: every post was copied from My posts alone checked: int = 0 # new posts that were read back to see whether Telegram itself hides them hidden: list = field(default_factory=list) # [(old id, had media)] new posts Telegram itself hides (restriction mark) hidden_why: Optional[str] = None # what Telegram says about them dst_restricted: Optional[str] = None # what Telegram says about the destination channel (when copies were hidden) hidden_stop: bool = False # the run stopped by itself: Telegram hides every copy it read back aborted: Optional[str] = None stopped: bool = False first_new: Optional[int] = None last_new: Optional[int] = None phase: str = "copy" # copy | finish | profile | links final: Optional[FinalizeResult] = None final_error: Optional[str] = None cross: Optional[CrossResult] = None # links to the old posts in ALL connected channels, pointed at the copies cross_error: Optional[str] = None profile: Any = None # /shift -all: what happened to the name, description and photo (profile_copy.ProfileResult) profile_error: Optional[str] = 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": plain_of(f.rich) if f.rich is not None else f.text, "entities": [rich_entry(f.rich)] if f.rich is not None else ser_entities(f.entities), "buttons": ser_markup(f.markup), "media_kind": kind, "media_file_id": ref, "link_preview": bool(f.preview), }, } ) return rows CHECK_FIRST = 3 # the first copies of any run are read back ... CHECK_EVERY = 10 # ... then every tenth (all of them when Telegram restricts the source) STOP_IF_HIDDEN = 3 # /shift stops when Telegram hides this many new copies and not one of those read back is shown async def check_copies(client, dst_peer, unit: list, news: list, res: RepostResult) -> None: """Read the new posts back: does TELEGRAM itself hide them (a restriction mark, or its notice as the text)? That is not the bot copying a notice - Telegram restricts the new post as well, usually because of a file or a link in it. What is found goes into `res` (checked, hidden, hidden_why). Never fails the copy: it is only a look.""" try: got = await fetch_messages(client, dst_peer, [n.id for n in news]) except Exception as e: # a look that fails is not a failed copy (the post exists and is registered already) log.info("could not read the new posts back: %s", type(e).__name__) return for old, new in zip(unit, got): if is_skippable(new): continue res.checked += 1 why = message_restriction(new) if why: res.hidden.append((old.id, _has_media([old]))) res.hidden_why = res.hidden_why or why @dataclass class CopyJob: """Everything the copy loop needs to know; /repost and /shift each fill it in their own way.""" src_peer: Any dst_peer: Any # the same as src_peer for /repost opts: RepostOptions links: PostLinks # links.ids is `idmap`: old message id -> id of its copy, grows as we go first_id: int last_id: int done: set # old ids that already have a copy idmap: dict record: Callable[[list], Awaitable[None]] # saves [{old_id, new_id, post}] (and registers the posts) saved: Optional[SavedLookup] = None # what My posts saved of the source's posts use_saved: bool = False # a post Telegram holds back is rebuilt from its saved copy (otherwise it is left out) units: Optional[Callable[[], AsyncIterator[list]]] = None # where the posts come from (default: read the channel) from_db: bool = False # `units` are saved posts: they are all rebuilt, none is read from Telegram restricted: Optional[str] = None # what Telegram says about the source channel itself (a copyright strike), if anything no_media: bool = False # the copies get no files: text, formatting and buttons only (/shift --no-media) stop_if_hidden: int = 0 # stop when this many copies were read back and Telegram hides every one (0: never) async def copy_loop( client, job: CopyJob, *, delay: float = 1.2, progress: Optional[Callable[[RepostResult], Awaitable[None]]] = None, should_stop: Optional[Callable[[], bool]] = None, settle_pause: float = 0.7, ) -> RepostResult: """Copy the posts first_id..last_id of `job.src_peer` to the end of `job.dst_peer`, in order. Safe to run again to continue: posts in `job.done` are skipped.""" src, dst = job.src_peer, job.dst_peer seen: set = set() # old ids read in this run (a reply to one of them needs its copy) res, st, streak = RepostResult(used_saved=job.use_saved or job.from_db, from_db=job.from_db), _State(), 0 res.no_media = job.no_media units = job.units() if job.units is not None else iter_units(client, src, job.first_id, job.last_id) if job.no_media: units = single_posts(units) async for unit in units: if should_stop and should_stop(): res.stopped = True break ids = [m.id for m in unit] seen.update(ids) if all(i in job.done for i in ids): res.skipped_done += 1 continue rows = await job.saved(ids) if (job.saved and not job.from_db) else {} if job.from_db: source, why = "saved", None else: source, why = copy_source(unit, rows, restricted=job.restricted, use_saved=job.use_saved) if why: # Telegram restricts this post (or the whole channel): what it hands out now is not trusted res.held_why = res.held_why or why from_saved = job.from_db # saved posts already standing = None if source == "saved" and not job.from_db: standing = stand_ins(unit, rows) if standing is None: source = "skip" if source == "skip": log.info("post %s: Telegram shows only its notice (%s) and My posts has no saved copy to use", ids[0], why) res.held_back.append(ids[0]) continue if standing is not None: unit, from_saved = standing, True no_forward, left_out = bool(why), 0 if job.no_media: # the copy is the text with its formatting and buttons; its file stays where it is unit, left_out = without_files(unit) if not unit: # nothing but a file: there is nothing to post without it, and nothing waits for it res.file_only.append(ids[0]) seen.difference_update(ids) continue no_forward = no_forward or bool(left_out) # a copy by Telegram would carry the file finals = [final_state(m, job.opts, job.links) for m in unit] if any(is_placeholder_text(f.text) for f in finals): # Telegram's notice is never posted, whatever it came from log.info("post %s: its text is Telegram's notice - not copied", ids[0]) res.held_back.append(ids[0]) res.held_why = res.held_why or PLACEHOLDER_NOTE continue try: reply_to = new_parent(unit[0], job.idmap, seen) news = await copy_unit( client, src, unit, finals, st, reply_to, dst, from_saved=from_saved, no_forward=no_forward ) await settle_buttons(client, dst, finals, news, st, pause=settle_pause) 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 job.record(_rows(unit, finals, news)) 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, dst, [n.id for n in news]) res.failed.append((ids[0], type(e).__name__)) res.aborted = type(e).__name__ break streak = 0 job.done.update(ids) for old, new in zip(unit, news): job.idmap[old.id] = new.id if reply_parent(unit[0]) is not None and reply_to is None: res.reply_dropped += 1 if why and not from_saved: res.live_restricted += 1 res.files_left_out += left_out if why or res.copied_units < CHECK_FIRST or res.copied_units % CHECK_EVERY == 0: await check_copies(client, dst, unit, news, res) 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) if job.stop_if_hidden and res.checked >= job.stop_if_hidden and len(res.hidden) == res.checked: log.info("Telegram hides every copy read back (%s): stopping", res.checked) res.stopped = res.hidden_stop = True # more copies would only be more posts that nobody can see break await asyncio.sleep(delay) if res.hidden: # Telegram hides copies: is the destination itself restricted too? try: _top, res.dst_restricted, _err = await probe_channel(client, dst) except Exception: # only an extra for the report log.debug("could not ask Telegram about the destination channel", exc_info=True) res.forwarded, res.rebuilt, res.repaired = st.forwarded, st.rebuilt, st.repaired res.replies, res.reply_lost = st.replies, st.reply_lost res.from_saved, res.media_lost = st.from_saved, st.media_lost return res 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, settle_pause: float = 0.7, ) -> RepostResult: """Copy the posts of migration `mig` (ids first_id..last_id) in order; safe to run again to continue. When every post is copied (not stopped, no fatal error) the finishing touches follow: links to posts that were copied later are pointed at their copies, the copies of pinned posts are pinned, and then every connected channel is searched for links to the old posts (crosslinks.relink_after_repost).""" opts = RepostOptions( old=mig.old_username, new=mig.new_username, include_typed=mig.include_typed, include_posts=mig.include_posts ) peer = peer_of(ch) idmap = dict(await db.migration_pairs(mid=mig.id, alive_only=False)) job = CopyJob( src_peer=peer, dst_peer=peer, opts=opts, links=PostLinks.for_channel(ch, idmap), first_id=mig.first_id, last_id=mig.last_id, done=await db.migration_done_old_ids(mig.id), idmap=idmap, record=lambda rows: db.record_repost(mig.id, ch.id, rows, user_id), saved=saved_lookup(db, ch.id), # a post Telegram holds back is left alone (and so is not deleted) ) res = await copy_loop(client, job, delay=delay, progress=progress, should_stop=should_stop, settle_pause=settle_pause) if not res.stopped and not res.aborted: res.phase = "finish" if progress: await progress(res) try: res.final = await finalize_repost( client, db, ch, mig, delay=delay, settle_pause=settle_pause, should_stop=should_stop ) except Exception as e: # the copies are fine; only the finishing touches failed log.exception("finishing the repost (links, pins) failed") res.final_error = type(e).__name__ if not (res.final is not None and res.final.stopped): res.phase = "links" res.cross = CrossResult() async def watch(_cross) -> None: if progress: await progress(res) if progress: await progress(res) try: await relink_after_repost( client, db, ch, mig, delay=delay, settle_pause=settle_pause, progress=watch, should_stop=should_stop, result=res.cross, ) except Exception as e: # the copies are fine; only the links in the other channels were not updated log.exception("pointing the links of the other channels at the copies failed") res.cross_error = type(e).__name__ return res # --------------------------------------------------------------------- links to copies, pins async def relink_copies( client, db, ch, ids: dict, targets: list, *, delay: float = 1.2, settle_pause: float = 0.7, dry_run: bool = False, progress: Optional[Callable[[FinalizeResult], Awaitable[None]]] = None, should_stop: Optional[Callable[[], bool]] = None, links: Optional[PostLinks] = None, ) -> FinalizeResult: """Point every link to a post of this channel at the copy of that post (`ids`: old message id -> new one). `targets` are the message ids of `ch` to look at (the copies). Idempotent: a link that already points at a copy is left alone, because only OLD ids are replaced. With dry_run nothing is edited, the result only counts. `links` (optional) says which links to follow; the default is "links to posts of `ch` itself" (/repost).""" res = FinalizeResult() peer = peer_of(ch) links = links if links is not None else PostLinks.for_channel(ch, ids) targets = sorted(set(targets)) streak, last_name = 0, None for i in range(0, len(targets), CHUNK): chunk = targets[i : i + CHUNK] for mid, m in zip(chunk, await fetch_messages(client, peer, chunk)): if is_skippable(m): res.missing += 1 continue res.checked += 1 rl = relink_message(m, links) if rl is None: continue if dry_run: res.relinked += 1 res.links += rl.links continue if should_stop and should_stop(): res.stopped = True return res rich = is_rich_message(m) # a rich message's text is blocks: only its buttons change, the text is not sent again try: if rich: await flood_retry(lambda: edit_raw(client, peer, m.id, markup=rl.markup)) else: await flood_retry( lambda: edit_raw( client, peer, m.id, text=rl.text, entities=rl.entities, markup=rl.markup, preview=preview_flag(m) ) ) if link_urls(rl.markup): await ensure_markup(client, peer, m.id, rl.markup, pause=settle_pause) except errors.MessageNotModifiedError: continue except Exception as e: # one bad post must not stop the others name = type(e).__name__ log.warning("could not update the links of %s: %s %s", m.id, name, e) res.failed.append((m.id, name)) streak = streak + 1 if name == last_name else 1 last_name = name if name in FATAL or streak >= ABORT_AFTER: res.aborted = name return res continue streak, last_name = 0, None res.relinked += 1 res.links += rl.links try: saved = dict(buttons=ser_markup(rl.markup)) if not rich: saved.update(text=rl.text, entities=ser_entities(rl.entities)) await db.update_post_by_message(ch.id, m.id, **saved) except Exception: # the channel is right; the saved copy will be corrected by the next sync log.warning("could not save the new links of %s", m.id, exc_info=True) if progress: await progress(res) await asyncio.sleep(delay) return res async def pin_copies(client, ch, pairs: list, res: FinalizeResult, *, delay: float = 1.2) -> None: """Pin the copy of every post that is pinned among the originals (oldest first, so the newest pin is on top). The "pinned a message" notices Telegram adds to the channel are deleted again right away.""" peer = peer_of(ch) pairs = sorted(pairs) todo: list = [] for i in range(0, len(pairs), CHUNK): chunk = pairs[i : i + CHUNK] olds = await fetch_messages(client, peer, [o for o, _ in chunk]) news = await fetch_messages(client, peer, [n for _, n in chunk]) for (_, new_id), src, cur in zip(chunk, olds, news): if is_skippable(src) or is_skippable(cur): continue if getattr(src, "pinned", False) and not getattr(cur, "pinned", False): todo.append(new_id) res.pin_wanted = len(todo) for new_id in todo: try: out = await flood_retry( lambda: client(functions.messages.UpdatePinnedMessageRequest(peer=peer, id=new_id, silent=True)) ) except errors.MessageNotModifiedError: continue # already pinned except errors.RPCError as e: res.pin_error = type(e).__name__ return res.pinned += 1 await delete_ids(client, peer, [m.id for m in new_messages_from(out) if isinstance(m, types.MessageService)]) await asyncio.sleep(delay) async def finalize_repost( client, db, ch, mig, *, delay: float = 1.2, settle_pause: float = 0.7, progress: Optional[Callable[[FinalizeResult], Awaitable[None]]] = None, should_stop: Optional[Callable[[], bool]] = None, ) -> FinalizeResult: """After the copies exist: links that point at old posts get the copies' ids, copies of pinned posts get pinned.""" pairs = sorted(await db.migration_pairs(mid=mig.id)) res = await relink_copies( client, db, ch, dict(pairs), [n for _, n in pairs], delay=delay, settle_pause=settle_pause, progress=progress, should_stop=should_stop, ) if not res.stopped and not res.aborted: await pin_copies(client, ch, pairs, res, delay=delay) return res # ----------------------------------------------------------------------------- deleting @dataclass class DeleteResult: deleted: int = 0 by_userbot: int = 0 # of those, how many only the userbot could delete remaining: int = 0 blocked: bool = False # some messages are still there although Telegram did not report an error error: Optional[str] = None # the bot's Telegram error that stopped the run fallback_error: Optional[str] = None # why the userbot could not help tried_userbot: bool = False async def gone_ids(client, peer, ids: list) -> set: """Which of `ids` no longer exist in the channel.""" msgs = await fetch_messages(client, peer, ids) return {i for i, m in zip(ids, msgs) if m is None or isinstance(m, types.MessageEmpty)} @dataclass class BatchOutcome: gone: set = field(default_factory=set) # message ids that no longer exist afterwards bot_error: Optional[str] = None by_userbot: int = 0 fallback_error: Optional[str] = None tried_userbot: bool = False async def delete_batch(client, peer, ids: list, fallback: Optional[Callable[[list], Awaitable[Any]]] = None) -> BatchOutcome: """Delete up to 100 messages: the bot tries first, the result is checked by reading the messages back, and whatever is still there goes to `fallback(ids)` (the userbot), followed by a second check.""" out = BatchOutcome() try: await flood_retry(lambda: client(functions.channels.DeleteMessagesRequest(channel=peer, id=list(ids)))) except errors.RPCError as e: out.bot_error = type(e).__name__ out.gone = await gone_ids(client, peer, ids) left = [x for x in ids if x not in out.gone] if left and fallback is not None: out.tried_userbot = True try: await fallback(left) except Exception as e: # a userbot problem must not hide what the bot already did log.warning("the userbot could not delete %s message(s): %s", len(left), e) out.fallback_error = str(e) or type(e).__name__ again = await gone_ids(client, peer, left) out.by_userbot = len(again) out.gone |= again return out async def delete_pending( client, peer, todo: list, mark: Callable[[list], Awaitable[None]], *, delay: float = 0.5, progress: Optional[Callable[[DeleteResult], Awaitable[None]]] = None, fallback: Optional[Callable[[list], Awaitable[Any]]] = None, ) -> DeleteResult: """Delete the messages `todo` (oldest first, 100 per request). `mark(ids)` records the ones that are gone. The bot tries first and the result is always checked by reading the messages back. Whatever the bot could not delete (Telegram may refuse old posts) goes to `fallback(ids)` - the userbot - if there is one, and is checked again. A batch the bot cannot delete does not stop the run: newer posts may still be deletable.""" res = DeleteResult() use_fallback = fallback for i in range(0, len(todo), 100): ids = todo[i : i + 100] out = await delete_batch(client, peer, ids, use_fallback) res.tried_userbot = res.tried_userbot or out.tried_userbot res.by_userbot += out.by_userbot if out.fallback_error: res.fallback_error = out.fallback_error use_fallback = None # it will not work for the next batches either left = [x for x in ids if x not in out.gone] if out.gone: await mark(sorted(out.gone)) res.deleted += len(out.gone) if left and out.bot_error: res.error = res.error or out.bot_error # remembered for the report if out.bot_error in FATAL and use_fallback is None: break # e.g. no "Delete messages" right: the next batch would fail the same way if progress: await progress(res) await asyncio.sleep(delay) return res async def delete_copies( client, db, ch, mig, which: str, *, delay: float = 0.5, progress: Optional[Callable[[DeleteResult], Awaitable[None]]] = None, fallback: Optional[Callable[[list], Awaitable[Any]]] = None, ) -> DeleteResult: """which='old': remove the original posts; which='new': remove the copies (undo). Oldest first, 100 per request.""" todo = await db.migration_pending_ids(mig.id, which, 1_000_000) async def mark(ids: list) -> None: await db.mark_migration_deleted(mig.id, ch.id, which, ids) res = await delete_pending(client, peer_of(ch), todo, mark, delay=delay, progress=progress, fallback=fallback) counts = await db.migration_counts(mig.id) res.remaining = counts["old_left"] if which == "old" else counts["new_left"] res.blocked = res.remaining > 0 and not res.error return res