"""请求日志存储:完整记录每个业务请求的入参/出参/耗时/错误。 借鉴 ds2api 的设计: - 完整记录请求标头、请求体、响应体、响应标头、耗时、错误码 - 容量轮转:保留最近 N 条(默认 500,可配 0/100/200/500/1000) - 列表查询返回摘要(不带大 body),详情查询返回完整 - 敏感信息脱敏(authorization / api_key / x-xtc-admin-key 等) 记录范围:/v1/* 和 /admin/api/* 路径 不记录:/health /docs /openapi.json /redoc /admin(HTML 页面)静态资源 """ from __future__ import annotations import json import re import time from typing import Any, Optional from ..database import get_conn # ===== 常量 ===== # 敏感标头/key 脱敏 _SENSITIVE_HEADERS = { "authorization", "x-xtc-admin-key", "x-xtc-access-key", "x-xtc-access-token", "cookie", "x-api-key", "x-hf-token", "hf-token", } _SENSITIVE_BODY_KEYS = { "api_key", "apikey", "api_keys", "apikey", "key", "token", "access_token", "access_key", "admin_key", "password", "secret", "authorization", } # body 最大记录长度(超过截断) _MAX_BODY_LEN = 64 * 1024 # 64KB # 标头最大记录长度 _MAX_HEADERS_LEN = 16 * 1024 # 16KB # ===== 脱敏 ===== def _redact_headers(headers: dict) -> dict: """脱敏敏感标头:保留前 4 位 + ***。""" out = {} for k, v in headers.items(): kl = k.lower() if kl in _SENSITIVE_HEADERS: sv = str(v) if len(sv) <= 8: out[k] = "***" else: out[k] = sv[:4] + "***" + sv[-2:] else: out[k] = v return out def _redact_body(body: Any) -> Any: """递归脱敏 body 中的敏感字段。""" if isinstance(body, dict): out = {} for k, v in body.items(): if k.lower() in _SENSITIVE_BODY_KEYS: out[k] = "***REDACTED***" else: out[k] = _redact_body(v) return out if isinstance(body, list): return [_redact_body(item) for item in body] return body def _truncate(s: Optional[str], max_len: int) -> Optional[str]: if s is None: return None if len(s) <= max_len: return s return s[:max_len] + f"\n...[truncated, total {len(s)} bytes]" # 设备 ID 长度上限(防止异常长字符串刷屏/撑爆索引) _MAX_DEVICE_ID_LEN = 128 def _normalize_device_id(raw: Any) -> str: """归一化设备 ID:去空白、截断、空值返回空字符串(确保入库)。""" if raw is None: return "" s = str(raw).strip() if not s: return "" if len(s) > _MAX_DEVICE_ID_LEN: s = s[:_MAX_DEVICE_ID_LEN] return s # ===== 容量配置 ===== # 内存缓存:limit 只在管理员改设置时变化,缓存后避免每次 insert/list 都 SELECT kv。 _limit_cache: Optional[int] = None def get_limit() -> int: """读取容量上限。0 = 禁用记录。命中缓存直接返回。""" global _limit_cache if _limit_cache is not None: return _limit_cache try: with get_conn() as conn: row = conn.execute( "SELECT value FROM kv WHERE key = 'request_log_limit'" ).fetchone() _limit_cache = int(row["value"]) if row else 500 except Exception: return 500 return _limit_cache def set_limit(limit: int) -> int: """设置容量上限,并立即触发轮转。""" global _limit_cache limit = max(0, min(5000, int(limit))) now = int(time.time()) with get_conn() as conn: conn.execute( "INSERT INTO kv(key, value, updated_at) VALUES(?,?,?) " "ON CONFLICT(key) DO UPDATE SET value=excluded.value, updated_at=excluded.updated_at", ("request_log_limit", str(limit), now), ) _limit_cache = limit if limit > 0: _rotate(limit) return limit def _rotate(limit: int) -> int: """删除超限的旧记录(事件循环线程调用)。返回删除条数。""" with get_conn() as conn: return _do_rotate(conn, limit) def _do_rotate(conn, limit: int) -> int: """在给定连接上执行轮转。""" row = conn.execute( "SELECT id FROM request_log ORDER BY id DESC LIMIT 1 OFFSET ?", (limit,), ).fetchone() if not row: return 0 threshold = int(row["id"]) cur = conn.execute("DELETE FROM request_log WHERE id < ?", (threshold,)) return cur.rowcount # ===== 写入 ===== # 后台写线程的插入计数器,用于阈值触发轮转(单线程写,无需加锁)。 _insert_count = 0 def insert(record: dict) -> Optional[int]: """插入一条请求日志(异步:转交后台写线程,立即返回 None)。 limit=0 时不插入。脱敏/截断在事件循环上完成(开销小),INSERT + 轮转 在后台写线程执行,避免阻塞事件循环。 """ global _insert_count limit = get_limit() if limit <= 0: return None now = int(time.time()) args = ( now, record.get("method", ""), record.get("path", ""), record.get("query"), record.get("status_code"), record.get("elapsed_ms"), record.get("access_key"), record.get("client_ip"), record.get("user_agent"), record.get("provider"), record.get("model"), 1 if record.get("stream") else 0, 1 if record.get("ok") else 0, record.get("error_code"), record.get("error_message"), _truncate(record.get("request_headers"), _MAX_HEADERS_LEN), _truncate(record.get("request_body"), _MAX_BODY_LEN), _truncate(record.get("response_body"), _MAX_BODY_LEN), _truncate(record.get("response_headers"), _MAX_HEADERS_LEN), _normalize_device_id(record.get("device_id")), ) sql = ( """INSERT INTO request_log( ts, method, path, query, status_code, elapsed_ms, access_key, client_ip, user_agent, provider, model, stream, ok, error_code, error_message, request_headers, request_body, response_body, response_headers, device_id ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""" ) rotate_limit = limit def _task(c): global _insert_count c.execute(sql, args) _insert_count += 1 # 每 50 次写入触发一次轮转(阈值触发,避免每次都查) if _insert_count % 50 == 0: _do_rotate(c, rotate_limit) try: from ..db_writer import enqueue, is_started if is_started(): enqueue(_task) return None except Exception: pass # 回退:写线程未启动(如启动前/测试),在事件循环上内联写入 with get_conn() as conn: conn.execute(sql, args) return None # ===== 查询 ===== def _category_filter(category: Optional[str]) -> tuple[str, list]: """返回 (SQL 片段, 参数) 用于按请求类别过滤。 - ``business``(默认):只看业务请求 ``/v1/*``(对话/伪流式/OpenAI 兼容等) - ``admin``:只看管理面板访问 ``/admin/api/*`` - ``all`` 或其它:不过滤 """ cat = str(category or "business").lower() if cat == "admin": return " AND path LIKE '/admin/api/%'", [] if cat == "all": return "", [] # 默认 business:排除管理面板访问,只保留 /v1/* return " AND path LIKE '/v1/%'", [] def list_records( *, limit: int = 50, offset: int = 0, path: Optional[str] = None, method: Optional[str] = None, status: Optional[str] = None, access_key: Optional[str] = None, ok: Optional[bool] = None, hours: Optional[int] = None, keyword: Optional[str] = None, category: Optional[str] = None, device_id: Optional[str] = None, ) -> dict: """列表查询:返回摘要(不含 request_body / response_body / headers)。""" sql = "SELECT id, ts, method, path, query, status_code, elapsed_ms, access_key, client_ip, user_agent, provider, model, stream, ok, error_code, error_message, device_id FROM request_log WHERE 1=1" args: list = [] cat_sql, cat_args = _category_filter(category) if cat_sql: sql += cat_sql args += cat_args if path: sql += " AND path LIKE ?" args.append(f"%{path}%") if method: sql += " AND method = ?" args.append(method.upper()) if status: # 支持 2xx/3xx/4xx/5xx 通配 s = str(status).strip() if s.endswith("xx"): prefix = s[0] sql += " AND CAST(status_code/100 AS INTEGER) = ?" args.append(int(prefix)) else: sql += " AND status_code = ?" args.append(int(s)) if access_key: sql += " AND access_key = ?" args.append(access_key) if ok is not None: sql += " AND ok = ?" args.append(1 if ok else 0) if hours: since = int(time.time()) - hours * 3600 sql += " AND ts >= ?" args.append(since) if keyword: sql += " AND (path LIKE ? OR error_message LIKE ? OR error_code LIKE ?)" kw = f"%{keyword}%" args += [kw, kw, kw] if device_id: sql += " AND device_id = ?" args.append(device_id) # 总数 count_sql = f"SELECT COUNT(*) AS c FROM ({sql})" sql += " ORDER BY ts DESC, id DESC LIMIT ? OFFSET ?" args += [limit, offset] with get_conn() as conn: total = int(conn.execute(count_sql, args[:-2]).fetchone()["c"]) rows = conn.execute(sql, args).fetchall() items = [_row_to_summary(r) for r in rows] return {"items": items, "count": len(items), "total": total, "limit": limit, "offset": offset} def get_record(rid: int) -> Optional[dict]: """详情查询:返回完整记录(含 body / headers)。""" with get_conn() as conn: row = conn.execute( "SELECT * FROM request_log WHERE id = ?", (rid,), ).fetchone() if not row: return None return _row_to_detail(row) def clear_all() -> int: """清空全部请求日志。返回删除条数。""" with get_conn() as conn: cur = conn.execute("DELETE FROM request_log") return cur.rowcount def clear_device_records(device_id: str) -> int: """清空某设备的全部请求日志。返回删除条数。""" if not device_id: return 0 with get_conn() as conn: cur = conn.execute( "DELETE FROM request_log WHERE device_id = ?", (device_id,) ) return cur.rowcount def delete_record(rid: int) -> bool: with get_conn() as conn: cur = conn.execute("DELETE FROM request_log WHERE id = ?", (rid,)) return cur.rowcount > 0 # ===== 行转换 ===== def _row_to_summary(row) -> dict: """摘要:不含大字段。""" return { "id": int(row["id"]), "ts": int(row["ts"]), "method": row["method"], "path": row["path"], "query": row["query"], "status_code": row["status_code"], "elapsed_ms": int(row["elapsed_ms"]) if row["elapsed_ms"] is not None else None, "access_key": row["access_key"], "client_ip": row["client_ip"], "user_agent": row["user_agent"], "provider": row["provider"], "model": row["model"], "stream": bool(row["stream"]), "ok": bool(row["ok"]), "error_code": row["error_code"], "error_message": row["error_message"], "device_id": row["device_id"], } def _row_to_detail(row) -> dict: """详情:含全部字段。""" d = _row_to_summary(row) # 解析 JSON 字段 for k in ("request_headers", "request_body", "response_body", "response_headers"): v = row[k] if v is None: d[k] = None continue # 尝试 JSON 解析 try: d[k] = json.loads(v) except Exception: d[k] = v return d # ===== 统计 ===== def stats(hours: int = 24, category: Optional[str] = None) -> dict: """请求日志统计。``category`` 语义同 :func:`list_records`。""" since = int(time.time()) - hours * 3600 cat_sql, cat_args = _category_filter(category) base_cond = " WHERE ts >= ?" base_cond += cat_sql with get_conn() as conn: total = conn.execute( "SELECT COUNT(*) AS c FROM request_log" + base_cond, [since] + cat_args, ).fetchone() by_status = conn.execute( "SELECT CAST(status_code/100 AS INTEGER) AS bucket, COUNT(*) AS c " "FROM request_log" + base_cond + " GROUP BY bucket ORDER BY bucket", [since] + cat_args, ).fetchall() by_path = conn.execute( "SELECT path, COUNT(*) AS c, AVG(elapsed_ms) AS avg_ms, " "SUM(CASE WHEN ok=1 THEN 1 ELSE 0 END) AS ok_c " "FROM request_log" + base_cond + " GROUP BY path ORDER BY c DESC LIMIT 20", [since] + cat_args, ).fetchall() by_error = conn.execute( "SELECT error_code, COUNT(*) AS c FROM request_log " + base_cond + " AND ok = 0 AND error_code IS NOT NULL " "GROUP BY error_code ORDER BY c DESC LIMIT 20", [since] + cat_args, ).fetchall() slow = conn.execute( "SELECT id, ts, method, path, elapsed_ms, status_code, ok " "FROM request_log" + base_cond + " AND elapsed_ms IS NOT NULL " "ORDER BY elapsed_ms DESC LIMIT 10", [since] + cat_args, ).fetchall() return { "hours": hours, "total": int(total["c"]) if total else 0, "by_status": {str(r["bucket"]) + "xx": int(r["c"]) for r in by_status}, "by_path": [ { "path": r["path"], "count": int(r["c"]), "avg_ms": round(float(r["avg_ms"]), 1) if r["avg_ms"] else 0, "ok": int(r["ok_c"]), } for r in by_path ], "by_error": [{"code": r["error_code"], "count": int(r["c"])} for r in by_error], "slowest": [ { "id": int(r["id"]), "ts": int(r["ts"]), "method": r["method"], "path": r["path"], "elapsed_ms": int(r["elapsed_ms"]), "status_code": r["status_code"], "ok": bool(r["ok"]), } for r in slow ], } # ===== 设备视图 ===== def list_devices( *, hours: Optional[int] = None, limit: int = 100, offset: int = 0, keyword: Optional[str] = None, ) -> dict: """按 device_id 聚合:列出有记录的设备及其统计。 返回每个设备的:总请求数、成功数、失败数、首次/最近活跃时间、 最近使用的 provider/model/client_ip/user_agent。 """ cond = " WHERE device_id != ''" args: list = [] if hours: since = int(time.time()) - hours * 3600 cond += " AND ts >= ?" args.append(since) if keyword: cond += " AND device_id LIKE ?" args.append(f"%{keyword}%") # 外层 SELECT 做 count + 最近值聚合 sql = ( "SELECT device_id, COUNT(*) AS total, " "SUM(CASE WHEN ok=1 THEN 1 ELSE 0 END) AS ok_c, " "SUM(CASE WHEN ok=0 THEN 1 ELSE 0 END) AS fail_c, " "MIN(ts) AS first_seen, MAX(ts) AS last_seen " "FROM request_log" + cond + " GROUP BY device_id " "ORDER BY last_seen DESC LIMIT ? OFFSET ?" ) args += [limit, offset] count_sql = ( "SELECT COUNT(*) AS c FROM (SELECT 1 FROM request_log" + cond + " GROUP BY device_id)" ) with get_conn() as conn: total = int(conn.execute(count_sql, args[:-2]).fetchone()["c"]) rows = conn.execute(sql, args).fetchall() items = [] for r in rows: did = r["device_id"] # 取该设备最近一条请求的 provider/model/ip/ua last = conn.execute( "SELECT provider, model, client_ip, user_agent, path, status_code, ts " "FROM request_log WHERE device_id = ? " "ORDER BY ts DESC, id DESC LIMIT 1", (did,), ).fetchone() items.append({ "device_id": did, "total": int(r["total"]), "ok": int(r["ok_c"]), "fail": int(r["fail_c"]), "first_seen": int(r["first_seen"]) if r["first_seen"] else None, "last_seen": int(r["last_seen"]) if r["last_seen"] else None, "last_provider": last["provider"] if last else None, "last_model": last["model"] if last else None, "last_client_ip": last["client_ip"] if last else None, "last_user_agent": last["user_agent"] if last else None, "last_path": last["path"] if last else None, "last_status": last["status_code"] if last else None, }) return {"items": items, "count": len(items), "total": total, "limit": limit, "offset": offset} def list_device_records( *, device_id: str, limit: int = 50, offset: int = 0, hours: Optional[int] = None, keyword: Optional[str] = None, scope: Optional[str] = None, ) -> dict: """某设备的最近请求记录(简化视图:含 request_body 用于提取文本)。 keyword: 关键词,在 request_body / response_body 上做 LIKE 搜索。 scope: 筛选范围,'req' 仅请求、'resp' 仅响应、'both' 或 None 两者都搜。 """ did = _normalize_device_id(device_id) if not did: return {"items": [], "count": 0, "total": 0, "limit": limit, "offset": offset} sql = ( "SELECT id, ts, method, path, query, status_code, elapsed_ms, " "access_key, client_ip, provider, model, stream, ok, " "error_code, error_message, request_body, response_body " "FROM request_log WHERE device_id = ?" ) args: list = [did] if hours: since = int(time.time()) - hours * 3600 sql += " AND ts >= ?" args.append(since) # 文本筛选 kw = (keyword or "").strip() if kw: s = (scope or "both").strip().lower() like = f"%{kw}%" if s == "req": sql += " AND request_body LIKE ?" args.append(like) elif s == "resp": sql += " AND response_body LIKE ?" args.append(like) else: # both / 默认:任一命中即可 sql += " AND (request_body LIKE ? OR response_body LIKE ?)" args += [like, like] count_sql = f"SELECT COUNT(*) AS c FROM ({sql})" sql += " ORDER BY ts DESC, id DESC LIMIT ? OFFSET ?" args += [limit, offset] with get_conn() as conn: total = int(conn.execute(count_sql, args[:-2]).fetchone()["c"]) rows = conn.execute(sql, args).fetchall() items = [_row_to_device_record(r) for r in rows] return {"items": items, "count": len(items), "total": total, "limit": limit, "offset": offset} def _row_to_device_record(row) -> dict: """设备视图专用记录:含请求/响应摘要 + 完整文本 + 原始 body(供下载)。""" # request_body 是 JSON 字符串(已在中间件层脱敏 json.dumps) rb_raw = row["request_body"] rb_parsed = None if rb_raw: try: rb_parsed = json.loads(rb_raw) except Exception: rb_parsed = rb_raw text_summary = _extract_text_summary(rb_parsed) text_full = _extract_text_full(rb_parsed) # response_body:流式取摘要+完整,非流式取完整 resp_raw = row["response_body"] resp_parsed = None if resp_raw: try: resp_parsed = json.loads(resp_raw) except Exception: resp_parsed = resp_raw resp_summary = _extract_response_summary(resp_parsed) resp_full = _extract_response_full(resp_parsed) return { "id": int(row["id"]), "ts": int(row["ts"]), "method": row["method"], "path": row["path"], "status_code": row["status_code"], "elapsed_ms": int(row["elapsed_ms"]) if row["elapsed_ms"] is not None else None, "access_key": row["access_key"], "client_ip": row["client_ip"], "provider": row["provider"], "model": row["model"], "stream": bool(row["stream"]), "ok": bool(row["ok"]), "error_code": row["error_code"], "error_message": row["error_message"], "text_summary": text_summary, "text_full": text_full, "response_summary": resp_summary, "response_full": resp_full, "request_raw": rb_raw or "", "response_raw": resp_raw or "", } def _extract_text_summary(body: Any) -> str: """从请求体提取用户输入文本(短摘要,用于卡片预览,≤200 字符)。""" full = _extract_text_full(body) if len(full) <= 200: return full return full[:200] + " …" def _extract_text_full(body: Any) -> str: """从请求体提取完整用户输入文本(不截断)。 兼容 OpenAI messages 形态、XTC input 简化形态、多模态 content 数组。 """ if body is None: return "" if isinstance(body, str): return body if not isinstance(body, dict): return str(body) # 1) 简化形态:input 字段 inp = body.get("input") if isinstance(inp, str) and inp.strip(): return inp.strip() # 2) messages 形态 msgs = body.get("messages") if isinstance(msgs, list) and msgs: parts: list[str] = [] for m in msgs: if not isinstance(m, dict): continue role = str(m.get("role") or "").strip() content = m.get("content") text = "" if isinstance(content, str): text = content elif isinstance(content, list): # OpenAI 多模态:[{type:text,text:...},{type:image_url,...}] chunks = [] for c in content: if isinstance(c, dict): if c.get("type") == "text" and c.get("text"): chunks.append(str(c["text"])) elif c.get("type") == "image_url": chunks.append("[图片]") text = " ".join(chunks) if text: prefix = f"[{role}] " if role else "" parts.append(prefix + text.strip()) return "\n".join(parts) return "" def _extract_response_summary(resp: Any) -> str: """从响应体提取简要文本(≤200 字符,用于卡片预览)。""" full = _extract_response_full(resp) if len(full) <= 200: return full return full[:200] + " …" def _extract_response_full(resp: Any) -> str: """从响应体提取完整文本(不截断)。 - 流式:合并所有 chunk 的 delta 文本 - 非流式:取 choices[0].message.content - 错误:拼错误信息 """ if resp is None: return "" if isinstance(resp, str): return resp if isinstance(resp, dict): # 流式摘要 if resp.get("type") == "streaming": chunks = resp.get("sampled_chunks") or [] total = resp.get("total_chunks") bytes_ = resp.get("total_bytes") # 合并所有 chunk 的 delta 文本 text_parts: list[str] = [] for c in chunks: if not isinstance(c, str): continue # SSE data: {...} for line in c.splitlines(): line = line.strip() if line.startswith("data:"): payload = line[5:].strip() if not payload or payload == "[DONE]": continue try: obj = json.loads(payload) except Exception: continue txt = _extract_delta_text(obj) if txt: text_parts.append(txt) joined = "".join(text_parts) header = f"[流式 {total} chunks / {bytes_} B]" return (header + "\n" + joined) if joined else header # OpenAI 非流式响应 choices = resp.get("choices") if isinstance(choices, list) and choices: first = choices[0] if isinstance(first, dict): msg = first.get("message") or {} content = msg.get("content") if isinstance(msg, dict) else None if isinstance(content, str) and content: return content if isinstance(content, list): texts = [] for c in content: if isinstance(c, dict) and c.get("type") == "text": texts.append(str(c.get("text") or "")) return "".join(texts) # 错误响应 if resp.get("error"): err = resp["error"] if isinstance(err, dict): return f"[错误] {err.get('code','')} {err.get('message','')}" return f"[错误] {err}" return str(resp) def _extract_delta_text(obj: Any) -> str: """从单个 OpenAI 流式 chunk 中提取增量文本。""" if not isinstance(obj, dict): return "" choices = obj.get("choices") if not isinstance(choices, list) or not choices: return "" first = choices[0] if not isinstance(first, dict): return "" delta = first.get("delta") or {} if isinstance(delta, dict): content = delta.get("content") if isinstance(content, str): return content return ""