Spaces:
Paused
Paused
| """Links that point at other posts of the same channel (t.me/name/115, t.me/c/123456/115, tg://...&post=115). | |
| When a channel is reposted its posts get new message ids, so a link to post 115 has to point at the copy of post 115. | |
| Only the digits of the message id change; everything around them (?single, ?comment=3, the visible text) stays. | |
| Links can sit in three places: a hyperlink behind text, a link typed into the text, and a button. | |
| """ | |
| from __future__ import annotations | |
| import copy | |
| import re | |
| from dataclasses import dataclass, field | |
| from typing import Optional | |
| from telethon import types | |
| from telethon.helpers import add_surrogate, del_surrogate | |
| from .linkswap import apply_span_replacements | |
| from .richmsg import is_rich_message | |
| from .tgutil import button_url, with_url | |
| _HOST = r"(?:t\.me|telegram\.me|telegram\.dog)" | |
| _NAME = r"[A-Za-z][A-Za-z0-9_]{2,31}" | |
| # https://t.me/name/115 https://t.me/c/123456/115 (also t.me/s/name/115, the web preview) | |
| _WEB = re.compile( | |
| rf"^(?:https?://)?(?:www\.)?{_HOST}/(?:s/)?(?P<seg>c/(?P<cid>\d+)|(?P<name>{_NAME}))/(?P<id>\d+)(?![A-Za-z0-9_])", re.I | |
| ) | |
| _TYPED = re.compile( | |
| rf"(?<![\w@./=?&#-])(?:https?://)?(?:www\.)?{_HOST}/(?:s/)?(?:c/\d+|{_NAME})/\d+(?![A-Za-z0-9_])", re.I | |
| ) | |
| # tg://resolve?domain=name&post=115 tg://privatepost?channel=123456&post=115 | |
| _TG = re.compile(r"^tg://(?P<kind>resolve|privatepost)\?(?P<query>[^#]*)", re.I) | |
| _POST_PARAM = re.compile(r"(?<=[?&])post=(\d+)") | |
| class Ref: | |
| """A link to a post: which channel (username in lower case, or the numeric id) and where the id's digits sit.""" | |
| name: Optional[str] | |
| cid: Optional[int] | |
| msg_id: int | |
| start: int # the digits of the message id ... | |
| end: int | |
| seg_start: Optional[int] = None # ... and the channel part in front of them ("name" or "c/123"); web links only | |
| def parse_ref(url: str) -> Optional[Ref]: | |
| """What `url` points at, if it is a link to a post of some channel; None for every other link.""" | |
| url = url or "" | |
| m = _WEB.match(url) | |
| if m: | |
| return Ref( | |
| name=m.group("name").lower() if m.group("name") else None, | |
| cid=int(m.group("cid")) if m.group("cid") else None, | |
| msg_id=int(m.group("id")), | |
| start=m.start("id"), | |
| end=m.end("id"), | |
| seg_start=m.start("seg"), | |
| ) | |
| m = _TG.match(url) | |
| if m: | |
| params = dict(p.split("=", 1) for p in m.group("query").split("&") if "=" in p) | |
| post = _POST_PARAM.search(url) | |
| if post is None: | |
| return None | |
| if m.group("kind").lower() == "resolve" and re.fullmatch(r"[A-Za-z0-9_]+", params.get("domain", "")): | |
| return Ref(params["domain"].lower(), None, int(post.group(1)), post.start(1), post.end(1)) | |
| if m.group("kind").lower() == "privatepost" and params.get("channel", "").isdigit(): | |
| return Ref(None, int(params["channel"]), int(post.group(1)), post.start(1), post.end(1)) | |
| return None | |
| class PostLinks: | |
| """The links that point into one channel, and where its posts went: `ids` maps old message id -> new one. | |
| By default the copies are in the same channel (/repost): only the message id changes. With `target_id` the | |
| copies are in another channel (/shift): the channel part of the link changes too.""" | |
| username: Optional[str] | |
| channel_id: int | |
| ids: dict = field(default_factory=dict) | |
| target_username: Optional[str] = None | |
| target_id: Optional[int] = None | |
| def for_channel(cls, ch, ids: Optional[dict] = None) -> "PostLinks": | |
| return cls(username=(ch.username or None), channel_id=int(ch.id), ids=ids if ids is not None else {}) | |
| # -------------------------------------------------------------------------------------- one url | |
| def ref_of(self, url: str) -> Optional[Ref]: | |
| """The reference in `url` if it points at a post of THIS channel.""" | |
| r = parse_ref(url) | |
| if r is None: | |
| return None | |
| if r.name is not None and self.username and r.name == self.username.lower(): | |
| return r | |
| if r.cid is not None and r.cid == self.channel_id: | |
| return r | |
| return None | |
| def edit_of(self, url: str) -> Optional[tuple]: | |
| """(start, end, new text) to apply to `url` so it points at the copy of its post; None if it stays as it is.""" | |
| r = self.ref_of(url) | |
| if r is None or r.msg_id not in self.ids: | |
| return None | |
| new_id = str(self.ids[r.msg_id]) | |
| if self.target_id is None: | |
| return r.start, r.end, new_id | |
| if r.seg_start is None: # tg:// links are not moved to another channel | |
| return None | |
| channel = self.target_username or f"c/{self.target_id}" | |
| return r.seg_start, r.end, f"{channel}/{new_id}" | |
| def remap_url(self, url: str) -> str: | |
| e = self.edit_of(url) | |
| return url if e is None else url[: e[0]] + e[2] + url[e[1] :] | |
| # --------------------------------------------------------------------------------- whole message | |
| def count(self, text: str, entities, markup) -> int: | |
| """How many links in a message point at a post of this channel (whether or not the post has a new id).""" | |
| n = 0 | |
| sur = add_surrogate(text or "") | |
| for m in _TYPED.finditer(sur): | |
| if self.ref_of(m.group(0)): | |
| n += 1 | |
| for e in entities or []: | |
| if isinstance(e, types.MessageEntityTextUrl) and self.ref_of(e.url): | |
| n += 1 | |
| if isinstance(markup, types.ReplyInlineMarkup): | |
| for row in markup.rows: | |
| for b in row.buttons: | |
| u = button_url(b) | |
| if u is not None and self.ref_of(u): | |
| n += 1 | |
| return n | |
| def remap_text(self, text: str, entities) -> tuple: | |
| """-> (text, entities, typed_links_changed, hyperlinks_changed). Offsets of all entities stay right.""" | |
| ents = list(entities or []) | |
| sur = add_surrogate(text or "") | |
| repls = [] | |
| for m in _TYPED.finditer(sur): | |
| e = self.edit_of(m.group(0)) | |
| if e is not None: | |
| repls.append((m.start() + e[0], m.start() + e[1], e[2])) | |
| if repls: | |
| new_sur, new_ents = apply_span_replacements(sur, ents, repls) | |
| else: | |
| new_sur, new_ents = sur, [copy.copy(e) for e in ents] | |
| hyper = 0 | |
| for e in new_ents: | |
| if isinstance(e, types.MessageEntityTextUrl): | |
| new_url = self.remap_url(e.url) | |
| if new_url != e.url: | |
| e.url = new_url | |
| hyper += 1 | |
| return del_surrogate(new_sur), new_ents, len(repls), hyper | |
| def remap_markup(self, markup) -> tuple: | |
| """-> (markup, buttons_changed); the keyboard object is only rebuilt when a button changes.""" | |
| if not isinstance(markup, types.ReplyInlineMarkup): | |
| return markup, 0 | |
| rows, changed = [], 0 | |
| for row in markup.rows: | |
| btns = [] | |
| for b in row.buttons: | |
| u = button_url(b) | |
| new = self.remap_url(u) if u is not None else None | |
| if new is not None and new != u: | |
| btns.append(with_url(b, new)) | |
| changed += 1 | |
| else: | |
| btns.append(b) | |
| rows.append(types.KeyboardButtonRow(btns)) | |
| return (types.ReplyInlineMarkup(rows) if changed else markup), changed | |
| class Relinked: | |
| """A message whose links to other posts of the channel were pointed at the copies.""" | |
| text: str | |
| entities: list | |
| markup: Optional[types.ReplyInlineMarkup] | |
| text_changed: bool # the text or its entities differ | |
| markup_changed: bool | |
| n_typed: int | |
| n_hyper: int | |
| n_buttons: int | |
| def links(self) -> int: | |
| return self.n_typed + self.n_hyper + self.n_buttons | |
| def relink_message(msg, links: PostLinks) -> Optional[Relinked]: | |
| """The new text / entities / buttons of `msg` once its links are remapped; None when nothing changes.""" | |
| if is_rich_message(msg): # the text of a rich message is blocks: only its buttons are changed | |
| text, entities, n_typed, n_hyper = "", [], 0, 0 | |
| else: | |
| text, entities, n_typed, n_hyper = links.remap_text(msg.message or "", msg.entities) | |
| markup, n_buttons = links.remap_markup(msg.reply_markup) | |
| if not (n_typed or n_hyper or n_buttons): | |
| return None | |
| return Relinked( | |
| text=text, | |
| entities=entities, | |
| markup=markup, | |
| text_changed=bool(n_typed or n_hyper), | |
| markup_changed=bool(n_buttons), | |
| n_typed=n_typed, | |
| n_hyper=n_hyper, | |
| n_buttons=n_buttons, | |
| ) | |
| def resolve_chain(step: dict) -> dict: | |
| """Several reposts in a row: {1: 5, 5: 9} -> {1: 9, 5: 9} (a post that moved twice points at its latest copy).""" | |
| out = {} | |
| for old in step: | |
| cur, seen = old, 0 | |
| while cur in step and seen < 100: | |
| cur, seen = step[cur], seen + 1 | |
| out[old] = cur | |
| return out | |