feat: localize operational runtime notifications

Co-Authored-By: Codex <noreply@openai.com>

Author: Pigbibi

.github/workflows/execution-report-heartbeat.yml

@@ -57,6 +57,7 @@ jobs:
       CLOUD_RUN_SERVICES: ${{ secrets.CLOUD_RUN_SERVICES }}
       CLOUD_RUN_SERVICE_TARGETS_JSON: ${{ vars.CLOUD_RUN_SERVICE_TARGETS_JSON || secrets.CLOUD_RUN_SERVICE_TARGETS_JSON }}
       CLOUD_SCHEDULER_MAIN_TIME: ${{ vars.CLOUD_SCHEDULER_MAIN_TIME }}
+      NOTIFY_LANG: ${{ vars.NOTIFY_LANG || 'zh' }}
       GLOBAL_TELEGRAM_CHAT_ID: ${{ secrets.GLOBAL_TELEGRAM_CHAT_ID }}
       TELEGRAM_TOKEN: ${{ secrets.TELEGRAM_TOKEN }}
       TELEGRAM_TOKEN_SECRET_NAME: ${{ vars.TELEGRAM_TOKEN_SECRET_NAME }}

.github/workflows/runtime-guard.yml

@@ -55,6 +55,7 @@ jobs:
       CLOUD_RUN_SERVICES: ${{ secrets.CLOUD_RUN_SERVICES }}
       CLOUD_RUN_SERVICE_TARGETS_JSON: ${{ vars.CLOUD_RUN_SERVICE_TARGETS_JSON || secrets.CLOUD_RUN_SERVICE_TARGETS_JSON }}
       CLOUD_RUN_REGION: ${{ vars.CLOUD_RUN_REGION }}
+      NOTIFY_LANG: ${{ vars.NOTIFY_LANG || 'zh' }}
       GLOBAL_TELEGRAM_CHAT_ID: ${{ secrets.GLOBAL_TELEGRAM_CHAT_ID }}
       TELEGRAM_TOKEN: ${{ secrets.TELEGRAM_TOKEN }}
       TELEGRAM_TOKEN_SECRET_NAME: ${{ vars.TELEGRAM_TOKEN_SECRET_NAME }}

scripts/cloud_run_runtime_guard.py

@@ -14,6 +14,12 @@
 import urllib.request
 from typing import Any
 
