Spaces:
Paused
Paused
| """Telegram rich messages: headings, lists, tables, details, math, maps, media blocks, buttons ... | |
| A rich message is not text plus entities: it is a tree of blocks. Telegram builds that tree itself from HTML or Markdown | |
| (InputRichMessageHTML / InputRichMessageMarkdown), so text typed in those languages is handed over unchanged - every block | |
| and every formatting Telegram knows is supported without this bot having to know it. A rich message that is sent to the | |
| bot natively (or forwarded to it) is kept block for block (InputRichMessage). | |
| How the bot keeps one: as a single entry in the post's `entities` list - ``{"k": "rich", "f": "html" | "markdown" | | |
| "blocks", "s": source, ...}`` - next to a plain-text version in `text` (for lists, panels and the like). Whatever sends or | |
| edits a post looks for that entry with `rich_of`. No database change is needed, and exports, backups and the undo log carry | |
| it like any other formatting. | |
| """ | |
| from __future__ import annotations | |
| import asyncio | |
| import copy | |
| import hashlib | |
| import logging | |
| import re | |
| from dataclasses import dataclass, replace | |
| from html.parser import HTMLParser | |
| from types import SimpleNamespace | |
| from typing import List, Optional | |
| from telethon import errors, functions, types, utils | |
| from telethon.tl.tlobject import TLObject | |
| from .tgutil import _b64, _unb64, flood_retry | |
| log = logging.getLogger(__name__) | |
| KIND = "rich" # the "k" of the entry in a post's entities | |
| MAX_SOURCE = 32768 # characters Telegram allows in a rich message (500 blocks, 16 levels and 50 media as well) | |
| class RichError(ValueError): | |
| """A rich message that can't be used; the text is safe to show to the owner.""" | |
| class Rich: | |
| """A rich message as the bot keeps it.""" | |
| fmt: str # "html" | "markdown" (typed: Telegram reads it) | "blocks" (received: base64 of an InputRichMessage) | |
| source: str | |
| rtl: bool = False | |
| noautolink: bool = False # links, @mentions, #tags ... in the text are not made clickable by Telegram | |
| fp: Optional[str] = None # fingerprint of the blocks Telegram showed last time (what the sync compares) | |
| def entry(r: Rich) -> dict: | |
| """The entry that stands for `r` in a post's entities.""" | |
| e: dict = {"k": KIND, "f": r.fmt, "s": r.source} | |
| if r.rtl: | |
| e["r"] = 1 | |
| if r.noautolink: | |
| e["n"] = 1 | |
| if r.fp: | |
| e["p"] = r.fp | |
| return e | |
| def rich_of(obj) -> Optional[Rich]: | |
| """The rich message a post (or anything with a post's `entities`) holds, or None.""" | |
| for d in getattr(obj, "entities", None) or []: | |
| if isinstance(d, dict) and d.get("k") == KIND and d.get("f") in ("html", "markdown", "blocks") and d.get("s"): | |
| return Rich(d["f"], d["s"], bool(d.get("r")), bool(d.get("n")), d.get("p") or None) | |
| return None | |
| def is_rich(obj) -> bool: | |
| return rich_of(obj) is not None | |
| def real_entities(entities) -> list: | |
| """The formatting entries of a stored entities list, without the rich message's.""" | |
| return [d for d in entities or [] if not (isinstance(d, dict) and d.get("k") == KIND)] | |
| # ------------------------------------------------------------------------------------ what Telegram takes | |
| def to_input(r: Rich): | |
| """The InputRichMessage* object to send for `r`.""" | |
| flags = dict(rtl=True if r.rtl else None, noautolink=True if r.noautolink else None) | |
| if r.fmt == "html": | |
| return types.InputRichMessageHTML(html=r.source, **flags) | |
| if r.fmt == "markdown": | |
| return types.InputRichMessageMarkdown(markdown=r.source, **flags) | |
| inp = copy.copy(_unb64(r.source)) | |
| # (a message saved by an older version may still hold blocks in Telegram's own form) | |
| inp.blocks = inputify(inp.blocks, _user_ids(inp.users)) | |
| return inp | |
| def _input_geo(geo): | |
| if isinstance(geo, types.GeoPoint): | |
| return types.InputGeoPoint(lat=geo.lat, long=geo.long, accuracy_radius=geo.accuracy_radius) | |
| if isinstance(geo, types.InputGeoPoint): | |
| return geo | |
| return types.InputGeoPointEmpty() | |
| def _user_ids(users) -> set: | |
| return {u.user_id for u in users or [] if isinstance(getattr(u, "user_id", None), int)} | |
| def inputify(o, users: Optional[set] = None): | |
| """Blocks the way Telegram wants them SENT. What it hands out and what it takes are the same blocks except where a | |
| block holds something only Telegram can make: a map holds a stored point (with an access hash) when it comes in, and | |
| must hold an input point (latitude and longitude) when it goes out. | |
| `users`: the ids of the accounts that the message may mention (it carries their access data). A text mention of | |
| anybody else would be refused, so it stays as the plain name; None leaves all mentions alone. | |
| A thinking block (the "thinking ..." part of a message that is being streamed) is for drafts only - the Bot API says so | |
| of <tg-thinking> - and a message that is sent can't hold one: it goes out as an expandable section called "Thinking".""" | |
| if isinstance(o, types.PageBlockMap): | |
| return types.InputPageBlockMap( | |
| geo=_input_geo(o.geo), zoom=o.zoom, w=o.w, h=o.h, caption=inputify(o.caption, users) | |
| ) | |
| if isinstance(o, types.PageBlockThinking): | |
| return types.PageBlockDetails( | |
| blocks=[types.PageBlockParagraph(inputify(o.text, users))], title=types.TextPlain("Thinking") | |
| ) | |
| if users is not None and isinstance(o, types.TextMentionName) and o.user_id not in users: | |
| return inputify(o.text, users) | |
| if isinstance(o, list): | |
| return [inputify(x, users) for x in o] | |
| if isinstance(o, TLObject): | |
| new = copy.copy(o) | |
| for key, value in vars(o).items(): | |
| if isinstance(value, (list, TLObject)): | |
| setattr(new, key, inputify(value, users)) | |
| return new | |
| return o | |
| def blank_references(inp): | |
| """The same InputRichMessage with the file references of its photos and files left out (they may have expired).""" | |
| if not isinstance(inp, types.InputRichMessage): | |
| return inp | |
| out = copy.copy(inp) | |
| out.photos = [types.InputPhoto(p.id, p.access_hash, b"") for p in inp.photos or []] or None | |
| out.documents = [types.InputDocument(d.id, d.access_hash, b"") for d in inp.documents or []] or None | |
| return out | |
| async def with_rich(r: Rich, call): | |
| """call(InputRichMessage*): first as saved, then (if Telegram says a saved file reference expired) without them.""" | |
| try: | |
| return await call(to_input(r)) | |
| except errors.RPCError as e: | |
| if r.fmt != "blocks" or "FileReference" not in type(e).__name__: | |
| raise | |
| log.info("stored file reference of a rich message rejected (%s) - retrying without it", type(e).__name__) | |
| return await call(blank_references(to_input(r))) | |
| def sent_message(result): | |
| """The message a send request created (something with an `id`).""" | |
| for u in getattr(result, "updates", None) or []: | |
| if isinstance(u, (types.UpdateNewChannelMessage, types.UpdateNewMessage)): | |
| return u.message | |
| if getattr(result, "id", None) is not None and not isinstance(result, (list, tuple)): | |
| return SimpleNamespace(id=result.id) # UpdateShortSentMessage | |
| raise RuntimeError("Telegram did not say which message it created") | |
| async def send_rich(client, peer, r: Rich, *, markup=None, silent=None): | |
| """Send `r` as a new message to `peer`; the message that was created comes back.""" | |
| result = await with_rich( | |
| r, | |
| lambda inp: client( | |
| functions.messages.SendMessageRequest( | |
| peer=peer, message="", rich_message=inp, reply_markup=markup, silent=True if silent else None | |
| ) | |
| ), | |
| ) | |
| return sent_message(result) | |
| # --------------------------------------------------------------------------------- rich messages that are received | |
| def is_rich_message(m) -> bool: | |
| """The message carries a rich message (blocks) instead of plain text.""" | |
| return getattr(m, "rich_message", None) is not None | |
| def fingerprint(rm) -> str: | |
| """Identifies what a rich message shows (its blocks; the file references of its pictures change from fetch to fetch).""" | |
| h = hashlib.sha1() | |
| for b in getattr(rm, "blocks", None) or []: | |
| h.update(bytes(inputify(b))) # (a map's access hash is not part of what it shows) | |
| h.update(b"r" if getattr(rm, "rtl", None) else b"-") | |
| return h.hexdigest()[:20] | |
| def _input_files(items, getter, empty) -> list: | |
| out = [] | |
| for item in items or []: | |
| try: | |
| inp = getter(item) | |
| except (TypeError, ValueError): | |
| continue | |
| if not isinstance(inp, empty): | |
| out.append(inp) | |
| return out | |
| def input_of(rm, users: Optional[list] = None): | |
| """The InputRichMessage that sends a received RichMessage again: same blocks, same pictures and files.""" | |
| photos = _input_files(getattr(rm, "photos", None), utils.get_input_photo, types.InputPhotoEmpty) | |
| docs = _input_files(getattr(rm, "documents", None), utils.get_input_document, types.InputDocumentEmpty) | |
| return types.InputRichMessage( | |
| blocks=inputify(list(rm.blocks), _user_ids(users)), | |
| rtl=True if getattr(rm, "rtl", None) else None, | |
| photos=photos or None, | |
| documents=docs or None, | |
| users=users or None, | |
| ) | |
| def rich_from_received(rm, users: Optional[list] = None) -> Rich: | |
| return Rich("blocks", _b64(input_of(rm, users)), bool(getattr(rm, "rtl", None)), False, fingerprint(rm)) | |
| def mention_ids(blocks) -> list: | |
| """User ids that the blocks mention by link (a text mention needs the account's access data to be sent again).""" | |
| found: list = [] | |
| def walk(o) -> None: | |
| if isinstance(o, dict): | |
| if o.get("_") == "TextMentionName" and isinstance(o.get("user_id"), int): | |
| found.append(o["user_id"]) | |
| for v in o.values(): | |
| walk(v) | |
| elif isinstance(o, list): | |
| for v in o: | |
| walk(v) | |
| for b in blocks or []: | |
| walk(b.to_dict()) | |
| return list(dict.fromkeys(found)) | |
| async def input_users(client, ids: list) -> list: | |
| out = [] | |
| for uid in ids: | |
| try: | |
| peer = await client.get_input_entity(uid) | |
| except Exception as e: # noqa: BLE001 - a mention that can't be resolved is sent as the plain name | |
| log.info("could not resolve the mentioned user %s: %s", uid, type(e).__name__) | |
| continue | |
| if isinstance(peer, types.InputPeerUser): | |
| out.append(types.InputUser(peer.user_id, peer.access_hash)) | |
| return out | |
| async def received(client, event) -> Optional[Rich]: | |
| """The rich message that `event`'s message is (sent natively or forwarded to the bot), or None for an ordinary message.""" | |
| m = event.message | |
| rm = getattr(m, "rich_message", None) | |
| if rm is None: | |
| return None | |
| if getattr(rm, "part", None): # Telegram sent a part of it; the rest has to be asked for | |
| full = None | |
| try: | |
| got = await client(functions.messages.GetRichMessageRequest(peer=await event.get_input_chat(), id=m.id)) | |
| full = got if isinstance(got, types.RichMessage) else getattr(got, "rich_message", None) | |
| except errors.RPCError as e: | |
| log.info("could not read the whole rich message: %s", type(e).__name__) | |
| if full is None or getattr(full, "part", None): | |
| raise RichError("Telegram only handed me a part of that rich message. Send it again, or type it as HTML / Markdown.") | |
| rm = full | |
| if not getattr(rm, "blocks", None): | |
| raise RichError("That rich message has no content.") | |
| return rich_from_received(rm, await input_users(client, mention_ids(rm.blocks))) | |
| # --------------------------------------------------------------- when Telegram refuses a block of a received message | |
| _BLOCK_NAMES = { | |
| "Heading1": "heading", "Heading2": "heading", "Heading3": "heading", "Heading4": "heading", "Heading5": "heading", | |
| "Heading6": "heading", "Title": "heading", "Subtitle": "heading", "Header": "heading", "Subheader": "heading", | |
| "Paragraph": "paragraph", "Preformatted": "code", "Blockquote": "quote", "BlockquoteBlocks": "quote", | |
| "Pullquote": "pull quote", "List": "list", "OrderedList": "numbered list", "Details": "expandable section", | |
| "Math": "math", "Map": "map", "ButtonRow": "buttons", "Photo": "picture", "Video": "video", "Audio": "audio", | |
| "Document": "file", "Collage": "collage", "Slideshow": "slideshow", "Table": "table", "Divider": "rule", | |
| "Anchor": "anchor", "Footer": "footer", "Unsupported": "a block this version does not know", | |
| "Thinking": "thinking block", "Embed": "embedded page", "EmbedPost": "embedded post", "Channel": "channel card", | |
| "RelatedArticles": "related articles", "AuthorDate": "author line", "Kicker": "kicker", "Cover": "cover", | |
| } | |
| MAX_PROBES = 60 # blocks that are looked through one by one (at most) when Telegram refuses a message | |
| def block_name(b) -> str: | |
| """What a block is called to the owner: "map", "table" ...""" | |
| raw = re.sub(r"^(?:Input)?PageBlock", "", type(b).__name__) | |
| return _BLOCK_NAMES.get(raw, raw.lower() or "block") | |
| def is_block_invalid(e: BaseException) -> bool: | |
| """Telegram says one of the blocks it was given is not acceptable (RICH_MESSAGE_BLOCK_INVALID).""" | |
| return "RICH_MESSAGE_BLOCK_INVALID" in (getattr(e, "message", None) or str(e)) | |
| async def refused_blocks(client, peer, r: Rich, *, pause: float = 0.4) -> Optional[List[int]]: | |
| """The numbers (0, 1, 2 ...) of the top-level blocks of a received rich message that Telegram will not take back. | |
| Called once the whole message has been refused. Found by sending groups of the blocks to `peer` - a chat with the | |
| owner - quietly, and deleting them again straight away. None when it can't be told: the message is typed (Telegram | |
| reads that itself), it has too many blocks to look through, no single block is at fault (every part is taken), or | |
| something else than a refused block went wrong.""" | |
| if r.fmt != "blocks": | |
| return None | |
| whole = to_input(r) | |
| blocks = list(whole.blocks) | |
| if not blocks or len(blocks) > MAX_PROBES: | |
| return None | |
| async def accepted(idx: List[int]) -> bool: | |
| part = copy.copy(whole) | |
| part.blocks = [blocks[i] for i in idx] | |
| try: | |
| made = await flood_retry( | |
| lambda: client( | |
| functions.messages.SendMessageRequest(peer=peer, message="", rich_message=part, silent=True) | |
| ) | |
| ) | |
| except errors.RPCError as e: | |
| if is_block_invalid(e): | |
| return False | |
| raise | |
| try: | |
| await client(functions.messages.DeleteMessagesRequest(id=[sent_message(made).id], revoke=True)) | |
| except Exception: # noqa: BLE001 - a test message that stays is only clutter | |
| log.info("could not delete a test message of the rich-message check") | |
| await asyncio.sleep(pause) | |
| return True | |
| async def find(idx: List[int], known: bool = False) -> List[int]: | |
| if not known and await accepted(idx): | |
| return [] | |
| if len(idx) == 1: | |
| return list(idx) | |
| mid = len(idx) // 2 | |
| return await find(idx[:mid]) + await find(idx[mid:]) | |
| try: | |
| bad = await find(list(range(len(blocks))), known=True) # (the whole message was refused: no need to ask again) | |
| except Exception as e: # noqa: BLE001 - this is only a help for the owner | |
| log.info("could not find the refused blocks: %s", type(e).__name__) | |
| return None | |
| return bad or None | |
| def block_labels(r: Rich, idx: List[int]) -> str: | |
| """"4 (map), 9 (table)": the blocks numbered `idx` (from 0), counted from 1 and named.""" | |
| blocks = to_input(r).blocks | |
| return ", ".join(f"{i + 1} ({block_name(blocks[i])})" for i in sorted(set(idx)) if 0 <= i < len(blocks)) | |
| def block_count(r: Rich) -> int: | |
| return len(to_input(r).blocks) if r.fmt == "blocks" else 0 | |
| def digest(r: Rich) -> str: | |
| """Short mark of what `r` holds (so a button made for one message is not used on another).""" | |
| return hashlib.sha1(f"{r.fmt}:{r.source}".encode()).hexdigest()[:6] | |
| def pack_blocks(idx: List[int]) -> str: | |
| """Block numbers as one short word (the bits of a number, in base 36) - small enough for a button.""" | |
| n, out = sum(1 << i for i in set(idx)), "" | |
| while n: | |
| n, d = divmod(n, 36) | |
| out = "0123456789abcdefghijklmnopqrstuvwxyz"[d] + out | |
| return out or "0" | |
| def unpack_blocks(word: str) -> List[int]: | |
| n = int(word, 36) | |
| return [i for i in range(n.bit_length()) if n >> i & 1] | |
| def without_blocks(r: Rich, drop: List[int]) -> Rich: | |
| """The received rich message `r` without the top-level blocks numbered `drop` (what Telegram saw of it changes, so the | |
| fingerprint is forgotten).""" | |
| inp = to_input(r) | |
| inp.blocks = [b for i, b in enumerate(inp.blocks) if i not in set(drop)] | |
| return Rich("blocks", _b64(inp), r.rtl, r.noautolink, None) | |
| # ------------------------------------------------------------------------- plain text (lists, panels, previews) | |
| _BLOCKY = ("h1", "h2", "h3", "h4", "h5", "h6", "p", "ul", "ol", "li", "table", "tr", "details", "summary", "figure", | |
| "figcaption", "aside", "blockquote", "pre", "footer", "hr", "br", "tg-math-block", "tg-button-row", "tg-collage", | |
| "tg-slideshow", "tg-map", "img", "video", "audio", "tg-document", "caption", "cite") | |
| class _Plain(HTMLParser): | |
| def __init__(self) -> None: | |
| super().__init__(convert_charrefs=True) | |
| self.out: List[str] = [] | |
| self.skip = 0 | |
| def _break(self, tag: str) -> None: | |
| """A line break for a block; two blocks in a row are only one line apart (a <br> is always a break).""" | |
| if tag == "br" or (self.out and not self.out[-1].endswith("\n")): | |
| self.out.append("\n") | |
| def handle_starttag(self, tag, attrs): | |
| if tag in ("script", "style"): | |
| self.skip += 1 | |
| if tag in _BLOCKY: | |
| self._break(tag) | |
| if tag == "li": | |
| self.out.append("• ") | |
| elif tag in ("td", "th"): | |
| self.out.append(" | " if self.out and not self.out[-1].endswith(("\n", " | ")) else "") | |
| elif tag == "input" and any(k == "type" and v == "checkbox" for k, v in attrs): | |
| self.out.append("☑ " if any(k == "checked" for k, _ in attrs) else "☐ ") | |
| def handle_startendtag(self, tag, attrs): | |
| self.handle_starttag(tag, attrs) | |
| def handle_endtag(self, tag): | |
| if tag in ("script", "style") and self.skip: | |
| self.skip -= 1 | |
| if tag in _BLOCKY: | |
| self._break(tag) | |
| def handle_data(self, data): | |
| if not self.skip: | |
| self.out.append(data) | |
| def _tidy(text: str) -> str: | |
| lines = [" ".join(line.split()) for line in text.replace("\r", "").split("\n")] | |
| return re.sub(r"\n{3,}", "\n\n", "\n".join(lines)).strip() | |
| def _plain_html(source: str) -> str: | |
| p = _Plain() | |
| try: | |
| p.feed(source) | |
| p.close() | |
| except Exception: # noqa: BLE001 - a snippet is not worth failing for | |
| return _tidy(re.sub(r"<[^>]*>", " ", source)) | |
| return _tidy("".join(p.out)) | |
| _MD_EMPHASIS = re.compile(r"(\*\*|__|~~|\|\||==)(?=\S)(.+?)(?<=\S)\1") | |
| _MD_SINGLE = re.compile(r"(?<![\w*])([*_])(?=\S)([^*_\n]+?)(?<=\S)\1(?![\w*])") | |
| def _plain_markdown(source: str) -> str: | |
| out: List[str] = [] | |
| fenced = False | |
| for line in source.replace("\r", "").split("\n"): | |
| if re.match(r"^\s*(```|~~~)", line): | |
| fenced = not fenced | |
| continue | |
| if fenced: | |
| out.append(line) | |
| continue | |
| if re.match(r"^\s*\|?\s*:?-+:?\s*(\|\s*:?-+:?\s*)+\|?\s*$", line): # a table's delimiter row | |
| continue | |
| if re.match(r"^\s*(-{3,}|\*{3,}|_{3,})\s*$", line): | |
| out.append("") | |
| continue | |
| line = re.sub(r"^\s{0,3}#{1,6}\s+", "", line) | |
| line = re.sub(r"^(\s*)(?:>\s?)+", r"\1", line) | |
| line = re.sub(r"^(\s*)[-*+]\s+\[[xX]\]\s+", r"\1☑ ", line) | |
| line = re.sub(r"^(\s*)[-*+]\s+\[ \]\s+", r"\1☐ ", line) | |
| line = re.sub(r"^(\s*)[-*+]\s+", r"\1• ", line) | |
| if "|" in line and line.count("|") >= 2: | |
| line = " | ".join(c.strip() for c in line.strip().strip("|").split("|")) | |
| line = re.sub(r"!\[([^\]]*)\]\([^)]*?(?:\s+\"([^\"]*)\")?\)", lambda m: m.group(2) or m.group(1), line) | |
| line = re.sub(r"\[([^\]]+)\]\([^)]*\)", r"\1", line) | |
| line = re.sub(r"^\[\^([^\]\s]+)\]:[ \t]*", r"\1: ", line) # a footnote's text, led by its mark | |
| line = re.sub(r"\[\^[^\]]+\]", "", line) | |
| line = _MD_EMPHASIS.sub(r"\2", line) | |
| line = _MD_SINGLE.sub(r"\2", line) | |
| line = line.replace("`", "") | |
| line = re.sub(r"\\([\\`*_{}\[\]()#+\-.!|~>$=])", r"\1", line) | |
| out.append(line) | |
| return _tidy("\n".join(out)) | |
| def _plain_blocks(blocks) -> str: | |
| out: List[str] = [] | |
| def brk() -> None: | |
| if out and not out[-1].endswith("\n"): | |
| out.append("\n") | |
| def walk(o) -> None: | |
| if isinstance(o, dict): | |
| name = str(o.get("_", "")) | |
| if name == "PageTableRow": | |
| brk() | |
| for n, cell in enumerate(o.get("cells") or []): | |
| if n: | |
| out.append(" | ") | |
| walk(cell) | |
| brk() | |
| return | |
| item = name.startswith(("PageListItem", "PageListOrderedItem")) | |
| if name.startswith("PageBlock") or item: | |
| brk() | |
| if item: | |
| out.append(("☑ " if o.get("checked") else "☐ ") if o.get("checkbox") else "• ") | |
| for key in sorted(o, key=lambda k: k != "title"): # a title (of a <details>, a table) is read first | |
| v = o[key] | |
| if key in ("text", "source") and isinstance(v, str): | |
| out.append(v) | |
| elif isinstance(v, (dict, list)): | |
| walk(v) | |
| if name.startswith("PageBlock") or item: | |
| brk() | |
| elif isinstance(o, list): | |
| for v in o: | |
| walk(v) | |
| for b in blocks or []: | |
| try: | |
| walk(b.to_dict()) | |
| except Exception: # noqa: BLE001 | |
| continue | |
| return _tidy("".join(out)) | |
| def plain_of(r: Rich) -> str: | |
| """What the rich message says, as plain text (what My posts shows and searches).""" | |
| if r.fmt == "html": | |
| return _plain_html(r.source) | |
| if r.fmt == "markdown": | |
| return _plain_markdown(r.source) | |
| try: | |
| inp = _unb64(r.source) | |
| return _plain_blocks(getattr(inp, "blocks", None)) | |
| except Exception: # noqa: BLE001 | |
| return "" | |
| def plain_of_message(m) -> str: | |
| return _plain_blocks(getattr(getattr(m, "rich_message", None), "blocks", None)) | |
| # ---------------------------------------------------------------------------- reading what the owner typed | |
| _FORCE = re.compile(r"\A[ \t]*<rich((?:[ \t]+(?:rtl|noautolink))*)[ \t]*>[ \t]*(?:\r?\n|\Z)", re.I) | |
| # HTML only rich messages have: headings, paragraphs, lists, tables, details, figures, media, maps, math, buttons, marks ... | |
| _RICH_TAGS = ( | |
| r"h[1-6]", "p", "ul", "ol", "li", "table", "thead", "tbody", "tfoot", "tr", "td", "th", "caption", "details", "summary", | |
| "figure", "figcaption", "aside", "hr", "img", "video", "audio", "footer", "cite", "input", "mark", "sub", "sup", | |
| "tg-document", "tg-map", "tg-collage", "tg-slideshow", "tg-math", "tg-math-block", "tg-button", "tg-button-row", | |
| "tg-time", "tg-reference", | |
| ) | |
| _RICH_HTML = re.compile( | |
| r"<(?:" + "|".join(t if t.startswith("h[") else re.escape(t) for t in _RICH_TAGS) + r")(?=[\s/>])[^<>]*>" | |
| r"|<a\s[^<>]*?\b(?:name\s*=|href\s*=\s*[\"']?#)", | |
| re.I, | |
| ) | |
| _BASIC_HTML = re.compile(r"<(?:b|strong|i|em|u|ins|s|strike|del|a|code|pre|blockquote|tg-spoiler|tg-emoji)(?=[\s/>])[^<>]*>", re.I) | |
| # Markdown only rich messages have (a bullet list or a rule alone does not count: plain posts use those as decoration) | |
| _MD_RICH = ( | |
| re.compile(r"^ {0,3}#{1,6}[ \t]+\S", re.M), # # heading | |
| re.compile(r"^[ \t]*(?:[-*+]|\d{1,9}[.)])[ \t]+\[[ xX]\](?:[ \t]+\S|[ \t]*$)", re.M), # - [x] task | |
| re.compile(r"^[^\n]*\|[^\n]*\n[ \t]*\|?[ \t]*:?-+:?[ \t]*(?:\|[ \t]*:?-+:?[ \t]*)+\|?[ \t]*$", re.M), # | table | | |
| re.compile(r"\$\$.+?\$\$", re.S), # $$ block math $$ | |
| re.compile(r"^\[\^[^\]\s]+\]:[ \t]+\S", re.M), # [^1]: footnote | |
| re.compile(r"(?<![=\w])==(?=\S)[^=\n]+?(?<=\S)==(?![=\w])"), # ==marked== | |
| re.compile(r"^[ \t]*!\[[^\]\n]*\]\((?!tg://(?:emoji|time)\b)[^)\s]+(?:[ \t]+\"[^\"\n]*\")?\)[ \t]*$", re.M), #  | |
| re.compile(r"!\[[^\]\n]*\]\(tg://time\?[^)\s]+\)"), # a formatted date | |
| ) | |
| _CODE = re.compile(r"```.*?```|~~~.*?~~~|<pre\b.*?</pre\s*>|<code\b.*?</code\s*>|`[^`\n]+`", re.S | re.I) | |
| def _blank(m: "re.Match") -> str: | |
| return re.sub(r"[^\n]", " ", m.group()) | |
| def detect(text: str, forced: bool = False) -> Optional[str]: | |
| """"html" or "markdown" when `text` needs a rich message (or `forced` says it is one), else None.""" | |
| bare = _CODE.sub(_blank, text) # what stands in code is not markup | |
| html_hit = bool(_RICH_HTML.search(bare)) | |
| if html_hit and bare.lstrip().startswith("<"): | |
| return "html" | |
| if any(rx.search(bare) for rx in _MD_RICH): | |
| return "markdown" | |
| if html_hit: | |
| return "html" | |
| if forced: | |
| return "html" if _BASIC_HTML.search(bare) else "markdown" | |
| return None | |
| def read_typed(text: str) -> Optional[Rich]: | |
| """The rich message that text typed as HTML or Markdown stands for, or None when it is an ordinary text. | |
| It is a rich message when it has something only rich messages have (a heading, a table, a task list, math, a picture, | |
| <details> ...), or when its first line is ``<rich>`` (``<rich rtl>``, ``<rich noautolink>``). Telegram reads the | |
| language itself; what is passed on is exactly what was typed (without that first line).""" | |
| text = text or "" | |
| force = _FORCE.match(text) | |
| body = text[force.end():] if force else text | |
| fmt = detect(body, forced=bool(force)) | |
| if fmt is None or not body.strip(): | |
| return None | |
| words = (force.group(1) or "").lower().split() if force else [] | |
| return Rich(fmt, body.strip(), rtl="rtl" in words, noautolink="noautolink" in words) | |
| def check(r: Rich) -> None: | |
| """RichError when Telegram would not take it (only the limit that is cheap to check here).""" | |
| if r.fmt != "blocks" and len(r.source) > MAX_SOURCE: | |
| raise RichError(f"A rich message can have at most {MAX_SOURCE} characters (yours has {len(r.source)}).") | |
| def with_fingerprint(r: Rich, fp: Optional[str]) -> Rich: | |
| return replace(r, fp=fp) | |
| # --------------------------------------------------------------------------------------- the /rich help pages | |
| RICH_HELP = ( | |
| "🧩 <b>Rich messages</b>\n\n" | |
| "Telegram's newer message type: headings, lists, tables, expandable sections, math, maps, pictures and buttons in one " | |
| "message (up to 32768 characters, 500 blocks, 16 nesting levels, 50 pictures/files, 20 table columns). The bot " | |
| "publishes your text as one when\n" | |
| "• you send it a rich message made in Telegram, or forward one: it is kept block for block, or\n" | |
| "• you type HTML or Markdown that has something only rich messages have: a heading, a table, a task list, math, a " | |
| "picture, <code><details></code>, <code><h1></code>, <code><table></code> ... or\n" | |
| "• the first line is just <code><rich></code> (<code><rich rtl></code> for right-to-left text, " | |
| "<code><rich noautolink></code> so links, @names and #tags are not made clickable).\n\n" | |
| "Telegram reads the text itself, so everything it supports works. 👁 Preview shows what it made of your text before " | |
| "you publish. Rich messages are for text posts - a photo's caption can't be rich. The post's own link buttons " | |
| "(➕ Button) work with them.\n\n" | |
| "<b>Markdown</b>\n" | |
| "<code>**bold** *italic* ~~strike~~ ||spoiler|| ==marked== `code`</code> (underline, subscript and superscript: HTML " | |
| "<code><u> <sub> <sup></code>)\n" | |
| "<code># Heading</code> … <code>###### Heading</code>\n" | |
| "<code>- item</code> <code>1. item</code> lists, <code>- [ ] task</code> <code>- [x] done</code>, <code>---</code> a rule\n" | |
| "<code>> quote</code> (a bare <code>></code> line is a new paragraph inside it)\n" | |
| "<code>```python</code> … <code>```</code> a code block with a language\n" | |
| "<code>$x^2$</code> inline math, <code>$$E = mc^2$$</code> or a <code>```math</code> block: a formula block\n" | |
| "<code>| a | b |</code> then <code>|:--|--:|</code> then rows: a table (colons set the alignment; cells hold inline " | |
| "formatting only)\n" | |
| "<code>text[^1]</code> and a line <code>[^1]: the note</code>: a footnote\n" | |
| "<code>![](https://…/pic.jpg \"caption\")</code> alone on a line: a picture, video, audio, GIF or file (https links " | |
| "only); put several such lines inside <code><tg-collage></code> or <code><tg-slideshow></code>\n" | |
| "<code></code> a date shown in the reader's time\n" | |
| "<code></code> a custom emoji\n" | |
| "<code>[name](tg://user?id=123)</code> <code>[mail](mailto:a@b.c)</code> <code>[call](tel:+123)</code>\n" | |
| "Raw HTML works inside Markdown, but Markdown is not read inside block tags (only inside <code><details></code>, " | |
| "<code><tg-collage></code> and <code><tg-slideshow></code>)." | |
| ) | |
| RICH_HELP_HTML = ( | |
| "🧩 <b>Rich HTML</b>\n\n" | |
| "<code><h1></code> … <code><h6></code> <code><p></code> <code><hr></code> " | |
| "<code><footer></code> <code><b> <i> <u> <s> <mark> <sub> <sup> " | |
| "<code> <tg-spoiler></code> <code><br></code>\n" | |
| "<code><pre><code class=\"language-python\"></code> a code block\n" | |
| "<code><ul></code> <code><ol start=\"3\" type=\"a\" reversed></code> " | |
| "<code><li value=\"7\" type=\"i\"></code> (type: a, A, i, I, 1) " | |
| "<code><input type=\"checkbox\" checked></code>\n" | |
| "<code><blockquote expandable>text<cite>who</cite></blockquote></code> and " | |
| "<code><aside>pull quote<cite>who</cite></aside></code>\n" | |
| "<code><details open><summary>Title</summary>…</details></code> an expandable section\n" | |
| "<code><table bordered striped compact><caption>…</caption><tr><th colspan=\"2\" " | |
| "align=\"center\" valign=\"middle\">…</code> (td and th: colspan, rowspan, align left/center/right, valign " | |
| "top/middle/bottom)\n" | |
| "<code><tg-math>x^2</tg-math></code> <code><tg-math-block>E = mc^2</tg-math-block></code> " | |
| "(raw LaTeX)\n" | |
| "<code><tg-time unix=\"1647531900\" format=\"wDT\">22:45 tomorrow</tg-time></code>\n" | |
| "<code><tg-emoji emoji-id=\"5368324170671202286\">🙂</tg-emoji></code>\n" | |
| "<code><a name=\"top\"></a></code> and <code><a href=\"#top\"></code> anchors; " | |
| "<code><tg-reference name=\"n1\">text</tg-reference></code> with <code><a href=\"#n1\"></code> " | |
| "a footnote\n" | |
| "<code><img src=\"https://…\"></code> <code><video src=…></code> <code><audio src=…></code> " | |
| "<code><tg-document src=…></code>, each on its own or in " | |
| "<code><figure>…<figcaption>caption<cite>credit</cite></figcaption></figure></code> " | |
| "(https links only; <code>tg-spoiler</code> on img/video hides it); several inside " | |
| "<code><tg-collage></code> or <code><tg-slideshow></code>\n" | |
| "<code><tg-map lat=\"41.9\" long=\"12.5\" zoom=\"14\"/></code> a map\n" | |
| "<code><tg-button-row align=\"center\"><tg-button type=\"url\" url=\"https://t.me\" " | |
| "style=\"success\">Open</tg-button></tg-button-row></code> buttons in the text. Types: url, " | |
| "web_app and login_url (url=), callback_data (data=), copy_text (text=), switch_inline_query, " | |
| "switch_inline_query_current_chat, switch_inline_query_chosen_chat (query=), disabled. Styles: primary, success, " | |
| "danger, link.\n" | |
| "<code>&lt; &gt; &amp;</code> stand for < > & when they must stay as they are.\n\n" | |
| "<b>Good to know</b>\n" | |
| "• A picture or video typed into a rich message must be a public https link. To use a file that exists only on " | |
| "Telegram, send the bot a rich message that already contains it.\n" | |
| "• A button inside the text that needs the bot (callback_data) only works for the bot that made it - for a forwarded " | |
| "rich message that is another bot.\n" | |
| "• /swap, /handleswap and the post-link changes of /repost change a rich post's buttons, not the links inside its text.\n" | |
| "• Older Telegram apps may not show rich messages." | |
| ) | |