数据剖析接口：POST /catalog/profile 同步剖析，未开启或同一数据源正在剖析时返回 409，返回每张表用的查询数、写入的发现和跳过的原因，记录署名

backend/app/api/catalog.py

@@ -5,9 +5,10 @@
 - PUT    /catalog/{table}        整份提交（乐观锁），改动过的项记为人工确认
 - POST   /catalog/{table}/review 单项审阅：确认、驳回、撤销审阅
 - POST   /catalog/draft          同步起草（注释、外键、命名推断，可选模型）
+- POST   /catalog/profile        同步剖析：对业务库发少量只读查询，核对关系、取码值候选（data/catalog_profile.py）
 
 规则都在 data/catalog.py，这里只做取源、转换形状和把异常翻成状态码：版本不符 409，结构不合规、审阅路径
-不对 422，源或表不存在 404。所有写操作记录请求头 X-Actor 的署名（自报、未认证）。
+不对 422，源或表不存在 404。所有写操作记录请求头 X-Actor 的署名（自报、未认证）。剖析没开启、正在剖析 409。
 
 上传表格的源按当前快照的表结构回答（table_versions.resolve_source），和查询实际用的库是同一个版本。
 """
@@ -20,7 +21,8 @@ from pydantic import BaseModel, Field
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.api.runs import actor_of
-from app.data import catalog, introspect, table_versions
+from app.data import catalog, catalog_profile, introspect, table_versions
+from app.data.engine import SnapshotTampered
 from app.db.base import get_session
 from app.db.models import DataSource
 
@@ -47,6 +49,11 @@ class CatalogDraftIn(BaseModel):
     use_model: bool = False
 
 
+class CatalogProfileIn(BaseModel):
+    #: 要剖析的表；不给或为空时按使用次数取前 10 张目录里有待核实关系的表
+    tables: list[str] | None = Field(default=None, max_length=200)
+
+
 async def _resolved(session: AsyncSession, source_id: str) -> tuple[DataSource, Any]:
     """(数据源记录, 用来读表结构的源)：上传源解析成绑定当前快照的视图。"""
     row = await session.get(DataSource, source_id)
@@ -209,6 +216,62 @@ async def draft_catalog(source_id: str, payload: CatalogDraftIn,
     }
 
 
+def _profile_table_out(t: catalog_profile.TableProfile) -> dict[str, Any]:
+    return {
+        "table_name": t.table,
+        "queries": t.queries,
+        "row_estimate": t.row_estimate.to_dict() if t.row_estimate else None,
+        "findings": t.findings,
+        "skipped": [s.to_dict() for s in t.skipped],
+        "date_ranges": t.date_ranges,
+        "added": t.added,
+        "updated": t.updated,
+        "removed": t.removed,
+        "version": t.version,
+        "error": t.error,
+    }
+
+
+@router.post("/{source_id}/catalog/profile")
+async def profile_catalog_tables(source_id: str, payload: CatalogProfileIn,
+                                 x_actor: str | None = Header(default=None),
+                                 session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
+    """同步剖析。返回每张表用了几条查询、写进目录的发现、跳过了什么及原因。
+
+    默认关闭：数据源设置里没开启数据剖析、数据源停用，409；同一个数据源已经在剖析，409。
+    """
+    row, source = await _resolved(session, source_id)
+    try:
+        catalog_profile.ensure_enabled(row)
+    except catalog_profile.ProfileDisabled as e:
+        raise HTTPException(409, str(e)) from e
+    cache = source.schema_cache or {}
+    if not cache.get("tables"):
+        raise HTTPException(409, f"{introspect.why_empty(cache)}，无法进行数据剖析。请先在数据源卡片上点「探查结构」")
+    actor = actor_of(x_actor)
+    try:
+        report = await catalog_profile.profile_catalog(session, row, source, tables=payload.tables or None,
+                                                       actor=actor)
+    except (catalog_profile.ProfileDisabled, catalog_profile.ProfileBusy) as e:
+        raise HTTPException(409, str(e)) from e
+    except SnapshotTampered as e:
+        raise HTTPException(409, str(e)) from e
+    rows = [_profile_table_out(t) for t in report.tables]
+    note = None
+    if not rows:
+        note = "目录里还没有待核实的关联关系，请先起草数据目录，或指定要剖析的表"
+    return {
+        "profiled_at": report.profiled_at,
+        "actor": actor,
+        "settings": report.settings.to_dict(),
+        "queries_used": report.queries_used,
+        "stopped": report.stopped,
+        "tables": rows,
+        "total": {k: sum(r[k] for r in rows) for k in ("added", "updated", "removed")},
+        "note": note,
+    }
+
+
 @router.post("/{source_id}/catalog/{table}/review")
 async def review_catalog_item(source_id: str, table: str, payload: CatalogReviewIn,
                               x_actor: str | None = Header(default=None),

backend/app/data/catalog_profile.py

@@ -354,7 +354,7 @@ class _Runner:
             # 上面已经按只读过了同一个守卫，查询层再拒只可能是时限到了
             if cut:
                 self.stopped = "deadline"
-            raise _Skip("timeout", f"查询超过 {_fmt(seconds)} 秒被中断，已跳过") from e
+            raise _Skip("timeout", f"查询超时：超过 {_fmt(seconds)} 秒被中断，已跳过") from e
         except SnapshotTampered:
             raise
         except Exception as e:  # noqa: BLE001 - 一条查询失败不该拖垮整次剖析，原因照实记下
@@ -677,8 +677,10 @@ def _relation_note(day: str, *, sample: int, matched: int, coverage: float, scan
         parts.append(f"被指向列是{how}" if how in ("主键", "唯一约束") else "被指向列经计数核对唯一")
     elif unique is False:
         parts.append("被指向列有重复值，不能作为关联的被指向键")
-    else:
+    elif how == "too_large":
         parts.append("被指向的表超过整表统计的行数上限，未核对被指向列是否唯一")
+    else:
+        parts.append("未能核对被指向列是否唯一")
     if child is not None:
         child_unique, child_how = child
         if child_unique is True:
@@ -686,8 +688,10 @@ def _relation_note(day: str, *, sample: int, matched: int, coverage: float, scan
             parts.append(f"子表一侧也唯一（{basis}），一对一")
         elif child_unique is False:
             parts.append("子表一侧有重复值，多对一")
-        else:
+        elif child_how == "too_large":
             parts.append("子表超过整表统计的行数上限，未核对子表一侧是否唯一，按多对一记")
+        else:
+            parts.append("未能核对子表一侧是否唯一，按多对一记")
     if not holds and not low:
         parts.append("暂不升为有确证")
     return f"{NOTE_PREFIX}{day}）：" + "；".join(parts) + "。"
@@ -696,8 +700,8 @@ def _relation_note(day: str, *, sample: int, matched: int, coverage: float, scan
 async def _check_relation(cx: _Context, table: str, rel: dict[str, Any], out: _TableOutcome,
                           result: TableProfile) -> None:
     """从子表取至多 sample_size 个不同的非空键值，去父表里数对上几个，得到覆盖率；再看父键、子键唯不唯一。"""
-    cols, to_table, to_cols = list(rel.get("columns") or []), str(rel.get("to_table") or ""), \
-        list(rel.get("to_columns") or [])
+    cols, to_cols = list(rel.get("columns") or []), list(rel.get("to_columns") or [])
+    to_table = str(rel.get("to_table") or "")
     if len(cols) != 1 or len(to_cols) != 1:
         raise _Skip("unsupported", "多列组成的关联关系暂不剖析")
     meta = cx.tables[table]
@@ -1165,12 +1169,9 @@ async def profile_catalog(session: AsyncSession, row: Any, source: Any, *, table
     - tables 为空时按使用次数取前 PROFILE_DEFAULT_TABLES 张目录里有待核实关系的表。
     - 每张表做完就写入，后面的表停下（预算、总时长）不影响前面已经写进去的。
     """