+from quant_platform_kit.common.operational_notification_localization import (
+    format_operational_alert,
+    operational_notification_text,
+    resolve_operational_notification_locale,
+)
+
 
 ERROR_SEVERITIES = {"ERROR", "CRITICAL", "ALERT", "EMERGENCY"}
 FAILURE_WORDS = (
@@ -60,6 +66,14 @@ def _env_bool(name: str, default: bool = False) -> bool:
     return value in {"1", "true", "yes", "y", "on"}
 
 
+def _notification_locale() -> str:
+    return resolve_operational_notification_locale(os.environ.get("NOTIFY_LANG"))
+
+
+def _notice(key: str, /, **values: object) -> str:
+    return operational_notification_text(_notification_locale(), key, **values)
+
+
 def _load_services() -> list[str]:
     services = []
     enabled_target_services = []
@@ -669,7 +683,8 @@ def main() -> int:
         services = _load_services()
     except RuntimeError as exc:
         services = []
-        issues.append(f"service configuration error: {exc}")
+        issues.append(_notice("runtime_guard_service_configuration_error"))
+        details.append(f"service configuration error: {exc}")
     scheduler_pattern = (
         os.environ.get("RUNTIME_GUARD_SCHEDULER_JOB_PATTERN")
         or _scheduler_job_pattern_for_services(services)
@@ -683,7 +698,8 @@ def main() -> int:
         try:
             entries = _run_gcloud_logging(project, log_filter, limit)
         except RuntimeError as exc:
-            issues.append(f"Cloud Run log query failed for {service}: {exc}")
+            issues.append(_notice("runtime_guard_cloud_run_log_query_failed", service=service))
+            details.append(f"Cloud Run log query failed for {service}: {exc}")
             continue
         queried_services.add(service)
         failures = [entry for entry in entries if _is_failure(entry)]
@@ -692,7 +708,13 @@ def main() -> int:
         success_count_by_service[service] = service_success_count
         success_count += service_success_count
         if failures:
-            issues.append(f"{len(failures)} Cloud Run failure log(s) for {service}")
+            issues.append(
+                _notice(
+                    "runtime_guard_cloud_run_failure_logs",
+                    count=len(failures),
+                    service=service,
+                )
+            )
             details.extend(_summarize(entry) for entry in failures[:5])
 
     if services and require_success:
@@ -702,8 +724,11 @@ def main() -> int:
             queried_services,
         ):
             issues.append(
-                f"no successful Cloud Run request found for {service} "
-                f"in the last {lookback_minutes} minutes"
+                _notice(
+                    "runtime_guard_no_successful_request",
+                    service=service,
+                    lookback_minutes=lookback_minutes,
+                )
             )
 
     if check_scheduler and scheduler_pattern:
@@ -732,10 +757,16 @@ def main() -> int:
                     continue
                 failures.append(entry)
             if failures:
-                issues.append(f"{len(failures)} Cloud Scheduler failure log(s)")
+                issues.append(
+                    _notice(
+                        "runtime_guard_scheduler_failure_logs",
+                        count=len(failures),
+                    )
+                )
                 details.extend(_summarize(entry) for entry in failures[:5])
         except RuntimeError as exc:
-            issues.append(f"Cloud Scheduler log query failed: {exc}")
+            issues.append(_notice("runtime_guard_scheduler_log_query_failed"))
+            details.append(f"Cloud Scheduler log query failed: {exc}")
     elif check_scheduler:
         print("Skipping Cloud Scheduler check because no scheduler job pattern could be derived.", file=sys.stderr)
 
@@ -752,18 +783,15 @@ def main() -> int:
             f"{os.environ['GITHUB_SERVER_URL']}/{os.environ['GITHUB_REPOSITORY']}"
             f"/actions/runs/{os.environ['GITHUB_RUN_ID']}"
         )
-    message_lines = [
-        f"[Runtime Guard] {name}",
-        f"Project: {project}",
-        f"Lookback: {lookback_minutes} minutes",
-        "Issues:",
-        *[f"- {issue}" for issue in issues],
-    ]
-    if details:
-        message_lines.extend(["Details:", *details[:10]])
-    if run_url:
-        message_lines.append(f"Workflow: {run_url}")
-    message = "\n".join(message_lines)
+    message = format_operational_alert(
+        locale=_notification_locale(),
+        alert_type="runtime_guard",
+        name=name,
+        context={"project": project, "lookback_minutes": lookback_minutes},
+        issues=issues,
+        technical_details=details[:10],
+        workflow_url=run_url,
+    )
     print(message)
     _send_telegram(message[:3900])
     return 1 if fail_workflow else 0

scripts/execution_report_heartbeat.py

@@ -14,6 +14,12 @@
 from typing import Any
 from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
 
+from quant_platform_kit.common.operational_notification_localization import (
+    format_operational_alert,
+    operational_notification_text,
+    resolve_operational_notification_locale,
+)
+
 try:
     from scripts.runtime_heartbeat_policy import (
         filter_due_targets,
@@ -76,6 +82,14 @@ def _env_bool(name: str, default: bool = False) -> bool:
     return value in {"1", "true", "yes", "y", "on"}
 
 
+def _notification_locale() -> str:
+    return resolve_operational_notification_locale(os.environ.get("NOTIFY_LANG"))
+
+
+def _notice(key: str, /, **values: object) -> str:
+    return operational_notification_text(_notification_locale(), key, **values)
+
+
 def _enabled_value(value: Any, *, default: bool = True) -> bool:
     if value is None:
         return default
@@ -1035,7 +1049,14 @@ def main(now: dt.datetime | None = None) -> int:
             project=project,
         )
     except RuntimeError as exc:
