patrickandClaude Sonnet 5.5
cdcf7f673a
Telegram: Plain-Text-Fallback bei Parse-Fehler, tag-sichere Kürzung auf 4096. Matrix: bytebewusste Kürzung. Downtime: lesender Abgleich nach dem POST, generische Fehlermeldung im Chat. Agent: Platzhalter für nicht ausgeführte Tool-Calls, Text im letzten Schritt erzwingen, Zeitreserve, Kompaktierung, Limit pro Antwort. CLAUDE.md: Fallbacks für Phase 6d. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
204 lines
8.3 KiB
Python
204 lines
8.3 KiB
Python
import asyncio
|
|
import re
|
|
from html import unescape
|
|
|
|
import httpx
|
|
|
|
from ..constants import DEFAULT_TELEGRAM_API_BASE
|
|
from .base import Button, Buttons, ChatError, MessageRef, json_body, mask_secret
|
|
|
|
BACKEND_TELEGRAM = "telegram"
|
|
PARSE_MODE = "HTML"
|
|
DEFAULT_TIMEOUT = 15.0
|
|
MAX_RETRY_AFTER = 30.0
|
|
DEFAULT_RETRY_AFTER = 1.0
|
|
STATUS_TOO_MANY = 429
|
|
NOT_MODIFIED = "message is not modified"
|
|
ALLOWED_UPDATES = ["message", "callback_query"]
|
|
ANSWER_MAX = 200
|
|
ERROR_MAX = 300
|
|
MESSAGE_MAX = 4096
|
|
TRUNCATION_MARK = "…"
|
|
|
|
CANT_PARSE_ENTITIES = "can't parse entities"
|
|
URL_BUTTON_ERRORS = ("button_url_invalid", "keyboard button url")
|
|
|
|
TAG_OPEN_RE = re.compile(r"<([a-zA-Z][a-zA-Z0-9]*)[^>]*>")
|
|
TAG_CLOSE_RE = re.compile(r"</([a-zA-Z][a-zA-Z0-9]*)>")
|
|
TAG_OR_ENTITY_RE = re.compile(r"<[^>]*>?$|&[a-zA-Z0-9#]*;?$")
|
|
STRIP_TAGS_RE = re.compile(r"<[^>]+>")
|
|
|
|
|
|
def _strip_tags(text: str) -> str:
|
|
return unescape(STRIP_TAGS_RE.sub("", text))
|
|
|
|
|
|
def _close_open_tags(truncated: str) -> str:
|
|
stack: list[str] = []
|
|
pos = 0
|
|
while True:
|
|
open_match = TAG_OPEN_RE.search(truncated, pos)
|
|
close_match = TAG_CLOSE_RE.search(truncated, pos)
|
|
if open_match and (not close_match or open_match.start() < close_match.start()):
|
|
stack.append(open_match.group(1))
|
|
pos = open_match.end()
|
|
continue
|
|
if close_match:
|
|
if stack and stack[-1] == close_match.group(1):
|
|
stack.pop()
|
|
pos = close_match.end()
|
|
continue
|
|
break
|
|
closing = "".join(f"</{tag}>" for tag in reversed(stack))
|
|
return truncated + closing
|
|
|
|
|
|
def truncate_html(text: str, limit: int = MESSAGE_MAX) -> str:
|
|
if len(text) <= limit:
|
|
return text
|
|
mark_len = len(TRUNCATION_MARK)
|
|
cut = max(limit - mark_len, 0)
|
|
while cut > 0:
|
|
truncated = text[:cut]
|
|
# Nicht mitten in einem Tag oder einer &entity; aufhören, sonst wird das HTML ungültig.
|
|
boundary = TAG_OR_ENTITY_RE.search(truncated)
|
|
if boundary:
|
|
truncated = truncated[:boundary.start()]
|
|
closed = _close_open_tags(truncated) + TRUNCATION_MARK
|
|
if len(closed) <= limit:
|
|
return closed
|
|
cut -= 1
|
|
# Fallback: als Plain Text kürzen, wenn die HTML-sichere Kürzung das Limit trotzdem sprengt.
|
|
plain = _strip_tags(text)
|
|
return plain[:max(limit - mark_len, 0)] + TRUNCATION_MARK
|
|
|
|
|
|
def _button(b: Button) -> dict:
|
|
return {"text": b.label, "url": b.url} if b.url else {"text": b.label, "callback_data": b.data}
|
|
|
|
|
|
def _markup(buttons: Buttons) -> dict:
|
|
return {"inline_keyboard": [[_button(b) for b in row] for row in buttons]}
|
|
|
|
|
|
def _without_url_buttons(buttons: Buttons) -> Buttons:
|
|
rows = (tuple(b for b in row if not b.url) for row in buttons)
|
|
return tuple(row for row in rows if row)
|
|
|
|
|
|
class TelegramError(ChatError):
|
|
pass
|
|
|
|
|
|
class TelegramAdapter:
|
|
name = BACKEND_TELEGRAM
|
|
|
|
def __init__(self, token: str, default_target: str, api_base: str = DEFAULT_TELEGRAM_API_BASE,
|
|
timeout: float = DEFAULT_TIMEOUT, transport: httpx.AsyncBaseTransport | None = None,
|
|
sleep=asyncio.sleep):
|
|
self.default_target = default_target
|
|
self._token = token
|
|
self._api_base = api_base.rstrip("/")
|
|
self._sleep = sleep
|
|
self._client = httpx.AsyncClient(timeout=timeout, transport=transport)
|
|
|
|
def _mask(self, text: str) -> str:
|
|
return mask_secret(text, self._token)
|
|
|
|
async def _call(self, method: str, payload: dict | None = None) -> dict:
|
|
url = f"{self._api_base}/bot{self._token}/{method}"
|
|
for attempt in range(2):
|
|
try:
|
|
response = await self._client.post(url, json=payload or {})
|
|
except (httpx.HTTPError, httpx.InvalidURL) as exc:
|
|
raise TelegramError(self._mask(f"{type(exc).__name__}: {exc}")) from None
|
|
body = json_body(response)
|
|
if response.status_code == STATUS_TOO_MANY and attempt == 0:
|
|
retry_after = body.get("parameters", {}).get("retry_after", DEFAULT_RETRY_AFTER)
|
|
await self._sleep(min(float(retry_after), MAX_RETRY_AFTER))
|
|
continue
|
|
if response.status_code < 400 and body.get("ok"):
|
|
return body.get("result") or {}
|
|
description = str(body.get("description", ""))
|
|
raise TelegramError(self._mask(f"HTTP {response.status_code}: {description}")[:ERROR_MAX])
|
|
raise TelegramError("Telegram nicht erreichbar (Rate-Limit)")
|
|
|
|
async def _call_buttons(self, method: str, payload: dict, buttons: Buttons) -> dict:
|
|
try:
|
|
return await self._call(method, payload)
|
|
except TelegramError as exc:
|
|
# Telegram lehnt manche URLs (z. B. localhost) ab; ohne Link-Button soll der Alarm trotzdem ankommen.
|
|
if not any(m in str(exc).lower() for m in URL_BUTTON_ERRORS) or not any(b.url for row in buttons for b in row):
|
|
raise
|
|
remaining = _without_url_buttons(buttons)
|
|
payload = {k: v for k, v in payload.items() if k != "reply_markup"}
|
|
if remaining:
|
|
payload["reply_markup"] = _markup(remaining)
|
|
return await self._call(method, payload)
|
|
|
|
async def _send_plain_on_parse_error(self, method: str, payload: dict, buttons: Buttons,
|
|
html: bool) -> dict:
|
|
try:
|
|
return await self._call_buttons(method, payload, buttons)
|
|
except TelegramError as exc:
|
|
if not html or CANT_PARSE_ENTITIES not in str(exc).lower():
|
|
raise
|
|
plain_payload = {k: v for k, v in payload.items() if k != "parse_mode"}
|
|
plain_payload["text"] = _strip_tags(payload["text"])[:MESSAGE_MAX]
|
|
return await self._call_buttons(method, plain_payload, buttons)
|
|
|
|
async def send(self, target: str, text: str, buttons: Buttons = (), reply_to: str | None = None,
|
|
html: bool = True, silent: bool = False) -> MessageRef:
|
|
text = truncate_html(text) if html else text[:MESSAGE_MAX]
|
|
payload: dict = {"chat_id": target, "text": text, "disable_web_page_preview": True}
|
|
if silent:
|
|
payload["disable_notification"] = True
|
|
if html:
|
|
payload["parse_mode"] = PARSE_MODE
|
|
if buttons:
|
|
payload["reply_markup"] = _markup(buttons)
|
|
if reply_to:
|
|
payload["reply_parameters"] = {"message_id": int(reply_to), "allow_sending_without_reply": True}
|
|
result = await self._send_plain_on_parse_error("sendMessage", payload, buttons, html)
|
|
return MessageRef(self.name, str(target), str(result["message_id"]))
|
|
|
|
async def edit(self, ref: MessageRef, text: str, buttons: Buttons = (), silent: bool = False) -> None:
|
|
text = truncate_html(text)
|
|
payload: dict = {"chat_id": ref.target, "message_id": int(ref.message_id), "text": text,
|
|
"parse_mode": PARSE_MODE, "disable_web_page_preview": True}
|
|
if buttons:
|
|
payload["reply_markup"] = _markup(buttons)
|
|
try:
|
|
await self._send_plain_on_parse_error("editMessageText", payload, buttons, html=True)
|
|
except TelegramError as exc:
|
|
if NOT_MODIFIED not in str(exc).lower():
|
|
raise
|
|
|
|
async def get_updates(self, offset: int | None, timeout: int) -> list[dict]:
|
|
payload: dict = {"timeout": timeout, "allowed_updates": ALLOWED_UPDATES}
|
|
if offset is not None:
|
|
payload["offset"] = offset
|
|
result = await self._call("getUpdates", payload)
|
|
return result if isinstance(result, list) else []
|
|
|
|
async def answer_callback(self, callback_id: str, text: str = "") -> None:
|
|
payload = {"callback_query_id": callback_id}
|
|
if text:
|
|
payload["text"] = text[:ANSWER_MAX]
|
|
await self._call("answerCallbackQuery", payload)
|
|
|
|
async def clear_buttons(self, target: str, message_id: str) -> None:
|
|
await self._call("editMessageReplyMarkup", {
|
|
"chat_id": target, "message_id": int(message_id), "reply_markup": {"inline_keyboard": []}})
|
|
|
|
async def username(self) -> str | None:
|
|
result = await self._call("getMe")
|
|
return result.get("username")
|
|
|
|
async def check(self) -> str:
|
|
result = await self._call("getMe")
|
|
return f"Bot @{result.get('username', '?')}"
|
|
|
|
async def aclose(self) -> None:
|
|
await self._client.aclose()
|