-    settings = settings or profile_settings(getattr(row, "options", None))
-    name = getattr(row, "name", "") or ""
-    if not settings.enabled:
-        raise ProfileDisabled(f"数据源「{name}」未开启数据剖析。剖析会对业务库发查询，请先在数据源设置中开启数据剖析")
+    settings = ensure_enabled(row, settings)
     if source.id in _RUNNING:
-        raise ProfileBusy(f"数据源「{name}」正在进行数据剖析，请等待完成后再试")
+        raise ProfileBusy(f"数据源「{getattr(row, 'name', '')}」正在进行数据剖析，请等待完成后再试")
     _RUNNING.add(source.id)
     try:
         return await _profile(session, row, source, tables, actor, settings)
@@ -1178,6 +1179,18 @@ async def profile_catalog(session: AsyncSession, row: Any, source: Any, *, table
         _RUNNING.discard(source.id)
 
 
+def ensure_enabled(row: Any, settings: ProfileSettings | None = None) -> ProfileSettings:
+    """这个数据源能不能剖析：数据源停用了、或者没开启数据剖析，抛 ProfileDisabled（message 给人看）。
+    返回要用的设置（不给就按 row.options 读）。接口在做别的检查之前先调它：开关是第一道门。"""
+    settings = settings or profile_settings(getattr(row, "options", None))
+    name = getattr(row, "name", "") or ""
+    if getattr(row, "enabled", True) is False:
+        raise ProfileDisabled(f"数据源「{name}」已停用，无法进行数据剖析。请先在数据源设置中启用")
+    if not settings.enabled:
+        raise ProfileDisabled(f"数据源「{name}」未开启数据剖析。剖析会对业务库发查询，请先在数据源设置中开启数据剖析")
+    return settings
+
+
 async def _profile(session: AsyncSession, row: Any, source: Any, tables: list[str] | None, actor: str | None,
                    settings: ProfileSettings) -> ProfileReport:
     at = catalog.now_iso()
@@ -1209,7 +1222,7 @@ async def _profile(session: AsyncSession, row: Any, source: Any, tables: list[st
     return report
 
 
-__all__ = ["CODES_MAX", "IN_CHUNK", "MAX_CODE_LEN", "MAX_LITERAL_LEN", "NOTE_PREFIX", "PROFILE_DEFAULT_TABLES", "PROFILE_OPTION", "ProfileBusy",
-           "ProfileDisabled", "ProfileReport", "ProfileSettings", "ProfileSkip", "SqlDialect", "TableProfile",
-           "TableSize", "estimate_size", "profile_catalog", "profile_settings", "profile_settings_problem",
-           "sql_literal"]
+__all__ = ["CODES_MAX", "IN_CHUNK", "MAX_CODE_LEN", "MAX_LITERAL_LEN", "NOTE_PREFIX", "PROFILE_DEFAULT_TABLES",
+           "PROFILE_OPTION", "ProfileBusy", "ProfileDisabled", "ProfileReport", "ProfileSettings", "ProfileSkip",
+           "SqlDialect", "TableProfile", "TableSize", "ensure_enabled", "estimate_size", "profile_catalog",
+           "profile_settings", "profile_settings_problem", "sql_literal"]

backend/tests/test_catalog_profile.py

@@ -0,0 +1,430 @@
+"""数据剖析接口：POST /api/datasources/{id}/catalog/profile（阶段 2A）。
+
+剖析是数据目录里唯一对业务库发查询的地方，这里逐条钉住方案定的约束：默认关闭、只读、每条查询都过守卫、
+查询次数和单条时限、总时长、遮罩的列不取样、大表不做整表统计、同一个数据源同时只有一次剖析；
+以及剖析的结论怎么进目录：命名推断对上数据的升为有确证（来源记为数据剖析，和外键分开）、对不上的写明
+覆盖率、人工确认过的只补覆盖率和基数、状态列得到码值候选、只有一个日期列的表提议业务日期。
+
+所有测试都在合成的景区库上做（tests/fixtures/catalog/scenic.py），数据和名称全部虚构。
+"""
+from __future__ import annotations
+
+import asyncio
+import sqlite3
+import uuid
+from pathlib import Path
+from types import SimpleNamespace
+from urllib.parse import quote
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+from sqlalchemy import event
+from sqlalchemy.engine import Engine
+
+from app.data import catalog, catalog_profile
+from app.data import engine as data_engine
+from app.data.catalog_profile import PROFILE_OPTION, ProfileSettings
+from app.data.engine import engines
+from app.db.base import SessionLocal
+from app.db.models import DataSource
+from app.main import app as _fastapi_app
+from tests.fixtures.catalog import scenic
+
+ACTOR = {"X-Actor": quote("周宁")}
+
+
+@pytest.fixture
+async def client():
+    async with AsyncClient(transport=ASGITransport(app=_fastapi_app), base_url="http://test") as c:
+        yield c
+
+
+def _on(**settings) -> dict:
+    return {PROFILE_OPTION: {"enabled": True, **settings}}
+
+
+@pytest.fixture
+async def make_source(client):
+    """登记一个接上 path 的数据源并走真实的探查接口；draft 给了就先起草这些表的目录。"""
+    made: list[str] = []
+
+    async def _make(path: str, *, options: dict | None = None, draft: list[str] | None = None) -> str:
+        async with SessionLocal() as session:
+            row = DataSource(name=f"prof_{uuid.uuid4().hex[:10]}", kind="sqlite", database=str(path),
+                             options=options or {})
+            session.add(row)
+            await session.commit()
+            sid = row.id
+        made.append(sid)
+        r = await client.post(f"/api/datasources/{sid}/introspect")
+        assert r.status_code == 200, r.text
+        if draft is not None:
+            r = await client.post(f"/api/datasources/{sid}/catalog/draft", json={"tables": draft})
+            assert r.status_code == 200, r.text
+        return sid
+
+    yield _make
+    for sid in made:
+        await engines.invalidate(sid)
+
+
+class _Spy:
+    """记下两样东西：守卫放行（guard.check 返回）的语句，和真正发到这个业务库的语句。
+
+    后者挂在 SQLAlchemy 的 before_cursor_execute 上，任何一条到达驱动的语句都逃不掉——包括绕开查询层
+    另开连接发的。断言「发到库的 ⊆ 守卫放行的」就是「所有查询都经过守卫」。
+    """
+
+    def __init__(self, db_name: str) -> None:
+        self.db_name = db_name
+        self.checked: list[str] = []
+        self.executed: list[str] = []
+
+    def clear(self) -> None:
+        self.checked.clear()
+        self.executed.clear()
+
+    def all_guarded(self) -> bool:
+        return bool(self.executed) and all(s in self.checked for s in self.executed)
+
+    def mentions(self, word: str) -> list[str]:
+        return [s for s in self.executed if word.lower() in s.lower()]
+
+
+@pytest.fixture
+def spy(monkeypatch, scenic_db):
+    holder = SimpleNamespace(spy=None)
+
+    def watch(path: str) -> _Spy:
+        holder.spy = _Spy(Path(path).name)
+        return holder.spy
+
+    real_check = data_engine.check
+
+    def checking(sql, **kw):
+        out = real_check(sql, **kw)
+        if holder.spy is not None:
+            holder.spy.checked.append(out)
+        return out
+
+    def on_execute(conn, cursor, statement, parameters, context, executemany):
+        if holder.spy is not None and holder.spy.db_name in str(conn.engine.url):
+            holder.spy.executed.append(statement)
+
+    monkeypatch.setattr(data_engine, "check", checking)
+    event.listen(Engine, "before_cursor_execute", on_execute)
+    yield watch
+    event.remove(Engine, "before_cursor_execute", on_execute)
+
+
+def _url(sid: str) -> str:
+    return f"/api/datasources/{sid}/catalog/profile"
+
+
+async def _notes(client, sid: str, table: str) -> dict:
+    r = await client.get(f"/api/datasources/{sid}/catalog/{table}")
+    assert r.status_code == 200, r.text
+    return r.json()
+
+
+def _relation(notes: dict, column: str) -> dict:
+    return next(r for r in notes.get("relations") or [] if r["columns"] == [column])
+
+
+def _table(body: dict, name: str) -> dict:
+    return next(t for t in body["tables"] if t["table_name"] == name)
+
+
+def _reasons(table: dict) -> set[str]:
+    return {s["reason"] for s in table["skipped"]}
+
+
+# ---------------------------------------------------------------- 开关与设置
+
+
+async def test_profile_is_off_by_default_and_says_where_to_turn_it_on(client, make_source, scenic_db, spy):
+    sid = await make_source(scenic_db, draft=["visits"])
+    watch = spy(scenic_db)
+    watch.clear()
+    r = await client.post(_url(sid), json={"tables": ["visits"]}, headers=ACTOR)
+    assert r.status_code == 409
+    detail = r.json()["detail"]
+    assert "数据源设置" in detail and "开启数据剖析" in detail
+    assert watch.executed == []          # 一条查询都没发
+
+
+async def test_settings_are_saved_and_validated_through_datasource_update(client, make_source, scenic_db):
+    sid = await make_source(scenic_db)
+    r = await client.patch(f"/api/datasources/{sid}", json={"options": _on(max_queries=0)})
+    assert r.status_code == 422 and "查询次数上限" in r.json()["detail"]
+    r = await client.patch(f"/api/datasources/{sid}", json={"options": _on(max_queries=40, sample_size=500)})
+    assert r.status_code == 200, r.text
+    assert r.json()["options"][PROFILE_OPTION] == {"enabled": True, "max_queries": 40, "sample_size": 500}
+    # 设好以后就能剖析；返回里带着这次实际用的设置
+    r = await client.post(_url(sid), json={"tables": ["channels"]}, headers=ACTOR)
+    assert r.status_code == 200, r.text
+    assert r.json()["settings"]["max_queries"] == 40 and r.json()["settings"]["sample_size"] == 500
+
+
+# ---------------------------------------------------------------- 剖析的结论
+
+
+async def test_profile_verifies_name_relations_and_proposes_codes_and_business_date(client, make_source, scenic_db,
+                                                                                    spy):
+    sid = await make_source(scenic_db, options=_on(), draft=["visits", "channels"])
+    watch = spy(scenic_db)
+    watch.clear()
+    r = await client.post(_url(sid), json={"tables": ["visits", "channels"]}, headers=ACTOR)
+    assert r.status_code == 200, r.text
+    body = r.json()
+
+    # 所有发到业务库的语句都经过了守卫
+    assert watch.all_guarded(), watch.executed
+    assert body["queries_used"] == len(watch.executed) <= body["settings"]["max_queries"]
+    assert body["queries_used"] == sum(t["queries"] for t in body["tables"])
+    assert body["stopped"] is None and body["actor"] == "周宁" and body["profiled_at"]
+
+    notes = (await _notes(client, sid, "visits"))
+    assert notes["updated_by"] == "周宁"
+    notes = notes["notes"]
+    # 命名推断出来的两条关系：数据对得上、父表主键唯一 → 有确证，来源记为数据剖析
+    for column, parent in (("gate_id", "gates"), ("member_id", "members")):
+        rel = _relation(notes, column)
+        assert rel["to_table"] == parent
+        assert (rel["source"], rel["status"]) == ("profile", "verified")
+        assert rel["coverage"] == 1.0 and rel["cardinality"] == "many_to_one"
+        assert "数据剖析" in rel["note"] and "抽样" in rel["note"]
+    # 外键约束来的关系不动：外键是外键，剖析是剖析
+    park = _relation(notes, "park_id")
+    assert (park["source"], park["status"], park["coverage"]) == ("fk", "verified", None)
+
+    # 状态列：码值候选，值有了、含义留空，推断
+    codes = notes["columns"]["status"]["codes"]
+    assert codes["value"] == {"1": "", "0": ""}
+    assert (codes["source"], codes["status"]) == ("profile", "proposed") and "1500" in codes["note"]
+    # 只有一个日期类列：提议为业务日期，备注里写明取值范围
+    bdate = notes["business_date"]
+    assert bdate["value"] == {"column": "visit_time"} and (bdate["source"], bdate["status"]) == ("profile", "proposed")
+    assert "2026-07" in bdate["note"] and "2026-09" in bdate["note"]
+
+    channels = (await _notes(client, sid, "channels"))["notes"]
+    assert set(channels["columns"]["channel_type"]["codes"]["value"]) == {"线上直销", "线下", "分销"}
+    assert "channel_code" not in channels.get("columns", {})       # 唯一约束列不是码值
+
+    visits = _table(body, "visits")
+    kinds = sorted(f["kind"] for f in visits["findings"])
+    assert kinds == ["business_date", "codes", "relation", "relation"]
+    gate = next(f for f in visits["findings"] if f["kind"] == "relation" and f["columns"] == ["gate_id"])
+    assert gate["status"] == "verified" and gate["path"] == f"relations.{_relation(notes, 'gate_id')['id']}"
+    assert gate["sample"] == 8 and gate["matched"] == 8 and gate["summary"]
+    assert visits["row_estimate"] == {"rows": 1500, "method": "count", "at_least": None}
+    assert visits["added"] == 2 and visits["updated"] == 2 and visits["error"] is None
+
+
+async def test_wrong_name_relation_stays_proposed_with_coverage_in_note(client, make_source, tmp_path):
+    """推错的命名关系：parking_records.lot_id 按名字对上了新建的 lots 表，但停车记录里的区号大多不在里面。"""
+    path = scenic.build(tmp_path / "lots.db")
+    db = sqlite3.connect(path)
+    try:
+        db.execute("CREATE TABLE lots (id INTEGER PRIMARY KEY, name TEXT NOT NULL)")
+        db.executemany("INSERT INTO lots VALUES (?, ?)", [(i, f"{i} 号停车区") for i in (1, 2, 3)])
+        db.executemany("INSERT INTO parking_records VALUES (?, ?, ?, ?, ?, ?)",
+                       [(i, 1 + i % 20, None, "2026-08-01 09:00:00", None, 10.0) for i in range(1, 101)])
+        db.commit()
+    finally:
+        db.close()
+    sid = await make_source(path, options=_on(), draft=["parking_records"])
+    before = _relation((await _notes(client, sid, "parking_records"))["notes"], "lot_id")
+    assert (before["to_table"], before["source"], before["status"]) == ("lots", "name", "proposed")
+
+    r = await client.post(_url(sid), json={"tables": ["parking_records"]}, headers=ACTOR)
+    assert r.status_code == 200, r.text
+    rel = _relation((await _notes(client, sid, "parking_records"))["notes"], "lot_id")
+    assert rel["status"] == "proposed" and rel["source"] == "profile"
+    assert rel["coverage"] == 0.15
+    assert "抽样覆盖率 15%" in rel["note"] and "可能不是这条关系" in rel["note"] and "20 个" in rel["note"]
+    finding = _table(r.json(), "parking_records")["findings"][0]
+    assert finding["status"] == "proposed" and finding["coverage"] == 0.15
+    # 两个日期类列：不提议业务日期，取值范围只在返回里
+    ranges = {d["column"] for d in _table(r.json(), "parking_records")["date_ranges"]}
+    assert ranges == {"entered_at"}           # exited_at 全是空值
+    assert "business_date" not in (await _notes(client, sid, "parking_records"))["notes"]
+
+
+async def test_confirmed_relation_only_gets_coverage_and_cardinality(client, make_source, scenic_db):
+    sid = await make_source(scenic_db, options=_on(), draft=["visits"])
+    detail = await _notes(client, sid, "visits")
+    gate = _relation(detail["notes"], "gate_id")
+    r = await client.post(f"/api/datasources/{sid}/catalog/visits/review",
+                          json={"path": f"relations.{gate['id']}", "action": "confirm",
+                                "if_version": detail["version"]})
+    assert r.status_code == 200, r.text
+
+    r = await client.post(_url(sid), json={"tables": ["visits"]}, headers=ACTOR)
+    assert r.status_code == 200, r.text
+    rel = _relation((await _notes(client, sid, "visits"))["notes"], "gate_id")
+    assert (rel["source"], rel["status"]) == ("name", "confirmed")
+    assert rel["coverage"] == 1.0 and rel["cardinality"] == "many_to_one"
+    finding = next(f for f in _table(r.json(), "visits")["findings"] if f.get("columns") == ["gate_id"])
+    assert finding["status"] == "confirmed" and finding["confirmed"] is True
+
+
+async def test_rerun_with_same_data_changes_nothing(client, make_source, scenic_db):
+    """同一天、数据没变：第二次剖析不改目录、不升版本（冻结进快照的目录不会无故变）。"""
+    sid = await make_source(scenic_db, options=_on(), draft=["visits"])
+    first = (await client.post(_url(sid), json={"tables": ["visits"]})).json()
+    second = (await client.post(_url(sid), json={"tables": ["visits"]})).json()
+    a, b = _table(first, "visits"), _table(second, "visits")
+    assert (b["added"], b["updated"], b["removed"]) == (0, 0, 0)
+    assert b["version"] == a["version"]
+
+
+# ---------------------------------------------------------------- 安全约束
+
+
+async def test_masked_columns_are_never_queried(client, make_source, scenic_db, spy):
+    sid = await make_source(scenic_db, options={**_on(), "mask_columns": "member_id, Channel_Type"},
+                            draft=["visits", "channels"])
+    watch = spy(scenic_db)
+    watch.clear()
+    r = await client.post(_url(sid), json={"tables": ["visits", "channels"]}, headers=ACTOR)
+    assert r.status_code == 200, r.text
+    assert watch.all_guarded()
+    assert watch.mentions("member_id") == [] and watch.mentions("channel_type") == []
+    visits, channels = _table(r.json(), "visits"), _table(r.json(), "channels")
+    masked = [s for s in visits["skipped"] + channels["skipped"] if s["reason"] == "masked"]
+    assert {s["target"] for s in masked} == {"member_id → members.id", "channel_type"}
+    assert all("遮罩" in s["detail"] for s in masked)
+    notes = (await _notes(client, sid, "visits"))["notes"]
+    assert _relation(notes, "member_id")["source"] == "name"                    # 没动
+    assert _relation(notes, "gate_id")["status"] == "verified"                  # 其余照常
+    assert "channel_type" not in ((await _notes(client, sid, "channels"))["notes"].get("columns") or {})
+
+
+async def test_query_budget_stops_profiling_and_reports_the_rest(client, make_source, scenic_db, spy):
+    sid = await make_source(scenic_db, options=_on(max_queries=3), draft=["visits"])
+    watch = spy(scenic_db)
+    watch.clear()
+    r = await client.post(_url(sid), json={"tables": ["visits", "channels"]}, headers=ACTOR)
+    assert r.status_code == 200, r.text
+    body = r.json()
+    assert body["queries_used"] == 3 == len(watch.executed) and body["stopped"] == "budget"
+    visits, channels = _table(body, "visits"), _table(body, "channels")
+    assert "budget" in _reasons(visits)
+    assert channels["queries"] == 0 and _reasons(channels) == {"budget"}
+    assert all("查询次数" in s["detail"] for s in visits["skipped"] + channels["skipped"] if s["reason"] == "budget")
+    # 没查完的关系不改：原来的命名推断原样留着
+    rel = _relation((await _notes(client, sid, "visits"))["notes"], "member_id")
+    assert (rel["source"], rel["status"]) == ("name", "proposed")
+
+
+async def test_single_query_timeout_is_recorded_and_profiling_continues(client, make_source, scenic_db, monkeypatch):
+    """闸口的抽样查询换成一条停不下来的递归查询：1 秒时限到点由数据库掐断，记为超时，其余照常。"""
+    sid = await make_source(scenic_db, options=_on(query_timeout_s=1), draft=["visits"])
+    real = data_engine.run_query
+    endless = "WITH RECURSIVE r(x) AS (SELECT 1 UNION ALL SELECT x + 1 FROM r) SELECT COUNT(*) FROM r"
+
+    async def slow_gate(source, sql, *, limits=None):
+        if "gate_id" in sql and "DISTINCT" in sql and "COUNT" not in sql:
+            assert limits is not None and limits.timeout_seconds == 1
+            return await real(source, endless, limits=limits)
+        return await real(source, sql, limits=limits)
+
+    monkeypatch.setattr(data_engine, "run_query", slow_gate)
+    r = await client.post(_url(sid), json={"tables": ["visits"]}, headers=ACTOR)
+    assert r.status_code == 200, r.text
+    visits = _table(r.json(), "visits")
+    timeout = [s for s in visits["skipped"] if s["reason"] == "timeout"]
+    assert [s["target"] for s in timeout] == ["gate_id → gates.id"] and "超时" in timeout[0]["detail"]
+    assert r.json()["stopped"] is None
+    notes = (await _notes(client, sid, "visits"))["notes"]
+    assert _relation(notes, "gate_id")["source"] == "name"                       # 超时的不改
+    assert _relation(notes, "member_id")["status"] == "verified"                 # 其余照常
+
+
+async def test_total_deadline_stops_profiling(make_source, scenic_db, monkeypatch):
+    sid = await make_source(scenic_db, draft=["visits"])
+    real = data_engine.run_query
+
+    async def slow(source, sql, *, limits=None):
+        await asyncio.sleep(0.5)
+        return await real(source, sql, limits=limits)
+
+    monkeypatch.setattr(data_engine, "run_query", slow)
+    async with SessionLocal() as session:
+        row = await session.get(DataSource, sid)
+        settings = ProfileSettings(enabled=True, max_total_s=1.4)
+        report = await catalog_profile.profile_catalog(session, row, row, tables=["visits"], actor=None,
+                                                       settings=settings)
+    assert report.stopped == "deadline"
+    assert 1 <= report.queries_used <= 3
+    reasons = {s.reason for t in report.tables for s in t.skipped}
+    assert "deadline" in reasons
+
+
+async def test_large_tables_get_no_full_scans(client, make_source, scenic_db, spy):
+    """行数超过整表统计上限（这里设成 100 行）：不做 COUNT(DISTINCT)、MIN/MAX，抽样和取值分布都带上限。"""
+    sid = await make_source(scenic_db, options=_on(max_scan_rows=100), draft=["visits"])
+    watch = spy(scenic_db)
+    watch.clear()
+    r = await client.post(_url(sid), json={"tables": ["visits"]}, headers=ACTOR)
+    assert r.status_code == 200, r.text
+    visits = _table(r.json(), "visits")
+    assert visits["row_estimate"] == {"rows": None, "method": "count", "at_least": 101}
+    on_visits = [s for s in watch.executed if "from visits" in s.lower()]
+    assert on_visits and all("LIMIT" in s for s in on_visits), on_visits
+    assert not any("MIN(" in s for s in on_visits)
+    assert "too_large" in _reasons(visits)
+    assert any("太大" in s["detail"] or "上限" in s["detail"] for s in visits["skipped"] if s["reason"] == "too_large")
+    # 父表有主键，覆盖率又够：照样有确证；子表一侧没核对唯一，按多对一记
+    rel = _relation((await _notes(client, sid, "visits"))["notes"], "gate_id")
+    assert rel["status"] == "verified" and rel["cardinality"] == "many_to_one"
+
+
+# ---------------------------------------------------------------- 接口
+
+
+async def test_concurrent_profile_on_same_source_returns_409(client, make_source, scenic_db, monkeypatch):
+    sid = await make_source(scenic_db, options=_on(), draft=["visits"])
+    real = data_engine.run_query
+    started, release = asyncio.Event(), asyncio.Event()
+
+    async def held(source, sql, *, limits=None):
+        started.set()
+        await release.wait()
+        return await real(source, sql, limits=limits)
+
+    monkeypatch.setattr(data_engine, "run_query", held)
+    first = asyncio.create_task(client.post(_url(sid), json={"tables": ["visits"]}, headers=ACTOR))
+    await asyncio.wait_for(started.wait(), 10)
+    second = await client.post(_url(sid), json={"tables": ["visits"]}, headers=ACTOR)
+    assert second.status_code == 409 and "正在进行数据剖析" in second.json()["detail"]
+    release.set()
+    done = await asyncio.wait_for(first, 30)
+    assert done.status_code == 200, done.text
+    # 做完以后可以再来
+    again = await client.post(_url(sid), json={"tables": ["channels"]}, headers=ACTOR)
+    assert again.status_code == 200, again.text
+
+
+async def test_default_tables_are_used_tables_with_relation_candidates(client, make_source, scenic_db):
+    sid = await make_source(scenic_db, options=_on(), draft=[])
+    r = await client.post(_url(sid), json={}, headers=ACTOR)
+    assert r.status_code == 200, r.text
+    names = [t["table_name"] for t in r.json()["tables"]]
+    assert names and len(names) <= catalog_profile.PROFILE_DEFAULT_TABLES
+    async with SessionLocal() as session:
+        entries = await catalog.read_catalog(session, sid)
+    for name in names:
+        rels = entries[name].notes.get("relations") or []
+        assert any(rel["source"] not in ("fk",) for rel in rels), name
+    assert "parks" not in names and "channels" not in names          # 没有待核实关系的表不选
+
+
+async def test_unknown_table_and_missing_source(client, make_source, scenic_db):
+    sid = await make_source(scenic_db, options=_on(), draft=["visits"])
+    r = await client.post(_url(sid), json={"tables": ["nope"]}, headers=ACTOR)
+    assert r.status_code == 200
+    assert _table(r.json(), "nope")["error"] and r.json()["queries_used"] == 0
+    r = await client.post(_url("0" * 32), json={}, headers=ACTOR)
+    assert r.status_code == 404