性能优化（一）基础层：事件循环解冻 + 规则编译期 API + 接口小优化

- db.py 四端点、ai.py explain/gen_regex 从 async def 改 def：内部是同步阻塞的
  DB 建连（最长 10s）与 LLM 调用（最长 30s×2），挂事件循环上会冻结整个后端；
  def 走 FastAPI 线程池。explain 按 /ai/rule 同款补 run_in_executor
- rule_runner 新增 compile_rule() 编译期 API（regex→编译好的 Pattern；
  number/date/string→沙箱 exec 一次取出 check 闭包，skip_null/date 归一化烘焙
  进闭包）；run_rule 收敛为薄封装（单值测试端点继续用）；_exec_user_function
  拆为 _compile_user_function + _sandbox_namespace
- tasks.py list_tasks 批量计数（逐任务 3~4 条 SQL → 3 条 group_by 聚合）
- app.py 加 GZipMiddleware（minimum_size=1024）

检测引擎接入预编译（queries.py）与流式扫描在下一笔。
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

web/backend/app.py

@@ -18,6 +18,7 @@ except Exception:
 
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
+from fastapi.middleware.gzip import GZipMiddleware
 from fastapi.responses import HTMLResponse
 
 from web.backend._logging import setup_logging, get_logger
@@ -49,6 +50,9 @@ app.add_middleware(
     allow_methods=["*"],
     allow_headers=["*"],
 )
+# 2026-09-24：gzip 压缩大 JSON 响应（starlette 自带，无新依赖）。
+# 查询结果 / 任务列表这类响应行数多、文本重复度高，压缩比通常 5~10 倍。
+app.add_middleware(GZipMiddleware, minimum_size=1024)
 
 
 from web.backend.routers.ai import router as ai_router  # noqa: E402

web/backend/core/rule_runner.py

@@ -192,26 +192,33 @@ class RuleRunError(Exception):
     """规则执行失败（语法错 / 缺 check / 运行时异常），检测引擎按不合规处理并把原因暴露给 issues"""
 
 
