feat(telegram): block-level markdown to html, wikilinks, tables, quotes, fence-safe chunking
This commit is contained in:
@@ -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}"
|
||||
|
||||
@@ -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)
|
||||
_INLINE_CODE = re.compile(r"(`[^`\n]+`)")
|
||||
_HEADING = re.compile(r"^#{1,6}\s+(.+)$", re.MULTILINE)
|
||||
_BOLD = re.compile(r"\*\*(.+?)\*\*|__(.+?)__", re.DOTALL)
|
||||
_SNAP = 200
|
||||
|
||||
_FENCE_OPEN = re.compile(r"^\s*```(.*)$")
|
||||
_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] = []
|
||||
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()
|
||||
return "\n".join(_render(block) for block in _blocks(text)).strip()
|
||||
|
||||
|
||||
def _inline(text: str) -> str:
|
||||
parts = _INLINE_CODE.split(text)
|
||||
for i, part in enumerate(parts):
|
||||
if i % 2:
|
||||
parts[i] = f"<code>{html.escape(part[1:-1])}</code>"
|
||||
continue
|
||||
s = html.escape(part, quote=False)
|
||||
s = _HEADING.sub(r"<b>\1</b>", s)
|
||||
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)
|
||||
s = _LINK.sub(r'<a href="\2">\1</a>', s)
|
||||
parts[i] = _BULLET.sub(r"\1• ", s)
|
||||
return "".join(parts)
|
||||
def to_html_tail(text: str, tail_chars: int) -> str:
|
||||
start = max(len(text) - tail_chars, 0)
|
||||
if start:
|
||||
newline = text.find("\n", start, start + _SNAP)
|
||||
if newline >= 0:
|
||||
start = newline + 1
|
||||
opening = _open_fence(text[:start])
|
||||
tail = text[start:]
|
||||
return to_html(f"{opening}\n{tail}" if opening else tail)
|
||||
|
||||
|
||||
def chunks(text: str, limit: int = LIMIT) -> list[str]:
|
||||
text = text.strip()
|
||||
if len(text) <= limit:
|
||||
return [text] if text else []
|
||||
if not text:
|
||||
return []
|
||||
if _fits(text, limit):
|
||||
return [text]
|
||||
out: list[str] = []
|
||||
while len(text) > limit:
|
||||
cut = _cut_point(text, limit)
|
||||
out.append(text[:cut].rstrip())
|
||||
text = text[cut:].lstrip()
|
||||
if text:
|
||||
out.append(text)
|
||||
cur: list[str] = []
|
||||
for block in _blocks(text):
|
||||
for piece in _pieces(block, limit):
|
||||
if cur and not _fits("\n".join([*cur, piece]), limit):
|
||||
_flush(out, cur)
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user