""" 数据分析管理后端 API - 事件采集、聚合统计、仪表盘配置、趋势分析 - 路由前缀: /app/api/admin/analytics """ import uuid import json from datetime import datetime, timedelta, date from fastapi import APIRouter, Query, HTTPException, Depends from ...database import SessionLocal, get_db from ..auth import get_current_user from sqlalchemy import text from sqlalchemy.orm import Session router = APIRouter() CST_TZ = None def _format_cst_time(dt_val): global CST_TZ if CST_TZ is None: from datetime import timezone CST_TZ = timezone(timedelta(hours=8)) if dt_val is None: return None if isinstance(dt_val, str): return dt_val if dt_val.tzinfo is None: dt_val = dt_val.replace(tzinfo=CST_TZ) return dt_val.strftime("%Y-%m-%d %H:%M:%S") # ============================================================ # 统计概览 # ============================================================ @router.get("/overview") async def analytics_overview( days: int = Query(30, description="统计天数"), db: Session = Depends(get_db), user=Depends(get_current_user) ): """数据分析总览""" since = datetime.now() - timedelta(days=days) # User stats total_users = db.execute(text("SELECT COUNT(*) FROM users")).scalar() active_users = db.execute(text(""" SELECT COUNT(DISTINCT user_id) FROM token_usage_logs WHERE created_at >= :since """), {"since": since}).scalar() new_users = db.execute(text(""" SELECT COUNT(*) FROM users WHERE created_at >= :since """), {"since": since}).scalar() # API stats api_calls = db.execute(text(""" SELECT COALESCE(SUM(call_count), 0) FROM api_call_stats WHERE updated_at >= :since """), {"since": since}).scalar() avg_response = db.execute(text(""" SELECT COALESCE(AVG(total_response_time/NULLIF(call_count,0)), 0) FROM api_call_stats WHERE updated_at >= :since """), {"since": since}).scalar() error_rate = db.execute(text(""" SELECT COALESCE(SUM(error_count)/NULLIF(SUM(call_count),0)*100, 0) FROM api_call_stats WHERE updated_at >= :since """), {"since": since}).scalar() # Token stats tokens_used = db.execute(text(""" SELECT COALESCE(SUM(total_tokens), 0) FROM token_usage_logs WHERE created_at >= :since """), {"since": since}).scalar() total_cost = db.execute(text(""" SELECT COALESCE(SUM(cost), 0) FROM token_usage_logs WHERE created_at >= :since """), {"since": since}).scalar() # Chat stats total_chats = db.execute(text("SELECT COUNT(*) FROM conversations")).scalar() total_messages = db.execute(text("SELECT COUNT(*) FROM messages")).scalar() # Revenue (from billing_transactions) revenue = db.execute(text(""" SELECT COALESCE(SUM(amount), 0) FROM billing_transactions WHERE transaction_type='recharge' AND status='completed' AND created_at >= :since """), {"since": since}).scalar() return {"code": 200, "data": { "total_users": total_users, "active_users": active_users, "new_users": new_users, "api_calls": api_calls, "avg_response_time": round(float(avg_response), 2), "error_rate": round(float(error_rate), 2), "tokens_used": tokens_used, "total_cost": float(total_cost), "total_chats": total_chats, "total_messages": total_messages, "revenue": float(revenue), "period_days": days, }} # ============================================================ # 趋势分析 # ============================================================ @router.get("/trends/api-calls") async def api_call_trends( days: int = Query(7, description="天数"), db: Session = Depends(get_db), user=Depends(get_current_user) ): """API调用趋势(按天)""" since = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d') rows = db.execute(text(""" SELECT DATE(created_at) as d, COUNT(*) as calls, COALESCE(SUM(CASE WHEN status='error' THEN 1 ELSE 0 END), 0) as errors FROM analytics_events WHERE event_type='api_call' AND created_at >= :since GROUP BY DATE(created_at) ORDER BY d """), {"since": since}).fetchall() if not rows: # Fallback to api_call_stats rows = db.execute(text(""" SELECT DATE(updated_at) as d, SUM(call_count) as calls, SUM(error_count) as errors FROM api_call_stats WHERE updated_at >= :since GROUP BY DATE(updated_at) ORDER BY d """), {"since": since}).fetchall() data = [{"date": str(r[0]), "calls": int(r[1] or 0), "errors": int(r[2] or 0)} for r in rows] return {"code": 200, "data": data} @router.get("/trends/tokens") async def token_usage_trends( days: int = Query(7, description="天数"), db: Session = Depends(get_db), user=Depends(get_current_user) ): """Token使用趋势(按天+按模型)""" since = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d') rows = db.execute(text(""" SELECT DATE(created_at) as d, model_name, SUM(total_tokens) as tokens, SUM(cost) as cost FROM token_usage_logs WHERE created_at >= :since GROUP BY DATE(created_at), model_name ORDER BY d, tokens DESC """), {"since": since}).fetchall() # Group by date by_date = {} for r in rows: d = str(r[0]) if d not in by_date: by_date[d] = {"date": d, "total_tokens": 0, "models": {}} by_date[d]["total_tokens"] += int(r[2] or 0) model = r[1] or "unknown" by_date[d]["models"][model] = int(r[2] or 0) # Model distribution model_dist = {} for r in rows: model = r[1] or "unknown" model_dist[model] = model_dist.get(model, 0) + int(r[2] or 0) return {"code": 200, "data": { "daily": list(by_date.values()), "model_distribution": model_dist, }} @router.get("/trends/users") async def user_growth_trends( days: int = Query(30, description="天数"), db: Session = Depends(get_db), user=Depends(get_current_user) ): """用户增长趋势""" since = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d') rows = db.execute(text(""" SELECT DATE(created_at) as d, COUNT(*) as new_users FROM users WHERE created_at >= :since GROUP BY DATE(created_at) ORDER BY d """), {"since": since}).fetchall() data = [{"date": str(r[0]), "new_users": int(r[1])} for r in rows] # Cumulative cumulative = 0 total_before = db.execute(text("SELECT COUNT(*) FROM users WHERE created_at < :since"), {"since": since}).scalar() cumulative = total_before for item in data: cumulative += item["new_users"] item["cumulative"] = cumulative return {"code": 200, "data": data} @router.get("/trends/revenue") async def revenue_trends( days: int = Query(30, description="天数"), db: Session = Depends(get_db), user=Depends(get_current_user) ): """收入趋势""" since = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d') rows = db.execute(text(""" SELECT DATE(created_at) as d, COALESCE(SUM(CASE WHEN transaction_type='recharge' THEN amount ELSE 0 END), 0) as revenue, COALESCE(SUM(CASE WHEN transaction_type='consumption' THEN amount ELSE 0 END), 0) as consumption FROM billing_transactions WHERE status='completed' AND created_at >= :since GROUP BY DATE(created_at) ORDER BY d """), {"since": since}).fetchall() data = [{"date": str(r[0]), "revenue": float(r[1] or 0), "consumption": float(r[2] or 0)} for r in rows] return {"code": 200, "data": data} # ============================================================ # 模型使用分析 # ============================================================ @router.get("/models/usage") async def model_usage_analysis( days: int = Query(30, description="天数"), db: Session = Depends(get_db), user=Depends(get_current_user) ): """模型使用分析""" since = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d') rows = db.execute(text(""" SELECT model_name, COUNT(*) as call_count, SUM(total_tokens) as total_tokens, SUM(prompt_tokens) as prompt_tokens, SUM(completion_tokens) as completion_tokens, SUM(cost) as total_cost, AVG(total_tokens) as avg_tokens FROM token_usage_logs WHERE created_at >= :since AND model_name IS NOT NULL GROUP BY model_name ORDER BY total_tokens DESC """), {"since": since}).fetchall() data = [{ "model_name": r[0], "call_count": int(r[1]), "total_tokens": int(r[2] or 0), "prompt_tokens": int(r[3] or 0), "completion_tokens": int(r[4] or 0), "total_cost": float(r[5] or 0), "avg_tokens": round(float(r[6] or 0), 0), } for r in rows] return {"code": 200, "data": data} # ============================================================ # 仪表盘配置 # ============================================================ @router.get("/dashboards") async def list_dashboards(db: Session = Depends(get_db), user=Depends(get_current_user)): """获取仪表盘配置列表""" rows = db.execute(text(""" SELECT ad.*, u.username as created_by_name FROM analytics_dashboards ad LEFT JOIN users u ON ad.created_by = u.id ORDER BY ad.is_default DESC, ad.created_at DESC """)).fetchall() cols = ['id','tenant_id','name','description','config','is_default', 'created_by','created_at','updated_at','created_by_name'] items = [] for r in rows: d = _format_cst_time_dict(r, cols) items.append(d) return {"code": 200, "data": items} def _format_cst_time_dict(row, columns): result = {} for col, val in zip(columns, row): if isinstance(val, datetime): result[col] = _format_cst_time(val) elif isinstance(val, str) and val and val.startswith('[') or (isinstance(val, str) and val and val.startswith('{')): try: result[col] = json.loads(val) except: result[col] = val else: result[col] = val return result @router.post("/dashboards/create") async def create_dashboard( name: str = Query(...), description: str = Query(""), config: str = Query(..., description="仪表盘配置JSON"), is_default: int = Query(0), db: Session = Depends(get_db), user=Depends(get_current_user) ): """创建仪表盘""" did = str(uuid.uuid4()) try: config_json = json.loads(config) except: config_json = json.loads('[{"type":"stat","title":"暂无数据","metric":"none","chart_type":"number"}]') if is_default: db.execute(text("UPDATE analytics_dashboards SET is_default=0")) db.execute(text(""" INSERT INTO analytics_dashboards (id, name, description, config, is_default, created_by) VALUES (:id, :name, :desc, :config, :default, :uid) """), { "id": did, "name": name, "desc": description, "config": json.dumps(config_json), "default": is_default, "uid": user.id }) db.commit() return {"code": 200, "data": {"id": did, "message": "仪表盘创建成功"}} @router.delete("/dashboards/{dashboard_id}") async def delete_dashboard(dashboard_id: str, db: Session = Depends(get_db), user=Depends(get_current_user)): """删除仪表盘""" db.execute(text("DELETE FROM analytics_dashboards WHERE id=:id AND is_default=0"), {"id": dashboard_id}) db.commit() return {"code": 200, "data": {"message": "仪表盘已删除"}} # ============================================================ # 事件记录 # ============================================================ @router.post("/events/record") async def record_event( event_type: str = Query(..., description="事件类型"), event_category: str = Query("general"), event_data: str = Query("", description="事件数据JSON"), duration_ms: int = Query(None), status: str = Query("success"), db: Session = Depends(get_db), user=Depends(get_current_user) ): """记录分析事件""" data = None if event_data: try: data = json.loads(event_data) except: data = {"raw": event_data} db.execute(text(""" INSERT INTO analytics_events (tenant_id, user_id, event_type, event_category, event_data, duration_ms, status) VALUES (:tid, :uid, :et, :ec, :data, :dur, :status) """), { "tid": None, "uid": user.id, "et": event_type, "ec": event_category, "data": json.dumps(data) if data else None, "dur": duration_ms, "status": status }) db.commit() return {"code": 200, "data": {"message": "事件已记录"}} @router.get("/events") async def list_events( page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100), event_type: str = Query(""), event_category: str = Query(""), status: str = Query(""), days: int = Query(7, description="查询天数"), db: Session = Depends(get_db), user=Depends(get_current_user) ): """获取事件列表""" since = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d') where = "ae.created_at >= :since" params = {"since": since} if event_type: where += " AND ae.event_type=:et" params["et"] = event_type if event_category: where += " AND ae.event_category=:ec" params["ec"] = event_category if status: where += " AND ae.status=:st" params["st"] = status total = db.execute(text(f"SELECT COUNT(*) FROM analytics_events ae WHERE {where}"), params).scalar() offset = (page - 1) * page_size rows = db.execute(text(f""" SELECT ae.*, u.username FROM analytics_events ae LEFT JOIN users u ON ae.user_id = u.id WHERE {where} ORDER BY ae.created_at DESC LIMIT :limit OFFSET :offset """), {**params, "limit": page_size, "offset": offset}).fetchall() cols = ['id','tenant_id','user_id','event_type','event_category','event_data', 'ip_address','user_agent','duration_ms','status','created_at','username'] items = [] for r in rows: d = {} for col, val in zip(cols, r): if isinstance(val, datetime): d[col] = _format_cst_time(val) elif isinstance(val, str) and val and (val.startswith('{') or val.startswith('[')): try: d[col] = json.loads(val) except: d[col] = val else: d[col] = val items.append(d) return {"code": 200, "data": {"items": items, "total": total, "page": page, "page_size": page_size}}