Spaces:
Paused
Paused
| """/shift: copy posts of one channel into another channel - text, formatting, media, albums, buttons and replies | |
| - and register the copies in My posts. The source is only read, never changed or deleted. | |
| It is the /repost copy loop pointed at two channels. Links from a shifted post to another post of the source channel | |
| follow the content: when that post was shifted too, the link points at its copy in the destination. | |
| Three optional extras are remembered with the shift (job marks), so that continuing a stopped shift keeps them: | |
| -c afterwards every link to a post of the source, in ANY connected channel, is pointed at the copy | |
| (crosslinks.relink_after_shift; the source itself is only read); | |
| -all afterwards the destination gets the source's name, description and profile photo | |
| (profile_copy.copy_profile); | |
| --no-media the copies get no files: every post is copied as its text, with its formatting and buttons (mark "nomedia"; | |
| repost_engine.without_files). A post that is only a file has nothing to copy and is left out. | |
| When Telegram hides the first new copies (read back right after posting) the shift stops by itself - more copies would | |
| only be more posts nobody can see - and goes on only when the owner presses Continue (mark "hiddenok"). | |
| A source that Telegram restricts (a copyright strike: "This channel is unavailable due to copyright infringement", its | |
| posts marked or shown as a notice) is copied from what My posts saved of its posts - every post that has a saved copy, | |
| however intact Telegram's version looks (repost_engine.copy_source). A post without a saved copy is copied as Telegram | |
| hands it out if that is the post itself, and skipped if it is only the notice. A source that can't be read at all (a | |
| banned or private channel) is copied from My posts alone: mark "fromdb", set when the plan was made that way. | |
| """ | |
| from __future__ import annotations | |
| import logging | |
| from types import SimpleNamespace | |
| from typing import Any, Awaitable, Callable, Optional | |
| from .crosslinks import CrossResult, relink_after_shift, shift_links | |
| from .profile_copy import copy_profile | |
| from .repost_engine import ( | |
| STOP_IF_HIDDEN, | |
| CopyJob, | |
| DeleteResult, | |
| RepostOptions, | |
| RepostPlan, | |
| RepostResult, | |
| copy_loop, | |
| delete_pending, | |
| plan_from_saved, | |
| relink_copies, | |
| saved_lookup, | |
| saved_units, | |
| ) | |
| from .restrictions import probe_channel | |
| from .tgutil import peer_of | |
| log = logging.getLogger(__name__) | |
| def source_of(shift) -> SimpleNamespace: | |
| """The source channel as stored with the shift (it need not be registered in the bot).""" | |
| return SimpleNamespace( | |
| id=shift.src_channel_id, access_hash=shift.src_access_hash, username=shift.src_username, title=shift.src_title | |
| ) | |
| async def plan_from_my_posts( | |
| db, src, first_id=None, last_id=None, unreadable=None, no_media: bool = False | |
| ) -> Optional[RepostPlan]: | |
| """The plan of a shift made from My posts alone, for a source that can't be read; None when My posts has nothing | |
| of that channel (in that range).""" | |
| plan = plan_from_saved( | |
| src, await db.sent_posts_from(src.id, first_id or 1), RepostOptions(), first_id=first_id, last_id=last_id, | |
| unreadable=unreadable, no_media=no_media, | |
| ) | |
| return None if plan.error else plan | |
| async def run_shift( | |
| client, | |
| db, | |
| shift, | |
| dst, | |
| user_id: int, | |
| *, | |
| delay: float = 1.2, | |
| settle_pause: float = 0.7, | |
| progress: Optional[Callable[[RepostResult], Awaitable[None]]] = None, | |
| should_stop: Optional[Callable[[], bool]] = None, | |
| ) -> RepostResult: | |
| """Copy the posts shift.first_id..shift.last_id of the source into `dst` (a registered channel), in order. | |
| Safe to run again to continue: posts that already have a copy are skipped.""" | |
| src = source_of(shift) | |
| idmap = dict(await db.shift_pairs(shift.id, alive_only=False)) | |
| links = shift_links(src, dst, idmap) | |
| marks = await db.marks_of(shift.id) | |
| from_db = "fromdb" in marks # the source can't be read: its posts come from My posts alone | |
| restricted = None | |
| if not from_db: # does Telegram restrict the source channel itself (a copyright strike)? Then My posts is used | |
| try: | |
| _top, restricted, _err = await probe_channel(client, peer_of(src)) | |
| except Exception: # unknown is treated as "not restricted": marks on single posts and the read-back still guard | |
| log.warning("could not ask Telegram whether the source channel is restricted", exc_info=True) | |
| job = CopyJob( | |
| src_peer=peer_of(src), | |
| dst_peer=peer_of(dst), | |
| opts=RepostOptions(), | |
| links=links, | |
| first_id=shift.first_id, | |
| last_id=shift.last_id, | |
| done=await db.shift_done_ids(shift.id), | |
| idmap=idmap, | |
| record=lambda rows: db.record_shift(shift.id, dst.id, rows, user_id), | |
| saved=saved_lookup(db, src.id), # posts Telegram holds back (a copyright strike) are copied from My posts | |
| use_saved=True, | |
| units=(lambda: saved_units(db, src.id, shift.first_id, shift.last_id)) if from_db else None, | |
| from_db=from_db, | |
| restricted=restricted, | |
| no_media="nomedia" in marks, | |
| stop_if_hidden=0 if "hiddenok" in marks else STOP_IF_HIDDEN, # the owner said to go on although Telegram hides them | |
| ) | |
| 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: | |
| pairs = await db.shift_pairs(shift.id) | |
| res.final = await relink_copies( | |
| client, db, dst, dict(pairs), [n for _, n in pairs], links=links, delay=delay, | |
| settle_pause=settle_pause, should_stop=should_stop, | |
| ) | |
| except Exception as e: # the copies are fine; only the links between them were not updated | |
| log.exception("pointing the links of a shift at the copies failed") | |
| res.final_error = type(e).__name__ | |
| marks = await db.marks_of(shift.id) | |
| if "profile" in marks and "profile_done" not in marks: | |
| await shift_profile(client, db, shift, dst, user_id, res, progress) | |
| if "relink" in marks and 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_shift( | |
| client, db, shift, src, dst, 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 shifted posts failed") | |
| res.cross_error = type(e).__name__ | |
| return res | |
| async def shift_profile(client, db, shift, dst, user_id: int, res: RepostResult, progress=None) -> None: | |
| """-all: the destination takes over the source's name, description and photo (the outcome goes into `res`).""" | |
| res.phase = "profile" | |
| if progress: | |
| await progress(res) | |
| try: | |
| res.profile = await copy_profile(client, source_of(shift), dst) | |
| if res.profile.title: # the destination has a new name: keep the bot's own list right | |
| await db.save_channel(dst.id, dst.access_hash, res.profile.title, dst.username, user_id) | |
| if res.profile.ok: | |
| await db.set_mark(shift.id, "profile_done") | |
| except Exception as e: # the copies are fine; only the channel's name / description / photo were not copied | |
| log.exception("copying the channel profile failed") | |
| res.profile_error = type(e).__name__ | |
| async def delete_shift_copies( | |
| client, | |
| db, | |
| dst, | |
| shift, | |
| *, | |
| delay: float = 0.5, | |
| progress: Optional[Callable[[DeleteResult], Awaitable[None]]] = None, | |
| fallback: Optional[Callable[[list], Awaitable[Any]]] = None, | |
| ) -> DeleteResult: | |
| """Undo: remove the copies of this shift from the destination channel (the source is not touched).""" | |
| todo = await db.shift_pending_new(shift.id, 1_000_000) | |
| async def mark(ids: list) -> None: | |
| await db.mark_shift_deleted(shift.id, dst.id, ids) | |
| res = await delete_pending(client, peer_of(dst), todo, mark, delay=delay, progress=progress, fallback=fallback) | |
| res.remaining = (await db.shift_counts(shift.id))["new_left"] | |
| res.blocked = res.remaining > 0 and not res.error | |
| return res | |