-def _exec_user_function(code: str, value: Any) -> bool:
-    """在受限命名空间里执行用户提供的 Python 函数源码，调用 check(value) 返回 bool。
-
-    抛出 RuleRunError：
-      - SyntaxError / IndentationError 等语法错
-      - 缺少 check 函数定义
-      - check() 调用时抛出的异常
+def _sandbox_namespace() -> dict[str, Any]:
+    """构建受限 exec 命名空间（compile_rule / validate_user_code 共用）。
+
+    - __builtins__ 只放白名单 + 白名单版 __import__
+    - __name__ 用 "__main__" 而非 "<user_rule>"：dataclasses._is_type 内部会
+      `sys.modules.get(cls.__module__).__dict__`，__module__ 从 __name__ 推导，
+      必须指向 sys.modules 里真实存在的条目
+    - __qualname__：enum / dataclass 的 metaclass 会查
     """
-    namespace: dict[str, Any] = {
+    ns: dict[str, Any] = {
         "__builtins__": _SAFE_BUILTINS,
-        # 2026-08-21：enum.Enum / dataclasses.dataclass 的 metaclass 会查 __name__/__qualname__，
-        # 真 `class X:` 语句由 __build_class__ 自动注入；AI 生成的代码偶尔会写
-        # `@dataclasses.dataclass class P: ...` 这种也靠这个。
-        # 用 "__main__" 而不是 "<user_rule>"：dataclasses._is_type 内部会
-        # `sys.modules.get(cls.__module__).__dict__`，__module__ 从 __name__ 推导，
-        # 必须指向 sys.modules 里真实存在的条目。__main__ 是 Python 启动时一定在 sys.modules 里的。
         "__name__":     "__main__",
         "__qualname__": "user_rule",
     }
-    namespace.update(_PRELOAD_MODULES)    # 预注入模块
+    ns.update(_PRELOAD_MODULES)    # 预注入模块
+    return ns
+
+
+def _compile_user_function(code: str):
+    """在受限命名空间里 compile+exec 用户源码一次，返回其中的 check 可调用对象。
+
+    抛出 RuleRunError：语法错 / 模块级执行错误 / 缺 check 定义。
+    2026-09-24：从原 _exec_user_function（每行 compile+exec）拆出编译期 ——
+    检测引擎全表扫描只编译一次，逐行只调 check(value)。
+    """
+    if not code or not code.strip():
+        raise RuleRunError("规则代码为空，请填写 check(value) 函数")
 
     # compile() 先走一遍拿到语法错的具体行号
     try:
@@ -219,6 +226,7 @@ def _exec_user_function(code: str, value: Any) -> bool:
     except SyntaxError as e:
         raise RuleRunError(f"语法错误：{e.msg or e}") from e
 
+    namespace = _sandbox_namespace()
     try:
         exec(compiled, namespace)
     except Exception as e:
@@ -227,13 +235,7 @@ def _exec_user_function(code: str, value: Any) -> bool:
     check = namespace.get("check")
     if not callable(check):
         raise RuleRunError("代码中必须定义 check(value) 函数")
-
-    try:
-        result = check(value)
-    except Exception as e:
-        raise RuleRunError(f"check() 调用异常：{type(e).__name__}: {e}") from e
-
-    return bool(result)
+    return check
 
 
 # ── date 类型 value 预处理（patch） ─────────────────────
@@ -295,6 +297,81 @@ def _patch_date_value(val) -> Optional[datetime.datetime]:
 
 
 # ── 主入口 ──────────────────────────────────────────────
+# 2026-09-24：重构为「编译期 + 执行期」两段。
+#   compile_rule(...)  → check(value) -> bool 的可调用（编译失败抛 RuleRunError）
+#   run_rule(...)      → 编译一次 + 执行一次（单值测试场景用，如 ai.py /rule/test）
+# 检测引擎（queries.py）全表扫描改为 /start 时 compile_rule 一次、逐行只调闭包，
+# 不再每行 compile+exec 重建沙箱命名空间（10 万行 × N 条 code 规则的量级差距）。
+# 语义变化：code 规则的语法错/缺 check 从「每行报执行失败」变为编译期一次性暴露；
+# 用户 check 函数的模块级语句只执行一次，闭包状态跨行保持（规则应是纯函数）。
+def compile_rule(
+    rule_type: str,
+    regex: Optional[str] = None,
+    code: Optional[str] = None,
+    skip_null: bool = False,
+):
+    """编译一条规则，返回 check(value) -> bool 的可调用对象。
+
+    编译失败（regex 语法错 / code 语法错 / 缺 check / rule_type 非法）抛 RuleRunError，
+    调用方把坏规则标记出来（检测引擎对每行判不合规并注明原因）。
+    执行期异常同样包成 RuleRunError 抛出（含 date 值无法解析的情况）。
+    skip_null=True 时空值直接 True 短路（烘焙进闭包，只有一份逻辑）。
+    """
+    rt = rule_type or "regex"
+
+    if rt == "regex":
+        pattern_src = (regex or "").strip() or r"^.+$"   # 空 regex = 非空校验
+        try:
+            pat = _re.compile(pattern_src)
+        except _re.error as e:
+            raise RuleRunError(f"正则编译失败：{e}") from e
+
+        def _check_regex(value: Any) -> bool:
+            val_str = "" if value is None else str(value)
+            return pat.search(val_str) is not None
+
+        fn = _check_regex
+
+    elif rt in ("number", "date", "string"):
+        check = _compile_user_function(code or "")
+
+        if rt == "date":
+            def _check_date(value: Any) -> bool:
+                # 先 patch：保证 check 收到的是 datetime.datetime 对象
+                dt = _patch_date_value(value)
+                if dt is None:
+                    raise RuleRunError(f"无法解析为日期：{value!r}")
+                try:
+                    return bool(check(dt))
+                except RuleRunError:
+                    raise
+                except Exception as e:
+                    raise RuleRunError(f"check() 调用异常：{type(e).__name__}: {e}") from e
+            fn = _check_date
+        else:
+            def _check_code(value: Any) -> bool:
+                try:
+                    return bool(check(value))
+                except RuleRunError:
+                    raise
+                except Exception as e:
+                    raise RuleRunError(f"check() 调用异常：{type(e).__name__}: {e}") from e
+            fn = _check_code
+
+    else:
+        raise RuleRunError(f"未知的规则类型：{rule_type!r}")
+
+    if not skip_null:
+        return fn
+
+    def _skip_null_wrapper(value: Any) -> bool:
+        if _is_null_like(value):
+            return True          # 空值直接合规，不跑规则
+        return fn(value)
+
+    return _skip_null_wrapper
+
+
 def run_rule(
     rule_type: str,
     regex: Optional[str],
@@ -302,7 +379,7 @@ def run_rule(
     value: Any,
     skip_null: bool = False,
 ) -> bool:
-    """执行一条规则，返回 True = 通过 / False = 不通过。
+    """执行一条规则，返回 True = 通过 / False = 不通过（单值场景：编译一次 + 执行一次）。
 
     参数：
       rule_type: 'regex' | 'number' | 'date' | 'string'
@@ -310,46 +387,15 @@ def run_rule(
       code:      number/date/string 类型用；regex 传 None
       value:     数据源该字段的原始值（字符串或 None，或 datetime 对象等）
       skip_null: 为 True 且 value 是 None/空字符串 → 直接 True 跳过，不进 check/正则
-                 （2026-08-21：UI 上「跳过空值」开关，省掉每个 check 函数自己写 if value is None）
 
     date 类型的 value 会被 _patch_date_value 预先归一化成 datetime 对象，
-    用户 check(value) 收到的永远是 datetime.datetime，可以直接 value.date() / .year / .strftime() 用。
-    value 无法解析成日期时抛 RuleRunError，检测引擎把错误带进 issues 让用户看到原因。
-
-    抛出 RuleRunError：
-      - regex 编译失败
-      - date 值无法解析为日期
-      - 用户函数语法错 / 缺 check / 运行时异常
-      - rule_type 非法
-    """
-    rt = rule_type or "regex"
-
-    # 2026-08-21：跳过空值开关 —— 空值直接 True，不跑规则
-    if skip_null and _is_null_like(value):
-        return True
+    用户 check(value) 收到的永远是 datetime.datetime。value 无法解析成日期时抛
+    RuleRunError，调用方把错误带进 issues 让用户看到原因。
 
