数据剖析设置：按数据源存在 options.catalog_profile 里，默认关闭；保存数据源时校验取值范围，拼连接串时摘掉

backend/app/api/datasources.py

@@ -23,6 +23,7 @@ from app.core.errors import AUTH, NETWORK, TIMEOUT, classify, explain, raw
 from app.core.config import settings
 from app.core.crypto import encrypt, mask
 from app.data import catalog as data_catalog
+from app.data.catalog_profile import profile_settings_problem
 from app.data import introspect as introspect_mod
 from app.data import table_versions
 from app.data.engine import (
@@ -336,7 +337,8 @@ _MASK_FIELD = {
 
 
 def _refuse_bad_options(options: dict[str, Any] | None) -> None:
-    problem = query_timeout_problem(options) or mask_columns_problem(options)
+    problem = (query_timeout_problem(options) or mask_columns_problem(options)
+               or profile_settings_problem(options))
     if problem:
         raise HTTPException(422, problem)
 

backend/app/data/catalog_profile.py

@@ -0,0 +1,119 @@
+"""数据剖析：对业务库发少量只读查询，验证数据目录里推断出来的关联关系，补上基数、覆盖率和码值候选。
+
+**为什么要有它。** 命名推断只看列名，可能推错，也给不出基数和覆盖率；而只有 verified、confirmed 的关系才会
+触发错误级的 SQL 检查。剖析拿数据说话：从子表抽一批不同的键值去父表里对，对得上、父键唯一的升为 verified
+（来源记为 profile，和外键约束的 fk 分开），对不上的保持 proposed 并在备注里写明覆盖率。顺带给状态类的列
+取一次取值分布（码值候选，含义留给人填），给只有一个日期列的表提议业务日期。
+
+**这是数据目录里唯一会对业务库发查询的地方**，约束都是硬的：
+
+- 默认关闭，按数据源在 options.catalog_profile 里开启（profile_settings）；
+- 每条查询先过守卫的只读判定（不论数据源本身可写与否），再交给查询层 run_query：同一套连接、守卫和
+  时限，不另开连接；
+- 每条查询有时限（不超过数据源自己的查询时限），每次剖析有查询次数上限和总时长上限；
+- 表有多大先看数据库的统计信息，没有统计信息就数到上限为止，不做全表 COUNT；超过上限的表不做
+  COUNT(DISTINCT)、MIN/MAX 这类整表统计，抽样和取值分布也只看前若干行；
+- 数据源设置里遮罩的列不取样，也不出现在任何一条剖析查询里；
+- 只取低基数列的取值分布（最多 21 个取值），不取明细行。
+"""
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass
+from typing import Any
+
+from app.data.engine import CATALOG_PROFILE_OPTION
+
+#: 数据源 options 里剖析设置的键
+PROFILE_OPTION = CATALOG_PROFILE_OPTION
+
+
+# ==========================================================================
+# 设置
+# ==========================================================================
+
+
+@dataclass(frozen=True)
+class ProfileSettings:
+    """一个数据源的剖析设置。缺省即「关闭」：剖析会对业务库发查询，必须有人按源明确打开。"""
+
+    enabled: bool = False
+    #: 一次剖析最多发几条查询（读统计信息、抽样、核对都算）。用完就停，剩下的记为「查询次数用完」
+    max_queries: int = 60
+    #: 单条查询的时限（秒）。实际取它和数据源自己的查询时限中较小的一个
+    query_timeout_s: float = 10
+    #: 每条关系从子表抽多少个不同的键值去父表里对
+    sample_size: int = 2000
+    #: 行数不超过它的表才做 COUNT(DISTINCT)、MIN/MAX 这类整表统计；更大的、或者不知道多大的表只看前若干行。
+    #: 填 0 表示一律不做整表统计
+    max_scan_rows: int = 100_000
+    #: 一次剖析的总时长上限（秒）。接口是同步的，不能让一个请求无限地挂着
+    max_total_s: float = 120
+
+    def to_dict(self) -> dict[str, Any]:
+        return asdict(self)
+
+
+#: 数值项：(中文叫法, 下限, 上限, 是否必须是整数)。叫法和将来设置界面上的标签一致，报错里只写它
+_NUMBER_FIELDS: dict[str, tuple[str, float, float, bool]] = {
+    "max_queries": ("查询次数上限", 1, 500, True),
+    "query_timeout_s": ("单条查询时限（秒）", 1, 60, False),
+    "sample_size": ("抽样键值数", 10, 10_000, True),
+    "max_scan_rows": ("整表统计行数上限", 0, 10_000_000, True),
+    "max_total_s": ("总时长上限（秒）", 10, 600, False),
+}
+_ENABLED_LABEL = "开启数据剖析"
+_KNOWN_KEYS = frozenset({"enabled", *_NUMBER_FIELDS})
+
+
+def _fmt(value: float) -> str:
+    return str(int(value)) if float(value).is_integer() else f"{value:g}"
+
+
+def _number(value: Any, low: float, high: float, integer: bool) -> int | float | None:
+    """设置里的一个数：数字和数字写成的文字都认（表单存的常是文字）。不合规返回 None。"""
+    if value is None or isinstance(value, bool):
+        return None
+    try:
+        number = float(str(value).strip())
+    except ValueError:
+        return None
+    if number != number or not low <= number <= high:       # NaN 也在这里挡掉
+        return None
+    if number.is_integer():
+        return int(number)
+    return None if integer else number
+
+
+def profile_settings(options: dict[str, Any] | None) -> ProfileSettings:
+    """options → 剖析设置。读的时候宽松：库里存着的某一项不对（老数据、手改过），那一项用缺省，不报错；
+    开关只认 true，写成别的一律当作没开。"""
+    raw = (options or {}).get(PROFILE_OPTION)
+    if not isinstance(raw, dict):
+        return ProfileSettings()
+    values: dict[str, Any] = {"enabled": raw.get("enabled") is True}
+    for key, (_, low, high, integer) in _NUMBER_FIELDS.items():
+        number = _number(raw.get(key), low, high, integer)
+        if number is not None:
+            values[key] = number
+    return ProfileSettings(**values)
+
+
+def profile_settings_problem(options: dict[str, Any] | None) -> str | None:
+    """保存数据源时查一下剖析设置；没填、或者填对了返回 None。写的时候严格：不认识的项、超出范围的值都拒。"""
+    raw = (options or {}).get(PROFILE_OPTION)
+    if raw is None:
+        return None
+    if not isinstance(raw, dict):
+        return "数据剖析设置的格式不正确，请在数据源设置中重新填写"
+    if unknown := sorted(set(raw) - _KNOWN_KEYS):
+        return f"数据剖析设置中有无法识别的项「{'、'.join(map(str, unknown))}」，请删除后重试"
+    if "enabled" in raw and not isinstance(raw["enabled"], bool):
+        return f"数据剖析的「{_ENABLED_LABEL}」只能是开启或关闭"
+    for key, (label, low, high, integer) in _NUMBER_FIELDS.items():
+        if key in raw and _number(raw[key], low, high, integer) is None:
+            kind = "整数" if integer else "数"
+            return f"数据剖析的「{label}」需要填写 {_fmt(low)} 到 {_fmt(high)} 之间的{kind}；当前为「{raw[key]}」"
+    return None
+
+
+__all__ = ["PROFILE_OPTION", "ProfileSettings", "profile_settings", "profile_settings_problem"]

backend/app/data/engine.py

@@ -58,11 +58,16 @@ MAX_QUERY_TIMEOUT_S = 600
 #: 完整快照仍能按工件 id 取到，SQL 里给列起个别名也能绕开
 MASK_COLUMNS_OPTION = "mask_columns"
 
+#: 数据源 options 里的数据剖析设置（一个对象：开关、查询次数上限、单条时限……）。取值和校验见
+#: data/catalog_profile.py；放在这里只为让拼连接串时摘掉它
+CATALOG_PROFILE_OPTION = "catalog_profile"
+
 # options 里这些 key 是 AgentLab 自己的配置，不是驱动参数，拼 URL 时要摘掉。
 # schema：探查哪个 schema（企业库里只读账号名下常常什么都没有，数据在别处）
 # query_timeout_s：查询时限，由数据层按语句下发给数据库，见 _server_deadline
 # mask_columns：证据面板遮罩的列，驱动不认识它，拼进连接串会被当成未知参数拒掉
-_NON_DRIVER_OPTIONS = frozenset({"schema", QUERY_TIMEOUT_OPTION, MASK_COLUMNS_OPTION})
+# catalog_profile：数据剖析设置，同上；它还是个对象，拼进去连 URL 都不成形
+_NON_DRIVER_OPTIONS = frozenset({"schema", QUERY_TIMEOUT_OPTION, MASK_COLUMNS_OPTION, CATALOG_PROFILE_OPTION})
 
 _MASK_SPLIT = re.compile(r"[,，、;；\n]+")
 

backend/tests/test_catalog_profile_units.py

@@ -0,0 +1,59 @@
+"""数据剖析（阶段 2A）里不连库的部分：设置的解析与校验、SQL 字面量、各方言的剖析 SQL 都过得了守卫、
+目录对剖析结果的容纳（待填含义的码值、撤销审阅回到剖析的结论）。"""
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+import pytest
+
+from app.data.catalog_profile import PROFILE_OPTION, ProfileSettings, profile_settings, profile_settings_problem
+from app.data.engine import build_url
+
+# ---------------------------------------------------------------- 设置
+
+
+def test_settings_default_off_and_parse_leniently():
+    s = profile_settings({})
+    assert s == ProfileSettings()
+    assert s.enabled is False and s.max_queries == 60 and s.query_timeout_s == 10 and s.sample_size == 2000
+    assert s.max_scan_rows == 100_000 and s.max_total_s == 120
+    on = profile_settings({PROFILE_OPTION: {"enabled": True, "max_queries": "30", "sample_size": 500}})
+    assert on.enabled is True and on.max_queries == 30 and on.sample_size == 500
+    # 库里存着的值不对（老数据、手改过）：那一项用缺省，不因此开启，也不报错
+    odd = profile_settings({PROFILE_OPTION: {"enabled": "yes", "max_queries": 0, "query_timeout_s": "x"}})
+    assert odd.enabled is False and odd.max_queries == 60 and odd.query_timeout_s == 10
+    assert profile_settings({PROFILE_OPTION: "on"}) == ProfileSettings()
+
+
+@pytest.mark.parametrize("value, label", [
+    ({"max_queries": 0}, "查询次数上限"),
+    ({"max_queries": 501}, "查询次数上限"),
+    ({"max_queries": 2.5}, "查询次数上限"),
+    ({"query_timeout_s": 61}, "单条查询时限"),
+    ({"sample_size": 5}, "抽样键值数"),
+    ({"max_scan_rows": -1}, "整表统计行数上限"),
+    ({"max_total_s": 5}, "总时长上限"),
+    ({"enabled": "true"}, "开启数据剖析"),
+    ({"max_queries": True}, "查询次数上限"),
+])
+def test_settings_problem_names_the_field_in_chinese(value, label):
+    problem = profile_settings_problem({PROFILE_OPTION: value})
+    assert problem and label in problem and "数据剖析" in problem
+
+
+def test_settings_problem_rejects_unknown_keys_and_wrong_shape():
+    assert "无法识别" in profile_settings_problem({PROFILE_OPTION: {"enabled": True, "rows": 1}})
+    assert profile_settings_problem({PROFILE_OPTION: ["enabled"]})
+    assert profile_settings_problem({}) is None
+    assert profile_settings_problem({PROFILE_OPTION: None}) is None
+    assert profile_settings_problem({PROFILE_OPTION: {"enabled": True, "max_queries": 500, "query_timeout_s": 1.5,
+                                                      "sample_size": 10, "max_scan_rows": 0,
+                                                      "max_total_s": 600}}) is None
+
+
+def test_profile_option_never_reaches_the_connection_url():
+    """剖析设置是 AgentLab 自己的配置：拼进连接串，驱动会把它当未知参数拒掉。"""
+    source = SimpleNamespace(kind="mysql", host="db.example", port=3306, database="shop", username="ro",
+                             password=None, options={"charset": "utf8mb4", PROFILE_OPTION: {"enabled": True}})
+    url = build_url(source)
+    assert "charset=utf8mb4" in url and PROFILE_OPTION not in url and "enabled" not in url