diff --git a/backend/app/data/catalog.py b/backend/app/data/catalog.py index 92617fdd..463a5077 100644 --- a/backend/app/data/catalog.py +++ b/backend/app/data/catalog.py @@ -15,7 +15,7 @@ notes 结构（每个「项」都是 {value, source, status}，可选 note、upd "valid_filter": 项(str，SQL 条件片段), "dedup": 项(str), "columns": {列名: {"label": 项, "meaning": 项, "unit": 项, "measure": 项("flow"|"stock"|"ratio"|"identifier"|"status"|"attribute"), - "codes": 项(dict[str, str])}}, + "codes": 项(dict[码值, 含义]；含义可为空串，表示待填写)}}, "relations": [{"id", "columns", "to_table", "to_columns", "cardinality", "coverage", "source", "status", "note"?}]} @@ -181,8 +181,10 @@ def _business_date_ok(value: Any) -> bool: def _codes_ok(value: Any) -> bool: + """码值对照：{码值: 含义}，都是文字。含义可以是空串——数据剖析只知道列里出现过哪些取值，含义要等人填； + 以前要求含义非空，剖析得到的码值候选就写不进来。""" return (isinstance(value, dict) and bool(value) - and all(isinstance(k, str) and isinstance(v, str) and v.strip() for k, v in value.items())) + and all(isinstance(k, str) and isinstance(v, str) for k, v in value.items())) #: 每个字段的值怎么查、不合规时怎么说 @@ -1411,6 +1413,17 @@ def _date_text(value: Mapping[str, Any]) -> str: return "；".join(parts) +def _codes_text(codes: Mapping[str, Any]) -> str: + """码值对照的文字：有含义的写「1=有效」，含义待填写的单独说明——不能写成「1=」，模型会当成含义就是空。""" + filled = [f"{k}={v}" for k, v in codes.items() if str(v).strip()] + pending = [str(k) for k, v in codes.items() if not str(v).strip()] + if not pending: + return "、".join(filled) + if not filled: + return f"{'、'.join(pending)}（含义待填写）" + return f"{'、'.join(filled)}，{'、'.join(pending)} 的含义待填写" + + 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 []) @@ -1461,7 +1474,7 @@ def _notes_lines(notes: Mapping[str, Any] | None, *, meta: Mapping[str, Any] | N if name == "measure": text = MEASURE_LABEL.get(value, str(value)) elif name == "codes": - text = "、".join(f"{k}={v}" for k, v in value.items()) + text = _codes_text(value) else: text = str(value) # 第一段是中文名（没有中文名时是含义），不加前缀；其余写明是什么 diff --git a/backend/app/data/catalog_profile.py b/backend/app/data/catalog_profile.py index acb3244c..28750a79 100644 --- a/backend/app/data/catalog_profile.py +++ b/backend/app/data/catalog_profile.py @@ -22,6 +22,7 @@ import copy import logging import re import time +from collections.abc import Awaitable, Callable from dataclasses import asdict, dataclass, field from decimal import Decimal from typing import Any @@ -525,13 +526,37 @@ class _RelationOutcome: note: str +@dataclass +class _CodesOutcome: + """一列的取值分布：[(码值, 行数)]，按行数从多到少。""" + + column: str + values: list[tuple[str, int]] + rows: int + note: str + + +@dataclass +class _DateOutcome: + """表里唯一一个日期类列，提议作为业务日期。""" + + column: str + low: str + high: str + note: str + + @dataclass class _TableOutcome: """一张表所有检查的结论，以及没查成、要原样保留旧结论的槽位。""" relations: list[_RelationOutcome] = field(default_factory=list) - #: 没查成的关系：目录里剖析以前写的结论原样留着（不然会被 covered={"profile"} 当成「这次没有」删掉） + codes: list[_CodesOutcome] = field(default_factory=list) + business_date: _DateOutcome | None = None + #: 没查成的项：目录里剖析以前写的结论原样留着（不然会被 covered={"profile"} 当成「这次没有」删掉） keep_relations: set[str] = field(default_factory=set) + keep_codes: set[str] = field(default_factory=set) + keep_business_date: bool = False @dataclass @@ -556,6 +581,13 @@ class _Context: result.skipped += [ProfileSkip("row_estimate", table, s.reason, s.detail) for s in skips] return self.sizes[table] + async def size_with_data(self, table: str, result: TableProfile) -> TableSize: + """同 size，但确知是空表时直接跳过这一项（抛 _Skip），不再为它发查询。""" + size = await self.size(table, result) + if size.rows == 0: + raise _Skip("no_data", f"{table} 是空表，无法剖析") + return size + def is_masked(self, column: str) -> bool: return column.lower() in self.masked @@ -684,7 +716,7 @@ async def _check_relation(cx: _Context, table: str, rel: dict[str, Any], out: _T raise _Skip("masked", f"{owner}.{name} 在数据源设置中被遮罩，不取样", carry=True) d, s = cx.dialect, cx.settings - size = await cx.size(table, result) + size = await cx.size_with_data(table, result) scan_cap = cx.scan_cap(size) child_table, parent_table = d.table(meta, table), d.table(pmeta, parent) res = await cx.run(d.distinct_sample_sql(child_table, d.quote(col), s.sample_size, scan_cap=scan_cap), @@ -717,6 +749,167 @@ async def _check_relation(cx: _Context, table: str, rel: dict[str, Any], out: _T matched=matched, note=note)) +# ========================================================================== +# 检查：码值候选、日期列 +# ========================================================================== + +#: 码值候选最多几个取值。查询取 CODES_MAX + 1 个：取满了说明不是低基数列 +CODES_MAX = 20 +#: 码值的长度上限。更长的不像状态码，像说明文字 +MAX_CODE_LEN = 64 +#: 像状态、类型、渠道的列名里会有的词 +_CODE_WORDS = frozenset({"status", "state", "type", "kind", "category", "channel", "level", "method", "mode", + "reason", "stage", "phase", "grade", "flag", "source", "result"}) +#: 以这些词结尾的列不是码值：标识、名称、说明、时间、金额、数量（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: + """整数，或者短文字（定长不超过 MAX_CODE_LEN；SQLite 的 TEXT 不带长度，也算）。""" + t = (type_name or "").upper() + if not t or any(k in t for k in ("CLOB", "BLOB", "BINARY", "JSON", "DATE", "TIME", "INTERVAL", "REAL", "FLOAT", + "DOUBLE", "BOOL")): + return False + if "INT" in t: + return True + size = re.search(r"\(\s*(\d+)\s*(?:,\s*(\d+)\s*)?\)", t) + if any(k in t for k in ("NUMBER", "NUMERIC", "DECIMAL")): + return size is None or not size.group(2) or int(size.group(2)) == 0 + if any(k in t for k in ("CHAR", "TEXT", "STRING")): + return size is None or int(size.group(1)) <= MAX_CODE_LEN + return False + + +def _looks_like_code(name: str) -> bool: + words = _words(name) + return bool(words) and words[-1] not in _NOT_CODE_TAILS and any(w in _CODE_WORDS for w in words) + + +def _code_columns(meta: dict[str, Any], notes: dict[str, Any]) -> list[str]: + """要取值分布的列：目录里度量类型记为状态的、剖析以前提过码值候选的（再核一次）、或者类型和命名像状态、 + 类型、渠道的整数或短文字列。主键、唯一约束列、关系里的键列不算；码值人工确认或驳回过的不再取。 + 被遮罩的列也列出来，由检查记为「被遮罩」跳过——要让人看得见为什么它没有码值。""" + pk = [c.lower() for c in meta.get("primary_key") or []] + unique = {str(g[0]).lower() for g in meta.get("unique") or [] if isinstance(g, list) and len(g) == 1} + keys = {str(c).lower() for r in notes.get("relations") or [] if isinstance(r, dict) for c in r.get("columns") or []} + described = notes.get("columns") if isinstance(notes.get("columns"), dict) else {} + out: list[str] = [] + for col in meta.get("columns") or []: + name = str(col.get("name") or "") + items = described.get(name) if isinstance(described.get(name), dict) else {} + codes, measure = items.get("codes"), items.get("measure") + if isinstance(codes, dict) and codes.get("status") in ("confirmed", "rejected"): + continue + status_measure = (isinstance(measure, dict) and measure.get("value") == "status" + and measure.get("status") != "rejected") + profiled = isinstance(codes, dict) and codes.get("source") == "profile" + if not (status_measure or profiled or _looks_like_code(name)): + continue + lowered = name.lower() + if (pk == [lowered]) or lowered in unique or lowered in keys or not _code_type_ok(col.get("type")): + continue + out.append(name) + return out + + +def _code_key(value: Any) -> str | None: + """取值分布里的一个取值 → 码值（文字）。只认整数和不太长的文字。""" + if value is None or isinstance(value, bool): + return None + if isinstance(value, int): + return str(value) + if isinstance(value, float): + return str(int(value)) if value.is_integer() else None + if isinstance(value, str) and value.strip() and len(value) <= MAX_CODE_LEN and not _CONTROL.search(value): + return value + return None + + +def _share(count: int, total: int) -> str: + if total <= 0: + return "0%" + ratio = count / total + return "不足 1%" if 0 < ratio < 0.01 else f"{round(ratio * 100)}%" + + +async def _check_codes(cx: _Context, table: str, column: str, out: _TableOutcome, result: TableProfile) -> None: + """GROUP BY 取出现最多的 CODES_MAX + 1 个取值：不超过 CODES_MAX 个就是码值候选（值有了，含义等人填）。""" + if cx.is_masked(column): + raise _Skip("masked", f"{table}.{column} 在数据源设置中被遮罩，不取值", carry=False) + d, meta = cx.dialect, cx.tables[table] + size = await cx.size_with_data(table, result) + scan_cap = cx.scan_cap(size) + res = await cx.run(d.value_counts_sql(d.table(meta, table), d.quote(column), CODES_MAX + 1, scan_cap=scan_cap), + max_rows=CODES_MAX + 1) + if not res.rows: + raise _Skip("no_data", f"{table}.{column} 没有非空值，无法取值分布") + if len(res.rows) > CODES_MAX: + raise _Skip("high_cardinality", f"{table}.{column} 的取值超过 {CODES_MAX} 个，不作为码值候选", carry=False) + values: list[tuple[str, int]] = [] + for row in res.rows: + key = _code_key(row[0]) + if key is None: + raise _Skip("unsupported", f"{table}.{column} 的取值不是整数或短文字，不作为码值候选", carry=False) + values.append((key, _as_int(row[1]) or 0)) + if len(values) >= 3 and all(n == 1 for _, n in values): + # 每个取值只出现一次（票种编码 T01…T09）：这是一列标识，不是状态码 + raise _Skip("high_cardinality", f"{table}.{column} 的取值各不相同，像标识而不是码值，不作为码值候选", + carry=False) + rows = sum(n for _, n in values) + scope = f"按前 {scan_cap} 行统计" if scan_cap else "统计全表" + spread = "、".join(f"{k}（{_share(n, rows)}）" for k, n in values) + note = f"{NOTE_PREFIX}{cx.day}）：{scope}，{rows} 行非空值共 {len(values)} 个取值：{spread}。" + out.codes.append(_CodesOutcome(column=column, values=values, rows=rows, note=note)) + + +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")) + + +async def _check_dates(cx: _Context, table: str, columns: list[str], total: int, out: _TableOutcome, + result: TableProfile) -> None: + """日期类列的 MIN、MAX，作为业务日期的参考。整表统计，只对行数不超过上限的表做。表里只有一个日期类列 + （total 是表结构里日期类列的个数，含被遮罩的）时提议它作为业务日期。""" + d, meta = cx.dialect, cx.tables[table] + size = await cx.size_with_data(table, result) + if not size.within(cx.settings.max_scan_rows): + raise _Skip("too_large", f"{table} 的行数超过整表统计的行数上限（{cx.settings.max_scan_rows} 行），" + "不取日期列的最小值和最大值") + res = await cx.run(d.min_max_sql(d.table(meta, table), [d.quote(c) for c in columns])) + row = res.rows[0] if res.rows else [] + for i, column in enumerate(columns): + low, high = (row[2 * i], row[2 * i + 1]) if len(row) >= 2 * i + 2 else (None, None) + if low is None or high is None or not (_DATE_TEXT.match(str(low)) and _DATE_TEXT.match(str(high))): + continue + result.date_ranges.append({"column": column, "min": str(low), "max": str(high)}) + if total != 1: + return + if not result.date_ranges: + raise _Skip("no_data", f"{table}.{columns[0]} 没有可识别的日期值") + found = result.date_ranges[0] + note = (f"{NOTE_PREFIX}{cx.day}）：表里只有这一个日期类列，取值从 {str(found['min'])[:19]} 到 " + f"{str(found['max'])[:19]}，可作为业务日期的参考。") + out.business_date = _DateOutcome(column=found["column"], low=found["min"], high=found["max"], note=note) + + # ========================================================================== # 结论进目录 # ========================================================================== @@ -738,6 +931,25 @@ def _relation_finding(o: _RelationOutcome, rel: dict[str, Any], *, status: str, "summary": summary} +def _codes_value(c: _CodesOutcome, item: dict[str, Any] | None) -> dict[str, str]: + """码值候选的值：观察到的取值，含义先空着。原来有别的来源（模型起草、数据库注释）写过含义的，对得上的取值 + 沿用它的含义；它写了、数据里没出现的取值也留着——码值是「可能的取值」，没出现不等于不存在。""" + old = item.get("value") if isinstance(item, dict) and isinstance(item.get("value"), dict) else {} + value = {key: str(old.get(key) or "") for key, _ in c.values} + for key, meaning in old.items(): + if key not in value and isinstance(meaning, str) and meaning.strip(): + value[key] = meaning + return value + + +def _codes_finding(c: _CodesOutcome, value: dict[str, str]) -> dict[str, Any]: + pending = sum(1 for v in value.values() if not v.strip()) + summary = f"{len(c.values)} 个取值" + (f"，{pending} 个含义待填写" if pending else "") + return {"kind": "codes", "path": f"columns.{c.column}.codes", "column": c.column, + "values": [{"value": k, "rows": n} for k, n in c.values], "rows": c.rows, "status": "proposed", + "summary": summary} + + def _patch_confirmed(rel: dict[str, Any], o: _RelationOutcome, at: str) -> bool: """人工确认过的关系只补覆盖率和基数：覆盖率是测量值，照新的写；基数只在原来没有时补，不改人定的； 备注只在原来为空、或者是剖析自己写的时候换成新的。状态、来源不动。返回改没改。""" @@ -764,8 +976,10 @@ def _apply(existing: dict[str, Any] | None, out: _TableOutcome, *, - 没被确认、驳回的关系：记为 source=profile，覆盖率够、父键唯一的 verified，否则 proposed；按 merge_notes 的规则并入（剖析顶得掉命名推断、模型起草，顶不掉外键和人工）。 - 人工确认过的关系：只补覆盖率和基数（_patch_confirmed）。 - - covered={"profile"}：这次完整检查过、却没再得出的剖析结论删掉（比如取值变多、不再像码值的列）； - 没查成的检查（超时、预算用完……）把剖析以前写的结论原样带上，不因为这次没查成就删。 + - 码值候选：profile / proposed，含义沿用别的来源写过的（_codes_value）；人工确认、驳回过的码值不碰。 + - 业务日期：只有一个日期类列时提议，profile / proposed；已有别的来源定的业务日期不顶掉。 + - covered={"profile"}：这次完整检查过、却没再得出的剖析结论删掉（比如取值变多、不再像码值的列，列被遮罩 + 之后它的码值）；没查成的检查（超时、预算用完……）把剖析以前写的结论原样带上，不因为这次没查成就删。 """ existing = existing or {} current = {str(r["id"]): r for r in existing.get("relations") or [] if isinstance(r, dict) and r.get("id")} @@ -790,6 +1004,35 @@ def _apply(existing: dict[str, Any] | None, out: _TableOutcome, *, if rel is not None and rel.get("source") == "profile": relations.append(copy.deepcopy(rel)) draft: dict[str, Any] = {"relations": relations} if relations else {} + + described = existing.get("columns") if isinstance(existing.get("columns"), dict) else {} + columns: dict[str, dict[str, Any]] = {} + for c in out.codes: + item = (described.get(c.column) or {}).get("codes") if isinstance(described.get(c.column), dict) else None + if isinstance(item, dict) and item.get("status") in ("confirmed", "rejected"): + continue + columns[c.column] = {"codes": catalog.make_item(_codes_value(c, item), "profile", "proposed", note=c.note)} + findings.append(_codes_finding(c, columns[c.column]["codes"]["value"])) + for column in out.keep_codes: + item = (described.get(column) or {}).get("codes") if isinstance(described.get(column), dict) else None + if isinstance(item, dict) and item.get("source") == "profile": + columns[column] = {"codes": copy.deepcopy(item)} + if columns: + draft["columns"] = columns + + current_date = existing.get("business_date") if isinstance(existing.get("business_date"), dict) else None + ours = current_date is None or (current_date.get("source") == "profile" + and current_date.get("status") not in ("confirmed", "rejected")) + if out.business_date is not None and ours: + # 已有别的来源（模型起草、人工）定的业务日期不顶掉：按来源的可信程度剖析排得更前，但它只知道 + # 「表里只有这一个日期列」，不比写了规则和时区的那条更懂业务 + b = out.business_date + draft["business_date"] = catalog.make_item({"column": b.column}, "profile", "proposed", note=b.note) + findings.append({"kind": "business_date", "path": "business_date", "column": b.column, "min": b.low, + "max": b.high, "status": "proposed", "summary": f"提议按 {b.column} 作为业务日期"}) + elif out.keep_business_date and current_date is not None and current_date.get("source") == "profile": + draft["business_date"] = copy.deepcopy(current_date) + merged, stats = catalog.merge_notes(existing, draft, covered={"profile"}, at=at) for rel in merged.get("relations") or []: o = confirmed.get(str(rel.get("id"))) @@ -812,19 +1055,35 @@ class _Check: kind: str target: str path: str | None - run: Any - #: 没查成时要原样保留的旧结论（关系编号） - keep_relation: str | None = None + run: Callable[[], Awaitable[None]] + #: 没查成时调用：把目录里剖析以前对这一项写的结论标记为原样保留 + keep: Callable[[], None] def _plan(cx: _Context, table: str, notes: dict[str, Any], out: _TableOutcome, result: TableProfile) -> list[_Check]: """这张表要做的检查，按价值排：关系、码值、日期。先列全再执行：中途停下时，没做的每一项都要记下原因。""" + meta = cx.tables[table] checks: list[_Check] = [] for rel in _relation_candidates(notes): - checks.append(_Check("relation", _relation_target(rel), f"relations.{rel['id']}", + rid = str(rel["id"]) + checks.append(_Check("relation", _relation_target(rel), f"relations.{rid}", lambda rel=rel: _check_relation(cx, table, rel, out, result), - keep_relation=str(rel["id"]))) + lambda rid=rid: out.keep_relations.add(rid))) + for column in _code_columns(meta, notes): + checks.append(_Check("codes", column, f"columns.{column}.codes", + lambda column=column: _check_codes(cx, table, column, out, result), + lambda column=column: out.keep_codes.add(column))) + dates = [str(c["name"]) for c in meta.get("columns") or [] if isinstance(c, dict) and _is_date_column(c)] + for column in dates: + if cx.is_masked(column): + result.skipped.append(ProfileSkip("date", column, "masked", + f"{table}.{column} 在数据源设置中被遮罩，不取最小值和最大值")) + usable = [c for c in dates if not cx.is_masked(c)] + if usable: + checks.append(_Check("date", "、".join(usable), "business_date" if len(dates) == 1 else None, + lambda: _check_dates(cx, table, usable, len(dates), out, result), + lambda: setattr(out, "keep_business_date", True))) return checks @@ -839,15 +1098,14 @@ async def _profile_table(cx: _Context, table: str, notes: dict[str, Any], result await check.run() except _Skip as e: result.skipped.append(ProfileSkip(check.kind, check.target, e.reason, e.detail, check.path)) - if check.keep_relation and e.carry: - out.keep_relations.add(check.keep_relation) + if e.carry: + check.keep() except _Stop as e: detail = _STOP_DETAIL[e.reason].format(max_queries=cx.settings.max_queries, max_total_s=_fmt(cx.settings.max_total_s)) for rest in checks[i:]: result.skipped.append(ProfileSkip(rest.kind, rest.target, e.reason, detail, rest.path)) - if rest.keep_relation: - out.keep_relations.add(rest.keep_relation) + rest.keep() break if table in cx.sizes: result.row_estimate = cx.sizes[table] @@ -951,7 +1209,7 @@ async def _profile(session: AsyncSession, row: Any, source: Any, tables: list[st return report -__all__ = ["IN_CHUNK", "MAX_LITERAL_LEN", "NOTE_PREFIX", "PROFILE_DEFAULT_TABLES", "PROFILE_OPTION", "ProfileBusy", +__all__ = ["CODES_MAX", "IN_CHUNK", "MAX_CODE_LEN", "MAX_LITERAL_LEN", "NOTE_PREFIX", "PROFILE_DEFAULT_TABLES", "PROFILE_OPTION", "ProfileBusy", "ProfileDisabled", "ProfileReport", "ProfileSettings", "ProfileSkip", "SqlDialect", "TableProfile", "TableSize", "estimate_size", "profile_catalog", "profile_settings", "profile_settings_problem", "sql_literal"] diff --git a/backend/tests/test_catalog_profile_units.py b/backend/tests/test_catalog_profile_units.py index 9913c2bc..d41ecf2d 100644 --- a/backend/tests/test_catalog_profile_units.py +++ b/backend/tests/test_catalog_profile_units.py @@ -152,6 +152,16 @@ def test_dialect_quoting_and_row_limits(): # ---------------------------------------------------------------- 目录对剖析结果的容纳 +def test_codes_with_pending_meanings_are_valid_and_render_as_pending(): + """剖析得到的码值只有取值、含义待人填写：目录要收得下，给模型看时写明含义待填写。""" + notes = {"columns": {"status": {"codes": make_item({"1": "", "0": ""}, "profile", "proposed")}, + "channel": {"codes": make_item({"WEB": "官网", "BOX": ""}, "profile", "proposed")}}} + assert catalog.validate_notes(notes) == [] + text = catalog.render_table_notes(notes) + assert "1、0（含义待填写）" in text + assert "WEB=官网，BOX 的含义待填写" in text + # 码值仍得是对照表：值不是文字的照样拒收 + assert catalog.validate_notes({"columns": {"s": {"codes": make_item({"1": 2}, "profile")}}}) def _profile_rel(coverage, cardinality, status):