mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 02:31:58 +00:00
数据目录阶段 1A：catalog.py 补齐合并与审阅、注释 / 外键 / 命名推断 / 模型起草、渲染、关系图、使用次数和冻结内容
This commit is contained in:
ThearchyHelios
+1316
-8
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,379 @@
|
||||
"""数据目录的起草：数据库注释、外键约束、命名推断、模型起草，以及并入已有目录。
|
||||
|
||||
这一阶段起草不发数据库查询（数据剖析归阶段 2），只看探查缓存里的结构。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import uuid
|
||||
from types import SimpleNamespace
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from app.data import catalog
|
||||
from app.data.catalog import make_item
|
||||
from app.db.base import SessionLocal
|
||||
from app.db.models import DataSource
|
||||
|
||||
|
||||
def _col(name: str, type_: str = "INTEGER", comment: str | None = None, nullable: bool = True) -> dict:
|
||||
return {"name": name, "type": type_, "nullable": nullable, "comment": comment}
|
||||
|
||||
|
||||
def _table(*cols: dict, pk=("id",), fks=None, unique=None, comment=None) -> dict:
|
||||
meta = {"qualified": "", "schema": None, "columns": list(cols), "primary_key": list(pk), "comment": comment,
|
||||
"is_view": False}
|
||||
if fks is not None:
|
||||
meta["foreign_keys"] = fks
|
||||
if unique is not None:
|
||||
meta["unique"] = unique
|
||||
return meta
|
||||
|
||||
|
||||
def _source(tables: dict, **kw) -> SimpleNamespace:
|
||||
for name, meta in tables.items():
|
||||
meta["qualified"] = name
|
||||
return SimpleNamespace(id=kw.pop("id", "src"), name="scenic", kind="sqlite", origin=kw.pop("origin", "manual"),
|
||||
schema_cache={"tables": tables, **kw})
|
||||
|
||||
|
||||
def _rels(notes: dict) -> set[tuple]:
|
||||
return {(tuple(r["columns"]), r["to_table"], tuple(r["to_columns"]), r["source"], r["status"])
|
||||
for r in notes.get("relations", [])}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 数据库注释
|
||||
|
||||
|
||||
def test_comments_become_labels_or_meanings():
|
||||
src = _source({"visits": _table(
|
||||
_col("id"), _col("amount", "REAL", comment="实收金额"),
|
||||
_col("status", comment="状态：1 表示有效，0 表示作废；作废记录不计入客流"),
|
||||
comment="入园记录")})
|
||||
draft = catalog.draft_structure(src, "visits")
|
||||
assert draft.notes["label"] == make_item("入园记录", "comment")
|
||||
assert draft.notes["columns"]["amount"]["label"] == make_item("实收金额", "comment")
|
||||
assert draft.notes["columns"]["status"]["meaning"]["value"].startswith("状态：1 表示有效")
|
||||
assert "label" not in draft.notes["columns"]["status"]
|
||||
assert "comment" in draft.covered
|
||||
|
||||
|
||||
def test_long_table_comment_goes_to_description():
|
||||
src = _source({"visits": _table(_col("id"), comment="游客每次检票入园记一行，团队票按人数拆成多行")})
|
||||
draft = catalog.draft_structure(src, "visits")
|
||||
assert "label" not in draft.notes
|
||||
assert draft.notes["description"]["value"].startswith("游客每次检票入园")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("kw", [{"origin": "upload"}, {"import_mode": "recipe"}])
|
||||
def test_imported_table_comments_are_not_copied(kw):
|
||||
"""导入表格的说明是按核对结果生成的（recipe_notes），数据目录不重复它们。"""
|
||||
src = _source({"日报": _table(_col("日期", "TEXT", comment="格式 YYYY-MM-DD"), _col("客流", comment="单位：人次"),
|
||||
pk=(), comment="粒度：每个日期")}, **kw)
|
||||
assert catalog.system_notes_source(src)
|
||||
draft = catalog.draft_structure(src, "日报")
|
||||
assert "label" not in draft.notes and "description" not in draft.notes
|
||||
assert "columns" not in draft.notes
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 外键约束
|
||||
|
||||
|
||||
def test_foreign_keys_become_verified_relations():
|
||||
src = _source({
|
||||
"visits": _table(_col("id"), _col("park_id"), fks=[
|
||||
{"columns": ["park_id"], "to_table": "parks", "to_columns": ["id"]}]),
|
||||
"parks": _table(_col("id"), fks=[]),
|
||||
})
|
||||
draft = catalog.draft_structure(src, "visits")
|
||||
assert _rels(draft.notes) == {(("park_id",), "parks", ("id",), "fk", "verified")}
|
||||
rel = draft.notes["relations"][0]
|
||||
assert rel["cardinality"] == "many_to_one"
|
||||
assert rel["id"] == catalog.relation_id("visits", ["park_id"], "parks", ["id"])
|
||||
assert "fk" in draft.covered
|
||||
|
||||
|
||||
def test_foreign_key_without_target_columns_points_at_primary_key_and_one_to_one():
|
||||
src = _source({
|
||||
"member_profiles": _table(_col("member_id"), _col("nickname", "TEXT"), pk=("member_id",),
|
||||
fks=[{"columns": ["member_id"], "to_table": "members", "to_columns": []}]),
|
||||
"members": _table(_col("id"), fks=[]),
|
||||
})
|
||||
rel = catalog.draft_structure(src, "member_profiles").notes["relations"][0]
|
||||
assert rel["to_columns"] == ["id"] and rel["cardinality"] == "one_to_one"
|
||||
|
||||
|
||||
def test_old_cache_without_foreign_keys_does_not_cover_fk():
|
||||
"""旧缓存里没有 foreign_keys：不知道有没有外键，不能据此删掉已有的外键关系。"""
|
||||
src = _source({"visits": _table(_col("id"), _col("park_id")), "parks": _table(_col("id"))})
|
||||
draft = catalog.draft_structure(src, "visits")
|
||||
assert "fk" not in draft.covered
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 命名推断
|
||||
|
||||
|
||||
def _infer(tables: dict, table: str) -> set[tuple]:
|
||||
src = _source(tables)
|
||||
return {(tuple(r["columns"]), r["to_table"], tuple(r["to_columns"]))
|
||||
for r in catalog.infer_name_relations(src.schema_cache, table)}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("column, target", [
|
||||
("channel_id", "channels"), # 复数
|
||||
("category_id", "categories"), # y → ies
|
||||
("storeId", "stores"), # 驼峰
|
||||
("storeID", "stores"), # xxxID
|
||||
("ticket_type_id", "ticket_types"), # 多段下划线
|
||||
("memberTagId", "memberTags"), # 驼峰表名
|
||||
("member_tag_id", "memberTags"), # 下划线列名对驼峰表名
|
||||
("park_id", "park"), # 单数表名
|
||||
("CHANNEL_ID", "CHANNELS"), # 大写（Oracle）
|
||||
])
|
||||
def test_name_inference_positive(column, target):
|
||||
tables = {"facts": _table(_col("id"), _col(column)), target: _table(_col("id"))}
|
||||
pk = "id"
|
||||
assert _infer(tables, "facts") == {((column,), target, (pk,))}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("tables, why", [
|
||||
({"employees": _table(_col("id"), _col("employee_id", "TEXT"))}, "自引用"),
|
||||
({"facts": _table(_col("id"), _col("guide_id")), "guide": _table(_col("id")), "guides": _table(_col("id"))},
|
||||
"多个候选"),
|
||||
({"facts": _table(_col("id"), _col("lot_id")), "parking_lots": _table(_col("id"))}, "没有对应的表"),
|
||||
({"facts": _table(_col("id"), _col("store_id", "VARCHAR(20)")), "stores": _table(_col("id"))}, "类型对不上"),
|
||||
({"facts": _table(_col("id"), _col("day_id")), "days": _table(_col("park_id"), _col("d"), pk=("park_id", "d"))},
|
||||
"被指向的表是复合主键"),
|
||||
({"facts": _table(_col("id"), _col("tag_id")), "tags": _table(_col("name", "TEXT"), pk=())}, "被指向的表没有主键"),
|
||||
({"facts": _table(_col("id"), _col("x_id")), "xs": _table(_col("id"))}, "词根太短"),
|
||||
({"facts": _table(_col("id"), _col("paid")), "pas": _table(_col("id"))}, "不是 _id 结尾"),
|
||||
({"facts": _table(_col("id"), _col("store_id"), fks=[{"columns": ["store_id"], "to_table": "shops",
|
||||
"to_columns": ["id"]}]),
|
||||
"stores": _table(_col("id")), "shops": _table(_col("id"))}, "已有外键约束的列"),
|
||||
])
|
||||
def test_name_inference_negative(tables, why):
|
||||
assert _infer(tables, next(iter(tables))) == set(), why
|
||||
|
||||
|
||||
def test_name_inference_one_to_one_when_column_is_own_key():
|
||||
tables = {"member_profiles": _table(_col("member_id"), pk=("member_id",)), "members": _table(_col("id"))}
|
||||
rel = catalog.infer_name_relations(_source(tables).schema_cache, "member_profiles")[0]
|
||||
assert rel["cardinality"] == "one_to_one" and rel["source"] == "name" and rel["status"] == "proposed"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 模型起草
|
||||
|
||||
|
||||
class FakeModel:
|
||||
"""假模型：结构化输出返回 replies[表名] 拼成的 {"tables": [...]}；fail 里的表所在的那一批抛异常。"""
|
||||
|
||||
def __init__(self, replies: dict[str, dict], fail: set[str] = frozenset()) -> None:
|
||||
self.replies, self.fail, self.calls = replies, set(fail), []
|
||||
|
||||
def with_structured_output(self, schema: Any, **_: Any) -> "FakeModel":
|
||||
assert schema["title"]
|
||||
return self
|
||||
|
||||
async def ainvoke(self, messages: list[Any]) -> dict:
|
||||
text = "\n".join(str(getattr(m, "content", m)) for m in messages)
|
||||
self.calls.append(text)
|
||||
names = [t for t in self.replies if f"表 {t}" in text]
|
||||
if any(t in self.fail for t in names):
|
||||
raise RuntimeError("网关超时")
|
||||
return {"tables": [{"table": t, **self.replies[t]} for t in names]}
|
||||
|
||||
|
||||
_VISITS_REPLY = {
|
||||
"label": "入园记录", "grain": "每张门票每次检票一行", "kind": "fact", "keys": ["ticket_no"],
|
||||
"business_date": {"column": "visit_time", "rule": "按检票时间计", "timezone": "+08:00"},
|
||||
"columns": [
|
||||
{"name": "visitor_count", "label": "入园人数", "meaning": "这次检票进园的人数", "unit": "人", "measure": "flow"},
|
||||
{"name": "status", "label": "状态", "measure": "status"},
|
||||
{"name": "ghost", "label": "不存在的列"}, # 丢掉
|
||||
{"name": "ticket_no", "measure": "money"}, # 度量类型不在取值里：丢掉这一项
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def _scenic_tables() -> dict:
|
||||
return {
|
||||
"visits": _table(_col("id"), _col("ticket_no", "TEXT"), _col("visit_time", "TEXT"), _col("visitor_count"),
|
||||
_col("status"), fks=[]),
|
||||
"channels": _table(_col("id"), _col("name", "TEXT"), fks=[]),
|
||||
}
|
||||
|
||||
|
||||
async def test_model_draft_maps_fields_and_drops_invalid_values():
|
||||
src = _source(_scenic_tables())
|
||||
model = FakeModel({"visits": _VISITS_REPLY, "channels": {"label": "渠道", "kind": "planet",
|
||||
"keys": ["nope"]}})
|
||||
got = await catalog.draft_with_model(model, src, ["visits", "channels"])
|
||||
visits = got["visits"].notes
|
||||
assert visits["label"] == make_item("入园记录", "llm")
|
||||
assert visits["grain"]["value"] == "每张门票每次检票一行" and visits["kind"]["value"] == "fact"
|
||||
assert visits["keys"]["value"] == ["ticket_no"]
|
||||
assert visits["business_date"]["value"] == {"column": "visit_time", "rule": "按检票时间计", "timezone": "+08:00"}
|
||||
assert visits["columns"]["visitor_count"]["unit"] == make_item("人", "llm")
|
||||
assert visits["columns"]["status"]["measure"]["value"] == "status"
|
||||
assert "ghost" not in visits["columns"] and "ticket_no" not in visits["columns"]
|
||||
channels = got["channels"].notes
|
||||
assert channels == {"label": make_item("渠道", "llm")} # 表类型、主键不合规的都丢掉
|
||||
assert got["visits"].covered == frozenset({"llm"})
|
||||
assert catalog.validate_notes(visits) == []
|
||||
|
||||
|
||||
async def test_model_prompt_carries_structure_and_no_data():
|
||||
src = _source(_scenic_tables())
|
||||
model = FakeModel({"visits": _VISITS_REPLY})
|
||||
await catalog.draft_with_model(model, src, ["visits"])
|
||||
prompt = model.calls[0]
|
||||
assert "表 visits" in prompt and "ticket_no" in prompt and "visit_time" in prompt
|
||||
|
||||
|
||||
async def test_model_failure_only_affects_its_batch():
|
||||
tables = {f"t{i:02d}": _table(_col("id"), *[_col(f"c{j}") for j in range(60)]) for i in range(4)}
|
||||
src = _source(tables)
|
||||
model = FakeModel({name: {"label": f"表{name}"} for name in tables}, fail={"t00"})
|
||||
got = await catalog.draft_with_model(model, src, list(tables))
|
||||
assert isinstance(got["t00"], str) and "网关超时" in got["t00"]
|
||||
assert all(got[n].notes["label"]["value"] == f"表{n}" for n in ("t01", "t02", "t03"))
|
||||
assert len(model.calls) >= 2 # 按表分批，不是一次全发
|
||||
|
||||
|
||||
async def test_model_draft_respects_system_notes_of_imported_tables():
|
||||
"""导入表格：表说明已经写了粒度、列说明写了单位，模型起草的粒度和单位不进目录，免得重复或矛盾。"""
|
||||
src = _source({"日报": _table(_col("日期", "TEXT", comment="格式 YYYY-MM-DD"), _col("客流"), pk=(),
|
||||
comment="粒度：每个日期")}, origin="upload")
|
||||
model = FakeModel({"日报": {"label": "客流日报", "grain": "每天一行", "columns": [
|
||||
{"name": "日期", "meaning": "统计日期", "unit": "天", "measure": "attribute"},
|
||||
{"name": "客流", "meaning": "当天入园人次", "unit": "人次", "measure": "flow"}]}})
|
||||
got = (await catalog.draft_with_model(model, src, ["日报"]))["日报"].notes
|
||||
assert "grain" not in got and got["label"]["value"] == "客流日报"
|
||||
assert "meaning" not in got["columns"]["日期"] and "unit" not in got["columns"]["日期"]
|
||||
assert got["columns"]["日期"]["measure"]["value"] == "attribute"
|
||||
assert got["columns"]["客流"]["unit"]["value"] == "人次" # 这一列没有系统说明，照常起草
|
||||
assert "粒度：每个日期" in model.calls[0] # 系统说明交给模型作背景
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 起草并写入
|
||||
|
||||
|
||||
async def _db_source(tables: dict, **kw) -> DataSource:
|
||||
for name, meta in tables.items():
|
||||
meta["qualified"] = name
|
||||
async with SessionLocal() as session:
|
||||
row = DataSource(name=f"cat_{uuid.uuid4().hex[:10]}", kind="sqlite", database=":memory:",
|
||||
schema_cache={"tables": tables}, **kw)
|
||||
session.add(row)
|
||||
await session.commit()
|
||||
return row
|
||||
|
||||
|
||||
async def test_draft_catalog_writes_and_is_idempotent():
|
||||
tables = _scenic_tables()
|
||||
tables["visits"]["foreign_keys"] = [{"columns": ["channel_id"], "to_table": "channels", "to_columns": ["id"]}]
|
||||
tables["visits"]["columns"].append(_col("channel_id"))
|
||||
tables["visits"]["columns"].append(_col("gate_id"))
|
||||
tables["gates"] = _table(_col("id"), fks=[])
|
||||
row = await _db_source(tables)
|
||||
model = FakeModel({"visits": _VISITS_REPLY})
|
||||
async with SessionLocal() as session:
|
||||
report = await catalog.draft_catalog(session, row, tables=["visits"], use_model=True, model=model,
|
||||
actor="王敏")
|
||||
result = report.tables[0]
|
||||
assert result.table == "visits" and result.error is None and result.version == 1
|
||||
# 表级 5 项（中文名、粒度、表类型、业务主键、业务日期）+ 列级 6 项 + 外键和命名推断各一条关系
|
||||
assert (result.added, result.updated, result.removed) == (13, 0, 0)
|
||||
assert report.model_used and report.model_error is None
|
||||
async with SessionLocal() as session:
|
||||
entry = await catalog.read_entry(session, row.id, "visits")
|
||||
assert entry.updated_by == "王敏"
|
||||
assert _rels(entry.notes) == {(("channel_id",), "channels", ("id",), "fk", "verified"),
|
||||
(("gate_id",), "gates", ("id",), "name", "proposed")}
|
||||
again = await catalog.draft_catalog(session, row, tables=["visits"], use_model=True, model=model, actor=None)
|
||||
assert (again.tables[0].added, again.tables[0].updated, again.tables[0].removed) == (0, 0, 0)
|
||||
assert again.tables[0].version == 1 # 没有变化不升版本
|
||||
|
||||
|
||||
async def test_redraft_keeps_confirmed_and_rejected_items():
|
||||
tables = _scenic_tables()
|
||||
tables["visits"]["columns"].append(_col("gate_id"))
|
||||
tables["gates"] = _table(_col("id"), fks=[])
|
||||
row = await _db_source(tables)
|
||||
model = FakeModel({"visits": _VISITS_REPLY})
|
||||
async with SessionLocal() as session:
|
||||
await catalog.draft_catalog(session, row, tables=["visits"], use_model=True, model=model, actor=None)
|
||||
entry = await catalog.read_entry(session, row.id, "visits")
|
||||
rid = entry.notes["relations"][0]["id"]
|
||||
entry = await catalog.review_entry(session, row.id, "visits", "grain", "confirm", if_version=entry.version,
|
||||
actor="李雷")
|
||||
entry = await catalog.review_entry(session, row.id, "visits", f"relations.{rid}", "reject",
|
||||
if_version=entry.version, actor="李雷")
|
||||
changed = FakeModel({"visits": {**_VISITS_REPLY, "grain": "每次入园一行", "label": "游客入园"}})
|
||||
report = await catalog.draft_catalog(session, row, tables=["visits"], use_model=True, model=changed,
|
||||
actor=None)
|
||||
after = await catalog.read_entry(session, row.id, "visits")
|
||||
assert after.notes["grain"]["value"] == "每张门票每次检票一行" and after.notes["grain"]["status"] == "confirmed"
|
||||
assert after.notes["relations"][0]["status"] == "rejected" # 驳回的关系留着，下次起草不再提出
|
||||
assert after.notes["label"]["value"] == "游客入园" # 同一来源、没确认过的项照常更新
|
||||
assert report.tables[0].updated == 1
|
||||
|
||||
|
||||
async def test_draft_catalog_model_unavailable_still_writes_other_sources(monkeypatch):
|
||||
tables = _scenic_tables()
|
||||
tables["visits"]["comment"] = "入园记录"
|
||||
row = await _db_source(tables)
|
||||
|
||||
async def unavailable(session):
|
||||
raise catalog.CatalogModelUnavailable("未配置模型接入，无法起草。请到「设置 → 模型接入」添加")
|
||||
|
||||
monkeypatch.setattr(catalog, "resolve_draft_model", unavailable)
|
||||
async with SessionLocal() as session:
|
||||
report = await catalog.draft_catalog(session, row, tables=["visits"], use_model=True, actor=None)
|
||||
entry = await catalog.read_entry(session, row.id, "visits")
|
||||
assert not report.model_used and "模型接入" in report.model_error
|
||||
label = entry.notes["label"]
|
||||
assert label["updated_at"] and {k: v for k, v in label.items() if k != "updated_at"} == make_item("入园记录",
|
||||
"comment")
|
||||
|
||||
|
||||
async def test_draft_catalog_model_failure_on_one_table(monkeypatch):
|
||||
row = await _db_source(_scenic_tables())
|
||||
model = FakeModel({"visits": _VISITS_REPLY, "channels": {"label": "渠道"}}, fail={"visits"})
|
||||
monkeypatch.setattr(catalog, "MODEL_BATCH_TABLES", 1)
|
||||
async with SessionLocal() as session:
|
||||
report = await catalog.draft_catalog(session, row, tables=["visits", "channels"], use_model=True,
|
||||
model=model, actor=None)
|
||||
by = {r.table: r for r in report.tables}
|
||||
assert by["visits"].model_error and "网关超时" in by["visits"].model_error
|
||||
assert by["channels"].model_error is None and by["channels"].added == 1
|
||||
|
||||
|
||||
async def test_draft_catalog_defaults_to_most_used_tables(monkeypatch):
|
||||
tables = {f"t{i:02d}": _table(_col("id"), fks=[]) for i in range(25)}
|
||||
row = await _db_source(tables)
|
||||
|
||||
async def usage(session, source):
|
||||
return {"t24": 9, "t03": 5, "t10": 5}
|
||||
|
||||
monkeypatch.setattr(catalog, "table_usage", usage)
|
||||
async with SessionLocal() as session:
|
||||
report = await catalog.draft_catalog(session, row, use_model=False, actor=None)
|
||||
picked = [r.table for r in report.tables]
|
||||
assert len(picked) == catalog.DRAFT_DEFAULT_TABLES == 20
|
||||
assert picked[:3] == ["t24", "t03", "t10"] # 次数相同按表结构里的顺序
|
||||
assert picked[3:6] == ["t00", "t01", "t02"]
|
||||
|
||||
|
||||
async def test_draft_catalog_unknown_table_is_reported():
|
||||
row = await _db_source(_scenic_tables())
|
||||
async with SessionLocal() as session:
|
||||
report = await catalog.draft_catalog(session, row, tables=["nope"], use_model=False, actor=None)
|
||||
assert report.tables[0].error and "nope" in report.tables[0].error
|
||||
|
||||
|
||||
def test_draft_prompt_constants_are_model_only():
|
||||
"""提示词常量的名字带 _SYSTEM / _PROMPT：文案检查按名字把它们排除在界面文案之外。"""
|
||||
assert catalog.CATALOG_DRAFT_SYSTEM and catalog.CATALOG_DRAFT_PROMPT
|
||||
json.dumps(catalog.CATALOG_DRAFT_PROMPT_SCHEMA, ensure_ascii=False)
|
||||
@@ -0,0 +1,183 @@
|
||||
"""数据目录的合并、单项审阅和人工编辑规则（纯函数）。
|
||||
|
||||
- 起草结果并入已有目录：人工确认过、驳回过的项一律不动；同一来源的项可以更新；不同来源按可信程度
|
||||
（人工 > 外键 > 数据剖析 > 数据库注释 > 模型起草、命名推断）决定谁留下；这一轮完整算过的来源，
|
||||
已经没有的旧项删掉（外键约束被删了，原来那条「有确证」的关系不能留着）。
|
||||
- 单项审阅：confirm / reject / reset，路径写法 grain、columns.amount.measure、relations.<编号>。
|
||||
- 人工提交整份目录：改动过的项记为 human / confirmed，没动过的项保持原来的来源和状态。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from app.data import catalog
|
||||
from app.data.catalog import make_item
|
||||
|
||||
AT = "2026-10-03T08:00:00+00:00"
|
||||
|
||||
|
||||
def _rel(columns, to_table, to_columns, source="name", status=None, table="visits", **kw):
|
||||
return {"id": catalog.relation_id(table, columns, to_table, to_columns), "columns": columns,
|
||||
"to_table": to_table, "to_columns": to_columns, "cardinality": "many_to_one", "coverage": None,
|
||||
"source": source, "status": status or catalog.initial_status(source), **kw}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 合并
|
||||
|
||||
|
||||
def test_merge_adds_new_items_and_counts():
|
||||
draft = {"label": make_item("入园记录", "llm"),
|
||||
"columns": {"amount": {"unit": make_item("元", "llm")}},
|
||||
"relations": [_rel(["gate_id"], "gates", ["id"])]}
|
||||
merged, stats = catalog.merge_notes({}, draft, at=AT)
|
||||
assert stats == catalog.MergeStats(added=3, updated=0, removed=0)
|
||||
assert merged["label"]["value"] == "入园记录" and merged["label"]["updated_at"] == AT
|
||||
assert merged["columns"]["amount"]["unit"]["status"] == "proposed"
|
||||
assert merged["relations"][0]["source"] == "name"
|
||||
|
||||
|
||||
def test_merge_never_overwrites_confirmed_or_rejected():
|
||||
existing = {"label": make_item("入园明细", "human"),
|
||||
"grain": make_item("每张票一行", "llm", "rejected"),
|
||||
"relations": [_rel(["member_id"], "members", ["id"], status="rejected")]}
|
||||
draft = {"label": make_item("入园记录", "llm"), "grain": make_item("每张票一行", "llm"),
|
||||
"relations": [_rel(["member_id"], "members", ["id"])]}
|
||||
merged, stats = catalog.merge_notes(existing, draft, covered={"llm", "name"}, at=AT)
|
||||
assert merged == existing
|
||||
assert stats == catalog.MergeStats(0, 0, 0)
|
||||
|
||||
|
||||
def test_merge_same_source_updates_value():
|
||||
existing = {"grain": make_item("每次入园一行", "llm", at="2026-09-01T00:00:00+00:00")}
|
||||
merged, stats = catalog.merge_notes(existing, {"grain": make_item("每张门票一次检票一行", "llm")}, at=AT)
|
||||
assert merged["grain"]["value"] == "每张门票一次检票一行" and merged["grain"]["updated_at"] == AT
|
||||
assert stats.updated == 1
|
||||
|
||||
|
||||
def test_merge_same_value_is_not_an_update():
|
||||
existing = {"grain": make_item("每次入园一行", "llm", at="2026-09-01T00:00:00+00:00")}
|
||||
merged, stats = catalog.merge_notes(existing, {"grain": make_item("每次入园一行", "llm")}, at=AT)
|
||||
assert merged == existing and stats == catalog.MergeStats(0, 0, 0)
|
||||
|
||||
|
||||
def test_merge_higher_rank_source_replaces_lower_but_not_the_reverse():
|
||||
# 外键约束（有确证）顶掉同一条关系的命名推断
|
||||
existing = {"relations": [_rel(["member_id"], "members", ["id"], source="name")]}
|
||||
merged, stats = catalog.merge_notes(existing, {"relations": [_rel(["member_id"], "members", ["id"], source="fk")]},
|
||||
at=AT)
|
||||
assert merged["relations"][0]["source"] == "fk" and merged["relations"][0]["status"] == "verified"
|
||||
assert stats.updated == 1
|
||||
# 数据库注释比模型起草可信：模型起草不顶掉注释，注释可以顶掉模型起草
|
||||
existing = {"label": make_item("入园记录", "comment")}
|
||||
merged, _ = catalog.merge_notes(existing, {"label": make_item("游客入园", "llm")}, at=AT)
|
||||
assert merged["label"]["source"] == "comment"
|
||||
merged, _ = catalog.merge_notes({"label": make_item("游客入园", "llm")}, {"label": make_item("入园记录", "comment")},
|
||||
at=AT)
|
||||
assert merged["label"] == {**make_item("入园记录", "comment"), "updated_at": AT}
|
||||
|
||||
|
||||
def test_merge_removes_stale_items_only_for_covered_sources():
|
||||
stale_fk = _rel(["store_id"], "stores", ["id"], source="fk")
|
||||
keep_name = _rel(["gate_id"], "gates", ["id"], source="name")
|
||||
existing = {"relations": [stale_fk, keep_name], "label": make_item("入园记录", "llm")}
|
||||
# 这一轮完整读过外键约束（covered 含 fk）而那条外键已经没了：删掉；命名推断和模型起草这一轮没算，留着
|
||||
merged, stats = catalog.merge_notes(existing, {}, covered={"fk"}, at=AT)
|
||||
assert merged["relations"] == [keep_name] and merged["label"]["value"] == "入园记录"
|
||||
assert stats.removed == 1
|
||||
|
||||
|
||||
def test_merge_keeps_column_entries_tidy():
|
||||
existing = {"columns": {"amount": {"unit": make_item("元", "llm")}}}
|
||||
merged, stats = catalog.merge_notes(existing, {}, covered={"llm"}, at=AT)
|
||||
assert merged == {} and stats.removed == 1
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 单项审阅
|
||||
|
||||
|
||||
def _notes():
|
||||
return {"grain": make_item("每次入园一行", "llm"),
|
||||
"columns": {"amount": {"measure": make_item("flow", "llm")}},
|
||||
"relations": [_rel(["park_id"], "parks", ["id"], source="fk"), _rel(["gate_id"], "gates", ["id"])]}
|
||||
|
||||
|
||||
def test_review_confirm_reject_reset_on_each_kind_of_path():
|
||||
notes = _notes()
|
||||
rid = notes["relations"][1]["id"]
|
||||
out = catalog.review_item(notes, "grain", "confirm", at=AT)
|
||||
assert out["grain"]["status"] == "confirmed" and out["grain"]["source"] == "llm" and out["grain"]["updated_at"] == AT
|
||||
out = catalog.review_item(out, "columns.amount.measure", "reject", at=AT)
|
||||
assert out["columns"]["amount"]["measure"]["status"] == "rejected"
|
||||
out = catalog.review_item(out, f"relations.{rid}", "confirm", at=AT)
|
||||
assert out["relations"][1]["status"] == "confirmed"
|
||||
# reset 回到来源的初始状态：外键是 verified，其余是 proposed
|
||||
fk_id = notes["relations"][0]["id"]
|
||||
out = catalog.review_item(catalog.review_item(out, f"relations.{fk_id}", "reject", at=AT),
|
||||
f"relations.{fk_id}", "reset", at=AT)
|
||||
assert out["relations"][0]["status"] == "verified"
|
||||
out = catalog.review_item(out, "columns.amount.measure", "reset", at=AT)
|
||||
assert out["columns"]["amount"]["measure"]["status"] == "proposed"
|
||||
assert notes["grain"]["status"] == "proposed" # 不改入参
|
||||
|
||||
|
||||
def test_review_reset_of_human_item_removes_it():
|
||||
notes = {"label": make_item("入园记录", "human")}
|
||||
assert catalog.review_item(notes, "label", "reset", at=AT) == {}
|
||||
|
||||
|
||||
def test_review_column_names_may_contain_dots():
|
||||
notes = {"columns": {"a.b": {"unit": make_item("元", "llm")}}}
|
||||
out = catalog.review_item(notes, "columns.a.b.unit", "confirm", at=AT)
|
||||
assert out["columns"]["a.b"]["unit"]["status"] == "confirmed"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("path", ["", "nope", "grain.extra", "columns.amount", "columns.amount.formula",
|
||||
"columns.missing.measure", "relations.r000000000000", "label"])
|
||||
def test_review_bad_paths(path):
|
||||
with pytest.raises(catalog.CatalogPathError):
|
||||
catalog.review_item(_notes(), path, "confirm", at=AT)
|
||||
|
||||
|
||||
def test_review_bad_action():
|
||||
with pytest.raises(catalog.CatalogPathError):
|
||||
catalog.review_item(_notes(), "grain", "approve", at=AT)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 人工编辑
|
||||
|
||||
|
||||
def test_human_edit_marks_changed_items_as_human_confirmed_and_keeps_the_rest():
|
||||
existing = _notes()
|
||||
submitted = {
|
||||
"grain": {**existing["grain"], "value": "每张门票每次检票一行"}, # 改了值
|
||||
"columns": {"amount": {"measure": existing["columns"]["amount"]["measure"], # 没动
|
||||
"unit": {"value": "元"}}}, # 新填，没写来源和状态
|
||||
"relations": existing["relations"][:1], # 删掉一条关系
|
||||
"dedup": {"value": "同一票号只计一次", "source": "llm", "status": "verified"}, # 冒充别的来源、状态
|
||||
}
|
||||
out = catalog.apply_human_edit(existing, submitted, table="visits", at=AT)
|
||||
assert out["grain"] == {"value": "每张门票每次检票一行", "source": "human", "status": "confirmed", "updated_at": AT}
|
||||
assert out["columns"]["amount"]["measure"] == existing["columns"]["amount"]["measure"]
|
||||
assert out["columns"]["amount"]["unit"] == {"value": "元", "source": "human", "status": "confirmed",
|
||||
"updated_at": AT}
|
||||
assert out["dedup"]["source"] == "human" and out["dedup"]["status"] == "confirmed"
|
||||
assert len(out["relations"]) == 1 and out["relations"][0] == existing["relations"][0]
|
||||
assert catalog.validate_notes(out) == []
|
||||
|
||||
|
||||
def test_human_edit_status_only_change_keeps_source():
|
||||
existing = _notes()
|
||||
submitted = {**existing, "grain": {**existing["grain"], "status": "confirmed"}}
|
||||
out = catalog.apply_human_edit(existing, submitted, table="visits", at=AT)
|
||||
assert out["grain"]["source"] == "llm" and out["grain"]["status"] == "confirmed"
|
||||
# 不能把推断改成「有确证」：verified 只来自外键约束和数据剖析
|
||||
submitted = {**existing, "grain": {**existing["grain"], "status": "verified"}}
|
||||
assert catalog.apply_human_edit(existing, submitted, table="visits", at=AT)["grain"]["status"] == "proposed"
|
||||
|
||||
|
||||
def test_human_edit_relation_id_follows_its_ends():
|
||||
existing = _notes()
|
||||
moved = {**existing["relations"][1], "to_table": "parks", "to_columns": ["id"], "columns": ["gate_id"]}
|
||||
out = catalog.apply_human_edit(existing, {"relations": [moved]}, table="visits", at=AT)
|
||||
assert out["relations"][0]["id"] == catalog.relation_id("visits", ["gate_id"], "parks", ["id"])
|
||||
assert out["relations"][0]["source"] == "human"
|
||||
@@ -0,0 +1,187 @@
|
||||
"""数据目录交给模型的文本、关系图与连接路径、冻结内容、使用次数。"""
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
|
||||
from app.data import catalog
|
||||
from app.data.catalog import make_item
|
||||
from app.db.base import SessionLocal
|
||||
from app.db.models import Artifact, DataSource, Run
|
||||
|
||||
MARK = "（推断，未确认）"
|
||||
|
||||
|
||||
def _rel(table, columns, to_table, to_columns, source="fk", status=None, cardinality="many_to_one"):
|
||||
return {"id": catalog.relation_id(table, columns, to_table, to_columns), "columns": columns, "to_table": to_table,
|
||||
"to_columns": to_columns, "cardinality": cardinality, "coverage": None, "source": source,
|
||||
"status": status or catalog.initial_status(source)}
|
||||
|
||||
|
||||
VISITS = {
|
||||
"label": make_item("入园记录", "human"),
|
||||
"grain": make_item("每张门票每次检票一行", "llm"),
|
||||
"kind": make_item("fact", "llm", "confirmed"),
|
||||
"keys": make_item(["ticket_no"], "human"),
|
||||
"business_date": make_item({"column": "visit_time", "rule": "按检票时间计", "timezone": "+08:00"}, "human"),
|
||||
"valid_filter": make_item("status = 1", "human"),
|
||||
"dedup": make_item("同一票号只计一次", "llm", "rejected"),
|
||||
"columns": {
|
||||
"visitor_count": {"meaning": make_item("这次检票进园的人数", "human"), "unit": make_item("人", "llm"),
|
||||
"measure": make_item("flow", "llm", "confirmed")},
|
||||
"status": {"codes": make_item({"1": "有效", "0": "作废"}, "human")},
|
||||
},
|
||||
"relations": [_rel("visits", ["park_id"], "parks", ["id"]),
|
||||
_rel("visits", ["gate_id"], "gates", ["id"], source="name"),
|
||||
_rel("visits", ["member_id"], "members", ["id"], source="name", status="rejected")],
|
||||
}
|
||||
|
||||
|
||||
def test_full_render_shows_every_fact_and_marks_inferences():
|
||||
text = catalog.render_table_notes(VISITS)
|
||||
assert "中文名：入园记录" in text
|
||||
assert f"粒度：每张门票每次检票一行{MARK}" in text
|
||||
assert "表类型：事实表" in text and "业务主键：ticket_no" in text
|
||||
assert "业务日期：visit_time" in text and "按检票时间计" in text and "+08:00" in text
|
||||
assert "有效记录条件：status = 1" in text
|
||||
assert "visitor_count：这次检票进园的人数" in text and f"单位：人{MARK}" in text and "流量" in text
|
||||
assert "1=有效" in text and "0=作废" in text
|
||||
assert "park_id → parks.id" in text and f"gate_id → gates.id（多对一）{MARK}" in text
|
||||
# 驳回的项模型看不到
|
||||
assert "同一票号只计一次" not in text and "members" not in text
|
||||
# 有确证、已确认的不标推断
|
||||
assert "表类型：事实表（" not in text and "中文名：入园记录（" not in text
|
||||
|
||||
|
||||
def test_render_of_empty_or_all_rejected_notes_is_empty():
|
||||
assert catalog.render_table_notes({}) == ""
|
||||
assert catalog.render_table_notes({"dedup": make_item("x", "llm", "rejected")}) == ""
|
||||
|
||||
|
||||
def test_render_beside_system_notes_only_adds_what_they_do_not_cover():
|
||||
"""导入表格：表和列已有按核对结果生成的说明。保留说明，只追加人工确认过的、或说明没有覆盖的项。"""
|
||||
meta = {"comment": "粒度：每个日期。客流 等于 各入口之和（导入时已逐日核对）",
|
||||
"columns": [{"name": "日期", "type": "TEXT", "comment": "格式 YYYY-MM-DD"},
|
||||
{"name": "客流", "type": "INTEGER", "comment": None}]}
|
||||
notes = {"grain": make_item("每天一行", "llm"), "label": make_item("客流日报", "llm"),
|
||||
"description": make_item("人工补充的说明", "human"),
|
||||
"columns": {"日期": {"meaning": make_item("统计日期", "llm"), "measure": make_item("attribute", "llm")},
|
||||
"客流": {"unit": make_item("人次", "llm")}}}
|
||||
text = catalog.render_table_notes(notes, meta=meta, system_notes=True)
|
||||
assert "每天一行" not in text and "统计日期" not in text # 说明已经覆盖、又没人确认的：不追加
|
||||
assert "客流日报" in text and "人工补充的说明" in text # 说明没有覆盖的、人工确认过的：追加
|
||||
assert "属性" in text and "人次" in text
|
||||
|
||||
|
||||
def test_compact_index_one_line_per_table():
|
||||
cache = {"tables": {"visits": {"qualified": "visits", "columns": []},
|
||||
"parks": {"qualified": "main.parks", "columns": []},
|
||||
"gates": {"qualified": "gates", "columns": []}}}
|
||||
text = catalog.render_table_index(cache, {"visits": VISITS, "parks": {"label": make_item("景区", "llm")}})
|
||||
lines = text.splitlines()
|
||||
assert len(lines) == 3
|
||||
assert lines[0].startswith("visits｜入园记录｜每张门票每次检票一行") and lines[0].endswith(MARK)
|
||||
assert lines[1].startswith("main.parks｜景区") and MARK in lines[1]
|
||||
assert lines[2] == "gates"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 关系图
|
||||
|
||||
|
||||
NOTES = {
|
||||
"visits": {"relations": [_rel("visits", ["park_id"], "parks", ["id"]),
|
||||
_rel("visits", ["member_id"], "members", ["id"], source="name", status="rejected")]},
|
||||
"channel_visits": {"relations": [_rel("channel_visits", ["channel_id"], "channels", ["id"]),
|
||||
_rel("channel_visits", ["visit_id"], "visits", ["id"])]},
|
||||
"orders": {"relations": [_rel("orders", ["channel_id"], "channels", ["id"]),
|
||||
_rel("orders", ["member_id"], "members", ["id"])]},
|
||||
}
|
||||
|
||||
|
||||
def test_graph_adjacency_is_bidirectional_and_skips_rejected():
|
||||
graph = catalog.relation_graph(NOTES)
|
||||
assert {e.to_table for e in graph["visits"]} == {"parks", "channel_visits"}
|
||||
back = next(e for e in graph["channels"] if e.to_table == "orders")
|
||||
assert back.from_columns == ("id",) and back.to_columns == ("channel_id",) and back.cardinality == "one_to_many"
|
||||
assert all(e.to_table != "members" for e in graph["visits"])
|
||||
|
||||
|
||||
def test_join_paths_up_to_two_hops():
|
||||
graph = catalog.relation_graph(NOTES)
|
||||
paths = catalog.join_paths(graph, "visits", "channels")
|
||||
assert [tuple(e.to_table for e in p) for p in paths] == [("channel_visits", "channels")]
|
||||
assert paths[0][0].condition() == "visits.id = channel_visits.visit_id"
|
||||
assert paths[0][1].condition() == "channel_visits.channel_id = channels.id"
|
||||
assert catalog.join_paths(graph, "visits", "parks")[0][0].to_table == "parks"
|
||||
# 三跳以上不给：visits → channel_visits → channels → orders
|
||||
assert catalog.join_paths(graph, "visits", "orders") == []
|
||||
# 驳回的关系不出现在路径里
|
||||
assert catalog.join_paths(graph, "visits", "members") == []
|
||||
|
||||
|
||||
def test_graph_can_fill_in_relations_from_structure():
|
||||
"""没起草过目录的表，关系图可以按表结构现推（外键 + 命名推断），驳回过的照样排除。"""
|
||||
cache = {"tables": {
|
||||
"visits": {"columns": [{"name": "id", "type": "INTEGER"}, {"name": "gate_id", "type": "INTEGER"},
|
||||
{"name": "member_id", "type": "INTEGER"}],
|
||||
"primary_key": ["id"], "foreign_keys": []},
|
||||
"gates": {"columns": [{"name": "id", "type": "INTEGER"}], "primary_key": ["id"], "foreign_keys": []},
|
||||
"members": {"columns": [{"name": "id", "type": "INTEGER"}], "primary_key": ["id"], "foreign_keys": []},
|
||||
}}
|
||||
rejected = {"visits": {"relations": [_rel("visits", ["member_id"], "members", ["id"], source="name",
|
||||
status="rejected")]}}
|
||||
graph = catalog.relation_graph(rejected, schema_cache=cache)
|
||||
assert {e.to_table for e in graph["visits"]} == {"gates"}
|
||||
assert graph["gates"][0].status == "proposed"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 冻结
|
||||
|
||||
|
||||
def test_frozen_catalog_strips_rejected_and_keeps_versions():
|
||||
entries = {"visits": catalog.CatalogEntry("visits", VISITS, 4),
|
||||
"gone": catalog.CatalogEntry("gone", {"label": make_item("x", "human")}, 2),
|
||||
"empty": catalog.CatalogEntry("empty", {"dedup": make_item("x", "llm", "rejected")}, 1)}
|
||||
frozen = catalog.frozen_catalog(entries, ["visits", "empty", "parks"])
|
||||
assert set(frozen) == {"visits"} # 表结构里没有的、全被驳回的都不冻结
|
||||
assert frozen["visits"]["version"] == 4
|
||||
assert "dedup" not in frozen["visits"]["notes"]
|
||||
assert len(frozen["visits"]["notes"]["relations"]) == 2
|
||||
assert catalog.frozen_catalog({}, ["visits"]) == {}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 使用次数
|
||||
|
||||
|
||||
async def test_table_usage_counts_run_queries_per_table(monkeypatch):
|
||||
sid = uuid.uuid4().hex
|
||||
name = f"use_{sid[:8]}"
|
||||
cache = {"tables": {"visits": {"qualified": "main.visits"}, "parks": {"qualified": "main.parks"},
|
||||
"明细": {"qualified": "main.明细"}}}
|
||||
run_a, run_b = uuid.uuid4().hex, uuid.uuid4().hex
|
||||
legacy = {sid[:8] + "legacy1": {"source": name, "sql": 'SELECT * FROM "Visits" v JOIN main.parks p ON 1=1'}}
|
||||
monkeypatch.setattr(catalog, "_load_artifact", lambda aid: legacy.get(aid))
|
||||
async with SessionLocal() as session:
|
||||
session.add(DataSource(id=sid, name=name, kind="sqlite", database=":memory:", schema_cache=cache))
|
||||
session.add_all([Run(id=run_a, graph={}), Run(id=run_b, graph={})])
|
||||
session.add_all([
|
||||
# 新写法：查询当时把源和表名记在工件引用的 meta 里
|
||||
Artifact(id=sid[:8] + "q1", run_id=run_a, kind="query_snapshot",
|
||||
meta={"source_id": sid, "source": name, "tables": ["visits"]}),
|
||||
Artifact(id=sid[:8] + "q2", run_id=run_b, kind="query_snapshot",
|
||||
meta={"source_id": sid, "source": name, "tables": ["visits", "明细"]}),
|
||||
# 同一个查询快照被两次运行引用：各算一次
|
||||
Artifact(id=sid[:8] + "q2", run_id=run_a, kind="query_snapshot",
|
||||
meta={"source_id": sid, "source": name, "tables": ["visits", "明细"]}),
|
||||
# 别的数据源、工具库试用（不属于任何运行）、别的工件：不算
|
||||
Artifact(id=sid[:8] + "q3", run_id=run_a, kind="query_snapshot",
|
||||
meta={"source_id": "other", "source": "other", "tables": ["visits"]}),
|
||||
Artifact(id=sid[:8] + "q4", run_id="playground", kind="query_snapshot",
|
||||
meta={"source_id": sid, "source": name, "tables": ["visits"]}),
|
||||
Artifact(id=sid[:8] + "s1", run_id=run_a, kind="schema_snapshot", meta={"source": name}),
|
||||
# 老的查询快照引用没有 meta：读工件内容里的源名和 SQL
|
||||
Artifact(id=sid[:8] + "legacy1", run_id=run_b, kind="query_snapshot", meta={}),
|
||||
])
|
||||
await session.commit()
|
||||
source = await session.get(DataSource, sid)
|
||||
usage = await catalog.table_usage(session, source)
|
||||
assert usage == {"visits": 4, "明细": 2, "parks": 1}
|
||||
@@ -127,7 +127,7 @@ def test_validate_accepts_full_example():
|
||||
notes = {
|
||||
"label": _label("入园记录"), "description": _label("每次检票入园记一行", "llm", "proposed"),
|
||||
"grain": _label("每次入园"), "keys": _label(["visit_id"]), "kind": _label("fact"),
|
||||
"business_date": _label({"column": "visit_date", "rule": "按检票时间计", "timezone": "Asia/Shanghai"}),
|
||||
"business_date": _label({"column": "visit_date", "rule": "按检票时间计", "timezone": "+08:00"}),
|
||||
"valid_filter": _label("status = 1"), "dedup": _label("同一票号只计一次"),
|
||||
"columns": {"amount": {"label": _label("实收金额"), "meaning": _label("扣除优惠后的金额"),
|
||||
"unit": _label("元"), "measure": _label("flow"),
|
||||
|
||||
Reference in New Issue
Block a user