525 lines
21 KiB
Python
525 lines
21 KiB
Python
"""统计服务：日志查询、趋势数据、知识库健康度、Token 成本。"""
|
||
import csv
|
||
import io
|
||
import json
|
||
from insurance.db.compat import db
|
||
from sqlalchemy import text
|
||
|
||
|
||
class StatsService:
|
||
"""统计报表业务逻辑。"""
|
||
|
||
def _parse_model_name(self, provider: str = "", model_id: str = "", model_json=None) -> str:
|
||
"""Extract provider/model from Dify app model config fields."""
|
||
if provider and model_id:
|
||
return f"{provider}/{model_id}"
|
||
|
||
if not model_json:
|
||
return ""
|
||
|
||
try:
|
||
model_data = json.loads(model_json) if isinstance(model_json, str) and model_json.startswith("{") else model_json
|
||
except (json.JSONDecodeError, TypeError):
|
||
return ""
|
||
|
||
if not isinstance(model_data, dict):
|
||
return ""
|
||
|
||
parsed_provider = model_data.get("provider", "")
|
||
parsed_model = model_data.get("name", "") or model_data.get("model", "") or model_data.get("model_id", "")
|
||
if parsed_provider and parsed_model and not isinstance(parsed_model, dict):
|
||
return f"{parsed_provider}/{parsed_model}"
|
||
|
||
nested = model_data.get("model", {})
|
||
if isinstance(nested, dict):
|
||
nested_provider = nested.get("provider", "") or parsed_provider
|
||
nested_model = nested.get("name", "") or nested.get("model", "") or nested.get("model_id", "")
|
||
if nested_provider and nested_model:
|
||
return f"{nested_provider}/{nested_model}"
|
||
|
||
return ""
|
||
|
||
def _get_default_model_name(self) -> str:
|
||
"""Get the latest configured app model name for records saved before model_id was populated."""
|
||
try:
|
||
result = db.session.execute(text("""
|
||
SELECT provider, model_id, model
|
||
FROM app_model_configs
|
||
ORDER BY updated_at DESC
|
||
""")).fetchall()
|
||
for provider, model_id, model_json in result:
|
||
model_name = self._parse_model_name(provider, model_id, model_json)
|
||
if model_name:
|
||
return model_name
|
||
except Exception:
|
||
pass
|
||
return "unknown"
|
||
|
||
def get_chat_logs(self, params: dict) -> dict:
|
||
"""查询对话记录（从本地 insurance_chat_records 表获取）。"""
|
||
# 构建 WHERE 条件
|
||
conditions = ["u.role = 'user'"]
|
||
bind_params = {}
|
||
|
||
if params.get("user_id"):
|
||
conditions.append("u.user_id = :user_id")
|
||
bind_params["user_id"] = params["user_id"]
|
||
if params.get("start_date"):
|
||
conditions.append("u.created_at >= :start_date")
|
||
bind_params["start_date"] = params["start_date"]
|
||
if params.get("end_date"):
|
||
conditions.append("u.created_at <= :end_date")
|
||
bind_params["end_date"] = params["end_date"] + " 23:59:59"
|
||
if params.get("keyword"):
|
||
conditions.append("""
|
||
(u.content LIKE :keyword
|
||
OR EXISTS (
|
||
SELECT 1 FROM insurance_chat_records a2
|
||
WHERE a2.session_id = u.session_id
|
||
AND a2.role = 'assistant'
|
||
AND a2.content LIKE :keyword
|
||
))
|
||
""")
|
||
bind_params["keyword"] = f"%{params['keyword']}%"
|
||
if params.get("rating"):
|
||
conditions.append("""
|
||
EXISTS (
|
||
SELECT 1 FROM insurance_chat_records a3
|
||
WHERE a3.session_id = u.session_id
|
||
AND a3.role = 'assistant'
|
||
AND a3.rating = :rating
|
||
)
|
||
""")
|
||
bind_params["rating"] = params["rating"]
|
||
|
||
where_clause = " AND ".join(conditions)
|
||
|
||
# 查询总数（只计算用户消息数，每条用户消息代表一个问答对）
|
||
count_sql = text(f"""
|
||
SELECT COUNT(*) as total
|
||
FROM insurance_chat_records u
|
||
WHERE {where_clause}
|
||
""")
|
||
total_result = db.session.execute(count_sql, bind_params)
|
||
total = total_result.scalar() or 0
|
||
|
||
# 查询数据（用户问题 + 助手回答配对）
|
||
# 使用子查询确保每条用户消息只匹配一条助手消息（兼容 PostgreSQL 和 SQLite）
|
||
query_sql = text(f"""
|
||
SELECT
|
||
u.id,
|
||
u.user_id as user_id,
|
||
COALESCE(NULLIF(m.real_name, ''), NULLIF(m.username, ''), u.user_id) as "user",
|
||
COALESCE(NULLIF(m.real_name, ''), NULLIF(m.username, ''), u.user_id) as user_name,
|
||
m.username as username,
|
||
u.content as query,
|
||
COALESCE(
|
||
(SELECT a.content FROM insurance_chat_records a
|
||
WHERE a.session_id = u.session_id
|
||
AND a.role = 'assistant'
|
||
AND a.created_at > u.created_at
|
||
ORDER BY a.created_at ASC LIMIT 1),
|
||
''
|
||
) as answer,
|
||
COALESCE(
|
||
(SELECT a.rating FROM insurance_chat_records a
|
||
WHERE a.session_id = u.session_id
|
||
AND a.role = 'assistant'
|
||
AND a.created_at > u.created_at
|
||
ORDER BY a.created_at ASC LIMIT 1),
|
||
''
|
||
) as feedback,
|
||
COALESCE(
|
||
(SELECT a.correction FROM insurance_chat_records a
|
||
WHERE a.session_id = u.session_id
|
||
AND a.role = 'assistant'
|
||
AND a.created_at > u.created_at
|
||
ORDER BY a.created_at ASC LIMIT 1),
|
||
''
|
||
) as correction,
|
||
u.created_at
|
||
FROM insurance_chat_records u
|
||
LEFT JOIN wecom_user_mapping m ON u.user_id = CAST(m.id AS TEXT)
|
||
WHERE {where_clause}
|
||
ORDER BY u.created_at DESC
|
||
LIMIT :limit OFFSET :offset
|
||
""")
|
||
bind_params["limit"] = params["page_size"]
|
||
bind_params["offset"] = (params["page"] - 1) * params["page_size"]
|
||
|
||
result = db.session.execute(query_sql, bind_params)
|
||
items = [row._asdict() for row in result]
|
||
|
||
# 格式化时间字段
|
||
for item in items:
|
||
if item.get("created_at"):
|
||
item["created_at"] = str(item["created_at"])
|
||
|
||
return {"code": 0, "data": {"items": items, "total": total}}
|
||
|
||
def export_chat_logs(self, params: dict) -> str:
|
||
"""导出问答记录为 CSV（带 UTF-8 BOM，解决 Excel 乱码问题）。"""
|
||
result = self.get_chat_logs({**params, "page": 1, "page_size": 10000})
|
||
output = io.StringIO()
|
||
# 添加 UTF-8 BOM，让 Excel 正确识别中文编码
|
||
output.write('')
|
||
writer = csv.writer(output)
|
||
writer.writerow(["ID", "用户", "问题", "回答", "评分", "纠错内容", "时间"])
|
||
for item in result.get("data", {}).get("items", []):
|
||
writer.writerow([
|
||
item.get("id", ""),
|
||
item.get("user", ""),
|
||
item.get("query", ""),
|
||
item.get("answer", ""),
|
||
item.get("feedback", ""),
|
||
item.get("correction", ""),
|
||
item.get("created_at", ""),
|
||
])
|
||
return output.getvalue()
|
||
|
||
def get_system_logs(self, params: dict) -> dict:
|
||
"""查询系统操作日志。"""
|
||
# 构建WHERE条件
|
||
conditions = ["1=1"]
|
||
bind_params = {}
|
||
|
||
if params.get("action"):
|
||
conditions.append("action = :action")
|
||
bind_params["action"] = params["action"]
|
||
if params.get("user_id"):
|
||
conditions.append("user_id = :user_id")
|
||
bind_params["user_id"] = params["user_id"]
|
||
|
||
where_clause = " AND ".join(conditions)
|
||
|
||
# 查询总数
|
||
count_sql = text(f"""
|
||
SELECT COUNT(*) as total
|
||
FROM system_operation_logs
|
||
WHERE {where_clause}
|
||
""")
|
||
total_result = db.session.execute(count_sql, bind_params)
|
||
total = total_result.scalar() or 0
|
||
|
||
# 查询数据
|
||
query_sql = text(f"""
|
||
SELECT id, user_id, action, target_type, target_id, detail, ip, user_agent, created_at
|
||
FROM system_operation_logs
|
||
WHERE {where_clause}
|
||
ORDER BY created_at DESC
|
||
LIMIT :limit OFFSET :offset
|
||
""")
|
||
bind_params["limit"] = params["page_size"]
|
||
bind_params["offset"] = (params["page"] - 1) * params["page_size"]
|
||
|
||
result = db.session.execute(query_sql, bind_params)
|
||
items = [row._asdict() for row in result]
|
||
|
||
return {"code": 0, "data": {"total": total, "items": items}}
|
||
|
||
def get_overview(self) -> dict:
|
||
"""使用概览：今日问答数、活跃用户、问答记录总数。"""
|
||
from datetime import datetime
|
||
|
||
result = {
|
||
"today_chats": 0,
|
||
"active_users": 0,
|
||
"total_documents": 0,
|
||
}
|
||
|
||
today_start = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
|
||
|
||
try:
|
||
# 今日问答数：统计 insurance_chat_records 表中今日的用户消息数
|
||
count_sql = text("""
|
||
SELECT COUNT(*) as total
|
||
FROM insurance_chat_records
|
||
WHERE role = 'user'
|
||
AND created_at >= :today_start
|
||
""")
|
||
total_result = db.session.execute(count_sql, {"today_start": today_start})
|
||
result["today_chats"] = total_result.scalar() or 0
|
||
except Exception:
|
||
pass
|
||
|
||
try:
|
||
# 活跃用户数：统计今日在 insurance_chat_records 表中有消息的去重用户数
|
||
active_sql = text("""
|
||
SELECT COUNT(DISTINCT user_id) as total
|
||
FROM insurance_chat_records
|
||
WHERE role = 'user'
|
||
AND created_at >= :today_start
|
||
""")
|
||
active_result = db.session.execute(active_sql, {"today_start": today_start})
|
||
result["active_users"] = active_result.scalar() or 0
|
||
except Exception:
|
||
pass
|
||
|
||
try:
|
||
# 文档总数：统计所有历史问答对数量（每条用户消息代表一个问答对）
|
||
doc_sql = text("""
|
||
SELECT COUNT(*) as total
|
||
FROM insurance_chat_records
|
||
WHERE role = 'user'
|
||
""")
|
||
doc_result = db.session.execute(doc_sql)
|
||
result["total_documents"] = doc_result.scalar() or 0
|
||
except Exception:
|
||
pass
|
||
|
||
return {"code": 0, "data": result}
|
||
|
||
def get_trend(self, metric: str, start_date: str, end_date: str, granularity: str) -> dict:
|
||
"""趋势数据（折线图数据源）。"""
|
||
from datetime import datetime, timedelta
|
||
|
||
# 如果未提供日期，默认最近 30 天
|
||
now = datetime.now()
|
||
if not start_date or not end_date:
|
||
end = now
|
||
start = now - timedelta(days=30)
|
||
else:
|
||
try:
|
||
start = datetime.strptime(start_date, "%Y-%m-%d")
|
||
end = datetime.strptime(end_date, "%Y-%m-%d")
|
||
except ValueError:
|
||
return {"code": 1001, "message": "日期格式错误，请使用 YYYY-MM-DD 格式", "data": None}
|
||
|
||
# 根据指标选择查询（使用 insurance_chat_records 表）
|
||
if metric == "chat_count":
|
||
# 每日问答数（只统计用户消息，每条代表一个问答对）
|
||
sql = text("""
|
||
SELECT DATE(created_at) as date, COUNT(*) as value
|
||
FROM insurance_chat_records
|
||
WHERE role = 'user'
|
||
AND created_at >= :start_date AND created_at <= :end_date
|
||
GROUP BY DATE(created_at)
|
||
ORDER BY date DESC
|
||
""")
|
||
elif metric == "active_users":
|
||
# 每日活跃用户数
|
||
sql = text("""
|
||
SELECT DATE(created_at) as date, COUNT(DISTINCT user_id) as value
|
||
FROM insurance_chat_records
|
||
WHERE role = 'user'
|
||
AND created_at >= :start_date AND created_at <= :end_date
|
||
GROUP BY DATE(created_at)
|
||
ORDER BY date DESC
|
||
""")
|
||
else:
|
||
return {"code": 1001, "message": f"不支持的指标: {metric}", "data": None}
|
||
|
||
try:
|
||
result = db.session.execute(sql, {
|
||
"start_date": start.strftime("%Y-%m-%d"),
|
||
"end_date": end.strftime("%Y-%m-%d") + " 23:59:59",
|
||
}).fetchall()
|
||
|
||
# 转换为字典，方便查找
|
||
data_map = {row[0].strftime("%Y-%m-%d") if hasattr(row[0], 'strftime') else str(row[0]): row[1] for row in result}
|
||
|
||
# 补全日期（没有数据的日期填 0）
|
||
data_points = []
|
||
current = start
|
||
while current <= end:
|
||
date_str = current.strftime("%Y-%m-%d")
|
||
data_points.append({
|
||
"date": date_str,
|
||
"value": data_map.get(date_str, 0),
|
||
})
|
||
current += timedelta(days=1)
|
||
data_points.reverse()
|
||
|
||
except Exception as e:
|
||
return {"code": 5001, "message": f"获取趋势数据失败: {str(e)}", "data": None}
|
||
|
||
return {
|
||
"code": 0,
|
||
"data": {
|
||
"metric": metric,
|
||
"granularity": granularity,
|
||
"data": data_points,
|
||
},
|
||
}
|
||
|
||
def get_kb_health(self, days: int) -> dict:
|
||
"""知识库健康度。"""
|
||
import requests
|
||
from flask import current_app
|
||
|
||
base_url = current_app.config.get("BAODAN_API_URL", "http://localhost:5001")
|
||
api_key = current_app.config.get("BAODAN_CHAT_API_KEY", "")
|
||
|
||
result = {
|
||
"hot_questions": [],
|
||
"missed_questions": [],
|
||
"coverage": {},
|
||
}
|
||
|
||
try:
|
||
# 获取热门问题（从对话日志中统计）
|
||
resp = requests.get(
|
||
f"{base_url}/v1/messages",
|
||
params={"limit": 100, "user": ""},
|
||
headers={"Authorization": f"Bearer {api_key}"},
|
||
timeout=10,
|
||
)
|
||
if resp.status_code == 200:
|
||
messages = resp.json().get("data", [])
|
||
# 统计问题频次
|
||
question_counts = {}
|
||
for msg in messages:
|
||
query = msg.get("query", "")
|
||
if query:
|
||
question_counts[query] = question_counts.get(query, 0) + 1
|
||
|
||
# 取Top20
|
||
sorted_questions = sorted(question_counts.items(), key=lambda x: x[1], reverse=True)[:20]
|
||
result["hot_questions"] = [
|
||
{"question": q, "count": c}
|
||
for q, c in sorted_questions
|
||
]
|
||
except Exception:
|
||
pass
|
||
|
||
return {"code": 0, "data": result}
|
||
|
||
def get_token_usage(self, params: dict) -> dict:
|
||
"""Token 使用统计：总量 + 按模型分组 + 按用户分组 + 日趋势。
|
||
|
||
查询 insurance_chat_records 表。
|
||
"""
|
||
import logging
|
||
logging.info(f"[TOKEN_USAGE] 收到请求: params={params}")
|
||
|
||
default_model = self._get_default_model_name()
|
||
conditions = ["1=1"]
|
||
bind_params = {"default_model": default_model}
|
||
|
||
# 时间范围筛选
|
||
if params.get("start_date"):
|
||
conditions.append("created_at >= :start_date")
|
||
bind_params["start_date"] = params["start_date"]
|
||
if params.get("end_date"):
|
||
conditions.append("created_at <= :end_date")
|
||
bind_params["end_date"] = params["end_date"] + " 23:59:59"
|
||
|
||
# 模型筛选
|
||
if params.get("models"):
|
||
model_list = params["models"]
|
||
placeholders = ", ".join([f":model_{i}" for i in range(len(model_list))])
|
||
conditions.append(f"COALESCE(NULLIF(model_id, ''), :default_model) IN ({placeholders})")
|
||
for i, model in enumerate(model_list):
|
||
bind_params[f"model_{i}"] = model
|
||
|
||
where_clause = " AND ".join(conditions)
|
||
logging.info(f"[TOKEN_USAGE] SQL WHERE: {where_clause}, params={bind_params}")
|
||
|
||
try:
|
||
# 1. 总量统计（只统计 assistant 消息的 answer_tokens，user 消息的 message_tokens）
|
||
total_sql = text(f"""
|
||
SELECT
|
||
COALESCE(SUM(CASE WHEN role = 'assistant' THEN answer_tokens ELSE message_tokens END), 0) as total_tokens,
|
||
COALESCE(SUM(message_tokens), 0) as total_input_tokens,
|
||
COALESCE(SUM(answer_tokens), 0) as total_output_tokens
|
||
FROM insurance_chat_records
|
||
WHERE {where_clause}
|
||
""")
|
||
total_result = db.session.execute(total_sql, bind_params).fetchone()
|
||
logging.info(f"[TOKEN_USAGE] 总量统计: total_tokens={total_result.total_tokens if total_result else 0}, input={total_result.total_input_tokens if total_result else 0}, output={total_result.total_output_tokens if total_result else 0}")
|
||
|
||
# 2. 按模型分组
|
||
model_sql = text(f"""
|
||
SELECT
|
||
COALESCE(NULLIF(model_id, ''), :default_model) as model,
|
||
COALESCE(SUM(CASE WHEN role = 'assistant' THEN answer_tokens ELSE message_tokens END), 0) as total_tokens,
|
||
COUNT(*) as message_count
|
||
FROM insurance_chat_records
|
||
WHERE {where_clause}
|
||
GROUP BY COALESCE(NULLIF(model_id, ''), :default_model)
|
||
ORDER BY total_tokens DESC
|
||
""")
|
||
model_result = db.session.execute(model_sql, bind_params).fetchall()
|
||
by_model = [row._asdict() for row in model_result]
|
||
logging.info(f"[TOKEN_USAGE] 按模型统计: {len(by_model)} 个模型, data={by_model[:3]}")
|
||
|
||
# 3. 按用户分组
|
||
user_sql = text(f"""
|
||
SELECT
|
||
user_id,
|
||
COALESCE(SUM(CASE WHEN role = 'assistant' THEN answer_tokens ELSE message_tokens END), 0) as total_tokens,
|
||
COUNT(*) as message_count
|
||
FROM insurance_chat_records
|
||
WHERE {where_clause}
|
||
GROUP BY user_id
|
||
ORDER BY total_tokens DESC
|
||
LIMIT 50
|
||
""")
|
||
user_result = db.session.execute(user_sql, bind_params).fetchall()
|
||
by_user = [row._asdict() for row in user_result]
|
||
logging.info(f"[TOKEN_USAGE] 按用户统计: {len(by_user)} 个用户, data={by_user[:3]}")
|
||
|
||
# 4. 日趋势（最近30天）
|
||
trend_sql = text(f"""
|
||
SELECT
|
||
DATE(created_at) as date,
|
||
COALESCE(SUM(CASE WHEN role = 'assistant' THEN answer_tokens ELSE message_tokens END), 0) as total_tokens
|
||
FROM insurance_chat_records
|
||
WHERE {where_clause}
|
||
GROUP BY DATE(created_at)
|
||
ORDER BY date DESC
|
||
LIMIT 30
|
||
""")
|
||
trend_result = db.session.execute(trend_sql, bind_params).fetchall()
|
||
daily_trend = [row._asdict() for row in trend_result]
|
||
|
||
return {
|
||
"code": 0,
|
||
"data": {
|
||
"total_tokens": total_result.total_tokens if total_result else 0,
|
||
"total_input_tokens": total_result.total_input_tokens if total_result else 0,
|
||
"total_output_tokens": total_result.total_output_tokens if total_result else 0,
|
||
"by_model": by_model,
|
||
"by_user": by_user,
|
||
"daily_trend": daily_trend,
|
||
},
|
||
}
|
||
except Exception as e:
|
||
import logging
|
||
logging.warning("get_token_usage 查询失败: %s", e)
|
||
return {
|
||
"code": 0,
|
||
"data": {
|
||
"total_tokens": 0,
|
||
"total_input_tokens": 0,
|
||
"total_output_tokens": 0,
|
||
"by_model": [],
|
||
"by_user": [],
|
||
"daily_trend": [],
|
||
},
|
||
}
|
||
|
||
def get_token_models(self) -> dict:
|
||
"""获取可用的模型列表。查询 insurance_chat_records 表。"""
|
||
try:
|
||
default_model = self._get_default_model_name()
|
||
sql = text("""
|
||
SELECT DISTINCT COALESCE(NULLIF(model_id, ''), :default_model) as model
|
||
FROM insurance_chat_records
|
||
ORDER BY model
|
||
""")
|
||
result = db.session.execute(sql, {"default_model": default_model}).fetchall()
|
||
models = [row[0] for row in result if row[0]]
|
||
return {"code": 0, "data": models}
|
||
except Exception:
|
||
return {"code": 0, "data": []}
|
||
|
||
def get_token_cost(self, start_date: str, end_date: str, group_by: str = "model") -> dict:
|
||
"""Token 成本统计（兼容旧接口）。"""
|
||
params = {
|
||
"start_date": start_date,
|
||
"end_date": end_date,
|
||
}
|
||
result = self.get_token_usage(params)
|
||
return result
|