From 0fa498ac18751e93660e393a01f9be84b1401417 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Sun, 4 Oct 2026 01:03:32 +0800 Subject: [PATCH] =?UTF-8?q?=E6=95=B0=E6=8D=AE=E7=9B=AE=E5=BD=95=20F1-A1?= =?UTF-8?q?=EF=BC=9A=E5=AF=B9=E8=AF=9D=E6=8F=90=E6=A1=88=E7=A1=AE=E8=AE=A4?= =?UTF-8?q?=E5=85=B3=E8=81=94=E5=85=B3=E7=B3=BB=E6=97=B6=E4=B8=8D=E5=86=8D?= =?UTF-8?q?=E6=8A=B9=E6=8E=89=E5=9F=BA=E6=95=B0=E5=92=8C=E8=A6=86=E7=9B=96?= =?UTF-8?q?=E7=8E=87=E2=80=94=E2=80=94=E8=A6=86=E7=9B=96=E7=8E=87=E6=98=AF?= =?UTF-8?q?=E5=89=96=E6=9E=90=E6=B5=8B=E5=BE=97=E7=9A=84=EF=BC=8C=E4=B8=80?= =?UTF-8?q?=E5=BE=8B=E6=B2=BF=E7=94=A8=EF=BC=9B=E5=9F=BA=E6=95=B0=E6=B2=A1?= =?UTF-8?q?=E5=86=99=E6=97=B6=E6=B2=BF=E7=94=A8=E5=B7=B2=E6=9C=89=E7=9A=84?= =?UTF-8?q?=EF=BC=8C=E6=96=B0=E5=A2=9E=E7=9A=84=E6=8C=89=E7=9B=AE=E6=A0=87?= =?UTF-8?q?=E8=A1=A8=E7=9A=84=E4=B8=BB=E9=94=AE=E6=88=96=E5=94=AF=E4=B8=80?= =?UTF-8?q?=E7=BA=A6=E6=9D=9F=E6=8E=A8=E7=AE=97=EF=BC=8C=E6=8E=A8=E7=AE=97?= =?UTF-8?q?=E4=B8=8D=E4=BA=86=E7=95=99=E7=A9=BA=E5=B9=B6=E5=9C=A8=E6=8F=90?= =?UTF-8?q?=E6=A1=88=E8=BF=99=E4=B8=80=E9=A1=B9=E4=B8=8A=E8=AF=B4=E6=98=8E?= =?UTF-8?q?=EF=BC=88PatchChange.note=EF=BC=89=E3=80=82=E4=BB=A5=E5=89=8D?= =?UTF-8?q?=E5=89=96=E6=9E=90=E6=B5=8B=E5=BE=97=E5=A4=9A=E5=AF=B9=E4=B8=80?= =?UTF-8?q?=E7=9A=84=E5=85=B3=E7=B3=BB=E7=BB=8F=E6=8F=90=E6=A1=88=E7=A1=AE?= =?UTF-8?q?=E8=AE=A4=E5=90=8E=E5=9F=BA=E6=95=B0=E5=8F=98=E7=A9=BA=EF=BC=8C?= =?UTF-8?q?=E4=B8=80=E5=AF=B9=E5=A4=9A=E5=85=B3=E8=81=94=E5=90=8E=E9=87=8D?= =?UTF-8?q?=E5=A4=8D=E8=AE=A1=E7=AE=97=E7=9A=84=E6=A3=80=E6=9F=A5=E9=9D=99?= =?UTF-8?q?=E9=BB=98=E5=A4=B1=E6=95=88?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/data/catalog.py | 73 ++++++++++-- backend/tests/test_catalog_semantics.py | 151 ++++++++++++++++++++++++ 2 files changed, 213 insertions(+), 11 deletions(-) create mode 100644 backend/tests/test_catalog_semantics.py diff --git a/backend/app/data/catalog.py b/backend/app/data/catalog.py index 8b9b9ad6..10db4596 100644 --- a/backend/app/data/catalog.py +++ b/backend/app/data/catalog.py @@ -746,6 +746,8 @@ class PatchChange: - after：保存后的值（码值是补充后的完整对照）；value：要交回保存的原样取值（码值只有补充的那几个， 409 之后在最新的目录上重新补）； - state：change 值有变化；confirm 值相同、但还不是已确认（保存即确认）；same 已经是这个值且已确认。 + - note：服务端对这一项的说明，给人看；没有时为 None，as_dict 里也不出现。目前只有一种：新增的关联关系没写 + 基数、又推算不出来（_relation_after）。 """ path: str @@ -755,10 +757,14 @@ class PatchChange: value: Any reason: str state: str + note: str | None = None def as_dict(self) -> dict[str, Any]: - return {"path": self.path, "before": self.before, "before_status": self.before_status, "after": self.after, - "value": self.value, "reason": self.reason, "state": self.state} + out = {"path": self.path, "before": self.before, "before_status": self.before_status, "after": self.after, + "value": self.value, "reason": self.reason, "state": self.state} + if self.note: + out["note"] = self.note + return out @dataclass(frozen=True) @@ -785,7 +791,11 @@ def _formula_like(key: Slot, value: Any) -> bool: def _patch_relation(table: str, raw: Any, path_id: str | None, tables: Mapping[str, Any] | None, columns: set[str] | None) -> tuple[dict[str, Any] | None, str | None]: - """关联关系的取值 → (关系的值部分, 问题)。值写两端（columns、to_table、to_columns），基数可选。""" + """关联关系的取值 → (关系的值部分, 问题)。值写两端（columns、to_table、to_columns），基数可选。 + + 返回的是提案原样的取值：基数没写就是 None，覆盖率一律 None（模型写了也不认）。保存后的值由 _relation_after + 并上目录里已有的测量结果再算。 + """ if not isinstance(raw, Mapping): return None, "关联关系的值应写明本表字段、目标表和目标字段" to_table = raw.get("to_table") @@ -809,6 +819,35 @@ def _patch_relation(table: str, raw: Any, path_id: str | None, tables: Mapping[s return value, None +def _relation_after(table: str, value: Mapping[str, Any], prev: Mapping[str, Any] | None, + tables: Mapping[str, Any] | None) -> tuple[dict[str, Any], str | None]: + """提案里一条关系保存后的值：(值, 给人看的说明)。 + + 以前直接拿提案的取值整份替换：覆盖率写成空、基数取提案原值（模型多半不写）。数据剖析测得多对一、覆盖率 100% + 的关系，经对话「确认一下」就丢了基数，一对多关联后重复计算的检查跟着失效。所以： + - 覆盖率是数据剖析测出来的，一律沿用目录里这条关系的（包括被驳回的那条留下的），没有就是空； + - 基数：提案写了用提案的；没写时，目录里已有这条关系就沿用它的；新增的按目标表的主键或唯一约束推算 + （目标列是键，本表这几列也是键就是一对一，否则多对一，同外键约束的 _cardinality）；推算不了就留空， + 并说明为什么空着。 + """ + out = dict(value) + out["coverage"] = prev.get("coverage") if prev is not None else None + if out.get("cardinality") is not None: + return out, None + if prev is not None: + out["cardinality"] = prev.get("cardinality") + return out, None + meta = tables.get(table) if tables is not None else None + target = tables.get(str(out.get("to_table"))) if tables is not None else None + if not isinstance(meta, Mapping) or not isinstance(target, Mapping): + return out, "没有写明基数，表结构里也查不到两端的主键和唯一约束，基数暂时空着；一对多关联后重复计算的检查用不上这条关系" + if not _is_key(target, list(out.get("to_columns") or [])): + return out, (f"没有写明基数，目标字段不是 {out.get('to_table')} 的主键或唯一约束，推算不出，基数暂时空着；" + "一对多关联后重复计算的检查用不上这条关系，可以在数据目录里补上基数") + out["cardinality"] = _cardinality(meta, list(out.get("columns") or [])) + return out, None + + def _merge_codes(before: Mapping[str, str], value: Mapping[str, str]) -> dict[str, str]: """码值的补充：给出的码加进去或改掉原来的含义。含义为空表示「只知道有这个码、含义待填写」（数据剖析的码值 候选就是这样），所以提案里含义为空的码不覆盖已有的含义，只在原来没有这个码时加进去。""" @@ -833,6 +872,8 @@ def plan_patch(notes: Mapping[str, Any] | None, changes: Any, *, table: str, - 码值是补充：在现有（没被驳回的）码值上加或改给出的码，不整份替换——「status=9 表示作废」不该把已确认的 其余码值删掉。含义为空的码（数据剖析写进来的「含义待填写」）可以被补上含义；提案里含义为空的码只在原来 没有这个码时加进去，不会把已有的含义抹掉（_merge_codes）。 + - 关联关系不抹掉测量结果：覆盖率沿用目录里的，基数没写时沿用或推算（_relation_after）。两端、基数、覆盖率都和 + 目录里一样时只是确认，来源不变。 - 看起来像计算公式的文字不收（_FORMULA）：公式进口径卡。 - 同一个路径出现两次只认第一次。 """ @@ -904,8 +945,10 @@ def plan_patch(notes: Mapping[str, Any] | None, changes: Any, *, table: str, prev = slots.get(key) live = prev if prev is not None and prev.get("status") != "rejected" else None before = _value_of(key, live) if live is not None else None - after = value - if key[0] == "c" and key[2] == "codes" and isinstance(before, dict): + after, note = value, None + if key[0] == "r": + after, note = _relation_after(table, value, prev, tables) + elif key[0] == "c" and key[2] == "codes" and isinstance(before, dict): after = _merge_codes(before, value) if before is not None and before == after: state = "same" if live.get("status") == "confirmed" else "confirm" @@ -913,7 +956,7 @@ def plan_patch(notes: Mapping[str, Any] | None, changes: Any, *, table: str, state = "change" out.append(PatchChange(path=norm, before=copy.deepcopy(before), before_status=live.get("status") if live is not None else None, - after=after, value=value, reason=reason, state=state)) + after=after, value=value, reason=reason, state=state, note=note)) return PatchPlan(out, problems) @@ -938,7 +981,10 @@ def apply_patch(notes: Mapping[str, Any] | None, changes: Any, *, table: str, slots[key] = {**prev, "status": "confirmed", "updated_at": at} continue if key[0] == "r": - slots[key] = {"id": key[1], **change.after, "source": "human", "status": "confirmed", "updated_at": at} + rel = {"id": key[1], **change.after, "source": "human", "status": "confirmed", "updated_at": at} + if prev is not None and prev.get("note"): + rel["note"] = prev["note"] # 覆盖率沿用了，说明它怎么测的那段备注也留着 + slots[key] = rel else: slots[key] = {"value": change.after, "source": "human", "status": "confirmed", "updated_at": at} out = _assemble(slots) @@ -1006,11 +1052,16 @@ def _comment_is_label(text: str) -> bool: return len(text) <= _LABEL_MAX and not _SENTENCE_MARKS.search(text) -def _cardinality(meta: Mapping[str, Any], columns: list[str]) -> str: - """本表这几列指向别的表时的基数：这几列恰好是本表的主键或某个唯一约束，就是一对一，否则多对一。""" - wanted = {c.lower() for c in columns} +def _is_key(meta: Mapping[str, Any], columns: list[str]) -> bool: + """这几列恰好是这张表的主键或某个唯一约束（不分大小写、不计顺序）。""" + wanted = {str(c).lower() for c in columns} keys = [meta.get("primary_key") or []] + [u for u in meta.get("unique") or [] if isinstance(u, list)] - return "one_to_one" if any(wanted == {c.lower() for c in k} for k in keys if k) else "many_to_one" + return bool(wanted) and any(wanted == {str(c).lower() for c in k} for k in keys if k) + + +def _cardinality(meta: Mapping[str, Any], columns: list[str]) -> str: + """本表这几列指向别的表（的键）时的基数：这几列恰好是本表的主键或某个唯一约束，就是一对一，否则多对一。""" + return "one_to_one" if _is_key(meta, columns) else "many_to_one" def _relation(table: str, columns: list[str], to_table: str, to_columns: list[str], source: str, *, diff --git a/backend/tests/test_catalog_semantics.py b/backend/tests/test_catalog_semantics.py new file mode 100644 index 00000000..4e5dc86f --- /dev/null +++ b/backend/tests/test_catalog_semantics.py @@ -0,0 +1,151 @@ +"""数据目录核心语义的回归（2026-10 完整性审查 F1）。 + +- A1：经对话提案确认一条关系，不抹掉数据剖析测得的基数和覆盖率；新增关系没写基数时按目标表的键推算， + 推算不了就留空并在提案的这一项上说明。否则扇出检查（fanout_sum）看不到基数，静默失效。 +- A2：在目录页去掉推断或外键来的关系即驳回，保留编号；只有人工新建的关系才真正删除。检查器、关系图 + （助手挑表也用它）、重新起草三处都不能再出现被驳回的关系。 +- A4：码值只补了一部分时不当作完整清单。「码值不在码值表中」只对标了「已列全」的码值报；「已列全」只有人工 + 明确确认、或者剖析没截断地看完全表取值时才有。 +- B2：隔天重新剖析、数据没变，目录不升版本（剖析写的备注里只有日期变了不算变化）。 +- B3：计算公式在人工整份提交、模型起草两条路径上也拦下（以前只拦提案）。 +- 一致性：日期列的判定、表名的匹配、拆词各只有一份。 + +数据和名称全部虚构（tests/fixtures/catalog/scenic.py）。 +""" +from __future__ import annotations + +import sqlite3 + +import pytest + +from app.data import catalog, catalog_profile, sqlcheck +from app.db.base import SessionLocal +from app.db.models import DataSource +from tests.fixtures.catalog import scenic, scenic_notes +from tests.fixtures.catalog.fake_model import FakeModel +from tests.test_catalog_patch import _schema +from tests.test_catalog_profile import _notes, _on, _relation, _table, _url, client, make_source # noqa: F401 + +AT = "2026-10-04T08:00:00+00:00" +DAY1, DAY2, DAY3 = "2026-10-01T08:00:00+00:00", "2026-10-02T08:00:00+00:00", "2026-10-03T08:00:00+00:00" + + +async def _checker(sid: str) -> sqlcheck.SqlChecker | None: + async with SessionLocal() as session: + row = await session.get(DataSource, sid) + return (await sqlcheck.load_checkers(session, [row])).get(row.name) + + +async def _found(sid: str, sql: str) -> list[tuple]: + checker = await _checker(sid) + return [(c.code, c.level, c.table, c.column) for c in (checker.check(sql) if checker else [])] + + +async def _graph(sid: str) -> dict[str, list[catalog.JoinEdge]]: + """助手挑表用的关系图：库里的目录原样（含驳回项）加表结构现推。""" + async with SessionLocal() as session: + row = await session.get(DataSource, sid) + entries = await catalog.read_catalog(session, sid) + return catalog.relation_graph({t: e.notes for t, e in entries.items()}, schema_cache=row.schema_cache) + + +async def _put(client, sid: str, table: str, notes: dict, version: int): + return await client.put(f"/api/datasources/{sid}/catalog/{table}", json={"notes": notes, "if_version": version}) + + +def _profile_rel(table: str, columns: list[str], to_table: str, to_columns: list[str], *, + cardinality: str | None = "many_to_one", coverage: float | None = 1.0, + status: str = "verified") -> dict: + return {"id": catalog.relation_id(table, columns, to_table, to_columns), "columns": columns, + "to_table": to_table, "to_columns": to_columns, "cardinality": cardinality, "coverage": coverage, + "source": "profile", "status": status, + "note": "数据剖析（2026-10-01）：子表抽样 8 个不同键值，父表对上 8 个，覆盖率 100%；被指向列是主键。"} + + +# ========================================================================== +# A1：提案不抹掉基数和覆盖率 +# ========================================================================== + + +async def test_patch_confirming_a_profiled_relation_keeps_cardinality_and_coverage(scenic_db): + tables = (await _schema(scenic_db))["tables"] + gate = _profile_rel("visits", ["gate_id"], "gates", ["id"]) + notes = {"relations": [gate]} + value = {"columns": ["gate_id"], "to_table": "gates", "to_columns": ["id"]} + for path in ("relations.new", f"relations.{gate['id']}"): + [change] = catalog.plan_patch(notes, [{"path": path, "value": value}], table="visits", tables=tables).changes + assert change.after["cardinality"] == "many_to_one" and change.after["coverage"] == 1.0 + # 两端、基数、覆盖率都和目录里的一样：保存即确认，来源不变 + assert change.state == "confirm" + out = catalog.apply_patch(notes, [{"path": "relations.new", "value": value}], table="visits", tables=tables, + at=AT) + [rel] = out["relations"] + assert (rel["source"], rel["status"], rel["cardinality"], rel["coverage"]) == ( + "profile", "confirmed", "many_to_one", 1.0) + assert rel["note"] == gate["note"] + + +async def test_patch_may_change_cardinality_but_never_coverage(scenic_db): + tables = (await _schema(scenic_db))["tables"] + gate = _profile_rel("visits", ["gate_id"], "gates", ["id"], coverage=0.97) + value = {"columns": ["gate_id"], "to_table": "gates", "to_columns": ["id"], "cardinality": "one_to_one", + "coverage": 0.2} # 覆盖率是测量值：模型写了也不认 + [change] = catalog.plan_patch({"relations": [gate]}, [{"path": "relations.new", "value": value}], + table="visits", tables=tables).changes + assert change.state == "change" + assert (change.after["cardinality"], change.after["coverage"]) == ("one_to_one", 0.97) + out = catalog.apply_patch({"relations": [gate]}, [{"path": "relations.new", "value": value}], table="visits", + tables=tables, at=AT) + [rel] = out["relations"] + assert (rel["source"], rel["status"], rel["cardinality"], rel["coverage"]) == ( + "human", "confirmed", "one_to_one", 0.97) + assert rel["note"] == gate["note"] # 覆盖率留着，说明它的剖析备注也留着 + + +async def test_new_relation_without_cardinality_is_inferred_from_target_keys(scenic_db): + tables = (await _schema(scenic_db))["tables"] + + def plan(table: str, value: dict) -> catalog.PatchChange: + [change] = catalog.plan_patch({}, [{"path": "relations.new", "value": value}], table=table, + tables=tables).changes + return change + + # 目标是主键、本表这一列不唯一：多对一 + gate = plan("visits", {"columns": ["gate_id"], "to_table": "gates", "to_columns": ["id"]}) + assert (gate.after["cardinality"], gate.after["coverage"], gate.note) == ("many_to_one", None, None) + assert "note" not in gate.as_dict() + # 目标是唯一约束、本表这一列也唯一（票号有唯一索引）：一对一 + ticket = plan("visits", {"columns": ["ticket_no"], "to_table": "orders", "to_columns": ["order_no"]}) + assert ticket.after["cardinality"] == "one_to_one" + # 目标列不是主键也不是唯一约束：推算不了，基数留空，这一项上说明 + park = plan("visits", {"columns": ["park_id"], "to_table": "gates", "to_columns": ["park_id"]}) + assert park.after["cardinality"] is None + assert park.note and "基数" in park.note and park.as_dict()["note"] == park.note + # 不知道表结构（尚未探查）：不推算，也说明 + [blind] = catalog.plan_patch({}, [{"path": "relations.new", "value": { + "columns": ["gate_id"], "to_table": "gates", "to_columns": ["id"]}}], table="visits", tables=None).changes + assert blind.after["cardinality"] is None and blind.note + + +async def test_confirming_a_profiled_relation_by_patch_keeps_the_fanout_check(client, make_source, scenic_db): + """审查探针 A1：剖析得出多对一、覆盖率 1.0，经对话提案确认这条关系之后，扇出检查照样报 error。""" + sql = "SELECT COUNT(g.id) AS n FROM gates g JOIN visits v ON v.gate_id = g.id" + sid = await make_source(scenic_db, options=_on(), draft=["visits", "gates"]) + r = await client.post(_url(sid), json={"tables": ["visits"]}) + assert r.status_code == 200, r.text + detail = await _notes(client, sid, "visits") + rel = _relation(detail["notes"], "gate_id") + assert (rel["source"], rel["status"], rel["cardinality"], rel["coverage"]) == ( + "profile", "verified", "many_to_one", 1.0) + assert ("fanout_sum", "error", "gates", "id") in await _found(sid, sql) + + changes = [{"path": "relations.new", "value": {"columns": ["gate_id"], "to_table": "gates", "to_columns": ["id"]}, + "reason": "用户说 gate_id 对应 gates.id"}] + r = await client.post(f"/api/datasources/{sid}/catalog/visits/patch", + json={"changes": changes, "if_version": detail["version"]}) + assert r.status_code == 200, r.text + after = _relation(r.json()["notes"], "gate_id") + assert (after["status"], after["cardinality"], after["coverage"]) == ("confirmed", "many_to_one", 1.0) + assert ("fanout_sum", "error", "gates", "id") in await _found(sid, sql) + +