From 6f6798153e035a58196a3a81ed4143bb71014201 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Sat, 3 Oct 2026 09:27:25 +0800 Subject: [PATCH] =?UTF-8?q?=E6=95=B0=E6=8D=AE=E7=9B=AE=E5=BD=95=E9=98=B6?= =?UTF-8?q?=E6=AE=B5=201A=EF=BC=9Acatalog.py=20=E8=A1=A5=E9=BD=90=E5=90=88?= =?UTF-8?q?=E5=B9=B6=E4=B8=8E=E5=AE=A1=E9=98=85=E3=80=81=E6=B3=A8=E9=87=8A?= =?UTF-8?q?=20/=20=E5=A4=96=E9=94=AE=20/=20=E5=91=BD=E5=90=8D=E6=8E=A8?= =?UTF-8?q?=E6=96=AD=20/=20=E6=A8=A1=E5=9E=8B=E8=B5=B7=E8=8D=89=E3=80=81?= =?UTF-8?q?=E6=B8=B2=E6=9F=93=E3=80=81=E5=85=B3=E7=B3=BB=E5=9B=BE=E3=80=81?= =?UTF-8?q?=E4=BD=BF=E7=94=A8=E6=AC=A1=E6=95=B0=E5=92=8C=E5=86=BB=E7=BB=93?= =?UTF-8?q?=E5=86=85=E5=AE=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/data/catalog.py | 1324 +++++++++++++++++++++++++- backend/tests/test_catalog_draft.py | 379 ++++++++ backend/tests/test_catalog_merge.py | 183 ++++ backend/tests/test_catalog_render.py | 187 ++++ backend/tests/test_catalog_store.py | 2 +- 5 files changed, 2066 insertions(+), 9 deletions(-) create mode 100644 backend/tests/test_catalog_draft.py create mode 100644 backend/tests/test_catalog_merge.py create mode 100644 backend/tests/test_catalog_render.py diff --git a/backend/app/data/catalog.py b/backend/app/data/catalog.py index f05900dc..ae2a482d 100644 --- a/backend/app/data/catalog.py +++ b/backend/app/data/catalog.py @@ -30,9 +30,14 @@ human 人工填写。状态（status）：proposed 推断（命名推断、模 """ from __future__ import annotations +import asyncio import copy +import hashlib import json -from dataclasses import dataclass +import re +from collections import Counter +from collections.abc import Iterable, Mapping +from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Any @@ -40,7 +45,8 @@ from sqlalchemy import select, update from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession -from app.db.models import CatalogNote +from app.data.names import name_key +from app.db.models import Artifact, CatalogNote, Run # ========================================================================== # 取值 @@ -368,16 +374,1318 @@ def relation_id(table: str, columns: list[str], to_table: str, to_columns: list[ 列按 (本表列, 被指向列) 成对排序后再算，复合外键列的书写顺序不影响编号；不区分大小写（外键约束 和命名推断拿到的大小写可能不同）。编号里不含句点：审阅接口的路径按句点切分（relations.<编号>）。 """ - import hashlib - pairs = sorted(zip((c.lower() for c in columns), (c.lower() for c in to_columns))) raw = json.dumps([table.lower(), to_table.lower(), pairs], ensure_ascii=False, separators=(",", ":")) return "r" + hashlib.sha256(raw.encode()).hexdigest()[:12] +# ========================================================================== +# 槽位：把 notes 摊平成「一项一个键」，合并、审阅、人工编辑都在这上面做 +# ========================================================================== + +#: 槽位键：("t", 字段) 表级项、("c", 列名, 字段) 列级项、("r", 关系编号) 关系 +Slot = tuple[str, ...] + +#: 关系里决定「是不是同一条、值有没有变」的字段（来源、状态、备注、时间之外的部分） +_RELATION_VALUE_KEYS = ("columns", "to_table", "to_columns", "cardinality", "coverage") + + +def _slots(notes: Mapping[str, Any] | None) -> dict[Slot, dict[str, Any]]: + """notes → {槽位: 项}（深拷贝，不改入参）。顺序：表级项、列级项、关系，各自保持原来的顺序。""" + out: dict[Slot, dict[str, Any]] = {} + notes = notes or {} + for name in TABLE_FIELDS: + if isinstance(notes.get(name), dict): + out[("t", name)] = copy.deepcopy(notes[name]) + columns = notes.get("columns") if isinstance(notes.get("columns"), dict) else {} + for col, items in columns.items(): + for name in COLUMN_FIELDS: + if isinstance(items, dict) and isinstance(items.get(name), dict): + out[("c", col, name)] = copy.deepcopy(items[name]) + relations = notes.get("relations") if isinstance(notes.get("relations"), list) else [] + for rel in relations: + if isinstance(rel, dict) and rel.get("id"): + out[("r", str(rel["id"]))] = copy.deepcopy(rel) + return out + + +def _assemble(slots: Mapping[Slot, dict[str, Any]]) -> dict[str, Any]: + """槽位 → notes。空的列、空的关系列表不留：目录内容相同则序列化结果相同（冻结的哈希靠它）。""" + notes: dict[str, Any] = {} + columns: dict[str, dict[str, Any]] = {} + relations: list[dict[str, Any]] = [] + for key, item in slots.items(): + if key[0] == "t": + notes[key[1]] = item + elif key[0] == "c": + columns.setdefault(key[1], {})[key[2]] = item + else: + relations.append(item) + if columns: + notes["columns"] = columns + if relations: + notes["relations"] = relations + return notes + + +def _value_of(key: Slot, item: Mapping[str, Any]) -> Any: + """一项的「值」：比较人工有没有改动时只看它。""" + if key[0] == "r": + return {k: item.get(k) for k in _RELATION_VALUE_KEYS} + return item.get("value") + + +def _core(item: Mapping[str, Any]) -> dict[str, Any]: + """去掉修改时间的项：判断起草结果和已有的是不是一回事。""" + return {k: v for k, v in item.items() if k != "updated_at"} + + +def visible_notes(notes: Mapping[str, Any] | None) -> dict[str, Any]: + """去掉被驳回的项和关系。给模型看、冻结进证据、将来的检查都只用这一份。""" + return _assemble({k: v for k, v in _slots(notes).items() if v.get("status") != "rejected"}) + + +def status_counts(notes: Mapping[str, Any] | None) -> dict[str, int]: + """各状态的项数（表级项、列级项、关系都算），四种状态都有键。""" + counts = dict.fromkeys(ITEM_STATUSES, 0) + for item in _slots(notes).values(): + if item.get("status") in counts: + counts[item["status"]] += 1 + return counts + + +# ========================================================================== +# 合并：起草结果并入已有目录 +# ========================================================================== + +#: 来源的可信程度。不同来源对同一个槽位各有说法时，高的留下；同级（模型起草和命名推断）先到先得 +_SOURCE_RANK = {"human": 5, "fk": 4, "profile": 3, "comment": 2, "llm": 1, "name": 1} +#: 人工定过的状态：起草永远不碰 +_LOCKED = ("confirmed", "rejected") + + +@dataclass(frozen=True) +class MergeStats: + """一次合并新增、更新、删除了几项。""" + + added: int = 0 + updated: int = 0 + removed: int = 0 + + def __add__(self, other: "MergeStats") -> "MergeStats": + return MergeStats(self.added + other.added, self.updated + other.updated, self.removed + other.removed) + + +def merge_notes(existing: Mapping[str, Any] | None, draft: Mapping[str, Any] | None, *, + covered: Iterable[str] = (), at: str | None = None) -> tuple[dict[str, Any], MergeStats]: + """把起草结果并入已有目录，返回 (新目录, 计数)。不改入参。 + + 规则（逐个槽位）： + - 已有项是 confirmed 或 rejected：不动。驳回的留着，同一项下次起草就不会再冒出来。 + - 起草结果里没有这一项：已有项的来源在 covered 里（这一轮完整算过这个来源）就删掉，否则留着。 + 外键约束被删掉以后，原来那条「有确证」的关系不能还挂着；而这一轮没调模型，就不能因此删掉模型起草的项。 + - 同一来源：值或状态有变化就更新。 + - 不同来源：来源更可信的（_SOURCE_RANK）顶掉不如它的，反过来不行。 + + 新增和更新的项盖上 at（缺省为当前时间）作为 updated_at；内容没变的项保持原样，修改时间也不动。 + """ + at = at or now_iso() + covered = set(covered) + old, new = _slots(existing), _slots(draft) + out: dict[Slot, dict[str, Any]] = {} + added = updated = removed = 0 + for key, item in old.items(): + cand = new.get(key) + if item.get("status") in _LOCKED: + out[key] = item + elif cand is None: + if item.get("source") in covered: + removed += 1 + else: + out[key] = item + elif (cand.get("source") == item.get("source") + or _SOURCE_RANK.get(cand.get("source"), 0) > _SOURCE_RANK.get(item.get("source"), 0)): + if _core(cand) != _core(item): + out[key] = {**_core(cand), "updated_at": at} + updated += 1 + else: + out[key] = item + else: + out[key] = item + for key, cand in new.items(): + if key not in old: + out[key] = {**_core(cand), "updated_at": at} + added += 1 + return _assemble(out), MergeStats(added, updated, removed) + + +# ========================================================================== +# 单项审阅与人工编辑 +# ========================================================================== + +#: 单项审阅的操作：confirm 确认、reject 驳回、reset 撤销审阅（回到来源的初始状态；人工填写的项直接删掉） +REVIEW_ACTIONS = ("confirm", "reject", "reset") + + +class CatalogPathError(ValueError): + """审阅的路径或操作不对：路径写错、项不存在、操作不认识。message 给人看。""" + + +def parse_path(path: str) -> Slot: + """审阅路径 → 槽位。 + + 写法：表级项直接写字段名（grain）；列级项 columns.<列名>.<字段>（列名里可以有句点，按最后一个句点切出 + 字段）；关系 relations.<编号>。 + """ + path = (path or "").strip() + if path in TABLE_FIELDS: + return ("t", path) + if path.startswith("columns."): + col, _, name = path[len("columns."):].rpartition(".") + if col and name in COLUMN_FIELDS: + return ("c", col, name) + if path.startswith("relations.") and path[len("relations."):]: + return ("r", path[len("relations."):]) + raise CatalogPathError("无法识别要审阅的项，请重新载入后再试") + + +def describe_slot(key: Slot) -> str: + """槽位的中文说法：「表的粒度」「列 amount 的度量类型」「关联关系 r1a2b3」。""" + if key[0] == "t": + return f"表的{TABLE_FIELD_LABEL.get(key[1], key[1])}" + if key[0] == "c": + return f"列 {key[1]} 的{COLUMN_FIELD_LABEL.get(key[2], key[2])}" + return f"关联关系 {key[1]}" + + +def review_item(notes: Mapping[str, Any] | None, path: str, action: str, *, at: str | None = None) -> dict[str, Any]: + """对一项做审阅，返回新目录（不改入参）。路径不对、项不存在、操作不认识都抛 CatalogPathError。""" + if action not in REVIEW_ACTIONS: + raise CatalogPathError("不支持这种审阅操作，只能确认、驳回或撤销审阅") + key = parse_path(path) + slots = _slots(notes) + item = slots.get(key) + if item is None: + raise CatalogPathError(f"找不到{describe_slot(key)}，可能已被修改，请重新载入") + if action == "reset" and item.get("source") == "human": + # 人工填写的项没有「推断时的样子」可回：撤销就是删掉，下次起草可以重新提出 + del slots[key] + return _assemble(slots) + item["status"] = {"confirm": "confirmed", "reject": "rejected"}.get(action) or initial_status(item.get("source")) + item["updated_at"] = at or now_iso() + return _assemble(slots) + + +def _normalize_submitted(submitted: Mapping[str, Any], table: str) -> dict[str, Any]: + """人工提交的目录：补上没写的来源和状态（下面会按规则改写），关系编号按两端重算。""" + out = copy.deepcopy(dict(submitted)) + + def fill(item: Any) -> None: + if isinstance(item, dict) and "value" in item: + item.setdefault("source", "human") + item.setdefault("status", "confirmed") + + for name in TABLE_FIELDS: + fill(out.get(name)) + columns = out.get("columns") if isinstance(out.get("columns"), dict) else {} + for items in columns.values(): + for name in COLUMN_FIELDS: + fill(items.get(name) if isinstance(items, dict) else None) + relations = out.get("relations") if isinstance(out.get("relations"), list) else [] + for rel in relations: + if not isinstance(rel, dict): + continue + rel.setdefault("source", "human") + rel.setdefault("status", "confirmed") + rel.setdefault("cardinality", None) + rel.setdefault("coverage", None) + cols, to_table, to_cols = rel.get("columns"), rel.get("to_table"), rel.get("to_columns") + if _str_list_ok(cols) and isinstance(to_table, str) and to_table and _str_list_ok(to_cols): + # 编号由两端决定：改了指向就是另一条关系，沿用旧编号会让「同一条关系同一个编号」失效 + rel["id"] = relation_id(table, cols, to_table, to_cols) + return out + + +def apply_human_edit(existing: Mapping[str, Any] | None, submitted: Mapping[str, Any], *, table: str, + at: str | None = None) -> dict[str, Any]: + """人工提交的整份目录 → 要写入的目录（不改入参）。 + + - 值改过的项、新填的项：记为 human / confirmed。提交里写的来源和状态不作数——客户端不能冒充外键约束 + 或数据剖析。 + - 值没动的项：保持原来的来源和状态；提交里把状态改成确认、驳回或来源的初始状态的，照改（等同单项审阅）； + 改成别的状态的不认（不能把推断改成「有确证」）。 + - 已有、但提交里没有的项：删掉。 + - 关系编号按两端重算（_normalize_submitted）。 + + 结构不合规抛 CatalogInvalid。 + """ + if not isinstance(submitted, Mapping): + raise CatalogInvalid(["数据目录应为对象"]) + normalized = _normalize_submitted(submitted, table) + if problems := validate_notes(normalized): + raise CatalogInvalid(problems) + at = at or now_iso() + old, new = _slots(existing), _slots(normalized) + out: dict[Slot, dict[str, Any]] = {} + for key, item in new.items(): + prev = old.get(key) + if prev is None or _value_of(key, item) != _value_of(key, prev): + fresh = {k: v for k, v in item.items() if k != "updated_at"} + fresh.update(source="human", status="confirmed", updated_at=at) + out[key] = fresh + continue + kept = dict(prev) + status = item.get("status") + if status != prev.get("status") and status in ("confirmed", "rejected", initial_status(prev.get("source"))): + kept["status"] = status + kept["updated_at"] = at + if item.get("note") != prev.get("note"): + if item.get("note"): + kept["note"] = item["note"] + else: + kept.pop("note", None) + kept["updated_at"] = at + out[key] = kept + return _assemble(out) + + +async def review_entry(session: AsyncSession, source_id: str, table: str, path: str, action: str, *, + if_version: int, actor: str | None) -> CatalogEntry: + """单项审阅并写入（乐观锁同 write_entry）。版本不符抛 CatalogConflict，路径不对抛 CatalogPathError。""" + entry = await read_entry(session, source_id, table) + current = entry.version if entry else 0 + if if_version != current: + raise CatalogConflict(table, current) + notes = review_item(entry.notes if entry else {}, path, action) + return await write_entry(session, source_id, table, notes, if_version=current, actor=actor) + + +async def save_human_edit(session: AsyncSession, source_id: str, table: str, submitted: Mapping[str, Any], *, + if_version: int, actor: str | None) -> CatalogEntry: + """人工提交整份目录并写入（规则见 apply_human_edit，乐观锁同 write_entry）。""" + entry = await read_entry(session, source_id, table) + current = entry.version if entry else 0 + if if_version != current: + raise CatalogConflict(table, current) + notes = apply_human_edit(entry.notes if entry else {}, submitted, table=table) + return await write_entry(session, source_id, table, notes, if_version=current, actor=actor) + + +# ========================================================================== +# 起草：数据库注释、外键约束、命名推断（只看探查缓存，不发数据库查询） +# ========================================================================== + + +@dataclass(frozen=True) +class TableDraft: + """一张表的起草结果。covered：这一轮完整算过的来源（merge_notes 据此删掉这些来源里已经没有的旧项）。""" + + notes: dict[str, Any] + covered: frozenset[str] = frozenset() + + +def system_notes_source(source: Any) -> bool: + """这个源的表说明、列说明是不是系统生成的（导入表格）。 + + 上传表格建的源（origin=upload）：简单导入时未规整的表写着 UNSHAPED_NOTE，按配方导入时表和列的说明 + 由 recipe_notes 按核对结果生成（schema_cache 顶层 import_mode=recipe）。这些说明有严格的规矩（跟着 + 核对结果走、不带数字和坐标），目录不能重复它们，也不能和它们矛盾：起草时不拷进中文名和含义，渲染时 + 保留说明、只追加说明没有覆盖的项（render_table_notes）。 + """ + if (getattr(source, "origin", None) or "manual") == "upload": + return True + return (getattr(source, "schema_cache", None) or {}).get("import_mode") == "recipe" + + +def resolve_table_name(schema_cache: Mapping[str, Any] | None, name: str) -> str | None: + """在表结构里找表，返回 schema_cache["tables"] 的键。全名、只有表名、大小写不一致都认。""" + tables = (schema_cache or {}).get("tables") or {} + if name in tables: + return name + short = name.rsplit(".", 1)[-1] + if short in tables: + return short + lowered = short.lower() + return next((n for n in tables if n.lower() == lowered), None) + + +#: 数据库注释多长、带不带句读，决定它是「名字」还是「解释」 +_LABEL_MAX = 24 +_SENTENCE_MARKS = re.compile(r"[，。；;,.!?！？：:\n]") + + +def _comment_is_label(text: str) -> bool: + """短、不带句读的注释是名字（「实收金额」），否则是一段解释（「状态：1 表示有效…」）。""" + 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} + 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" + + +def _relation(table: str, columns: list[str], to_table: str, to_columns: list[str], source: str, *, + cardinality: str | None) -> dict[str, Any]: + return {"id": relation_id(table, columns, to_table, to_columns), "columns": list(columns), "to_table": to_table, + "to_columns": list(to_columns), "cardinality": cardinality, "coverage": None, "source": source, + "status": initial_status(source)} + + +def fk_relations(schema_cache: Mapping[str, Any] | None, table: str) -> list[dict[str, Any]] | None: + """外键约束 → 关系（fk / verified）。缓存里没有 foreign_keys（升级前探查的）返回 None：不知道有没有外键。 + + 外键没写被指向的列（SQLite 的 REFERENCES t 可以省略列）时取被指向表的主键；列数对不上的跳过。 + """ + tables = (schema_cache or {}).get("tables") or {} + meta = tables.get(table) + if not isinstance(meta, dict) or not isinstance(meta.get("foreign_keys"), list): + return None + out: dict[str, dict[str, Any]] = {} + for fk in meta["foreign_keys"]: + if not isinstance(fk, dict): + continue + cols, to_table = list(fk.get("columns") or []), fk.get("to_table") + to_cols = list(fk.get("to_columns") or []) + if not cols or not isinstance(to_table, str) or not to_table: + continue + if not to_cols: + to_cols = list((tables.get(to_table) or {}).get("primary_key") or []) + if len(to_cols) != len(cols): + continue + rel = _relation(table, cols, to_table, to_cols, "fk", cardinality=_cardinality(meta, cols)) + out.setdefault(rel["id"], rel) + return list(out.values()) + + +# ---- 命名推断 + +#: xxx_id（不分大小写）、xxxId、xxxID。没有分隔的 paid、uuid、PAID 都不算 +_SNAKE_ID = re.compile(r"^(?P.+?)_id$", re.IGNORECASE) +_CAMEL_ID = re.compile(r"^(?P.*[a-z0-9])(?:Id|ID)$") +_WORDS = re.compile(r"[A-Z]+(?=[A-Z][a-z])|[A-Z]?[a-z]+|[A-Z]+|\d+") + + +def _words(name: str) -> list[str]: + """名字切成小写的词：下划线和驼峰都是分隔（memberTag、member_tag、MEMBER_TAG 都是 member + tag）。""" + return [w.lower() for part in name.split("_") for w in _WORDS.findall(part)] + + +def _singulars(word: str) -> set[str]: + """一个词可能的单数形式（含它自己）。两边都取这个集合、有交集就算同一个词：categories 和 category、 + courses 和 course、status 和 statuses 都对得上，不靠一套「复数变单数」的规则硬猜。""" + out = {word} + if word.endswith("ies") and len(word) > 3: + out.add(word[:-3] + "y") + if word.endswith("es") and len(word) > 2: + out.add(word[:-2]) + if word.endswith("s") and not word.endswith("ss") and len(word) > 1: + out.add(word[:-1]) + return out + + +def _name_index(tables: Mapping[str, Any]) -> dict[tuple[str, str], set[str]]: + """{(前面几个词拼起来, 最后一个词的单数形式): 表名}。""" + index: dict[tuple[str, str], set[str]] = {} + for name in tables: + words = _words(name) + if not words: + continue + prefix = "".join(words[:-1]) + for form in _singulars(words[-1]): + index.setdefault((prefix, form), set()).add(name) + return index + + +def _type_family(type_name: str | None) -> str | None: + """列类型的大类：数字、文字、时间；认不出来返回 None（不据此排除）。""" + t = (type_name or "").upper() + if not t: + return None + if any(k in t for k in ("CHAR", "TEXT", "CLOB", "STRING", "UUID")): + return "text" + if any(k in t for k in ("INT", "NUMBER", "NUMERIC", "DECIMAL", "SERIAL", "REAL", "FLOAT", "DOUBLE")): + return "number" + if any(k in t for k in ("DATE", "TIME")): + return "time" + return None + + +def _id_stem(column: str) -> str | None: + m = _SNAKE_ID.match(column) or _CAMEL_ID.match(column) + return m.group("stem") if m else None + + +def infer_name_relations(schema_cache: Mapping[str, Any] | None, table: str, *, + _index: dict[tuple[str, str], set[str]] | None = None) -> list[dict[str, Any]]: + """按列名推断关系（name / proposed）：xxx_id、xxxId、xxxID 对上表名的单复数、驼峰和下划线变体，指向那张表的主键。 + + 推不准就不推。以下情况一律放过： + - 列上已经有外键约束（外键说了算）； + - 词根太短（x_id）； + - 没有对得上的表，或者对得上的表不止一张（guide 和 guides 都在）； + - 对上的是本表自己（employees.employee_id 是工号，不是自引用）； + - 被指向的表没有主键，或主键不止一列； + - 两边的类型大类不同（文字列指向整数主键）。 + """ + tables = (schema_cache or {}).get("tables") or {} + meta = tables.get(table) + if not isinstance(meta, dict): + return [] + index = _index if _index is not None else _name_index(tables) + constrained = {c.lower() for fk in meta.get("foreign_keys") or [] if isinstance(fk, dict) + for c in fk.get("columns") or []} + out: list[dict[str, Any]] = [] + for col in meta.get("columns") or []: + name = str(col.get("name") or "") + stem = _id_stem(name) + if stem is None or name.lower() in constrained: + continue + words = _words(stem) + if not words or len("".join(words)) < 2: + continue + prefix = "".join(words[:-1]) + candidates: set[str] = set() + for form in _singulars(words[-1]): + candidates |= index.get((prefix, form), set()) + if len(candidates) != 1: + continue + target = next(iter(candidates)) + if target == table: + continue + target_meta = tables.get(target) or {} + pk = list(target_meta.get("primary_key") or []) + if len(pk) != 1: + continue + pk_type = next((c.get("type") for c in target_meta.get("columns") or [] if c.get("name") == pk[0]), None) + mine, theirs = _type_family(col.get("type")), _type_family(pk_type) + if mine and theirs and mine != theirs: + continue + out.append(_relation(table, [name], target, pk, "name", cardinality=_cardinality(meta, [name]))) + return out + + +def draft_structure(source: Any, table: str) -> TableDraft: + """只看表结构就能起草的部分：数据库注释、外键约束、命名推断。table 必须是 schema_cache["tables"] 的键。 + + - 数据库注释 → 中文名或含义（comment / proposed）：短而不带句读的是名字，记进表或列的中文名；否则是 + 一段解释，记进表的说明或列的含义。导入表格的源例外（system_notes_source）：说明是系统生成的，不拷。 + - 外键约束 → 关系（fk / verified）。 + - 命名推断 → 关系（name / proposed）。 + """ + cache = getattr(source, "schema_cache", None) or {} + meta = (cache.get("tables") or {}).get(table) + if not isinstance(meta, dict): + raise KeyError(table) + notes: dict[str, Any] = {} + if not system_notes_source(source): + comment = (meta.get("comment") or "").strip() + if comment: + notes["label" if _comment_is_label(comment) else "description"] = make_item(comment, "comment") + columns: dict[str, dict[str, Any]] = {} + for col in meta.get("columns") or []: + text = (col.get("comment") or "").strip() + if text: + columns[col["name"]] = {("label" if _comment_is_label(text) else "meaning"): make_item(text, "comment")} + if columns: + notes["columns"] = columns + fks = fk_relations(cache, table) + relations = list(fks or []) + seen = {r["id"] for r in relations} + relations += [r for r in infer_name_relations(cache, table) if r["id"] not in seen] + if relations: + notes["relations"] = relations + covered = {"comment", "name"} | ({"fk"} if fks is not None else set()) + return TableDraft(notes=notes, covered=frozenset(covered)) + + +# ========================================================================== +# 起草：模型（可选，按表分批，结构化输出） +# ========================================================================== + +#: 每批最多几张表、合计多少列。一批太大模型容易漏表、输出被截断；太小则调用次数多 +MODEL_BATCH_TABLES = 4 +MODEL_BATCH_COLUMNS = 120 +#: 同时发出的批数 +MODEL_CONCURRENCY = 3 +#: 一批最多等多久 +MODEL_TIMEOUT_S = 120 +#: 输出额度。一批四张表的目录远用不完；给太大的额度有的服务方会直接拒收 +MODEL_MAX_TOKENS = 8192 +#: 不指定表时起草使用次数最多的前几张 +DRAFT_DEFAULT_TABLES = 20 + + +class CatalogModelUnavailable(RuntimeError): + """没有可用于起草的模型（没有接入、停用、只有演示接入、缺 key）。message 给人看；在调用模型之前判定。""" + + +CATALOG_DRAFT_SYSTEM = """你在为一个业务数据库编写数据目录：说明每张表、每一列在业务上是什么。只根据给出的表结构推断，\ +不要编造不存在的表和列。 + +要写的内容： +- label：表的中文名，简短（例如「入园记录」）。 +- grain：一行代表什么（例如「每张门票每次检票一行」）。 +- kind：表类型，fact（事实表，记录业务事件）、dimension（维度表，描述对象）、snapshot（快照表，某一时点的状态）、\ +log（日志表）、config（配置表）之一。 +- keys：业务主键，即在业务上唯一确定一行的列（可以和数据库主键不同），写列名。 +- business_date：业务日期按哪一列算，rule 写一句规则，timezone 写时区（不确定就留空）。 +- columns：每一列的 label（中文名）、meaning（含义，一句话）、unit（单位，例如「元」「人」，没有单位就留空）、\ +measure（度量类型：flow 流量，可跨期加总；stock 存量，时点值，不能跨期加总；ratio 比率；identifier 标识；\ +status 状态；attribute 属性）。 + +规矩： +- 只写数据事实，不写计算公式，不写指标怎么算。 +- 拿不准的字段留空，不要猜。列名、表名照原样写，不要翻译或改写。 +- 标着「系统说明」的内容是导入时按核对结果生成的，以它为准：不要重复，也不要与之矛盾。 +- 标着「已确认」的目录项由人工确认过，以它为准。""" + +CATALOG_DRAFT_PROMPT = "请为下面 {n} 张表编写数据目录，每张表输出一项，table 写表名。\n\n{tables}" + +#: 拼表结构描述用的固定片段（写给模型） +CATALOG_DRAFT_PROMPT_LABELS = { + "table": "表 {name}", + "view": "表 {name}（视图）", + "comment": "数据库注释：{text}", + "system": "系统说明：{text}", + "columns": "列：", + "column": "- {name} {type}", + "pk": "主键", + "not_null": "非空", + "fk": "外键 → {target}", + "col_comment": "注释：{text}", + "col_system": "系统说明：{text}", + "unique": "唯一约束：{cols}", + "confirmed": "已确认的目录项：", +} + +CATALOG_DRAFT_PROMPT_SCHEMA: dict[str, Any] = { + "title": "catalog_draft", + "description": "数据目录草稿，每张表一项", + "type": "object", + "properties": { + "tables": { + "type": "array", + "items": { + "type": "object", + "properties": { + "table": {"type": "string", "description": "表名，照原样写"}, + "label": {"type": "string", "description": "表的中文名"}, + "grain": {"type": "string", "description": "一行代表什么"}, + "kind": {"type": "string", "enum": list(TABLE_KINDS)}, + "keys": {"type": "array", "items": {"type": "string"}, "description": "业务主键的列名"}, + "business_date": { + "type": "object", + "properties": {"column": {"type": "string"}, "rule": {"type": "string"}, + "timezone": {"type": "string"}}, + "required": ["column"], + }, + "columns": { + "type": "array", + "items": { + "type": "object", + "properties": { + "name": {"type": "string", "description": "列名，照原样写"}, + "label": {"type": "string"}, "meaning": {"type": "string"}, + "unit": {"type": "string"}, + "measure": {"type": "string", "enum": list(COLUMN_MEASURES)}, + }, + "required": ["name"], + }, + }, + }, + "required": ["table"], + }, + }, + }, + "required": ["tables"], +} + +_P = CATALOG_DRAFT_PROMPT_LABELS + + +async def resolve_draft_model(session: AsyncSession) -> tuple[Any, str]: + """(模型, 模型 id)：用设置里助手的模型，关掉思考、温度 0。没有可用的模型抛 CatalogModelUnavailable。 + + 和 recipe_ai.resolve_draft_model 一样不用 get_chat_model：它在没有接入时静默退回演示模型，拿演示模型 + 起草出来的目录会被当成真的写进库里。数据层不导入接口层，设置键取 recipe_ai 里那份（与助手的同值， + 那边有测试核对）。 + """ + from app.core.errors import not_configured + from app.data.recipe_ai import COPILOT_SETTING_KEY + from app.db.models import Setting + from app.providers.factory import ModelSpec, ProviderNotConfigured, build_chat_model, resolve_provider + + row = await session.get(Setting, COPILOT_SETTING_KEY) + saved = (row.value if row else None) or {} + spec = ModelSpec(provider=saved.get("provider") or None, model=saved.get("model") or None, + max_tokens=MODEL_MAX_TOKENS, temperature=0, thinking="off") + unconfigured = "助手使用的模型尚未配置完成：{reason}。请到「设置 → 模型接入」检查" + try: + provider = await resolve_provider(session, spec.provider, spec.model) + except ProviderNotConfigured as e: + raise CatalogModelUnavailable(unconfigured.format(reason=not_configured(e, with_hint=False))) from e + if provider is None: + if spec.provider: + raise CatalogModelUnavailable(f"助手设置里的模型接入「{spec.provider}」不存在。请到「设置 → 模型接入」检查") + raise CatalogModelUnavailable("未配置模型接入，无法由助手起草数据目录。请到「设置 → 模型接入」添加") + if provider.kind == "mock": + raise CatalogModelUnavailable("当前只有演示用的模型接入，无法用于起草数据目录。" + "请到「设置 → 模型接入」添加真实的模型接入") + if not provider.enabled: + raise CatalogModelUnavailable(f"助手使用的模型接入「{provider.name}」已停用。" + "请到「设置 → 模型接入」启用它，或为助手换一个模型") + try: + model = build_chat_model(provider, spec) + except ProviderNotConfigured as e: + raise CatalogModelUnavailable(unconfigured.format(reason=not_configured(e, with_hint=False))) from e + return model, spec.model or provider.default_model or "" + + +def _confirmed_only(notes: Mapping[str, Any] | None) -> dict[str, Any]: + return _assemble({k: v for k, v in _slots(notes).items() if v.get("status") in ("confirmed", "verified")}) + + +def _describe_for_model(name: str, meta: Mapping[str, Any], *, system: bool, existing: Mapping[str, Any] | None) -> str: + """一张表交给模型的结构描述：列、类型、主键、外键、唯一约束、注释；不含任何数据值。""" + lines = [(_P["view"] if meta.get("is_view") else _P["table"]).format(name=name)] + comment = (meta.get("comment") or "").strip() + if comment: + lines.append((_P["system"] if system else _P["comment"]).format(text=comment)) + pk = set(meta.get("primary_key") or []) + fk_of: dict[str, str] = {} + for fk in meta.get("foreign_keys") or []: + if isinstance(fk, dict) and len(fk.get("columns") or []) == 1: + to_cols = fk.get("to_columns") or [] + fk_of[fk["columns"][0]] = f"{fk.get('to_table')}.{to_cols[0]}" if to_cols else str(fk.get("to_table")) + lines.append(_P["columns"]) + for col in meta.get("columns") or []: + marks = [] + if col["name"] in pk: + marks.append(_P["pk"]) + if not col.get("nullable", True): + marks.append(_P["not_null"]) + if col["name"] in fk_of: + marks.append(_P["fk"].format(target=fk_of[col["name"]])) + if text := (col.get("comment") or "").strip(): + marks.append((_P["col_system"] if system else _P["col_comment"]).format(text=text)) + line = _P["column"].format(name=col["name"], type=col.get("type") or "") + lines.append(line + ("，" + "，".join(marks) if marks else "")) + for group in meta.get("unique") or []: + if isinstance(group, list) and group: + lines.append(_P["unique"].format(cols="、".join(group))) + confirmed = _notes_lines(_confirmed_only(existing), meta=meta, system_notes=False) + if confirmed: + lines.append(_P["confirmed"]) + lines += confirmed + return "\n".join(lines) + + +def _batches(names: list[str], tables: Mapping[str, Any]) -> list[list[str]]: + """按表分批：每批不超过 MODEL_BATCH_TABLES 张、合计不超过 MODEL_BATCH_COLUMNS 列（一张表超了就自成一批）。""" + out: list[list[str]] = [] + batch: list[str] = [] + width = 0 + for name in names: + n = len((tables.get(name) or {}).get("columns") or []) + if batch and (len(batch) >= MODEL_BATCH_TABLES or width + n > MODEL_BATCH_COLUMNS): + out.append(batch) + batch, width = [], 0 + batch.append(name) + width += n + if batch: + out.append(batch) + return out + + +def _model_tables(raw: Any) -> list[dict[str, Any]] | None: + """模型的回复 → 每张表一项的列表。结构化输出给的对象、正文里的 JSON 都认；对不上格式返回 None。""" + if hasattr(raw, "model_dump"): + raw = raw.model_dump() + if isinstance(raw, str): + text = raw.strip() + start = min((i for i in (text.find("{"), text.find("[")) if i >= 0), default=-1) + if start < 0: + return None + end = text.rfind("}" if text[start] == "{" else "]") + try: + raw = json.loads(text[start:end + 1]) + except ValueError: + return None + if isinstance(raw, dict): + raw = raw.get("tables") + if not isinstance(raw, list): + return None + return [t for t in raw if isinstance(t, dict)] + + +async def _ask_model(model: Any, messages: list[Any]) -> list[dict[str, Any]]: + """问一批。先走结构化输出；模型不支持、或给回来的不是约定的结构，再退回纯文本、从正文里取 JSON。""" + try: + runnable = model.with_structured_output(CATALOG_DRAFT_PROMPT_SCHEMA) + except (AttributeError, NotImplementedError): + runnable = None + if runnable is not None: + parsed = _model_tables(await runnable.ainvoke(messages)) + if parsed is not None: + return parsed + from app.engine.state import message_text + + reply = await model.ainvoke(messages) + parsed = _model_tables(message_text(reply)) + if parsed is None: + raise ValueError("模型的回复不是约定的格式") + return parsed + + +def _clean_text(value: Any, limit: int = _TEXT_MAX) -> str | None: + if not isinstance(value, str): + return None + text = " ".join(value.split()) + return text if text and len(text) <= limit else None + + +def _from_model(entry: Mapping[str, Any], meta: Mapping[str, Any], *, system: bool) -> dict[str, Any]: + """模型给的一张表 → 目录项（llm / proposed）。不合规的值整项丢掉，不硬塞：列名对不上表结构的、 + 表类型和度量类型不在取值里的、业务主键里有不存在的列的。导入表格的源，说明已经覆盖的字段也丢掉。""" + columns = {c["name"]: c for c in meta.get("columns") or []} + notes: dict[str, Any] = {} + table_has_system_note = system and bool((meta.get("comment") or "").strip()) + for name in ("label", "grain"): + if name == "grain" and table_has_system_note: + continue + if (text := _clean_text(entry.get(name))) is not None: + notes[name] = make_item(text, "llm") + if entry.get("kind") in TABLE_KINDS: + notes["kind"] = make_item(entry["kind"], "llm") + keys = entry.get("keys") + if _str_list_ok(keys) and all(k in columns for k in keys): + notes["keys"] = make_item(list(dict.fromkeys(keys)), "llm") + date = entry.get("business_date") + if isinstance(date, dict) and date.get("column") in columns: + value = {"column": date["column"]} + for k in ("rule", "timezone"): + if (text := _clean_text(date.get(k))) is not None: + value[k] = text + notes["business_date"] = make_item(value, "llm") + out_cols: dict[str, dict[str, Any]] = {} + for col in entry.get("columns") or []: + if not isinstance(col, dict) or col.get("name") not in columns: + continue + name = col["name"] + has_system_note = system and bool((columns[name].get("comment") or "").strip()) + items: dict[str, Any] = {} + for field_name, limit in (("label", _TEXT_MAX), ("meaning", _TEXT_MAX), ("unit", 50)): + if has_system_note and field_name in ("meaning", "unit"): + continue + if (text := _clean_text(col.get(field_name), limit)) is not None: + items[field_name] = make_item(text, "llm") + if col.get("measure") in COLUMN_MEASURES: + items["measure"] = make_item(col["measure"], "llm") + if items: + out_cols[name] = items + if out_cols: + notes["columns"] = out_cols + return notes + + +async def draft_with_model(model: Any, source: Any, tables: list[str], *, + existing: Mapping[str, Mapping[str, Any]] | None = None) -> dict[str, TableDraft | str]: + """用模型起草这些表的中文名、粒度、表类型、业务主键、业务日期和每列的中文名、含义、单位、度量类型。 + + model 要有 with_structured_output(schema) 或 ainvoke(messages)（测试注入假模型）。按表分批（_batches）， + 最多 MODEL_CONCURRENCY 批同时发。返回 {表名: TableDraft 或失败原因}：一批失败只影响这一批的表， + 其余照常；失败的表不算 covered，已有的模型起草项不会因此被删。 + + 交给模型的只有表结构（列、类型、主键、外键、唯一约束、注释），不发任何数据值；existing 里已确认、 + 有确证的项作为「以此为准」的背景一起交过去。 + """ + from langchain_core.messages import HumanMessage, SystemMessage + + all_tables = (getattr(source, "schema_cache", None) or {}).get("tables") or {} + system = system_notes_source(source) + names = [t for t in tables if isinstance(all_tables.get(t), dict)] + results: dict[str, TableDraft | str] = {} + gate = asyncio.Semaphore(MODEL_CONCURRENCY) + + async def run(batch: list[str]) -> None: + described = "\n\n".join(_describe_for_model(n, all_tables[n], system=system, + existing=(existing or {}).get(n)) for n in batch) + messages = [SystemMessage(content=CATALOG_DRAFT_SYSTEM), + HumanMessage(content=CATALOG_DRAFT_PROMPT.format(n=len(batch), tables=described))] + try: + async with gate: + got = await asyncio.wait_for(_ask_model(model, messages), timeout=MODEL_TIMEOUT_S) + except asyncio.TimeoutError: + for n in batch: + results[n] = f"模型起草超时：{MODEL_TIMEOUT_S} 秒内没有收到回复" + return + except Exception as e: # noqa: BLE001 - 一批失败不影响其他批和其他来源 + from app.core.errors import describe_exception, first_line + + for n in batch: + results[n] = f"模型起草失败：{first_line(e) or describe_exception(e)}" + return + by_name = {str(t.get("table")): t for t in got} + for n in batch: + entry = by_name.get(n) + if entry is None: + results[n] = "模型的回复里没有这张表" + continue + results[n] = TableDraft(notes=_from_model(entry, all_tables[n], system=system), + covered=frozenset({"llm"})) + + await asyncio.gather(*(run(b) for b in _batches(names, all_tables))) + return {n: results[n] for n in names if n in results} + + +# ========================================================================== +# 起草并写入 +# ========================================================================== + + +@dataclass +class TableDraftResult: + """一张表这次起草的结果。error：这张表没起草（表结构里没有、写入一直冲突）；model_error：只是模型那部分失败。""" + + table: str + added: int = 0 + updated: int = 0 + removed: int = 0 + version: int = 0 + error: str | None = None + model_error: str | None = None + + +@dataclass +class DraftReport: + """一次起草的结果：每张表各一项，以及模型用没用上。""" + + tables: list[TableDraftResult] = field(default_factory=list) + #: 这次有没有调模型、调的是哪个 + model_used: bool = False + model: str = "" + #: 模型整体用不了的原因（没有接入等）；单张表的模型失败记在 TableDraftResult.model_error + model_error: str | None = None + + +async def draft_catalog(session: AsyncSession, source: Any, *, tables: list[str] | None = None, + use_model: bool = False, model: Any = None, actor: str | None = None) -> DraftReport: + """起草这些表的目录并写入，同步完成。 + + - source：手工源的 DataSource，或上传源绑定当前快照的视图（table_versions.resolve_source）；按 source.id 存。 + - tables 为空时按使用次数（table_usage）取前 DRAFT_DEFAULT_TABLES 张，次数相同按表结构里的顺序。 + - 每张表：数据库注释、外键约束、命名推断照做；use_model 时再加模型起草。model 可注入（测试用假模型）， + 不给就按设置解析（resolve_draft_model）；模型用不了、某一批失败，都不影响其余来源照常写入。 + - 并入已有目录按 merge_notes 的规则：人工确认、驳回过的项不动。内容没变不写、不升版本。 + - 写入撞上别人刚改过（乐观锁）就重读重并，最多三次。 + """ + cache = getattr(source, "schema_cache", None) or {} + all_tables = cache.get("tables") or {} + report = DraftReport() + if tables: + wanted = list(dict.fromkeys(str(t) for t in tables)) + else: + usage = await table_usage(session, source) + order = {name: i for i, name in enumerate(all_tables)} + wanted = sorted(all_tables, key=lambda n: (-usage.get(n, 0), order[n]))[:DRAFT_DEFAULT_TABLES] + picked: list[str] = [] + for name in wanted: + key = resolve_table_name(cache, name) + if key is None: + report.tables.append(TableDraftResult(table=name, error=f"表结构里没有 {name}，请先重新探查结构")) + elif key not in picked: + picked.append(key) + + llm: dict[str, TableDraft | str] = {} + if use_model and picked: + try: + if model is None: + model, report.model = await resolve_draft_model(session) + else: + report.model = str(getattr(model, "model_name", "") or "") + except CatalogModelUnavailable as e: + report.model_error = str(e) + else: + report.model_used = True + existing = await read_catalog(session, source.id) + llm = await draft_with_model(model, source, picked, existing={t: e.notes for t, e in existing.items()}) + + at = now_iso() + for table in picked: + result = TableDraftResult(table=table) + structure = draft_structure(source, table) + modeled = llm.get(table) + if isinstance(modeled, str): + result.model_error = modeled + for _attempt in range(3): + entry = await read_entry(session, source.id, table) + version = entry.version if entry else 0 + merged, stats = merge_notes(entry.notes if entry else {}, structure.notes, covered=structure.covered, + at=at) + if isinstance(modeled, TableDraft): + merged, more = merge_notes(merged, modeled.notes, covered=modeled.covered, at=at) + stats += more + result.added, result.updated, result.removed = stats.added, stats.updated, stats.removed + result.version = version + if entry is None and not merged: + break # 什么也没起草出来，不建空目录 + try: + written = await write_entry(session, source.id, table, merged, if_version=version, actor=actor) + except CatalogConflict: + continue + result.version = written.version + break + else: + result.error = "这张表的数据目录正被其他人修改，请稍后再起草" + result.added = result.updated = result.removed = 0 + report.tables.append(result) + return report + + +# ========================================================================== +# 渲染：交给模型看的文本 +# ========================================================================== + +#: proposed 的项在模型可见文本里的标记 +INFERRED_MARK = "（推断，未确认）" +KIND_LABEL = {"fact": "事实表", "dimension": "维度表", "snapshot": "快照表", "log": "日志表", "config": "配置表"} +MEASURE_LABEL = {"flow": "流量，可跨期加总", "stock": "存量，不能跨期加总", "ratio": "比率，不能直接加总", + "identifier": "标识", "status": "状态", "attribute": "属性"} +CARDINALITY_LABEL = {"many_to_one": "多对一", "one_to_one": "一对一", "one_to_many": "一对多"} +#: 单表目录的标题行 +NOTES_HEADING = "数据目录（标「推断，未确认」的项是推断，使用前请核实）：" + + +def _mark(item: Mapping[str, Any]) -> str: + return INFERRED_MARK if item.get("status") == "proposed" else "" + + +def _shown(item: Any, *, covered: bool) -> bool: + """这一项给不给模型看：被驳回的不给；系统说明已经覆盖的，只给人工确认过的。""" + if not isinstance(item, dict) or item.get("status") == "rejected": + return False + return not covered or item.get("status") == "confirmed" + + +def _date_text(value: Mapping[str, Any]) -> str: + parts = [str(value.get("column") or "")] + if value.get("rule"): + parts.append(f"规则：{value['rule']}") + if value.get("timezone"): + parts.append(f"时区：{value['timezone']}") + return "；".join(parts) + + +def _relation_text(rel: Mapping[str, Any]) -> str: + left = ", ".join(rel.get("columns") or []) + right = ", ".join(f"{rel.get('to_table')}.{c}" for c in rel.get("to_columns") or []) + extra = [] + if rel.get("cardinality") in CARDINALITY_LABEL: + extra.append(CARDINALITY_LABEL[rel["cardinality"]]) + coverage = rel.get("coverage") + if isinstance(coverage, (int, float)) and not isinstance(coverage, bool): + extra.append(f"覆盖率 {coverage:.0%}") + return f"{left} → {right}" + (f"（{'，'.join(extra)}）" if extra else "") + _mark(rel) + + +def _notes_lines(notes: Mapping[str, Any] | None, *, meta: Mapping[str, Any] | None, + system_notes: bool) -> list[str]: + """单表目录的正文行（不含标题），两格缩进起。""" + notes = notes or {} + meta = meta or {} + table_covered = system_notes and bool((meta.get("comment") or "").strip()) + lines: list[str] = [] + for name in TABLE_FIELDS: + item = notes.get(name) + if not _shown(item, covered=table_covered and name in ("grain", "description")): + continue + value = item["value"] + if name == "kind": + text = KIND_LABEL.get(value, str(value)) + elif name == "keys": + text = ", ".join(value) + elif name == "business_date": + text = _date_text(value) + else: + text = str(value) + lines.append(f" {TABLE_FIELD_LABEL[name]}：{text}{_mark(item)}") + + col_comment = {c.get("name"): bool((c.get("comment") or "").strip()) for c in meta.get("columns") or []} + order = [c.get("name") for c in meta.get("columns") or []] + columns = notes.get("columns") if isinstance(notes.get("columns"), dict) else {} + col_lines: list[str] = [] + for col in [c for c in order if c in columns] + [c for c in columns if c not in order]: + items = columns[col] if isinstance(columns[col], dict) else {} + covered = system_notes and col_comment.get(col, False) + parts: list[str] = [] + for name in COLUMN_FIELDS: + item = items.get(name) + if not _shown(item, covered=covered and name in ("meaning", "unit")): + continue + value = item["value"] + if name == "measure": + text = MEASURE_LABEL.get(value, str(value)) + elif name == "codes": + text = "、".join(f"{k}={v}" for k, v in value.items()) + else: + text = str(value) + # 第一段是中文名（没有中文名时是含义），不加前缀；其余写明是什么 + plain = not parts and name in ("label", "meaning") + parts.append(("" if plain else f"{COLUMN_FIELD_LABEL[name]}：") + text + _mark(item)) + if parts: + col_lines.append(f" {col}：{'；'.join(parts)}") + if col_lines: + lines.append(" 列：") + lines += col_lines + + relations = notes.get("relations") if isinstance(notes.get("relations"), list) else [] + rel_lines = [f" {_relation_text(r)}" for r in relations if isinstance(r, dict) and r.get("status") != "rejected"] + if rel_lines: + lines.append(" 关联关系：") + lines += rel_lines + return lines + + +def render_table_notes(notes: Mapping[str, Any] | None, *, meta: Mapping[str, Any] | None = None, + system_notes: bool = False) -> str: + """单表目录的完整版，给 db_schema 工具和助手挑表后补上下文用。没有可给模型看的内容时返回空串。 + + - 被驳回的项、关系不出现；proposed 的项标「推断，未确认」，有确证和人工确认的不标。 + - meta 是这张表在 schema_cache 里的结构：列按表结构的顺序列出。 + - system_notes（导入表格的源）：表和列已有系统生成的说明。表有说明时，粒度、说明只在人工确认过时追加； + 列有说明时，含义、单位同理；其余的项（中文名、表类型、业务主键、度量类型、码值、关系……）说明里不写， + 照常追加。 + """ + lines = _notes_lines(notes, meta=meta, system_notes=system_notes) + return "\n".join([NOTES_HEADING, *lines]) if lines else "" + + +def render_table_index(schema_cache: Mapping[str, Any] | None, notes_by_table: Mapping[str, Mapping[str, Any]], *, + tables: Iterable[str] | None = None) -> str: + """紧凑的表目录，每张表一行：表名｜中文名｜粒度。给助手按任务挑表用。 + + 表名用全名（模型照着写 SQL）；中文名或粒度是推断的，行尾标「推断，未确认」。没有目录的表只写表名。 + tables 给了就只列这些（按给的顺序），否则按表结构的顺序列全部。 + """ + all_tables = (schema_cache or {}).get("tables") or {} + lines = [] + for name in (list(tables) if tables is not None else list(all_tables)): + meta = all_tables.get(name) + if not isinstance(meta, dict): + continue + notes = notes_by_table.get(name) or {} + label = notes.get("label") if _shown(notes.get("label"), covered=False) else None + grain = notes.get("grain") if _shown(notes.get("grain"), covered=False) else None + parts = [str(meta.get("qualified") or name)] + if label: + parts.append(str(label["value"])) + if grain: + parts.append(str(grain["value"]) if label else f"粒度：{grain['value']}") + mark = INFERRED_MARK if any(i and i.get("status") == "proposed" for i in (label, grain)) else "" + lines.append("｜".join(parts) + mark) + return "\n".join(lines) + + +# ========================================================================== +# 关系图 +# ========================================================================== + +_FLIP = {"many_to_one": "one_to_many", "one_to_many": "many_to_one", "one_to_one": "one_to_one"} + + +@dataclass(frozen=True) +class JoinEdge: + """关系图里的一条有向边：从 from_table 的 from_columns 连到 to_table 的 to_columns。 + + 每条关系在图里有正反两条边；cardinality 是从 from_table 看过去的基数。 + """ + + from_table: str + from_columns: tuple[str, ...] + to_table: str + to_columns: tuple[str, ...] + relation_id: str + cardinality: str | None + source: str + status: str + + def reversed(self) -> "JoinEdge": + return JoinEdge(self.to_table, self.to_columns, self.from_table, self.from_columns, self.relation_id, + _FLIP.get(self.cardinality or ""), self.source, self.status) + + def condition(self) -> str: + """连接条件：visits.id = channel_visits.visit_id（多列用 AND 连接）。""" + return " AND ".join(f"{self.from_table}.{a} = {self.to_table}.{b}" + for a, b in zip(self.from_columns, self.to_columns)) + + +def relation_graph(notes_by_table: Mapping[str, Mapping[str, Any]], *, + schema_cache: Mapping[str, Any] | None = None) -> dict[str, list[JoinEdge]]: + """邻接表 {表: [从这张表出发的边]}，每条关系正反各一条边；被驳回的关系不进图，自己连自己的也不进。 + + schema_cache 给了的话，没起草过目录的关系按表结构现推（外键约束 + 命名推断）补进来：助手挑表时不必等 + 每张表都起草过。目录里已有的关系（含驳回的）以目录为准。 + """ + rels: dict[str, tuple[str, Mapping[str, Any]]] = {} + rejected: set[str] = set() + for table, notes in notes_by_table.items(): + relations = (notes or {}).get("relations") + for rel in relations if isinstance(relations, list) else []: + if not isinstance(rel, dict) or not rel.get("id"): + continue + if rel.get("status") == "rejected": + rejected.add(rel["id"]) + else: + rels.setdefault(rel["id"], (table, rel)) + if schema_cache: + all_tables = schema_cache.get("tables") or {} + index = _name_index(all_tables) + for table in all_tables: + inferred = (fk_relations(schema_cache, table) or []) + infer_name_relations(schema_cache, table, + _index=index) + for rel in inferred: + if rel["id"] not in rels and rel["id"] not in rejected: + rels[rel["id"]] = (table, rel) + graph: dict[str, list[JoinEdge]] = {} + for rid, (table, rel) in rels.items(): + if rel.get("to_table") == table: + continue + edge = JoinEdge(table, tuple(rel.get("columns") or ()), str(rel.get("to_table")), + tuple(rel.get("to_columns") or ()), rid, rel.get("cardinality"), str(rel.get("source")), + str(rel.get("status"))) + graph.setdefault(edge.from_table, []).append(edge) + graph.setdefault(edge.to_table, []).append(edge.reversed()) + return graph + + +def join_paths(graph: Mapping[str, list[JoinEdge]], start: str, end: str, *, + max_hops: int = 2) -> list[tuple[JoinEdge, ...]]: + """两张表之间不超过 max_hops 跳（最多 2）的连接路径，短的在前，同样长的推断边少的在前。""" + paths: list[tuple[JoinEdge, ...]] = [(e,) for e in graph.get(start, []) if e.to_table == end] + if max_hops >= 2: + for first in graph.get(start, []): + mid = first.to_table + if mid in (start, end): + continue + paths += [(first, second) for second in graph.get(mid, []) if second.to_table == end] + return sorted(paths, key=lambda p: (len(p), sum(e.status == "proposed" for e in p))) + + +# ========================================================================== +# 使用次数 +# ========================================================================== + +#: 数据源查询工具存查询快照用的工件类型（tools/datasource.py） +QUERY_SNAPSHOT_KIND = "query_snapshot" + +#: 老的查询快照引用（meta 里没记源和表名）读过一次的结果：{工件 id: (源名, 表名)}。工件内容按哈希寻址、 +#: 永不改变，缓存不会过期 +_LEGACY_QUERY_TABLES: dict[str, tuple[str | None, tuple[str, ...]]] = {} + + +def _load_artifact(artifact_id: str) -> Any: + from app.core.artifact_store import load + + return load(artifact_id) + + +def _legacy_query(artifact_id: str) -> tuple[str | None, tuple[str, ...]]: + if artifact_id not in _LEGACY_QUERY_TABLES: + from app.engine.evidence import sql_tables + + content = _load_artifact(artifact_id) + if isinstance(content, dict): + found = (content.get("source"), tuple(sql_tables(str(content.get("sql") or "")))) + else: + found = (None, ()) + _LEGACY_QUERY_TABLES[artifact_id] = found + return _LEGACY_QUERY_TABLES[artifact_id] + + +def query_snapshot_meta(source: Any, sql: str) -> dict[str, Any]: + """查询快照工件引用的 meta：源 id、源名、SQL 里的表名。数据源查询工具存快照时写进去，table_usage 据此计数。 + + 表名的取法和证据台账一样（evidence.sql_tables：FROM / JOIN 后面的表，去掉注释和字符串）。meta 只在 + 工件引用表里，不进工件内容：查询快照的内容哈希不变。 + """ + from app.engine.evidence import sql_tables + + return {"source_id": getattr(source, "id", None), "source": getattr(source, "name", None), + "tables": sql_tables(sql or "")} + + +async def table_usage(session: AsyncSession, source: Any) -> dict[str, int]: + """每张表被运行查询过的次数：{schema_cache 的表名: 次数}，没被查过的表不出现。 + + 取法：数据源查询工具每次成功查询都存一份查询快照工件（tools/datasource.py），工件引用表 artifacts 里 + 按 (内容哈希, 运行) 各记一行。数「属于某次运行（runs 里有这次运行）、属于这个源、SQL 里用到了这张表」的 + 行：同一次运行里结果一模一样的重复查询只算一次，工具库里的试用（不属于任何运行）不算。 + + 表名在查询当时就记进了引用的 meta（query_snapshot_meta），这里只读 artifacts 一张表；升级前的老引用 + 没有 meta，退回读工件内容里的源名和 SQL，读过的按工件 id 缓存（内容永不改变）。SQL 里写的表名去掉 + schema 前缀后按 name_key（只把 ASCII 字母转小写，和 SQLite 比较标识符的口径一致）对到表结构上，对不上的不计。 + """ + rows = (await session.execute( + select(Artifact.id, Artifact.meta).join(Run, Run.id == Artifact.run_id) + .where(Artifact.kind == QUERY_SNAPSHOT_KIND) + )).all() + tables = (getattr(source, "schema_cache", None) or {}).get("tables") or {} + lookup: dict[str, str] = {} + for name in tables: + lookup.setdefault(name_key(name), name) + counts: Counter[str] = Counter() + for artifact_id, meta in rows: + meta = meta if isinstance(meta, dict) else {} + if isinstance(meta.get("tables"), list): + source_id, source_name, names = meta.get("source_id"), meta.get("source"), meta["tables"] + else: + source_id = None + source_name, names = _legacy_query(str(artifact_id)) + if source_id is not None: + if source_id != source.id: + continue + elif source_name != source.name: + continue + hit = {lookup.get(name_key(str(n).rsplit(".", 1)[-1])) for n in names} + counts.update(t for t in hit if t is not None) + return dict(counts) + + +# ========================================================================== +# 冻结进证据 +# ========================================================================== + + +def frozen_catalog(entries: Mapping[str, CatalogEntry], tables: Iterable[str]) -> dict[str, dict[str, Any]]: + """要冻结进表结构快照的目录：{表名: {"version": 版本, "notes": 去掉驳回项的目录}}。 + + 只收表结构快照里有的表（tables）；去掉驳回项以后什么都不剩的表不收。一张也没有时返回空 dict， + 调用方据此不加这个键：没有目录的源，快照内容和以前一字不差。目录不变（版本和内容都不变）则返回值 + 不变，快照哈希也就不变。 + """ + wanted = set(tables) + out: dict[str, dict[str, Any]] = {} + for name in sorted(entries): + if name not in wanted: + continue + notes = visible_notes(entries[name].notes) + if notes: + out[name] = {"version": entries[name].version, "notes": notes} + return out + + __all__ = [ - "CARDINALITIES", "COLUMN_FIELDS", "COLUMN_FIELD_LABEL", "COLUMN_MEASURES", "CatalogConflict", "CatalogEntry", - "CatalogInvalid", "ITEM_SOURCES", "ITEM_STATUSES", "TABLE_FIELDS", "TABLE_FIELD_LABEL", "TABLE_KINDS", - "initial_status", "make_item", "now_iso", "read_catalog", "read_entry", "relation_id", "same_notes", - "validate_notes", "write_entry", + "CARDINALITIES", "CARDINALITY_LABEL", "CATALOG_DRAFT_PROMPT", "CATALOG_DRAFT_PROMPT_SCHEMA", + "CATALOG_DRAFT_SYSTEM", "COLUMN_FIELDS", "COLUMN_FIELD_LABEL", "COLUMN_MEASURES", "CatalogConflict", + "CatalogEntry", "CatalogInvalid", "CatalogModelUnavailable", "CatalogPathError", "DRAFT_DEFAULT_TABLES", + "DraftReport", "INFERRED_MARK", "ITEM_SOURCES", "ITEM_STATUSES", "JoinEdge", "KIND_LABEL", "MEASURE_LABEL", + "MergeStats", "QUERY_SNAPSHOT_KIND", "REVIEW_ACTIONS", "TABLE_FIELDS", "TABLE_FIELD_LABEL", "TABLE_KINDS", + "TableDraft", "TableDraftResult", "apply_human_edit", "describe_slot", "draft_catalog", "draft_structure", + "draft_with_model", "fk_relations", "frozen_catalog", "infer_name_relations", "initial_status", "join_paths", + "make_item", "merge_notes", "now_iso", "parse_path", "query_snapshot_meta", "read_catalog", "read_entry", + "relation_graph", "relation_id", "render_table_index", "render_table_notes", "resolve_draft_model", + "resolve_table_name", "review_entry", "review_item", "same_notes", "save_human_edit", "status_counts", + "system_notes_source", "table_usage", "validate_notes", "visible_notes", "write_entry", ] diff --git a/backend/tests/test_catalog_draft.py b/backend/tests/test_catalog_draft.py new file mode 100644 index 00000000..34d4f7bc --- /dev/null +++ b/backend/tests/test_catalog_draft.py @@ -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) diff --git a/backend/tests/test_catalog_merge.py b/backend/tests/test_catalog_merge.py new file mode 100644 index 00000000..8427505e --- /dev/null +++ b/backend/tests/test_catalog_merge.py @@ -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" diff --git a/backend/tests/test_catalog_render.py b/backend/tests/test_catalog_render.py new file mode 100644 index 00000000..ce7d3091 --- /dev/null +++ b/backend/tests/test_catalog_render.py @@ -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} diff --git a/backend/tests/test_catalog_store.py b/backend/tests/test_catalog_store.py index 751c4e34..000a78ef 100644 --- a/backend/tests/test_catalog_store.py +++ b/backend/tests/test_catalog_store.py @@ -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"),