diff --git a/apps/api/src/api/v1/stats.py b/apps/api/src/api/v1/stats.py index 8471a3bc9..934fa2167 100644 --- a/apps/api/src/api/v1/stats.py +++ b/apps/api/src/api/v1/stats.py @@ -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, + )