Automated sync from GitHub

.gitattributes

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

app/common.py

@@ -17,6 +17,7 @@ from .buttons_sync import ensure_markup, link_urls
 from .config import Config
 from .db import Database
 from .replace_engine import fetch_messages, is_skippable
+from .richmsg import rich_of, send_rich, with_rich
 from .tgutil import build_markup, de_entities, edit_raw, esc, explain_rpc, fetch_media, media_info, peer_of, with_media
 from .userbot import UserbotError, UserbotRefused, edit_problem_text
 
@@ -141,6 +142,9 @@ def require_owner(ctx: Ctx, event) -> None:
 async def send_post(client, peer, post):
     ents = de_entities(post.entities)
     markup = build_markup(post.buttons)
+    rich = rich_of(post)
+    if rich is not None and not post.media_file_id:  # a rich message: Telegram builds its blocks from what was kept
+        return await send_rich(client, peer, rich, markup=markup)
     if post.media_file_id:
         return await with_media(
             post.media_file_id,
@@ -158,6 +162,10 @@ async def push_edit(client, ch, p, *, replace_media: bool = False) -> None:
     peer = peer_of(ch)
     ents = de_entities(p.entities)
     markup = build_markup(p.buttons)
+    rich = rich_of(p)
+    if rich is not None and not p.media_kind:
+        await with_rich(rich, lambda inp: edit_raw(client, peer, p.message_id, rich_message=inp, markup=markup))
+        return
     if replace_media and p.media_file_id:
         await with_media(
             p.media_file_id,
@@ -264,7 +272,10 @@ async def edit_by_userbot(ctx: Ctx, ch, p, *, replace_media: bool, bot_error: st
             await notify(text)
 
     fields = dict(
-        text=p.text or "", entities=de_entities(p.entities), preview=None if p.media_kind else bool(p.link_preview)
+        text=p.text or "",
+        entities=de_entities(p.entities),
+        preview=None if p.media_kind or rich_of(p) else bool(p.link_preview),
+        rich=None if p.media_kind else rich_of(p),
     )
     try:
         await tell("⏳ Telegram refused the bot, so the userbot is making the edit.")

app/db.py

@@ -826,8 +826,9 @@ class Database:
                 q = select(Post).where(Post.channel_id == channel_id, Post.message_id == message_id)
                 p = (await s.execute(q)).scalar_one_or_none()
                 if p is not None:
-                    p.text = state.get("text", "")
-                    p.entities = state.get("entities", [])
+                    if not state.get("rich"):  # a rich message's text was never changed: it stays as saved
+                        p.text = state.get("text", "")
+                        p.entities = state.get("entities", [])
                     p.buttons = state.get("buttons", [])
                     p.updated_at = utcnow()
             if all_done:

app/handlers/basic.py

@@ -9,6 +9,7 @@ from telethon import events
 
 from ..common import Ctx, UserError, cmd, clear_state, discard_temp, guard, say, show
 from ..picker import PAGE_CB
+from ..richmsg import RICH_HELP, RICH_HELP_HTML
 from ..richtext import CHEAT_SHEET
 from ..tgutil import esc
 
@@ -21,6 +22,7 @@ HELP = (
     "/posts - your posts and drafts: edit text, caption, media and buttons, or delete them - a published post is deleted in the channel and in My posts (a button post also has “Mass replace links”)\n"
     "/autopost 114 20 - post “Episodes 01 to 20”, “21 to 40” ... up to 114, each with a download button, then fill in the links\n"
     "/formatting - bold, italic, links, quotes ...: Telegram's own formatting is kept, or type HTML / Markdown instead\n"
+    "/rich - rich messages: headings, lists, tables, expandable sections, math, maps, pictures, buttons ... typed as HTML / Markdown, or sent as a Telegram rich message\n"
     "/cancel - stop what you're doing\n\n"
     "<b>Channels</b>\n"
     "/addchannel - register a channel (the bot must be an admin there)\n"
@@ -92,6 +94,12 @@ def register(ctx: Ctx) -> None:
     async def h_formatting(event):
         await say(event, CHEAT_SHEET)
 
+    @client.on(cmd("rich"))
+    @guard(ctx)
+    async def h_rich(event):
+        await say(event, RICH_HELP)
+        await say(event, RICH_HELP_HTML)
+
     @client.on(cmd("cancel"))
     @guard(ctx)
     async def h_cancel(event):

app/handlers/broadcast.py

@@ -38,7 +38,7 @@ from ..db import utcnow
 from ..expiry import human, parse_duration, stamp
 from ..tgutil import esc, explain_rpc, flood_retry, peer_of
 from ..ui import CANCEL
-from .panel import extract_content, markup_note
+from .panel import markup_note, read_content
 
 log = logging.getLogger(__name__)
 
@@ -165,7 +165,7 @@ def register(ctx: Ctx) -> None:
             event,
             f"📣 <b>Broadcast</b> to {len(chans)} channel(s)\n{when}\n\n"
             "✍️ <b>Send the message now.</b> Text, or a photo / video / GIF / file with a caption. Formatting is kept "
-            "(or type HTML / Markdown - /formatting shows how).",
+            "(or type HTML / Markdown - /formatting shows how; rich blocks: /rich).",
             CANCEL,
         )
 
@@ -176,7 +176,7 @@ def register(ctx: Ctx) -> None:
             return  # the other items of an album: only the first one is used
         if m.grouped_id:
             st["group"] = m.grouped_id
-        content = extract_content(m, ctx.cfg.markup_input)
+        content = await read_content(ctx, event)
         set_state(ctx, event.sender_id, "bc_buttons", ttl=st["ttl"], content=content)
         await say(event, markup_note(content.how) + BUTTON_HELP, [[Button.inline("⏭ No buttons", "brs")], CANCEL[0]])
 

app/handlers/create.py

@@ -11,7 +11,7 @@ from ..common import Ctx, UserError, clear_state, cmd, discard_temp, guard, on_c
 from ..suggest import make_suggestion
 from ..tgutil import esc, short
 from ..ui import CANCEL, KIND_NAME
-from .panel import extract_content, markup_note, show_panel
+from .panel import markup_note, read_content, show_panel
 
 log = logging.getLogger(__name__)
 
@@ -32,7 +32,8 @@ def register(ctx: Ctx) -> None:
         await show(
             event,
             "✍️ <b>Send the post now.</b>\nText, or a photo / video / GIF / file with a caption. "
-            "Formatting (bold, links, spoilers...) is kept. No Telegram editor at hand? Type HTML or Markdown - /formatting shows how.",
+            "Formatting (bold, links, spoilers...) is kept. No Telegram editor at hand? Type HTML or Markdown - /formatting shows how; "
+            "headings, lists, tables, math and the other rich blocks: /rich.",
             CANCEL,
         )
 
@@ -153,7 +154,7 @@ def register(ctx: Ctx) -> None:
             return  # the other items of an album: only the first one is used
         if m.grouped_id:
             st["group"] = m.grouped_id
-        content = extract_content(m, ctx.cfg.markup_input)
+        content = await read_content(ctx, event)
         post = await db.create_post(
             channel_id=st["cid"],
             created_by=event.sender_id,

app/handlers/panel.py

@@ -1,12 +1,15 @@
 """Helpers shared by the create and post-management handlers (not a handler module itself)."""
 from __future__ import annotations
 
+from typing import Optional
+
 from telethon import types
 
 from ..buttons import valid
 from ..common import Ctx, UserError, is_cb, say, show
+from ..richmsg import Rich, RichError, check as check_rich, entry as rich_entry, plain_of, read_typed, received
 from ..richtext import message_text
-from ..tgutil import classify_media, pack_file_id, utf16_len
+from ..tgutil import classify_media, pack_file_id, ser_entities, utf16_len
 from ..ui import button_keyboard, button_text, panel_keyboard, panel_text
 
 CAPTION_MAX = 1024
@@ -53,27 +56,80 @@ def check_lengths(text: str, has_media: bool) -> None:
 
 
 class Content(dict):
-    """Post fields (text, entities, media_kind, media_file_id); `how` says whether the text was read as "html" or "markdown"."""
+    """Post fields (text, entities, media_kind, media_file_id).
+
+    `how` says what the text was read as: "html" or "markdown" (turned into formatting), "rich" (a rich message sent
+    natively, kept as it is), "rich-html" / "rich-markdown" (typed, and Telegram builds the blocks), or None."""
 
     how = None
 
 
 def markup_note(how, sep: str = "\n\n") -> str:
     """A line for the owner when their text was read as HTML or Markdown, followed by `sep` (empty otherwise)."""
-    return f"🎨 Formatting read from {'HTML' if how == 'html' else 'Markdown'} - see /formatting.{sep}" if how else ""
+    if not how:
+        return ""
+    if how == "rich":
+        return f"🧩 Rich message kept as it is - 👁 Preview shows it.{sep}"
+    if how.startswith("rich-"):
+        lang = "HTML" if how == "rich-html" else "Markdown"
+        return f"🧩 Rich message read from {lang} - 👁 Preview shows what Telegram makes of it (/rich).{sep}"
+    return f"🎨 Formatting read from {'HTML' if how == 'html' else 'Markdown'} - see /formatting.{sep}"
+
+
+def is_rich_how(how) -> bool:
+    return bool(how) and how.startswith("rich")
 
 
-def extract_content(m, markup: bool = True) -> Content:
+def read_text(m, markup: bool = True, rich: Optional[Rich] = None, allow_rich: bool = True) -> tuple:
+    """(text, entities, how) of what the owner sent - see Content.how.
+
+    A rich message (`rich`: sent natively, kept block for block) or text typed as HTML / Markdown that needs blocks
+    becomes a rich post: `text` is its plain-text version and its only entity is the rich entry. Otherwise it is an
+    ordinary text with Telegram's own formatting, or HTML / Markdown turned into formatting. A message that has
+    Telegram formatting of its own is used as it is. `allow_rich` is False where a rich message can't be (a caption)."""
+    how = None
+    if rich is not None:
+        if not allow_rich:
+            raise UserError("A caption can't be a rich message. Rich messages are for text posts.")
+        how = "rich"
+    elif allow_rich and markup and not ser_entities(m.entities):
+        rich = read_typed(m.raw_text or "")
+        how = f"rich-{rich.fmt}" if rich is not None else None
+    if rich is None:
+        return message_text(m, markup)
+    try:
+        check_rich(rich)
+    except RichError as e:
+        raise UserError(str(e))
+    return plain_of(rich), [rich_entry(rich)], how
+
+
+async def read_native(ctx: Ctx, event) -> Optional[Rich]:
+    """The rich message the owner sent natively (or forwarded), or None for an ordinary message."""
+    try:
+        return await received(ctx.client, event)
+    except RichError as e:
+        raise UserError(str(e))
+
+
+def extract_content(m, markup: bool = True, rich: Optional[Rich] = None) -> Content:
     """A message from the user -> post fields (text, entities, media).
 
-    Telegram's own formatting is kept as it is; without any, `markup` lets HTML or Markdown in the text become formatting."""
+    Telegram's own formatting is kept as it is; without any, `markup` lets HTML or Markdown in the text become formatting
+    (or a rich message, when it has blocks only those have). `rich`: the message is a rich message sent natively."""
     kind = classify_media(m)
     if m.media and kind is None and not isinstance(m.media, types.MessageMediaWebPage):
         raise UserError("That kind of message can't be used in a post. Send text, a photo, video, GIF, audio or a file.")
-    text, entities, how = message_text(m, markup)
-    if not kind and not text.strip():
+    text, entities, how = read_text(m, markup, rich if not kind else None, allow_rich=not kind)
+    if not kind and not text.strip() and not is_rich_how(how):
         raise UserError("Send some text, or a photo / video / GIF / file.")
-    check_lengths(text, bool(kind))
+    if not is_rich_how(how):
+        check_lengths(text, bool(kind))
     content = Content(text=text, entities=entities, media_kind=kind, media_file_id=pack_file_id(m) if kind else None)
     content.how = how
     return content
+
+
+async def read_content(ctx: Ctx, event) -> Content:
+    """extract_content for a message that just arrived (a rich message sent natively is read, in full, first)."""
+    return extract_content(event.message, ctx.cfg.markup_input, await read_native(ctx, event))

app/handlers/posts.py

@@ -36,14 +36,17 @@ from ..common import (
 )
 from ..db import utcnow
 from ..repost_engine import delete_batch
-from ..richtext import message_text
+from ..richmsg import is_rich
 from ..tgutil import classify_media, esc, explain_rpc, pack_file_id, peer_of, post_link
 from ..ui import CANCEL, post_label
 from ..userbot import delete_problem_text
 from .panel import (
     check_lengths,
+    is_rich_how,
     load_post,
     markup_note,
+    read_native,
+    read_text,
     show_button_screen,
     show_panel,
 )
@@ -123,7 +126,8 @@ def register(ctx: Ctx) -> None:
         post, _ = await load_post(ctx, parts[0])
         set_state(ctx, event.sender_id, "edit_text", pid=post.id)
         what = "caption" if post.media_kind else "text"
-        await say(event, f"✏️ Send the new {what}. Formatting is kept - or type HTML / Markdown (/formatting shows how).", CANCEL)
+        more = "" if post.media_kind else " Rich messages (headings, lists, tables ...) too: /rich."
+        await say(event, f"✏️ Send the new {what}. Formatting is kept - or type HTML / Markdown (/formatting shows how).{more}", CANCEL)
 
     @on_text(ctx, "edit_text")
     async def h_edit_text(event, st):
@@ -131,10 +135,12 @@ def register(ctx: Ctx) -> None:
         if classify_media(m):
             raise UserError("Send text only here. To change the media, use 🖼 Media.")
         post, ch = await load_post(ctx, st["pid"])
-        text, entities, how = message_text(m, ctx.cfg.markup_input)
-        if not post.media_kind and not text.strip():
+        native = await read_native(ctx, event)
+        text, entities, how = read_text(m, ctx.cfg.markup_input, native, allow_rich=not post.media_kind)
+        if not post.media_kind and not text.strip() and not is_rich_how(how):
             raise UserError("A text post can't be empty.")
-        check_lengths(text, bool(post.media_kind))
+        if not is_rich_how(how):
+            check_lengths(text, bool(post.media_kind))
         post, report = await apply_edit_reported(ctx, post, ch, text=text, entities=entities)
         clear_state(ctx, event.sender_id)
         await show_panel(ctx, event, post.id, note=markup_note(how, "\n") + note_for(post, report), new=True)
@@ -143,6 +149,11 @@ def register(ctx: Ctx) -> None:
     @on_cb(ctx, "em")
     async def cb_edit_media(event, parts):
         post, _ = await load_post(ctx, parts[0])
+        if is_rich(post):
+            raise UserError(
+                "A rich message has no media slot. Put pictures into the text itself (see /rich), "
+                "or send new text without rich blocks first."
+            )
         set_state(ctx, event.sender_id, "edit_media", pid=post.id)
         extra = "" if post.media_kind else "\nThe current text becomes its caption."
         await say(event, f"🖼 Send the new photo / video / GIF / audio / file.{extra}", CANCEL)
@@ -355,9 +366,14 @@ def register(ctx: Ctx) -> None:
     @on_cb(ctx, "pv")
     async def cb_preview(event, parts):
         post, _ = await load_post(ctx, parts[0])
-        if not (post.text or "").strip() and not post.media_file_id:
+        if not (post.text or "").strip() and not post.media_file_id and not is_rich(post):
             raise UserError("The post is empty.")
-        await send_post(client, await event.get_input_chat(), post)
+        try:
+            await send_post(client, await event.get_input_chat(), post)
+        except errors.RPCError as e:
+            if not is_rich(post):
+                raise
+            raise UserError("❌ " + explain_rpc(e) + "\nTelegram could not make that rich message - see /rich for the syntax.")
         miss = missing_links(post.buttons)
         if miss:
             await say(event, f"👆 Preview. ⚠️ {miss} button(s) without a link are not shown.")
@@ -369,7 +385,7 @@ def register(ctx: Ctx) -> None:
             raise UserError("This post is already published.")
         if missing_links(post.buttons):
             raise UserError("Some buttons (⚠️) have no link yet. Open them and add a link first.")
-        if not (post.text or "").strip() and not post.media_file_id:
+        if not (post.text or "").strip() and not post.media_file_id and not is_rich(post):
             raise UserError("The post is empty.")
         await show(
             event,

app/handlers/replace.py

@@ -26,6 +26,7 @@ from ..common import (
 )
 from ..linkswap import normalize_username
 from ..replace_engine import ApplyResult, ChannelScan, ScanOptions, apply_channel, scan_channel, undo_batch
+from ..richmsg import is_rich
 from ..richtext import message_text
 from ..sync_engine import SyncOptions, sync_ids
 from ..tgutil import classify_media, esc, explain_rpc, get_rights, pack_file_id
@@ -265,6 +266,8 @@ def register(ctx: Ctx) -> None:
                     raise UserError("That post no longer exists in the channel, or I can't add it to My posts.")
             if post.status != "sent":
                 raise UserError("That post is not published yet.")
+            if is_rich(post):
+                raise UserError("That post is a rich message - it has no media or caption to replace. Edit its text in My posts.")
             caption, entities, how = message_text(reply, cfg.markup_input)  # Telegram's formatting, or HTML / Markdown
             check_lengths(caption, True)
 

app/handleswap.py

@@ -13,6 +13,7 @@ from telethon import types
 from telethon.helpers import add_surrogate, del_surrogate
 
 from .linkswap import MsgChange, apply_span_replacements
+from .richmsg import is_rich_message
 from .tgutil import button_url, is_callback_button, with_text
 
 # a "@name" that follows one of these is part of something else: an e-mail (a@bro), a link (t.me/x@bro, ?u=@bro) ...
@@ -44,8 +45,9 @@ def find_mention_spans(sur: str, entities, name: str) -> tuple:
 def compute_handle_changes(msg, old: str, new: str) -> tuple:
     """(MsgChange or None, number of mentions skipped because they are part of a link) - same shape as
     linkswap.compute_message_changes, so the /swap engine can apply it."""
-    text = msg.message or ""
-    ents = list(msg.entities or [])
+    rich = is_rich_message(msg)  # the text of a rich message is blocks: only its buttons are changed
+    text = "" if rich else (msg.message or "")
+    ents = [] if rich else list(msg.entities or [])
     sur = add_surrogate(text)
     spans, skipped = find_mention_spans(sur, ents, old) if text else ([], 0)
     new_sur, new_ents = sur, [copy.copy(e) for e in ents]

app/linkswap.py

@@ -13,6 +13,7 @@ from typing import Optional
 from telethon import types
 from telethon.helpers import add_surrogate, del_surrogate
 
+from .richmsg import is_rich_message
 from .tgutil import button_url, is_callback_button, with_url
 
 USERNAME_RX = re.compile(r"^[A-Za-z0-9_]{1,32}$")
@@ -140,8 +141,9 @@ class MsgChange:
 
 def compute_message_changes(msg, old: str, new: str, *, include_typed: bool = True, include_posts: bool = False) -> tuple:
     """Returns (MsgChange or None, number_of_skipped_post_links)."""
-    text = msg.message or ""
-    ents = list(msg.entities or [])
+    rich = is_rich_message(msg)  # the text of a rich message is blocks: only its buttons are changed
+    text = "" if rich else (msg.message or "")
+    ents = [] if rich else list(msg.entities or [])
     skipped = 0
     n_typed = n_links = n_btn = 0
 

app/postlinks.py

@@ -15,6 +15,7 @@ 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)"
@@ -195,7 +196,10 @@ class Relinked:
 
 def relink_message(msg, links: PostLinks) -> Optional[Relinked]:
     """The new text / entities / buttons of `msg` once its links are remapped; None when nothing changes."""
-    text, entities, n_typed, n_hyper = links.remap_text(msg.message or "", msg.entities)
+    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

app/replace_engine.py

@@ -10,6 +10,7 @@ from telethon import errors, functions, types
 
 from .handleswap import compute_handle_changes
 from .linkswap import compute_message_changes
+from .richmsg import entry as rich_entry, is_rich_message, plain_of_message, rich_from_received, rich_of
 from .tgutil import (
     build_markup,
     de_entities,
@@ -230,6 +231,9 @@ async def apply_channel(
                 preview = preview_flag(m)
                 markup = change.new_markup if change.new_markup is not None else m.reply_markup
                 old_state = state_of(m.message, m.entities, m.reply_markup, preview)
+                rich = is_rich_message(m)
+                if rich:  # its text is blocks this run never touches: undo puts the buttons back and nothing else
+                    old_state["rich"] = True
                 try:
                     await flood_retry(lambda: _do_edit(client, peer, m, change, markup, preview))
                 except errors.MessageNotModifiedError:
@@ -256,11 +260,20 @@ async def apply_channel(
                 }
                 if kind:
                     adopt["media_kind"], adopt["media_file_id"] = kind, fid
+                new_state = state_of(change.new_text, change.new_entities, markup, preview)
+                if rich:
+                    new_state["rich"] = True
+                    saved = await db.find_by_message(ch.id, m.id)
+                    if saved is not None and rich_of(saved) is not None:
+                        del adopt["text"], adopt["entities"]  # the saved copy keeps what it was sent from
+                    else:  # a rich message nobody saved yet: it is taken over as its blocks
+                        adopt["text"] = plain_of_message(m)
+                        adopt["entities"] = [rich_entry(rich_from_received(m.rich_message))]
                 buf.append(
                     {
                         "message_id": m.id,
                         "old_state": old_state,
-                        "new_state": state_of(change.new_text, change.new_entities, markup, preview),
+                        "new_state": new_state,
                         "adopt": adopt,
                     }
                 )
@@ -298,17 +311,22 @@ async def undo_batch(
             continue
         old = c.old_state or {}
         try:
-            await flood_retry(
-                lambda: edit_raw(
-                    client,
-                    peer_of(ch),
-                    c.message_id,
-                    text=old.get("text", ""),
-                    entities=de_entities(old.get("entities")),
-                    markup=build_markup(old.get("buttons")),
-                    preview=old.get("preview"),
+            if old.get("rich"):  # a rich message: only its buttons were changed, so only they are put back
+                markup = build_markup(old.get("buttons"))
+                if markup is not None:
+                    await flood_retry(lambda: edit_raw(client, peer_of(ch), c.message_id, markup=markup))
+            else:
+                await flood_retry(
+                    lambda: edit_raw(
+                        client,
+                        peer_of(ch),
+                        c.message_id,
+                        text=old.get("text", ""),
+                        entities=de_entities(old.get("entities")),
+                        markup=build_markup(old.get("buttons")),
+                        preview=old.get("preview"),
+                    )
                 )
-            )
         except errors.MessageNotModifiedError:
             pass
         except errors.RPCError as e:

app/repost_engine.py

@@ -42,6 +42,7 @@ 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,
@@ -89,6 +90,16 @@ class Final:
     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:
@@ -99,8 +110,11 @@ def final_state(src, opts: RepostOptions, links: Optional[PostLinks] = None) ->
         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(
@@ -130,6 +144,7 @@ def final_state(src, opts: RepostOptions, links: Optional[PostLinks] = None) ->
         dropped=dropped,
         n_post_links=n_post_links,
         n_relinked=n_relinked,
+        rich=rich,
     )
 
 
@@ -212,7 +227,9 @@ def copy_source(unit: list, rows: dict, *, restricted: Optional[str] = None, use
         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):
+        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
 
@@ -223,6 +240,8 @@ def stand_in(m, row):
     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)
@@ -267,7 +286,7 @@ def without_files(unit: list) -> tuple:
             c = copy.copy(m)
             c.media, c.saved_ref, c.invert_media = None, None, None
             lost += 1
-        if (c.message or "").strip():
+        if (c.message or "").strip() or rich_for(c) is not None:
             out.append(c)
     return out, lost
 
@@ -592,7 +611,22 @@ async def resend_unit(client, peer, unit: list, finals: list, reply_to: Optional
     """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)
-    if len(unit) > 1:
+    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)
@@ -634,7 +668,7 @@ async def resend_unit(client, peer, unit: list, finals: list, reply_to: Optional
                     reply_to=reply,
                 )
 
-    result = await flood_retry(lambda: client(make()))
+    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])
@@ -847,8 +881,8 @@ def _rows(unit: list, finals: list, news: list) -> list:
                 "old_id": src.id,
                 "new_id": new.id,
                 "post": {
-                    "text": f.text,
-                    "entities": ser_entities(f.entities),
+                    "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,
@@ -1131,12 +1165,16 @@ async def relink_copies(
             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:
-                await flood_retry(
-                    lambda: edit_raw(
-                        client, peer, m.id, text=rl.text, entities=rl.entities, markup=rl.markup, preview=preview_flag(m)
+                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:
@@ -1155,9 +1193,10 @@ async def relink_copies(
             res.relinked += 1
             res.links += rl.links
             try:
-                await db.update_post_by_message(
-                    ch.id, m.id, text=rl.text, entities=ser_entities(rl.entities), buttons=ser_markup(rl.markup)
-                )
+                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:

app/richmsg.py

@@ -0,0 +1,537 @@
+"""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 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 .tgutil import _b64, _unb64
+
+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."""
+
+
+@dataclass(frozen=True)
+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)
+    return _unb64(r.source)
+
+
+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(b))
+    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=list(rm.blocks),
+        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)))
+
+
+# ------------------------------------------------------------------------- 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),  # ![](picture)
+    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>&lt;details&gt;</code>, <code>&lt;h1&gt;</code>, <code>&lt;table&gt;</code> ... or\n"
+    "• the first line is just <code>&lt;rich&gt;</code> (<code>&lt;rich rtl&gt;</code> for right-to-left text, "
+    "<code>&lt;rich noautolink&gt;</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>&lt;u&gt; &lt;sub&gt; &lt;sup&gt;</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>&gt; quote</code> (a bare <code>&gt;</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>&lt;tg-collage&gt;</code> or <code>&lt;tg-slideshow&gt;</code>\n"
+    "<code>![22:45 tomorrow](tg://time?unix=1647531900&amp;format=wDT)</code> a date shown in the reader's time\n"
+    "<code>![](tg://emoji?id=5368324170671202286)</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>&lt;details&gt;</code>, "
+    "<code>&lt;tg-collage&gt;</code> and <code>&lt;tg-slideshow&gt;</code>)."
+)
+
+RICH_HELP_HTML = (
+    "🧩 <b>Rich HTML</b>\n\n"
+    "<code>&lt;h1&gt;</code> … <code>&lt;h6&gt;</code> <code>&lt;p&gt;</code> <code>&lt;hr&gt;</code> "
+    "<code>&lt;footer&gt;</code> <code>&lt;b&gt; &lt;i&gt; &lt;u&gt; &lt;s&gt; &lt;mark&gt; &lt;sub&gt; &lt;sup&gt; "
+    "&lt;code&gt; &lt;tg-spoiler&gt;</code> <code>&lt;br&gt;</code>\n"
+    "<code>&lt;pre&gt;&lt;code class=\"language-python\"&gt;</code> a code block\n"
+    "<code>&lt;ul&gt;</code> <code>&lt;ol start=\"3\" type=\"a\" reversed&gt;</code> "
+    "<code>&lt;li value=\"7\" type=\"i\"&gt;</code> (type: a, A, i, I, 1) "
+    "<code>&lt;input type=\"checkbox\" checked&gt;</code>\n"
+    "<code>&lt;blockquote expandable&gt;text&lt;cite&gt;who&lt;/cite&gt;&lt;/blockquote&gt;</code> and "
+    "<code>&lt;aside&gt;pull quote&lt;cite&gt;who&lt;/cite&gt;&lt;/aside&gt;</code>\n"
+    "<code>&lt;details open&gt;&lt;summary&gt;Title&lt;/summary&gt;…&lt;/details&gt;</code> an expandable section\n"
+    "<code>&lt;table bordered striped compact&gt;&lt;caption&gt;…&lt;/caption&gt;&lt;tr&gt;&lt;th colspan=\"2\" "
+    "align=\"center\" valign=\"middle\"&gt;…</code> (td and th: colspan, rowspan, align left/center/right, valign "
+    "top/middle/bottom)\n"
+    "<code>&lt;tg-math&gt;x^2&lt;/tg-math&gt;</code> <code>&lt;tg-math-block&gt;E = mc^2&lt;/tg-math-block&gt;</code> "
+    "(raw LaTeX)\n"
+    "<code>&lt;tg-time unix=\"1647531900\" format=\"wDT\"&gt;22:45 tomorrow&lt;/tg-time&gt;</code>\n"
+    "<code>&lt;tg-emoji emoji-id=\"5368324170671202286\"&gt;🙂&lt;/tg-emoji&gt;</code>\n"
+    "<code>&lt;a name=\"top\"&gt;&lt;/a&gt;</code> and <code>&lt;a href=\"#top\"&gt;</code> anchors; "
+    "<code>&lt;tg-reference name=\"n1\"&gt;text&lt;/tg-reference&gt;</code> with <code>&lt;a href=\"#n1\"&gt;</code> "
+    "a footnote\n"
+    "<code>&lt;img src=\"https://…\"&gt;</code> <code>&lt;video src=…&gt;</code> <code>&lt;audio src=…&gt;</code> "
+    "<code>&lt;tg-document src=…&gt;</code>, each on its own or in "
+    "<code>&lt;figure&gt;…&lt;figcaption&gt;caption&lt;cite&gt;credit&lt;/cite&gt;&lt;/figcaption&gt;&lt;/figure&gt;</code> "
+    "(https links only; <code>tg-spoiler</code> on img/video hides it); several inside "
+    "<code>&lt;tg-collage&gt;</code> or <code>&lt;tg-slideshow&gt;</code>\n"
+    "<code>&lt;tg-map lat=\"41.9\" long=\"12.5\" zoom=\"14\"/&gt;</code> a map\n"
+    "<code>&lt;tg-button-row align=\"center\"&gt;&lt;tg-button type=\"url\" url=\"https://t.me\" "
+    "style=\"success\"&gt;Open&lt;/tg-button&gt;&lt;/tg-button-row&gt;</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>&amp;lt; &amp;gt; &amp;amp;</code> stand for &lt; &gt; &amp; 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."
+)

