Spaces:

a3216
/

xtc-backend

Running

App Files Files Community

a3216 commited on Jun 29

Commit

41577fb

verified ·

1 Parent(s): bc29ff1

sync from GitHub dc52e4f: fix: 修复文件上传999报错与Web管理手机端适配

Browse files

Auto-synced from GitHub commit dc52e4f9ae5eb2d5b5e506091153c6d13a8bbf29

Files changed (8) hide show

app/admin_html.py +84 -1
app/api/_common.py +36 -19
app/api/pseudo_stream.py +14 -2
app/api/xtc.py +7 -4
app/db_writer.py +76 -0
app/main.py +4 -0
app/services/config_store.py +36 -22
app/services/request_log_store.py +86 -49

app/admin_html.py CHANGED Viewed

@@ -171,10 +171,63 @@ _HTML_TEMPLATE = """<!DOCTYPE html>
   details.req-block summary { padding: 8px 12px; cursor: pointer; font-size: 13px; font-weight: 500; }
   details.req-block[open] summary { border-bottom: 1px solid var(--border); }
   details.req-block pre { margin: 0; border: none; border-radius: 0; max-height: 320px; }
 </style>
 </head>
 <body data-disabled="__DISABLED_ATTR__">
 <header>
   <h1>XTC 后台管理</h1>
   <div class="row" style="gap:8px">
     <span class="svc-status" id="svcStatus">服务: 检测中</span>
@@ -491,13 +544,43 @@ function esc(s) {
   return String(s==null?'':s).replace(/[&<>"']/g, c => ({'&':'&amp;','<':'&lt;','>':'&gt;','"':'&quot;',"'":'&#39;'}[c]));
 }
-// Tab 切换
 document.querySelectorAll('nav.tabs button').forEach(btn => {
   btn.onclick = () => {
     document.querySelectorAll('nav.tabs button').forEach(b => b.classList.remove('active'));
     btn.classList.add('active');
     const tab = btn.dataset.tab;
     document.querySelectorAll('.tab-content').forEach(c => c.classList.toggle('active', c.dataset.tab === tab));
     if (tab === 'overview') loadOverview();
     else if (tab === 'providers') loadProviders();
     else if (tab === 'tokens') loadTokens();

   details.req-block summary { padding: 8px 12px; cursor: pointer; font-size: 13px; font-weight: 500; }
   details.req-block[open] summary { border-bottom: 1px solid var(--border); }
   details.req-block pre { margin: 0; border: none; border-radius: 0; max-height: 320px; }
+  /* ===== 响应式：手机适配 ===== */
+  /* 移动端菜单按钮（默认隐藏，窄屏显示） */
+  .menu-toggle { display: none; }
+  /* 表格横向滚动容器（窄屏防溢出） */
+  .table-wrap { overflow-x: auto; -webkit-overflow-scrolling: touch; }
+  .table-wrap table { min-width: 480px; }
+  /* 窄屏（<768px）：侧边栏改为可折叠抽屉，主内容全宽 */
+  @media (max-width: 768px) {
+    header { padding: 10px 12px; gap: 8px; }
+    header h1 { font-size: 16px; }
+    .menu-toggle {
+      display: inline-block;
+      background: var(--panel2); border: 1px solid var(--border); color: var(--text);
+      padding: 6px 10px; border-radius: 6px; font-size: 18px; line-height: 1;
+      cursor: pointer; flex: 0 0 auto;
+    }
+    .layout { flex-direction: column; }
+    nav.tabs {
+      width: 100%; flex-shrink: 1; flex-wrap: wrap; gap: 4px;
+      padding: 0 8px 8px; border-right: none; border-bottom: 1px solid var(--border);
+      display: none;  /* 默认折叠，点菜单按钮展开 */
+    }
+    nav.tabs.open { display: flex; }
+    nav.tabs button { flex: 1 1 auto; text-align: center; }
+    main { padding: 12px; }
+    /* KPI/stats 网格窄屏降为 2 列 */
+    .kpi-grid, .stats-grid { grid-template-columns: repeat(2, 1fr); gap: 8px; }
+    .kpi .value { font-size: 18px; }
+    /* filter-row 控件窄屏单列堆叠 */
+    .filter-row { gap: 6px; }
+    .filter-row input, .filter-row select, .filter-row .grow, .filter-row .shrink {
+      flex: 1 1 100%; min-width: 0;
+    }
+    .filter-row button { flex: 1 1 calc(50% - 3px); }
+    /* 分段选择器窄屏允许换行 */
+    .seg { flex-wrap: wrap; width: 100%; }
+    .seg button { flex: 1 1 auto; }
+    /* 流量分析横排窄屏换行 */
+    .ring-wrap, .traffic-head { flex-wrap: wrap; }
+    .bar-chart { height: 70px; }
+    .msg-bubble { max-width: 95%; }
+    /* 卡片内边距收紧 */
+    .card { padding: 12px; }
+  }
+  /* 极窄屏（<400px）：KPI 单列 */
+  @media (max-width: 400px) {
+    .kpi-grid, .stats-grid { grid-template-columns: 1fr; }
+  }
 </style>
 </head>
 <body data-disabled="__DISABLED_ATTR__">
 <header>
+  <button class="menu-toggle" onclick="toggleNav()" aria-label="菜单">≡</button>
   <h1>XTC 后台管理</h1>
   <div class="row" style="gap:8px">
     <span class="svc-status" id="svcStatus">服务: 检测中</span>
   return String(s==null?'':s).replace(/[&<>"']/g, c => ({'&':'&amp;','<':'&lt;','>':'&gt;','"':'&quot;',"'":'&#39;'}[c]));
 }
+// 手机端侧边栏抽屉开关
+function toggleNav() {
+  document.querySelector('nav.tabs').classList.toggle('open');
+}
+// 自动给裸 table 包裹横向滚动容器（窄屏防溢出）。
+// 用 MutationObserver 监听动态插入的表格，避免逐处改 innerHTML。
+(function wrapTables() {
+  function wrapAll(root) {
+    root.querySelectorAll('table').forEach(t => {
+      if (t.parentElement && t.parentElement.classList.contains('table-wrap')) return;
+      const w = document.createElement('div');
+      w.className = 'table-wrap';
+      t.parentNode.insertBefore(w, t);
+      w.appendChild(t);
+    });
+  }
+  if (document.readyState === 'loading') {
+    document.addEventListener('DOMContentLoaded', () => {
+      wrapAll(document);
+      new MutationObserver(muts => muts.forEach(m => m.addedNodes.forEach(n => {
+        if (n.nodeType === 1) wrapAll(n);
+      }))).observe(document.body, { childList: true, subtree: true });
+    });
+  } else {
+    wrapAll(document);
+  }
+})();
 document.querySelectorAll('nav.tabs button').forEach(btn => {
   btn.onclick = () => {
     document.querySelectorAll('nav.tabs button').forEach(b => b.classList.remove('active'));
     btn.classList.add('active');
     const tab = btn.dataset.tab;
     document.querySelectorAll('.tab-content').forEach(c => c.classList.toggle('active', c.dataset.tab === tab));
+    // 手机端选完 tab 后自动收起侧边栏
+    document.querySelector('nav.tabs').classList.remove('open');
     if (tab === 'overview') loadOverview();
     else if (tab === 'providers') loadProviders();
     else if (tab === 'tokens') loadTokens();

app/api/_common.py CHANGED Viewed

@@ -65,26 +65,43 @@ def record_usage(
     ok: bool,
     error_code: Optional[str] = None,
 ) -> None:
     try:
-        now = int(time.time())
-        with get_conn() as conn:
-            conn.execute(
-                "INSERT INTO usage_log(ts, access_key, provider, model, prompt_tokens, completion_tokens, total_tokens, ok, error_code) "
-                "VALUES(?,?,?,?,?,?,?,?,?)",
-                (
-                    now,
-                    access_key,
-                    provider,
-                    model,
-                    int((usage or {}).get("prompt_tokens") or 0),
-                    int((usage or {}).get("completion_tokens") or 0),
-                    int((usage or {}).get("total_tokens") or 0),
-                    1 if ok else 0,
-                    error_code,
-                ),
-            )
-    except Exception as e:
-        print(f"[usage] record failed: {e}")
     # Webhook 通知（fire-and-forget，失败不影响主流程）
     try:

     ok: bool,
     error_code: Optional[str] = None,
 ) -> None:
+    now = int(time.time())
+    sql = (
+        "INSERT INTO usage_log(ts, access_key, provider, model, prompt_tokens, completion_tokens, total_tokens, ok, error_code) "
+        "VALUES(?,?,?,?,?,?,?,?,?)"
+    )
+    args = (
+        now,
+        access_key,
+        provider,
+        model,
+        int((usage or {}).get("prompt_tokens") or 0),
+        int((usage or {}).get("completion_tokens") or 0),
+        int((usage or {}).get("total_tokens") or 0),
+        1 if ok else 0,
+        error_code,
+    )
+    def _task(c):
+        try:
+            c.execute(sql, args)
+        except Exception as e:
+            print(f"[usage] record failed: {e}")
+    # 转交后台写线程，避免在事件循环上同步 INSERT
     try:
+        from ..db_writer import enqueue, is_started
+        if is_started():
+            enqueue(_task)
+        else:
+            with get_conn() as conn:
+                _task(conn)
+    except Exception:
+        try:
+            with get_conn() as conn:
+                _task(conn)
+        except Exception as e:
+            print(f"[usage] record failed: {e}")
     # Webhook 通知（fire-and-forget，失败不影响主流程）
     try:

app/api/pseudo_stream.py CHANGED Viewed

@@ -25,6 +25,7 @@ from ._common import (
     record_usage,
     select_provider_and_key,
 )
 router = APIRouter(prefix="/v1/xtc/chat/pseudo", tags=["pseudo-stream"])
@@ -35,8 +36,19 @@ async def pseudo_start(
     _key: str = Depends(require_access_key),
     x_provider: Optional[str] = Header(default=None, alias="x-provider"),
 ):
-    """启动伪流式会话：立即返回 session_id，后台异步执行对话。"""
-    body = await request.json()
     if not isinstance(body, dict):
         raise HttpError("invalid body", status=400, code="bad_request")

     record_usage,
     select_provider_and_key,
 )
+from .xtc import _parse_multipart
 router = APIRouter(prefix="/v1/xtc/chat/pseudo", tags=["pseudo-stream"])
     _key: str = Depends(require_access_key),
     x_provider: Optional[str] = Header(default=None, alias="x-provider"),
 ):
+    """启动伪流式会话：立即返回 session_id，后台异步执行对话。
+    支持 JSON 与 multipart/form-data 两种请求体。multipart 用于带文件/图片
+    的场景：传统做法是带文件走 /v1/xtc/chat 的长连接（stream=false，后端等
+    上游完整返回），但手表平台的 request.upload 默认超时较短（约 30s），
+    上游 LLM 生成耗时超过该值会被平台单方面掐断并报 999。让带文件的请求也
+    走伪流式（立即返回 session_id，后台处理），可避开长连接超时。
+    """
+    content_type = (request.headers.get("content-type") or "").lower()
+    if "multipart/form-data" in content_type:
+        body = await _parse_multipart(request)
+    else:
+        body = await request.json()
     if not isinstance(body, dict):
         raise HttpError("invalid body", status=400, code="bad_request")

app/api/xtc.py CHANGED Viewed

@@ -5,6 +5,7 @@
 from __future__ import annotations
 import json
 from typing import Any, AsyncIterator, Optional
 from fastapi import APIRouter, Depends, Header, Query, Request
@@ -33,6 +34,10 @@ from ._common import (
 router = APIRouter(prefix="/v1/xtc", tags=["xtc"])
 @router.get("/providers")
 async def list_providers(_key: str = Depends(require_access_key)) -> dict:
@@ -388,13 +393,11 @@ async def _simple_stream_gen(
     access_key: str,
 ) -> AsyncIterator[str]:
     """simple 模式：把 OpenAI chunk 流转换为 {type: text/thought/done, delta}。"""
-    import re
     in_thought = False
     thought_open_sent = False
     thought_close_sent = False
-    open_tag_re = re.compile(r"<thought>\s*$", re.IGNORECASE)
-    close_tag_re = re.compile(r"^\s*</thought>", re.IGNORECASE)
     try:
         if provider.type == "gemini":

 from __future__ import annotations
 import json
+import re
 from typing import Any, AsyncIterator, Optional
 from fastapi import APIRouter, Depends, Header, Query, Request
 router = APIRouter(prefix="/v1/xtc", tags=["xtc"])
+# simple 流式：thought 标签识别（提到模块级，避免每次调用重新 import + compile）
+_THOUGHT_OPEN_RE = re.compile(r"<thought>\s*$", re.IGNORECASE)
+_THOUGHT_CLOSE_RE = re.compile(r"^\s*</thought>", re.IGNORECASE)
 @router.get("/providers")
 async def list_providers(_key: str = Depends(require_access_key)) -> dict:
     access_key: str,
 ) -> AsyncIterator[str]:
     """simple 模式：把 OpenAI chunk 流转换为 {type: text/thought/done, delta}。"""
     in_thought = False
     thought_open_sent = False
     thought_close_sent = False
+    open_tag_re = _THOUGHT_OPEN_RE
+    close_tag_re = _THOUGHT_CLOSE_RE
     try:
         if provider.type == "gemini":

app/db_writer.py ADDED Viewed

	@@ -0,0 +1,76 @@

+"""后台 DB 写入线程：把分析类写入（usage_log / request_log 等）移出事件循环。
+为什么是「单独线程 + 独立连接」而不是 asyncio.to_thread：
+- sqlite3.Connection 即使设了 check_same_thread=False，也不允许跨线程并发使用
+  （并发会抛 ProgrammingError 或损坏游标状态）。主事件循环线程负责读，本写线程用
+  独立连接负责写，靠 WAL 模式实现读写并发，互不阻塞。
+- 写入是 fire-and-forget 的分析数据，丢失不影响主流程：任务异常仅记日志。
+"""
+from __future__ import annotations
+import logging
+import queue
+import sqlite3
+import threading
+from typing import Callable, Optional
+_log = logging.getLogger(__name__)
+_q: "queue.Queue[Optional[Callable[[sqlite3.Connection], None]]]" = queue.Queue()
+_conn: Optional[sqlite3.Connection] = None
+_thread: Optional[threading.Thread] = None
+_started = False
+def start() -> None:
+    """启动后台写线程。必须在 init_db() 之后调用（依赖已建好的 schema）。"""
+    global _conn, _thread, _started
+    if _started:
+        return
+    _started = True
+    from .database import _resolve_db_path
+    db_path = _resolve_db_path()
+    _conn = sqlite3.connect(db_path, check_same_thread=False, isolation_level=None)
+    _conn.row_factory = sqlite3.Row
+    _conn.execute("PRAGMA journal_mode=WAL")
+    _conn.execute("PRAGMA synchronous=NORMAL")
+    def _loop() -> None:
+        while True:
+            item = _q.get()
+            if item is None:  # 关闭信号
+                break
+            try:
+                item(_conn)  # type: ignore[arg-type]
+            except Exception as e:
+                _log.warning("db_writer task failed: %s", e)
+    _thread = threading.Thread(target=_loop, daemon=True, name="xtc-db-writer")
+    _thread.start()
+def stop() -> None:
+    """停止后台写线程，drain 队列。"""
+    global _conn, _started
+    if not _started:
+        return
+    _q.put(None)
+    if _thread is not None:
+        _thread.join(timeout=2.0)
+    if _conn is not None:
+        try:
+            _conn.close()
+        except Exception:
+            pass
+    _conn = None
+    _started = False
+def enqueue(fn: Callable[[sqlite3.Connection], None]) -> None:
+    """提交一个写任务。``fn`` 接收写线程的连接，在写线程内同步执行。"""
+    _q.put(fn)
+def is_started() -> bool:
+    return _started

app/main.py CHANGED Viewed

@@ -11,6 +11,7 @@ from fastapi.responses import JSONResponse
 from .config import get_settings
 from .database import close_db, init_db
 from .errors import HttpError, http_error_handler, unhandled_exception_handler
 from .http_client import close_http_client
 from .services import config_store, pseudo_store
@@ -21,6 +22,8 @@ async def lifespan(app: FastAPI):
     # 启动
     logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
     init_db()
     # 预热配置
     try:
         cfg = await config_store.load_config()
@@ -40,6 +43,7 @@ async def lifespan(app: FastAPI):
     except asyncio.CancelledError:
         pass
     await close_http_client()
     close_db()

 from .config import get_settings
 from .database import close_db, init_db
+from .db_writer import start as start_db_writer, stop as stop_db_writer
 from .errors import HttpError, http_error_handler, unhandled_exception_handler
 from .http_client import close_http_client
 from .services import config_store, pseudo_store
     # 启动
     logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
     init_db()
+    # 启动后台 DB 写线程（usage_log / request_log 等分析类写入移出事件循环）
+    start_db_writer()
     # 预热配置
     try:
         cfg = await config_store.load_config()
     except asyncio.CancelledError:
         pass
     await close_http_client()
+    stop_db_writer()
     close_db()

app/services/config_store.py CHANGED Viewed

@@ -28,6 +28,10 @@ _HUB_CONFIG_KEY = "app_config.json"
 _lock = asyncio.Lock()
 def _now_ts() -> int:
     return int(time.time())
@@ -93,39 +97,49 @@ def _seed_from_env() -> Optional[AppConfig]:
 async def load_config() -> AppConfig:
-    """加载配置（DB 优先 → HF Hub 拉取 → env 种子）。"""
     async with _lock:
         cfg = _load_from_db()
-        if cfg.providers:
-            return cfg.normalize()
-        # 尝试从 HF Hub 拉取配置备份
-        try:
-            hub_cfg = await _load_from_hub()
-            if hub_cfg and hub_cfg.providers:
-                _save_to_db(hub_cfg)
-                return hub_cfg.normalize()
-        except Exception as e:
-            print(f"[config_store] HF Hub load failed: {e}")
-        # 用 env 种子
-        seeded = _seed_from_env()
-        if seeded:
-            _save_to_db(seeded)
             try:
-                await _push_to_hub(seeded)
             except Exception as e:
-                print(f"[config_store] HF Hub push failed: {e}")
-            return seeded.normalize()
-        return AppConfig(providers=[], default_provider_id=None).normalize()
 async def save_config(config: AppConfig) -> AppConfig:
     """保存配置：写 DB + 异步推 HF Hub + Webhook 通知。"""
     config.normalize()
     async with _lock:
         _save_to_db(config)
     # HF Hub 同步失败不影响主流程
     try:
         await _push_to_hub(config)

 _lock = asyncio.Lock()
+# 内存缓存：命中后直接返回深拷贝，避免每请求 SELECT kv + json.loads + pydantic 校验，
+# 也消除 _lock 造成的并发请求串行化。save_config 时更新，保证写后即生效。
+_cache: Optional[AppConfig] = None
 def _now_ts() -> int:
     return int(time.time())
 async def load_config() -> AppConfig:
+    """加载配置（DB 优先 → HF Hub 拉取 → env 种子）。
+    热路径命中内存缓存时直接返回深拷贝（无 DB、无校验、无锁争用）。
+    """
+    global _cache
+    if _cache is not None:
+        return _cache.model_copy(deep=True)
     async with _lock:
+        # double-check：并发请求可能在等锁期间已被前一个持有者填充
+        if _cache is not None:
+            return _cache.model_copy(deep=True)
         cfg = _load_from_db()
+        if not cfg.providers:
+            # 尝试从 HF Hub 拉取配置备份
             try:
+                hub_cfg = await _load_from_hub()
+                if hub_cfg and hub_cfg.providers:
+                    _save_to_db(hub_cfg)
+                    cfg = hub_cfg
             except Exception as e:
+                print(f"[config_store] HF Hub load failed: {e}")
+        if not cfg.providers:
+            # 用 env 种子
+            seeded = _seed_from_env()
+            if seeded:
+                _save_to_db(seeded)
+                try:
+                    await _push_to_hub(seeded)
+                except Exception as e:
+                    print(f"[config_store] HF Hub push failed: {e}")
+                cfg = seeded
+        cfg = cfg.normalize()
+        _cache = cfg
+        return cfg.model_copy(deep=True)
 async def save_config(config: AppConfig) -> AppConfig:
     """保存配置：写 DB + 异步推 HF Hub + Webhook 通知。"""
+    global _cache
     config.normalize()
     async with _lock:
         _save_to_db(config)
+        _cache = config.model_copy(deep=True)
     # HF Hub 同步失败不影响主流程
     try:
         await _push_to_hub(config)

app/services/request_log_store.py CHANGED Viewed

@@ -79,20 +79,29 @@ def _truncate(s: Optional[str], max_len: int) -> Optional[str]:
 # ===== 容量配置 =====
 def get_limit() -> int:
-    """读取容量上限。0 = 禁用记录。"""
     try:
         with get_conn() as conn:
             row = conn.execute(
                 "SELECT value FROM kv WHERE key = 'request_log_limit'"
             ).fetchone()
-            return int(row["value"]) if row else 500
     except Exception:
         return 500
 def set_limit(limit: int) -> int:
     """设置容量上限，并立即触发轮转。"""
     limit = max(0, min(5000, int(limit)))
     now = int(time.time())
     with get_conn() as conn:
@@ -101,70 +110,98 @@ def set_limit(limit: int) -> int:
             "ON CONFLICT(key) DO UPDATE SET value=excluded.value, updated_at=excluded.updated_at",
             ("request_log_limit", str(limit), now),
         )
     if limit > 0:
         _rotate(limit)
     return limit
 def _rotate(limit: int) -> int:
-    """删除超限的旧记录。返回删除条数。"""
     with get_conn() as conn:
-        # 找到要保留的最小 id
-        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
 # ===== 写入 =====
 def insert(record: dict) -> Optional[int]:
-    """插入一条请求日志。返回 id（如果 limit=0 则不插入返回 None）。"""
     limit = get_limit()
     if limit <= 0:
         return None
     now = int(time.time())
     with get_conn() as conn:
-        cur = conn.execute(
-            """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
-            ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
-            (
-                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),
-            ),
-        )
-        rid = int(cur.lastrowid)
-    # 轮转（异步感觉没必要，每次插入后检查一次成本低）
-    # 用阈值触发，避免每次都查 count
-    if rid % 50 == 0:
-        _rotate(limit)
-    return rid
 # ===== 查询 =====

 # ===== 容量配置 =====
+# 内存缓存：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:
             "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),
+    )
+    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
+        ) 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
 # ===== 查询 =====