220 lines
7.8 KiB
Python
220 lines
7.8 KiB
Python
"""
|
||
Drift Report Repository - PostgreSQL 實作
|
||
==========================================
|
||
Phase 25 P2 B4: drift_reports 表 DB 持久化
|
||
|
||
職責: DriftReport 的 CRUD 操作（取代 in-memory dict）
|
||
設計: raw SQL via SQLAlchemy text()（表由 phase9 migration 建立）
|
||
|
||
版本: v1.0
|
||
建立: 2026-04-09 (台北時區)
|
||
建立者: Claude Sonnet 4.6 (B4 drift_reports 持久化)
|
||
"""
|
||
|
||
import json
|
||
from datetime import datetime
|
||
|
||
import structlog
|
||
from sqlalchemy import text
|
||
|
||
from src.db.base import get_db_context
|
||
from src.models.drift import (
|
||
DriftIntent,
|
||
DriftInterpretation,
|
||
DriftItem,
|
||
DriftLevel,
|
||
DriftReport,
|
||
DriftStatus,
|
||
)
|
||
|
||
logger = structlog.get_logger(__name__)
|
||
|
||
_MAX_REPORTS = 200 # DB 最多保留筆數（定期清理）
|
||
|
||
|
||
def _report_to_row(report: DriftReport) -> dict:
|
||
"""DriftReport → DB row dict"""
|
||
return {
|
||
"report_id": report.report_id,
|
||
"namespace": report.namespace,
|
||
"triggered_by": report.triggered_by,
|
||
"scanned_at": report.scanned_at,
|
||
"high_count": report.high_count,
|
||
"medium_count": report.medium_count,
|
||
"info_count": report.info_count,
|
||
"items": json.dumps([item.model_dump() for item in report.items]),
|
||
"interpretation": json.dumps(report.interpretation.model_dump()) if report.interpretation else None,
|
||
"status": report.status.value,
|
||
"created_at": report.created_at,
|
||
"resolved_at": report.resolved_at,
|
||
}
|
||
|
||
|
||
def _row_to_report(row) -> DriftReport:
|
||
"""DB row → DriftReport"""
|
||
items = []
|
||
for item_data in (row.items or []):
|
||
item_data["drift_level"] = DriftLevel(item_data.get("drift_level", "medium"))
|
||
items.append(DriftItem(**item_data))
|
||
|
||
interpretation = None
|
||
if row.interpretation:
|
||
d = row.interpretation
|
||
interpretation = DriftInterpretation(
|
||
intent=DriftIntent(d.get("intent", "unknown")),
|
||
explanation=d.get("explanation", ""),
|
||
risk=d.get("risk", "MEDIUM"),
|
||
confidence=d.get("confidence", 0.0),
|
||
)
|
||
|
||
return DriftReport(
|
||
report_id=row.report_id,
|
||
namespace=row.namespace,
|
||
triggered_by=row.triggered_by,
|
||
scanned_at=row.scanned_at,
|
||
high_count=row.high_count,
|
||
medium_count=row.medium_count,
|
||
info_count=row.info_count,
|
||
items=items,
|
||
interpretation=interpretation,
|
||
status=DriftStatus(row.status),
|
||
created_at=row.created_at,
|
||
resolved_at=row.resolved_at,
|
||
)
|
||
|
||
|
||
class DriftReportRepository:
|
||
"""drift_reports 表的 CRUD 操作"""
|
||
|
||
async def save(self, report: DriftReport) -> None:
|
||
"""新增或更新漂移報告（upsert）"""
|
||
row = _report_to_row(report)
|
||
async with get_db_context() as db:
|
||
await db.execute(
|
||
text("""
|
||
INSERT INTO drift_reports
|
||
(report_id, namespace, triggered_by, scanned_at,
|
||
high_count, medium_count, info_count,
|
||
items, interpretation, status, created_at, resolved_at)
|
||
VALUES
|
||
(:report_id, :namespace, :triggered_by, :scanned_at,
|
||
:high_count, :medium_count, :info_count,
|
||
CAST(:items AS jsonb), CAST(:interpretation AS jsonb), :status, :created_at, :resolved_at)
|
||
ON CONFLICT (report_id) DO UPDATE SET
|
||
items = EXCLUDED.items,
|
||
interpretation = EXCLUDED.interpretation,
|
||
status = EXCLUDED.status,
|
||
resolved_at = EXCLUDED.resolved_at,
|
||
high_count = EXCLUDED.high_count,
|
||
medium_count = EXCLUDED.medium_count,
|
||
info_count = EXCLUDED.info_count
|
||
"""),
|
||
row,
|
||
)
|
||
logger.info("drift_report_saved", report_id=report.report_id, namespace=report.namespace)
|
||
|
||
async def get(self, report_id: str) -> DriftReport | None:
|
||
"""依 report_id 查詢"""
|
||
async with get_db_context() as db:
|
||
result = await db.execute(
|
||
text("SELECT * FROM drift_reports WHERE report_id = :report_id"),
|
||
{"report_id": report_id},
|
||
)
|
||
row = result.fetchone()
|
||
return _row_to_report(row) if row else None
|
||
|
||
# 2026-04-19 Claude Opus 4.7 修 drift_view 按鈕 AttributeError
|
||
# 其他 repo 慣例皆 get_by_id，對齊介面供 telegram_gateway 呼叫
|
||
async def get_by_id(self, report_id: str) -> DriftReport | None:
|
||
return await self.get(report_id)
|
||
|
||
async def list_recent(self, limit: int = 50) -> list[DriftReport]:
|
||
"""列出最近 N 筆（倒序）"""
|
||
async with get_db_context() as db:
|
||
result = await db.execute(
|
||
text("SELECT * FROM drift_reports ORDER BY created_at DESC LIMIT :limit"),
|
||
{"limit": limit},
|
||
)
|
||
rows = result.fetchall()
|
||
return [_row_to_report(r) for r in rows]
|
||
|
||
async def update_status(self, report_id: str, status: DriftStatus, resolved_at: datetime | None = None) -> None:
|
||
"""更新處理狀態"""
|
||
async with get_db_context() as db:
|
||
await db.execute(
|
||
text("""
|
||
UPDATE drift_reports
|
||
SET status = :status, resolved_at = :resolved_at
|
||
WHERE report_id = :report_id
|
||
"""),
|
||
{"report_id": report_id, "status": status.value, "resolved_at": resolved_at},
|
||
)
|
||
|
||
async def update_interpretation(self, report_id: str, interpretation: DriftInterpretation) -> None:
|
||
"""更新 Nemotron 意圖分析結果"""
|
||
async with get_db_context() as db:
|
||
await db.execute(
|
||
text("""
|
||
UPDATE drift_reports
|
||
SET interpretation = CAST(:interpretation AS jsonb)
|
||
WHERE report_id = :report_id
|
||
"""),
|
||
{
|
||
"report_id": report_id,
|
||
"interpretation": json.dumps(interpretation.model_dump()),
|
||
},
|
||
)
|
||
|
||
|
||
async def update_narrative(self, report_id: str, narrative: str) -> None:
|
||
"""更新 AI 人話摘要 (Phase 30 ADR-067)"""
|
||
async with get_db_context() as db:
|
||
await db.execute(
|
||
text("UPDATE drift_reports SET narrative_text = :narrative WHERE report_id = :report_id"),
|
||
{"report_id": report_id, "narrative": narrative},
|
||
)
|
||
|
||
async def get_repeat_state(
|
||
self,
|
||
report: DriftReport,
|
||
*,
|
||
include_values: bool = True,
|
||
) -> dict:
|
||
"""Return stable fingerprint repeat state for a drift report."""
|
||
from src.services.drift_repeat_state import build_drift_repeat_state
|
||
|
||
async with get_db_context() as db:
|
||
result = await db.execute(
|
||
text("""
|
||
SELECT
|
||
report_id,
|
||
namespace,
|
||
status,
|
||
scanned_at,
|
||
created_at,
|
||
items
|
||
FROM drift_reports
|
||
WHERE namespace = :namespace
|
||
AND created_at > now() - interval '24 hours'
|
||
ORDER BY scanned_at DESC
|
||
LIMIT 200
|
||
"""),
|
||
{"namespace": report.namespace},
|
||
)
|
||
rows = [dict(row) for row in result.mappings().all()]
|
||
return build_drift_repeat_state(
|
||
report,
|
||
rows,
|
||
include_values=include_values,
|
||
)
|
||
|
||
|
||
_drift_repo: DriftReportRepository | None = None
|
||
|
||
|
||
def get_drift_repository() -> DriftReportRepository:
|
||
global _drift_repo
|
||
if _drift_repo is None:
|
||
_drift_repo = DriftReportRepository()
|
||
return _drift_repo
|