feat(telegram): block-level markdown to html, wikilinks, tables, quotes, fence-safe chunking

src/beaver_gateway/frontends/telegram/drafts.py

@@ -21,7 +21,7 @@ from typing import TYPE_CHECKING
 
 from aiogram.exceptions import TelegramAPIError, TelegramBadRequest
 
-from beaver_gateway.frontends.telegram.render import to_html
+from beaver_gateway.frontends.telegram.render import to_html, to_html_tail
 
 if TYPE_CHECKING:
     from aiogram import Bot
@@ -112,12 +112,12 @@ class Draft:
     async def _push(self) -> None:
         self._dirty = False
         self._last_sent = time.monotonic()
-        text = self._render()
+        rendered, plain = self._render()
         try:
             try:
-                await self._send(to_html(text), "HTML")
+                await self._send(rendered, "HTML")
             except TelegramBadRequest:
-                await self._send(text, None)
+                await self._send(plain, None)
         except TelegramAPIError as exc:
             self._broken = True
             _log.warning(
@@ -133,6 +133,9 @@ class Draft:
             parse_mode=parse_mode,
         )
 
-    def _render(self) -> str:
+    def _render(self) -> tuple[str, str]:
         tail = self.text[-_TAIL:]
-        return f"{self.status}\n\n{tail}" if tail.strip() else self.status
+        if not tail.strip():
+            return to_html(self.status), self.status
+        rendered = f"{to_html(self.status)}\n\n{to_html_tail(self.text, _TAIL)}"
+        return rendered, f"{self.status}\n\n{tail}"

src/beaver_gateway/frontends/telegram/render.py

@@ -4,20 +4,38 @@ from __future__ import annotations
 
 import html
 import re
-from typing import Any
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Any, Literal
 
-__all__ = ["LIMIT", "chunks", "status_label", "to_html"]
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+__all__ = ["LIMIT", "chunks", "status_label", "to_html", "to_html_tail"]
 
 LIMIT = 4000
-_FENCE = re.compile(r"```[^\n]*\n(.*?)(?:```|$)", re.DOTALL)
+_SNAP = 200
-_INLINE_CODE = re.compile(r"(`[^`\n]+`)")
+
-_HEADING = re.compile(r"^#{1,6}\s+(.+)$", re.MULTILINE)
+_FENCE_OPEN = re.compile(r"^\s*```(.*)$")
-_BOLD = re.compile(r"\*\*(.+?)\*\*|__(.+?)__", re.DOTALL)
+_FENCE_CLOSE = re.compile(r"^\s*```\s*$")
+_LANG = re.compile(r"[\w+#.-]+")
+_QUOTE = re.compile(r"^\s*(?:>\s?)+")
+_TABLE_SEP = re.compile(r"^\s*\|?\s*:?-+:?\s*(?:\|\s*:?-+:?\s*)*\|?\s*$")
+_CELL_SPLIT = re.compile(r"(?<!\\)\|")
+_HEADING = re.compile(r"^\s{0,3}#{1,6}\s+(.+?)\s*#*\s*$")
+_RULE = re.compile(r"^\s*([-*_])(?:\s*\1){2,}\s*$")
+_TASK = re.compile(r"^(\s*)[-*+]\s+\[([ xX])\]\s*(.*)$")
+_BULLET = re.compile(r"^(\s*)[-*+]\s+(.*)$")
+_NUMBERED = re.compile(r"^(\s*)(\d+)[.)]\s+(.*)$")
+_TOKEN = re.compile(
+    r"`[^`\n]+`|\[\[[^\]\n]+\]\]|!?\[[^\]]*\]\((?:https?|tg)://[^)\s]+\)"
+)
+_WIKILINK = re.compile(r"\[\[([^\]|#]+)(?:#([^\]|]+))?(?:\|([^\]]+))?\]\]")
+_LINK = re.compile(r"(!?)\[([^\]]*)\]\(([^)\s]+)\)")
+_BOLD = re.compile(r"\*\*(.+?)\*\*|__(.+?)__")
 _ITALIC = re.compile(r"(?<![\w*])\*(?!\s)(.+?)(?<!\s)\*(?![\w*])")
 _ITALIC_U = re.compile(r"(?<![\w_])_(?!\s)(.+?)(?<!\s)_(?![\w_])")
