mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 00:41:59 +00:00
ThearchyHelios
975dde1146
- 均分：超过 40 列的表，列分段数是列数除以 40 向上取整，各段列数平均（相差不超过 1），不再留下 2 到 8 列的尾段（实测 会思考的模型常对这种小段只给空壳）；表级那段、窄表拼段、parts_total 不变。 - 拆不开的段（不到 10 列）只给空壳：再问一次，结构化输出换纯文本，已经走纯文本的原样再问，仍是空壳才记没写出来。 - 超时当作这一段太大：不再原样重试同样大小的段，按额度用完的办法拆小再问；网络错误、429、5xx 仍原样退避重试。 - 段大小随模型调整：一次起草里额度用完或超时累计 2 次，之后新开始的段列数上限减半（40 → 20 → 10，最低 10），新开始 的段超过上限就先按上限均分开（多表的段按表重新拼）；任务记当前的上限 part_columns。 - 封顶：一段（连同它拆出来的、补问的、重试的）最多调用 8 次，到了次数还没写完的记没写出来、原因写清。止损照旧。 - 提醒：最近 6 次调用有 4 次以上没写出来（额度用完、超时、空壳、出错），任务的 notice 写一句提醒（次数、原因，建议在 「设置 → 偏好设置」的「起草模型」换一个模型、减少表或停止）；之后最近 6 次里少于 2 次就清掉。 - 用量：进度回调 usage 换成 call_done（CallReport：用量、没写出来的原因种类、当前的提醒和列数上限），任务的用量加 timeouts、failed_calls；超时的调用 token 照实记 0。 - 留空的列：请模型写了、模型也列出了，中文名和含义却都没给（或者给的都没通过校验）的列记在 ModelOutcome.blank，逐表 记 blank_columns 和最多 5 个列名 blank_sample，任务上是合计，目录清单的 last_draft 也带列数；表的状态不因此变。已确认 的、库注释已经说清的列不算。补问时把漏的列列出来、字段却留空的，算留空，不再记成「漏了这些列」。 - 公平排队：宽表的表级那段写完，它的列分段排到已经排着的段后面（先来先服务），不再插到队首让窄表一直等。 - 原因：几段因为同一个原因没写出来时说「这几段」，同一段拆出来的几块仍是「这一段」；只有真拆小过才补「已拆小重试仍 没写出」（截断后只补剩下的不算）；每段写上列的范围，例如「第 5 段（列名 到 列名，共 33 列）」，窄表写出是哪几列。 - 库：catalog_draft_jobs 加 timeouts、failed_calls、part_columns、notice，catalog_draft_tasks 加 blank_columns、 blank_sample，旧库启动时补列（存量回填 0 和空）。假模型加 hang（头几次调用一直不回，用来模拟超时）。
518 lines
28 KiB
Python
518 lines
28 KiB
Python
"""数据目录的起草：数据库注释、外键约束、命名推断、模型起草，以及并入已有目录。
|
||
|
||
这一阶段起草不发数据库查询（数据剖析归阶段 2），只看探查缓存里的结构。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import uuid
|
||
from types import SimpleNamespace
|
||
|
||
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
|
||
from tests.fixtures.catalog.fake_model import FakeModel
|
||
from tests.fixtures.sources import drop_source
|
||
|
||
|
||
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("comment, role", [
|
||
("实收金额", "label"),
|
||
("Paid amount", "label"), # 带空格的英文短语照旧当名字
|
||
("状态：1 表示有效，0 表示作废", "note"),
|
||
("paid_amt", None), ("PAID-AMT", None), ("Paid Amt", None), # 和列名相同（忽略大小写、下划线、连字符、空白）
|
||
("member_level_code", None), ("ord.status", None), ("GMV", None), ("v2", None), # 标识符写法
|
||
("member-level", "label"), # 带连字符、和名字不同：不是标识符写法
|
||
])
|
||
def test_comment_role_by_text_and_name(comment, role):
|
||
assert catalog.comment_role(comment, "paid_amt") == role
|
||
|
||
|
||
def test_a_comment_that_is_just_the_name_or_an_identifier_is_not_a_label():
|
||
"""注释就是表名本身（列注释就是列名）、或者是标识符写法：不当中文名，也不当说明（不然它比模型起草可信，模型起草的
|
||
中文名、含义永远进不了目录）；交给模型的表结构里照旧有这条注释。"""
|
||
src = _source({"member_card": _table(
|
||
_col("id"), _col("card_no", "TEXT", comment="CARD-NO"), _col("open_dt", "TEXT", comment="open_date"),
|
||
_col("bal", "REAL", comment="卡内余额"), comment="MEMBER CARD")})
|
||
draft = catalog.draft_structure(src, "member_card")
|
||
assert "label" not in draft.notes and "description" not in draft.notes
|
||
assert draft.notes["columns"] == {"bal": {"label": make_item("卡内余额", "comment")}}
|
||
assert "comment" in draft.covered
|
||
# schema 前缀不算：注释写的是不带前缀的表名，同样不当中文名
|
||
src = _source({"core.member_card": _table(_col("id"), comment="MEMBER_CARD")})
|
||
assert "label" not in catalog.draft_structure(src, "core.member_card").notes
|
||
|
||
|
||
async def test_a_redraft_drops_an_old_label_that_was_just_the_name_and_takes_the_model_one():
|
||
"""以前按老规矩把「注释就是表名」记成了中文名（注释的级别高于模型起草）：以后起草时这条注释不再当中文名，旧的那一项
|
||
随这一轮的注释一起删掉，模型起草的中文名就能进目录。已有目录不另做迁移，起草到这张表时才变。"""
|
||
tables = {"member_card": _table(_col("id"), _col("bal", "REAL"), comment="MEMBER_CARD", fks=[])}
|
||
row = await _db_source(tables)
|
||
async with SessionLocal() as session:
|
||
await catalog.write_entry(session, row.id, "member_card", {"label": make_item("MEMBER_CARD", "comment")},
|
||
if_version=0, actor=None)
|
||
report = await catalog.draft_catalog(session, row, tables=["member_card"], use_model=True,
|
||
model=FakeModel({"member_card": {"label": "会员卡"}}))
|
||
entry = await catalog.read_entry(session, row.id, "member_card")
|
||
assert entry.notes["label"]["value"] == "会员卡" and entry.notes["label"]["source"] == "llm"
|
||
assert (report.tables[0].added, report.tables[0].removed) == (1, 1)
|
||
|
||
|
||
@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_falls_back_to_a_table_sharing_the_prefix():
|
||
"""库里没有 tags，只有带前缀的 store_tags 和 menu_tag：和引用它的表共用 store 前缀的那张才是它要指的。
|
||
以前这种命名推不出关系，SQL 检查跟着误报「对不上任何关系」。"""
|
||
tables = {
|
||
"store_tag_links": _table(_col("id"), _col("tagId"), _col("storeID")),
|
||
"store_tags": _table(_col("id")),
|
||
"menu_tag": _table(_col("id")),
|
||
"store": _table(_col("id")),
|
||
}
|
||
assert _infer(tables, "store_tag_links") == {
|
||
(("tagId",), "store_tags", ("id",)), (("storeID",), "store", ("id",))}
|
||
|
||
|
||
@pytest.mark.parametrize("tables, why", [
|
||
# 直接对得上就用直接的，不去找带前缀的
|
||
({"shop_category_links": _table(_col("id"), _col("category_id")), "categories": _table(_col("id")),
|
||
"shop_categories": _table(_col("id"))}, {(("category_id",), "categories", ("id",))}),
|
||
# 引用它的表没有可共用的前缀：带前缀的表不止一张时照旧不推
|
||
({"links": _table(_col("id"), _col("tag_id")), "store_tags": _table(_col("id")),
|
||
"menu_tags": _table(_col("id"))}, set()),
|
||
])
|
||
def test_prefix_fallback_only_when_direct_match_is_missing(tables, why):
|
||
assert _infer(tables, next(iter(tables))) == 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"
|
||
|
||
|
||
# ---------------------------------------------------------------- 模型起草
|
||
|
||
|
||
_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_without_structured_output_falls_back_to_text():
|
||
"""有的网关不支持工具调用：结构化输出一调就报错，退回纯文本、从正文里取 JSON。"""
|
||
from langchain_core.messages import AIMessage
|
||
|
||
class TextOnly:
|
||
def with_structured_output(self, schema, **_):
|
||
raise NotImplementedError
|
||
|
||
async def ainvoke(self, messages):
|
||
return AIMessage(content='草稿如下：{"tables": [{"table": "visits", "label": "入园记录"}]}')
|
||
|
||
got = await catalog.draft_with_model(TextOnly(), _source(_scenic_tables()), ["visits"])
|
||
assert got["visits"].notes == {"label": make_item("入园记录", "llm")}
|
||
|
||
|
||
# ---- 实测见过的几种退化回复（A2）：一项都没写进去时不能说「没有变化」
|
||
|
||
|
||
async def test_model_single_table_object_without_wrapper_is_recognized():
|
||
"""纯文本兜底时只回了单个表对象，没有 tables 外壳：照样认出来。"""
|
||
model = FakeModel({"visits": {"label": "入园记录"}}, structured="error", text="single")
|
||
got = await catalog.draft_with_model(model, _source(_scenic_tables()), ["visits"])
|
||
assert got["visits"].notes == {"label": make_item("入园记录", "llm")}
|
||
|
||
|
||
async def test_hollow_structured_output_retries_with_text():
|
||
"""结构化输出只给了 {"tables":[{"table":"visits"}]} 这样的空壳、这一段又拆不开（不到 10 列）：改走纯文本再问一次，
|
||
用纯文本给的内容。纯文本只给了表级的项、一列都没列出来：再补问一次没给的列（只问缺的，data/catalog_draft.py）；
|
||
补问后仍没给，这几列算没写出来，表是只写出一部分（旧的模型起草项不删）。"""
|
||
model = FakeModel({"visits": {"label": "入园记录", "kind": "fact"}}, structured="hollow", text="tables",
|
||
list_all=False)
|
||
got = await catalog.draft_with_model(model, _source(_scenic_tables()), ["visits"])
|
||
assert model.modes == ["structured", "text", "text"]
|
||
assert got["visits"].notes == {"label": make_item("入园记录", "llm"), "kind": make_item("fact", "llm")}
|
||
assert got["visits"].covered == frozenset() and "有 5 列（id 到 status）没写出来" in got["visits"].error
|
||
|
||
|
||
async def test_hollow_structured_then_single_text_object():
|
||
model = FakeModel({"visits": {"label": "入园记录"}}, structured="hollow", text="single")
|
||
got = await catalog.draft_with_model(model, _source(_scenic_tables()), ["visits"])
|
||
assert got["visits"].notes == {"label": make_item("入园记录", "llm")}
|
||
|
||
|
||
async def test_table_with_no_usable_content_is_a_model_error():
|
||
"""表名对上了、一个字段都没有（两条路都是空壳）：这张表记模型失败，不算起草成功。"""
|
||
model = FakeModel({"visits": {"label": "入园记录"}}, structured="hollow", text="hollow")
|
||
got = await catalog.draft_with_model(model, _source(_scenic_tables()), ["visits"])
|
||
assert got["visits"] == catalog.MODEL_EMPTY_REASON
|
||
# 给了字段、但没有一项合规（列名对不上、取值不在范围里）同样算没有可用内容
|
||
junk = FakeModel({"visits": {"kind": "planet", "columns": [{"name": "ghost", "label": "不存在的列"}]}})
|
||
got = await catalog.draft_with_model(junk, _source(_scenic_tables()), ["visits"])
|
||
assert got["visits"] == catalog.MODEL_EMPTY_REASON
|
||
|
||
|
||
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] # 系统说明交给模型作背景
|
||
|
||
|
||
# ---------------------------------------------------------------- 起草并写入
|
||
|
||
|
||
_CREATED: list[str] = []
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
async def _drop_created_sources():
|
||
yield
|
||
while _CREATED:
|
||
await drop_source(_CREATED.pop())
|
||
|
||
|
||
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()
|
||
_CREATED.append(row.id)
|
||
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_empty_model_reply_is_not_no_change():
|
||
"""模型一项可用内容都没给：这张表记 model_error，没有 model_items，界面据此不说「没有变化」。
|
||
模型给了内容、只是和现有目录一致：没有 model_error，model_items 是模型给的项数。"""
|
||
row = await _db_source(_scenic_tables())
|
||
hollow = FakeModel({"visits": {"label": "入园记录"}}, structured="hollow", text="hollow")
|
||
async with SessionLocal() as session:
|
||
report = await catalog.draft_catalog(session, row, tables=["visits"], use_model=True, model=hollow)
|
||
result = report.tables[0]
|
||
assert result.model_error == catalog.MODEL_EMPTY_REASON and result.model_items is None
|
||
assert (result.added, result.updated, result.removed) == (0, 0, 0)
|
||
|
||
model = FakeModel({"visits": {"label": "入园记录", "kind": "fact"}})
|
||
async with SessionLocal() as session:
|
||
first = await catalog.draft_catalog(session, row, tables=["visits"], use_model=True, model=model)
|
||
again = await catalog.draft_catalog(session, row, tables=["visits"], use_model=True, model=model)
|
||
assert first.tables[0].model_items == 2 and first.tables[0].added == 2
|
||
assert again.tables[0].model_error is None and again.tables[0].model_items == 2
|
||
assert (again.tables[0].added, again.tables[0].updated, again.tables[0].removed) == (0, 0, 0)
|
||
# 不用模型时没有这个数
|
||
async with SessionLocal() as session:
|
||
plain = await catalog.draft_catalog(session, row, tables=["visits"], use_model=False)
|
||
assert plain.tables[0].model_items is None
|
||
|
||
|
||
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)
|