347 lines
11 KiB
Python
347 lines
11 KiB
Python
"""Model markdown → Telegram HTML, chunking, and the status line for drafts."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import html
|
||
import re
|
||
from dataclasses import dataclass
|
||
from typing import TYPE_CHECKING, Any, Literal
|
||
|
||
from beaver_gateway.frontends.telegram.texts import TOOL_LABELS
|
||
|
||
if TYPE_CHECKING:
|
||
from collections.abc import Callable, Mapping
|
||
|
||
__all__ = ["LIMIT", "chunks", "status_label", "to_html", "to_html_tail"]
|
||
|
||
LIMIT = 4000
|
||
_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_])")
|
||
_STRIKE = re.compile(r"~~(.+?)~~")
|
||
_SPOILER = re.compile(r"\|\|(.+?)\|\|")
|
||
|
||
|
||
_Kind = Literal["fence", "quote", "table", "text"]
|
||
|
||
|
||
@dataclass(slots=True)
|
||
class _Block:
|
||
kind: _Kind
|
||
lines: list[str]
|
||
|
||
|
||
def to_html(text: str) -> str:
|
||
return "\n".join(_render(block) for block in _blocks(text)).strip()
|
||
|
||
|
||
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 not text:
|
||
return []
|
||
if _fits(text, limit):
|
||
return [text]
|
||
out: list[str] = []
|
||
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 status_label(
|
||
name: str,
|
||
tool_input: dict[str, Any] | None = None,
|
||
labels: Mapping[str, str] | None = None,
|
||
) -> str:
|
||
label = (labels if labels is not None else TOOL_LABELS).get(name)
|
||
if label is not None:
|
||
return label
|
||
if name.startswith("mcp__"):
|
||
parts = name.split("__", 2)
|
||
server = parts[1]
|
||
tool = parts[2] if len(parts) == 3 else ""
|
||
return f"{server}: {tool}" if tool else server
|
||
hint = ""
|
||
if tool_input:
|
||
for key in ("description", "command", "file_path", "pattern", "query"):
|
||
value = tool_input.get(key)
|
||
if isinstance(value, str) and value.strip():
|
||
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
|