Convert AI responses to Telegram HTML with blockquote wrapping
Markdown (headings, bold, italic, strikethrough, code blocks, links) is converted to Telegram-compatible HTML tags before sending. Each reply chunk is wrapped in <blockquote> for visual distinction. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -1,5 +1,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import html
|
||||
import re
|
||||
from textwrap import wrap
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
@@ -10,6 +12,40 @@ if TYPE_CHECKING:
|
||||
from app.commands.core import CommandContext
|
||||
|
||||
|
||||
_PH = "\x00{}\x00"
|
||||
_PH_RE = re.compile(r"\x00(\d+)\x00")
|
||||
|
||||
|
||||
def _md_to_tg(text: str) -> str:
|
||||
"""Convert markdown-ish text to Telegram HTML."""
|
||||
saved: list[str] = []
|
||||
|
||||
def _save(s: str) -> str:
|
||||
saved.append(s)
|
||||
return _PH.format(len(saved) - 1)
|
||||
|
||||
def _fence(m: re.Match) -> str:
|
||||
lang = (m.group(1) or "").strip()
|
||||
code = html.escape(m.group(2) or "")
|
||||
cls = f' class="language-{html.escape(lang)}"' if lang else ""
|
||||
return _save(f"<pre><code{cls}>{code}</code></pre>")
|
||||
|
||||
text = re.sub(r"```(\w*)\n?(.*?)```", _fence, text, flags=re.DOTALL)
|
||||
text = re.sub(r"`([^`\n]+)`", lambda m: _save(f"<code>{html.escape(m.group(1))}</code>"), text)
|
||||
|
||||
text = html.escape(text)
|
||||
|
||||
text = re.sub(r"^#{1,6}\s+(.+)$", r"<b>\1</b>", text, flags=re.MULTILINE)
|
||||
text = re.sub(r"\*\*(.+?)\*\*", r"<b>\1</b>", text, flags=re.DOTALL)
|
||||
text = re.sub(r"__(.+?)__", r"<b>\1</b>", text, flags=re.DOTALL)
|
||||
text = re.sub(r"(?<!\*)\*(?!\*)(.+?)(?<!\*)\*(?!\*)", r"<i>\1</i>", text, flags=re.DOTALL)
|
||||
text = re.sub(r"(?<!_)_(?!_)(.+?)(?<!_)_(?!_)", r"<i>\1</i>", text, flags=re.DOTALL)
|
||||
text = re.sub(r"~~(.+?)~~", r"<s>\1</s>", text, flags=re.DOTALL)
|
||||
text = re.sub(r"\[(.+?)\]\((.+?)\)", r'<a href="\2">\1</a>', text)
|
||||
|
||||
return _PH_RE.sub(lambda m: saved[int(m.group(1))], text)
|
||||
|
||||
|
||||
command_registry = CommandRegistry()
|
||||
DEFAULT_REGISTRY = command_registry
|
||||
|
||||
@@ -69,11 +105,13 @@ async def _reply_long(ctx: "CommandContext", text: str) -> None:
|
||||
if not chunks:
|
||||
await ctx.reply("")
|
||||
return
|
||||
first = await ctx.reply(chunks[0])
|
||||
for chunk in chunks[1:]:
|
||||
html_chunks = [f"<blockquote>{_md_to_tg(c)}</blockquote>" for c in chunks]
|
||||
first = await ctx.reply(html_chunks[0], parse_mode="html")
|
||||
for chunk in html_chunks[1:]:
|
||||
await ctx.client.send_message(
|
||||
ctx.message.chat.id,
|
||||
chunk,
|
||||
parse_mode="html",
|
||||
reply_to_message_id=getattr(first, "id", None),
|
||||
)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user