From a6f7572a9ee9cfd5f1c32edd43321956f7872c1e Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Wed, 21 Oct 2026 13:12:05 +0800 Subject: [PATCH] =?UTF-8?q?=E6=8A=A5=E5=91=8A=E6=96=87=E6=A1=A3=E9=87=8C?= =?UTF-8?q?=E7=B3=BB=E7=BB=9F=E5=86=99=E7=9A=84=E3=80=8C=E5=88=86=E6=9E=90?= =?UTF-8?q?=E8=8C=83=E5=9B=B4=E3=80=8D=EF=BC=88engine/analysis=5Fscope?= =?UTF-8?q?=EF=BC=8C=E5=A5=91=E7=BA=A6=20ScopeBlock=EF=BC=89=EF=BC=9A?= =?UTF-8?q?=E6=8C=89=E5=86=99=E4=BD=9C=E7=9B=AE=E5=BD=95=E9=87=8C=E7=9A=84?= =?UTF-8?q?=20A=20=E6=9D=A1=E7=9B=AE=E7=85=A7=E5=AE=9E=E5=86=99=E7=8E=B0?= =?UTF-8?q?=E8=B1=A1=EF=BC=88=E5=8F=AF=E6=AF=94=E6=80=A7=E3=80=81=E5=9F=BA?= =?UTF-8?q?=E7=BA=BF=E3=80=81=E8=B6=8B=E5=8A=BF=EF=BC=89=E3=80=81=E6=8B=86?= =?UTF-8?q?=E8=BF=87=EF=BC=88=E6=8C=89=E4=B8=8B=E9=92=BB=E9=93=BE=EF=BC=8C?= =?UTF-8?q?=E4=B8=8A=E4=B8=80=E5=B1=82=E5=9C=A8=E5=89=8D=EF=BC=89=E3=80=81?= =?UTF-8?q?=E6=8E=92=E9=99=A4=EF=BC=88=E5=88=86=E6=95=A3=E7=9A=84=E6=8B=86?= =?UTF-8?q?=E8=A7=A3=E3=80=81=E5=B7=AE=E5=BC=82=E5=8F=AF=E8=83=BD=E5=8F=AA?= =?UTF-8?q?=E6=98=AF=E6=A0=B7=E6=9C=AC=E6=B3=A2=E5=8A=A8=E7=9A=84=E6=A3=80?= =?UTF-8?q?=E9=AA=8C=EF=BC=89=E3=80=81=E7=AE=97=E4=B8=8D=E4=BA=86=EF=BC=88?= =?UTF-8?q?=E5=B8=A6=E5=8E=9F=E5=9B=A0=EF=BC=89=E3=80=81=E6=A3=80=E9=AA=8C?= =?UTF-8?q?=E6=AC=A1=E6=95=B0=EF=BC=9Bagent=20=E5=86=99=E7=9A=84=E5=81=87?= =?UTF-8?q?=E8=AE=BE=E4=B8=8D=E8=BF=9B=E8=BF=99=E4=B8=80=E6=AE=B5=E3=80=82?= =?UTF-8?q?=E6=B2=A1=E6=8B=86=E7=9A=84=E5=80=99=E9=80=89=E7=BB=B4=E5=BA=A6?= =?UTF-8?q?=E7=A1=AE=E5=AE=9A=E5=9C=B0=E4=BB=8E=E6=95=B0=E6=8D=AE=E7=9B=AE?= =?UTF-8?q?=E5=BD=95=E7=AE=97=EF=BC=9A=E6=8B=86=E8=A7=A3=E6=89=80=E7=94=A8?= =?UTF-8?q?=E6=9F=A5=E8=AF=A2=20SQL=20=E9=87=8C=E7=9A=84=E4=BA=8B=E5=AE=9E?= =?UTF-8?q?=E8=A1=A8=EF=BC=88=E7=9B=AE=E5=BD=95=E6=A0=87=E4=BA=86=20fact?= =?UTF-8?q?=20/=20log=EF=BC=8C=E6=B2=A1=E6=A0=87=E4=B8=94=E5=8F=AA?= =?UTF-8?q?=E8=AF=BB=E4=B8=80=E5=BC=A0=E8=A1=A8=E6=97=B6=E5=B0=B1=E6=98=AF?= =?UTF-8?q?=E5=AE=83=EF=BC=89=E4=B8=8A=E6=A0=87=E6=88=90=E5=B1=9E=E6=80=A7?= =?UTF-8?q?=E3=80=81=E7=8A=B6=E6=80=81=E3=80=81=E4=B8=8D=E9=81=AE=E7=BD=A9?= =?UTF-8?q?=EF=BC=88=E6=95=B0=E6=8D=AE=E6=BA=90=E8=AE=BE=E7=BD=AE=E5=92=8C?= =?UTF-8?q?=E6=9F=A5=E8=AF=A2=E5=BD=93=E6=97=B6=E8=AE=B0=E4=B8=8B=E7=9A=84?= =?UTF-8?q?=E9=81=AE=E7=BD=A9=E5=88=97=EF=BC=89=E3=80=81=E5=8F=96=E5=80=BC?= =?UTF-8?q?=E4=B8=8D=E5=A4=9A=EF=BC=88=E7=A0=81=E5=80=BC=E8=A1=A8=E4=B8=8D?= =?UTF-8?q?=E8=B6=85=E8=BF=87=2050=20=E4=B8=AA=E5=8F=96=E5=80=BC=E6=88=96?= =?UTF-8?q?=E5=B7=B2=E5=88=97=E5=85=A8=EF=BC=89=E7=9A=84=E5=88=97=EF=BC=8C?= =?UTF-8?q?=E5=8A=A0=E4=B8=8A=E5=85=B3=E8=81=94=E7=9A=84=E7=BB=B4=E5=BA=A6?= =?UTF-8?q?=E8=A1=A8=EF=BC=88=E6=A0=87=E4=BA=86=20dimension=20=E6=88=96?= =?UTF-8?q?=E5=A4=9A=E5=AF=B9=E4=B8=80=EF=BC=89=E4=B8=8A=E5=90=8C=E6=A0=B7?= =?UTF-8?q?=E7=9A=84=E5=88=97=EF=BC=9B=E6=8B=86=E8=A7=A3=E6=8C=89=E7=9A=84?= =?UTF-8?q?=E5=88=97=E5=AF=B9=E4=B8=8D=E4=B8=8A=E7=9B=AE=E5=BD=95=E3=80=81?= =?UTF-8?q?=E8=AF=BB=E4=B8=8D=E5=88=B0=E7=9B=AE=E5=BD=95=E3=80=81=E6=BA=90?= =?UTF-8?q?=E8=BF=98=E6=B2=A1=E6=9C=89=E7=9B=AE=E5=BD=95=E3=80=81=E5=8F=96?= =?UTF-8?q?=E4=B8=8D=E5=9B=9E=E6=9F=A5=E8=AF=A2=E3=80=81=E8=AE=A4=E4=B8=8D?= =?UTF-8?q?=E5=87=BA=E4=BA=8B=E5=AE=9E=E8=A1=A8=E7=9A=84=EF=BC=8Cnot=5Fdec?= =?UTF-8?q?omposed=20=E4=B8=BA=20None=E3=80=81note=20=E7=85=A7=E5=AE=9E?= =?UTF-8?q?=E5=86=99=E6=A0=B8=E5=AF=B9=E4=B8=8D=E4=BA=86=E3=80=82=E7=9B=AE?= =?UTF-8?q?=E5=BD=95=E7=BB=8F=E8=BF=99=E6=AC=A1=E8=BF=90=E8=A1=8C=E7=9A=84?= =?UTF-8?q?=E7=9B=AE=E5=BD=95=E8=A7=86=E5=9B=BE=E8=AF=BB=EF=BC=88data/cata?= =?UTF-8?q?log=5Fview=EF=BC=8C=E5=85=B3=E6=8E=89=E7=9A=84=E5=BD=93?= =?UTF-8?q?=E6=A0=B8=E5=AF=B9=E4=B8=8D=E4=BA=86=EF=BC=89=EF=BC=8C=E5=87=BA?= =?UTF-8?q?=E9=94=99=E4=B8=8D=E6=8C=A1=E6=8A=A5=E5=91=8A=E3=80=82=E5=A5=91?= =?UTF-8?q?=E7=BA=A6=E7=9A=84=20ScopeBlock=20=E5=8A=A0=E4=B8=80=E9=A1=B9?= =?UTF-8?q?=20corrected=EF=BC=88=E6=A3=80=E9=AA=8C=E4=B9=8B=E9=97=B4?= =?UTF-8?q?=E6=9C=89=E6=B2=A1=E6=9C=89=E6=A0=A1=E6=AD=A3=EF=BC=9A=E5=8F=AA?= =?UTF-8?q?=E5=81=9A=E4=B8=80=E6=AC=A1=E4=B8=BA=20true=E3=80=81=E4=B8=8D?= =?UTF-8?q?=E6=AD=A2=E4=B8=80=E6=AC=A1=E4=B8=BA=20false=E3=80=81=E6=B2=A1?= =?UTF-8?q?=E5=81=9A=E4=B8=BA=20None=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/analysis/contract.py | 2 + backend/app/engine/analysis_scope.py | 352 +++++++++++++++++++++++++++ backend/tests/test_analysis_scope.py | 206 ++++++++++++++++ 3 files changed, 560 insertions(+) create mode 100644 backend/app/engine/analysis_scope.py create mode 100644 backend/tests/test_analysis_scope.py diff --git a/backend/app/analysis/contract.py b/backend/app/analysis/contract.py index e70d4cf8..72117262 100644 --- a/backend/app/analysis/contract.py +++ b/backend/app/analysis/contract.py @@ -360,6 +360,8 @@ class ScopeBlock(TypedDict, total=False): not_decomposed: list[str] | None # 候选维度里没拆过的叫法；None = 候选维度核对不了 note: str | None # 核对不了时的一句说明 tests: int # 这次做了几次差异检验 + corrected: bool | None # 检验校正过没有：只做了一次为 True（一次里比了多对的已按次数校正），不止一次为 + # False（几次之间没有校正），没做为 None #: 复算核对时不比的字段：hypothesis 是 agent 写的话，不是算出来的 diff --git a/backend/app/engine/analysis_scope.py b/backend/app/engine/analysis_scope.py new file mode 100644 index 00000000..c5755f54 --- /dev/null +++ b/backend/app/engine/analysis_scope.py @@ -0,0 +1,352 @@ +"""报告文档里系统写的「分析范围」（docs/design/analysis-round/00-总方案.md §3.6，契约 contract.ScopeBlock）。 + +报告里用到了分析结论（这次运行有分析结论）时，系统按写作目录里的 A 条目照实写一段：查了什么、排除了什么、哪些没查。 +它是报告文档的结构化字段 analysis_scope（D17），不进正文、不过裸数字和说法核对（不是模型写的），写作者改不了，随文档 +封存；网页、导出、可验证报告各自按它渲染，编号点得开。 + +- **现象**（phenomena）：做成了的可比性、基线、趋势结论 +- **拆过**（decomposed）：做成了的变化拆解，按下钻链的先后（上一层在前，下钻的紧跟在它后面） +- **排除**（excluded）：判定为分散的拆解、差异可能只是样本波动的检验——拆了没发现集中也是结论 +- **算不了**（refused）：各种算子算不了的结论，带原因 +- **没拆**（not_decomposed）：候选维度里没拆过的。候选维度确定地从数据目录算：拆解所用查询的 SQL 里的事实表（目录标了 + fact / log；没标的、查询只读一张表时就是它）上标成属性、状态（attribute / status）、不遮罩、取值不多（码值表不超过 + CANDIDATE_MAX 个取值，或者已列全）的列，加上它关联的维度表（目录标了 dimension，或者关系是多对一）上同样的列。 + 拆解按的列对不上目录、读不到目录、认不出事实表的，not_decomposed 为 None，note 写一句为什么核对不了——不乱列 +- **检验**（tests）：这次做了几次差异检验（做成了的）；corrected：一次检验比了多对的按次数校正过，几次检验之间没有校正， + 做了不止一次时为 False（契约之外加的一项，见交付说明） + +agent 写的 hypothesis 不进这一段（模型写的字不进系统渲染的正文，细案 01 §7.3）。build 是纯函数（目录、遮罩由调用方读好 +交进来）；scope_of 给报告节点用，读这次运行的数据目录视图（data/catalog_view），读不到当核对不了，不挡生成。 +""" +from __future__ import annotations + +import logging +from collections.abc import Callable, Iterable, Mapping +from typing import Any + +from app.analysis.contract import ScopeBlock, ScopeItem, catalog_result +from app.data.names import name_key + +logger = logging.getLogger(__name__) + +#: 候选维度「取值不多」的上限：码值表里的取值不超过这么多（取值观察最多也只取这么多，data/catalog_profile_observe） +CANDIDATE_MAX = 50 +#: 现象：可比性、基线、趋势 +_PHENOMENA = ("comparable", "baseline", "trend") +#: 事实表的表类型（数据目录的 kind）：流水、日志也是一行一件事 +_FACT_KINDS = ("fact", "log") +#: 候选维度的度量类型 +_DIMENSION_MEASURES = ("attribute", "status") +#: 拆解的输入快照（下钻的上一层不是快照） +_DATA_ROLES = ("data", "base", "target") + +NOTE_NO_CATALOG = "这次运行读不到数据目录，候选维度核对不了" +NOTE_NO_NOTES = "拆解所用的数据源还没有数据目录，候选维度核对不了" +NOTE_NO_DECOMPOSE = "这次没有按维度拆解，候选维度没有核对" +NOTE_NO_QUERY = "拆解所用的查询取不回来，候选维度核对不了" +NOTE_NO_FACT = "拆解所用查询的表在数据目录里认不出哪张是事实表，候选维度核对不了" + + +def _alias_order(alias: str) -> tuple[int, str]: + tail = alias[1:] + return (int(tail), alias) if tail.isdigit() else (1 << 30, alias) + + +def _value(item: Any) -> Any: + """目录里一个项的值：驳回的（rejected）当没有。""" + if not isinstance(item, Mapping) or item.get("status") == "rejected": + return None + return item.get("value") + + +def _dims(result: Mapping[str, Any]) -> list[str]: + """拆解按的维度列（params.by），乘积拆解没有维度。""" + if result.get("op") != "decompose" or result.get("mode") == "factor": + return [] + by = (result.get("params") or {}).get("by") + return [str(b) for b in (by if isinstance(by, list) else [by] if isinstance(by, str) and by else [])] + + +def _dim_labels(result: Mapping[str, Any]) -> list[str]: + labels = ((result.get("table") or {}).get("labels") or {}) + return [str(labels.get(d) or d) for d in _dims(result)] + + +def _code(result: Mapping[str, Any]) -> str | None: + """结论的判定码；没有判定（算不了）为 None。""" + verdict = result.get("verdict") + return verdict.get("code") if isinstance(verdict, Mapping) else None + + +def _item(alias: str, result: Mapping[str, Any], parent: str | None) -> ScopeItem: + return {"alias": alias, "op": str(result.get("op") or ""), "mode": result.get("mode"), + "dims": _dim_labels(result), "verdict": _code(result), + "phrase": str(result.get("phrase") or ""), "parent": parent, + "reason": result.get("reason") if result.get("status") != "ok" else None} + + +def _parent(result: Mapping[str, Any], by_artifact: Mapping[str, str]) -> str | None: + """下钻的上一层：输入里 role 为 parent 的那份结论（按工件对回写作目录里的编号）。""" + for item in result.get("inputs") or []: + if isinstance(item, Mapping) and item.get("role") == "parent" and item.get("artifact") in by_artifact: + return by_artifact[str(item["artifact"])] + return None + + +def _chain(items: list[ScopeItem]) -> list[ScopeItem]: + """按下钻链排：上一层在前，下钻的紧跟在它后面（同一层按编号）。上一层不在这几项里的当作最上层。""" + known = {i["alias"] for i in items} + children: dict[str | None, list[ScopeItem]] = {} + for item in items: + children.setdefault(item["parent"] if item["parent"] in known else None, []).append(item) + out: list[ScopeItem] = [] + seen: set[str] = set() + + def walk(parent: str | None) -> None: + for item in children.get(parent, []): + if item["alias"] in seen: + continue + seen.add(item["alias"]) + out.append(item) + walk(item["alias"]) + + walk(None) + out.extend(i for i in items if i["alias"] not in seen) # 链成了环（不该有）也照列 + return out + + +def results(catalog: Mapping[str, Any]) -> list[tuple[str, dict[str, Any], dict[str, Any]]]: + """写作目录里的分析条目，按编号：[(A 编号, 条目, 结论)]。""" + found = [(str(alias), entry, dict(result)) for alias, entry in catalog.items() + if isinstance(entry, dict) and (result := catalog_result(entry)) is not None] + return sorted(found, key=lambda t: _alias_order(t[0])) + + +# -------------------------------------------------------------------------- +# 候选维度 +# -------------------------------------------------------------------------- + + +def _table_notes(tables: Mapping[str, Any], name: str) -> tuple[str, dict[str, Any]] | None: + """SQL 里写的表名在数据目录里的那一张：按 name_key 比，schema 前缀去掉再比一次。""" + wanted = {name_key(name), name_key(name.rsplit(".", 1)[-1])} + for table, notes in tables.items(): + if isinstance(notes, Mapping) and (name_key(str(table)) in wanted + or name_key(str(table).rsplit(".", 1)[-1]) in wanted): + return str(table), dict(notes) + return None + + +def _few_values(column: Mapping[str, Any]) -> bool: + """取值不多：码值表列出的取值不超过 CANDIDATE_MAX 个，或者标了已列全。""" + codes = column.get("codes") + if not isinstance(codes, Mapping) or codes.get("status") == "rejected": + return False + values = codes.get("value") + return codes.get("complete") is True or (isinstance(values, Mapping) and 0 < len(values) <= CANDIDATE_MAX) + + +def _candidates(table: str, notes: Mapping[str, Any], masked: set[str]) -> list[tuple[str, str, str]]: + """一张表上的候选维度：(表, 列, 叫法)，按目录里的顺序。""" + out = [] + for column, body in (notes.get("columns") or {}).items(): + if not isinstance(body, Mapping) or name_key(str(column)) in masked: + continue + if _value(body.get("measure")) in _DIMENSION_MEASURES and _few_values(body): + label = _value(body.get("label")) + out.append((table, str(column), str(label) if isinstance(label, str) and label.strip() else str(column))) + return out + + +def _related(notes: Mapping[str, Any], tables: Mapping[str, Any]) -> list[tuple[str, dict[str, Any]]]: + """事实表关联的维度表：关系没被驳回、对面那张表目录里标了 dimension 或关系是多对一。""" + out = [] + for rel in notes.get("relations") or []: + if not isinstance(rel, Mapping) or rel.get("status") == "rejected" or not rel.get("to_table"): + continue + found = _table_notes(tables, str(rel["to_table"])) + if found and (_value(found[1].get("kind")) == "dimension" or rel.get("cardinality") == "many_to_one"): + out.append(found) + return out + + +def _fact_tables(names: list[str], tables: Mapping[str, Any]) -> list[tuple[str, dict[str, Any]]] | None: + """查询里的事实表；认不出的为 None。""" + found = [hit for name in names if (hit := _table_notes(tables, name))] + facts = [(t, n) for t, n in found if _value(n.get("kind")) in _FACT_KINDS] + if facts: + return facts + if len(names) == 1 and len(found) == 1 and _value(found[0][1].get("kind")) in (None, ""): + return found # 只读一张表、目录没标表类型：就是它 + return None + + +def _names(raw: Any) -> list[str]: + """遮罩列：列表，或者「a, b」这样的一段文字（同数据源设置里的写法）。""" + from app.data.engine import masked_columns + + return masked_columns({"mask_columns": raw}) if raw else [] + + +def _undecomposed(decomposes: list[dict[str, Any]], catalog: Mapping[str, Any], + sources: Mapping[str, Mapping[str, Any]] | None, + load: Callable[[str], Any]) -> tuple[list[str] | None, str | None]: + """(没拆过的候选维度的叫法, 核对不了时的说明)。""" + from app.engine.evidence import sql_tables + + if sources is None: + return None, NOTE_NO_CATALOG + if not decomposes: + return None, NOTE_NO_DECOMPOSE + queries = {str(e.get("artifact")): e for e in catalog.values() + if isinstance(e, dict) and e.get("kind") == "query" and e.get("artifact")} + pool: dict[tuple[str, str, str], str] = {} # (源, 表, 列 name_key) → 叫法，按目录里的顺序 + used: set[tuple[str, str, str]] = set() + for result in decomposes: + snapshots = [i.get("artifact") for i in result.get("inputs") or [] + if isinstance(i, Mapping) and i.get("role") in _DATA_ROLES and i.get("artifact")] + if not snapshots: + return None, NOTE_NO_QUERY + known: dict[str, tuple[str, str, str]] = {} # 这份拆解能对上的列：name_key → (源, 表, 列) + for artifact in snapshots: + try: + snapshot = load(str(artifact)) + except (ValueError, OSError): + snapshot = None + if not isinstance(snapshot, Mapping): + return None, NOTE_NO_QUERY + source = str(snapshot.get("source") or (queries.get(str(artifact)) or {}).get("source") or "") + facts_of = sources.get(source) if source else None + if facts_of is None: + return None, NOTE_NO_CATALOG + tables = facts_of.get("tables") or {} + if not tables: + return None, NOTE_NO_NOTES + masked = {name_key(str(c)) for c in [*_names(facts_of.get("masked")), *_names(snapshot.get("mask_columns"))]} + facts = _fact_tables(sql_tables(str(snapshot.get("sql") or "")), tables) + if facts is None: + return None, NOTE_NO_FACT + for table, notes in [*facts, *(r for t, n in facts for r in _related(n, tables))]: + for column in (notes.get("columns") or {}): + known.setdefault(name_key(str(column)), (source, table, name_key(str(column)))) + for _, column, label in _candidates(table, notes, masked): + pool.setdefault((source, table, name_key(column)), label) + for dim, label in zip(_dims(result), _dim_labels(result)): + hit = known.get(name_key(dim)) + if hit is None: + return None, f"候选维度核对不了：拆解按的「{label}」对不上数据目录里的列" + used.add(hit) + return [label for key, label in pool.items() if key not in used], None + + +# -------------------------------------------------------------------------- +# 组装 +# -------------------------------------------------------------------------- + + +def build(catalog: Mapping[str, Any], *, sources: Mapping[str, Mapping[str, Any]] | None = None, + load: Callable[[str], Any] | None = None) -> ScopeBlock | None: + """按写作目录里的分析条目算「分析范围」；目录里没有分析结论返回 None（不写这一项）。 + + sources 是 {数据源名: {"tables": {表: 数据目录 notes}, "masked": [遮罩列]}}（读不到目录时为 None，候选维度核对不了）； + load 按 sha256 取回拆解的输入快照（缺省 artifact_store.load），认 SQL 用的是哪几张表。""" + if load is None: + from app.core.artifact_store import load as load_artifact + + load = load_artifact + found = results(catalog) + if not found: + return None + by_artifact = {str(e.get("artifact")): alias for alias, e, _ in found if e.get("artifact")} + items = {alias: _item(alias, r, _parent(r, by_artifact)) for alias, _, r in found} + ok = [(alias, r) for alias, _, r in found if r.get("status") == "ok"] + decomposes = [r for _, r in ok if r.get("op") == "decompose"] + excluded = [items[a] for a, r in ok + if (r.get("op") == "decompose" and _code(r) == "dispersed") + or (r.get("op") == "differ" and _code(r) == "no_evidence")] + tests = sum(1 for _, r in ok if r.get("op") == "differ") + try: + missing, note = _undecomposed([r for r in decomposes if _dims(r)], catalog, sources, load) + except Exception: # noqa: BLE001 - 分析范围是附带的说明，算不出来照实说核对不了，不挡报告 + logger.exception("分析范围的候选维度没有算成") + missing, note = None, "候选维度核对不了：核对时出了错" + return { + "phenomena": [items[a] for a, r in ok if r.get("op") in _PHENOMENA], + "decomposed": _chain([items[a] for a, r in ok if r.get("op") == "decompose"]), + "excluded": excluded, + "refused": [items[a] for a, _, r in found if r.get("status") != "ok"], + "not_decomposed": missing, + "note": note, + "tests": tests, + "corrected": None if not tests else tests == 1, + } + + +async def source_facts(names: Iterable[str], view: Any = None) -> dict[str, dict[str, Any]] | None: + """几个数据源（按名字）在这次运行的目录视图下的数据目录和遮罩列：{源名: {"tables": {表: notes}, "masked": [列]}}。 + 目录视图关掉的、读库出错的返回 None（候选维度核对不了）；没有目录的源 tables 为空。""" + from sqlalchemy import select + + from app.data import catalog_view + from app.data.engine import masked_columns + from app.db.base import SessionLocal + from app.db.models import DataSource + + wanted = [n for n in dict.fromkeys(str(n) for n in names) if n] + view = catalog_view.CatalogView.parse(view) + if view.mode == "off": + return None + try: + async with SessionLocal() as session: + rows = list((await session.execute(select(DataSource).where(DataSource.name.in_(wanted)))).scalars()) \ + if wanted else [] + entries = await catalog_view.read_many(session, view, [r.id for r in rows]) + except Exception: # noqa: BLE001 - 目录是附带的说明，读不到照实说核对不了 + logger.exception("分析范围读取数据目录失败") + return None + return {row.name: {"tables": {t: e.notes for t, e in entries.get(row.id, {}).items()}, + "masked": masked_columns(row.options)} for row in rows} + + +async def scope_of(catalog: Mapping[str, Any], view: Any = None, *, + load: Callable[[str], Any] | None = None) -> ScopeBlock | None: + """报告节点用：读好拆解所用查询的数据源的目录，算「分析范围」。没有分析结论返回 None。""" + if load is None: + from app.core.artifact_store import load as load_artifact + + load = load_artifact + found = results(catalog) + if not found: + return None + sources: dict[str, dict[str, Any]] | None = None + try: + sources = await source_facts(_source_names(found, catalog, load), view) + except Exception: # noqa: BLE001 - 分析范围是附带的说明，读不到目录照实说核对不了，不挡报告 + logger.exception("分析范围没有读成数据目录") + return build(catalog, sources=sources, load=load) + + +def _source_names(found: list[tuple[str, dict[str, Any], dict[str, Any]]], catalog: Mapping[str, Any], + load: Callable[[str], Any]) -> list[str]: + """分析结论的输入快照来自哪几个数据源：写作目录里那次查询记的源名，没有就读快照。""" + names: list[str] = [] + for _, _, result in found: + for item in result.get("inputs") or []: + if not (isinstance(item, Mapping) and item.get("role") in _DATA_ROLES and item.get("artifact")): + continue + entry = next((e for e in catalog.values() if isinstance(e, dict) and e.get("kind") == "query" + and e.get("artifact") == item["artifact"]), None) + name = (entry or {}).get("source") + if not name: + try: + snapshot = load(str(item["artifact"])) + except (ValueError, OSError): + snapshot = None + name = snapshot.get("source") if isinstance(snapshot, Mapping) else None + if name: + names.append(str(name)) + return names + + +__all__ = ["CANDIDATE_MAX", "NOTE_NO_CATALOG", "NOTE_NO_DECOMPOSE", "NOTE_NO_FACT", "NOTE_NO_NOTES", "NOTE_NO_QUERY", "build", + "results", "scope_of", "source_facts"] diff --git a/backend/tests/test_analysis_scope.py b/backend/tests/test_analysis_scope.py new file mode 100644 index 00000000..d9e956e9 --- /dev/null +++ b/backend/tests/test_analysis_scope.py @@ -0,0 +1,206 @@ +"""报告文档里系统写的「分析范围」（engine/analysis_scope.py，00-总方案 §3.6，契约 contract.ScopeBlock）。 + +- 现象、拆过（按下钻链）、排除（分散、差异可能只是样本波动）、算不了、检验次数 +- 没拆：候选维度确定地从数据目录算——拆解所用查询的事实表上标成属性、状态、不遮罩、取值不多的列，加上它关联的维度表； + 拆解按的列对不上目录、读不到目录、认不出事实表时只说核对不了，不乱列 +- agent 写的假设不进这一段 +""" +from __future__ import annotations + +import sqlite3 +import uuid + +import pytest + +from app.analysis.testing import fake_result +from app.data import catalog as data_catalog +from app.data.catalog import make_item +from app.data.catalog_view import CatalogView +from app.data.introspect import introspect +from app.db.base import SessionLocal +from app.db.models import DataSource +from app.engine import analysis_scope as scope +from app.engine.evidence import make_eid +from tests.fixtures.sources import drop_source + +SQL = "SELECT o.region, o.week, SUM(o.amount) AS gmv FROM orders o JOIN stores s ON s.id = o.store_id GROUP BY 1, 2" +SNAP = "1" * 64 +SNAPSHOTS = {SNAP: {"source": "shop", "sql": SQL, "columns": ["region", "week", "gmv"], "rows": []}} + + +def codes(*values: str, complete: bool = False) -> dict: + item = make_item({v: "" for v in values}, "profile", "verified") + return {**item, "complete": True} if complete else item + + +def column(label: str, measure: str | None = None, codes_item: dict | None = None) -> dict: + out = {"label": make_item(label, "human")} + if measure: + out["measure"] = make_item(measure, "human") + if codes_item: + out["codes"] = codes_item + return out + + +NOTES = { + "orders": { + "kind": make_item("fact", "human"), + "columns": { + "region": column("地区", "attribute", codes("华东", "华南", "华北", "西南")), + "channel": column("渠道", "attribute", codes("线上", "线下")), + "status": column("订单状态", "status", codes("1", "2", "9", complete=True)), + "member_level": column("会员等级", "attribute"), # 没有码值：取值多不多说不清 + "customer_id": column("客户编号", "identifier", codes("c1", "c2")), + "phone": column("手机号", "attribute", codes("1", "2")), # 遮罩列 + "amount": column("实收金额", "flow"), + "week": column("周", "attribute", codes(*(f"W{i}" for i in range(60)))), # 取值太多 + }, + "relations": [{"id": "r1", "columns": ["store_id"], "to_table": "stores", "to_columns": ["id"], + "cardinality": "many_to_one", "source": "fk", "status": "verified"}], + }, + "stores": { + "kind": make_item("dimension", "human"), + "columns": {"store_type": column("门店类型", "attribute", codes("旗舰", "社区")), + "city": column("城市", "attribute", codes("上海", "杭州"))}, + }, +} +SOURCES = {"shop": {"tables": NOTES, "masked": ["phone"]}} + + +def entry(alias: str, result: dict) -> dict: + artifact = "a" * 63 + alias[-1] + return {"alias": alias, "kind": "analysis", "artifact": artifact, "eid": make_eid("analysis", artifact, {}), + "locator": {}, "result": result} + + +def data(snapshot: str = SNAP) -> list[dict]: + return [{"ref": "Q1", "artifact": snapshot, "role": "data", "columns": ["region", "week", "gmv"], "rows": "all"}] + + +def build(catalog: dict, *, sources=SOURCES, snapshots=SNAPSHOTS): + def load(artifact): + if artifact not in snapshots: + raise ValueError("取不回来") + return snapshots[artifact] + + return scope.build(catalog, sources=sources, load=load) + + +def test_no_conclusions_no_scope(): + assert scope.build({"m:gmv": {"kind": "metric"}}, sources=SOURCES) is None + + +def test_the_sections_follow_the_verdicts(): + region = fake_result(inputs=data()) + channel = fake_result(verdict="dispersed", inputs=[*data(), {"ref": "A1", "artifact": "a" * 63 + "1", + "role": "parent"}], + params={"by": ["channel"], "label": "销售额"}, hypothesis="华东是不是线上拖累的") + channel["table"]["columns"][0] = "channel" + channel["table"]["labels"] = {"channel": "渠道"} + catalog = { + "A1": entry("A1", region), + "A2": entry("A2", fake_result("baseline", None, "outside_band")), + "A3": entry("A3", channel), + "A4": entry("A4", fake_result("differ", "rate", "no_evidence")), + "A5": entry("A5", fake_result("trend", None, None, status="refused", reason_code="history_short")), + "A6": entry("A6", fake_result("differ", "rate", "differs")), + } + block = build(catalog) + assert [i["alias"] for i in block["phenomena"]] == ["A2"] + assert [(i["alias"], i["parent"], i["dims"], i["verdict"]) for i in block["decomposed"]] == [ + ("A1", None, ["地区"], "driver"), ("A3", "A1", ["渠道"], "dispersed")] + assert [i["alias"] for i in block["excluded"]] == ["A3", "A4"] + assert [(i["alias"], i["reason"]) for i in block["refused"]] == [("A5", "历史只有 5 期，至少要 8 期")] + assert block["tests"] == 2 and block["corrected"] is False + assert block["decomposed"][0]["phrase"] == region["phrase"] + assert "华东是不是线上拖累的" not in str(block), "agent 写的假设不进分析范围" + + +def test_drill_chains_keep_each_parent_ahead_of_its_children(): + def child(of: str) -> dict: + return fake_result(inputs=[*data(), {"ref": of, "artifact": "a" * 63 + of[-1], "role": "parent"}]) + + catalog = {"A1": entry("A1", fake_result(inputs=data())), "A2": entry("A2", fake_result(inputs=data())), + "A3": entry("A3", child("A1")), "A4": entry("A4", child("A3"))} + assert [i["alias"] for i in build(catalog)["decomposed"]] == ["A1", "A3", "A4", "A2"] + + +def test_candidate_dimensions_come_from_the_fact_table_and_its_dimensions(): + block = build({"A1": entry("A1", fake_result(inputs=data()))}) + # 地区拆过了；会员等级没有码值、客户编号是标识、手机号遮罩、金额是流量、周取值太多：都不是候选 + assert block["not_decomposed"] == ["渠道", "订单状态", "门店类型", "城市"] and block["note"] is None + assert block["tests"] == 0 and block["corrected"] is None + + +def test_a_dimension_that_does_not_match_the_catalog_is_reported_not_guessed(): + result = fake_result(inputs=data(), params={"by": ["地区名"], "label": "销售额"}) + result["table"]["columns"][0] = "地区名" + result["table"]["labels"] = {"地区名": "地区名"} + block = build({"A1": entry("A1", result)}) + assert block["not_decomposed"] is None and block["note"] == "候选维度核对不了：拆解按的「地区名」对不上数据目录里的列" + + +@pytest.mark.parametrize("kwargs, note", [ + ({"sources": None}, scope.NOTE_NO_CATALOG), + ({"sources": {"shop": {"tables": {}, "masked": []}}}, scope.NOTE_NO_NOTES), + ({"sources": {}}, scope.NOTE_NO_CATALOG), + ({"snapshots": {}}, scope.NOTE_NO_QUERY), + ({"sources": {"shop": {"tables": {"orders": {"columns": NOTES["orders"]["columns"]}, + "stores": NOTES["stores"]}, "masked": []}}}, scope.NOTE_NO_FACT), +]) +def test_what_cannot_be_checked_is_said_plainly(kwargs, note): + block = build({"A1": entry("A1", fake_result(inputs=data()))}, **kwargs) + assert block["not_decomposed"] is None and block["note"] == note + assert [i["alias"] for i in block["decomposed"]] == ["A1"], "核对不了的只是候选维度，别的照写" + + +def test_a_single_unlabelled_table_counts_as_the_fact_table(): + notes = {"orders": {k: v for k, v in NOTES["orders"].items() if k != "kind"}} + sql = "SELECT region, SUM(amount) AS gmv FROM orders GROUP BY region" + block = build({"A1": entry("A1", fake_result(inputs=data()))}, sources={"shop": {"tables": notes, "masked": []}}, + snapshots={SNAP: {**SNAPSHOTS[SNAP], "sql": sql}}) + assert block["not_decomposed"] == ["渠道", "订单状态", "手机号"] + + +def test_without_a_dimension_decomposition_the_candidates_are_not_checked(): + block = build({"A1": entry("A1", fake_result("baseline", None, "within_band", inputs=data()))}) + assert block["not_decomposed"] is None and block["note"] == scope.NOTE_NO_DECOMPOSE + + +@pytest.fixture +async def shop(tmp_path): + path = tmp_path / "scope_shop.db" + db = sqlite3.connect(path) + db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, region TEXT, channel TEXT, store_id INTEGER, " + "phone TEXT, amount REAL, week TEXT);" + "CREATE TABLE stores (id INTEGER PRIMARY KEY, store_type TEXT, city TEXT);") + db.commit() + db.close() + async with SessionLocal() as session: + row = DataSource(name=f"scope_{uuid.uuid4().hex[:6]}", kind="sqlite", database=str(path), + options={"mask_columns": "phone"}) + session.add(row) + await session.commit() + row.schema_cache = await introspect(row) + await session.commit() + await session.refresh(row) + for table, notes in NOTES.items(): + await data_catalog.write_entry(session, row.id, table, notes, if_version=0, actor=None) + session.expunge(row) + yield row + await drop_source(row.id) + + +async def test_the_report_reads_the_catalog_through_the_run_view(shop): + facts = await scope.source_facts([shop.name, "nope"]) + assert set(facts) == {shop.name} and facts[shop.name]["masked"] == ["phone"] + assert set(facts[shop.name]["tables"]) == {"orders", "stores"} + assert await scope.source_facts([shop.name], CatalogView(mode="off")) is None + + snapshot = {**SNAPSHOTS[SNAP], "source": shop.name} + catalog = {"Q1": {"alias": "Q1", "kind": "query", "artifact": SNAP, "source": shop.name}, + "A1": entry("A1", fake_result(inputs=data()))} + block = await scope.scope_of(catalog, None, load=lambda a: snapshot) + assert block["not_decomposed"] == ["渠道", "订单状态", "门店类型", "城市"] + block = await scope.scope_of(catalog, {"mode": "off"}, load=lambda a: snapshot) + assert block["not_decomposed"] is None and block["note"] == scope.NOTE_NO_CATALOG