style(security): normalize coverage evaluator imports
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 1s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 2m52s
CD Pipeline / build-and-deploy (push) Failing after 3m56s
CD Pipeline / post-deploy-checks (push) Has been skipped

This commit is contained in:
ogt
2026-07-14 16:28:56 +08:00
parent 0e184339af
commit 5ec6fd3c30

View File

@@ -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