-    if rt == "regex":
-        pattern_src = (regex or "").strip() or r"^.+$"   # 空 regex = 非空校验
-        try:
-            pat = _re.compile(pattern_src)
-        except _re.error as e:
-            raise RuleRunError(f"正则编译失败：{e}") from e
-        val_str = "" if value is None else str(value)
-        return pat.search(val_str) is not None
-
-    if rt in ("number", "date", "string"):
-        if not code or not code.strip():
-            raise RuleRunError("规则代码为空，请填写 check(value) 函数")
-        if rt == "date":
-            # 先 patch：保证 check 收到的是 datetime.datetime 对象
-            dt = _patch_date_value(value)
-            if dt is None:
-                raise RuleRunError(f"无法解析为日期：{value!r}")
-            return _exec_user_function(code, dt)
-        # number / string：原样传 value 进 check
-        return _exec_user_function(code, value)
-
-    raise RuleRunError(f"未知的规则类型：{rule_type!r}")
+    全表扫描请用 compile_rule()（一次编译、逐行调用），本函数保留给
+    ai.py /rule/test 这类单值测试端点。
+    """
+    return compile_rule(rule_type, regex, code, skip_null=skip_null)(value)
 
 
 def _is_null_like(v: Any) -> bool:
@@ -380,8 +426,7 @@ def validate_user_code(code: str) -> tuple[bool, str]:
         compiled = compile(code, "<user_rule>", "exec")
     except SyntaxError as e:
         return False, f"语法错误：{e.msg or e}"
-    namespace: dict[str, Any] = {"__builtins__": _SAFE_BUILTINS}
-    namespace.update(_PRELOAD_MODULES)
+    namespace = _sandbox_namespace()
     try:
         exec(compiled, namespace)
     except Exception as e:

web/backend/routers/ai.py

@@ -14,7 +14,7 @@
     POST /api/ai/regex      body: {desc}              → {ok, regex, note}  （rule_type 默认为 regex）
     POST /api/regex/test    body: {regex, value}      → {ok, match, error} （rule_type 默认为 regex）
 
-「测试」端点统一走 Python（regex 走 re.compile、number/date 走 rule_runner._exec_user_function），
+「测试」端点统一走 Python（regex 走 re.compile、number/date 走 rule_runner.run_rule → compile_rule），
 保证和检测引擎语义一致。
 """
 
