112 lines
3.8 KiB
Python
112 lines
3.8 KiB
Python
"""Model markdown → Telegram HTML, chunking, and the status line for drafts."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import html
|
|
import re
|
|
from typing import Any
|
|
|
|
__all__ = ["LIMIT", "chunks", "status_label", "to_html"]
|
|
|
|
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)
|
|
_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"~~(.+?)~~")
|
|
|
|
_LABELS: dict[str, str] = {
|
|
"Read": "читаю vault",
|
|
"Glob": "ищу файлы",
|
|
"Grep": "ищу в vault",
|
|
"Edit": "правлю файл",
|
|
"Write": "пишу файл",
|
|
"MultiEdit": "правлю файлы",
|
|
"Bash": "выполняю команду",
|
|
"WebSearch": "ищу в сети",
|
|
"WebFetch": "читаю страницу",
|
|
"Task": "запустил сабагента",
|
|
"Agent": "запустил сабагента",
|
|
"AskUserQuestion": "спрашиваю",
|
|
"TodoWrite": "планирую",
|
|
"Skill": "открываю скилл",
|
|
"mcp__gateway__spawn": "открываю разговор",
|
|
"mcp__gateway__read_conversation": "читаю разговор",
|
|
"mcp__gateway__say": "говорю",
|
|
"mcp__gateway__schedule": "ставлю напоминание",
|
|
"mcp__gateway__inject": "передаю в другой разговор",
|
|
}
|
|
|
|
|
|
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()
|
|
|
|
|
|
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 chunks(text: str, limit: int = LIMIT) -> list[str]:
|
|
text = text.strip()
|
|
if len(text) <= limit:
|
|
return [text] if text else []
|
|
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)
|
|
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:
|
|
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
|