app/richtext.py

@@ -546,5 +546,7 @@ CHEAT_SHEET = (
     "<b>Example</b>\n"
     "<pre>**Attack on Titan** - Episode 12\n"
     "__Hindi__ | ||spoiler||\n"
-    "[Watch here](https://t.me/yourchannel)</pre>"
+    "[Watch here](https://t.me/yourchannel)</pre>\n\n"
+    "🧩 Headings, lists, tables, expandable sections, math, maps, pictures and buttons are <b>rich messages</b> - "
+    "see /rich."
 )

app/suggest.py

@@ -9,6 +9,7 @@ from typing import Optional
 from telethon.helpers import add_surrogate, del_surrogate
 
 from .linkswap import apply_span_replacements
+from .richmsg import is_rich
 from .tgutil import de_entities, ser_entities
 
 RANGE_RX = re.compile(r"(?P<a>\d+)(?P<sep>\s*(?:to|[-–—~&])\s*)(?P<pre>(?:ep?\.?\s*)?)(?P<b>\d+)", re.I)
@@ -64,6 +65,8 @@ class Suggestion:
 
 
 def build_suggestion(post) -> Optional[Suggestion]:
+    if is_rich(post):  # its text lives in blocks Telegram made: the numbers can't be moved on in them
+        return None
     rows: list = []
     pending: list = []
     first: Optional[Advance] = None

