diff --git a/backend/app/api/copilot.py b/backend/app/api/copilot.py index 91022193..1ec9ebe4 100644 --- a/backend/app/api/copilot.py +++ b/backend/app/api/copilot.py @@ -25,6 +25,7 @@ from app.api.copilot_gen import ( # noqa: F401 # 起别名：本文件底下有个叫 explain 的接口，generate 里又有个局部变量叫 raw，同名会互相盖掉 from app.core.errors import explain as explain_error, graph_error, raw as raw_error from app.data.sqlcheck import CHECK_CODES +from app.data.sqlcheck_columns import sql_unknown_columns from app.db.base import get_session from app.db.models import Workflow from app.engine.layout import CORRIDOR_MIN, MIN_ROW_GAP, NODE_W, _height, auto_layout @@ -1347,80 +1348,9 @@ def _catalog_issues(node_id: str, sql: str, checker: Any) -> list[dict[str, Any] # -------------------------------------------------------------------------- -# SQL 里的列名对照数据源结构 -# -# 只做有把握的判断，宁可漏判不可误判：这条会把图打回去让模型改。CTE、FROM 里的子查询、 -# 引号括起来的标识符、结构里找不到的表、结构没探查过或者探查被截断的，一律不判。 +# SQL 里的列名对照数据源结构（sql_unknown_columns）在 data/sqlcheck_columns.py：表结构变化的影响面也用它 # -------------------------------------------------------------------------- -#: @ 开头的是变量（@region、@@session），: 开头的是参数，. 后面的是限定名 -_SQL_IDENT = re.compile(r"(? str: - """带引号、名字里有非 ASCII 字符的标识符（中文别名 AS "月份"、AS `曝光量`）连引号一起换成空格。 - - 这项核对只认 ASCII 的列名，中文名本来就不判；不抹掉的话，SQL 里一出现带引号的标识符就整条放弃核对，照「结果列 - 起中文别名」写的 SQL 会把猜错的英文列名一起放过去。英文的带引号标识符（"StatDate"）大小写、空格都可能，照旧不判。""" - from app.data.guard import _GENERIC, _tokens - - out = [] - for token in _tokens(sql, _GENERIC): - if token.kind == "ident" and not token.text.isascii(): - out.append(" " * len(token.text)) - else: - out.append(token.text) - return "".join(out) - - def sql_source_prefix(sql: str, source: Any) -> list[str]: """SQL 里 FROM / JOIN 后面写成「数据源名.表名」的表（去重、按出现顺序）。 @@ -1444,66 +1374,6 @@ def sql_source_prefix(sql: str, source: Any) -> list[str]: return list(dict.fromkeys(f"{name}.{m.group(1)}" for m in pattern.finditer(text))) -def sql_unknown_columns(sql: str, source: Any) -> tuple[list[str], list[str]] | None: - """SQL 里用到、数据源结构里查不到的列：(查不到的列, 这几张表的全部列)。判断不了、或者都查得到时返回 None。""" - from app.data import introspect as _introspect - from app.data.guard import _blank_quoted, first_verb, strip_comments - - cache = getattr(source, "schema_cache", None) or {} - if not cache.get("tables") or cache.get("failed") or first_verb(sql) != "select": - return None - text = _blank_unicode_identifiers(re.sub(r"\{\{.*?\}\}", " 0 ", strip_comments(sql), flags=re.S)) - if '"' in text.replace("''", "") and _blank_quoted(text) != _blank_quoted(text, identifiers=False): - return None # 引号括起来的标识符：大小写、空格都可能，不猜 - text = _blank_quoted(text) - if re.search(r"\bwith\b", text, re.I) or re.search(r"\b(?:from|join)\s*\(", text, re.I) or "`" in text \ - or "[" in text: - return None # [方括号] 标识符、数组下标：同样不猜 - tables: dict[str, dict[str, Any]] = {} # 表名、别名（小写）→ 表结构 - spans: list[tuple[int, int]] = [] - for m in _FROM_ITEMS.finditer(text): - spans.append(m.span(1)) - for item in m.group(1).split(","): - parsed = _TABLE_ITEM.match(item) - if not parsed: - return None - name, alias = parsed.group(1), parsed.group(2) - meta = _introspect.find_table(source, name) - if meta is None: - return None # 结构里找不到的表（视图、同义词、截断了的结构）不判 - tables[name.lower()] = tables[name.rsplit(".", 1)[-1].lower()] = meta - if alias and alias.lower() not in _SQL_WORDS: - tables[alias.lower()] = meta - if not tables: - return None - known = list(dict.fromkeys(c["name"] for meta in {id(m): m for m in tables.values()}.values() - for c in meta.get("columns") or [])) - columns = {c.lower() for c in known} - # 「SELECT created_at FROM」里 created_at 前面是关键字，它是列，不是别名 - aliases = {a.lower() for a in _ALIAS_AS.findall(text)} | { - m.group(2).lower() for m in _ALIAS_BARE.finditer(text) if m.group(1).lower() not in _OPERAND_BEFORE} | { - a.lower() for a in _WINDOW_NAME.findall(text)} - names = {m.start(1) for pattern in (_NAME_AFTER, _DATEPART_ARG) for m in pattern.finditer(text)} - unknown: list[str] = [] - for m in _QUALIFIED.finditer(text): - if any(a <= m.start() < b for a, b in spans): - continue - meta = tables.get(m.group(1).lower()) - col = m.group(2) - if meta is not None and col != "*" and col.lower() not in {c["name"].lower() for c in meta.get("columns") or []}: - unknown.append(f"{m.group(1)}.{col}") - for m in _SQL_IDENT.finditer(text): - word = m.group(1) - low = word.lower() - if any(a <= m.start() < b for a, b in spans) or low in _SQL_WORDS or low in columns or low in aliases \ - or low in tables or m.start() in names or re.match(r"\s*[.(]", text[m.end():]) \ - or text[m.end():m.end() + 1] == "'": # N'…'、X'…'、_utf8mb4'…'：字面量的前缀 - continue - unknown.append(word) - unknown = list(dict.fromkeys(unknown)) - return (unknown, known) if unknown else None - - def _source_of(tool: str) -> str | None: from app.tools.datasource import QUERY_PREFIX, SCHEMA_PREFIX diff --git a/backend/app/api/workflows.py b/backend/app/api/workflows.py index 91cefb94..2dd765c0 100644 --- a/backend/app/api/workflows.py +++ b/backend/app/api/workflows.py @@ -12,7 +12,9 @@ from app.core.errors import graph_error from app.api.coded import CodedHTTPException from app.api.runs import actor_of from app.core.artifact_store import graph_hash -from app.data.catalog_impact import catalog_drift, catalog_versions_for, recorded_versions +from app.data.catalog_impact import ( + catalog_drift, catalog_versions_for, open_structure_issues, recorded_versions, schema_drift, schema_versions_for, +) from app.db.base import get_session from app.db.models import Run, Workflow, WorkflowVersion from app.engine.governance import publish_issues @@ -345,6 +347,10 @@ class VersionDetailOut(VersionOut): catalog_versions: dict[str, Any] | None = None #: 发布之后目录有变化的表：[{source, source_id, table, label, published, current, impact}] catalog_changes: list[dict[str, Any]] = Field(default_factory=list) + #: 发布时记下的表结构版本 {源名: {表名: 结构版本}}（只记写死的 SQL 直接引用的表，next-round 03 B1）；记录之前发布的为 null + schema_versions: dict[str, Any] | None = None + #: 发布之后结构版本变了的表：[{source, source_id, table, label, published, current, removed, changes, more}] + schema_changes: list[dict[str, Any]] = Field(default_factory=list) #: 入口字段换成表单（agentlab.form/1）：发起表单、应用入口用同一个渲染器（X2） input_schema: dict[str, Any] = Field(default_factory=dict) #: 账本（L4）：固定这一版的应用、生效记录数、正式运行数 @@ -388,6 +394,8 @@ async def get_version( published=workflow.published_version == snapshot.version, level=snapshot.level, catalog_versions=recorded_versions(snapshot.catalog_versions), catalog_changes=await catalog_drift(session, snapshot.catalog_versions), + schema_versions=snapshot.schema_versions, + schema_changes=await schema_drift(session, snapshot.schema_versions), input_schema=input_schema(fields), published_at=snapshot.published_at, published_by=snapshot.published_by, retired_at=snapshot.retired_at, retired_by=snapshot.retired_by, retire_note=snapshot.retire_note, usage=await usage(session, workflow_id, snapshot.version), @@ -535,7 +543,9 @@ async def gate_issues(session: AsyncSession, graph: Any, level: str) -> list[dic checkers = await _sql_checkers(session, spec) providers = await _providers(session, spec) with with_subgraphs(await load_subgraphs(session, spec)): - return [i.model_dump() for i in publish_issues(spec, level=level, checkers=checkers, providers=providers)] + issues = [i.model_dump() for i in publish_issues(spec, level=level, checkers=checkers, providers=providers)] + # 这一版直接引用的表上还没处理的结构变化（next-round 03 B1）：只是警告，不拦发布 + return issues + await open_structure_issues(session, spec) async def commit_publish(session: AsyncSession, workflow: Workflow, snapshot: WorkflowVersion, level: str, @@ -554,6 +564,8 @@ async def commit_publish(session: AsyncSession, workflow: Workflow, snapshot: Wo snapshot.level = level # 发布时的目录版本也记在这一版上：重新发布同一版等于按现在的目录重新认可它，跟着更新 snapshot.catalog_versions = await catalog_versions_for(session, spec) + # 表的结构版本同理（next-round 03 B1）：正式运行发起时和现在的比，变了写 schema.drift，只提醒、不拦 + snapshot.schema_versions = await schema_versions_for(session, spec) # 发布人、发布时间、用到的能力版本也记在这一版上（F5）：工作流上的 published_by 只记最近一次。退役过的版本重新 # 发布等于重新认可它，退役信息清掉 snapshot.published_at = utcnow() @@ -673,7 +685,10 @@ async def check_graph(session: AsyncSession, graph: Any, level: str) -> dict[str checkers = await _sql_checkers(session, spec) providers = await _providers(session, spec) with with_subgraphs(await load_subgraphs(session, spec)): - return check(spec, level=level, versions=versions, cards=cards, checkers=checkers, providers=providers) + out = check(spec, level=level, versions=versions, cards=cards, checkers=checkers, providers=providers) + # 这一版直接引用的表上还没处理的结构变化（next-round 03 B1）：发布弹窗照别的警告一样列出来，不拦发布 + out["issues"] = [*out["issues"], *await open_structure_issues(session, spec)] + return out @router.post("/{workflow_id}/autofix") diff --git a/backend/app/core/events.py b/backend/app/core/events.py index bf891b8f..3a39eb14 100644 --- a/backend/app/core/events.py +++ b/backend/app/core/events.py @@ -77,6 +77,10 @@ class EventType(StrEnum): # 每张表带 impact：direct 写死的 SQL 用到的，possible Agent 可能查询的。 # 只提醒、不拦运行；落在封存范围内，出具物上看得到「当时的目录和发布时不一样」 CATALOG_DRIFT = "catalog.drift" + # 从发布版本发起的正式运行：发布之后这一版直接引用的表的结构版本变了（data/catalog_impact.schema_drift，next-round + # 03 B1）。每张表带发布时和现在的结构版本、是否已删除、还没处理的结构变化。和 catalog.drift 分开记（老的渲染只认整数的 + # 目录版本），界面上合到同一块横幅里。只提醒、不拦运行；落在封存范围内 + SCHEMA_DRIFT = "schema.drift" # 报告撰写节点的自查结果：报告文档工件 id、统计、违规清单。落在封存范围内， # 证据接口靠它找到文档——不去翻不受封存保护的 artifacts 表 REPORT_CHECKED = "report.checked" diff --git a/backend/app/data/catalog_impact.py b/backend/app/data/catalog_impact.py index f8f3daa5..c0d330e4 100644 --- a/backend/app/data/catalog_impact.py +++ b/backend/app/data/catalog_impact.py @@ -10,6 +10,11 @@ Agent、协作成员绑定了查询工具的源里所有有目录的表。从这一版发起正式运行时和当前的比（catalog_drift），有变化 只提醒、不拦运行——目录是给人和助手看的说明，不是运行的输入；拦下来等于让一次文字修订卡住正式出具。 +- **表结构变化**（next-round 03 B1）：结构变化的影响面（structure_impact：目录里说到被删、被改的列的项和状态，人工确认过的 + 升为「要处理」；已发布模板写死的 SQL 换上新结构再跑 sql_unknown_columns，说得出「这条 SQL 现在会报列不存在」）；发布时记下 + 直接引用的表的结构版本（schema_versions_for，写法照 catalog_versions_for），正式运行发起时比对（schema_drift，变了写 + schema.drift，只提醒、不拦）；发布前检查列出这一版用到的表上还没处理的结构变化（open_structure_issues，警告）。 + 表名的取法和证据台账同一个口径：evidence.sql_tables 取 FROM / JOIN 后面的表（去掉注释、字符串、CTE）， 去掉 schema 前缀后按 name_key 对到表结构上（和 catalog.table_usage 一致）。哪些节点算「写死的查询」以 sqlcheck.tool_query 为准（tool_queries），发布记录、影响面、SQL 检查和助手的上下文认的是同一种节点。 @@ -402,5 +407,325 @@ async def catalog_drift(session: AsyncSession, recorded: Mapping[str, Any] | Non return [] -__all__ = ["IMPACTS", "CatalogVersions", "agent_bindings", "agent_sources", "catalog_drift", "catalog_versions_for", - "graph_sql_tables", "recorded_versions", "sql_table_keys", "table_impact", "template_refs", "tool_queries"] +# ========================================================================== +# 表结构变化（next-round 03 B1）：影响面、发布时的结构版本、正式运行发起时比对、发布前检查 +# ========================================================================== + +#: 发布时记下的结构版本：{源名: {表名: 结构版本}}，只记写死的 SQL 直接引用的表（写法照 catalog_versions_for） +SchemaVersions = dict[str, dict[str, int]] + + +async def published_specs(session: AsyncSession) -> list[tuple[Any, Any, Any]]: + """每个发布过的模板、它当前的已发布版本、解析好的图：[(Workflow, WorkflowVersion, GraphSpec)]。图解析不了的跳过（发起 + 不了正式运行，也就谈不上影响）。一次后台探查里各张表的影响面共用这一份，不每张表重读、重解析一遍。""" + from app.engine.schema import GraphSpec + + out: list[tuple[Any, Any, Any]] = [] + for workflow, version in await _published(session): + try: + spec = GraphSpec.model_validate(version.graph or {}) + except Exception: # noqa: BLE001 + continue + out.append((workflow, version, spec)) + return out + + +def _rel_columns(rel: Mapping[str, Any], side: str) -> set[str]: + return {str(c).lower() for c in rel.get(side) or []} + + +def catalog_refs(notes_by_table: Mapping[str, Mapping[str, Any]], table: str, columns: set[str] | None, *, + kind: str | None = None, before_meta: Mapping[str, Any] | None = None, + source_kind: str | None = None) -> tuple[list[dict[str, Any]], bool]: + """目录里说到这张表这几列的项：([{table, path, label, status, source}], 有效记录条件用没用到它们)。 + + columns 为 None 是整张表（表删了）：这张表的每一项，外加别的表指向它的关联关系。否则是说到这几列的： + columns.<列>.*、业务主键、业务日期、有效记录条件（按 sqlcheck.parse_valid_filter 取它用到的列，按变化之前的结构解析）、 + 这张表的关联关系里用到这几列的，以及别的表指向这几列的关联关系。kind 是 fk_changed 时只看外键约束来的关联关系。 + 列名按不分大小写比。""" + from app.data import catalog, catalog_review, sqlcheck + + own = notes_by_table.get(table) or {} + wanted = {c.lower() for c in columns} if columns is not None else None + out: list[dict[str, Any]] = [] + rules_touched = False + + def add(owner: str, key: tuple[str, ...], item: Mapping[str, Any]) -> None: + out.append({"table": owner, "path": catalog_review.slot_path(key), "label": catalog_review.item_label(key, item), + "status": item.get("status"), "source": item.get("source")}) + + for key, item in catalog._slots(own).items(): + if wanted is None: + add(table, key, item) + if key == ("t", "valid_filter"): + rules_touched = True + continue + if kind == "fk_changed": + if key[0] == "r" and item.get("source") == "fk" and _rel_columns(item, "columns") & wanted: + add(table, key, item) + continue + hit = False + if key[0] == "c": + hit = key[1].lower() in wanted + elif key == ("t", "keys"): + hit = any(str(c).lower() in wanted for c in item.get("value") or []) + elif key == ("t", "business_date"): + hit = str((item.get("value") or {}).get("column") or "").lower() in wanted + elif key == ("t", "valid_filter"): + parsed = sqlcheck.parse_valid_filter(item.get("value"), before_meta, kind=source_kind, table=table) + hit = any(c.lower() in wanted for c in parsed.columns) + rules_touched = rules_touched or hit + elif key[0] == "r": + hit = bool(_rel_columns(item, "columns") & wanted) + if hit: + add(table, key, item) + target = name_key(table) + for other, notes in notes_by_table.items(): + if other == table: + continue + for key, item in catalog._slots(notes).items(): + if key[0] != "r" or name_key(str(item.get("to_table") or "").rsplit(".", 1)[-1]) != target: + continue + if kind == "fk_changed": + continue + if wanted is None or _rel_columns(item, "to_columns") & wanted: + add(other, key, item) + return out, rules_touched + + +def _short(column: str) -> str: + return str(column).rsplit(".", 1)[-1].lower() + + +def template_hits(specs: list[tuple[Any, Any, Any]], source_name: str, table: str, *, + tables_after: Mapping[str, Any] | None, removed_columns: set[str] | None, + table_removed: bool = False) -> list[dict[str, Any]]: + """已发布模板里用到这张表的：[{workflow_id, name, version, level, impact, nodes: [{node_id, label, type, impact, + breaks, unknown_columns?}]}]，直接引用的在前。 + + 直接引用（写死的 SQL）的那几个节点，拿新结构（tables_after）再跑一遍 sql_unknown_columns（data/sqlcheck_columns）：报出的 + 列里有被删的列（removed_columns），就是「这条 SQL 现在会报列不存在」（breaks）。表真删了，直接引用的节点都 breaks。 + 判断不了的 SQL（CTE、引号标识符……）不说会坏。""" + from types import SimpleNamespace + + from app.data.sqlcheck_columns import sql_unknown_columns + + removed = {c.lower() for c in removed_columns or set()} + known = list(tables_after) if tables_after is not None else None + probe = SimpleNamespace(name=source_name, schema_cache={"tables": dict(tables_after or {})}) + out: list[dict[str, Any]] = [] + for workflow, version, spec in specs: + refs = template_refs(spec, source_name, table, known if not table_removed else None) + if not refs: + continue + sqls = {getattr(node, "id", None): sql for node, name, sql in tool_queries(getattr(spec, "nodes", []) or []) + if name == source_name} + nodes: list[dict[str, Any]] = [] + for ref in refs: + entry = {k: ref[k] for k in ("node_id", "label", "type", "impact")} + entry["breaks"] = False + if ref["impact"] == "direct" and ref["node_id"] in sqls: + if table_removed: + entry["breaks"] = True + elif removed: + found = sql_unknown_columns(sqls[ref["node_id"]], probe) + hit = [c for c in (found[0] if found else []) if _short(c) in removed] + if hit: + entry.update(breaks=True, unknown_columns=hit) + nodes.append(entry) + level = version.level or (workflow.status if workflow.status in ("published", "governed") else "published") + out.append({"workflow_id": workflow.id, "name": workflow.name, "version": version.version, "level": level, + "impact": "direct" if any(n["impact"] == "direct" for n in nodes) else "possible", "nodes": nodes}) + out.sort(key=lambda t: (t["impact"] != "direct", t["name"])) + return out + + +async def structure_impact(session: AsyncSession, row: Any, table: str, diffs: list[Any], *, + before_meta: Mapping[str, Any] | None, tables_after: Mapping[str, Any] | None, + specs: list[tuple[Any, Any, Any]] | None = None, + notes_by_table: Mapping[str, Mapping[str, Any]] | None = None, + ) -> list[tuple[dict[str, Any], str]]: + """一张表的每条结构变化（schema_diff.Diff，或者表级的那几种）→ (影响面, 严重程度)，和 diffs 一一对应。 + + 影响面 {catalog: [{table, path, label, status, source, needs_action}], templates: [...], rules_touched}： + - 删了表、删了列、类型或可空变了：目录里说到它们的项（catalog_refs），已发布模板里用到这张表的（template_hits）； + - 外键变了：外键约束来的关联关系；其余（加列、注释、离开范围、读不到结构）没有影响面。 + 严重程度：删了表、删了列时，人工确认过的项还说着它（needs_action）、或者写死的 SQL 现在会报列不存在，是 action（要处理）； + 否则按种类的缺省（schema_diff.DEFAULT_SEVERITY）。读库出错只记日志、影响面为空：影响面是提示，不挡探查。""" + from app.data import catalog + from app.data import schema_diff as sd + + try: + if notes_by_table is None: + notes_by_table = {t: e.notes for t, e in (await catalog.read_catalog(session, row.id)).items()} + if specs is None: + specs = await published_specs(session) + except Exception: # noqa: BLE001 + logger.exception("计算 %s 的结构变化影响面时读库出错，本次不给影响面", table) + return [({"catalog": [], "templates": [], "rules_touched": False}, d.severity) for d in diffs] + out: list[tuple[dict[str, Any], str]] = [] + for d in diffs: + catalog_items: list[dict[str, Any]] = [] + templates: list[dict[str, Any]] = [] + rules = False + destructive = d.kind in sd.DESTRUCTIVE + if d.kind == sd.TABLE_REMOVED: + catalog_items, rules = catalog_refs(notes_by_table, table, None, before_meta=before_meta, + source_kind=getattr(row, "kind", None)) + templates = template_hits(specs, row.name, table, tables_after=tables_after, removed_columns=None, + table_removed=True) + elif d.kind in (sd.COLUMN_REMOVED, sd.COLUMN_TYPE, sd.COLUMN_NULLABLE, sd.FK) and d.column: + catalog_items, rules = catalog_refs(notes_by_table, table, {d.column}, kind=d.kind, before_meta=before_meta, + source_kind=getattr(row, "kind", None)) + if d.kind in (sd.COLUMN_REMOVED, sd.COLUMN_TYPE): + templates = template_hits(specs, row.name, table, tables_after=tables_after, + removed_columns={d.column} if d.kind == sd.COLUMN_REMOVED else None) + for item in catalog_items: + item["needs_action"] = destructive and item.get("status") == "confirmed" + breaks = any(n.get("breaks") for t in templates for n in t["nodes"]) + severity = "action" if destructive and (breaks or any(i["needs_action"] for i in catalog_items)) else d.severity + out.append(({"catalog": catalog_items, "templates": templates, "rules_touched": rules}, severity)) + return out + + +async def _index_rows(session: AsyncSession, row: Any, names: Iterable[str]) -> dict[str, Any]: + """对象索引里这个源（配置的 schema 下）这几张表的行：{name_key: 行}。""" + from app.data import source_objects + from app.db.models_sources import SourceObject + + keys = {name_key(str(n).rsplit(".", 1)[-1]) for n in names} + if not keys: + return {} + rows = (await session.execute(select(SourceObject).where( + SourceObject.source_id == row.id, SourceObject.schema_name == source_objects.index_schema(row), + SourceObject.name_key.in_(list(keys))))).scalars() + return {r.name_key: r for r in rows} + + +async def schema_versions_for(session: AsyncSession, spec: Any) -> SchemaVersions | None: + """发布时要记下的结构版本：{源名: {表名: 结构版本}}，只记调用工具节点写死的 SQL 直接引用的表（写法照 + catalog_versions_for）。对象索引里没有的表、结构版本还是 0 的表（升级后还没按对象索引读过）不记：不知道那时是哪一版， + 之后不能说它变了。读库出错返回 None（和没记过一样，不提醒）——记不下版本不该挡住发布。""" + try: + sources = await _sources_by_name(session, graph_sql_tables(spec).keys()) + tables_by_source = {name: _schema_tables(view) for name, (_, view) in sources.items()} + used = graph_sql_tables(spec, {k: v for k, v in tables_by_source.items() if v is not None}) + out: SchemaVersions = {} + for name, tables in sorted(used.items()): + if name not in sources: + continue + found = await _index_rows(session, sources[name][0], tables) + pinned = {t: int(found[name_key(t)].struct_version) for t in sorted(tables) + if name_key(t) in found and (found[name_key(t)].struct_version or 0) > 0} + if pinned: + out[name] = pinned + return out + except Exception: # noqa: BLE001 + logger.exception("记录发布时的结构版本失败，本次不记") + return None + + +async def _pending_structure(session: AsyncSession, source_id: str, table: str) -> list[Any]: + from app.data import schema_diff as sd + from app.data.source_changes import PENDING + from app.db.models_sources import SourceChange + + return list((await session.execute(select(SourceChange).where( + SourceChange.source_id == source_id, SourceChange.table_name == table, + SourceChange.kind.in_(sd.STRUCT_KINDS), SourceChange.status.in_(PENDING)) + .order_by(SourceChange.seq))).scalars()) + + +#: 漂移里每张表最多列几条没处理的变化 +DRIFT_CHANGES_SHOWN = 5 + + +async def schema_drift(session: AsyncSession, recorded: Mapping[str, Any] | None) -> list[dict[str, Any]]: + """发布时记下的结构版本和现在比：变了的表 [{source, source_id, table, label, published, current, removed, changes, + more}]。current 是现在的结构版本（表已删除时为 None、removed 为真）；changes 是这张表上在发布之后测到、还没处理的结构 + 变化（[{id, kind, column, severity, summary}]，最多 DRIFT_CHANGES_SHOWN 条，more 是另外还有几条）。 + + 没记过（加这一列之前发布的版本）返回空列表；现在找不到的源跳过；对象索引里找不到的表跳过（换了 schema、索引重建）： + 不知道它现在是哪一版，不能说它变了。读库出错返回空列表：提醒是附加的。""" + from app.data import catalog, source_changes + + if not isinstance(recorded, Mapping) or not recorded: + return [] + try: + sources = await _sources_by_name(session, [str(k) for k in recorded]) + out: list[dict[str, Any]] = [] + for name in sorted(recorded): + pinned = recorded.get(name) + if name not in sources or not isinstance(pinned, Mapping): + continue + row = sources[name][0] + found = await _index_rows(session, row, pinned) + entries = None + for table in sorted(pinned): + then = pinned[table] + obj = found.get(name_key(table)) + if obj is None or not isinstance(then, int): + continue + removed = obj.removed_at is not None + now = None if removed else int(obj.struct_version or 0) + if not removed and now == then: + continue + pending = [c for c in await _pending_structure(session, row.id, obj.name) + if (c.struct_version_after or 0) > then or c.kind == "table_removed"] + if entries is None: + entries = await catalog.read_catalog(session, row.id) + out.append({"source": name, "source_id": row.id, "table": obj.name, + "label": _label(entries.get(obj.name)), "published": then, "current": now, + "removed": removed, + "changes": [{"id": c.id, "kind": c.kind, "column": c.column_name, "severity": c.severity, + "summary": source_changes.summary(c)} for c in pending[:DRIFT_CHANGES_SHOWN]], + "more": max(0, len(pending) - DRIFT_CHANGES_SHOWN)}) + return out + except Exception: # noqa: BLE001 + logger.exception("比对发布时的结构版本失败，本次不提醒") + return [] + + +#: 发布前检查里「这张表上还有没处理的结构变化」的机读码 +SCHEMA_CHANGE_OPEN = "schema.change_open" + + +async def open_structure_issues(session: AsyncSession, spec: Any) -> list[dict[str, Any]]: + """发布前检查要列的：这一版写死的 SQL 直接引用的表上，还没处理的结构变化（警告、要处理两级；说明性的不列）。每张表 + 一条警告，挂在第一个直接引用它的节点上，不拦发布：结构变了不一定影响这条 SQL，真删了列查询本身会照实报错。读库出错 + 返回空列表。""" + from app.data import source_changes + + try: + sources = await _sources_by_name(session, graph_sql_tables(spec).keys()) + tables_by_source = {name: _schema_tables(view) for name, (_, view) in sources.items()} + used = graph_sql_tables(spec, {k: v for k, v in tables_by_source.items() if v is not None}) + issues: list[dict[str, Any]] = [] + for name, tables in sorted(used.items()): + if name not in sources: + continue + row = sources[name][0] + for table in sorted(tables): + pending = [c for c in await _pending_structure(session, row.id, table) + if c.severity in ("warning", "action")] + if not pending: + continue + refs = [r for r in template_refs(spec, name, table, tables_by_source.get(name)) if r["impact"] == "direct"] + shown = "；".join(source_changes.summary(c) for c in pending[:3]) + more = f"等 {len(pending)} 条" if len(pending) > 3 else "" + issues.append({ + "level": "warning", "node_id": refs[0]["node_id"] if refs else None, "edge_id": None, "field": None, + "code": SCHEMA_CHANGE_OPEN, "fix": None, + "message": (f"数据源「{name}」的表 {table} 有还没处理的结构变化：{shown}{more}。这一版写死的 SQL 用到了" + "这张表，发布前请核对这几处 SQL 是否还对（不拦发布）"), + }) + return issues + except Exception: # noqa: BLE001 + logger.exception("发布前检查结构变化时出错，本次不列") + return [] + + +__all__ = ["IMPACTS", "SCHEMA_CHANGE_OPEN", "CatalogVersions", "SchemaVersions", "agent_bindings", "agent_sources", + "catalog_drift", "catalog_refs", "catalog_versions_for", "graph_sql_tables", "open_structure_issues", + "published_specs", "recorded_versions", "schema_drift", "schema_versions_for", "sql_table_keys", + "structure_impact", "table_impact", "template_hits", "template_refs", "tool_queries"] diff --git a/backend/app/data/introspect_jobs.py b/backend/app/data/introspect_jobs.py index 244377a9..137f2b6a 100644 --- a/backend/app/data/introspect_jobs.py +++ b/backend/app/data/introspect_jobs.py @@ -19,6 +19,12 @@ - **保留**（prune）：每个源保留最近 KEEP_JOBS 个结束了的探查任务。 任务最后的状态：有表读不到是 partial，否则 succeeded；整体出错（数据源没了、连不上）是 failed。 + +**结构变化**（第 3 波，next-round 03 B1）：探查顺带把前后差异记进 source_changes（data/source_changes.py），每条带影响面 +（data/catalog_impact.structure_impact）。刷新对象索引时记新出现的表、真删除的表（第一次建索引只建基线）；开始执行时记 +离开范围的、有数据目录的表，以及范围规则和上一次探查不同（scope_changed）；逐批读完时拿上一次的结构和这一次的比 +（data/schema_diff.diff_table），在范围里却读不到结构的记 table_reflect_failed（不算删除）。升级后第一次读到（结构版本 0） +只建基线。记变化出错只记日志，不挡探查。任务上记 trigger（manual 人点的 / schedule 定时维护）和定时的 fire_at。 """ from __future__ import annotations @@ -271,6 +277,7 @@ async def preview(session: AsyncSession, row: Any, *, schema: str | None = None, return out async with so.source_lock(row.id): stats = await so.refresh_index(session, row.id, configured, result, legacy=row.schema_cache) + await _record_index_changes(session, row, stats, _detected(None, "manual", None)) await session.commit() plan = await so.plan_for(session, row, rules) out.update(stored=True, objects=[o.name for o in result.objects], scope=plan.summary(), added=stats.added, @@ -278,9 +285,10 @@ async def preview(session: AsyncSession, row: Any, *, schema: str | None = None, return out -async def prepare(session: AsyncSession, row: Any, *, actor: str | None, - trigger: str = "manual") -> tuple[IntrospectJob, int]: - """建一个排队的探查任务（还没开始，调用方接着 manager.start）。返回 (任务, 几张表)。 +async def prepare(session: AsyncSession, row: Any, *, actor: str | None, trigger: str = "manual", + fire_at: str | None = None, schedule_id: str | None = None) -> tuple[IntrospectJob, int]: + """建一个排队的探查任务（还没开始，调用方接着 manager.start）。返回 (任务, 几张表)。trigger 是 manual（人点的）或 + schedule（定时维护：fire_at、schedule_id 记进任务的 scope，留痕这一次是哪一次触发建的）。 上传表格、记录表抛 IntrospectRefused；同源已有排队或进行中的探查任务抛 IntrospectBusy；对象清单拿不到抛 IntrospectFailed（失败照实记进 schema_cache）；规则命中超过上限抛 ScopeTooLarge（对象索引照样刷新了）。 @@ -306,6 +314,7 @@ async def prepare(session: AsyncSession, row: Any, *, actor: str | None, rules = so.ScopeRules.from_options(row.options) plan = await so.plan_for(session, row, rules) if plan.over: + await _record_index_changes(session, row, stats, _detected(None, trigger, actor, fire_at)) await session.commit() raise ScopeTooLarge(plan) summary = {**plan.summary(rules), "index_method": stats.method, "index_capped": stats.capped, @@ -313,6 +322,10 @@ async def prepare(session: AsyncSession, row: Any, *, actor: str | None, "adopted": stats.adopted} if candidates: summary["available_schemas"] = candidates + if fire_at: + summary["fire_at"] = fire_at + if schedule_id: + summary["schedule_id"] = schedule_id job = IntrospectJob(id=new_id(), source_id=source_id, status="queued", trigger=trigger, schema_name=schema, scope=summary, created_by=actor) session.add(job) @@ -323,6 +336,7 @@ async def prepare(session: AsyncSession, row: Any, *, actor: str | None, raise IntrospectBusy(busy_message(SimpleNamespace(name=name), None)) from e session.add_all([IntrospectTask(job_id=job.id, seq=i, table_name=table, reason=reason) for i, (table, reason) in enumerate(plan.in_scope.items())]) + await _record_index_changes(session, row, stats, _detected(job.id, trigger, actor, fire_at)) await session.commit() try: await prune(session, source_id) @@ -345,6 +359,93 @@ async def prune(session: AsyncSession, source_id: str, *, keep: int = KEEP_JOBS) return len(victims) +# ========================================================================== +# 结构变化（next-round 03 B1） +# ========================================================================== + +#: 一次刷新对象索引最多记几条「新出现的表」（几千个按月新建的分区表这类：照实记前面的，其余只进日志） +ADDED_RECORDED = 200 + + +def _detected(job_id: str | None, trigger: str, actor: str | None, fire_at: str | None = None, + task_seq: int | None = None) -> Any: + from app.data.source_changes import Detected + + return Detected(job_kind="introspect", job_id=job_id, task_seq=task_seq, trigger=trigger or "manual", + actor=actor, fire_at=fire_at) + + +async def _record_index_changes(session: AsyncSession, row: Any, stats: so.RefreshStats, detected: Any) -> None: + """刷新对象索引得出的表级变化（不提交）：新出现的表（table_added；这个 schema 第一次建索引时全是新的，只建基线）、 + 真删除的表（table_removed，带影响面：目录里这张表的项、别的表指向它的关联关系、写死它的已发布模板）。出错只记日志。""" + if not stats.removed_names and not (stats.added_names and not stats.first_build): + return + from app.data import catalog_impact + from app.data import schema_diff as sd + from app.data import source_changes as sc + + try: + now = utcnow().isoformat() + changes: list[sc.NewChange] = [] + added = [] if stats.first_build else list(stats.added_names) + if len(added) > ADDED_RECORDED: + logger.warning("数据源 %s 这次新出现 %d 个对象，变化记录只记前 %d 个", row.name, len(added), ADDED_RECORDED) + for name in added[:ADDED_RECORDED]: + changes.append(sc.NewChange(kind=sd.TABLE_ADDED, table=name, severity="info", key=now)) + if stats.removed_names: + schema = so.index_schema(row) + gone = {r.name: r for r in (await session.execute(select(SourceObject).where( + SourceObject.source_id == row.id, SourceObject.schema_name == schema, + SourceObject.name.in_(stats.removed_names)))).scalars()} + tables_after = dict((row.schema_cache or {}).get("tables") or {}) + specs = await catalog_impact.published_specs(session) + for name in stats.removed_names: + obj = gone.get(name) + version = int(obj.struct_version or 0) if obj is not None else 0 + meta = obj.meta if obj is not None and isinstance(obj.meta, dict) else None + [(affects, severity)] = await catalog_impact.structure_impact( + session, row, name, [sd.Diff(sd.TABLE_REMOVED)], before_meta=meta, + tables_after={k: v for k, v in tables_after.items() if k != name}, specs=specs) + changes.append(sc.NewChange( + kind=sd.TABLE_REMOVED, table=name, severity=severity, affects=affects, + before={"kind": obj.kind if obj is not None else "table"}, struct_version_before=version or None, + key=(obj.removed_at.isoformat() if obj is not None and obj.removed_at else now))) + await sc.record_all(session, row, changes, detected) + except Exception: # noqa: BLE001 - 记变化出错不挡探查 + logger.exception("数据源 %s 记录表级的结构变化时出错", getattr(row, "name", "?")) + + +async def _record_scope_changes(session: AsyncSession, row: Any, job: IntrospectJob, leaving: list[str], + entering: int) -> None: + """开始执行、把范围写进对象索引时（不提交）：有数据目录的表离开了范围（table_left_scope，不算删除），和范围规则跟上一次 + 探查不同（scope_changed：前后的规则、进出各几张）。出错只记日志。""" + from app.data import catalog + from app.data import schema_diff as sd + from app.data import source_changes as sc + + try: + detected = _detected(job.id, job.trigger, job.created_by, (job.scope or {}).get("fire_at")) + changes: list[sc.NewChange] = [] + if leaving: + noted = set((await catalog.read_catalog(session, row.id)).keys()) + for name in leaving: + if name in noted: + changes.append(sc.NewChange(kind=sd.TABLE_LEFT_SCOPE, table=name, severity="info", key=job.id)) + previous = (await session.execute( + select(IntrospectJob).where(IntrospectJob.source_id == job.source_id, IntrospectJob.id != job.id, + IntrospectJob.started_at.is_not(None), + IntrospectJob.created_at <= job.created_at) + .order_by(IntrospectJob.created_at.desc()).limit(1))).scalar_one_or_none() + before = (previous.scope or {}).get("rules") if previous is not None else None + after = (job.scope or {}).get("rules") + if isinstance(before, dict) and isinstance(after, dict) and before != after: + changes.append(sc.NewChange(kind=sd.SCOPE, table="", severity="info", before={"rules": before}, + after={"rules": after, "entered": entering, "left": len(leaving)}, key=job.id)) + await sc.record_all(session, row, changes, detected) + except Exception: # noqa: BLE001 + logger.exception("数据源 %s 记录探查范围的变化时出错", getattr(row, "name", "?")) + + # ========================================================================== # 收尾、停止、启动时的整理 # ========================================================================== @@ -425,12 +526,32 @@ async def startup() -> int: # ========================================================================== +def _side(diff: Any, which: str) -> Any: + """一条差异的前（before）或后（after），疑似改名的另一半记在这一侧（{…, rename}）。""" + value = getattr(diff, which) + if diff.rename: + base = dict(value) if isinstance(value, dict) else ({"value": value} if value is not None else {}) + return {**base, "rename": dict(diff.rename)} + return value + + +def json_key(value: Any) -> str: + import json + + return json.dumps(value, sort_keys=True, ensure_ascii=False, default=str) + + class _IntrospectRun(runtime.JobRun): """一个探查任务在后台的一次执行（进队、停止、收尾的框架在 runtime.JobRun）。""" kind = "introspect" model = IntrospectJob + def __init__(self, job_id: str) -> None: + super().__init__(job_id) + #: 已发布模板和解析好的图（catalog_impact.published_specs）：这次探查里各批的影响面共用，第一次用到时读 + self._specs: list[Any] | None = None + async def _stopped(self) -> None: if self.stop_reason == "user": await self._end("cancelled", None, STOPPED_ERROR) @@ -510,8 +631,15 @@ class _IntrospectRun(runtime.JobRun): counts = {r: 0 for r in so.SCOPE_REASONS} for reason in reasons.values(): counts[reason] = counts.get(reason, 0) + 1 + before = set((await session.execute(select(SourceObject.name).where( + SourceObject.source_id == source_id, SourceObject.schema_name == schema, + SourceObject.in_scope.is_(True)))).scalars()) plan = so.ScopePlan(in_scope=reasons, counts=counts, cap=0, index_total=0, out_of_scope=0) await so.apply_scope(session, source_id, schema, plan) + row = await session.get(DataSource, source_id) + if job is not None and row is not None: + await _record_scope_changes(session, row, job, sorted(before - set(reasons)), + len(set(reasons) - before)) await session.commit() async def _recheck(self, session: AsyncSession, source_id: str) -> Any: @@ -540,8 +668,15 @@ class _IntrospectRun(runtime.JobRun): async def _save_batch(self, session: AsyncSession, source_id: str, schema: str, batch: list[tuple[int, str]], got: so.Reflected) -> None: - """一批读完：读到的结构写进对象索引，逐表记 ok / failed 和结构版本。""" + """一批读完：读到的结构写进对象索引，逐表记 ok / failed 和结构版本；拿上一次的结构和这一次的比，变化记进 + source_changes（同一个事务）。""" + old = {r.name: r for r in (await session.execute( + select(SourceObject.name, SourceObject.meta, SourceObject.struct_version, SourceObject.fingerprint, + SourceObject.reflect_error).where( + SourceObject.source_id == source_id, SourceObject.schema_name == schema, + SourceObject.name.in_([name for _, name in batch])))).all()} versions = await so.save_reflected(session, source_id, schema, got) + await self._record_diffs(session, source_id, batch, got, old, versions) now = utcnow() for seq, name in batch: if name in got.tables: @@ -554,6 +689,60 @@ class _IntrospectRun(runtime.JobRun): .execution_options(synchronize_session=False)) await session.commit() + async def _record_diffs(self, session: AsyncSession, source_id: str, batch: list[tuple[int, str]], + got: so.Reflected, old: dict[str, Any], versions: dict[str, int]) -> None: + """这一批的结构差异（不提交）：读到的表指纹变了、又不是基线（schema_diff.is_baseline）的，逐条比出变化、算影响面； + 以前读得到、这次读不到的记 table_reflect_failed。出错只记日志，不挡探查。""" + from app.data import catalog_impact + from app.data import schema_diff as sd + from app.data import source_changes as sc + + try: + todo: list[tuple[int, str, list[Any], dict[str, Any] | None]] = [] + failed: list[tuple[int, str]] = [] + for seq, name in batch: + prev = old.get(name) + if prev is None: + continue + if name in got.tables: + meta = got.tables[name] + if sd.is_baseline(prev.struct_version, prev.meta) or so.fingerprint(meta) == prev.fingerprint: + continue + diffs = sd.diff_table(prev.meta, meta) + if diffs: + todo.append((seq, name, diffs, prev.meta)) + elif name in got.skipped and isinstance(prev.meta, dict) and prev.meta and not prev.reflect_error: + failed.append((seq, name)) + if not todo and not failed: + return + job = await session.get(IntrospectJob, self.job_id) + row = await session.get(DataSource, source_id) + if job is None or row is None: + return + fire_at = (job.scope or {}).get("fire_at") + if self._specs is None: + self._specs = await catalog_impact.published_specs(session) + tables_after = {**dict((row.schema_cache or {}).get("tables") or {}), **got.tables} + for seq, name, diffs, before in todo: + impacts = await catalog_impact.structure_impact(session, row, name, diffs, before_meta=before, + tables_after=tables_after, specs=self._specs) + after_v, before_v = versions.get(name), old[name].struct_version + changes = [sc.NewChange(kind=d.kind, table=name, column=d.column, severity=severity, + before=_side(d, "before"), after=_side(d, "after"), affects=affects, + struct_version_before=before_v, struct_version_after=after_v, + key=(after_v, d.key, json_key(d.before), json_key(d.after))) + for d, (affects, severity) in zip(diffs, impacts)] + await sc.record_all(session, row, changes, + _detected(self.job_id, job.trigger, job.created_by, fire_at, seq)) + for seq, name in failed: + await sc.record(session, row, sc.NewChange( + kind=sd.TABLE_REFLECT_FAILED, table=name, severity="warning", + after={"error": got.skipped.get(name)}, struct_version_before=old[name].struct_version, + key=(old[name].struct_version, got.skipped.get(name))), + _detected(self.job_id, job.trigger, job.created_by, fire_at, seq)) + except Exception: # noqa: BLE001 + logger.exception("探查任务 %s 记录结构变化时出错", self.job_id) + async def _finish(self, session: AsyncSession, source_id: str) -> None: """读完：物化 schema_cache、盖同步时间；有表读不到是 partial，否则 succeeded。""" failed = int((await session.execute(select(func.count()).select_from(IntrospectTask).where( diff --git a/backend/app/data/schema_diff.py b/backend/app/data/schema_diff.py new file mode 100644 index 00000000..43bd7c38 --- /dev/null +++ b/backend/app/data/schema_diff.py @@ -0,0 +1,216 @@ +"""表结构的前后差异（docs/design/next-round/03-数据接入与目录维护.md B1）：纯函数，不连库、不读目录。 + +**为什么要有它。** 以前重新探查是整份替换 schema_cache，不比前后：删了一列，那一列上人工确认过的目录项、写死了这一列的 +已发布模板都收不到提醒，正式运行要到查询那一步才报错。现在每张表有结构版本（data/source_objects：指纹变了才加 1）， +后台探查读到新结构时拿上一次的结构和这一次的比（diff_table），变化写进 source_changes（data/source_changes.py），再由 +影响面（data/catalog_impact.structure_impact）说清哪些目录项、哪些写死的 SQL 会受影响。 + +**表级**的四种由调用方判（它知道对象索引和范围）：table_added（对象索引里新出现的）、table_removed（对象索引里也没了， +真删除）、table_left_scope（离开了探查范围）、table_reflect_failed（在范围里、这次读不到结构）。后两种**不算删除** +（is_deletion）：范围外、读不到结构时，目录和模板里说到它的照旧有效，只是这次没看到。 + +**列级**按列名比：column_added、column_removed、column_type_changed（比的是按方言编译的完整写法，VARCHAR2(20 CHAR)、 +ENUM('a','b')，见 introspect.type_text）、column_nullable_changed、comment_changed（列注释，表注释的 column 为 None）。 +**键**：主键、唯一约束的增删（key_changed），外键的增删（fk_changed）。反射时读不到唯一约束、外键的表不写那两个键： +两边有一边没有就不比（不知道 ≠ 删了）。 + +**疑似改名**：被删的列和新加的列类型相同、在列清单里的位置相同（夹在同样的两列之间），而且这样的配对只有一对时，两条变化都带 rename， +说法只写「可能是改名（未确认）」——库里只看得到删了一列、加了一列，不下结论。 + +**基线**（is_baseline）：升级后第一次按对象索引读到这张表（结构版本还是 0：升级前同步探查的老缓存接进来的那份、或者 +从没读过）只建基线、不出变化。列类型的写法第 1 波从通用改成了方言完整写法（目录-D10），拿老缓存的 VARCHAR2 和新读的 +VARCHAR2(20 CHAR) 比，所有 Oracle 列都会被说成「类型变了」。 +""" +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass, field +from typing import Any + +#: 表级的四种（调用方判） +TABLE_ADDED = "table_added" +TABLE_REMOVED = "table_removed" +TABLE_LEFT_SCOPE = "table_left_scope" +TABLE_REFLECT_FAILED = "table_reflect_failed" +#: 列级、键 +COLUMN_ADDED = "column_added" +COLUMN_REMOVED = "column_removed" +COLUMN_TYPE = "column_type_changed" +COLUMN_NULLABLE = "column_nullable_changed" +COMMENT = "comment_changed" +KEY = "key_changed" +FK = "fk_changed" +#: 探查范围的规则改了（源级，table 为空） +SCOPE = "scope_changed" + +STRUCT_KINDS = (TABLE_ADDED, TABLE_REMOVED, TABLE_LEFT_SCOPE, TABLE_REFLECT_FAILED, COLUMN_ADDED, COLUMN_REMOVED, + COLUMN_TYPE, COLUMN_NULLABLE, KEY, FK, COMMENT, SCOPE) + +#: 没有影响面时的严重程度（影响面会升级：已确认的项、写死的 SQL 指向被删的列时是 action） +DEFAULT_SEVERITY = { + TABLE_ADDED: "info", TABLE_REMOVED: "warning", TABLE_LEFT_SCOPE: "info", TABLE_REFLECT_FAILED: "warning", + COLUMN_ADDED: "info", COLUMN_REMOVED: "warning", COLUMN_TYPE: "warning", COLUMN_NULLABLE: "info", + KEY: "warning", FK: "warning", COMMENT: "info", SCOPE: "info", +} + +#: 会让说到它的目录项、写死的 SQL 失效的：表真删了、列删了 +DESTRUCTIVE = frozenset({TABLE_REMOVED, COLUMN_REMOVED}) + + +def is_deletion(kind: str) -> bool: + """这种变化是不是删除。离开范围、读不到结构都不是：范围外的表还在库里，读不到结构的上一次的结构照旧留着。""" + return kind in DESTRUCTIVE + + +def is_baseline(old_version: int | None, old_meta: Mapping[str, Any] | None) -> bool: + """这一次读到的结构只建基线、不和以前比：结构版本还是 0（升级前的老缓存接进来的、从没按对象索引读过），或者上一次的 + 结构不在手上（离开过范围，结构清掉了）。""" + return not old_version or not isinstance(old_meta, Mapping) or not old_meta.get("columns") + + +@dataclass(frozen=True) +class Diff: + """一条变化。column 是列级变化的那一列（表注释、键、外键为 None）；before / after 是前后各是什么（列的类型、可空、注释， + 键的列组，外键的两端）；rename 是疑似改名的另一半 {from, to}（只说可能）；key 是键的种类（primary / unique）。""" + + kind: str + column: str | None = None + before: Any = None + after: Any = None + rename: dict[str, str] | None = None + key: str | None = None + extra: dict[str, Any] = field(default_factory=dict) + + @property + def severity(self) -> str: + return DEFAULT_SEVERITY.get(self.kind, "info") + + def to_dict(self) -> dict[str, Any]: + out: dict[str, Any] = {"kind": self.kind, "column": self.column, "before": self.before, "after": self.after} + if self.rename: + out["rename"] = dict(self.rename) + if self.key: + out["key"] = self.key + if self.extra: + out.update(self.extra) + return out + + +def _columns(meta: Mapping[str, Any] | None) -> list[dict[str, Any]]: + return [c for c in (meta or {}).get("columns") or [] if isinstance(c, Mapping) and c.get("name") is not None] + + +def _text(value: Any) -> str | None: + text = str(value).strip() if value is not None else "" + return text or None + + +def _groups(raw: Any) -> set[tuple[str, ...]]: + return {tuple(str(c) for c in g) for g in raw or [] if isinstance(g, (list, tuple)) and g} + + +def _fks(raw: Any) -> dict[tuple[Any, ...], dict[str, Any]]: + out: dict[tuple[Any, ...], dict[str, Any]] = {} + for fk in raw or []: + if not isinstance(fk, Mapping): + continue + cols = tuple(str(c) for c in fk.get("columns") or []) + key = (cols, str(fk.get("to_table") or ""), tuple(str(c) for c in fk.get("to_columns") or [])) + out[key] = {"columns": list(key[0]), "to_table": key[1], "to_columns": list(key[2])} + return out + + +def _slots_of(names: list[str], common: set[str]) -> dict[str, tuple[str | None, str | None]]: + """每一列在列清单里的「位置」：它前后最近的、两次都在的那两列。按下标比会被别处的删列、加列带偏（删了前面一列，后面 + 的下标全都前移一位），按前后的邻居比才说得出「还在原来那个位置」。""" + out: dict[str, tuple[str | None, str | None]] = {} + for i, name in enumerate(names): + before = next((n for n in reversed(names[:i]) if n in common), None) + after = next((n for n in names[i + 1:] if n in common), None) + out[name] = (before, after) + return out + + +def diff_table(before: Mapping[str, Any] | None, after: Mapping[str, Any] | None) -> list[Diff]: + """同一张表前后两次的结构 → 变化（列级、键）。顺序：删掉的列、加上的列（各按列清单的顺序），再是各列的类型、可空、 + 注释，表注释，主键、唯一约束，外键。两边一样返回空列表。是不是基线（该不该比）由调用方先用 is_baseline 判。""" + old_cols = _columns(before) + new_cols = _columns(after) + old_by = {str(c["name"]): c for c in old_cols} + new_by = {str(c["name"]): c for c in new_cols} + removed = [n for n in old_by if n not in new_by] + added = [n for n in new_by if n not in old_by] + common = set(old_by) & set(new_by) + old_pos = _slots_of([str(c["name"]) for c in old_cols], common) + new_pos = _slots_of([str(c["name"]) for c in new_cols], common) + + # 疑似改名：被删的和新加的类型相同、位置相同（夹在同样的两列之间）；一个被删的只配得上一个新加的、反过来也是，才算 + pairs: dict[str, str] = {} + for r in removed: + hits = [a for a in added if old_pos[r] == new_pos[a] and old_by[r].get("type") == new_by[a].get("type")] + if len(hits) == 1: + back = [x for x in removed if old_pos[x] == new_pos[hits[0]] + and old_by[x].get("type") == new_by[hits[0]].get("type")] + if back == [r]: + pairs[r] = hits[0] + reverse = {a: r for r, a in pairs.items()} + + out: list[Diff] = [] + for name in removed: + col = old_by[name] + out.append(Diff(COLUMN_REMOVED, name, before={"type": col.get("type"), "nullable": col.get("nullable", True)}, + rename={"from": name, "to": pairs[name]} if name in pairs else None)) + for name in added: + col = new_by[name] + out.append(Diff(COLUMN_ADDED, name, after={"type": col.get("type"), "nullable": col.get("nullable", True)}, + rename={"from": reverse[name], "to": name} if name in reverse else None)) + for name, col in new_by.items(): + old = old_by.get(name) + if old is None: + continue + if (old.get("type") or "") != (col.get("type") or ""): + out.append(Diff(COLUMN_TYPE, name, before=old.get("type"), after=col.get("type"))) + if bool(old.get("nullable", True)) != bool(col.get("nullable", True)): + out.append(Diff(COLUMN_NULLABLE, name, before=bool(old.get("nullable", True)), + after=bool(col.get("nullable", True)))) + if _text(old.get("comment")) != _text(col.get("comment")): + out.append(Diff(COMMENT, name, before=_text(old.get("comment")), after=_text(col.get("comment")))) + if _text((before or {}).get("comment")) != _text((after or {}).get("comment")): + out.append(Diff(COMMENT, None, before=_text((before or {}).get("comment")), + after=_text((after or {}).get("comment")))) + + old_pk = [str(c) for c in (before or {}).get("primary_key") or []] + new_pk = [str(c) for c in (after or {}).get("primary_key") or []] + if old_pk != new_pk: + out.append(Diff(KEY, None, before=old_pk or None, after=new_pk or None, key="primary")) + # 唯一约束、外键：反射读不到时不写这个键，两边都有才比 + if "unique" in (before or {}) and "unique" in (after or {}): + old_u, new_u = _groups(before.get("unique")), _groups(after.get("unique")) # type: ignore[union-attr] + for g in sorted(old_u - new_u): + out.append(Diff(KEY, g[0] if len(g) == 1 else None, before=list(g), after=None, key="unique")) + for g in sorted(new_u - old_u): + out.append(Diff(KEY, g[0] if len(g) == 1 else None, before=None, after=list(g), key="unique")) + if "foreign_keys" in (before or {}) and "foreign_keys" in (after or {}): + old_f, new_f = _fks(before.get("foreign_keys")), _fks(after.get("foreign_keys")) # type: ignore[union-attr] + for k in sorted(set(old_f) - set(new_f)): + cols = old_f[k]["columns"] + out.append(Diff(FK, cols[0] if len(cols) == 1 else None, before=old_f[k], after=None)) + for k in sorted(set(new_f) - set(old_f)): + cols = new_f[k]["columns"] + out.append(Diff(FK, cols[0] if len(cols) == 1 else None, before=None, after=new_f[k])) + return out + + +def touched_columns(diffs: list[Diff]) -> dict[str, set[str]]: + """各条变化说到的列，按种类分：{"removed": {…}, "changed": {…}}。removed 是被删的列（含疑似改名的旧名），changed 是 + 类型、可空变了的列。影响面按它找目录里说到这些列的项。""" + removed = {d.column for d in diffs if d.kind == COLUMN_REMOVED and d.column} + changed = {d.column for d in diffs if d.kind in (COLUMN_TYPE, COLUMN_NULLABLE) and d.column} + return {"removed": removed, "changed": changed} + + +__all__ = [ + "COLUMN_ADDED", "COLUMN_NULLABLE", "COLUMN_REMOVED", "COLUMN_TYPE", "COMMENT", "DEFAULT_SEVERITY", "DESTRUCTIVE", + "Diff", "FK", "KEY", "SCOPE", "STRUCT_KINDS", "TABLE_ADDED", "TABLE_LEFT_SCOPE", "TABLE_REFLECT_FAILED", + "TABLE_REMOVED", "diff_table", "is_baseline", "is_deletion", "touched_columns", +] diff --git a/backend/app/data/source_changes.py b/backend/app/data/source_changes.py new file mode 100644 index 00000000..3e43d61c --- /dev/null +++ b/backend/app/data/source_changes.py @@ -0,0 +1,492 @@ +"""数据源的变化记录（docs/design/next-round/03-数据接入与目录维护.md B3）：结构变化、取值变化的记录、去重、处理和拉取。 + +**为什么要有它。** 以前表结构变了、码值多了一个新取值都没有地方记：审阅页看不到，通知发不了，口径账本的召回也没有可以 +订阅的源。现在全站只有 source_changes 这一份记录（格式 agentlab-change/1）： + +- **记**（record）：后台探查读到的结构差异（data/schema_diff → introspect_jobs）、取值观察和同口径历史的比较 + （data/value_drift → catalog_profile_observe）都经这里写。fingerprint 去重：同一个变化再次测到只更新 last_seen_at、 + seen_count（还开着的顺带换上最新的影响面和严重程度），不另起一条。 +- **处理**（handle / resolve）：动作都落到目录已有的写入接口上——清理目录里说到被删的列的项（catalog.write_entry， + 整份写、乐观锁）、加入码值（catalog.save_patch：apply_patch 配 _merge_codes，含义待填写，依据写这条变化的编号）、重新 + 起草（建起草任务）、重新剖析（建剖析任务）、知道了、误报（必须写理由）。每个动作写 resolution，动了目录的记下目录版本， + 建了任务的记下任务编号。 +- **拉取**（list_changes / affecting）：给召回按 seq 游标拉；seq 自增、不复用。触发召回的只有结构类、处理时改了目录的、 + 以及 rules_touched 的新码值（triggers_recall）；纯取值漂移只算数据质量提醒。 +- **说法**（summary）：一条变化给人看的一句（接口、日志用）；通知里不放它（不放表名、列名、取值，D9），见 + notify/service.notify_source_changed。 + +不记遮罩列的取值（观察查询本来就不碰遮罩列）。 +""" +from __future__ import annotations + +import hashlib +import json +import logging +from collections.abc import Iterable, Mapping +from dataclasses import dataclass +from datetime import datetime +from typing import Any + +from sqlalchemy import func, or_, select +from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncSession + +from app.data import schema_diff as sd +from app.data.names import name_key +from app.db.base import utcnow +from app.db.models_sources import SourceChange + +logger = logging.getLogger(__name__) + +SCHEMA = "agentlab-change/1" + +#: 取值类的变化（data/value_drift） +CODES_NEW = "codes_new" +CODES_SHIFT = "codes_shift" +DATE_STALLED = "date_stalled" +DATE_GAP = "date_gap" +VOLUME_DROP = "volume_drop" +ROWS_DROP = "rows_drop" +ROWS_JUMP = "rows_jump" +COVERAGE_DROP = "coverage_drop" +VALUE_KINDS = (CODES_NEW, CODES_SHIFT, DATE_STALLED, DATE_GAP, VOLUME_DROP, ROWS_DROP, ROWS_JUMP, COVERAGE_DROP) +KINDS = sd.STRUCT_KINDS + VALUE_KINDS + +SEVERITIES = ("info", "warning", "action") +SEVERITY_RANK = {"info": 1, "warning": 2, "action": 3} +STATUSES = ("open", "acknowledged", "resolved", "dismissed") +#: 还要人看的：开着的、知道了但还没处理的 +PENDING = ("open", "acknowledged") + +#: 处理动作：clean_catalog 清理目录里说到这些列的项、add_codes 加入码值、redraft 重新起草这张表、reprofile 重新剖析、 +#: acknowledge 知道了、dismiss 误报（必须写理由） +ACTIONS = ("clean_catalog", "add_codes", "redraft", "reprofile", "acknowledge", "dismiss") +#: 动作之后的状态 +ACTION_STATUS = {"clean_catalog": "resolved", "add_codes": "resolved", "redraft": "resolved", "reprofile": "resolved", + "acknowledge": "acknowledged", "dismiss": "dismissed"} +#: 理由最长几个字 +NOTE_MAX = 500 + +#: 变化的叫法（接口给人看的 summary 用；前端第 4 波的变化标签页在 lib/terms.ts 另有一份，逐字对照） +KIND_LABEL = { + sd.TABLE_ADDED: "新出现的表", sd.TABLE_REMOVED: "表已删除", sd.TABLE_LEFT_SCOPE: "表离开了探查范围", + sd.TABLE_REFLECT_FAILED: "这次读不到表结构", sd.COLUMN_ADDED: "新加了列", sd.COLUMN_REMOVED: "删除了列", + sd.COLUMN_TYPE: "列的类型变了", sd.COLUMN_NULLABLE: "列的可空变了", sd.KEY: "键变了", sd.FK: "外键变了", + sd.COMMENT: "注释变了", sd.SCOPE: "探查范围的规则变了", + CODES_NEW: "出现了新码值", CODES_SHIFT: "码值的比例变化大", DATE_STALLED: "最新业务日期没有前进", + DATE_GAP: "有的日子没有数据", VOLUME_DROP: "按日行数大幅下跌", ROWS_DROP: "行数大幅下跌", ROWS_JUMP: "行数大幅上涨", + COVERAGE_DROP: "关联的覆盖率下降", +} + + +class ChangeNotFound(Exception): + pass + + +class ChangeFinished(Exception): + """这条变化已经处理过（resolved / dismissed）了。""" + + +class ChangeInvalid(ValueError): + """处理动作不对：不认识的动作、误报没写理由、这种变化做不了这个动作。message 给人看。""" + + +# ========================================================================== +# 记 +# ========================================================================== + + +def fingerprint(*parts: Any) -> str: + """去重用的指纹：几段内容的规范 JSON 的 sha256。""" + text = json.dumps(list(parts), sort_keys=True, ensure_ascii=False, separators=(",", ":"), default=str) + return hashlib.sha256(text.encode("utf-8")).hexdigest() + + +@dataclass +class Detected: + """谁、经什么测到的（detected_by）：job_kind 是 introspect / profile / on_demand，trigger 是 manual / schedule / + on_demand，actor 是署名。""" + + job_kind: str + job_id: str | None = None + task_seq: int | None = None + trigger: str = "manual" + actor: str | None = None + fire_at: str | None = None + + def to_dict(self) -> dict[str, Any]: + out = {"job_kind": self.job_kind, "job_id": self.job_id, "task_seq": self.task_seq, "trigger": self.trigger, + "actor": self.actor} + if self.fire_at: + out["fire_at"] = self.fire_at + return out + + +@dataclass +class NewChange: + """要记的一条变化（还没去重）。key 是去重指纹的组成部分（除了源、种类、表、列以外的那部分：结构版本、码值……）。""" + + kind: str + table: str + column: str | None = None + severity: str = "info" + before: Any = None + after: Any = None + basis: dict[str, Any] | None = None + affects: dict[str, Any] | None = None + struct_version_before: int | None = None + struct_version_after: int | None = None + key: Any = None + + +async def record(session: AsyncSession, source: Any, change: NewChange, detected: Detected) -> tuple[SourceChange, bool]: + """记一条变化（不提交）。同一个指纹已经有了：只更新 last_seen_at、seen_count（还开着的另换上最新的影响面、前后值、 + 严重程度），返回 (那一条, False)；新的返回 (新的一条, True)。""" + if change.kind not in KINDS: + raise ValueError(f"不认识的变化种类：{change.kind}") + if change.severity not in SEVERITIES: + raise ValueError(f"不认识的严重程度：{change.severity}") + fp = fingerprint(change.kind, change.table, change.column, change.key) + now = utcnow() + found = (await session.execute(select(SourceChange).where( + SourceChange.source_id == source.id, SourceChange.fingerprint == fp))).scalar_one_or_none() + if found is not None: + found.last_seen_at = now + found.seen_count = int(found.seen_count or 0) + 1 + if found.status == "open": + found.affects, found.before, found.after = change.affects, change.before, change.after + if SEVERITY_RANK[change.severity] > SEVERITY_RANK.get(found.severity, 0): + found.severity = change.severity + return found, False + row = SourceChange(source_id=source.id, source_name=str(getattr(source, "name", "") or ""), + table_name=change.table or "", column_name=change.column, kind=change.kind, + severity=change.severity, detected_at=now, last_seen_at=now, seen_count=1, + detected_by=detected.to_dict(), basis=change.basis, before=change.before, after=change.after, + affects=change.affects, struct_version_before=change.struct_version_before, + struct_version_after=change.struct_version_after, fingerprint=fp, status="open", + schema_tag=SCHEMA) + try: + async with session.begin_nested(): + session.add(row) + except IntegrityError: + # 同一个指纹刚被别处写进去（单进程下几乎不会）：按已有的那条更新 + found = (await session.execute(select(SourceChange).where( + SourceChange.source_id == source.id, SourceChange.fingerprint == fp))).scalar_one() + found.last_seen_at, found.seen_count = now, int(found.seen_count or 0) + 1 + return found, False + return row, True + + +async def record_all(session: AsyncSession, source: Any, changes: Iterable[NewChange], + detected: Detected) -> list[SourceChange]: + """一批变化逐条 record（不提交），返回这次新记的那几条。""" + fresh: list[SourceChange] = [] + for change in changes: + row, new = await record(session, source, change, detected) + if new: + fresh.append(row) + return fresh + + +# ========================================================================== +# 读 +# ========================================================================== + + +def _iso(value: Any) -> str | None: + return value.isoformat() if value is not None else None + + +def summary(row: Any) -> str: + """一条变化给人看的一句（接口、日志）：「orders：删除了列 amount（可能是改名为 total，未确认）」。""" + kind = row.kind + label = KIND_LABEL.get(kind, kind) + table = row.table_name or "" + column = row.column_name + where = f"{table}.{column}" if table and column else table + text = f"{where}：{label}" if where else label + after = row.after if isinstance(row.after, Mapping) else {} + before = row.before if isinstance(row.before, Mapping) else {} + rename = after.get("rename") or before.get("rename") + if isinstance(rename, Mapping): + text += f"（可能是 {rename.get('from')} 改名为 {rename.get('to')}，未确认）" + elif kind == sd.COLUMN_TYPE: + text += f"（{row.before} → {row.after}）" + elif kind == CODES_NEW and isinstance(after.get("codes"), list): + text += f"（{'、'.join(str(c) for c in after['codes'][:8])}）" + return text + + +def change_out(row: SourceChange) -> dict[str, Any]: + """接口给的一条变化（agentlab-change/1）。""" + return { + "schema": row.schema_tag or SCHEMA, "seq": row.seq, "id": row.id, "source_id": row.source_id, + "source_name": row.source_name, "table": row.table_name or None, "column": row.column_name, "kind": row.kind, + "severity": row.severity, "summary": summary(row), "detected_at": _iso(row.detected_at), + "last_seen_at": _iso(row.last_seen_at), "seen_count": row.seen_count or 1, + "detected_by": dict(row.detected_by or {}), "basis": row.basis, "before": row.before, "after": row.after, + "affects": row.affects or {"catalog": [], "templates": [], "rules_touched": False}, + "struct_version_before": row.struct_version_before, "struct_version_after": row.struct_version_after, + "status": row.status, "resolution": row.resolution, + } + + +async def load(session: AsyncSession, source_id: str, change_id: str) -> SourceChange | None: + row = (await session.execute(select(SourceChange).where(SourceChange.id == change_id))).scalar_one_or_none() + return row if row is not None and row.source_id == source_id else None + + +async def counts(session: AsyncSession, source_id: str) -> dict[str, Any]: + """一个源还要人看的变化（开着的、知道了还没处理的）：{pending, by_severity: {action, warning, info}, open}。数据源卡片的 + 角标「N 个变化要处理」按 by_severity.action（第 4 波）。""" + rows = (await session.execute(select(SourceChange.status, SourceChange.severity, func.count()).where( + SourceChange.source_id == source_id, SourceChange.status.in_(PENDING)) + .group_by(SourceChange.status, SourceChange.severity))).all() + by = dict.fromkeys(SEVERITIES, 0) + opened = 0 + for status, severity, n in rows: + by[severity] = by.get(severity, 0) + n + if status == "open": + opened += n + return {"pending": sum(by.values()), "open": opened, "by_severity": by} + + +async def list_for_source(session: AsyncSession, source: Any, *, status: str | None = "pending", + kinds: Iterable[str] | None = None, table: str | None = None, limit: int = 100, + offset: int = 0) -> tuple[list[SourceChange], int]: + """一个源的变化（审阅页「变化」标签页）：按「使用次数权重 × 严重程度」排（catalog_review 的权重），同分新的在前。 + status：pending 开着的和知道了的（缺省）、all、或者某一个状态。返回 (这一页, 一共几条)。""" + from app.data import catalog_review + + filt = [SourceChange.source_id == source.id] + if status == "pending": + filt.append(SourceChange.status.in_(PENDING)) + elif status and status != "all": + filt.append(SourceChange.status == status) + if kinds: + filt.append(SourceChange.kind.in_(list(kinds))) + if table: + filt.append(SourceChange.table_name == table) + total = int((await session.execute(select(func.count()).select_from(SourceChange).where(*filt))).scalar() or 0) + rows = list((await session.execute(select(SourceChange).where(*filt))).scalars()) + usage = (await catalog_review.table_usage_counts(session, source)).queries if rows else {} + + def rank(r: SourceChange) -> tuple[float, int]: + weight = catalog_review.usage_weight(usage.get(r.table_name, 0)) * SEVERITY_RANK.get(r.severity, 1) + return -weight, -int(r.seq or 0) + + rows.sort(key=rank) + return rows[offset:offset + limit], total + + +async def list_changes(session: AsyncSession, *, since_seq: int = 0, source_id: str | None = None, + kinds: Iterable[str] | None = None, status: Iterable[str] | str | None = None, + limit: int = 200) -> list[dict[str, Any]]: + """按 seq 游标拉取（给召回）：seq 大于 since_seq 的，按 seq 从小到大，最多 limit 条。拉到的最后一条的 seq 就是下一次的 + since_seq。""" + filt = [SourceChange.seq > int(since_seq or 0)] + if source_id: + filt.append(SourceChange.source_id == source_id) + if kinds: + filt.append(SourceChange.kind.in_(list(kinds))) + if isinstance(status, str): + filt.append(SourceChange.status == status) + elif status: + filt.append(SourceChange.status.in_(list(status))) + rows = (await session.execute(select(SourceChange).where(*filt).order_by(SourceChange.seq) + .limit(max(1, min(int(limit), 1000))))).scalars() + return [change_out(r) for r in rows] + + +async def affecting(session: AsyncSession, source_name: str, tables: Iterable[str], since: int | datetime | None = None, + ) -> list[dict[str, Any]]: + """动到这个源这几张表的变化（给召回反查「这次答案用到的表之后有没有变」）：表名按 name_key 比；since 是 seq（拉取的 + 游标）或时刻（测到的时间晚于它的）。源级的变化（范围规则改了）不算。""" + keys = {name_key(str(t).rsplit(".", 1)[-1]) for t in tables if t} + if not keys: + return [] + filt = [SourceChange.source_name == source_name, SourceChange.table_name != ""] + if isinstance(since, datetime): + filt.append(or_(SourceChange.detected_at > since, SourceChange.last_seen_at > since)) + elif since: + filt.append(SourceChange.seq > int(since)) + rows = (await session.execute(select(SourceChange).where(*filt).order_by(SourceChange.seq))).scalars() + return [change_out(r) for r in rows if name_key(r.table_name) in keys] + + +def triggers_recall(change: Mapping[str, Any]) -> bool: + """这条变化触发不触发召回：结构类、处理时改了目录的（resolution 里有目录版本）、触及有效记录条件的新码值。纯取值漂移 + 只算数据质量提醒。""" + kind = change.get("kind") + if kind in sd.STRUCT_KINDS: + return True + if (change.get("resolution") or {}).get("catalog_version"): + return True + return kind == CODES_NEW and bool((change.get("affects") or {}).get("rules_touched")) + + +# ========================================================================== +# 处理 +# ========================================================================== + + +def resolve(row: SourceChange, *, action: str, by: str | None, note: str | None = None, + catalog_version: int | None = None, job_id: str | None = None) -> SourceChange: + """把一条变化记成处理过（不提交）：状态按动作（ACTION_STATUS），resolution 写谁、什么时候、什么动作、理由、目录版本、 + 任务编号。已经 resolved / dismissed 的抛 ChangeFinished；误报没写理由抛 ChangeInvalid。""" + if action not in ACTIONS: + raise ChangeInvalid(f"不认识的处理动作：{action}。可选：{'、'.join(ACTIONS)}") + if row.status in ("resolved", "dismissed"): + raise ChangeFinished("这条变化已经处理过了") + text = (note or "").strip() + if action == "dismiss" and not text: + raise ChangeInvalid("标为误报要写明理由") + if len(text) > NOTE_MAX: + raise ChangeInvalid(f"理由最多 {NOTE_MAX} 个字") + resolution: dict[str, Any] = {"by": by, "at": utcnow().isoformat(), "action": action, "note": text or None} + if catalog_version is not None: + resolution["catalog_version"] = catalog_version + if job_id is not None: + resolution["job_id"] = job_id + row.status = ACTION_STATUS[action] + row.resolution = resolution + return row + + +def _catalog_paths(row: SourceChange) -> list[dict[str, Any]]: + """影响面里这张表上的目录项（清理目录时删的就是它们）。""" + items = ((row.affects or {}).get("catalog") or []) + return [i for i in items if isinstance(i, Mapping) and i.get("table") == row.table_name and i.get("path")] + + +def without_paths(notes: Mapping[str, Any], paths: Iterable[str]) -> dict[str, Any]: + """目录去掉这几项（路径同审阅接口：字段名、columns.<列>.<字段>、relations.<编号>）。""" + from app.data import catalog + + slots = catalog._slots(notes) + for path in paths: + try: + slots.pop(catalog.parse_path(path), None) + except catalog.CatalogPathError: + continue + return catalog._assemble(slots) + + +async def handle(session: AsyncSession, source: Any, row: SourceChange, action: str, *, actor: str | None, + note: str | None = None, codes: Mapping[str, str] | None = None) -> dict[str, Any]: + """执行一个处理动作并记下 resolution（提交）。返回 {change, catalog_version?, job?}。 + + - clean_catalog：把影响面里这张表上说到这些列的项从目录里去掉（整份写入，乐观锁），记目录版本。影响面里没有目录项的 + 抛 ChangeInvalid。 + - add_codes：码值的新取值加进目录（catalog.save_patch，含义待填写；依据写「数据源变化 #编号」），只给 codes_new。codes + 给了就按它加（{码: 含义}，含义可以空），不给就加这条变化测到的新码值。 + - redraft：建一个只含这张表的起草任务；reprofile：建一个只含这张表的剖析任务。建不了（模型没配、剖析没开、同源正忙） + 抛 ChangeInvalid，原因照实。 + - acknowledge、dismiss（必须写理由）只改状态。 + """ + from app.data import catalog + + if action not in ACTIONS: + raise ChangeInvalid(f"不认识的处理动作：{action}。可选：{'、'.join(ACTIONS)}") + if row.status in ("resolved", "dismissed"): + raise ChangeFinished("这条变化已经处理过了") + if action == "dismiss" and not (note or "").strip(): + raise ChangeInvalid("标为误报要写明理由") + out: dict[str, Any] = {} + version: int | None = None + job_id: str | None = None + table = row.table_name + if action == "clean_catalog": + paths = [str(i["path"]) for i in _catalog_paths(row)] + if not paths: + raise ChangeInvalid("这条变化的影响面里没有目录项，不需要清理") + entry = await catalog.read_entry(session, source.id, table) + if entry is None: + raise ChangeInvalid(f"{table} 已经没有数据目录了") + notes = without_paths(entry.notes, paths) + written = await catalog.write_entry(session, source.id, table, notes, if_version=entry.version, actor=actor, + commit=False) + version = written.version + elif action == "add_codes": + if row.kind != CODES_NEW or not row.column_name: + raise ChangeInvalid("只有「出现了新码值」能加入码值") + wanted = dict(codes) if codes else {str(c): "" for c in ((row.after or {}).get("codes") or [])} + if not wanted or not all(isinstance(k, str) and isinstance(v, str) for k, v in wanted.items()): + raise ChangeInvalid("要加入的码值应为 {码值: 含义} 的对照，含义可以先空着") + entry = await catalog.read_entry(session, source.id, table) + current = entry.version if entry else 0 + meta_tables = (getattr(source, "schema_cache", None) or {}).get("tables") or None + changes = [{"path": f"columns.{row.column_name}.codes", "value": wanted, + "reason": f"数据源变化 #{row.id[:6]}：取值观察测到的新码值，含义待填写"}] + try: + notes = catalog.apply_patch(entry.notes if entry else {}, changes, table=table, tables=meta_tables) + except catalog.CatalogInvalid as e: + raise ChangeInvalid("；".join(e.problems) if hasattr(e, "problems") else str(e)) from e + written = await catalog.write_entry(session, source.id, table, notes, if_version=current, actor=actor, + commit=False) + version = written.version + elif action == "reprofile": + job_id = await _reprofile(session, source, table, actor) + out["job"] = {"kind": "profile", "id": job_id} + elif action == "redraft": + job_id = await _redraft(session, source, table, actor, row.id) + out["job"] = {"kind": "draft", "id": job_id} + resolve(row, action=action, by=actor, note=note, catalog_version=version, job_id=job_id) + await session.commit() + if version is not None: + out["catalog_version"] = version + out["change"] = change_out(row) + return out + + +async def _reprofile(session: AsyncSession, source: Any, table: str, actor: str | None) -> str: + from app.data import catalog_profile, catalog_profile_jobs, table_versions + + try: + settings = catalog_profile.ensure_enabled(source) + except catalog_profile.ProfileDisabled as e: + raise ChangeInvalid(str(e)) from e + try: + view = await table_versions.resolve_source(session, source) + except table_versions.SnapshotMissing as e: + raise ChangeInvalid(str(e)) from e + try: + job, _tasks = await catalog_profile_jobs.create_job(session, source, view, tables=[table], scope="selected", + settings=settings, actor=actor) + except (catalog_profile_jobs.ProfileBusy, catalog_profile_jobs.ProfileNothing) as e: + raise ChangeInvalid(str(e)) from e + catalog_profile_jobs.manager.start(job.id) + return job.id + + +async def _redraft(session: AsyncSession, source: Any, table: str, actor: str | None, change_id: str) -> str: + """建一个只含这张表的起草任务。起草模型用不了时只按表结构起草（数据库注释、外键、命名推断），不挡这个动作。""" + from app.data import catalog, catalog_jobs, table_versions + + try: + view = await table_versions.resolve_source(session, source) + except table_versions.SnapshotMissing as e: + raise ChangeInvalid(str(e)) from e + if not ((getattr(view, "schema_cache", None) or {}).get("tables") or {}): + raise ChangeInvalid("这个数据源还没有表结构，不能起草") + use_model = True + try: + await catalog_jobs.check_model(session) + except catalog.CatalogModelUnavailable: + use_model = False + try: + job, _tasks = await catalog_jobs.create_job(session, source, view, tables=[table], use_model=use_model, + note=f"数据源变化 #{change_id[:6]}", actor=actor) + except catalog_jobs.DraftBusy as e: + raise ChangeInvalid(str(e)) from e + catalog_jobs.manager.start(job.id) + return job.id + + +__all__ = [ + "ACTIONS", "ACTION_STATUS", "CODES_NEW", "CODES_SHIFT", "COVERAGE_DROP", "ChangeFinished", "ChangeInvalid", + "ChangeNotFound", "DATE_GAP", "DATE_STALLED", "Detected", "KINDS", "KIND_LABEL", "NewChange", "PENDING", "ROWS_DROP", + "ROWS_JUMP", "SCHEMA", "SEVERITIES", "SEVERITY_RANK", "STATUSES", "VALUE_KINDS", "VOLUME_DROP", "affecting", + "change_out", "counts", "fingerprint", "handle", "list_changes", "list_for_source", "load", "record", "record_all", + "resolve", "summary", "triggers_recall", "without_paths", +] diff --git a/backend/app/data/source_objects.py b/backend/app/data/source_objects.py index adff0c3a..058fc989 100644 --- a/backend/app/data/source_objects.py +++ b/backend/app/data/source_objects.py @@ -446,6 +446,11 @@ class RefreshStats: note: str | None = None #: 第一次建对象索引时从升级前的老缓存接进来几张（legacy_seed） adopted: int = 0 + #: 这次新出现的、不在了（真删除）的对象名（结构变化按它记 table_added / table_removed，next-round 03 B1）； + #: first_build：这个 schema 的对象索引是这一次才建起来的——全是新的，只建基线、不记「新出现的表」 + added_names: list[str] = field(default_factory=list) + removed_names: list[str] = field(default_factory=list) + first_build: bool = False def _chunks(items: list[Any], size: int = 500) -> Iterable[list[Any]]: @@ -548,14 +553,18 @@ async def refresh_index(session: AsyncSession, source_id: str, schema: str, resu await session.execute(update(_SO).where(_SO.id.in_(chunk)).values(removed_at=None) .execution_options(synchronize_session=False)) gone: list[str] = [] + gone_names: list[str] = [] if not result.capped: - gone = [r.id for name, r in existing.items() if name not in seen and r.removed_at is None] + gone_names = [name for name, r in existing.items() if name not in seen and r.removed_at is None] + gone = [existing[name].id for name in gone_names] for chunk in _chunks(gone): - # 真删除：不在范围里了，结构（meta）留着——下一波的结构差异要拿它和以前比 + # 真删除：不在范围里了，结构（meta）留着——结构差异（data/schema_diff）要拿它和以前比、算影响面 await session.execute(update(_SO).where(_SO.id.in_(chunk)).values(removed_at=now, in_scope=False) .execution_options(synchronize_session=False)) return RefreshStats(total=len(seen), added=len(new_rows), removed=len(gone), method=result.method, - capped=result.capped, note=result.note, adopted=len(seeds)) + capped=result.capped, note=result.note, adopted=len(seeds), + added_names=[r["name"] for r in new_rows] if existing else [], removed_names=gone_names, + first_build=not existing) async def live_objects(session: AsyncSession, source_id: str, schema: str) -> list[tuple[str, str]]: diff --git a/backend/app/data/sqlcheck_columns.py b/backend/app/data/sqlcheck_columns.py new file mode 100644 index 00000000..ac40fa66 --- /dev/null +++ b/backend/app/data/sqlcheck_columns.py @@ -0,0 +1,142 @@ +"""SQL 里的列名对照数据源结构（sql_unknown_columns）：助手自查（api/copilot.py）和表结构变化的影响面 +（data/catalog_impact.py：已发布模板写死的 SQL 换上新结构再跑一遍，说得出「这条 SQL 现在会报列不存在」）共用这一份。 + +以前它在 api/copilot.py 里，影响面要用就得从 data 包反过来导入接口层；挪到这里，copilot 那边只改 import。 + +只做有把握的判断，宁可漏判不可误判：助手自查会把图打回去让模型改，影响面会把一条变化升为「要处理」。CTE、FROM 里的 +子查询、引号括起来的标识符、结构里找不到的表、结构没探查过或者探查被截断的，一律不判。 +""" +from __future__ import annotations + +import re +from typing import Any + +#: @ 开头的是变量（@region、@@session），: 开头的是参数，. 后面的是限定名 +_SQL_IDENT = re.compile(r"(? str: + """带引号、名字里有非 ASCII 字符的标识符（中文别名 AS "月份"、AS `曝光量`）连引号一起换成空格。 + + 这项核对只认 ASCII 的列名，中文名本来就不判；不抹掉的话，SQL 里一出现带引号的标识符就整条放弃核对，照「结果列 + 起中文别名」写的 SQL 会把猜错的英文列名一起放过去。英文的带引号标识符（"StatDate"）大小写、空格都可能，照旧不判。""" + from app.data.guard import _GENERIC, _tokens + + out = [] + for token in _tokens(sql, _GENERIC): + if token.kind == "ident" and not token.text.isascii(): + out.append(" " * len(token.text)) + else: + out.append(token.text) + return "".join(out) + + +def sql_unknown_columns(sql: str, source: Any) -> tuple[list[str], list[str]] | None: + """SQL 里用到、数据源结构里查不到的列：(查不到的列, 这几张表的全部列)。判断不了、或者都查得到时返回 None。""" + from app.data import introspect as _introspect + from app.data.guard import _blank_quoted, first_verb, strip_comments + + cache = getattr(source, "schema_cache", None) or {} + if not cache.get("tables") or cache.get("failed") or first_verb(sql) != "select": + return None + text = _blank_unicode_identifiers(re.sub(r"\{\{.*?\}\}", " 0 ", strip_comments(sql), flags=re.S)) + if '"' in text.replace("''", "") and _blank_quoted(text) != _blank_quoted(text, identifiers=False): + return None # 引号括起来的标识符：大小写、空格都可能，不猜 + text = _blank_quoted(text) + if re.search(r"\bwith\b", text, re.I) or re.search(r"\b(?:from|join)\s*\(", text, re.I) or "`" in text \ + or "[" in text: + return None # [方括号] 标识符、数组下标：同样不猜 + tables: dict[str, dict[str, Any]] = {} # 表名、别名（小写）→ 表结构 + spans: list[tuple[int, int]] = [] + for m in _FROM_ITEMS.finditer(text): + spans.append(m.span(1)) + for item in m.group(1).split(","): + parsed = _TABLE_ITEM.match(item) + if not parsed: + return None + name, alias = parsed.group(1), parsed.group(2) + meta = _introspect.find_table(source, name) + if meta is None: + return None # 结构里找不到的表（视图、同义词、截断了的结构）不判 + tables[name.lower()] = tables[name.rsplit(".", 1)[-1].lower()] = meta + if alias and alias.lower() not in _SQL_WORDS: + tables[alias.lower()] = meta + if not tables: + return None + known = list(dict.fromkeys(c["name"] for meta in {id(m): m for m in tables.values()}.values() + for c in meta.get("columns") or [])) + columns = {c.lower() for c in known} + # 「SELECT created_at FROM」里 created_at 前面是关键字，它是列，不是别名 + aliases = {a.lower() for a in _ALIAS_AS.findall(text)} | { + m.group(2).lower() for m in _ALIAS_BARE.finditer(text) if m.group(1).lower() not in _OPERAND_BEFORE} | { + a.lower() for a in _WINDOW_NAME.findall(text)} + names = {m.start(1) for pattern in (_NAME_AFTER, _DATEPART_ARG) for m in pattern.finditer(text)} + unknown: list[str] = [] + for m in _QUALIFIED.finditer(text): + if any(a <= m.start() < b for a, b in spans): + continue + meta = tables.get(m.group(1).lower()) + col = m.group(2) + if meta is not None and col != "*" and col.lower() not in {c["name"].lower() for c in meta.get("columns") or []}: + unknown.append(f"{m.group(1)}.{col}") + for m in _SQL_IDENT.finditer(text): + word = m.group(1) + low = word.lower() + if any(a <= m.start() < b for a, b in spans) or low in _SQL_WORDS or low in columns or low in aliases \ + or low in tables or m.start() in names or re.match(r"\s*[.(]", text[m.end():]) \ + or text[m.end():m.end() + 1] == "'": # N'…'、X'…'、_utf8mb4'…'：字面量的前缀 + continue + unknown.append(word) + unknown = list(dict.fromkeys(unknown)) + return (unknown, known) if unknown else None + + +__all__ = ["sql_unknown_columns"] diff --git a/backend/app/db/models_sources.py b/backend/app/db/models_sources.py index b684ffed..5495855e 100644 --- a/backend/app/db/models_sources.py +++ b/backend/app/db/models_sources.py @@ -19,7 +19,7 @@ from __future__ import annotations from datetime import datetime from typing import Any -from sqlalchemy import Boolean, ForeignKey, Index, Integer, String, Text, UniqueConstraint, text +from sqlalchemy import JSON, Boolean, ForeignKey, Index, Integer, String, Text, UniqueConstraint, text from sqlalchemy.orm import Mapped, mapped_column from app.db.base import Base, UTCDateTime, new_id, utcnow @@ -134,4 +134,101 @@ class IntrospectTask(Base): finished_at: Mapped[datetime | None] = mapped_column(UTCDateTime, default=None) -__all__ = ["IntrospectJob", "IntrospectTask", "SourceObject"] +# -------------------------------------------------------------------------- +# 第 3 波：取值观察、变化记录（docs/design/next-round/03-数据接入与目录维护.md B2、B3） +# -------------------------------------------------------------------------- + + +class TableObservation(Base): + """一次取值观察（剖析引擎的四类观察检查，data/catalog_profile_observe.py）：码值列的取值分布（codes_observe）、最新业务 + 日期（freshness）、最近 35 天按天的行数（daily_volume）、行数（rows）。 + + **为什么单独一张表。** 剖析任务的逐表一行只留最近 50 个任务，而且一张表一次剖析只有一份发现；取值漂移要拿同一口径的 + 历史比（新码值、比例剧变、断档、缺天、量跌、行数跌），得按「表 × 种类 × 列」留一串。每张表每种观察（码值按列）留最近 + KEEP 条，更早的删掉。 + + **口径（basis）。** 每条都记它看的是哪一部分数据：full 整表、first_rows（按存储顺序的前 N 行）、window（业务日期最近 N 天）、 + sample（块采样 p%）、stats / count（行数的来源）。basis_key 是口径的规范写法：只和 basis_key 相同的历史比，口径不同的照实说 + 不比。遮罩列不出现在任何一条观察查询里，取值也就不会进这张表。 + + 挂外键、级联删除：观察是数据源的附属记录。""" + + __tablename__ = "table_observations" + __table_args__ = ( + Index("ix_table_observations_lookup", "source_id", "table_name", "kind", "column_name", "at"), + ) + + #: 每张表每种观察（码值按列）留几条 + KEEP = 26 + + id: Mapped[str] = mapped_column(String(32), primary_key=True, default=new_id) + source_id: Mapped[str] = mapped_column(ForeignKey("data_sources.id", ondelete="CASCADE")) + table_name: Mapped[str] = mapped_column(String(255)) + #: 码值、业务日期观察的那一列；表级的（行数）是空串 + column_name: Mapped[str] = mapped_column(String(255), default="") + #: codes_observe | freshness | daily_volume | rows + kind: Mapped[str] = mapped_column(String(20)) + basis: Mapped[dict[str, Any]] = mapped_column(default=dict) + basis_key: Mapped[str] = mapped_column(String(200), default="") + #: 观察到的：码值 {values: [[码, 行数]…], rows, truncated}；最新业务日期 {max}；按日 {since, until, days: {日: 行数}}； + #: 行数 {rows, at_least, method} + data: Mapped[dict[str, Any]] = mapped_column(default=dict) + #: 哪个剖析任务的第几张表测到的 + job_id: Mapped[str | None] = mapped_column(String(32), default=None) + task_seq: Mapped[int | None] = mapped_column(Integer, default=None) + at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow) + + +class SourceChange(Base): + """数据源的一条变化（data/source_changes.py，格式 agentlab-change/1）：结构变化（重新探查后的前后差异）和取值变化（取值 + 观察和同口径历史的比较）。全站只有这一份记录：审阅页的「变化」、数据源卡片的角标、外部通知、口径账本的召回都读它。 + + - seq：自增，不复用（sqlite_autoincrement），召回按它当游标拉取；id 是给界面、链接用的编号； + - kind：结构类 table_added / table_removed / table_left_scope / table_reflect_failed / column_added / column_removed / + column_type_changed / column_nullable_changed / key_changed / fk_changed / comment_changed / scope_changed；取值类 + codes_new / codes_shift / date_stalled / date_gap / volume_drop / rows_drop / rows_jump / coverage_drop； + - severity：info / warning / action（现在大概率已经不对了：已确认的项、写死的 SQL 指向被删的列，或者「已列全」被撤销）； + - detected_by：{job_kind, job_id, task_seq, trigger: manual|schedule|on_demand, actor}；basis：{method, query_refs}； + - before / after：前后各是什么（码值只记非遮罩列的）；affects：{catalog: [{table, path, status, source}], templates: + [{workflow_id, name, version, impact, nodes}], rules_touched}；struct_version_before / after：结构变化前后的结构版本； + - fingerprint：去重——同一个变化再次测到只更新 last_seen_at 和 seen_count； + - status：open / acknowledged / resolved / dismissed；resolution：{by, at, action, note, catalog_version, job_id}。 + + 挂外键、级联删除：变化是数据源的附属记录。""" + + __tablename__ = "source_changes" + __table_args__ = ( + UniqueConstraint("source_id", "fingerprint"), + Index("ix_source_changes_source_status", "source_id", "status"), + Index("ix_source_changes_source_table", "source_id", "table_name"), + {"sqlite_autoincrement": True}, + ) + + seq: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + id: Mapped[str] = mapped_column(String(32), unique=True, default=new_id) + source_id: Mapped[str] = mapped_column(ForeignKey("data_sources.id", ondelete="CASCADE")) + #: 测到时这个源叫什么（之后改了名，这条记录说的仍是当时的名字） + source_name: Mapped[str] = mapped_column(String(100), default="") + #: 源级的变化（范围规则改了）是空串 + table_name: Mapped[str] = mapped_column(String(255), default="") + column_name: Mapped[str | None] = mapped_column(String(255), default=None) + kind: Mapped[str] = mapped_column(String(40)) + severity: Mapped[str] = mapped_column(String(10), default="info") + detected_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow) + last_seen_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow) + seen_count: Mapped[int] = mapped_column(Integer, default=1) + detected_by: Mapped[dict[str, Any]] = mapped_column(default=dict) + basis: Mapped[dict[str, Any] | None] = mapped_column(default=None) + before: Mapped[Any] = mapped_column(JSON, nullable=True, default=None) + after: Mapped[Any] = mapped_column(JSON, nullable=True, default=None) + affects: Mapped[dict[str, Any] | None] = mapped_column(default=None) + struct_version_before: Mapped[int | None] = mapped_column(Integer, default=None) + struct_version_after: Mapped[int | None] = mapped_column(Integer, default=None) + fingerprint: Mapped[str] = mapped_column(String(64)) + status: Mapped[str] = mapped_column(String(20), default="open") + resolution: Mapped[dict[str, Any] | None] = mapped_column(default=None) + #: 记录的格式（agentlab-change/1）：召回按它读，字段以后改要升版本 + schema_tag: Mapped[str] = mapped_column("schema", String(30), default="agentlab-change/1") + + +__all__ = ["IntrospectJob", "IntrospectTask", "SourceChange", "SourceObject", "TableObservation"] diff --git a/backend/app/engine/launch.py b/backend/app/engine/launch.py index 4e71dc26..9bdbf3fa 100644 --- a/backend/app/engine/launch.py +++ b/backend/app/engine/launch.py @@ -19,7 +19,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.core import artifact_store from app.core.events import EventType -from app.data.catalog_impact import catalog_drift +from app.data.catalog_impact import catalog_drift, schema_drift from app.apps.versions import is_published, subgraph_pin_problems from app.db.models import Run, Workflow, WorkflowVersion from app.engine.governance import governance_note, unresolved_caliber_upgrades @@ -165,9 +165,13 @@ async def launch_run( # 数据目录和口径卡不同：目录是给人和助手看的说明，不是运行的输入。发布之后变了只提醒、不拦—— # 拦下来等于让一次文字修订卡住正式出具 catalog_changes = await catalog_drift(session, snapshot.catalog_versions) + # 表结构同理（next-round 03 B1）：发布之后结构版本变了只提醒、不拦——只是加了一列不该卡住正式出具，真删了列查询 + # 本身会照实报错 + schema_changes = await schema_drift(session, snapshot.schema_versions) else: upgrade_events = [] catalog_changes = [] + schema_changes = [] if workflow_id: workflow = await session.get(Workflow, workflow_id) if not workflow: @@ -197,6 +201,10 @@ async def launch_run( # SQL 用到的，possible Agent 可能查询的（界面分开说） notes.append((EventType.CATALOG_DRIFT, None, {"version": run_version, "count": len(catalog_changes), "tables": catalog_changes})) + if schema_changes: + # 回看这次正式出具，能知道当时的表结构和发布时不一样：哪几张表、发布时和现在的结构版本、还没处理的变化 + notes.append((EventType.SCHEMA_DRIFT, None, + {"version": run_version, "count": len(schema_changes), "tables": schema_changes})) if capability_changes: # 能力的治理（读写效果、审批、遮罩……）发布之后改过：同目录漂移，只提醒、不拦，之后回看这次运行知道差在哪 notes.append(("log", None, { diff --git a/backend/tests/test_schema_changes.py b/backend/tests/test_schema_changes.py new file mode 100644 index 00000000..80e8779d --- /dev/null +++ b/backend/tests/test_schema_changes.py @@ -0,0 +1,263 @@ +"""表结构变化：前后差异进变化记录、影响面、结构版本与 schema.drift、发布前检查（next-round 03 B1）。 + +用一个 SQLite 夹具（虚构的店铺库：stores、orders、refunds）走一遍真实路径：后台探查建基线 → 写目录（人工确认的码值、 +有效记录条件、关联关系）→ 发布一个写死 SQL 的模板（记下结构版本）→ ALTER（删列、加列、改名、改类型、加注释）→ 重新探查： + +- 变化恰好就是这几条，疑似改名只说「可能」；第一次探查只建基线、不报变化； +- 删掉的列上人工确认过的项、写死这一列的已发布模板都列进影响面，这条变化是「要处理」（action），有效记录条件用到它记 + rules_touched； +- 再探查一次（结构没变）不重复记；表真删了记 table_removed，范围外 ≠ 删除； +- 从这一版发起的正式运行写一条 schema.drift、照常发起；发布前检查列出这张表上还没处理的结构变化（警告，不拦）。 +""" +from __future__ import annotations + +import asyncio +import sqlite3 +import uuid +from pathlib import Path + +import pytest +from httpx import ASGITransport, AsyncClient +from sqlalchemy import select + +from app.data import catalog +from app.db.base import SessionLocal +from app.db.models import Run, RunEvent, Workflow, WorkflowVersion +from app.db.models_sources import SourceChange, SourceObject +from app.engine.runner import run_manager +from app.main import app +from tests.fixtures.sources import drop_source, introspect_now + +SQL = "SELECT o.status, o.amount FROM orders o WHERE o.status = 'paid'" + + +def _build(path: Path) -> None: + db = sqlite3.connect(path) + db.executescript(""" + CREATE TABLE stores (id INTEGER PRIMARY KEY, name VARCHAR(40)); + CREATE TABLE orders (id INTEGER PRIMARY KEY, store_id INTEGER REFERENCES stores(id), status VARCHAR(10), + amount NUMERIC(12, 2), note VARCHAR(40), paid_on DATE); + CREATE TABLE refunds (id INTEGER PRIMARY KEY, order_id INTEGER, amount NUMERIC(12, 2)); + INSERT INTO stores VALUES (1, 'east'), (2, 'west'); + INSERT INTO orders VALUES (1, 1, 'paid', 10.5, 'a', '2026-10-01'), (2, 2, 'void', 3, 'b', '2026-10-02'); + """) + db.commit() + db.close() + + +def _alter(path: Path) -> None: + """删 status、加 channel、note 改名为 remark（同位置同类型）、amount 改类型（SQLite 没有改类型：整表重建）。""" + db = sqlite3.connect(path) + db.executescript(""" + CREATE TABLE orders_new (id INTEGER PRIMARY KEY, store_id INTEGER REFERENCES stores(id), + amount NUMERIC(14, 2), remark VARCHAR(40), paid_on DATE, channel VARCHAR(8)); + INSERT INTO orders_new SELECT id, store_id, amount, note, paid_on, 'web' FROM orders; + DROP TABLE orders; + ALTER TABLE orders_new RENAME TO orders; + """) + db.commit() + db.close() + + +@pytest.fixture(autouse=True) +async def engine_up(): + await run_manager.setup() + yield + await run_manager.shutdown() + + +@pytest.fixture +async def client(): + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + yield c + + +@pytest.fixture +async def shop(client, tmp_path): + path = tmp_path / "shop.db" + _build(path) + name = f"shop_{uuid.uuid4().hex[:8]}" + r = await client.post("/api/datasources", json={"name": name, "kind": "sqlite", "database": str(path), + "readonly": True}) + assert r.status_code == 201, r.text + sid = r.json()["id"] + made: list[str] = [] + yield {"id": sid, "name": name, "path": path, "made": made} + async with SessionLocal() as session: + for wid in made: + if (wf := await session.get(Workflow, wid)) is not None: + await session.delete(wf) + await session.commit() + await drop_source(sid) + + +async def _changes(sid: str) -> list[SourceChange]: + async with SessionLocal() as session: + return list((await session.execute(select(SourceChange).where(SourceChange.source_id == sid) + .order_by(SourceChange.seq))).scalars()) + + +async def _write_notes(sid: str) -> None: + async with SessionLocal() as session: + await catalog.write_entry(session, sid, "orders", { + "label": catalog.make_item("订单", "human"), + "valid_filter": catalog.make_item("status <> 'void'", "human"), + "columns": {"status": {"codes": catalog.make_item({"paid": "已支付", "void": "作废"}, "human", complete=True)}, + "note": {"label": catalog.make_item("备注", "llm")}}, + }, if_version=0, actor="测试员") + await catalog.write_entry(session, sid, "refunds", { + "relations": [{"id": catalog.relation_id("refunds", ["order_id"], "orders", ["id"]), + "columns": ["order_id"], "to_table": "orders", "to_columns": ["id"], + "cardinality": "many_to_one", "source": "human", "status": "confirmed"}], + }, if_version=0, actor="测试员") + + +def _graph(source: str) -> dict: + def node(nid, ntype, **config): + return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}} + + nodes = [node("start", "input"), node("fetch", "tool", tool=f"db_query__{source}", args={"sql": SQL}), + node("out", "output", fields=[{"name": "结果", "value": "{{ nodes.fetch }}"}])] + return {"nodes": nodes, "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]} + + +async def _publish(client, shop) -> str: + r = await client.post("/api/workflows", json={"name": f"订单日报-{uuid.uuid4().hex[:6]}", + "graph": _graph(shop["name"])}) + wf = r.json()["id"] + shop["made"].append(wf) + body = (await client.post(f"/api/workflows/{wf}/publish", json={"level": "published"})).json() + assert body["ok"] is True, body + return wf + + +async def _settle(run_id: str) -> Run: + for _ in range(400): + await asyncio.sleep(0.05) + async with SessionLocal() as session: + row = await session.get(Run, run_id) + if row.status in ("succeeded", "failed") and row.manifest_seq is not None: + return row + raise AssertionError(f"run {run_id} 没跑完：{row.status}") + + +async def test_first_introspect_is_baseline_and_alter_reports_exactly_the_changes(client, shop): + sid = shop["id"] + await introspect_now(client, sid) + assert await _changes(sid) == [], "第一次探查只建基线：不报「新出现的表」，也不报列" + async with SessionLocal() as session: + orders = (await session.execute(select(SourceObject).where( + SourceObject.source_id == sid, SourceObject.name == "orders"))).scalar_one() + assert orders.struct_version == 1 + await _write_notes(sid) + wf = await _publish(client, shop) + async with SessionLocal() as session: + snap = (await session.execute(select(WorkflowVersion).where( + WorkflowVersion.workflow_id == wf, WorkflowVersion.version == 1))).scalar_one() + assert snap.schema_versions == {shop["name"]: {"orders": 1}}, "只记写死的 SQL 直接引用的表" + detail = (await client.get(f"/api/workflows/{wf}/versions/1")).json() + assert detail["schema_versions"] == {shop["name"]: {"orders": 1}} and detail["schema_changes"] == [] + + _alter(shop["path"]) + await introspect_now(client, sid) + got = [(c.kind, c.column_name) for c in await _changes(sid) if c.table_name == "orders"] + assert sorted(got) == sorted([("column_removed", "status"), ("column_removed", "note"), ("column_added", "remark"), + ("column_added", "channel"), ("column_type_changed", "amount")]), got + by = {(c.kind, c.column_name): c for c in await _changes(sid)} + removed = by[("column_removed", "status")] + assert removed.struct_version_before == 1 and removed.struct_version_after == 2 + assert removed.severity == "action", "人工确认过的码值、有效记录条件、写死它的模板都还说着这一列" + paths = {(i["table"], i["path"]) for i in removed.affects["catalog"]} + assert ("orders", "columns.status.codes") in paths and ("orders", "valid_filter") in paths + assert removed.affects["rules_touched"] is True, "有效记录条件用到了被删的列" + assert all(i["needs_action"] for i in removed.affects["catalog"] if i["status"] == "confirmed") + [tpl] = removed.affects["templates"] + assert tpl["workflow_id"] == wf and tpl["impact"] == "direct" + [fetch] = [n for n in tpl["nodes"] if n["node_id"] == "fetch"] + assert fetch["breaks"] is True and any(c.endswith("status") for c in fetch["unknown_columns"]), fetch + # 疑似改名只说「可能」：note → remark 同位置同类型 + note = by[("column_removed", "note")] + assert note.before["rename"] == {"from": "note", "to": "remark"} and note.kind == "column_removed" + assert note.severity != "action", "只有推断的中文名说着这一列：不是要处理" + assert by[("column_type_changed", "amount")].before == "NUMERIC(12, 2)" + assert by[("column_type_changed", "amount")].after == "NUMERIC(14, 2)" + assert all(c.detected_by["job_kind"] == "introspect" and c.detected_by["trigger"] == "manual" + for c in await _changes(sid)) + + # 结构没再变：再探查一次不重复记 + count = len(await _changes(sid)) + await introspect_now(client, sid) + assert len(await _changes(sid)) == count + + # 正式运行：写一条 schema.drift，照常发起 + r = await client.post("/api/runs", json={"workflow_id": wf, "run_class": "formal"}) + assert r.status_code == 201, r.text + run_id = r.json()["id"] + await _settle(run_id) + async with SessionLocal() as session: + drift = [e.data for e in (await session.execute(select(RunEvent).where( + RunEvent.run_id == run_id, RunEvent.type == "schema.drift"))).scalars()] + assert len(drift) == 1 and drift[0]["count"] == 1 + [table] = drift[0]["tables"] + assert table["table"] == "orders" and table["published"] == 1 and table["current"] == 2 and not table["removed"] + assert {c["kind"] for c in table["changes"]} >= {"column_removed", "column_type_changed"} + detail = (await client.get(f"/api/workflows/{wf}/versions/1")).json() + assert [(c["table"], c["published"], c["current"]) for c in detail["schema_changes"]] == [("orders", 1, 2)] + + # 发布前检查：这一版直接引用的 orders 上有还没处理的结构变化（警告，不拦） + check = (await client.post(f"/api/workflows/{wf}/publish-check", json={"level": "published"})).json() + [issue] = [i for i in check["issues"] if i.get("code") == "schema.change_open"] + assert issue["level"] == "warning" and issue["node_id"] == "fetch" and "orders" in issue["message"] + + +async def test_table_removed_is_a_deletion_with_impact(client, shop): + sid = shop["id"] + await introspect_now(client, sid) + await _write_notes(sid) + db = sqlite3.connect(shop["path"]) + db.execute("DROP TABLE refunds") + db.commit() + db.close() + await introspect_now(client, sid) + [gone] = [c for c in await _changes(sid) if c.kind == "table_removed"] + assert gone.table_name == "refunds" and gone.struct_version_before == 1 + assert gone.severity == "action", "人工确认过的关联关系还在这张表上" + assert ("refunds", f"relations.{catalog.relation_id('refunds', ['order_id'], 'orders', ['id'])}") in { + (i["table"], i["path"]) for i in gone.affects["catalog"]} + + +async def test_leaving_scope_is_not_a_deletion(client, shop): + sid = shop["id"] + await introspect_now(client, sid) + await _write_notes(sid) + r = await client.put(f"/api/datasources/{sid}/introspect-scope", + json={"scope": {"mode": "rules", "include": ["stores", "orders"], "removed": ["refunds"]}}) + assert r.status_code == 200, r.text + await introspect_now(client, sid) + kinds = {(c.kind, c.table_name) for c in await _changes(sid)} + assert ("table_left_scope", "refunds") in kinds, "有目录的表离开了范围" + assert ("table_removed", "refunds") not in kinds, "范围外 ≠ 删除" + assert any(k == "scope_changed" for k, _ in kinds), "范围规则和上一次探查不同" + + +async def test_first_introspect_after_upgrade_only_builds_a_baseline(client, shop): + """升级前同步探查写的老缓存（类型是通用写法 VARCHAR、没有 scope 摘要）：第一次按对象索引读到只建基线，不把写法的差异 + 说成「类型变了」。""" + from app.db.models import DataSource + + sid = shop["id"] + legacy = {"tables": {"orders": {"qualified": "orders", "schema": None, "comment": None, "is_view": False, + "primary_key": ["id"], + "columns": [{"name": "id", "type": "INTEGER", "nullable": False}, + {"name": "status", "type": "VARCHAR", "nullable": True}]}}, + "truncated": False, "total": 1, "schema": None, "synced_at": "2026-09-01T00:00:00+00:00"} + async with SessionLocal() as session: + row = await session.get(DataSource, sid) + row.schema_cache = legacy + await session.commit() + await introspect_now(client, sid) + assert await _changes(sid) == [], "老缓存接进来的结构版本是 0：这一次只建基线" + async with SessionLocal() as session: + orders = (await session.execute(select(SourceObject).where( + SourceObject.source_id == sid, SourceObject.name == "orders"))).scalar_one() + assert orders.struct_version == 1 diff --git a/backend/tests/test_schema_diff.py b/backend/tests/test_schema_diff.py new file mode 100644 index 00000000..6f23800f --- /dev/null +++ b/backend/tests/test_schema_diff.py @@ -0,0 +1,124 @@ +"""表结构的前后差异（app/data/schema_diff.py，next-round 03 B1）：纯函数。 + +守的是：加列、删列、类型、可空、注释（列和表）、主键、唯一约束、外键各出一条；疑似改名只说「可能」（类型和位置都对上、 +而且只有一对）；读不到唯一约束、外键的一边不比；离开范围、读不到结构不算删除；升级后第一次读到（版本 0）只建基线。 +表、列都是虚构的通用名字。 +""" +from __future__ import annotations + +import copy + +from app.data import schema_diff as sd + + +def _meta(**over): + meta = { + "qualified": "orders", "schema": None, "comment": "订单", "is_view": False, + "columns": [ + {"name": "id", "type": "INTEGER", "nullable": False, "comment": None}, + {"name": "status", "type": "VARCHAR(10)", "nullable": True, "comment": "状态"}, + {"name": "amount", "type": "NUMERIC(12, 2)", "nullable": True, "comment": None}, + {"name": "paid_at", "type": "TIMESTAMP", "nullable": True, "comment": None}, + ], + "primary_key": ["id"], + "unique": [], + "foreign_keys": [{"columns": ["store_id"], "to_table": "stores", "to_columns": ["id"]}], + } + meta.update(over) + return meta + + +def _kinds(diffs): + return [(d.kind, d.column) for d in diffs] + + +def test_same_structure_has_no_changes(): + assert sd.diff_table(_meta(), _meta()) == [] + + +def test_added_removed_type_nullable_and_comment_each_get_one_change(): + before = _meta() + after = copy.deepcopy(before) + cols = after["columns"] + cols[1]["type"] = "VARCHAR(20)" # 类型变长 + cols[2]["nullable"] = False # 改成非空 + cols[3]["comment"] = "支付时间" # 补了注释 + cols.append({"name": "channel", "type": "VARCHAR(8)", "nullable": True, "comment": None}) + after["columns"] = [c for c in cols if c["name"] != "amount"] + [] + after["columns"].insert(2, {"name": "amount", "type": "NUMERIC(12, 2)", "nullable": False, "comment": None}) + after["comment"] = "订单明细" + diffs = sd.diff_table(before, after) + assert _kinds(diffs) == [ + ("column_added", "channel"), + ("column_type_changed", "status"), ("column_nullable_changed", "amount"), + ("comment_changed", "paid_at"), ("comment_changed", None), + ] + by = {(d.kind, d.column): d for d in diffs} + assert by[("column_type_changed", "status")].before == "VARCHAR(10)" + assert by[("column_type_changed", "status")].after == "VARCHAR(20)" + assert by[("column_nullable_changed", "amount")].before is True and by[("column_nullable_changed", "amount")].after is False + assert by[("comment_changed", None)].after == "订单明细" + + +def test_removed_column_is_a_deletion_and_left_scope_is_not(): + after = _meta(columns=[c for c in _meta()["columns"] if c["name"] != "amount"]) + [d] = sd.diff_table(_meta(), after) + assert d.kind == "column_removed" and d.column == "amount" and d.before["type"] == "NUMERIC(12, 2)" + assert d.rename is None, "没有新加的列，谈不上改名" + assert sd.is_deletion("column_removed") and sd.is_deletion("table_removed") + assert not sd.is_deletion("table_left_scope") and not sd.is_deletion("table_reflect_failed"), "范围外、读不到不算删除" + assert d.severity == "warning" + + +def test_same_type_same_position_is_only_a_possible_rename(): + after = copy.deepcopy(_meta()) + after["columns"][1] = {"name": "state", "type": "VARCHAR(10)", "nullable": True, "comment": "状态"} + diffs = sd.diff_table(_meta(), after) + assert _kinds(diffs) == [("column_removed", "status"), ("column_added", "state")] + assert diffs[0].rename == {"from": "status", "to": "state"} == diffs[1].rename + assert diffs[0].kind == "column_removed", "改名只说「可能」：照样是一条删列、一条加列" + + +def test_rename_needs_same_type_and_same_position(): + moved = copy.deepcopy(_meta()) + moved["columns"] = [moved["columns"][0], moved["columns"][2], moved["columns"][3], + {"name": "state", "type": "VARCHAR(10)", "nullable": True, "comment": None}] + diffs = sd.diff_table(_meta(), moved) + assert all(d.rename is None for d in diffs), "位置不同：不说改名" + retyped = copy.deepcopy(_meta()) + retyped["columns"][1] = {"name": "state", "type": "INTEGER", "nullable": True, "comment": None} + assert all(d.rename is None for d in sd.diff_table(_meta(), retyped)), "类型不同：不说改名" + + +def test_keys_and_foreign_keys(): + after = _meta(primary_key=["id", "status"], unique=[["paid_at"]], + foreign_keys=[{"columns": ["store_id"], "to_table": "shops", "to_columns": ["id"]}]) + diffs = sd.diff_table(_meta(), after) + assert [(d.kind, d.key, d.column) for d in diffs] == [ + ("key_changed", "primary", None), ("key_changed", "unique", "paid_at"), + ("fk_changed", None, "store_id"), ("fk_changed", None, "store_id")] + fk_gone, fk_new = diffs[2], diffs[3] + assert fk_gone.before["to_table"] == "stores" and fk_gone.after is None + assert fk_new.after["to_table"] == "shops" and fk_new.before is None + + +def test_constraints_not_read_on_one_side_are_not_compared(): + """反射读不到唯一约束、外键时不写那两个键：不知道 ≠ 删了。""" + after = _meta() + del after["unique"], after["foreign_keys"] + assert sd.diff_table(_meta(), after) == [] + + +def test_baseline_after_upgrade_reports_nothing(): + """升级前的老缓存接进来的结构版本是 0：第一次按对象索引读到只建基线，通用写法换成方言完整写法不算类型变了。""" + legacy = _meta() + assert sd.is_baseline(0, legacy) and sd.is_baseline(None, legacy) + assert sd.is_baseline(3, None) and sd.is_baseline(3, {"columns": []}), "上一次的结构不在手上：没法比" + assert not sd.is_baseline(1, legacy) + + +def test_to_dict_keeps_rename_and_key(): + d = sd.Diff("key_changed", None, before=["id"], after=None, key="primary") + assert d.to_dict() == {"kind": "key_changed", "column": None, "before": ["id"], "after": None, "key": "primary"} + assert sd.touched_columns([sd.Diff("column_removed", "a"), sd.Diff("column_type_changed", "b"), + sd.Diff("column_added", "c")]) == {"removed": {"a"}, "changed": {"b"}}