@@ -172,10 +172,12 @@ class GenRegexResponse(BaseModel):
 
 
 @router.post("/ai/regex", response_model=GenRegexResponse, summary="[兼容] AI 生成 Python 正则")
-async def ai_gen_regex(req: GenRegexRequest):
+def ai_gen_regex(req: GenRegexRequest):
     """老端点，转发到 /ai/rule（rule_type='regex'）。
 
     保留原因：兼容前端历史调用，避免已有客户端集成失效。
+    2026-09-24：async def → def —— gen_rule 是同步 LLM 调用（最长 30s×2），
+    挂事件循环上会冻结整个后端。
     """
     logger.info(f"POST /api/ai/regex  (legacy) desc={req.desc!r}")
     code, note = gen_rule(req.desc, "regex")
@@ -254,15 +256,21 @@ async def ai_explain_issue(req: ExplainIssueRequest):
         f"rule_type={req.rule_type!r} rule_id={req.rule_id} rule_name={req.rule_name!r}"
     )
     started = time.monotonic()
-    result = _explain_issue(
-        field_key=req.field_key,
-        field_comment=req.field_comment,
-        value=req.value,
-        rule_type=req.rule_type,
-        rule_name=req.rule_name,
-        rule_desc=req.rule_desc,
-        code=req.code,
-        regex=req.regex,
+    # 2026-09-24：_explain_issue 是同步 LLM 调用（可能 1~3 次 × 30s timeout），
+    # 直接调用会把整个后端挂住 —— 放线程池执行后再返回（与 /ai/rule 同款做法）。
+    loop = asyncio.get_running_loop()
+    result = await loop.run_in_executor(
+        None,
+        lambda: _explain_issue(
+            field_key=req.field_key,
+            field_comment=req.field_comment,
+            value=req.value,
+            rule_type=req.rule_type,
+            rule_name=req.rule_name,
+            rule_desc=req.rule_desc,
+            code=req.code,
+            regex=req.regex,
+        ),
     )
     elapsed_ms = (time.monotonic() - started) * 1000
     logger.info(

web/backend/routers/db.py

@@ -95,8 +95,11 @@ def _log_failure_hint(msg: str) -> None:
 
 
 # ── 端点 ─────────────────────────────────────────────────
+# 2026-09-24：四个端点从 async def 改为 def —— 内部是同步阻塞的 DB 建连/查询
+# （connect_timeout 最长 10s），async 挂在事件循环上会冻结整个后端；
+# def 端点 FastAPI 自动放进 starlette 线程池，互不阻塞。
 @router.post("/test", response_model=TestConnectionResponse, summary="测试数据库连接")
-async def connect_test(req: TestConnectionRequest):
+def connect_test(req: TestConnectionRequest):
     """复用 web.backend.core.db_adapter，验证 db_type/host/port/user/password/database
 
     返回 TestConnectionResponse(ok, message, db_type, database_name)
@@ -136,7 +139,7 @@ async def connect_test(req: TestConnectionRequest):
 
 
 @router.post("/schemas", summary="列出可用的 schema / database / user")
-async def list_schemas(req: TestConnectionRequest):
+def list_schemas(req: TestConnectionRequest):
     """根据 db_type 列出可连接的 schema 列表
 
     - MySQL  = SHOW DATABASES
@@ -189,7 +192,7 @@ async def list_schemas(req: TestConnectionRequest):
 
 
 @router.post("/tables", summary="列出指定 schema 下的所有表")
-async def list_tables(req: TestConnectionRequest):
+def list_tables(req: TestConnectionRequest):
     """复用 db_adapter.list_tables()，走 info_schema/list_tables.<dialect>.sql 模板"""
     cfg = _build_db_config(req)
     logger.info("─" * 60)
@@ -220,7 +223,7 @@ class ListColumnsRequest(TestConnectionRequest):
 
 
 @router.post("/columns", summary="列出指定表的全部字段（数据字典）")
-async def list_columns(req: ListColumnsRequest):
+def list_columns(req: ListColumnsRequest):
     """复用 db_adapter.list_columns()，走 info_schema/list_columns.<dialect>.sql 模板
 
     接收连接信息 + table_name，返回该表的所有列元数据，用于任务弹窗

web/backend/routers/tasks.py

@@ -323,23 +323,34 @@ def _copy_fields(db: Session, src_task: Task, dst_task: Task) -> None:
             ))
 
 
-def _row_to_out(db: Session, t: Task, *, include_field_list: bool = False) -> TaskOut:
+def _row_to_out(
+    db: Session,
+    t: Task,
+    *,
+    include_field_list: bool = False,
+    counts: Optional[dict] = None,
+) -> TaskOut:
     """t.to_dict() 给基础字段；counts / field_list 都复用当前 session 直接 query。
 
     include_field_list=True 时再查 field+rule 全量（详情用）；
     列表页不查（避免 N+1）。
+    counts：2026-09-24 批量计数通道 —— list_tasks 传 {task_id: (fields, rules)}
+    预聚合结果进来，跳过下面的逐任务计数查询（原来每个任务 3~4 条 SQL）。
     """
     d = t.to_dict()
     conn = _parse_conn_json(t.conn_json)
 
-    fields_count = (
-        db.query(Field).filter(Field.task_id == t.id).count() if t.id else 0
-    )
-    task_field_ids = [f.id for f in db.query(Field).filter(Field.task_id == t.id).all()]
-    rules_count = (
-        db.query(FieldRule).filter(FieldRule.field_id.in_(task_field_ids)).count()
-        + db.query(Rule).join(Field, Rule.field_id == Field.id).filter(Field.task_id == t.id).count()
-    ) if task_field_ids else 0
+    if counts is not None and t.id in counts:
+        fields_count, rules_count = counts[t.id]
+    else:
+        fields_count = (
+            db.query(Field).filter(Field.task_id == t.id).count() if t.id else 0
+        )
+        task_field_ids = [f.id for f in db.query(Field).filter(Field.task_id == t.id).all()]
+        rules_count = (
+            db.query(FieldRule).filter(FieldRule.field_id.in_(task_field_ids)).count()
+            + db.query(Rule).join(Field, Rule.field_id == Field.id).filter(Field.task_id == t.id).count()
+        ) if task_field_ids else 0
 
     field_list: list[FieldOut] = []
     if include_field_list:
@@ -426,7 +437,38 @@ def list_tasks(
         like = f"%{keyword}%"
         q = q.filter((Task.name.like(like)) | (Task.source_table.like(like)))
     rows = q.order_by(Task.id.desc()).all()
-    return [_row_to_out(db, t) for t in rows]
+
+    # 2026-09-24：批量计数 —— 原来逐任务 3~4 条 SQL（fields 数 + FieldRule 数 +
+    # 老 Rule 数），现在 3 条 group_by 聚合一次算完（任务多时列表接口明显提速）。
+    task_ids = [t.id for t in rows]
+    counts: dict = {}
+    if task_ids:
+        from sqlalchemy import func
+        f_cnt = dict(
+            db.query(Field.task_id, func.count(Field.id))
+            .filter(Field.task_id.in_(task_ids))
+            .group_by(Field.task_id)
+            .all()
+        )
+        fr_cnt = dict(
+            db.query(Field.task_id, func.count(FieldRule.id))
+            .join(Field, FieldRule.field_id == Field.id)
+            .filter(Field.task_id.in_(task_ids))
+            .group_by(Field.task_id)
+            .all()
+        )
+        r_cnt = dict(
+            db.query(Field.task_id, func.count(Rule.id))
+            .join(Field, Rule.field_id == Field.id)
+            .filter(Field.task_id.in_(task_ids))
+            .group_by(Field.task_id)
+            .all()
+        )
+        counts = {
+            tid: (f_cnt.get(tid, 0), fr_cnt.get(tid, 0) + r_cnt.get(tid, 0))
+            for tid in task_ids
+        }
+    return [_row_to_out(db, t, counts=counts) for t in rows]
 
 
 @router.post("", response_model=TaskOut, summary="新建任务")