app/sync_engine.py

@@ -38,6 +38,7 @@ from .db import as_utc, utcnow
 from .replace_engine import fetch_messages
 from .repost_engine import delete_batch
 from .restrictions import message_restriction, norm_text, probe_channel
+from .richmsg import Rich, entry as rich_entry, is_rich_message, plain_of_message, rich_from_received, rich_of, with_fingerprint
 from .tgutil import classify_media, explain_rpc, media_from_ref, media_info, peer_of, ser_entities, ser_markup
 
 log = logging.getLogger(__name__)
@@ -127,12 +128,19 @@ def stored_media_key(ref: Optional[str]) -> Optional[tuple]:
         return ("unreadable", ref[:40])  # differs from anything real: the saved reference gets replaced
 
 
+def live_rich(m) -> Optional[Rich]:
+    """The rich message the live message is (as blocks), or None for an ordinary post."""
+    rm = getattr(m, "rich_message", None)
+    return rich_from_received(rm) if rm is not None else None
+
+
 def live_fields(m) -> dict:
     """The saved-post fields of a message as it is in the channel now."""
     kind, ref = media_info(m)
+    rich = live_rich(m)
     return {
-        "text": m.message or "",
-        "entities": ser_entities(m.entities),
+        "text": plain_of_message(m) if rich is not None else (m.message or ""),
+        "entities": [rich_entry(rich)] if rich is not None else ser_entities(m.entities),
         "buttons": ser_markup(m.reply_markup),
         "media_kind": kind,
         "media_file_id": ref,
@@ -140,8 +148,8 @@ def live_fields(m) -> dict:
 
 
 def message_supported(m) -> bool:
-    """Something My posts can show and edit: text, or a photo / video / GIF / audio / file."""
-    return bool((m.message or "").strip()) or classify_media(m) is not None
+    """Something My posts can show and edit: text, a rich message, or a photo / video / GIF / audio / file."""
+    return bool((m.message or "").strip()) or classify_media(m) is not None or is_rich_message(m)
 
 
 def adopt_fields(m) -> dict:
@@ -185,10 +193,24 @@ def diff_post(p, m) -> Diff:
     """What differs between the saved post `p` and the live message `m`."""
     live = live_fields(m)
     fields: dict = {}
-    if (p.text or "") != live["text"]:
-        fields["text"] = live["text"]
-    if list(p.entities or []) != live["entities"]:
-        fields["entities"] = live["entities"]
+    saved_rich, now_rich = rich_of(p), live_rich(m) if is_rich_message(m) else None
+    first_look = False
+    if saved_rich is None and now_rich is None:
+        if (p.text or "") != live["text"]:
+            fields["text"] = live["text"]
+        if list(p.entities or []) != live["entities"]:
+            fields["entities"] = live["entities"]
+    elif now_rich is not None:
+        # A rich message is compared by what Telegram shows of it (a fingerprint of its blocks), never by its text: the
+        # saved copy keeps the source the bot sent. A post that became rich in the channel is taken over as its blocks.
+        if saved_rich is None or (saved_rich.fp is not None and saved_rich.fp != now_rich.fp):
+            fields["text"], fields["entities"] = live["text"], live["entities"]
+        elif saved_rich.fp is None:  # the bot's own rich post, seen for the first time: remember what Telegram shows
+            fields["entities"] = [rich_entry(with_fingerprint(saved_rich, now_rich.fp))]
+            first_look = True
+    elif live["text"].strip() or live["media_kind"] is not None:
+        # the channel shows an ordinary post where a rich message was saved (an empty answer is not trusted)
+        fields["text"], fields["entities"] = live["text"], live["entities"]
     # media: replaced media is taken over; media that seems to have vanished is not (a message can't lose its media)
     if live["media_kind"] is not None:
         if p.media_kind != live["media_kind"] or stored_media_key(p.media_file_id) != media_key(m.media):
@@ -199,7 +221,9 @@ def diff_post(p, m) -> Diff:
             fields["buttons"] = live["buttons"]
     elif p.buttons:
         missing = True
-    quiet = set(fields) == {"text"} and (p.text or "").strip() == live["text"].strip()
+    quiet = (set(fields) == {"text"} and (p.text or "").strip() == live["text"].strip()) or (
+        first_look and set(fields) == {"entities"}
+    )
     return Diff(fields, missing, quiet)
 
 

app/tgutil.py

@@ -381,19 +381,31 @@ async def flood_retry(factory: Callable[[], Awaitable[Any]], tries: int = 4) ->
 
 
 async def edit_raw(
-    client, peer, message_id: int, *, text: Optional[str] = None, entities=None, markup=None, preview=None, media=None
+    client,
+    peer,
+    message_id: int,
+    *,
+    text: Optional[str] = None,
+    entities=None,
+    markup=None,
+    preview=None,
+    media=None,
+    rich_message=None,
 ):
     """messages.editMessage with every field explicit.
 
-    `markup=None` together with `text` removes the keyboard (that is how Telegram behaves);
+    `markup=None` together with `text` (or `rich_message`) removes the keyboard (that is how Telegram behaves);
     a markup-only edit (text=None) leaves the text alone. `media` (an InputMedia) replaces the photo / file.
+    `rich_message` (an InputRichMessage*) makes the message that rich message instead of text.
     """
     kw: dict = {"peer": peer, "id": message_id}
     if markup is not None:
         kw["reply_markup"] = markup
     if media is not None:
         kw["media"] = media
-    if text is not None:
+    if rich_message is not None:
+        kw["rich_message"] = rich_message
+    elif text is not None:
         kw["message"] = text
         kw["entities"] = list(entities or [])
         if preview is not None:

app/ui.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 from telethon import Button
 
 from .buttons import count, grid_label, link_positions, missing_links
+from .richmsg import rich_of
 from .tgutil import esc, post_link, short
 
 CANCEL = [[Button.inline("✖️ Cancel", "cx")]]
@@ -30,7 +31,12 @@ def panel_text(p, ch, note=None) -> str:
     lines = [f"<b>Post #{p.id}</b> · {state}", f"Channel: <b>{esc(ch.title)}</b>"]
     if p.status == "sent" and p.message_id:
         lines.append(f'<a href="{post_link(ch, p.message_id)}">Open in the channel</a>')
-    lines.append(f"Type: {KIND_ICON.get(p.media_kind, '📎')} {KIND_NAME.get(p.media_kind, p.media_kind)}")
+    rich = rich_of(p)
+    if rich is not None:
+        origin = {"html": "typed as HTML", "markdown": "typed as Markdown"}.get(rich.fmt, "made in Telegram")
+        lines.append(f"Type: 🧩 rich message ({origin})")
+    else:
+        lines.append(f"Type: {KIND_ICON.get(p.media_kind, '📎')} {KIND_NAME.get(p.media_kind, p.media_kind)}")
     n, miss = count(p.buttons), missing_links(p.buttons)
     lines.append(f"Buttons: {n}" + (f" (⚠️ {miss} without a link)" if miss else ""))
     if p.source == "adopted":
@@ -47,18 +53,17 @@ def panel_keyboard(p) -> list:
     kb = []
     for r, row in enumerate(p.buttons or []):
         kb.append([Button.inline(grid_label(b), f"bs:{pid}:{r}:{c}") for c, b in enumerate(row)])
+    rich = rich_of(p) is not None
     kb.append(
-        [
-            Button.inline("✏️ Caption" if p.media_kind else "✏️ Text", f"et:{pid}"),
-            Button.inline("🖼 Media", f"em:{pid}"),
-        ]
+        [Button.inline("✏️ Caption" if p.media_kind else "✏️ Text", f"et:{pid}")]
+        + ([] if rich else [Button.inline("🖼 Media", f"em:{pid}")])  # a rich message has no media slot
     )
     row3 = [Button.inline("➕ Button", f"ba:{pid}"), Button.inline("📋 Set buttons", f"bp:{pid}")]
     if p.buttons:
         row3.append(Button.inline("🧹 Clear", f"bc:{pid}"))
     kb.append(row3)
     extra = []
-    if not p.media_kind:
+    if not p.media_kind and not rich:
         extra.append(Button.inline(f"🔗 Link preview: {'on' if p.link_preview else 'off'}", f"lp:{pid}"))
     if p.media_kind and p.status == "draft":
         extra.append(Button.inline("🚫 Remove media", f"rm:{pid}"))

app/userbot.py

@@ -31,6 +31,7 @@ from typing import Optional
 from telethon import TelegramClient, errors, functions, types, utils
 from telethon.sessions import StringSession
 
+from .richmsg import with_rich
 from .tgutil import MediaFile, edit_raw, flood_retry, get_rights, media_from_ref, media_key, peer_of, with_media
 
 log = logging.getLogger(__name__)
@@ -524,17 +525,24 @@ class UserbotEditor(_ChannelJob):
                 log.info("could not delete the helper message in the userbot's chat: %s", type(e).__name__)
         await super().close()
 
-    async def edit(self, message_id: int, *, text: str, entities: list, media=None, preview=None) -> None:
+    async def edit(self, message_id: int, *, text: str, entities: list, media=None, preview=None, rich=None) -> None:
         """Make the post show `text` (with `entities`) and, when given, `media`: what media_by_reference() returned (the
-        file does not move at all) or a MediaFile (uploaded again)."""
+        file does not move at all) or a MediaFile (uploaded again). `rich` (a richmsg.Rich) makes it that rich message."""
         await self._begin()
         client = self.ub.client
         try:
             if isinstance(media, MediaFile):
                 media = await self._upload(media)
-            await flood_retry(
-                lambda: edit_raw(client, self.peer, message_id, text=text, entities=entities, preview=preview, media=media)
-            )
+            if rich is not None:
+                await flood_retry(
+                    lambda: with_rich(rich, lambda inp: edit_raw(client, self.peer, message_id, rich_message=inp))
+                )
+            else:
+                await flood_retry(
+                    lambda: edit_raw(
+                        client, self.peer, message_id, text=text, entities=entities, preview=preview, media=media
+                    )
+                )
         except errors.MessageNotModifiedError:
             return  # it already looks like that
         except errors.RPCError as e:

tests/fakes.py

@@ -303,7 +303,28 @@ class FakeChannelClient:
         else:
             m._hidden, m.reply_markup = None, markup
 
-    def _new(self, text, entities, markup, media=None, gid=None, invert=None, hide=False, reply_to=None):
+    def _rich(self, inp):
+        """What Telegram makes of an InputRichMessage*: for HTML / Markdown the "parse" is one paragraph holding the source
+        (the same source always gives the same blocks); a message sent as blocks keeps them, with its pictures and files."""
+        if isinstance(inp, (types.InputRichMessageHTML, types.InputRichMessageMarkdown)):
+            src = inp.html if isinstance(inp, types.InputRichMessageHTML) else inp.markdown
+            if not src.strip():
+                raise rpc("MessageEmptyError")
+            kind = "html" if isinstance(inp, types.InputRichMessageHTML) else "markdown"
+            return types.RichMessage(
+                blocks=[types.PageBlockParagraph(types.TextPlain(f"[{kind}]{src}"))], photos=[], documents=[], rtl=inp.rtl
+            )
+        if isinstance(inp, types.InputRichMessage):
+            photos = [
+                types.Photo(p.id, p.access_hash, p.file_reference, NOW, [], 1) for p in inp.photos or []
+            ]
+            docs = [
+                types.Document(d.id, d.access_hash, d.file_reference, NOW, "video/mp4", 1, 1, []) for d in inp.documents or []
+            ]
+            return types.RichMessage(blocks=list(inp.blocks), photos=photos, documents=docs, rtl=inp.rtl)
+        raise rpc("MediaInvalidError")
+
+    def _new(self, text, entities, markup, media=None, gid=None, invert=None, hide=False, reply_to=None, rich=None):
         self._top += 1
         self._pts += 1
         m = Message(
@@ -317,6 +338,7 @@ class FakeChannelClient:
             grouped_id=gid,
             invert_media=invert,
             reply_to=types.MessageReplyHeader(reply_to_msg_id=reply_to) if reply_to else None,
+            rich_message=rich,
         )
         if self.hide_keyboard_on_create > 0 and markup is not None:
             hide = True
@@ -458,7 +480,7 @@ class FakeChannelClient:
             out.append(
                 self._new(
                     src.message, src.entities, src.reply_markup, src.media, gid, getattr(src, "invert_media", None),
-                    hide=self.forward_hides_keyboard,
+                    hide=self.forward_hides_keyboard, rich=getattr(src, "rich_message", None),
                 )
             )
         if not out:
@@ -482,6 +504,11 @@ class FakeChannelClient:
 
     def _do_SendMessageRequest(self, req):
         rid = self._check_reply(req)
+        if req.rich_message is not None:
+            rich = self._rich(req.rich_message)
+            m = self._new("", None, req.reply_markup, None, None, None, reply_to=rid, rich=rich)
+            m.rich_input = req.rich_message  # what the bot asked for (the tests look at it)
+            return self._updates([m])
         return self._updates([self._new(req.message, req.entities, req.reply_markup, None, None, req.invert_media, reply_to=rid)])
 
     def _do_SendMediaRequest(self, req):
@@ -529,6 +556,19 @@ class FakeChannelClient:
             raise rpc("MessageAuthorRequiredError")
         if getattr(m, "via_other_bot", False):  # buttons or a post made through another bot: only that bot may edit
             raise rpc("MessageIdInvalidError")
+        if req.rich_message is not None:  # the post becomes this rich message; no keyboard sent = keyboard removed
+            rich = self._rich(req.rich_message)
+            if (
+                m.rich_message is not None
+                and bytes(m.rich_message) == bytes(rich)
+                and _bytes(req.reply_markup) == _bytes(self._effective(m))
+            ):
+                raise rpc("MessageNotModifiedError")
+            m.message, m.entities, m.rich_message, m.rich_input = "", None, rich, req.rich_message
+            self._set_markup(m, req.reply_markup)
+            self.edits.append(req)
+            self._pts += 1
+            return self._updates([])
         if req.message is None:  # markup-only edit
             text, ents = m.message, list(m.entities or [])
             markup = req.reply_markup if req.reply_markup is not None else self._effective(m)
@@ -541,11 +581,21 @@ class FakeChannelClient:
         ):
             raise rpc("MessageNotModifiedError")
         m.message, m.entities = text, (ents or None)
+        if req.message is not None:
+            m.rich_message = None
         self._set_markup(m, markup)
         self.edits.append(req)
         self._pts += 1
         return self._updates([])
 
+    def _do_GetRichMessageRequest(self, req):
+        """messages.getRichMessage: the whole rich message of a post (the tests may put a longer one in `full_rich`)."""
+        full = getattr(self, "full_rich", {}).get(req.id)
+        m = self.msgs.get(req.id)
+        if full is None and (m is None or m.rich_message is None):
+            raise rpc("MessageIdInvalidError")
+        return full if full is not None else m.rich_message
+
     def _do_DeleteMessagesRequest(self, req):
         ids = list(req.id)
         self.delete_requests.append(ids)
@@ -772,6 +822,18 @@ class FakeUserClient:
             media = self._by_reference(req.media)
         else:
             media = self._uploaded_media(req.media)
+        if req.rich_message is not None:
+            rich = chan._rich(req.rich_message)
+            if m.rich_message is not None and bytes(m.rich_message) == bytes(rich):
+                raise rpc("MessageNotModifiedError")
+            m.message, m.entities, m.rich_message, m.rich_input = "", None, rich, req.rich_message
+            if not chan.user_edit_keeps_markup:
+                chan._set_markup(m, None)
+            if chan.user_edit_frees_message:
+                m.via_other_bot = False
+            chan.user_edits.append(req)
+            chan._pts += 1
+            return chan._updates([])
         text = m.message if req.message is None else req.message
         ents = list(m.entities or []) if req.message is None else list(req.entities or [])
         if media is None and text == m.message and [_bytes(e) for e in ents] == [_bytes(e) for e in (m.entities or [])]:

tests/harness.py

@@ -180,6 +180,14 @@ class FakeTG(FakeChannelClient):
                 return None
         if isinstance(req, functions.channels.DeleteMessagesRequest):
             self.deleted.append(list(req.id))
+        if isinstance(req, functions.messages.SendMessageRequest) and req.rich_message is not None:
+            self.sent.append(("rich", req.peer, req.rich_message, {"reply_markup": req.reply_markup}))
+            if isinstance(req.peer, types.InputPeerUser):  # a chat with a person (the preview): nothing lands in a channel
+                if self.fail.get("SendMessageRequest") is not None:
+                    raise self.fail["SendMessageRequest"]
+                bytes(req)  # it must serialise
+                self._rich(req.rich_message)  # Telegram has to be able to make blocks of it
+                return types.UpdateShortSentMessage(id=next(self.ids), pts=0, pts_count=0, date=None, out=True)
         return await super().__call__(req)
 
 

tests/test_rich.py

@@ -0,0 +1,483 @@
+"""Rich messages (headings, lists, tables, <details>, math, maps, pictures, buttons ...): what is recognised, how a rich message is
+kept in a post, what is sent to Telegram for it and how one that is sent to the bot is read."""
+import asyncio
+from types import SimpleNamespace
+
+import pytest
+from telethon import errors, functions, types
+
+from app.richmsg import (
+    RICH_HELP,
+    RICH_HELP_HTML,
+    MAX_SOURCE,
+    Rich,
+    RichError,
+    blank_references,
+    check,
+    detect,
+    entry,
+    fingerprint,
+    input_users,
+    is_rich,
+    is_rich_message,
+    mention_ids,
+    plain_of,
+    plain_of_message,
+    read_typed,
+    real_entities,
+    received,
+    rich_from_received,
+    rich_of,
+    send_rich,
+    sent_message,
+    to_input,
+    with_rich,
+)
+from app.tgutil import de_entities, edit_raw
+
+from .fakes import photo_media, rpc, video_media
+
+T = types.TextPlain
+
+
+def blocks(word="Hello"):
+    return [types.PageBlockHeading1(T("Title")), types.PageBlockParagraph(types.TextBold(T(word)))]
+
+
+def rich_msg(bl=None, *, photos=None, documents=None, rtl=None, part=None):
+    """A rich message as Telegram hands it to the bot."""
+    return types.RichMessage(blocks=bl or blocks(), photos=photos or [], documents=documents or [], rtl=rtl, part=part)
+
+
+def same(a, b):
+    """The same TL objects (they have no usable ==)."""
+    return [bytes(x) for x in a] == [bytes(x) for x in b]
+
+
+# ============================================================================================ what is a rich message
+HTML_RICH = [
+    "<h1>Title</h1>",
+    "<h4 id='x'>Small</h4>",
+    "<p>one</p><p>two</p>",
+    "<ul><li>a</li><li>b</li></ul>",
+    "<ol start=\"3\"><li>a</li></ol>",
+    "<table bordered><tr><td>a</td></tr></table>",
+    "<details><summary>More</summary>text</details>",
+    "<tg-map lat=\"41.9\" long=\"12.5\" zoom=\"14\"/>",
+    "<tg-math-block>E = mc^2</tg-math-block>",
+    "<aside>pull<cite>me</cite></aside>",
+    "<hr>",
+    "<footer>fine print</footer>",
+    "<img src=\"https://x.org/a.jpg\">",
+    "<figure><img src=\"https://x.org/a.jpg\"><figcaption>c</figcaption></figure>",
+    "<tg-collage><img src=\"https://x.org/a.jpg\"></tg-collage>",
+    "<tg-button-row><tg-button type=\"url\" url=\"https://t.me\">Open</tg-button></tg-button-row>",
+    "text with <mark>marked</mark> words",
+    "H<sub>2</sub>O and x<sup>2</sup>",
+    "<input type=\"checkbox\" checked> done",
+    "<tg-time unix=\"1647531900\" format=\"wDT\">soon</tg-time>",
+    "<a name=\"top\"></a>go",
+    "back to <a href=\"#top\">the top</a>",
+    "<tg-reference name=\"n1\">note</tg-reference>",
+]
+MARKDOWN_RICH = [
+    "# Title",
+    "text\n\n### Sub heading",
+    "###### Smallest",
+    "- [ ] todo",
+    "1. [x] done",
+    "| a | b |\n|---|---|\n| 1 | 2 |",
+    "a | b\n:-:|--:\n1 | 2",
+    "$$E = mc^2$$",
+    "a note[^1]\n\n[^1]: the note",
+    "some ==marked== text",
+    "![](https://x.org/a.jpg)",
+    "![](https://x.org/a.mp4 \"caption\")",
+    "see ![22:45 tomorrow](tg://time?unix=1647531900&format=wDT)",
+]
+NOT_RICH = [
+    "Episode 12 is out",
+    "#anime #naruto",
+    "#1 hit",  # a hashtag with no space after the # is not a heading
+    "Price: $5 and $6",
+    "a == b",
+    "1 == 1 and 2 == 2",
+    "I <3 this",
+    "- one\n- two",
+    "* a\n* b",
+    "---",
+    "1) first\n2) second",
+    "**bold** and __under__ and ~~strike~~ and ||spoiler|| and `code`",
+    "<b>bold</b> <i>it</i> <u>u</u> <s>s</s> <a href=\"https://t.me\">x</a> <code>c</code> <tg-spoiler>s</tg-spoiler>",
+    "<blockquote>quote</blockquote> and <pre>block</pre>",
+    "<pre>&lt;h1&gt; stays text</pre>",
+    "`<h1>` is a tag",
+    "```\n# not a heading\n```",
+    "~~~\n| a | b |\n|---|---|\n~~~",
+    "a | b | c",
+    "5 > 3 and 2 < 4",
+    "@channel_name look at https://example.com/a#section",
+    "[watch](https://t.me/x) and ![smile](tg://emoji?id=1)",
+    "",
+    "   ",
+]
+
+
+@pytest.mark.parametrize("text", HTML_RICH)
+def test_html_with_rich_blocks_is_a_rich_message(text):
+    assert detect(text) == "html"
+    assert read_typed(text) == Rich("html", text)
+
+
+@pytest.mark.parametrize("text", MARKDOWN_RICH)
+def test_markdown_with_rich_blocks_is_a_rich_message(text):
+    assert detect(text) == "markdown"
+    assert read_typed(text) == Rich("markdown", text)
+
+
+@pytest.mark.parametrize("text", NOT_RICH)
+def test_ordinary_text_is_not_turned_into_a_rich_message(text):
+    assert detect(text) is None
+    assert read_typed(text) is None
+
+
+def test_html_wins_when_the_text_starts_with_a_tag_and_markdown_otherwise():
+    assert detect("<h1>Title</h1>\n\n- [ ] not a task here") == "html"
+    assert detect("# Title\n\nsome <mark>marked</mark> html inside") == "markdown"
+    assert detect("intro\n<h2>Sub</h2>") == "html"
+
+
+def test_the_rich_marker_makes_any_text_a_rich_message():
+    assert read_typed("<rich>\nhello **x**") == Rich("markdown", "hello **x**")
+    assert read_typed("<rich>\n<b>x</b> and <i>y</i>") == Rich("html", "<b>x</b> and <i>y</i>")
+    r = read_typed("<rich rtl noautolink>\nhello #tag @name")
+    assert (r.fmt, r.source, r.rtl, r.noautolink) == ("markdown", "hello #tag @name", True, True)
+    r = read_typed("  <RICH   rtl >\r\nمرحبا")  # spaces, any case and a Windows line end are fine
+    assert (r.source, r.rtl, r.noautolink) == ("مرحبا", True, False)
+    assert read_typed("<rich noautolink>\n# Title") == Rich("markdown", "# Title", noautolink=True)
+    # the marker is a line of its own, at the start
+    assert read_typed("<rich> hello") is None
+    assert read_typed("hello <rich>\nx") is None
+    assert read_typed("hello\n<rich>\nx") is None
+    assert read_typed("<rich>") is None and read_typed("<rich>\n  \n") is None
+
+
+def test_what_is_typed_is_handed_over_exactly():
+    src = "# Title  \n\n  indented\n\n| a | b |\n|---|---|\n| 1 | 2 |"
+    assert read_typed("\n" + src + "\n\n").source == src.strip()
+    assert read_typed(src).source == src
+    assert read_typed("<rich>\n" + src).source == src
+
+
+def test_code_is_not_markup():
+    assert detect("```python\n# comment\nprint('<h1>')\n```") is None
+    assert detect("use `# Heading` for a heading and <code><h1></code> in HTML") is None
+    assert detect("```\n- [ ] x\n```\n# Real heading") == "markdown"
+
+
+# ================================================================================================= keeping a rich message
+def test_a_rich_message_is_one_entry_among_the_entities_of_a_post():
+    r = Rich("html", "<h1>x</h1>", rtl=True, noautolink=True, fp="abc")
+    e = entry(r)
+    assert e == {"k": "rich", "f": "html", "s": "<h1>x</h1>", "r": 1, "n": 1, "p": "abc"}
+    assert entry(Rich("markdown", "# x")) == {"k": "rich", "f": "markdown", "s": "# x"}
+    post = SimpleNamespace(entities=[{"k": "bold", "o": 0, "l": 1}, e])
+    assert rich_of(post) == r and is_rich(post)
+    assert real_entities(post.entities) == [{"k": "bold", "o": 0, "l": 1}]
+    assert real_entities(None) == []
+    for bad in ({"k": "rich", "f": "pdf", "s": "x"}, {"k": "rich", "f": "html", "s": ""}, {"k": "bold", "o": 0, "l": 1}, "x"):
+        assert rich_of(SimpleNamespace(entities=[bad])) is None
+    assert rich_of(SimpleNamespace(entities=None)) is None and rich_of(SimpleNamespace()) is None
+    assert not is_rich(SimpleNamespace(entities=[]))
+
+
+def test_the_entry_is_not_formatting():
+    assert de_entities([entry(Rich("markdown", "# x"))]) == []
+    assert de_entities([entry(Rich("html", "<b>x</b>")), {"k": "bold", "o": 0, "l": 2}]) == [types.MessageEntityBold(0, 2)]
+
+
+# ============================================================================================ what is sent to Telegram
+def test_typed_text_goes_to_telegram_as_html_or_markdown_for_it_to_read():
+    i = to_input(Rich("html", "<h1>x</h1>", rtl=True))
+    assert isinstance(i, types.InputRichMessageHTML) and i.html == "<h1>x</h1>" and i.rtl is True and i.noautolink is None
+    i = to_input(Rich("markdown", "# x", noautolink=True))
+    assert isinstance(i, types.InputRichMessageMarkdown) and i.markdown == "# x" and i.noautolink is True and i.rtl is None
+    bytes(to_input(Rich("markdown", "# x")))  # serialises
+
+
+def test_a_rich_message_that_was_received_is_sent_again_block_for_block():
+    rm = rich_msg(photos=[photo_media(5).photo], documents=[video_media(81).document], rtl=True)
+    r = rich_from_received(rm)
+    assert (r.fmt, r.rtl, r.fp) == ("blocks", True, fingerprint(rm))
+    inp = to_input(r)
+    assert isinstance(inp, types.InputRichMessage) and inp.rtl is True
+    assert same(inp.blocks, rm.blocks)
+    assert [(p.id, p.access_hash, p.file_reference) for p in inp.photos] == [(5, 35, b"ref5")]
+    assert [(d.id, d.access_hash, d.file_reference) for d in inp.documents] == [(81, 81 * 7, b"fr81")]
+    bytes(inp)
+    plain = to_input(rich_from_received(rich_msg()))
+    assert plain.photos is None and plain.documents is None and plain.users is None and not plain.rtl
+
+
+def test_file_references_can_be_left_out():
+    inp = to_input(rich_from_received(rich_msg(photos=[photo_media(5).photo], documents=[video_media(81).document])))
+    blank = blank_references(inp)
+    assert blank.photos[0].file_reference == b"" and (blank.photos[0].id, blank.photos[0].access_hash) == (5, 35)
+    assert blank.documents[0].file_reference == b"" and blank.documents[0].id == 81
+    assert inp.photos[0].file_reference == b"ref5"  # the original is untouched
+    html = types.InputRichMessageHTML(html="x")
+    assert blank_references(html) is html
+
+
+def test_a_stale_file_reference_is_retried_without_it():
+    async def run():
+        r = rich_from_received(rich_msg(photos=[photo_media(5).photo]))
+        seen = []
+
+        async def call(inp):
+            seen.append(inp)
+            if len(seen) == 1:
+                raise rpc("FileReferenceExpiredError")
+            return "ok"
+
+        assert await with_rich(r, call) == "ok"
+        assert seen[0].photos[0].file_reference == b"ref5" and seen[1].photos[0].file_reference == b""
+
+        async def always(inp):
+            raise rpc("FileReferenceExpiredError")
+
+        with pytest.raises(errors.RPCError):
+            await with_rich(r, always)  # the second try failed as well: that error is shown
+        with pytest.raises(errors.RPCError):
+            await with_rich(Rich("html", "<p>x</p>"), always)  # typed text has no references to blame
+
+        async def other(inp):
+            raise rpc("MessageEmptyError")
+
+        with pytest.raises(errors.MessageEmptyError):
+            await with_rich(r, other)
+
+    asyncio.run(run())
+
+
+def test_send_rich_makes_one_raw_request_and_finds_the_message_it_created():
+    async def run():
+        sent = []
+
+        async def client(req):
+            sent.append(req)
+            return types.Updates(
+                updates=[types.UpdateNewChannelMessage(message=types.Message(id=41, peer_id=types.PeerChannel(1), date=None, message=""), pts=1, pts_count=1)],
+                users=[], chats=[], date=None, seq=0,
+            )
+
+        markup = types.ReplyInlineMarkup([])
+        m = await send_rich(client, types.InputPeerChannel(1, 5), Rich("markdown", "# x"), markup=markup, silent=True)
+        assert m.id == 41
+        (req,) = sent
+        assert isinstance(req, functions.messages.SendMessageRequest)
+        assert req.message == "" and req.reply_markup is markup and req.silent is True
+        assert isinstance(req.rich_message, types.InputRichMessageMarkdown) and req.rich_message.markdown == "# x"
+        bytes(req)
+
+    asyncio.run(run())
+
+
+def test_the_message_a_send_created_is_found_in_every_answer_telegram_gives():
+    short = types.UpdateShortSentMessage(id=7, pts=1, pts_count=1, date=None, out=True)
+    assert sent_message(short).id == 7
+    new = types.UpdateNewMessage(message=types.Message(id=8, peer_id=types.PeerUser(1), date=None, message=""), pts=1, pts_count=1)
+    assert sent_message(types.Updates(updates=[new], users=[], chats=[], date=None, seq=0)).id == 8
+    with pytest.raises(RuntimeError):
+        sent_message(types.Updates(updates=[], users=[], chats=[], date=None, seq=0))
+
+
+def test_edit_raw_makes_the_message_that_rich_message():
+    async def run():
+        sent = []
+
+        async def client(req):
+            sent.append(req)
+
+        inp = types.InputRichMessageMarkdown(markdown="# x")
+        peer = types.InputPeerChannel(1, 5)
+        await edit_raw(client, peer, 9, rich_message=inp, markup=None, text="ignored", entities=[types.MessageEntityBold(0, 1)], preview=True)
+        (req,) = sent
+        assert req.rich_message is inp and req.message is None and not req.entities and not req.no_webpage and req.reply_markup is None
+        bytes(req)
+        sent.clear()
+        await edit_raw(client, peer, 9, text="plain", entities=[], markup=None)
+        assert sent[0].rich_message is None and sent[0].message == "plain"
+
+    asyncio.run(run())
+
+
+def test_a_rich_message_longer_than_telegram_allows_is_refused():
+    check(Rich("html", "x" * MAX_SOURCE))
+    check(Rich("blocks", "x" * (MAX_SOURCE * 2)))  # received blocks are Telegram's own: nothing to count here
+    with pytest.raises(RichError) as exc:
+        check(Rich("markdown", "x" * (MAX_SOURCE + 1)))
+    assert "32768" in str(exc.value)
+
+
+# ===================================================================================== a rich message sent to the bot
+def test_the_fingerprint_follows_what_the_message_shows():
+    a, b = rich_msg(blocks("x")), rich_msg(blocks("x"), photos=[photo_media(5).photo])
+    assert fingerprint(a) == fingerprint(b)  # the pictures' file references change from fetch to fetch
+    assert fingerprint(a) != fingerprint(rich_msg(blocks("y")))
+    assert fingerprint(a) != fingerprint(rich_msg(blocks("x"), rtl=True))
+    assert len(fingerprint(a)) == 20
+
+
+def test_is_rich_message():
+    assert is_rich_message(SimpleNamespace(rich_message=rich_msg()))
+    assert not is_rich_message(SimpleNamespace(rich_message=None)) and not is_rich_message(SimpleNamespace())
+
+
+def test_people_mentioned_by_link_are_found_and_looked_up():
+    async def run():
+        bl = [types.PageBlockParagraph(types.TextConcat([T("hi "), types.TextMentionName(T("Bob"), 42), types.TextMentionName(T("Bob"), 42), types.TextMentionName(T("Al"), 43)]))]
+        assert mention_ids(bl) == [42, 43]
+        assert mention_ids(blocks()) == [] and mention_ids(None) == []
+
+        class Client:
+            async def get_input_entity(self, uid):
+                if uid == 42:
+                    return types.InputPeerUser(42, 777)
+                raise ValueError("unknown")
+
+        users = await input_users(Client(), [42, 43])
+        assert [(u.user_id, u.access_hash) for u in users] == [(42, 777)]  # the one that can't be found is left out
+        r = rich_from_received(rich_msg(bl), users)
+        assert [u.user_id for u in to_input(r).users] == [42]
+
+    asyncio.run(run())
+
+
+class Event:
+    """The message event of a rich message that arrived (what `received` reads)."""
+
+    def __init__(self, rm, mid=77):
+        self.message = SimpleNamespace(rich_message=rm, id=mid)
+
+    async def get_input_chat(self):
+        return types.InputPeerUser(1, 0)
+
+
+class Client:
+    def __init__(self, answer=None, error=None):
+        self.answer, self.error, self.requests = answer, error, []
+
+    async def __call__(self, req):
+        self.requests.append(req)
+        if self.error:
+            raise self.error
+        return self.answer
+
+    async def get_input_entity(self, uid):
+        raise ValueError(uid)
+
+
+def test_a_whole_rich_message_is_read_as_it_arrived():
+    async def run():
+        rm = rich_msg(photos=[photo_media(5).photo])
+        client = Client()
+        r = await received(client, Event(rm))
+        assert client.requests == []  # nothing more to ask
+        assert (r.fmt, r.fp) == ("blocks", fingerprint(rm)) and plain_of(r) == "Title\nHello"
+        assert await received(client, Event(None)) is None  # an ordinary message
+
+    asyncio.run(run())
+
+
+def test_a_rich_message_that_arrives_in_parts_is_fetched_whole():
+    async def run():
+        part = rich_msg([types.PageBlockParagraph(T("first"))], part=True)
+        client = Client(answer=rich_msg(blocks("whole")))
+        r = await received(client, Event(part, mid=9))
+        (req,) = client.requests
+        assert isinstance(req, functions.messages.GetRichMessageRequest) and req.id == 9
+        assert plain_of(r) == "Title\nwhole"
+
+    asyncio.run(run())
+
+
+def test_a_rich_message_that_cannot_be_had_whole_is_refused_in_plain_words():
+    async def run():
+        part = rich_msg([types.PageBlockParagraph(T("first"))], part=True)
+        for client in (
+            Client(answer=rich_msg(blocks(), part=True)),  # still only a part
+            Client(error=rpc("MessageIdInvalidError")),  # Telegram refuses to say more
+            Client(answer=None),  # nothing came back
+        ):
+            with pytest.raises(RichError) as exc:
+                await received(client, Event(part))
+            assert "part of that rich message" in str(exc.value)
+        with pytest.raises(RichError) as exc:
+            await received(Client(), Event(types.RichMessage(blocks=[], photos=[], documents=[])))
+        assert "no content" in str(exc.value)
+
+    asyncio.run(run())
+
+
+# ==================================================================================== the plain text of a rich message
+def test_the_plain_text_of_typed_html():
+    html = (
+        "<h1>Schedule</h1><p>Every <b>Monday</b></p><ul><li>one</li><li><input type=\"checkbox\" checked> two</li></ul>"
+        "<table><tr><th>Day</th><th>Show</th></tr><tr><td>Mon</td><td>Naruto</td></tr></table>"
+        "<details><summary>More</summary>hidden text</details><blockquote>quote<cite>me</cite></blockquote>"
+        "<tg-math-block>E = mc^2</tg-math-block>"
+    )
+    assert plain_of(Rich("html", html)) == (
+        "Schedule\nEvery Monday\n• one\n• ☑ two\nDay | Show\nMon | Naruto\nMore\nhidden text\nquote\nme\nE = mc^2"
+    )
+    assert plain_of(Rich("html", "line one<br>line two<br><br>after&nbsp;a gap &amp; more")) == "line one\nline two\n\nafter a gap & more"
+    assert plain_of(Rich("html", "a<script>var x = 1;</script>b")) == "ab"
+
+
+def test_the_plain_text_of_typed_markdown():
+    md = (
+        "# Title\n\nSome **bold**, *it* and ==mark== [link](https://t.me/x)\n\n- [x] done\n- [ ] todo\n- item\n\n"
+        "| a | b |\n|:--|--:|\n| 1 | 2 |\n\n> quoted\n\n```python\nprint(1)\n```\n\n$$E = mc^2$$\n\n"
+        "note[^1]\n\n[^1]: the note\n\n![](https://x.org/a.jpg \"a caption\")\n\n---\n\nend"
+    )
+    assert plain_of(Rich("markdown", md)) == (
+        "Title\n\nSome bold, it and mark link\n\n☑ done\n☐ todo\n• item\n\na | b\n1 | 2\n\nquoted\n\nprint(1)\n\n"
+        "$$E = mc^2$$\n\nnote\n\n1: the note\n\na caption\n\nend"
+    )
+
+
+def test_the_plain_text_of_a_rich_message_made_in_telegram():
+    bl = [
+        types.PageBlockHeading1(T("Title")),
+        types.PageBlockParagraph(types.TextConcat([T("Every "), types.TextBold(T("Monday"))])),
+        types.PageBlockList([types.PageListItemText(T("one")), types.PageListItemText(T("two"), checkbox=True, checked=True)]),
+        types.PageBlockTable(
+            T("sched"),
+            [
+                types.PageTableRow([types.PageTableCell(text=T("Day"), header=True), types.PageTableCell(text=T("Show"), header=True)]),
+                types.PageTableRow([types.PageTableCell(text=T("Mon")), types.PageTableCell(text=T("Naruto"))]),
+            ],
+        ),
+        types.PageBlockMath("E = mc^2"),
+        types.PageBlockDetails([types.PageBlockParagraph(T("hidden"))], T("More")),
+    ]
+    want = "Title\nEvery Monday\n• one\n☑ two\nsched\nDay | Show\nMon | Naruto\nE = mc^2\nMore\nhidden"
+    rm = rich_msg(bl)
+    assert plain_of(rich_from_received(rm)) == want
+    assert plain_of_message(SimpleNamespace(rich_message=rm)) == want
+    assert plain_of_message(SimpleNamespace(rich_message=None)) == ""
+    assert plain_of(Rich("blocks", "not base64 of anything")) == ""
+
+
+# ============================================================================================================== the help
+def test_the_help_pages_fit_in_a_message_and_cover_the_blocks():
+    for page in (RICH_HELP, RICH_HELP_HTML):
+        assert len(page) < 4096
+    help_all = RICH_HELP + RICH_HELP_HTML
+    for word in (
+        "&lt;rich&gt;", "# Heading", "- [ ] task", "```python", "$$E = mc^2$$", "[^1]", "tg://time", "tg://emoji", "==marked==",
+        "&lt;details", "&lt;table", "&lt;tg-math-block&gt;", "&lt;tg-map", "&lt;tg-collage&gt;", "&lt;tg-slideshow&gt;",
+        "&lt;tg-button-row", "&lt;aside&gt;", "&lt;figure&gt;", "&lt;tg-reference", "&lt;tg-time", "&lt;sub&gt;", "https",
+    ):
+        assert word in help_all, word

tests/test_rich_flows.py

@@ -0,0 +1,759 @@
+"""Rich messages through the bot: /new, the edits in My posts, the preview, /broadcast, the userbot, /swap, the sync and /repost."""
+import asyncio
+from types import SimpleNamespace
+
+from telethon import functions, types
+
+from app.replace_engine import ScanOptions, apply_channel, scan_channel, undo_batch
+from app.repost_engine import RepostOptions, final_state, run_repost, stand_in
+from app.richmsg import Rich, entry, fingerprint, rich_of
+from app.suggest import build_suggestion
+from app.sync_engine import diff_post, live_fields, message_supported, sync_channel
+from app.tgutil import button_url, pack_file_id
+
+from .fakes import FakeChannelClient, make_msg, markup, photo_media, rpc, url_btn, video_media
+from .harness import App, norm
+from .test_broadcast import broadcast, make_app as broadcast_app
+from .test_engine import setup as engine_setup
+from .test_flows import make_app
+from .test_markup_flows import msg, new_post
+from .test_replace_userbot import BUTTONS, CID, MID, make_app as userbot_app, replacement, user_edits
+from .test_repost import NEW, OLD, edit_requests, forwarded_ids, setup as repost_setup
+from .test_rich import blocks, rich_msg, same
+from .test_sync import old_msg, opts, world
+
+HEADING = "# Release notes\n\n- [x] fixed\n- [ ] todo\n\n| a | b |\n|---|---|\n| 1 | 2 |"
+TABLE = '<h2>Schedule</h2><table bordered><tr><th>Day</th><th>Show</th></tr><tr><td>Mon</td><td>Naruto</td></tr></table>'
+
+
+def native(rm=None, mid=77):
+    """A rich message sent to the bot (or forwarded to it)."""
+    return SimpleNamespace(
+        raw_text="", entities=[], media=None, grouped_id=None, forward=None, rich_message=rm or rich_msg(), id=mid
+    )
+
+
+def raw_edits(app):
+    """The messages.editMessage requests the bot sent, each once."""
+    out = []
+    for e in app.tg.edits:
+        r = e[2] if isinstance(e, tuple) and e[0] == "raw" else e
+        if isinstance(r, functions.messages.EditMessageRequest) and not any(r is o for o in out):
+            out.append(r)
+    return out
+
+
+def rich_sends(app):
+    return [e for e in app.tg.sent if e[0] == "rich"]
+
+
+def md_post(source):
+    return dict(text="x", entities=[entry(Rich("markdown", source))])
+
+
+# ============================================================================================== typed text: /new
+def test_markdown_with_a_heading_is_published_as_a_rich_message(tmp_path):
+    async def run():
+        app = await make_app(tmp_path)
+        pid, prompt = await new_post(app, HEADING)
+        assert "Rich message read from Markdown" in prompt and "/rich" in prompt and "buttons" in prompt.lower()
+        post = await app.db.get_post(pid)
+        assert post.status == "sent" and post.entities == [{"k": "rich", "f": "markdown", "s": HEADING}]
+        assert post.text.startswith("Release notes\n") and "☑ fixed" in post.text and "☐ todo" in post.text
+        kind, peer, inp, kw = app.tg.sent[-1]
+        assert kind == "rich" and isinstance(peer, types.InputPeerChannel) and peer.channel_id == 1
+        assert isinstance(inp, types.InputRichMessageMarkdown) and inp.markdown == HEADING and not inp.rtl
+        assert kw["reply_markup"] is None
+        live = app.tg.msgs[post.message_id]  # Telegram made a rich message of it
+        assert live.rich_message is not None and live.message == ""
+        assert [e for e in app.tg.sent if e[0] in ("text", "file")] == []  # nothing went out as an ordinary message
+        await app.db.close()
+
+    asyncio.run(run())
+
+
+def test_html_with_a_table_is_published_as_a_rich_message(tmp_path):
+    async def run():
+        app = await make_app(tmp_path)
+        pid, prompt = await new_post(app, TABLE)
+        assert "Rich message read from HTML" in prompt
+        post = await app.db.get_post(pid)
+        assert post.entities == [{"k": "rich", "f": "html", "s": TABLE}]
+        assert post.text == "Schedule\nDay | Show\nMon | Naruto"
+        kind, peer, inp, kw = app.tg.sent[-1]
+        assert kind == "rich" and isinstance(inp, types.InputRichMessageHTML) and inp.html == TABLE
+        await app.db.close()
+
+    asyncio.run(run())
+
+
+def test_the_rich_marker_turns_any_text_into_a_rich_message_with_its_options(tmp_path):
+    async def run():
+        app = await make_app(tmp_path)
+        pid, prompt = await new_post(app, "<rich rtl noautolink>\nhello **x** #tag @name")
+        assert "Rich message read from Markdown" in prompt
+        post = await app.db.get_post(pid)
+        assert post.text == "hello x #tag @name"
+        assert post.entities == [{"k": "rich", "f": "markdown", "s": "hello **x** #tag @name", "r": 1, "n": 1}]
+        kind, peer, inp, kw = app.tg.sent[-1]
+        assert isinstance(inp, types.InputRichMessageMarkdown) and inp.markdown == "hello **x** #tag @name"
+        assert inp.rtl is True and inp.noautolink is True
+        await app.db.close()
+
+    asyncio.run(run())
+
+
+def test_text_that_only_looks_a_little_like_markup_stays_an_ordinary_post(tmp_path):
+    async def run():
+        app = await make_app(tmp_path)
+        for text in ("#anime #naruto new episode", "Price: $5 and $6 - a == b - I <3 it\n- one\n- two\n---", "**bold** [x](https://t.me/x)"):
+            pid, prompt = await new_post(app, text)
+            assert "Rich message" not in prompt
+            assert rich_of(await app.db.get_post(pid)) is None
+            assert app.tg.sent[-1][0] == "text"
+        await app.db.close()
+
+    asyncio.run(run())
+
+
+def test_markup_input_off_keeps_typed_blocks_literal(tmp_path):
+    async def run():
+        app = App(tmp_path, markup_input=False)
+        await app.start()
+        await app.db.save_channel(1, 5, "Anime Channel", "animech", 1)
+        pid, prompt = await new_post(app, HEADING)
+        assert "Rich message" not in prompt
+        kind, peer, text, kw = app.tg.sent[-1]
+        assert kind == "text" and text == HEADING and kw["formatting_entities"] == []
+        # a rich message made in Telegram is not typed text: it is still kept
+        pid, prompt = await new_post(app, "", message=native())
+        assert "Rich message kept as it is" in prompt and app.tg.sent[-1][0] == "rich"
+        await app.db.close()
+
+    asyncio.run(run())
+
+
+def test_a_typed_rich_message_that_is_too_long_is_refused(tmp_path):
+    async def run():
+        app = await make_app(tmp_path)
+        await app.text("/new")
+        await app.text("# Title\n" + "x" * 33000, message=msg("# Title\n" + "x" * 33000))
+        assert "32768" in norm(app.out.last_text)
+        assert (await app.db.list_posts(1, None, 0, 10))[1] == 0  # nothing was saved
+        await app.db.close()
+
+    asyncio.run(run())
+
+
+# =================================================================================== a rich message sent to the bot
+def test_a_rich_message_sent_to_the_bot_is_published_block_for_block(tmp_path):
+    async def run():
+        app = await make_app(tmp_path)
+        rm = rich_msg(photos=[photo_media(5).photo], documents=[video_media(81).document], rtl=True)
+        pid, prompt = await new_post(app, "", message=native(rm))
+        assert "Rich message kept as it is" in prompt
+        post = await app.db.get_post(pid)
+        (e,) = post.entities
+        assert (e["k"], e["f"], e["r"], e["p"]) == ("rich", "blocks", 1, fingerprint(rm)) and post.text == "Title\nHello"
+        kind, peer, inp, kw = app.tg.sent[-1]
+        assert kind == "rich" and isinstance(inp, types.InputRichMessage) and inp.rtl
+        assert same(inp.blocks, rm.blocks)
+        assert [(p.id, p.access_hash, p.file_reference) for p in inp.photos] == [(5, 35, b"ref5")]
+        assert [d.id for d in inp.documents] == [81]
+        live = app.tg.msgs[post.message_id].rich_message
+        assert same(live.blocks, rm.blocks) and [p.id for p in live.photos] == [5]
+        await app.db.close()
+
+    asyncio.run(run())
+
+
+def test_a_rich_message_that_arrives_in_parts_is_asked_for_whole(tmp_path):
+    async def run():
+        app = await make_app(tmp_path)
+        part = rich_msg([types.PageBlockParagraph(types.TextPlain("first"))], part=True)
+        app.tg.full_rich = {77: rich_msg(blocks("whole"))}
+        pid, prompt = await new_post(app, "", message=native(part))
+        assert "Rich message kept as it is" in prompt
+        assert (await app.db.get_post(pid)).text == "Title\nwhole"
+        assert any(isinstance(r, functions.messages.GetRichMessageRequest) and r.id == 77 for r in app.tg.requests)
+        await app.db.close()
+
+    asyncio.run(run())
+
+
+def test_a_rich_message_that_cannot_be_had_whole_makes_no_post(tmp_path):
+    async def run():
+        app = await make_app(tmp_path)
+        part = rich_msg([types.PageBlockParagraph(types.TextPlain("first"))], part=True)
+        app.tg.full_rich = {77: rich_msg(blocks(), part=True)}
+        await app.text("/new")
+        await app.text("", message=native(part))
+        assert "part of that rich message" in norm(app.out.last_text)
+        assert (await app.db.list_posts(1, None, 0, 10))[1] == 0
+        assert app.ctx.state[app.uid]["mode"] == "new_content"  # still waiting for the message
+        await app.text("hello", message=msg("hello"))  # and a better one is taken
+        assert (await app.db.list_posts(1, None, 0, 10))[1] == 1
+        await app.db.close()
+
+    asyncio.run(run())
+
+
+# ================================================================================ buttons, preview, publishing
+def test_the_link_buttons_of_a_post_go_with_its_rich_message(tmp_path):
+    async def run():
+        app = await make_app(tmp_path)
+        await app.text("/new")
+        await app.text(HEADING, message=msg(HEADING))
+        await app.text("Download - https://t.me/x?start=a | Other - https://example.com", message=msg("x"))
+        pid = int(app.out.callback_data("Publish").split(":")[1])
+        await app.press(f"pb:{pid}")
+        await app.press(f"pby:{pid}")
+        kind, peer, inp, kw = app.tg.sent[-1]
+        rows = kw["reply_markup"].rows
+        assert kind == "rich" and [b.text for b in rows[0].buttons] == ["Download", "Other"]
+        assert [button_url(b) for b in rows[0].buttons] == ["https://t.me/x?start=a", "https://example.com"]
+        assert app.tg.msgs[(await app.db.get_post(pid)).message_id].reply_markup is not None
+        await app.db.close()
+
+    asyncio.run(run())
+
+
+def test_the_preview_of_a_rich_draft_goes_to_the_owner_and_not_to_the_channel(tmp_path):
+    async def run():
+        app = await make_app(tmp_path)
+        pid, _ = await new_post(app, HEADING, publish=False)
+        await app.press(f"pv:{pid}")
+        kind, peer, inp, kw = app.tg.sent[-1]
+        assert kind == "rich" and isinstance(peer, types.InputPeerUser) and inp.markdown == HEADING
+        assert app.tg.msgs == {} and (await app.db.get_post(pid)).status == "draft"
+        await app.db.close()
+
+    asyncio.run(run())
+
+
+def test_telegram_not_being_able_to_make_the_message_is_explained_in_the_preview_and_when_publishing(tmp_path):
+    async def run():
+        app = await make_app(tmp_path)
+        pid, _ = await new_post(app, "<rich>\n<tg-map lat=\"x\"/>", publish=False)
+        app.tg.fail["SendMessageRequest"] = rpc("MessageEmptyError")
+        await app.press(f"pv:{pid}")
+        assert "see /rich for the syntax" in norm(app.out.last_text)
+        await app.press(f"pb:{pid}")
+        await app.press(f"pby:{pid}")
+        assert "NOT published" in norm(app.out.last_text)
+        assert (await app.db.get_post(pid)).status == "draft" and app.tg.msgs == {}
+        await app.db.close()
+
+    asyncio.run(run())
+
+
+def test_the_panel_of_a_rich_post_has_no_media_or_link_preview(tmp_path):
+    async def run():
+        app = await make_app(tmp_path)
+        pid, _ = await new_post(app, HEADING, publish=False)
+        await app.press(f"po:{pid}")
+        text = norm(app.out.last_text)
+        assert "Type: 🧩 rich message (typed as Markdown)" in text and "Release notes" in text
+        labels = [b for row in app.out.last_buttons() for b in row]
+        assert "✏️ Text" in labels and "👁 Preview" in labels and "➕ Button" in labels
+        assert not any("Media" in b or "Link preview" in b for b in labels if b)
+        native_pid, _ = await new_post(app, "", message=native(), publish=False)
+        assert "made in Telegram" in norm((await _panel(app, native_pid)))
+        await app.db.close()
+
+    asyncio.run(run())
+
+
+async def _panel(app, pid):
+    await app.press(f"po:{pid}")
+    return app.out.last_text
+
+
+# ====================================================================================== editing in My posts
+def test_the_text_of_a_published_rich_post_is_changed_with_a_new_rich_message(tmp_path):
+    async def run():
+        app = await make_app(tmp_path)
+        pid, _ = await new_post(app, "# One\n\ntext")
+        before = app.tg.msgs[(await app.db.get_post(pid)).message_id].rich_message
+        await app.press(f"et:{pid}")
+        assert "/rich" in norm(app.out.last_text)
+        new = "# Two\n\n- [ ] a"
+        await app.text(new, message=msg(new))
+        (req,) = raw_edits(app)
+        assert isinstance(req.rich_message, types.InputRichMessageMarkdown) and req.rich_message.markdown == new
+        assert req.message is None and not req.entities
+        post = await app.db.get_post(pid)
+        assert post.entities == [{"k": "rich", "f": "markdown", "s": new}] and post.text == "Two\n\n☐ a"
+        assert bytes(app.tg.msgs[post.message_id].rich_message) != bytes(before)  # the channel shows the new one
+        assert "Rich message read from Markdown" in norm(app.out.last_text)
+        await app.db.close()
+
+    asyncio.run(run())
+
+
+def test_a_native_rich_message_can_replace_the_text_of_a_post(tmp_path):
+    async def run():
+        app = await make_app(tmp_path)
+        pid, _ = await new_post(app, "plain words")
+        await app.press(f"et:{pid}")
+        await app.text("", message=native(rich_msg(blocks("brand new"))))
+        (req,) = raw_edits(app)
+        assert isinstance(req.rich_message, types.InputRichMessage) and req.message is None
+        post = await app.db.get_post(pid)
+        assert rich_of(post).fmt == "blocks" and post.text == "Title\nbrand new"
+        assert "Rich message kept as it is" in norm(app.out.last_text)
+        await app.db.close()
+
+    asyncio.run(run())
+
+
+def test_a_plain_post_becomes_a_rich_one_and_a_rich_one_a_plain_one_again(tmp_path):
+    async def run():
+        app = await make_app(tmp_path)
+        pid, _ = await new_post(app, "hello")
+        await app.press(f"et:{pid}")
+        await app.text(HEADING, message=msg(HEADING))
+        assert isinstance(raw_edits(app)[-1].rich_message, types.InputRichMessageMarkdown)
+        assert rich_of(await app.db.get_post(pid)) is not None
+        await app.press(f"et:{pid}")
+        await app.text("plain again", message=msg("plain again"))
+        last = raw_edits(app)[-1]
+        assert last.message == "plain again" and last.rich_message is None
+        post = await app.db.get_post(pid)
+        assert rich_of(post) is None and post.text == "plain again" and post.entities == []
+        await app.db.close()
+
+    asyncio.run(run())
+
+
+def test_a_draft_can_be_changed_into_a_rich_message_before_it_is_published(tmp_path):
+    async def run():
+        app = await make_app(tmp_path)
+        pid, _ = await new_post(app, "first", publish=False)
+        await app.press(f"et:{pid}")
+        await app.text(TABLE, message=msg(TABLE))
+        post = await app.db.get_post(pid)
+        assert post.status == "draft" and rich_of(post).fmt == "html" and raw_edits(app) == []
+        await app.press(f"pb:{pid}")
+        await app.press(f"pby:{pid}")
+        assert app.tg.sent[-1][0] == "rich" and app.tg.sent[-1][2].html == TABLE
+        await app.db.close()
+
+    asyncio.run(run())
+
+
+def test_the_buttons_of_a_published_rich_post_are_changed_without_losing_the_message(tmp_path):
+    async def run():
+        app = await make_app(tmp_path)
+        pid, _ = await new_post(app, HEADING)
+        await app.press(f"ba:{pid}")
+        await app.text("Watch", message=msg("Watch"))
+        await app.text("https://example.com/w", message=msg("https://example.com/w"))
+        req = raw_edits(app)[-1]
+        assert isinstance(req.rich_message, types.InputRichMessageMarkdown) and req.rich_message.markdown == HEADING
+        assert [button_url(b) for b in req.reply_markup.rows[0].buttons] == ["https://example.com/w"]
+        post = await app.db.get_post(pid)
+        assert rich_of(post) is not None and post.buttons == [[{"t": "Watch", "u": "https://example.com/w"}]]
+        await app.db.close()
+
+    asyncio.run(run())
+
+
+def test_a_rich_post_has_no_media_slot(tmp_path):
+    async def run():
+        app = await make_app(tmp_path)
+        pid, _ = await new_post(app, HEADING)
+        await app.press(f"em:{pid}")
+        assert "has no media slot" in norm(app.out.last_text)
+        assert app.ctx.state.get(app.uid) is None or app.ctx.state[app.uid]["mode"] != "edit_media"
+        await app.db.close()
+
+    asyncio.run(run())
+
+
+def test_a_caption_is_never_a_rich_message(tmp_path):
+    async def run():
+        app = await make_app(tmp_path)
+        post = await app.db.create_post(
+            channel_id=1, message_id=7, status="sent", source="bot", text="old", entities=[], buttons=[],
+            media_kind="photo", media_file_id=pack_file_id(make_msg(1, "", media=photo_media(3))),
+        )
+        await app.press(f"et:{post.id}")
+        assert "Rich messages" not in norm(app.out.last_text)
+        await app.text("", message=native())  # a rich message as a caption
+        assert "caption can't be a rich message" in norm(app.out.last_text)
+        await app.text("# Heading\n\n- [ ] task", message=msg("# Heading\n\n- [ ] task"))  # typed blocks stay words
+        saved_post = await app.db.get_post(post.id)
+        assert rich_of(saved_post) is None and saved_post.text.startswith("# Heading") and saved_post.media_kind == "photo"
+        assert all(r.rich_message is None for r in raw_edits(app))
+        await app.db.close()
+
+    asyncio.run(run())
+
+
+def test_a_rich_message_can_not_be_made_a_media_post_by_the_media_edit(tmp_path):
+    async def run():
+        app = await make_app(tmp_path)
+        await app.text("/new")
+        await app.text("", message=SimpleNamespace(raw_text="", entities=[], media=photo_media(5), grouped_id=None, forward=None, rich_message=rich_msg()))
+        post = (await app.db.list_posts(1, None, 0, 10))[0][0]
+        assert post.media_kind == "photo" and rich_of(post) is None  # a photo wins: the message is not a rich post
+        await app.db.close()
+
+    asyncio.run(run())
+
+
+# =================================================================================================== /broadcast
+def test_a_broadcast_can_be_a_rich_message(tmp_path):
+    async def run():
+        app = await broadcast_app(tmp_path)
+        await app.text("/broadcast 1h")
+        assert "/rich" in norm(app.out.last_text)
+        text = "# Maintenance\n\n- tonight 22:00\n- [ ] backups"
+        await app.text(text, message=msg(text))
+        assert "Rich message read from Markdown" in norm(app.out.last_text)
+        await app.press(app.out.callback_data("No buttons"))
+        assert rich_sends(app)[-1][1].__class__ is types.InputPeerUser  # the preview was made as a rich message too
+        await app.press(app.out.callback_data("Send to all channels"))
+        for cid in (1, 2):
+            chan = app.tg.registry[cid]
+            (live,) = [m for m in chan.msgs.values() if m.rich_message is not None]
+            assert isinstance(live.rich_input, types.InputRichMessageMarkdown) and live.rich_input.markdown == text
+            (p,) = await app.db.sent_posts(cid)
+            assert p.message_id == live.id and p.entities == [{"k": "rich", "f": "markdown", "s": text}]
+            assert p.text.startswith("Maintenance")
+        assert len(await app.db.pending_deletes()) == 2  # it is still deleted when its time is up
+        await app.db.close()
+
+    asyncio.run(run())
+
+
+def test_a_broadcast_with_a_native_rich_message_and_buttons(tmp_path):
+    async def run():
+        app = await broadcast_app(tmp_path)
+        await broadcast(app, "/broadcast 1h", "", buttons="Open - https://t.me/x", message=native())
+        for cid in (1, 2):
+            (live,) = [m for m in app.tg.registry[cid].msgs.values() if m.rich_message is not None]
+            assert same(live.rich_message.blocks, blocks())
+            assert [button_url(b) for b in live.reply_markup.rows[0].buttons] == ["https://t.me/x"]
+        await app.db.close()
+
+    asyncio.run(run())
+
+
+# ======================================================================================================== /replace
+def test_replace_leaves_a_rich_post_alone(tmp_path):
+    async def run():
+        app, chan, user = await userbot_app(tmp_path)
+        live = chan.msgs[MID]
+        live.media, live.message, live.rich_message = None, "", rich_msg()  # the channel's post is a rich message
+        await app.text(f"/replace https://t.me/c/{CID}/{MID}", reply=replacement())
+        assert "is a rich message" in norm(app.out.last_text)
+        assert user_edits(user) == [] and not any(isinstance(r, functions.messages.EditMessageRequest) for r in chan.requests)
+        assert chan.msgs[MID].rich_message is not None
+        await app.db.close()
+
+    asyncio.run(run())
+
+
+# ================================================================================================ the userbot
+def test_a_rich_edit_that_telegram_refuses_the_bot_is_made_by_the_userbot(tmp_path):
+    async def run():
+        app, chan, user = await userbot_app(tmp_path)  # the post belongs to another bot: the bot may not edit it
+        live = chan.msgs[MID]
+        live.media, live.message, live.rich_message = None, "", chan._rich(types.InputRichMessageMarkdown(markdown="# old"))
+        post = await app.db.create_post(
+            channel_id=CID, message_id=MID, status="sent", source="adopted", text="old",
+            entities=[entry(Rich("markdown", "# old"))], buttons=BUTTONS,
+        )
+        await app.press(f"et:{post.id}")
+        new = "# New\n\n- [x] a"
+        await app.text(new, message=msg(new))
+        (req,) = user_edits(user)
+        assert isinstance(req.rich_message, types.InputRichMessageMarkdown) and req.rich_message.markdown == new
+        assert req.message is None and req.reply_markup is None  # a user account sends no keyboard
+        assert isinstance(chan.msgs[MID].rich_message, types.RichMessage) and chan.msgs[MID].rich_input.markdown == new
+        assert "the userbot made the edit" in norm(app.out.last_text) and "Rich message read from Markdown" in norm(app.out.last_text)
+        saved_post = await app.db.get_post(post.id)
+        assert saved_post.entities == [{"k": "rich", "f": "markdown", "s": new}] and saved_post.buttons == BUTTONS
+        await app.db.close()
+
+    asyncio.run(run())
+
+
+def test_the_userbot_editor_can_set_a_rich_message():
+    from .test_userbot import channel
+    from .test_userbot_edit import ready_editor
+
+    async def go():
+        bot = channel()
+        ub, made, e = await ready_editor(bot)
+        await e.edit(5, text="", entities=[], rich=Rich("html", "<h1>New</h1>"))
+        await e.close()
+        req = bot.user_edits[-1]
+        assert isinstance(req.rich_message, types.InputRichMessageHTML) and req.rich_message.html == "<h1>New</h1>"
+        assert req.message is None and bot.msgs[5].rich_message is not None and bot.msgs[5].message == ""
+
+    asyncio.run(go())
+
+
+# ============================================================================================ /swap and /handleswap
+OLD_LINK = "https://t.me/bharath?start=AAA"
+
+
+def live_rich_post(i=1, *, out=True):
+    m = make_msg(i, "", markup=markup([url_btn("Get", OLD_LINK)]), out=out)
+    m.rich_message = rich_msg(
+        [types.PageBlockParagraph(types.TextUrl(types.TextPlain("t.me/bharath"), "https://t.me/bharath?start=ZZZ", 0))]
+    )
+    return m
+
+
+def test_swap_changes_only_the_buttons_of_a_rich_post_and_undo_puts_them_back(tmp_path):
+    async def run():
+        live = live_rich_post()
+        db, ch, client = await engine_setup(tmp_path, {1: live})
+        typed = Rich("markdown", "# Title\n\n@bharath t.me/bharath")
+        await db.create_post(
+            channel_id=1, message_id=1, status="sent", source="bot", text="Title", entities=[entry(typed)],
+            buttons=[[{"t": "Get", "u": OLD_LINK}]],
+        )
+        o = ScanOptions("bharath", "goku")
+        scan = await scan_channel(client, ch, o, can_edit_others=True)
+        assert set(scan.infos) == {1}
+        await db.create_batch("b1", "bharath", "goku", 99)
+        res = await apply_channel(client, db, scan, o, "b1", 99, edit_delay=0)
+        assert (res.edited, res.failed) == (1, [])
+        req = client.edits[-1]
+        assert req.message is None and req.rich_message is None  # only the keyboard was sent
+        assert button_url(live.reply_markup.rows[0].buttons[0]) == "https://t.me/goku?start=AAA"
+        assert live.message == "" and live.rich_message is not None
+        p = await db.find_by_message(1, 1)
+        assert rich_of(p) == typed and p.text == "Title" and p.buttons == [[{"t": "Get", "u": "https://t.me/goku?start=AAA"}]]
+
+        ures = await undo_batch(client, db, "b1", edit_delay=0)
+        assert ures.restored == 1 and not ures.failed
+        assert client.edits[-1].message is None and client.edits[-1].rich_message is None
+        assert button_url(live.reply_markup.rows[0].buttons[0]) == OLD_LINK
+        p = await db.find_by_message(1, 1)
+        assert rich_of(p) == typed and p.text == "Title" and p.buttons == [[{"t": "Get", "u": OLD_LINK}]]
+        await db.close()
+
+    asyncio.run(run())
+
+
+def test_a_rich_post_nobody_saved_is_adopted_by_a_swap_as_its_blocks(tmp_path):
+    async def run():
+        live = live_rich_post()
+        db, ch, client = await engine_setup(tmp_path, {1: live})
+        o = ScanOptions("bharath", "goku")
+        scan = await scan_channel(client, ch, o, can_edit_others=True)
+        await db.create_batch("b1", "bharath", "goku", 99)
+        await apply_channel(client, db, scan, o, "b1", 99, edit_delay=0)
+        p = await db.find_by_message(1, 1)
+        r = rich_of(p)
+        assert p.source == "adopted" and r.fmt == "blocks" and r.fp == fingerprint(live.rich_message)
+        assert p.text == "t.me/bharath" and p.buttons == [[{"t": "Get", "u": "https://t.me/goku?start=AAA"}]]
+        await db.close()
+
+    asyncio.run(run())
+
+
+def test_swaps_leave_the_text_of_a_rich_post_alone(tmp_path):
+    async def run():
+        for mode in ("links", "handles"):
+            live = make_msg(1, "")
+            live.rich_message = rich_msg([types.PageBlockParagraph(types.TextPlain("@bharath https://t.me/bharath?start=A"))])
+            db, ch, client = await engine_setup(tmp_path, {1: live})
+            scan = await scan_channel(client, ch, ScanOptions("bharath", "goku", mode=mode), can_edit_others=True)
+            assert scan.infos == {}  # no button to change: nothing to do
+            await db.close()
+
+    asyncio.run(run())
+
+
+def test_a_handleswap_renames_the_buttons_of_a_rich_post_only(tmp_path):
+    async def run():
+        live = make_msg(1, "", markup=markup([url_btn("Join @bharath", "https://example.com")]), out=True)
+        live.rich_message = rich_msg()
+        db, ch, client = await engine_setup(tmp_path, {1: live})
+        o = ScanOptions("bharath", "goku", mode="handles")
+        scan = await scan_channel(client, ch, o, can_edit_others=True)
+        await db.create_batch("b1", "bharath", "goku", 99)
+        await apply_channel(client, db, scan, o, "b1", 99, edit_delay=0)
+        assert live.reply_markup.rows[0].buttons[0].text == "Join @goku" and client.edits[-1].message is None
+        assert live.rich_message is not None
+        await db.close()
+
+    asyncio.run(run())
+
+
+# ===================================================================================================== the sync
+class SavedLike:
+    """Anything with the Post fields."""
+
+    def __init__(self, m, **kw):
+        for k, v in live_fields(m).items():
+            setattr(self, k, v)
+        self.link_preview = False
+        for k, v in kw.items():
+            setattr(self, k, v)
+
+
+def rich_live(i, word="Hello", **kw):
+    m = old_msg(i, "", **kw)
+    m.rich_message = rich_msg(blocks(word))
+    return m
+
+
+def test_a_saved_rich_post_that_is_unchanged_shows_no_difference():
+    m = rich_live(1)
+    p = SavedLike(m)
+    d = diff_post(p, m)
+    assert d.fields == {} and not d.quiet and not d.buttons_missing
+    assert message_supported(m) and not message_supported(old_msg(2, ""))
+
+
+def test_a_rich_post_edited_in_the_channel_is_taken_over_as_its_new_blocks():
+    m = rich_live(1, "Hello")
+    p = SavedLike(m)
+    edited = rich_live(1, "Changed")
+    d = diff_post(p, edited)
+    assert set(d.fields) == {"text", "entities"} and d.fields["text"] == "Title\nChanged"
+    (e,) = d.fields["entities"]
+    assert e["f"] == "blocks" and e["p"] == fingerprint(edited.rich_message) and not d.quiet
+
+
+def test_the_bots_own_typed_rich_post_is_only_noted_the_first_time_the_sync_sees_it():
+    m = rich_live(1)
+    typed = Rich("markdown", HEADING)
+    p = SavedLike(m, text="Release notes", entities=[entry(typed)])
+    d = diff_post(p, m)
+    assert d.quiet and set(d.fields) == {"entities"}
+    (e,) = d.fields["entities"]
+    assert (e["f"], e["s"], e["p"]) == ("markdown", HEADING, fingerprint(m.rich_message))  # its source is kept, the fingerprint added
+    again = SavedLike(m, text="Release notes", entities=d.fields["entities"])
+    assert diff_post(again, m).fields == {}
+    changed = rich_live(1, "Edited by an admin")
+    assert set(diff_post(again, changed).fields) == {"text", "entities"}
+
+
+def test_a_rich_post_that_became_an_ordinary_one_or_the_other_way_round_is_followed():
+    rich = rich_live(1)
+    plain = old_msg(1, "now just words")
+    d = diff_post(SavedLike(rich), plain)
+    assert d.fields == {"text": "now just words", "entities": []}
+    d = diff_post(SavedLike(plain), rich)
+    assert d.fields["text"] == "Title\nHello" and d.fields["entities"][0]["f"] == "blocks"
+    empty = old_msg(1, "")  # Telegram shows nothing this time: that is not an edit
+    assert diff_post(SavedLike(rich), empty).fields == {}
+
+
+def test_the_sync_saves_a_rich_post_it_finds_and_leaves_it_alone_afterwards(tmp_path):
+    async def run():
+        chan, db, ch = await world(tmp_path, {1: old_msg(1, "plain"), 2: rich_live(2)}, save=False)
+        rep = await sync_channel(chan, db, ch, opts())
+        p = await db.find_by_message(1, 2)
+        assert p is not None and rich_of(p).fmt == "blocks" and p.text == "Title\nHello" and p.source == "adopted"
+        again = await sync_channel(chan, db, ch, opts())
+        assert again.edited == [] and again.adopted == [] and rep.error is None
+        chan.msgs[2].rich_message = rich_msg(blocks("Later"))  # somebody edits it in the channel
+        third = await sync_channel(chan, db, ch, opts())
+        assert third.edited == [2] and (await db.find_by_message(1, 2)).text == "Title\nLater"
+        await db.close()
+
+    asyncio.run(run())
+
+
+# ===================================================================================================== /repost
+def rich_channel():
+    welcome = make_msg(1, "Welcome")
+    same_post = make_msg(2, "")
+    same_post.rich_message = rich_msg(blocks("one"))
+    with_button = make_msg(3, "", markup=markup([url_btn("Get", f"https://t.me/{OLD}?start=AAA"), url_btn("Other", "https://example.com/x")]))
+    with_button.rich_message = rich_msg(blocks("two"), photos=[photo_media(5).photo])
+    return {1: welcome, 2: same_post, 3: with_button}
+
+
+def test_a_repost_copies_rich_posts_and_rebuilds_the_one_with_buttons(tmp_path):
+    async def run():
+        db, ch, chan, mig = await repost_setup(tmp_path, FakeChannelClient(rich_channel()), include_typed=True)
+        res = await run_repost(chan, db, ch, mig, 99, delay=0, settle_pause=0)
+        assert (res.copied_units, res.failed, res.aborted) == (3, [], None)
+        assert forwarded_ids(chan) == [[1], [2]] and edit_requests(chan) == []
+        new = {i: m for i, m in chan.msgs.items() if i > 3}
+        assert list(new) == [4, 5, 6]
+        assert new[4].message == "Welcome"
+        assert same(new[5].rich_message.blocks, blocks("one"))  # a copy by Telegram: still the same rich message
+        m = new[6]  # the post with buttons is created again, with the link swapped
+        assert isinstance(m.rich_input, types.InputRichMessage) and same(m.rich_input.blocks, blocks("two"))
+        assert [(p.id, p.file_reference) for p in m.rich_input.photos] == [(5, b"ref5")]
+        rows = m.reply_markup.rows
+        assert [button_url(b) for b in rows[0].buttons] == [f"https://t.me/{NEW}?start=AAA", "https://example.com/x"]
+        saved_row = await db.find_by_message(1, 6)
+        assert saved_row.text == "Title\ntwo" and rich_of(saved_row).fmt == "blocks"
+        assert saved_row.buttons[0][0]["u"] == f"https://t.me/{NEW}?start=AAA"
+        assert rich_of(await db.find_by_message(1, 5)) is not None
+        await db.close()
+
+    asyncio.run(run())
+
+
+def test_a_channel_that_forbids_forwarding_gets_its_rich_posts_rebuilt(tmp_path):
+    async def run():
+        chan = FakeChannelClient(rich_channel())
+        chan.restrict_forwards = True
+        db, ch, chan, mig = await repost_setup(tmp_path, chan)
+        res = await run_repost(chan, db, ch, mig, 99, delay=0, settle_pause=0)
+        assert (res.copied_units, res.failed) == (3, []) and forwarded_ids(chan) == [[1]]  # only the first, refused, try
+        new = {i: m for i, m in chan.msgs.items() if i > 3}
+        assert same(new[5].rich_input.blocks, blocks("one")) and same(new[6].rich_input.blocks, blocks("two"))
+        assert new[5].reply_markup is None and new[6].reply_markup is not None
+        await db.close()
+
+    asyncio.run(run())
+
+
+def test_a_post_held_back_by_telegram_is_rebuilt_from_its_saved_rich_message():
+    row = SimpleNamespace(
+        text="Release notes", entities=[entry(Rich("markdown", HEADING))], buttons=[], media_file_id=None, link_preview=False
+    )
+    live = make_msg(7, "This message couldn't be displayed on your device due to copyright infringement.")
+    final = final_state(stand_in(live, row), RepostOptions(OLD, NEW))
+    assert final.rich == Rich("markdown", HEADING) and final.text == "" and final.entities == []
+    assert final_state(make_msg(8, "plain"), RepostOptions(OLD, NEW)).rich is None
+
+
+# ============================================================================================== everything else
+def test_a_rich_post_is_kept_by_an_export_and_an_import(tmp_path):
+    async def run():
+        app = await make_app(tmp_path)
+        pid, _ = await new_post(app, HEADING)
+        data = await app.db.export_all()
+        (p,) = [x for x in data["posts"] if x["id"] == pid]
+        assert p["entities"] == [{"k": "rich", "f": "markdown", "s": HEADING}]
+        await app.db.close()
+
+    asyncio.run(run())
+
+
+def test_the_next_post_suggestion_leaves_rich_posts_out():
+    post = SimpleNamespace(
+        text="x", entities=[entry(Rich("markdown", HEADING))], buttons=[[{"t": "Download Episodes 1 to 20", "u": "https://t.me/x?start=a"}]],
+    )
+    assert build_suggestion(post) is None
+
+
+def test_the_rich_command_and_the_help_pages(tmp_path):
+    async def run():
+        app = await make_app(tmp_path)
+        await app.text("/rich")
+        said = [norm(t) for kind, t, _ in app.out.log if kind == "say"]
+        assert len(said) == 2 and "Rich messages" in said[0] and "<tg-map" in said[1] and "# Heading" in said[0]
+        await app.text("/help")
+        assert "/rich" in norm(app.out.last_text)
+        await app.text("/formatting")
+        assert "/rich" in norm(app.out.last_text)
+        await app.db.close()
+
+    asyncio.run(run())