QihuAI Design
a2504bce34
Root cause: ensure_structured_column() executed a synchronous blocking engine.connect().execute(SELECT information_schema.columns ...) inside the async request path (_build_chat_history / _maybe_compress_history), freezing the uvicorn event loop and hanging all endpoints (~200s, recovered only by docker restart). Fix: - chat_context.py: remove per-request ensure_structured_column() calls; run once idempotently at app startup (main.py lifespan). Use CURRENT_DATABASE() for PostgreSQL compatibility. - main.py: add startup-only ensure_structured_column() hook (try/except). - llm_client.py: trust_env=False + bounded httpx.Timeout/wait_for so broken endpoints fail fast instead of hanging the event loop. - chat_service.py: run media generation in a background daemon thread with its own event loop (_launch_media_task_bg) so SSE end does not cancel image/video tasks (#811); remove TEMP faulthandler diagnostic. Safe-align: persist accumulated live container changes (admin/*, knowledge, market, models, config, observability, mcp_client, platform_knowledge, im_bot_platform, chat_media, capability_registry, requirements) and add workflow_templates. Verified: live failover test (model-006 -> conn-refused) returns in ~1.9s with concurrent login 0.83s/200; event-loop responsiveness restored (sub-second).
425 lines
15 KiB
Python
425 lines
15 KiB
Python
"""
|
|
数据分析管理后端 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=false"))
|
|
|
|
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=false"), {"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}}
|