feat(api): 新增 trends 和 feedback 統計端點

- /stats/incidents/trends: 每日/週/月趨勢分析
- /stats/feedback/summary: 人類回饋摘要 (正/中/負比例 + 常見主題萃取)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
OG T
2026-03-24 09:52:11 +08:00
parent 765ee39a90
commit 3a95b35384

View File

@@ -383,3 +383,131 @@ async def get_affected_services(
)
for svc, stats in sorted_services
]
@router.get(
"/incidents/trends",
response_model=IncidentTrends,
summary="事件趨勢分析",
)
async def get_incident_trends(
days: int = Query(30, ge=7, le=365, description="統計區間 (天)"),
period: str = Query("daily", description="週期: daily/weekly/monthly"),
db: AsyncSession = Depends(get_db), # noqa: B008
) -> IncidentTrends:
"""
取得事件趨勢數據
支援週期:
- daily: 每日事件數
- weekly: 每週事件數
- monthly: 每月事件數
"""
since = datetime.utcnow() - timedelta(days=days)
# 取得所有事件的建立時間
result = await db.execute(
select(IncidentRecord.created_at).where(
IncidentRecord.created_at >= since
)
)
timestamps = [row[0] for row in result.all()]
# 依週期聚合
counts: dict[str, int] = {}
for ts in timestamps:
if period == "daily":
key = ts.strftime("%Y-%m-%d")
elif period == "weekly":
# ISO 週數
key = ts.strftime("%Y-W%W")
else: # monthly
key = ts.strftime("%Y-%m")
counts[key] = counts.get(key, 0) + 1
# 排序並轉換為 TrendPoint
sorted_data = sorted(counts.items(), key=lambda x: x[0])
trend_data = [TrendPoint(date=k, count=v) for k, v in sorted_data]
logger.info(
"stats_incident_trends",
period=period,
days=days,
data_points=len(trend_data),
)
return IncidentTrends(period=period, data=trend_data)
@router.get(
"/feedback/summary",
response_model=FeedbackSummary,
summary="人類回饋摘要",
)
async def get_feedback_summary(
days: int = Query(30, ge=1, le=365, description="統計區間 (天)"),
db: AsyncSession = Depends(get_db), # noqa: B008
) -> FeedbackSummary:
"""
取得人類回饋統計
從 Incident outcome 中萃取:
- 正面/中性/負面回饋比例
- 常見主題 (從 learning_notes 萃取)
"""
since = datetime.utcnow() - timedelta(days=days)
# 取得有 outcome 的事件
result = await db.execute(
select(IncidentRecord.outcome).where(
IncidentRecord.created_at >= since,
IncidentRecord.outcome.isnot(None),
)
)
outcomes = [row[0] for row in result.all() if row[0]]
# 統計回饋分數
positive = 0
neutral = 0
negative = 0
themes: dict[str, int] = {}
for o in outcomes:
score = o.get("effectiveness_score") or o.get("feedback_score")
if score:
if score >= 4:
positive += 1
elif score == 3:
neutral += 1
else:
negative += 1
# 萃取主題 (從 learning_notes)
notes = o.get("learning_notes") or o.get("notes") or ""
if notes:
# 簡單關鍵字萃取 (未來可用 NLP)
for keyword in ["timeout", "memory", "network", "disk", "cpu", "connection"]:
if keyword.lower() in notes.lower():
themes[keyword] = themes.get(keyword, 0) + 1
# 取前 5 個常見主題
sorted_themes = sorted(themes.items(), key=lambda x: x[1], reverse=True)[:5]
common_themes = [t[0] for t in sorted_themes]
total = positive + neutral + negative
logger.info(
"stats_feedback_summary",
total=total,
positive=positive,
negative=negative,
days=days,
)
return FeedbackSummary(
total_feedback=total,
positive_count=positive,
neutral_count=neutral,
negative_count=negative,
common_themes=common_themes,
)