Feature: Multi-tenant + Billing + Analytics - 3 new modules with 40+ API endpoints

backend/python/app/api/admin/analytics_admin.py

@@ -0,0 +1,424 @@
+"""
+数据分析管理后端 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}}