数据目录 F1 一致性:日期列的判定抽成 catalog.is_date_column,数据剖析(唯一日期列提议业务日期)和 SQL 检查(按业务日期以外的时间列统计)共用,只有时刻的 TIME、整数存的时间戳两边都不算;表名匹配统一按 name_key(只把 ASCII 字母转小写),resolve_table_name 和 SQL 检查里认表名不再按 Unicode 转小写,和使用次数、影响面同一个口径;切词合并成 catalog.name_words,剖析不再另写一份;更新过时的说明(错误级检查已经上线、剖析预算的叫法跟设置界面);码值比较不分大小写保留,注释写明是照顾 MySQL 默认排序规则、宁可漏报

This commit is contained in:
ThearchyHelios
2026-10-04 01:17:29 +08:00
parent f5e8a436f1
commit d805f64195
4 changed files with 116 additions and 56 deletions
+56 -24
View File
@@ -1,4 +1,4 @@
"""业务数据目录:每张表一份业务说明,给助手建图、db_schema 工具和后续的检查用。
"""业务数据目录:每张表一份业务说明,给助手建图、db_schema 工具和 SQL 检查(data/sqlcheck.py)用。
**为什么要有它。** 接进来的业务库往往一个注释都没有:几百张表、上千个字段,助手只看得到名字,模型就会
照着名字编字段、猜关联。目录把「一行代表什么、业务主键是哪几列、业务日期按哪列算、金额是什么单位、
@@ -23,12 +23,14 @@ notes 结构(每个「项」都是 {value, source, status},可选 note、upd
来源(source):comment 数据库注释、fk 外键约束、name 命名推断、profile 数据剖析(data/catalog_profile.py)、
llm 模型起草、human 人工填写。状态(status):proposed 推断(命名推断、模型起草、数据库注释)、verified 有确证
(外键约束;数据剖析核实了覆盖率和父键唯一的关系也是,见 profile_relation_holds)、confirmed 人工确认、
rejected 人工驳回。只有 confirmed 和 verified
将来会触发错误级检查;给模型看时 proposed 的项标「推断,未确认」;rejected 的项留着(防止下次起草又提出
来),但模型和检查都看不到。
rejected 人工驳回。SQL 检查只在依据全是 confirmed、verified
时报错误级(data/sqlcheck.py),依据里有 proposed 就降为提示;给模型看时 proposed 的项标「推断,未确认」;
rejected 的项留着(防止下次起草又提出来),但模型和检查都看不到。
本模块分几块:存取(带乐观锁)、结构校验、合并与审阅、起草、渲染、关系图、使用次数、冻结。存取以外的
函数尽量是纯函数,接口层、db_schema 工具和后续阶段(1B 助手挑表、2 数据剖析、4 维护闭环)共用。
本模块分几块:存取(带乐观锁)、结构校验、合并与审阅、对话提案、起草、渲染、关系图、使用次数、冻结。存取
以外的函数尽量是纯函数,接口层、db_schema 工具、助手挑表(api/copilot_context.py)、数据剖析
(data/catalog_profile.py)和 SQL 检查(data/sqlcheck.py)共用;几处两边都要用的判断也放在这里只写一份:
表名的匹配(resolve_table_name,按 name_key)、名字切词(name_words)、日期列的判定(is_date_column)。
"""
from __future__ import annotations
@@ -535,7 +537,7 @@ def _same_core(a: Mapping[str, Any], b: Mapping[str, Any]) -> bool:
def visible_notes(notes: Mapping[str, Any] | None) -> dict[str, Any]:
"""去掉被驳回的项和关系。给模型看、冻结进证据、将来的检查都只用这一份。"""
"""去掉被驳回的项和关系。给模型看、冻结进证据、SQL 检查(data/sqlcheck.py)都只用这一份。"""
return _assemble({k: v for k, v in _slots(notes).items() if v.get("status") != "rejected"})
@@ -1128,15 +1130,20 @@ def system_notes_source(source: Any) -> bool:
def resolve_table_name(schema_cache: Mapping[str, Any] | None, name: str) -> str | None:
"""在表结构里找表,返回 schema_cache["tables"] 的键。全名、只有表名、大小写不一致都认。"""
"""在表结构里找表,返回 schema_cache["tables"] 的键。全名、只有表名、大小写不一致都认。
大小写按 name_key 比:只把 ASCII 字母转小写,和 SQLite 比较标识符、使用次数(table_usage)、影响面
(catalog_impact)的口径一致。以前这里按 Unicode 转小写,表名里有非 ASCII 大写字母时(「CAFÉ_ORDERS」)
这里认得、那两处认不得,同一张表在目录页和影响面上对不上。
"""
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)
key = name_key(short)
return next((n for n in tables if name_key(n) == key), None)
#: 数据库注释多长、带不带句读,决定它是「名字」还是「解释」
@@ -1202,11 +1209,37 @@ _CAMEL_ID = re.compile(r"^(?P<stem>.*[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)。"""
def name_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)]
#: 名字像日期、时间的列,切词后的最后一个词:visit_time、ordered_at、joined_on、createdAt、biz_dt、load_ts……
_DATE_TAILS = frozenset({"at", "on", "date", "time", "day", "dt", "ts", "timestamp", "datetime"})
#: 单独做列名也算日期的词。at、on 单独做列名不是日期
_DATE_WORDS = frozenset({"date", "time", "day", "dt", "ts", "timestamp", "datetime"})
_TEXT_TYPES = ("CHAR", "TEXT", "STRING", "CLOB")
def is_date_column(name: str, type_name: str | None) -> bool:
"""日期类列:类型是日期、时间戳(DATE、DATETIME、TIMESTAMP……);或者类型是文字(或者不知道类型)、名字像日期
(SQLite 的日期常存成文字)。只有时刻的 TIME、整数存的时间戳不算:取值不是日期。
数据剖析(表里只有一个日期类列时提议它作业务日期)和 SQL 检查(按业务日期以外的时间列统计)共用这一个判断。
以前两边各写一套:剖析认的「唯一日期列」,SQL 检查可能认为还有别的时间列,反过来也一样,两边对不上。
"""
t = (type_name or "").upper()
if "DATE" in t or "TIMESTAMP" in t:
return True
if t and not any(k in t for k in _TEXT_TYPES):
return False
words = name_words(name)
return bool(words) and words[-1] in _DATE_TAILS and (len(words) > 1 or words[0] in _DATE_WORDS)
def _singulars(word: str) -> set[str]:
"""一个词可能的单数形式(含它自己)。两边都取这个集合、有交集就算同一个词:categories 和 category、
courses 和 course、status 和 statuses 都对得上,不靠一套「复数变单数」的规则硬猜。"""
@@ -1224,7 +1257,7 @@ 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)
words = name_words(name)
if not words:
continue
prefix = "".join(words[:-1])
@@ -1277,7 +1310,7 @@ def infer_name_relations(schema_cache: Mapping[str, Any] | None, table: str, *,
stem = _id_stem(name)
if stem is None or name.lower() in constrained:
continue
words = _words(stem)
words = name_words(stem)
if not words or len("".join(words)) < 2:
continue
prefix = "".join(words[:-1])
@@ -1287,7 +1320,7 @@ def infer_name_relations(schema_cache: Mapping[str, Any] | None, table: str, *,
if not candidates:
# 直接对不上:库里的表常带业务前缀(store_tags),列名却只写词根(tagId)。
# 只认和引用它的表共用前缀的那张,从最长的共用前缀试起;仍然不止一张就不推
own = _words(table)
own = name_words(table)
for k in range(len(own) - 1, 0, -1):
for form in _singulars(words[-1]):
candidates |= index.get(("".join(own[:k]) + prefix, form), set())
@@ -2148,14 +2181,13 @@ __all__ = [
"CatalogEntry", "CatalogInvalid", "CatalogModelUnavailable", "CatalogPathError", "DRAFT_DEFAULT_TABLES",
"DraftReport", "INFERRED_MARK", "ITEM_SOURCES", "ITEM_STATUSES", "JoinEdge", "KIND_LABEL", "MEASURE_LABEL",
"MergeStats", "NEW_RELATION_PATH", "PATCH_MAX_CHANGES", "PROFILE_NOTE_PREFIX", "PROFILE_VERIFY_COVERAGE",
"PatchChange", "PatchPlan",
"QUERY_SNAPSHOT_KIND", "REVIEW_ACTIONS", "TABLE_FIELDS", "TABLE_FIELD_LABEL", "TABLE_KINDS", "TableDraft",
"TableDraftResult", "UI_KIND_LABEL", "UI_MEASURE_LABEL", "apply_human_edit", "apply_patch", "codes_complete",
"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", "plan_patch",
"profile_relation_holds", "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", "save_patch", "status_counts", "structure_relation_ids",
"PatchChange", "PatchPlan", "QUERY_SNAPSHOT_KIND", "REVIEW_ACTIONS", "TABLE_FIELDS", "TABLE_FIELD_LABEL",
"TABLE_KINDS", "TableDraft", "TableDraftResult", "UI_KIND_LABEL", "UI_MEASURE_LABEL", "apply_human_edit",
"apply_patch", "codes_complete", "describe_slot", "draft_catalog", "draft_structure", "draft_with_model",
"fk_relations", "frozen_catalog", "infer_name_relations", "initial_status", "is_date_column", "join_paths",
"make_item", "merge_notes", "name_words", "now_iso", "parse_path", "plan_patch", "profile_relation_holds",
"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",
"same_profile_note", "save_human_edit", "save_patch", "status_counts", "structure_relation_ids",
"system_notes_source", "table_usage", "unreviewed_status", "validate_notes", "visible_notes", "write_entry",
]
+5 -18
View File
@@ -67,7 +67,8 @@ class ProfileSettings:
return asdict(self)
#: 数值项:(中文叫法, 下限, 上限, 是否必须是整数)。叫法和将来设置界面上的标签一致,报错里只写它
#: 数值项:(中文叫法, 下限, 上限, 是否必须是整数)。叫法和数据源设置里剖析预算的标签一致(frontend/src/lib/terms.ts),
#: 报错里只写它
_NUMBER_FIELDS: dict[str, tuple[str, float, float, bool]] = {
"max_queries": ("查询次数上限", 1, 500, True),
"query_timeout_s": ("单条查询时限(秒)", 1, 60, False),
@@ -770,16 +771,8 @@ _CODE_WORDS = frozenset({"status", "state", "type", "kind", "category", "channel
#: 以这些词结尾的列不是码值:标识、名称、说明、时间、金额、数量(channel_id、level_name、status_time……)
_NOT_CODE_TAILS = frozenset({"id", "no", "num", "number", "name", "desc", "description", "text", "remark", "note",
"at", "date", "time", "on", "amount", "price", "count", "qty", "rate", "url"})
#: 文字类型的列名以这些词结尾时,按日期类列看待(SQLite 里日期都存成文字:visit_time、ordered_at、joined_on)
_DATE_TAILS = frozenset({"at", "date", "time", "on", "dt", "day"})
#: 像日期的取值:2026-07-01、2026/07/01,后面可以跟时间
_DATE_TEXT = re.compile(r"^\d{4}[-/]\d{2}[-/]\d{2}")
_WORDS = re.compile(r"[A-Z]+(?=[A-Z][a-z])|[A-Z]?[a-z]+|[A-Z]+|\d+")
def _words(name: str) -> list[str]:
"""列名切成小写的词:下划线和驼峰都是分隔(和目录的命名推断同一个切法)。"""
return [w.lower() for part in name.split("_") for w in _WORDS.findall(part)]
def _code_type_ok(type_name: str | None) -> bool:
@@ -799,7 +792,7 @@ def _code_type_ok(type_name: str | None) -> bool:
def _looks_like_code(name: str) -> bool:
words = _words(name)
words = catalog.name_words(name)
return bool(words) and words[-1] not in _NOT_CODE_TAILS and any(w in _CODE_WORDS for w in words)
@@ -881,14 +874,8 @@ async def _check_codes(cx: _Context, table: str, column: str, out: _TableOutcome
def _is_date_column(col: dict[str, Any]) -> bool:
"""日期类列:类型是日期、时间戳;或者类型是文字、列名像日期(SQLite 的日期都存成文字)。"""
t = str(col.get("type") or "").upper()
if "DATE" in t or "TIMESTAMP" in t:
return True
if t and not any(k in t for k in ("CHAR", "TEXT", "STRING")):
return False
words = _words(str(col.get("name") or ""))
return bool(words) and words[-1] in _DATE_TAILS and (len(words) > 1 or words[0] in ("date", "time", "day"))
"""日期类列(catalog.is_date_column:和 SQL 检查认时间列是同一个判断)。"""
return catalog.is_date_column(str(col.get("name") or ""), col.get("type"))
async def _check_dates(cx: _Context, table: str, columns: list[str], total: int, out: _TableOutcome,
+16 -14
View File
@@ -41,6 +41,7 @@ from dataclasses import dataclass, field
from typing import Any
from app.data import catalog
from app.data.names import name_key
logger = logging.getLogger(__name__)
@@ -69,11 +70,6 @@ _EXCERPT_MAX = 160
_SENTINEL = "__agentlab_tpl__"
_TEMPLATE = re.compile(r"\{\{.*?\}\}", re.S)
#: 名字看得出是时间的列:xxx_at、xxx_time、xxx_date、xxx_on、createdAt…(SQLite 的日期常存成文本,只看类型认不全)
_TEMPORAL_NAME = re.compile(r"(?:^|_)(?:at|time|date|day|dt|ts|on|timestamp|datetime)$|[a-z0-9](?:At|Time|Date|On)$",
re.IGNORECASE)
_TEMPORAL_TYPE = re.compile(r"DATE|TIME", re.IGNORECASE)
def dialect_of(kind: str | None) -> str | None:
"""数据源类型对应的 sqlglot 方言;不支持的类型返回 None(不查)。"""
@@ -133,7 +129,8 @@ class SqlChecker:
self.source_name = source_name
self._raw = {str(k): v for k, v in (notes or {}).items() if isinstance(v, Mapping) and v}
self._visible = {k: catalog.visible_notes(v) for k, v in self._raw.items()}
self._visible_lower = {k.lower(): v for k, v in self._visible.items()}
# 表名按 name_key 比(只把 ASCII 字母转小写),和 catalog.resolve_table_name、使用次数、影响面同一个口径
self._visible_lower = {name_key(k): v for k, v in self._visible.items()}
self._edges: dict[str, list[catalog.JoinEdge]] | None = None
@property
@@ -161,7 +158,7 @@ class SqlChecker:
return (key, meta) if key and isinstance(meta, Mapping) else None
def notes_of(self, table: str) -> dict[str, Any]:
return self._visible.get(table) or self._visible_lower.get(table.lower()) or {}
return self._visible.get(table) or self._visible_lower.get(name_key(table)) or {}
def edges_between(self, a: str, b: str) -> list[catalog.JoinEdge]:
"""从 a 连到 b 的边(每条关系正反各一条)。关系图第一次用到时才建,之后复用。
@@ -169,7 +166,7 @@ class SqlChecker:
和助手挑表看到的是同一张图(catalog.relation_graph):目录里的关系,加上没起草过的表按外键、命名现推的;
被驳回的关系不进图,也不会被现推回来。
"""
return [e for e in self.edges_from(a) if e.to_table.lower() == b.lower()]
return [e for e in self.edges_from(a) if name_key(e.to_table) == name_key(b)]
def edges_from(self, a: str) -> list[catalog.JoinEdge]:
"""从 a 出发的全部边。关系图第一次用到时才建,之后复用。"""
@@ -177,9 +174,9 @@ class SqlChecker:
graph = catalog.relation_graph(self._raw, schema_cache=self.schema_cache)
index: dict[str, list[catalog.JoinEdge]] = {}
for table, edges in graph.items():
index.setdefault(table.lower(), []).extend(edges)
index.setdefault(name_key(table), []).extend(edges)
self._edges = index
return list(self._edges.get(a.lower(), []))
return list(self._edges.get(name_key(a), []))
# ---- 检查
@@ -302,10 +299,11 @@ class _Table:
return self.columns[column.lower()], str(item.get("status"))
def temporal(self, column: str) -> bool:
"""这一列是不是时间列(catalog.is_date_column:和数据剖析认日期列是同一个判断,剖析提议业务日期时认的
「唯一日期列」和这里认的时间列对得上)。"""
meta = next((c for c in self.meta.get("columns") or []
if isinstance(c, Mapping) and str(c.get("name")).lower() == column.lower()), None)
kind = str((meta or {}).get("type") or "")
return bool(_TEMPORAL_TYPE.search(kind) or _TEMPORAL_NAME.search(column))
return catalog.is_date_column(column, (meta or {}).get("type"))
#: 一列解析到的位置:(表, 列的真名)
@@ -548,7 +546,7 @@ class _ScopeCheck:
if x in skip or y in skip:
continue
a, b = self.tables[x], self.tables[y]
if a.name.lower() == b.name.lower():
if name_key(a.name) == name_key(b.name):
continue # 自己连自己:目录不记这种关系
pairs = group["pairs"]
candidates = self.checker.edges_between(a.name, b.name)
@@ -990,7 +988,11 @@ def _number(text: str) -> float | None:
def _known_code(literal: tuple[str, bool], codes: Mapping[str, Any]) -> bool:
"""值在不在码值表里:原样相等、不分大小写相等、或者两边都是数且数值相等(1 和 '1'、1.0 都算 1)。"""
"""值在不在码值表里:原样相等、不分大小写相等、或者两边都是数且数值相等(1 和 '1'、1.0 都算 1)。
不分大小写是有意的:MySQL 默认的排序规则比较字符串本来就不分大小写,status = 'paid' 照样筛得出 'PAID';
PostgreSQL、SQLite 区分大小写,这样比会漏报一些,但宁可漏报,不能把在 MySQL 上完全正确的写法报成错。
"""
text = literal[0]
keys = [str(k) for k in codes]
if text in keys or text.strip().lower() in {k.strip().lower() for k in keys}:
+39
View File
@@ -441,3 +441,42 @@ async def test_model_draft_drops_formula_items():
assert set(notes["columns"]["total_amount"]) == {"label", "unit"}
# ==========================================================================
# 一致性
# ==========================================================================
@pytest.mark.parametrize("name, type_, expected", [
("visit_time", "TEXT", True), ("ordered_at", "TEXT", True), ("joined_on", "VARCHAR(20)", True),
("createdAt", "", True), ("biz_date", "DATE", True), ("paid", "DATETIME", True), ("loaded", "TIMESTAMP", True),
("dt", "TEXT", True), ("date", "TEXT", True),
("created_at", "INTEGER", False), # 整数存的时间戳:取值不是日期,剖析和检查都不当日期列
("open_time", "TIME", False), # 只有时刻、没有日期
("on_hand", "INTEGER", False), ("status", "TEXT", False), ("at", "TEXT", False), ("ticket_no", "TEXT", False),
])
def test_date_column_rule_is_shared(name, type_, expected):
assert catalog.is_date_column(name, type_) is expected
# 剖析认的日期列(业务日期提议)和 SQL 检查认的时间列(按别的时间列统计)是同一套
assert catalog_profile._is_date_column({"name": name, "type": type_}) is expected
table = sqlcheck._Table(alias="t", name="t", meta={"columns": [{"name": name, "type": type_}]}, notes={},
order=0)
assert table.temporal(name) is expected
def test_table_names_match_by_name_key_everywhere():
"""表名只把 ASCII 字母转小写(name_key,和 SQLite 比较标识符一致):非 ASCII 的大小写不同就是两张表。"""
cache = {"tables": {"CAFÉ_ORDERS": {"columns": []}, "Visits": {"columns": []}}}
assert catalog.resolve_table_name(cache, "visits") == "Visits"
assert catalog.resolve_table_name(cache, "main.VISITS") == "Visits"
assert catalog.resolve_table_name(cache, "cafÉ_orders") == "CAFÉ_ORDERS"
assert catalog.resolve_table_name(cache, "café_orders") is None
checker = sqlcheck.SqlChecker(kind="sqlite", schema_cache=cache, notes={"CAFÉ_ORDERS": {
"label": catalog.make_item("咖啡订单", "human")}})
assert checker.notes_of("cafÉ_orders")["label"]["value"] == "咖啡订单"
assert checker.notes_of("café_orders") == {}
def test_word_splitting_has_one_implementation():
assert catalog.name_words("memberTagId") == ["member", "tag", "id"]
assert catalog.name_words("MEMBER_TAG") == ["member", "tag"]
assert not hasattr(catalog_profile, "_words") and not hasattr(catalog_profile, "_WORDS")