diff --git a/src/beaver_gateway/frontends/telegram/drafts.py b/src/beaver_gateway/frontends/telegram/drafts.py index 8659aa9..15a4912 100644 --- a/src/beaver_gateway/frontends/telegram/drafts.py +++ b/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}" diff --git a/src/beaver_gateway/frontends/telegram/render.py b/src/beaver_gateway/frontends/telegram/render.py index 52e59f5..247db74 100644 --- a/src/beaver_gateway/frontends/telegram/render.py +++ b/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) -_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"(? str: - out: list[str] = [] - pos = 0 - for match in _FENCE.finditer(text): - out.append(_inline(text[pos : match.start()])) - out.append(f"
{html.escape(match.group(1).rstrip())}
") - 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"{html.escape(part[1:-1])}" - continue - s = html.escape(part, quote=False) - s = _HEADING.sub(r"\1", s) - s = _BOLD.sub(lambda m: f"{m.group(1) or m.group(2)}", s) - s = _ITALIC.sub(r"\1", s) - s = _ITALIC_U.sub(r"\1", s) - s = _STRIKE.sub(r"\1", s) - s = _LINK.sub(r'\1', 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"
{body}
" 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'
{code}
' + return f"
{code}
" + + +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"
{html.escape(text)}
" + + +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"{_inline(match.group(1))}" + 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"{html.escape(token[1:-1])}" + 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'{label}' + + +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"{html.escape(label, quote=False)}" + + +def _format(text: str) -> str: + s = html.escape(text, quote=False) + s = _BOLD.sub(lambda m: f"{m.group(1) or m.group(2)}", s) + s = _ITALIC.sub(r"\1", s) + s = _ITALIC_U.sub(r"\1", s) + s = _STRIKE.sub(r"\1", s) + return _SPOILER.sub(r"\1", 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