From d55fdaf7ee14bb23f1fb9b6c1d9b59ecc008ea03 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 8 Sep 2026 23:47:13 +0800 Subject: [PATCH] fix: localize concise runtime failure notifications Co-Authored-By: Codex --- main.py | 40 +++++++---------------- notifications/telegram.py | 18 ++++++++++- tests/test_request_handling.py | 8 ++--- tests/test_runtime_notification_i18n.py | 42 +++++++++++++++++++++++++ 4 files changed, 75 insertions(+), 33 deletions(-) create mode 100644 tests/test_runtime_notification_i18n.py diff --git a/main.py b/main.py index 5d77307..61fc14b 100644 --- a/main.py +++ b/main.py @@ -549,36 +549,20 @@ def _runtime_error_notification_targets() -> tuple[tuple[str, str], ...]: def _runtime_error_notification_message(exc: Exception, *, route_label: str) -> str: - error_text = type(exc).__name__ - if str(NOTIFY_LANG or "").strip().lower().startswith("zh"): - return "\n".join( - ( - "LongBridge 策略运行失败", - f"服务: {os.getenv('K_SERVICE') or SECRET_NAME or 'longbridge-platform'}", - f"版本: {os.getenv('K_REVISION') or ''}", - f"路由: {route_label}", - f"策略: {STRATEGY_PROFILE}", - f"账户范围: {ACCOUNT_REGION}", - f"错误: {error_text}", - ) - ) - return "\n".join( - ( - "LongBridge strategy run failed", - f"service: {os.getenv('K_SERVICE') or SECRET_NAME or 'longbridge-platform'}", - f"revision: {os.getenv('K_REVISION') or ''}", - f"route: {route_label}", - f"strategy: {STRATEGY_PROFILE}", - f"account_scope: {ACCOUNT_REGION}", - f"error: {error_text}", - ) - ) + title = "runtime_failure_title" + return "\n".join(( + t(title), + t("strategy_label", name=strategy_display_name or STRATEGY_PROFILE), + t("runtime_failure_context", context=ACCOUNT_REGION), + t("runtime_failure_result"), + t("runtime_failure_action"), + )) def _notify_runtime_error(exc: Exception, *, route_label: str) -> bool: targets = _runtime_error_notification_targets() if not targets: - print("LongBridge runtime error notification skipped: no Telegram target configured.", flush=True) + print(t("runtime_notification_missing_target"), flush=True) return False message = _runtime_error_notification_message(exc, route_label=route_label) outcomes = [] @@ -596,9 +580,9 @@ def _notify_runtime_error(exc: Exception, *, route_label: str) -> bool: if isinstance(payload, dict) and payload.get("ok") is False: acknowledged = False outcomes.append(acknowledged) - except Exception as send_exc: + except Exception: print( - f"LongBridge runtime error Telegram send failed: {type(send_exc).__name__}", + t("runtime_notification_delivery_failed"), flush=True, ) outcomes.append(False) @@ -606,7 +590,7 @@ def _notify_runtime_error(exc: Exception, *, route_label: str) -> bool: def _handle_route_runtime_error(exc: Exception, *, route_label: str): - print(f"LongBridge route failed before strategy-cycle handling: {type(exc).__name__}", flush=True) + print(t("runtime_failure_log"), flush=True) _notify_runtime_error(exc, route_label=route_label) return "Error", 500 diff --git a/notifications/telegram.py b/notifications/telegram.py index 37c17e3..eced395 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 Mapping from typing import Any @@ -41,6 +43,13 @@ def _break_telegram_market_symbol_auto_links(value) -> str: I18N = { "zh": { + "runtime_failure_title": "⚠️ LongBridge 策略运行失败", + "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": "🔔 【调仓指令】", "pending_order_title": "⏳ 【订单待券商最终确认】", "dry_run_banner": "🧪 模拟运行模式，本次不会真实下单", @@ -233,6 +242,13 @@ def _break_telegram_market_symbol_auto_links(value) -> str: "strategy_plugin_guidance_taco_rebound_shadow_taco_rebound_notify_manual_review": "TACO 仅提示可能的反弹窗口；可考虑小仓位、分批、预设止损/失效条件的人工博弈，不建议一次性满仓。", }, "en": { + "runtime_failure_title": "⚠️ LongBridge strategy run 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 account authorization, connectivity 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": "🔔 【Trade Execution Report】", "pending_order_title": "⏳ 【Broker Order Pending Confirmation】", "dry_run_banner": "🧪 Dry run mode, no real orders will be submitted", @@ -437,8 +453,8 @@ def _break_telegram_market_symbol_auto_links(value) -> str: def build_translator(lang): + active_lang = resolve_operational_notification_locale(lang) def translate(key, **kwargs): - active_lang = lang if lang in I18N else "en" template = I18N[active_lang].get(key, key) return template.format(**kwargs) if kwargs else template diff --git a/tests/test_request_handling.py b/tests/test_request_handling.py index 1a9ae92..51a7bed 100644 --- a/tests/test_request_handling.py +++ b/tests/test_request_handling.py @@ -361,7 +361,7 @@ def fake_post(_url, *, json, timeout): self.assertEqual(len(observed["payloads"]), 1) self.assertEqual(observed["payloads"][0][0]["chat_id"], "chat-1") self.assertIn("LongBridge strategy run failed", observed["payloads"][0][0]["text"]) - self.assertIn("RuntimeError", observed["payloads"][0][0]["text"]) + self.assertIn("did not finish successfully", observed["payloads"][0][0]["text"]) self.assertNotIn("boom", observed["payloads"][0][0]["text"]) def test_handle_trigger_runtime_error_fallback_uses_chinese_copy(self): @@ -387,8 +387,8 @@ def fake_post(_url, *, json, timeout): self.assertEqual(body, "Error") text = observed["payloads"][0][0]["text"] self.assertIn("LongBridge 策略运行失败", text) - self.assertIn("服务:", text) - self.assertIn("错误: RuntimeError", text) + self.assertIn("运行目标：", text) + self.assertIn("未正常结束", text) self.assertNotIn("boom", text) def test_handle_trigger_rejects_get_without_running_strategy(self): @@ -1255,7 +1255,7 @@ def _raise_sensitive(self, *args, **kwargs): def _assert_sanitized(self, output): self.assertNotIn(self.marker, output) self.assertNotIn("Traceback", output) - self.assertIn("RuntimeError", output) + self.assertTrue("RuntimeError" in output or "runtime_setup_failed" in output) def test_compact_and_route_failure_omit_exception_body(self): import io diff --git a/tests/test_runtime_notification_i18n.py b/tests/test_runtime_notification_i18n.py new file mode 100644 index 0000000..577eb4d --- /dev/null +++ b/tests/test_runtime_notification_i18n.py @@ -0,0 +1,42 @@ +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.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"), route_label="/main") + 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