-        message = f"[Execution Report Heartbeat] {name}\nScheduler policy error: {exc}"
+        message = format_operational_alert(
+            locale=_notification_locale(),
+            alert_type="execution_report_heartbeat",
+            name=name,
+            context={"lookback_hours": lookback_hours},
+            issues=[_notice("heartbeat_scheduler_policy_error")],
+            technical_details=[str(exc)],
+        )
         print(message)
         _send_telegram(message)
         return 1 if fail_workflow else 0
@@ -1173,36 +1194,44 @@ def main(now: dt.datetime | None = None) -> int:
         return 0
 
     issues = []
+    technical_details = []
     if list_errors:
-        issues.extend(f"list failed: {item}" for item in list_errors[:3])
+        issues.append(_notice("heartbeat_list_failed"))
+        technical_details.extend(list_errors[:3])
     if not sorted_objects:
-        issues.append(f"no report object updated in the last {lookback_hours:g} hours")
+        issues.append(_notice("heartbeat_no_recent_report", lookback_hours=f"{lookback_hours:g}"))
     elif required_keys:
         missing = [key for key in required_keys if key not in accepted_by_service]
         issues.append(
-            "missing acceptable report for runtime target(s): "
-            + ", ".join(required_labels[key] for key in missing)
+            _notice(
+                "heartbeat_missing_acceptable_report",
+                targets=", ".join(required_labels[key] for key in missing),
+            )
         )
     else:
-        issues.append(f"no acceptable report among {min(len(sorted_objects), max_reports)} recent report object(s)")
+        issues.append(
+            _notice(
+                "heartbeat_no_acceptable_report",
+                count=min(len(sorted_objects), max_reports),
+            )
+        )
 
     run_url = ""
     if os.environ.get("GITHUB_SERVER_URL") and os.environ.get("GITHUB_REPOSITORY") and os.environ.get("GITHUB_RUN_ID"):
         run_url = (
             f"{os.environ['GITHUB_SERVER_URL']}/{os.environ['GITHUB_REPOSITORY']}"
             f"/actions/runs/{os.environ['GITHUB_RUN_ID']}"
         )
-    message_lines = [
-        f"[Execution Report Heartbeat] {name}",
-        f"Lookback: {lookback_hours:g} hours",
-        "Issues:",
-        *[f"- {issue}" for issue in issues],
-    ]
-    if inspected:
-        message_lines.extend(["Recent reports:", *inspected[:max_reports]])
-    if run_url:
-        message_lines.append(f"Workflow: {run_url}")
-    message = "\n".join(message_lines)
+    message = format_operational_alert(
+        locale=_notification_locale(),
+        alert_type="execution_report_heartbeat",
+        name=name,
+        context={"lookback_hours": f"{lookback_hours:g}"},
+        issues=issues,
+        recent_reports=inspected[:max_reports],
+        technical_details=technical_details,
+        workflow_url=run_url,
+    )
     print(message)
     _send_telegram(message[:3900])
     return 1 if fail_workflow else 0

tests/test_cloud_run_runtime_guard.py

@@ -8,6 +8,12 @@
 from scripts import cloud_run_runtime_guard as guard
 
 
+def test_runtime_guard_notification_language_is_configurable(monkeypatch):
+    monkeypatch.setenv("NOTIFY_LANG", "zh-CN")
+
+    assert guard._notice("runtime_guard_scheduler_log_query_failed") == "Cloud Scheduler 日志查询失败"
+
+
 def test_cloud_run_log_filter_includes_region_when_available():
     log_filter = guard._cloud_run_log_filter(
         "interactive-brokers-live-u1599-tqqq-service",