From 1ee0740b136b6fbe07da922bebb520fedf181271 Mon Sep 17 00:00:00 2001 From: Your Name Date: Mon, 18 May 2026 14:12:08 +0800 Subject: [PATCH] fix(telegram): harden detail history html fallback --- apps/api/src/services/telegram_gateway.py | 35 +++++++++-- .../tests/test_telegram_message_templates.py | 63 +++++++++++++++++++ 2 files changed, 94 insertions(+), 4 deletions(-) diff --git a/apps/api/src/services/telegram_gateway.py b/apps/api/src/services/telegram_gateway.py index 6e43e050..a80e8cfb 100644 --- a/apps/api/src/services/telegram_gateway.py +++ b/apps/api/src/services/telegram_gateway.py @@ -342,7 +342,7 @@ def _telegram_html_chunks(lines: list[str], limit: int = _TELEGRAM_HTML_CHUNK_LI current = [] current_len = 0 if line_len > limit: - chunks.append(line[:limit]) + chunks.append(_html_safe_plain_chunk(line, limit=limit)) continue current.append(line) current_len += line_len @@ -357,6 +357,17 @@ def _plain_text_from_html(text: str, limit: int = 3900) -> str: return html.unescape(plain)[:limit] +def _html_safe_plain_chunk(text: str, limit: int) -> str: + """Render one overlong HTML line as parse-safe text for HTML mode chunks.""" + plain = _plain_text_from_html(text, limit=limit) + escaped = html.escape(plain) + if len(escaped) <= limit: + return escaped + # Escaping may expand &, <, >. Trim once more after escaping; a partial HTML + # entity is still plain text to Telegram, while a partial tag is not. + return escaped[:limit] + + def _sanitize_telegram_error(text: str) -> str: """遮蔽 Telegram Bot URL 中的 token，避免例外字串污染 log / trace。""" return _TELEGRAM_BOT_URL_RE.sub(r"\1", text) @@ -5920,13 +5931,29 @@ class TelegramGateway: Returns: dict: API 回應 """ + payload_text = text[:500] + payload_parse_mode = parse_mode + if parse_mode and parse_mode.upper() == "HTML" and len(text) > 500: + payload_text = _plain_text_from_html(text, limit=500) + payload_parse_mode = None + payload = { "chat_id": chat_id or self.alert_chat_id, - "text": text[:500], # SOUL.md 字數限制 - "parse_mode": parse_mode, + "text": payload_text, # SOUL.md 字數限制 } + if payload_parse_mode: + payload["parse_mode"] = payload_parse_mode - return await self._send_request("sendMessage", payload) + try: + return await self._send_request("sendMessage", payload) + except TelegramGatewayError as exc: + if payload_parse_mode and payload_parse_mode.upper() == "HTML" and "HTTP error: 400" in str(exc): + fallback_payload = { + "chat_id": chat_id or self.alert_chat_id, + "text": _plain_text_from_html(text, limit=500), + } + return await self._send_request("sendMessage", fallback_payload) + raise async def _send_html_line_message( self, diff --git a/apps/api/tests/test_telegram_message_templates.py b/apps/api/tests/test_telegram_message_templates.py index d631fee7..8b6ce6b8 100644 --- a/apps/api/tests/test_telegram_message_templates.py +++ b/apps/api/tests/test_telegram_message_templates.py @@ -71,6 +71,20 @@ def test_telegram_html_chunks_preserve_complete_lines() -> None: assert all(chunk.count("") == chunk.count("") for chunk in chunks) +def test_telegram_html_chunks_render_single_overlong_html_line_as_safe_text() -> None: + """單行過長時不得切出未閉合 ，否則 Telegram 會 400。""" + line = "告警鍵: " + ("node<188>&" * 80) + "" + + chunks = telegram_gateway_module._telegram_html_chunks([line], limit=120) + + assert len(chunks) == 1 + assert len(chunks[0]) <= 120 + assert "" not in chunks[0] + assert "" not in chunks[0] + assert "告警鍵" in chunks[0] + assert "<" in chunks[0] + + def test_awooop_runs_url_for_incident_uses_public_incident_filter() -> None: """Telegram URL button 必須導到公開 AwoooP Run list，並帶 incident filter。""" url = telegram_gateway_module._awooop_runs_url_for_incident("INC-20260514-F85F21") @@ -224,6 +238,55 @@ async def test_info_callback_sends_history_when_answer_callback_is_stale(monkeyp assert sent_history == ["INC-20260513-79ED5E"] +@pytest.mark.asyncio +async def test_send_notification_retries_plain_text_on_html_parse_400(monkeypatch): + """簡短 HTML 若被 Telegram 拒收，要用純文字重送，不回報成 HTTP 400。""" + gateway = TelegramGateway() + sent_requests = [] + + async def fake_send_request(method, payload): + sent_requests.append((method, payload)) + if len(sent_requests) == 1: + raise telegram_gateway_module.TelegramGatewayError("HTTP error: 400") + return {"ok": True} + + monkeypatch.setattr(gateway, "_send_request", fake_send_request) + + result = await gateway.send_notification( + "⚠️ 無法取得歷史統計: broken", + chat_id="chat", + ) + + assert result == {"ok": True} + assert sent_requests[0][1]["parse_mode"] == "HTML" + assert "parse_mode" not in sent_requests[1][1] + assert sent_requests[1][1]["text"] == "⚠️ 無法取得歷史統計: broken" + + +@pytest.mark.asyncio +async def test_send_notification_long_html_uses_plain_text_without_cutting_tags(monkeypatch): + """send_notification 的 500 字限制不可切壞 HTML tag。""" + gateway = TelegramGateway() + sent_requests = [] + + async def fake_send_request(method, payload): + sent_requests.append((method, payload)) + return {"ok": True} + + monkeypatch.setattr(gateway, "_send_request", fake_send_request) + long_text = "📊 事件歷史統計\n告警鍵: " + ("node<188>&" * 90) + "" + + await gateway.send_notification(long_text, chat_id="chat") + + assert len(sent_requests) == 1 + payload = sent_requests[0][1] + assert "parse_mode" not in payload + assert len(payload["text"]) <= 500 + assert "" not in payload["text"] + assert "" not in payload["text"] + assert "事件歷史統計" in payload["text"] + + class TestTelegramMessageFormat: """測試現有 TelegramMessage 格式化"""