mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 02:41:58 +00:00
数据目录 F1-A1:对话提案确认关联关系时不再抹掉基数和覆盖率——覆盖率是剖析测得的,一律沿用;基数没写时沿用已有的,新增的按目标表的主键或唯一约束推算,推算不了留空并在提案这一项上说明(PatchChange.note)。以前剖析测得多对一的关系经提案确认后基数变空,一对多关联后重复计算的检查静默失效
This commit is contained in:
+62
-11
@@ -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, *,
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user