-_LINK = re.compile(r"\[([^\]]+)\]\((https?://[^)\s]+)\)")
-_BULLET = re.compile(r"^(\s*)[-*]\s+", re.MULTILINE)
 _STRIKE = re.compile(r"~~(.+?)~~")
+_SPOILER = re.compile(r"\|\|(.+?)\|\|")
 
 _LABELS: dict[str, str] = {
     "Read": "читаю vault",
@@ -41,57 +59,47 @@ _LABELS: dict[str, str] = {
     "mcp__gateway__inject": "передаю в другой разговор",
 }
 
+_Kind = Literal["fence", "quote", "table", "text"]
+
+
+@dataclass(slots=True)
+class _Block:
+    kind: _Kind
+    lines: list[str]
+
 
 def to_html(text: str) -> str:
-    out: list[str] = []
+    return "\n".join(_render(block) for block in _blocks(text)).strip()
-    pos = 0
-    for match in _FENCE.finditer(text):
-        out.append(_inline(text[pos : match.start()]))
-        out.append(f"<pre>{html.escape(match.group(1).rstrip())}</pre>")
-        pos = match.end()
-    out.append(_inline(text[pos:]))
-    return "".join(out).strip()
 
 
-def _inline(text: str) -> str:
+def to_html_tail(text: str, tail_chars: int) -> str:
-    parts = _INLINE_CODE.split(text)
+    start = max(len(text) - tail_chars, 0)
-    for i, part in enumerate(parts):
+    if start:
-        if i % 2:
+        newline = text.find("\n", start, start + _SNAP)
-            parts[i] = f"<code>{html.escape(part[1:-1])}</code>"
+        if newline >= 0:
-            continue
+            start = newline + 1
-        s = html.escape(part, quote=False)
+    opening = _open_fence(text[:start])
-        s = _HEADING.sub(r"<b>\1</b>", s)
+    tail = text[start:]
-        s = _BOLD.sub(lambda m: f"<b>{m.group(1) or m.group(2)}</b>", s)
+    return to_html(f"{opening}\n{tail}" if opening else tail)
-        s = _ITALIC.sub(r"<i>\1</i>", s)
-        s = _ITALIC_U.sub(r"<i>\1</i>", s)
-        s = _STRIKE.sub(r"<s>\1</s>", s)
-        s = _LINK.sub(r'<a href="\2">\1</a>', s)
-        parts[i] = _BULLET.sub(r"\1• ", s)
-    return "".join(parts)
 
 
 def chunks(text: str, limit: int = LIMIT) -> list[str]:
     text = text.strip()
-    if len(text) <= limit:
+    if not text:
-        return [text] if text else []
+        return []
+    if _fits(text, limit):
+        return [text]
     out: list[str] = []
-    while len(text) > limit:
+    cur: list[str] = []
-        cut = _cut_point(text, limit)
+    for block in _blocks(text):
-        out.append(text[:cut].rstrip())
+        for piece in _pieces(block, limit):
-        text = text[cut:].lstrip()
+            if cur and not _fits("\n".join([*cur, piece]), limit):
-    if text:
+                _flush(out, cur)
-        out.append(text)
+            cur.append(piece)
+    _flush(out, cur)
     return out
 
 
-def _cut_point(text: str, limit: int) -> int:
-    for sep in ("\n\n", "\n", ". ", " "):
-        cut = text.rfind(sep, limit // 2, limit)
-        if cut > 0:
-            return cut + len(sep)
-    return limit
-
-
 def status_label(name: str, tool_input: dict[str, Any] | None = None) -> str:
     label = _LABELS.get(name)
     if label is not None:
@@ -109,3 +117,245 @@ def status_label(name: str, tool_input: dict[str, Any] | None = None) -> str:
                 hint = value.strip().splitlines()[0][:60]
                 break
     return f"{name} {hint}" if hint else name
+
+
+def _blocks(text: str) -> list[_Block]:
+    lines = text.split("\n")
+    blocks: list[_Block] = []
+    i = 0
+    while i < len(lines):
+        line = lines[i]
+        if _FENCE_OPEN.match(line):
+            j = i + 1
+            while j < len(lines) and not _FENCE_CLOSE.match(lines[j]):
+                j += 1
+            end = min(j + 1, len(lines))
+            blocks.append(_Block("fence", lines[i:end]))
+            i = end
+        elif _QUOTE.match(line):
+            j = i + 1
+            while j < len(lines) and _QUOTE.match(lines[j]):
+                j += 1
+            blocks.append(_Block("quote", lines[i:j]))
+            i = j
+        elif _is_table_start(lines, i):
+            j = i + 2
+            while j < len(lines) and "|" in lines[j] and lines[j].strip():
+                j += 1
+            blocks.append(_Block("table", lines[i:j]))
+            i = j
+        elif not line.strip():
+            blocks.append(_Block("text", [line]))
+            i += 1
+        else:
+            j = i + 1
+            while j < len(lines) and _is_plain(lines, j):
+                j += 1
+            blocks.append(_Block("text", lines[i:j]))
+            i = j
+    return blocks
+
+
+def _is_table_start(lines: list[str], i: int) -> bool:
+    if "|" not in lines[i] or i + 1 >= len(lines):
+        return False
+    sep = lines[i + 1]
+    return "|" in sep and _TABLE_SEP.match(sep) is not None
+
+
+def _is_plain(lines: list[str], i: int) -> bool:
+    line = lines[i]
+    return bool(
+        line.strip()
+        and not _FENCE_OPEN.match(line)
+        and not _QUOTE.match(line)
+        and not _is_table_start(lines, i)
+    )
+
+
+def _open_fence(text: str) -> str | None:
+    opening: str | None = None
+    for line in text.split("\n"):
+        if opening is None:
+            if _FENCE_OPEN.match(line):
+                opening = line
+        elif _FENCE_CLOSE.match(line):
+            opening = None
+    return opening
+
+
+def _render(block: _Block) -> str:
+    if block.kind == "fence":
+        return _render_fence(block.lines)
+    if block.kind == "quote":
+        body = "\n".join(_line(_QUOTE.sub("", line, 1)) for line in block.lines)
+        return f"<blockquote>{body}</blockquote>" if body.strip() else ""
+    if block.kind == "table":
+        return _render_table(block.lines)
+    return "\n".join(_line(line) for line in block.lines)
+
+
+def _fence_parts(lines: list[str]) -> tuple[str, list[str]]:
+    body = lines[1:]
+    if body and _FENCE_CLOSE.match(body[-1]):
+        body = body[:-1]
+    return lines[0], body
+
+
+def _render_fence(lines: list[str]) -> str:
+    head, body = _fence_parts(lines)
+    code = html.escape("\n".join(body).rstrip())
+    if not code:
+        return ""
+    lang = _language(head)
+    if lang:
+        return f'<pre><code class="language-{lang}">{code}</code></pre>'
+    return f"<pre>{code}</pre>"
+
+
+def _language(head: str) -> str:
+    match = _FENCE_OPEN.match(head)
+    info = match.group(1).strip() if match else ""
+    token = info.split()[0] if info else ""
+    return token if _LANG.fullmatch(token) else ""
+
+
+def _render_table(lines: list[str]) -> str:
+    rows = [_cells(line) for i, line in enumerate(lines) if i != 1]
+    width = max(len(row) for row in rows)
+    rows = [[*row, *[""] * (width - len(row))] for row in rows]
+    widths = [max(len(row[c]) for row in rows) for c in range(width)]
+    text = "\n".join(_table_row(row, widths) for row in rows)
+    return f"<pre>{html.escape(text)}</pre>"
+
+
+def _table_row(row: list[str], widths: list[int]) -> str:
+    cells = (cell.ljust(w) for cell, w in zip(row, widths, strict=True))
+    return " | ".join(cells).rstrip(" |")
+
+
+def _cells(line: str) -> list[str]:
+    inner = line.strip()
+    inner = inner.removeprefix("|").removesuffix("|")
+    return [cell.strip().replace("\\|", "|") for cell in _CELL_SPLIT.split(inner)]
+
+
+def _line(line: str) -> str:
+    if _RULE.match(line):
+        return "———"
+    if match := _HEADING.match(line):
+        return f"<b>{_inline(match.group(1))}</b>"
+    if match := _TASK.match(line):
+        box = "☐" if match.group(2) == " " else "☑"
+        return f"{match.group(1)}{box} {_inline(match.group(3))}"
+    if match := _BULLET.match(line):
+        return f"{match.group(1)}• {_inline(match.group(2))}"
+    if match := _NUMBERED.match(line):
+        return f"{match.group(1)}{match.group(2)}. {_inline(match.group(3))}"
+    return _inline(line)
+
+
+def _inline(text: str) -> str:
+    out: list[str] = []
+    pos = 0
+    for match in _TOKEN.finditer(text):
+        out.append(_format(text[pos : match.start()]))
+        out.append(_token(match.group()))
+        pos = match.end()
+    out.append(_format(text[pos:]))
+    return "".join(out)
+
+
+def _token(token: str) -> str:
+    if token.startswith("`"):
+        return f"<code>{html.escape(token[1:-1])}</code>"
+    if token.startswith("[["):
+        return _wikilink(token)
+    match = _LINK.fullmatch(token)
+    if match is None:
+        return html.escape(token, quote=False)
+    href = html.escape(match.group(3), quote=True)
+    label = _format(match.group(2)) or href
+    return f'<a href="{href}">{label}</a>'
+
+
+def _wikilink(token: str) -> str:
+    match = _WIKILINK.fullmatch(token)
+    if match is None:
+        return html.escape(token, quote=False)
+    note, heading, alias = (g.strip() if g else "" for g in match.groups())
+    label = alias or (f"{note} › {heading}" if heading else note)  # noqa: RUF001
+    return f"<u>{html.escape(label, quote=False)}</u>"
+
+
+def _format(text: str) -> str:
+    s = html.escape(text, quote=False)
+    s = _BOLD.sub(lambda m: f"<b>{m.group(1) or m.group(2)}</b>", s)
+    s = _ITALIC.sub(r"<i>\1</i>", s)
+    s = _ITALIC_U.sub(r"<i>\1</i>", s)
+    s = _STRIKE.sub(r"<s>\1</s>", s)
+    return _SPOILER.sub(r"<tg-spoiler>\1</tg-spoiler>", s)
+
+
+def _fits(text: str, limit: int) -> bool:
+    return len(to_html(text)) <= limit
+
+
+def _flush(out: list[str], cur: list[str]) -> None:
+    chunk = "\n".join(cur).strip()
+    if chunk:
+        out.append(chunk)
+    cur.clear()
+
+
+def _pieces(block: _Block, limit: int) -> list[str]:
+    whole = "\n".join(block.lines)
+    if _fits(whole, limit):
+        return [whole]
+    if block.kind == "fence":
+        head, body = _fence_parts(block.lines)
+        return _pack(body, lambda ls: "\n".join([head, *ls, "```"]), limit)
+    if block.kind == "quote":
+        inner = [_QUOTE.sub("", line, 1) for line in block.lines]
+        return _pack(inner, lambda ls: "\n".join(f"> {s}" for s in ls), limit)
+    if block.kind == "table":
+        header = block.lines[:2]
+        return _pack(block.lines[2:], lambda ls: "\n".join(header + ls), limit)
+    return _pack(block.lines, "\n".join, limit)
+
+
+def _pack(lines: list[str], wrap: Callable[[list[str]], str], limit: int) -> list[str]:
+    pieces: list[str] = []
+    cur: list[str] = []
+    for line in lines:
+        if cur and not _fits(wrap([*cur, line]), limit):
+            pieces.append(wrap(cur))
+            cur = []
+        if _fits(wrap([line]), limit):
+            cur.append(line)
+        else:
+            pieces.extend(wrap([part]) for part in _shatter(line, wrap, limit))
+    if cur:
+        pieces.append(wrap(cur))
+    return pieces
+
+
+def _shatter(line: str, wrap: Callable[[list[str]], str], limit: int) -> list[str]:
+    parts: list[str] = []
+    while line:
+        cut = len(line)
+        while cut > 1 and not _fits(wrap([line[:cut].rstrip()]), limit):
+            rendered = len(to_html(wrap([line[:cut]])))
+            cut = min(cut - 1, _cut_point(line, max(1, cut * limit // rendered)))
+        if part := line[:cut].rstrip():
+            parts.append(part)
+        line = line[cut:].lstrip()
+    return parts
+
+
+def _cut_point(text: str, limit: int) -> int:
+    for sep in ("\n\n", "\n", ". ", " "):
+        cut = text.rfind(sep, limit // 2, limit)
+        if cut > 0:
+            return cut + len(sep)
+    return limit