From e98014242b14b416953c161b9b32cb2268959bd6 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 8 Sep 2026 23:47:14 +0800 Subject: [PATCH] fix: localize concise runtime failure notifications Co-Authored-By: Codex --- main.py | 54 +++++++++---------------- notifications/telegram.py | 21 +++++++++- tests/test_request_handling.py | 10 ++--- tests/test_runtime_notification_i18n.py | 53 ++++++++++++++++++++++++ 4 files changed, 97 insertions(+), 41 deletions(-) create mode 100644 tests/test_runtime_notification_i18n.py diff --git a/main.py b/main.py index 7972cf6..2469d5b 100644 --- a/main.py +++ b/main.py @@ -2,6 +2,8 @@ from __future__ import annotations +from notifications.telegram import build_translator + import os import re import traceback @@ -139,40 +141,24 @@ def _telegram_notification_targets() -> tuple[tuple[str, str], ...]: def _runtime_error_notification_message(exc: Exception) -> str: - error_text = _safe_exception_text(exc, include_type=True) - if len(error_text) > 1200: - error_text = error_text[:1197] + "..." - is_health_check = request.path == "/probe" - if str(os.getenv("QSL_NOTIFY_LANG") or os.getenv("NOTIFY_LANG") or "").strip().lower().startswith("zh"): - return "\n".join( - ( - "Firstrade 健康检查失败" if is_health_check else "Firstrade 策略运行失败", - f"服务: {os.getenv('K_SERVICE') or 'firstrade-quant-service'}", - f"版本: {os.getenv('K_REVISION') or ''}", - f"路由: {request.method} {request.path}", - f"策略: {os.getenv('STRATEGY_PROFILE') or ''}", - f"账户范围: {os.getenv('ACCOUNT_REGION') or ''}", - f"错误: {error_text}", - ) - ) - return "\n".join( - ( - "Firstrade health check failed" if is_health_check else "Firstrade strategy run failed", - f"service: {os.getenv('K_SERVICE') or 'firstrade-quant-service'}", - f"revision: {os.getenv('K_REVISION') or ''}", - f"route: {request.method} {request.path}", - f"strategy: {os.getenv('STRATEGY_PROFILE') or ''}", - f"account_scope: {os.getenv('ACCOUNT_REGION') or ''}", - f"error: {error_text}", - ) - ) + t = build_translator(os.getenv("QSL_NOTIFY_LANG") or os.getenv("NOTIFY_LANG")) + title = "runtime_probe_failure_title" if request.path == "/probe" else "runtime_failure_title" + strategy_name = os.getenv("STRATEGY_DISPLAY_NAME") or os.getenv("STRATEGY_PROFILE") or "Firstrade" + return "\n".join(( + t(title), + t("strategy_label", name=strategy_name), + t("runtime_failure_context", context=os.getenv('ACCOUNT_REGION') or os.getenv('K_SERVICE') or 'Firstrade'), + t("runtime_failure_result"), + t("runtime_failure_action"), + )) def _notify_runtime_error(exc: Exception) -> bool: + t = build_translator(os.getenv("QSL_NOTIFY_LANG") or os.getenv("NOTIFY_LANG")) targets = _telegram_notification_targets() if not targets: print( - "Firstrade runtime error notification skipped: no Telegram target configured.", + t("runtime_notification_missing_target"), flush=True, ) return False @@ -182,10 +168,11 @@ def _notify_runtime_error(exc: Exception) -> bool: for token, chat_id in targets: attempted = True try: - build_sender(token, chat_id)(message) - except Exception as send_exc: # pragma: no cover - build_sender normally handles this. + if build_sender(token, chat_id)(message) is False: + print(t("runtime_notification_delivery_failed"), flush=True) + except Exception: # pragma: no cover - build_sender normally handles this. print( - f"Firstrade runtime error Telegram send failed: {redact_sensitive_text(send_exc)}", + t("runtime_notification_delivery_failed"), flush=True, ) return attempted @@ -197,9 +184,8 @@ def _safe_exception_text(exc: Exception, *, include_type: bool = False) -> str: def _handle_strategy_run_exception(exc: Exception) -> bool: - print(f"Firstrade strategy run failed: {_safe_exception_text(exc, include_type=True)}", flush=True) - for line in traceback.format_exception(type(exc), exc, exc.__traceback__): - print(redact_sensitive_text(line.rstrip()), flush=True) + t = build_translator(os.getenv("QSL_NOTIFY_LANG") or os.getenv("NOTIFY_LANG")) + print(t("runtime_failure_log"), flush=True) return _notify_runtime_error(exc) diff --git a/notifications/telegram.py b/notifications/telegram.py index 77f1c55..0ae07d5 100644 --- a/notifications/telegram.py +++ b/notifications/telegram.py @@ -2,6 +2,8 @@ from __future__ import annotations +from quant_platform_kit.common.operational_notification_localization import resolve_operational_notification_locale + import re from collections.abc import Callable, Mapping from typing import Any @@ -146,6 +148,14 @@ def format_small_account_whole_share_bootstrap_notes( I18N = { "zh": { + "runtime_failure_title": "⚠️ Firstrade 策略运行失败", + "runtime_probe_failure_title": "⚠️ Firstrade 健康检查失败", + "runtime_failure_context": "运行目标：{context}", + "runtime_failure_result": "本次运行未正常结束，请查看最新执行报告。", + "runtime_failure_action": "下一步：检查账户会话及最新执行报告。", + "runtime_notification_missing_target": "异常通知未发送：未配置 Telegram 接收目标（notification_target_missing）", + "runtime_notification_delivery_failed": "异常通知发送失败（notification_delivery_failed）", + "runtime_failure_log": "策略运行失败（runtime_setup_failed）", "rebalance_title": "🔔 【调仓指令】", "heartbeat_title": "💓 【心跳检测】", "strategy_label": "🧭 策略: {name}", @@ -319,6 +329,14 @@ def format_small_account_whole_share_bootstrap_notes( "skip_symbols_reason": "{symbols}（{reason}）", }, "en": { + "runtime_failure_title": "⚠️ Firstrade strategy run failed", + "runtime_probe_failure_title": "⚠️ Firstrade health check failed", + "runtime_failure_context": "Target: {context}", + "runtime_failure_result": "The run did not finish successfully; check the latest execution report.", + "runtime_failure_action": "Next: Check the account session and the latest execution report.", + "runtime_notification_missing_target": "Runtime alert not sent: Telegram target is not configured (notification_target_missing)", + "runtime_notification_delivery_failed": "Runtime alert delivery failed (notification_delivery_failed)", + "runtime_failure_log": "Strategy run failed (runtime_setup_failed)", "rebalance_title": "🔔 【Rebalance Instruction】", "heartbeat_title": "💓 【Heartbeat】", "strategy_label": "🧭 Strategy: {name}", @@ -504,8 +522,7 @@ def format_small_account_whole_share_bootstrap_notes( def build_translator(lang: str | None) -> Callable[..., str]: - normalized = str(lang or "").lower() - active_lang = "zh" if normalized.startswith("zh") else "en" + active_lang = resolve_operational_notification_locale(lang) def translate(key: str, **kwargs) -> str: template = I18N[active_lang].get(key, I18N["en"].get(key, key)) diff --git a/tests/test_request_handling.py b/tests/test_request_handling.py index 3e0cb30..6767b23 100644 --- a/tests/test_request_handling.py +++ b/tests/test_request_handling.py @@ -455,7 +455,7 @@ def send(message): assert sent_messages[0][0] == "token-1" assert sent_messages[0][1] == "chat-1" assert "Firstrade health check failed" in sent_messages[0][2] - assert "RuntimeError: session denied" in sent_messages[0][2] + assert "RuntimeError: session denied" not in sent_messages[0][2] def test_run_endpoint_notifies_telegram_on_strategy_cycle_error(monkeypatch): @@ -493,8 +493,8 @@ def send(message): assert sent_messages[0][0] == "token-1" assert sent_messages[0][1] == "chat-1" assert "Firstrade strategy run failed" in sent_messages[0][2] - assert "ValueError: snapshot denied" in sent_messages[0][2] - assert "strategy: russell_top50_leader_rotation" in sent_messages[0][2] + assert "ValueError: snapshot denied" not in sent_messages[0][2] + assert "Strategy: russell_top50_leader_rotation" in sent_messages[0][2] def test_run_endpoint_error_notification_uses_chinese_copy(monkeypatch): @@ -526,7 +526,7 @@ def send(message): text = sent_messages[0][2] assert "Firstrade 策略运行失败" in text assert "策略: russell_top50_leader_rotation" in text - assert "错误: ValueError: snapshot denied" in text + assert "snapshot denied" not in text def test_run_endpoint_redacts_sensitive_error_text(monkeypatch): @@ -559,7 +559,7 @@ def send(message): assert response.status_code == 500 payload = response.get_json() assert "" in payload["error"] - assert "" in sent_messages[0][2] + assert "request failed" not in sent_messages[0][2] for raw_secret in ("supersecret123", "abcd1234efgh", "123456789:ABC", "key987654"): assert raw_secret not in payload["error"] assert raw_secret not in sent_messages[0][2] diff --git a/tests/test_runtime_notification_i18n.py b/tests/test_runtime_notification_i18n.py new file mode 100644 index 0000000..8b46bd5 --- /dev/null +++ b/tests/test_runtime_notification_i18n.py @@ -0,0 +1,53 @@ +import ast +import os +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from notifications.telegram import build_translator + + +@pytest.mark.parametrize("locale", ["zh-CN", "zh_TW", "ZH-hans", " zh "]) +def test_chinese_locale_variants_use_chinese(locale): + assert build_translator(locale)("strategy_label", name="test") == build_translator("zh")("strategy_label", name="test") + + +def runtime_function(name, **overrides): + tree = ast.parse((Path(__file__).resolve().parents[1] / "main.py").read_text()) + function = next(node for node in tree.body if isinstance(node, ast.FunctionDef) and node.name == name) + namespace = dict(os=os, request=SimpleNamespace(method="GET", path="/main"), + NOTIFY_LANG="zh-CN", SERVICE_NAME="Test service", SECRET_NAME="Test secret alias", + STRATEGY_PROFILE="test_strategy", ACCOUNT_REGION="SG", ACCOUNT_GROUP="Test account", + strategy_display_name="Test strategy", build_translator=build_translator, + t=build_translator("zh-CN")) + namespace["_safe_exception_text"] = lambda exc, **kwargs: str(exc) + namespace.update(overrides) + exec(compile(ast.Module(body=[function], type_ignores=[]), "main.py", "exec"), namespace) + return namespace[name] + + +@pytest.mark.parametrize("locale", ["zh-CN", "en"]) +def test_startup_alert_is_compact_localized_and_does_not_embed_provider_text(locale, monkeypatch): + monkeypatch.setenv("NOTIFY_LANG", locale) + monkeypatch.delenv("QSL_NOTIFY_LANG", raising=False) + monkeypatch.setenv("STRATEGY_PROFILE", "test_strategy") + fn = runtime_function("_runtime_error_notification_message", NOTIFY_LANG=locale, t=build_translator(locale)) + message = fn(RuntimeError("PRIVATE_PROVIDER_SENTINEL")) + assert len(message.splitlines()) <= 5 + assert "PRIVATE_PROVIDER_SENTINEL" not in message + assert "未提交订单" not in message + assert "no order" not in message.lower() + assert ("未正常结束" in message) is locale.startswith("zh") + assert ("did not finish successfully" in message) is (locale == "en") + assert "test_strategy" in message or "Test strategy" in message + + +def test_failed_fallback_delivery_logs_failure_and_preserves_attempted_semantics(capsys): + fn = runtime_function( + "_notify_runtime_error", _telegram_notification_targets=lambda: (("test", "test"),), + _runtime_error_notification_message=lambda exc: "safe failure", + build_sender=lambda token, chat: lambda message: False, + ) + assert fn(RuntimeError("private")) is True + assert "notification_delivery_failed" in capsys.readouterr().out