Files
wehub-resource-sync 4b6817381b
CI (OpenClaw E2E) / openclaw test (push) Has been cancelled
CI / coverage-report (push) Has been cancelled
CI / test-kubernetes (push) Has been cancelled
CI / should-run-thorough (push) Has been cancelled
CI / test-thorough (cloudwatch-demo) (push) Has been cancelled
CI / test-thorough (flink-ecs) (push) Has been cancelled
CI / test-thorough (upstream-lambda) (push) Has been cancelled
CI / test-thorough (prefect-ecs-fargate) (push) Has been cancelled
Release / build-binaries (zip, opensre.exe, onefile, windows-latest, windows-x64) (push) Has been cancelled
Benchmark image — build + push to ECR (any adapter) / build + push (push) Has been cancelled
CI / quality (ubuntu-latest) (push) Has been cancelled
CI / test (tools-runtime) (push) Has been cancelled
CI / test (e2e-general) (push) Has been cancelled
CI / test (cli-runtime) (push) Has been cancelled
CI / test (e2e-provider-and-openclaw) (push) Has been cancelled
CI / test (integrations-and-misc) (push) Has been cancelled
Release / verify (push) Has been cancelled
Release / build-python-dist (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, macos-15-intel, darwin-x64) (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, macos-latest, darwin-arm64) (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, ubuntu-22.04, linux-x64) (push) Has been cancelled
Release / publish-release (push) Has been cancelled
Release / publish-main-release (push) Has been cancelled
Interactive Shell Live (PR + post-merge) / turn-checks (no-LLM) (push) Has been cancelled
CodeQL / Analyze (python) (push) Has been cancelled
Interactive Shell Live (PR + post-merge) / turn-live shard ${{ matrix.shard_index }} (push) Has been cancelled
Release / prepare (push) Has been cancelled
Release / build-binaries (tar.gz, opensre, onedir, ubuntu-22.04-arm, linux-arm64) (push) Has been cancelled
Synthetic Deterministic Tests / Synthetic offline (deterministic) (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 13:10:45 +08:00

121 lines
4.6 KiB
Python

"""Convert agent Markdown and Slack mrkdwn into Telegram ``parse_mode=HTML``.
Telegram HTML supports a small tag subset (``<b> <i> <u> <s> <code> <pre> <a>``).
This module is the single source of truth for turning mixed Markdown / Slack text
into safe HTML. Callers should always send with a plain-text fallback when the API
rejects markup.
"""
from __future__ import annotations
import html
import re
_INLINE_CODE = re.compile(r"`([^`\n]+)`")
_MD_LINK = re.compile(r"\[([^\]]+)\]\((https?://[^\s)]+)\)")
_SLACK_LINK = re.compile(r"<(https?://[^|>]+)(?:\|([^>]+))?>")
# Alternation order matters: ``**``/``__`` spans win over Slack single-star bold.
_BOLD = re.compile(r"\*\*(.+?)\*\*|__(.+?)__|(?<!\*)\*([^*\n]+)\*(?!\*)")
_ITALIC = re.compile(r"(?<![\w_])_([^_\n]+?)_(?![\w_])")
_PLACEHOLDER = re.compile("\x00(\\d+)\x00")
_HEADER = re.compile(r"^#{1,6}\s+(.+)$")
_HRULE = re.compile(r"^(\*{3,}|-{3,}|_{3,})\s*$")
_LIST_ITEM = re.compile(r"^(\d+\.|[-*+•])\s+(.+)$")
_TELEGRAM_TAG = re.compile(r"<(?:b|i|u|s|code|pre|a)\b", re.IGNORECASE)
def looks_like_telegram_html(text: str) -> bool:
"""Return True when *text* already contains Telegram HTML markup."""
return bool(_TELEGRAM_TAG.search(text))
def render_markdown_as_telegram_html(text: str) -> str:
"""Convert mixed Markdown / Slack mrkdwn into Telegram-safe HTML."""
lines = text.splitlines()
out: list[str] = []
index = 0
while index < len(lines):
if lines[index].strip().startswith("```"):
index += 1
start = index
while index < len(lines) and not lines[index].strip().startswith("```"):
index += 1
code = "".join(f"{code_line}\n" for code_line in lines[start:index])
out.append(f"<pre>{html.escape(code, quote=False)}</pre>")
index += 1 # skip the closing fence (harmless past EOF)
elif _is_table_row(lines[index]):
rows: list[list[str]] = []
while index < len(lines) and _is_table_row(lines[index]):
rows.append([cell.strip() for cell in lines[index].strip()[1:-1].split("|")])
index += 1
out.extend(_table_lines(rows))
else:
out.append(_render_line(lines[index]))
index += 1
merged: list[str] = []
for part in out: # collapse blank runs left by rules and skipped rows
if part or not merged or merged[-1]:
merged.append(part)
return "\n".join(merged)
def _render_line(line: str) -> str:
if header := _HEADER.match(line):
return f"<b>{html.escape(header.group(1).strip())}</b>"
if _HRULE.match(line.strip()):
return ""
if item := _LIST_ITEM.match(line):
marker = item.group(1) if item.group(1).endswith(".") else "•"
return f"{marker} {_render_inline(item.group(2))}"
return _render_inline(line)
def _render_inline(line: str) -> str:
stash: list[str] = []
def keep(rendered: str) -> str:
stash.append(rendered)
return f"\x00{len(stash) - 1}\x00"
def link(label: str, url: str) -> str:
safe_label = label.replace("|", "¦").strip() or url
return f'<a href="{html.escape(url, quote=True)}">{html.escape(safe_label)}</a>'
# Stash rendered spans behind placeholders so escaping can't touch them.
text = _INLINE_CODE.sub(
lambda m: keep(f"<code>{html.escape(m.group(1), quote=False)}</code>"), line
)
text = _SLACK_LINK.sub(lambda m: keep(link(m.group(2) or m.group(1), m.group(1))), text)
text = _MD_LINK.sub(lambda m: keep(link(m.group(1), m.group(2))), text)
text = _BOLD.sub(
lambda m: keep(
f"<b>{html.escape(m.group(1) or m.group(2) or m.group(3), quote=False)}</b>"
),
text,
)
text = html.escape(text, quote=False)
text = _ITALIC.sub(r"<i>\1</i>", text)
return _PLACEHOLDER.sub(lambda m: stash[int(m.group(1))], text)
def _is_table_row(line: str) -> bool:
stripped = line.strip()
return stripped.startswith("|") and stripped.endswith("|") and "|" in stripped[1:-1]
def _table_lines(rows: list[list[str]]) -> list[str]:
"""Render table body rows (header dropped) as mobile-friendly bullets."""
lines: list[str] = []
for row in rows[1:]:
cells = [cell for cell in row if cell]
if not cells or all(set(cell) <= {"-", ":", " "} for cell in cells):
continue # separator row
head, *tail = (html.escape(cell) for cell in cells)
joined = " · ".join(tail)
lines.append(f"• <b>{head}</b> — {joined}" if joined else f"• {head}")
return lines
__all__ = ["looks_like_telegram_html", "render_markdown_as_telegram_html"]