diff --git a/apps/api/src/jobs/coverage_evaluator_job.py b/apps/api/src/jobs/coverage_evaluator_job.py index 1a9a92881..74029a06b 100644 --- a/apps/api/src/jobs/coverage_evaluator_job.py +++ b/apps/api/src/jobs/coverage_evaluator_job.py @@ -166,6 +166,7 @@ async def _fetch_red_summary() -> dict[str, Any] | None: 2026-04-19 P1.3: 加 total_scanned 供呼叫端算 red_ratio 做雙條件觸發. """ from sqlalchemy import text as _sql + from src.db.base import get_db_context try: @@ -265,6 +266,7 @@ async def _llm_analyze_coverage_gaps(red_summary: dict[str, Any]) -> dict[str, A """ try: import json as _j + from src.services.llm_json_parser import parse_llm_json_response from src.services.openclaw import get_openclaw @@ -296,8 +298,12 @@ async def _send_telegram_gaps( """推 coverage 缺口 Telegram 摘要 + 互動按鈕 (P0 修).""" try: import html + from src.core.config import settings - from src.services.ai_advisory_helpers import build_ai_advisory_keyboard, is_snoozed + from src.services.ai_advisory_helpers import ( + build_ai_advisory_keyboard, + is_snoozed, + ) from src.services.telegram_gateway import get_telegram_gateway target_chat_id = settings.SRE_GROUP_CHAT_ID @@ -359,6 +365,7 @@ async def _send_telegram_gaps( async def _get_latest_run_id() -> str | None: from sqlalchemy import text as _sql + from src.db.base import get_db_context try: @@ -389,6 +396,7 @@ async def _evaluate_monitoring(run_id: str) -> int: return 0 from sqlalchemy import text as _sql + from src.db.base import get_db_context try: @@ -462,6 +470,7 @@ async def _evaluate_alerting(run_id: str) -> int: - 無任何 match → red """ from sqlalchemy import text as _sql + from src.db.base import get_db_context try: @@ -509,6 +518,7 @@ async def _evaluate_km_coverage(run_id: str) -> int: 不是 'body' (前次 UndefinedColumnError). 同時加 title 匹配擴大覆蓋. """ from sqlalchemy import text as _sql + from src.db.base import get_db_context try: @@ -553,6 +563,7 @@ async def _evaluate_playbook_coverage(run_id: str) -> int: 沒對應 playbook 但 type 合理 → yellow; 否則保持 unknown """ from sqlalchemy import text as _sql + from src.db.base import get_db_context try: @@ -594,6 +605,7 @@ async def _evaluate_remediation_coverage(run_id: str) -> int: 沒 target 匹配但 asset 是 k8s_workload/container → red (應有修復能力但沒) """ from sqlalchemy import text as _sql + from src.db.base import get_db_context try: @@ -636,6 +648,7 @@ async def _evaluate_rule_matching_coverage(run_id: str) -> int: 沒觸發 → yellow (可能沒問題也可能沒覆蓋,中性) """ from sqlalchemy import text as _sql + from src.db.base import get_db_context try: @@ -683,6 +696,7 @@ async def _evaluate_rule_creation_coverage(run_id: str) -> int: 未來 Hermes 建出 AI rule 後會變 green """ from sqlalchemy import text as _sql + from src.db.base import get_db_context try: @@ -726,6 +740,7 @@ async def _evaluate_rule_creation_coverage(run_id: str) -> int: async def _log_aol(stats: dict[str, int], duration_ms: int, error: str | None) -> None: try: from sqlalchemy import text as _sql + from src.db.base import get_db_context aol_status = "failed" if error else "success" @@ -779,11 +794,13 @@ async def _auto_create_rules_for_uncovered_assets(run_id: str | None) -> int: - rule_name UNIQUE 鍵:CoverageAuto_{type}_{safe_key} - Redis 不可用時跳過冷卻檢查(不中斷主流程) """ - from sqlalchemy import text as _sql - from src.db.base import get_db_context import json as _j import re + from sqlalchemy import text as _sql + + from src.db.base import get_db_context + if not run_id: return 0