diff --git a/backend/app/api/evidence.py b/backend/app/api/evidence.py index 67e5a199..741dab46 100644 --- a/backend/app/api/evidence.py +++ b/backend/app/api/evidence.py @@ -104,6 +104,7 @@ from app.db.base import SessionLocal from app.db.models import DataSource, Run, RunEvent, Workflow from app.engine import analysis_claims, direct_select, recompute from app.engine.evidence import ( + CELL_DISPLAY, ENTITY_KINDS, GUESS_NOTE, GUESS_SCHEMA, @@ -116,6 +117,7 @@ from app.engine.evidence import ( analysis_recheck, build_catalog, closest_entities, + doc_cell_display, find_segment, guess_sources, input_eid, @@ -126,6 +128,7 @@ from app.engine.evidence import ( query_entry_fields, render_cell, render_metric, + resolve_ref, schema_column, schema_table, sql_tables, @@ -221,6 +224,9 @@ class _Sealed: #: 节点 id → node.started 上记的名字、类型。子工作流里的节点(「子工作流节点 id/…」)不在这次运行的图上,名字从这里取 labels: dict[str, str] = field(default_factory=dict) types: dict[str, str] = field(default_factory=dict) + #: 这一次请求里复算过的分析结论:结论工件 id → 复算的结论(_analysis_recheck)。导出页里同一份结论被判定语、具名的数、 + #: 整表的每一格、分析范围各引一处,各接一条链,复算只做一次。只活在这一次请求里:下一次请求重新取、重新验 + rechecks: dict[str, dict[str, Any]] = field(default_factory=dict) @property def sealed(self) -> bool: @@ -1000,11 +1006,16 @@ _OP_SETTINGS: dict[str, tuple[str, ...]] = { "differ": ("differ_alpha", "min_n", "min_expected"), "comparable": (), } -#: 输入的角色(analysis.contract.AnalysisInput.role)给人看的叫法 +#: 输入的角色(analysis.contract.AnalysisInput.role)给人看的叫法;比率拆解的 headline 是整体比率,不是总数 _INPUT_ROLES = {"data": "数据", "base": "基期", "target": "本期", "headline_base": "基期总数", "headline_target": "本期总数", "parent": "上一层拆解"} -#: 闭合核对:headline 的那一格对着结论里的哪个总数(细案 01 §2.1) -_HEADLINE_TOTALS = {"headline_base": "base_total", "headline_target": "target_total"} +_RATIO_ROLES = {"headline_base": "基期整体比率", "headline_target": "本期整体比率"} +#: 闭合核对:headline 的那一格对着结论里的哪个数(细案 01 §2.1、§2.2)。可加拆解对各段之和(base_total / target_total), +#: 比率拆解对整体比率,即各段分子之和 ÷ 分母之和(r_base_total / r_target_total) +_HEADLINE_TOTALS = { + "additive": {"headline_base": "base_total", "headline_target": "target_total"}, + "ratio": {"headline_base": "r_base_total", "headline_target": "r_target_total"}, +} #: 计算一步最多给结果表的几行(前 table_top_k 段,加上被引用的那一行前后各 WINDOW 行) MAX_ANALYSIS_ROWS = MAX_WINDOW_ROWS @@ -1041,17 +1052,22 @@ def _analysis_cited(cite: dict[str, Any], result: dict[str, Any] | None) -> dict return {"what": "phrase"} -def _analysis_rendered(cite: dict[str, Any], result: dict[str, Any]) -> str | None: - """按封存的结论重新渲染这一处引用(判定语按文本格、具名标量和结果表的格按单元格的规则)。渲染不出来是 None。""" - try: - cited = _analysis_cited(cite, result) - if cited["what"] == "phrase": - return render_cell(result.get("phrase"), kind="text") if isinstance(result.get("phrase"), str) else None - if cited["what"] == "value": - return render_cell(cited["raw"], cite.get("conv")) - return render_cell(cited["raw"], cite.get("conv"), plain=cite.get("plain") is True) - except RenderError: - return None +def _analysis_rendered(seg: dict[str, Any], cite: dict[str, Any], entry: dict[str, Any], result: dict[str, Any], + doc: dict[str, Any]) -> str | None: + """按封存的结论把这一处引用重新解析一遍,交回渲染出来的字(渲染不出来是 None)。和组装、出口复核同一个函数 + (evidence.resolve_ref,按文档记下的那一版单元格显示规则):判定语按文本格,具名标量按单元格的规则,结果表的格照结果表 + 记的列类型(文本码值 "1203" 不加千分位)、显示规则(ISO 日期时间的 T 换成空格)、标识列不分千位。片段上没记引用的 + (口径卡指标上的可比性、分析范围里的编号)按判定语渲染。""" + ref = seg.get("ref") + if not isinstance(ref, str) or not ref.strip(): + phrase = result.get("phrase") + try: + return render_cell(phrase, kind="text") if isinstance(phrase, str) else None + except RenderError: + return None + alias = str(cite.get("alias") or entry.get("alias") or "") + again = resolve_ref(ref, {alias: {**entry, "result": result}}, display=doc_cell_display(doc) or CELL_DISPLAY) + return again.get("rendered") if again.get("status") == "resolved" else None async def _analysis_chain(seg: dict[str, Any], cite: dict[str, Any], entry: dict[str, Any], doc: dict[str, Any], @@ -1082,7 +1098,7 @@ async def _analysis_chain(seg: dict[str, Any], cite: dict[str, Any], entry: dict in_ledger = sealed.trusted and (str(entry.get("node_id") or ""), artifact) in sealed.ledger \ if chained is None else sealed.trusted and chained eid = make_eid(LEDGER_KIND, artifact or None, {}) - rendered = _analysis_rendered(cite, result) if result is not None else None + rendered = _analysis_rendered(seg, cite, entry, result, doc) if result is not None else None judged: dict[str, Any] = { "step": "analysis", "alias": alias, "op": entry.get("op"), "mode": entry.get("mode"), "method_label": method_label(entry.get("op"), entry.get("mode")), @@ -1122,7 +1138,7 @@ async def _analysis_chain(seg: dict[str, Any], cite: dict[str, Any], entry: dict inputs, steps = await _analysis_inputs(alias, result, catalog, doc, sealed, masks, queries=queries) hidden = {str(c).lower() for q in steps for c in q.get("masked") or []} calc = _analysis_calc(alias, result, judged["cited"], hidden) - again = await _analysis_recheck(result, sealed) + again = await _analysis_recheck(result, sealed, artifact) recheck = {"step": "analysis_recompute", "alias": alias, **again} return [judged, method, calc, *inputs, *steps, recheck, seal] @@ -1170,11 +1186,15 @@ async def _analysis_inputs(alias: str, result: dict[str, Any], catalog: dict[str sealed: _Sealed, masks: _Masks, *, queries: bool = True ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: """输入:每份一步(analysis_input),再给每份输入快照一个查询步骤(_query_steps,高亮用到的列;整份用到的给前 - MAX_WINDOW_ROWS 行;queries 为假时不给)。headline 的那一格另核对和结论里的总数闭不闭合(照实写判定是不是 - unclosed,不在这里重算容差);下钻的上一层落到报告目录的哪个 A。""" + MAX_WINDOW_ROWS 行;queries 为假时不给)。headline 的那一格另核对和结论里的数闭不闭合(_HEADLINE_TOTALS:可加拆解是 + 各段之和,比率拆解是整体比率;closure.mode 写明是哪一种,照实写判定是不是 unclosed,不在这里重算容差);下钻的上一层 + 落到报告目录的哪个 A。""" values = result.get("values") if isinstance(result.get("values"), dict) else {} verdict = result.get("verdict") if isinstance(result.get("verdict"), dict) else {} unclosed = verdict.get("code") == "unclosed" + mode = str(result.get("mode") or "") + totals = _HEADLINE_TOTALS.get(mode, {}) if result.get("op") == "decompose" else {} + roles = {**_INPUT_ROLES, **(_RATIO_ROLES if mode == "ratio" else {})} steps: list[dict[str, Any]] = [] cells: dict[str, list[tuple[Any, Any]]] = {} for item in result.get("inputs") or []: @@ -1184,7 +1204,7 @@ async def _analysis_inputs(alias: str, result: dict[str, Any], catalog: dict[str role = str(item.get("role") or "") step: dict[str, Any] = { "step": "analysis_input", "of": alias, "ref": item.get("ref"), "role": role, - "role_label": _INPUT_ROLES.get(role, role or "输入"), "artifact": artifact or None, + "role_label": roles.get(role, role or "输入"), "artifact": artifact or None, **{k: item[k] for k in ("columns", "rows", "row", "column", "value") if k in item}, } if role == "parent": @@ -1197,10 +1217,9 @@ async def _analysis_inputs(alias: str, result: dict[str, Any], catalog: dict[str query = sealed.queries.get(artifact) step.update(query=_catalog_alias_of(catalog, artifact), node_id=(query or {}).get("node_id"), sealed=sealed.trusted and query is not None) - if role in _HEADLINE_TOTALS and _HEADLINE_TOTALS[role] in values: - total = values[_HEADLINE_TOTALS[role]] + if role in totals and totals[role] in values: with contextlib.suppress(RenderError): - step["closure"] = {"total_key": _HEADLINE_TOTALS[role], "total": render_cell(total), + step["closure"] = {"mode": mode, "total_key": totals[role], "total": render_cell(values[totals[role]]), "headline": render_cell(item.get("value")), "closed": not unclosed} steps.append(step) if artifact and query is not None: @@ -1222,10 +1241,21 @@ async def _analysis_inputs(alias: str, result: dict[str, Any], catalog: dict[str return steps, found -async def _analysis_recheck(result: dict[str, Any], sealed: _Sealed) -> dict[str, Any]: +async def _analysis_recheck(result: dict[str, Any], sealed: _Sealed, artifact: str | None = None) -> dict[str, Any]: """复算(evidence.analysis_recheck):输入快照交工件原内容,下钻的上一层交那份结论,都按 sha256 取回、复验哈希,在线程里 复算、比较。只认封存范围内的——输入快照要是这次运行交回过的查询(sealed.queries),上一层要是台账记过的分析结论; - 不在范围内、哈希对不上的一律当取不回(复算照实说取不回哪一份)。""" + 不在范围内、哈希对不上的一律当取不回(复算照实说取不回哪一份)。 + + artifact 是这份结论的工件 id(内容寻址):一次请求里同一份结论只复算一次,结果记在 sealed.rechecks。""" + if artifact and artifact in sealed.rechecks: + return dict(sealed.rechecks[artifact]) + again = await _recheck_now(result, sealed) + if artifact: + sealed.rechecks[artifact] = again + return dict(again) + + +async def _recheck_now(result: dict[str, Any], sealed: _Sealed) -> dict[str, Any]: snapshots = {str(i["artifact"]) for i in result.get("inputs") or [] if isinstance(i, dict) and i.get("artifact") and i.get("role") != "parent"} parents = {str(i["artifact"]) for i in result.get("inputs") or [] @@ -3260,22 +3290,24 @@ def scope_aliases(doc: dict[str, Any]) -> list[str]: return list(dict.fromkeys(out)) -async def _scope_chains(doc: dict[str, Any], sealed: _Sealed, report_node: str) -> dict[str, list[dict[str, Any]]]: - """分析范围里每个编号的出处链(和点开正文里的判定语同一条 _chain,按那条结论的判定语认):{编号: 链}。目录里没有的 - 编号不给(文档里写着、目录里却没有,只能是文档有问题,范围那一项点不开)。""" +def _scope_cite_key(alias: str) -> tuple: + """分析范围里一个编号的引用键:和正文里引用判定语([[a:A1]])的片段同一个键(_export_cite_key),共用一条链。""" + return (LEDGER_KIND, alias, "resolved", None, None, None, None, None, None) + + +async def _scope_chain(alias: str, doc: dict[str, Any], sealed: _Sealed, report_node: str) -> list[dict[str, Any]] | None: + """分析范围里一个编号的出处链(和点开正文里的判定语同一条 _chain,按那条结论的判定语认)。目录里没有的编号为 None + (文档里写着、目录里却没有,只能是文档有问题,范围那一项点不开)。""" catalog = doc.get("catalog") if isinstance(doc.get("catalog"), dict) else {} - out: dict[str, list[dict[str, Any]]] = {} - for alias in scope_aliases(doc): - entry = catalog.get(alias) - if not isinstance(entry, dict) or entry.get("kind") != LEDGER_KIND: - continue - result, _ = _analysis_of(str(entry.get("artifact") or "")) - seg = {"id": SCOPE_REF.format(alias=alias), "kind": LEDGER_KIND, "state": "deterministic", - "text": render_cell(result.get("phrase"), kind="text") if result and result.get("phrase") else alias, - "cite": {"alias": alias, "kind": LEDGER_KIND, "status": "resolved", "locator": {}, - "eid": entry.get("eid")}} - out[alias] = await _chain(seg, doc, sealed, report_node) - return out + entry = catalog.get(alias) + if not isinstance(entry, dict) or entry.get("kind") != LEDGER_KIND: + return None + result, _ = _analysis_of(str(entry.get("artifact") or "")) + seg = {"id": SCOPE_REF.format(alias=alias), "kind": LEDGER_KIND, "state": "deterministic", + "text": render_cell(result.get("phrase"), kind="text") if result and result.get("phrase") else alias, + "cite": {"alias": alias, "kind": LEDGER_KIND, "status": "resolved", "locator": {}, + "eid": entry.get("eid")}} + return await _chain(seg, doc, sealed, report_node) async def _export_withheld(sealed: _Sealed, doc: dict[str, Any]) -> frozenset[str]: @@ -3383,14 +3415,17 @@ async def export_report_html(run_id: str, report: str | None = None) -> Response refs[sid] = str(ck) if key is not None: memo[key] = str(ck) - # 系统写的「分析范围」:每个编号点得开,链和点开正文里的判定语是同一条(引用键相同的共用) - for alias, chain in (await _scope_chains(doc, sealed, chosen.node_id)).items(): - sid = SCOPE_REF.format(alias=alias) - ck = (LEDGER_KIND, alias, "resolved", None, None, None, None, None, None) + # 系统写的「分析范围」:每个编号点得开,链和点开正文里的判定语是同一条(引用键相同的共用:正文已经引了判定语的, + # 不再建一遍) + for alias in scope_aliases(doc): + ck = _scope_cite_key(alias) if ck not in memo: + chain = await _scope_chain(alias, doc, sealed, chosen.node_id) + if chain is None: + continue memo[ck] = str(ck) chains[str(ck)] = chain - refs[sid] = memo[ck] + refs[SCOPE_REF.format(alias=alias)] = memo[ck] async with SessionLocal() as session: run = await session.get(Run, run_id) if run is None: # 两次读之间被删掉了 @@ -4013,18 +4048,22 @@ def _scope_catalog(sealed: _Sealed, report_node: str) -> dict[str, Any]: doc.catalog 只留了正文用到的实体,给「是不是想写…」找候选不够;运行状态(checkpoint)不在封存 范围里,也不能用。图读不出来、找不到报告节点时退回全部封存台账。 + + 台账里的分析结论不进:候选只看表和字段,用不到结论的内容,进了目录就要把每份结论读盘、复验一遍哈希(build_catalog + 给 A 条目装整份结论)。查询、表和字段的编号不受影响(分析结论另编 A)。 """ from app.engine.nodes.report import report_catalog from app.engine.schema import GraphSpec + entries = [e for e in sealed.entries if e.get("kind") != LEDGER_KIND] try: spec = GraphSpec.model_validate(sealed.graph) node = spec.node_map().get(report_node) except Exception: # noqa: BLE001 - 老运行的图可能已经不合现在的格式,退回不分范围 node = None if node is None: - return build_catalog(nodes={}, ledger=sealed.entries) - return report_catalog({"evidence": sealed.entries, "nodes": {}, "input": {}}, spec, node) # type: ignore[arg-type] + return build_catalog(nodes={}, ledger=entries) + return report_catalog({"evidence": entries, "nodes": {}, "input": {}}, spec, node) # type: ignore[arg-type] def _suspicious_step(seg: dict[str, Any], cite: dict[str, Any], sealed: _Sealed, report_node: str) -> dict[str, Any]: diff --git a/backend/app/bundle/build.py b/backend/app/bundle/build.py index 259e20f1..ac8a10a9 100644 --- a/backend/app/bundle/build.py +++ b/backend/app/bundle/build.py @@ -63,7 +63,7 @@ from pathlib import Path from typing import Any from urllib.parse import quote -from app.analysis.contract import ARTIFACT_KIND, LEDGER_KIND, NODE_TYPE as ANALYSIS_NODE, is_result +from app.analysis.contract import ARTIFACT_KIND, LEDGER_KIND, NODE_TYPE as ANALYSIS_NODE, is_result, method_label from app.core.artifact_store import canonical_json, content_hash from app.core.events import EventType from app.engine.evidence import analysis_recheck, iter_units, query_entry_fields @@ -1161,8 +1161,9 @@ class _Artifacts: def _attach_analysis(artifacts: _Artifacts, store: _Store, item: dict[str, Any] | None, artifact: str, owner: str | None, cited_queries: list[str], depth: int = 0) -> None: """随文件附带的分析结论:它的输入快照(种类 query,遮罩照常认)、下钻的上一层结论(同样附带它的输入,最多 - _PARENT_DEPTH 层)一并列进清单,再把服务端的复算结论记在这份结论的条目上(recheck)。只给 id 的结论、被改过的 - (哈希对不上)不顺着它去附带别的:被改过的结论说输入是哪几份,不能替它圆。""" + _PARENT_DEPTH 层)一并列进清单,再把服务端的复算结论记在这份结论的条目上(recheck),连同方法的叫法(method_label, + contract.method_label,和导出页同一份:核对页只显示,不在页面里另起一套叫法)。只给 id 的结论、被改过的(哈希对不上) + 不顺着它去附带别的:被改过的结论说输入是哪几份,不能替它圆。""" if item is None or "b64" not in item or not store.intact(artifact) or "recheck" in item: return result = store.content(artifact) @@ -1176,6 +1177,7 @@ def _attach_analysis(artifacts: _Artifacts, store: _Store, item: dict[str, Any] continue artifacts.add(input_id, "query", owner) cited_queries.append(input_id) + item["method_label"] = method_label(result.get("op"), result.get("mode")) item["recheck"] = _recheck(result, store) diff --git a/backend/app/bundle/verifier.js b/backend/app/bundle/verifier.js index 90309994..767828fe 100644 --- a/backend/app/bundle/verifier.js +++ b/backend/app/bundle/verifier.js @@ -1983,6 +1983,14 @@ return parts } + /** 方法与版本那一行:服务端在工件条目上带的叫法(method_label,和导出页同一份「变化拆解(可加) · decompose.additive@1」) + * 加方法 id;条目上没带叫法的只写 id */ + function methodText(art, result) { + const label = art && isObj(art.a) && typeof art.a.method_label === 'string' ? art.a.method_label : '' + const id = String(result.method || '—') + return label ? `${label} · ${id}` : id + } + /** 口径卡指标上的可比性(口径卡执行时自动做的可比性检查):卡上记的判定和原因;那份结论随文件附带、核对通过的,再给 * 判定标准、方法与版本、各项检查和导出时服务端的复算。没附带的照卡上的记录写,说明为什么没有更多 */ function comparabilityBlock(c) { @@ -2002,7 +2010,7 @@ } parts.push(kv([ result.criterion ? [T.dCriterion, String(result.criterion)] : null, - [T.dMethod, String(result.method || '—')], + [T.dMethod, methodText(art, result)], ])) const table = isObj(result.table) ? result.table : null const columns = table && Array.isArray(table.columns) ? table.columns.map(String) : [] @@ -2030,7 +2038,7 @@ parts.push(kv([ [T.dPhrase, String(result.phrase || '—')], result.criterion ? [T.dCriterion, String(result.criterion)] : null, - [T.dMethod, String(result.method || '—')], + [T.dMethod, methodText(art, result)], ])) const caveats = (Array.isArray(result.caveats) ? result.caveats : []).filter(isObj).map((c) => String(c.text || '')) if (caveats.length) parts.push(h('p', { class: 'sub' }, T.dCaveats), h('ul', { class: 'items' }, caveats.map((c) => h('li', {}, c)))) diff --git a/backend/app/engine/compare.py b/backend/app/engine/compare.py index 61b7ed80..6afdab43 100644 --- a/backend/app/engine/compare.py +++ b/backend/app/engine/compare.py @@ -336,17 +336,22 @@ def _queries(side: Side) -> list[_Query]: def _analysis_items(side: Side) -> dict[str, tuple[dict[str, dict[str, Any]], int]]: - """封存范围内每张分析卡各项最后一次的结论(analysis.end,engine/nodes/analysis.py 每项一条):{卡: ({项: 事件数据}, - 执行了几次)},项按记下的先后。循环里执行了几次的,同一项后一次的盖过前一次。""" + """封存范围内每张分析卡各项最后一次做完的结论(analysis.end,engine/nodes/analysis.py 每项一条):{卡: ({项: 事件数据}, + 执行了几次)},项按记下的先后。只认后面跟着这张卡 node.finished 的那几次执行:执行到一半失败的(前几项已经发了 + analysis.end,后面一项让卡失败),那次的判定不是这次运行的结果,不列,也不算执行次数。循环里执行了几次的,同一项后 + 一次的盖过前一次。""" out: dict[str, tuple[dict[str, dict[str, Any]], int]] = {} + pending: dict[str, dict[str, dict[str, Any]]] = {} # 卡 → 这一次执行里已经做完的项(还没见到 node.finished) for _, etype, node_id, data in side.events: if not node_id or not isinstance(data, dict): continue - if etype == EventType.ANALYSIS_END and isinstance(data.get("item"), str): + if etype == EventType.NODE_STARTED: + pending.pop(node_id, None) + elif etype == EventType.ANALYSIS_END and isinstance(data.get("item"), str): + pending.setdefault(node_id, {})[data["item"]] = data + elif etype == EventType.NODE_FINISHED and node_id in pending: items, runs = out.setdefault(node_id, ({}, 0)) - items[data["item"]] = data - elif etype == EventType.NODE_FINISHED and node_id in out: - items, runs = out[node_id] + items.update(pending.pop(node_id)) out[node_id] = (items, runs + 1) return out diff --git a/backend/app/engine/evidence.py b/backend/app/engine/evidence.py index c77667e0..3cdbcec6 100644 --- a/backend/app/engine/evidence.py +++ b/backend/app/engine/evidence.py @@ -2106,8 +2106,11 @@ def resolve_support(ref: str, catalog: dict[str, Any], *, loader: Loader | _Snap def resolve_ref(ref: str, catalog: dict[str, Any], *, cells_allowed: bool = True, - loader: Loader | _Snapshots | None = None) -> dict[str, Any]: - """片段上记的 ref(m:gmv|万、i:week、v:Q1.r0.amount)→ Citation。证据接口点开片段时用它重新解析。""" + loader: Loader | _Snapshots | None = None, display: int | None = None) -> dict[str, Any]: + """片段上记的 ref(m:gmv|万、i:week、v:Q1.r0.amount)→ Citation。证据接口点开片段时用它重新解析。display 是按哪一版 + 单元格显示规则渲染(文档记下的那一版,doc_cell_display;复核按它),不给按 loader 的(缺省当前版)。""" + if display is not None: + loader = _snapshots(loader, display=display) return resolve_marker(_parse_ref(ref), catalog, cells_allowed=cells_allowed, loader=loader) @@ -4057,11 +4060,18 @@ def _analysis_sources(entry: dict[str, Any], catalog: dict[str, Any]) -> str: f",另有 {missing} 份输入不在这份目录里" if missing else ""]) +#: 写作目录里一条分析结论的几种样子,从全到简(_analysis_prompt 的 level):full 连同结果表的前 table_top_k 段;compact 不列 +#: 各段,结果表只写一句;brief 只留判定语和具名的数;head 只列编号、方法和来自哪几个查询 +ANALYSIS_LEVELS = ("full", "compact", "brief", "head") + + def _analysis_prompt(entry: dict[str, Any], catalog: dict[str, Any], snaps: _Snapshots, cells_allowed: bool, *, - table: bool = True) -> list[str]: + level: str = "full") -> list[str]: """一条分析结论在写作目录里的样子:编号、方法、来自哪几个查询,判定语([[a:A1]] 渲染出来的那一句)、判定标准、具名的数 - (可引用的键,值按渲染规则写)、附带说明、输入查询的 SQL 检查问题、算不了的原因;table 为真时再列结果表的前 - table_top_k 段,其余写「其余 n 段」。agent 写的假设不列:它不是结论,只在证据面板显示。 + (可引用的键,值按渲染规则写)、附带说明、输入查询的 SQL 检查问题、算不了的原因;level 为 full 时再列结果表的前 + table_top_k 段,其余写「其余 n 段」。目录放不下时往简里降(ANALYSIS_LEVELS):compact 结果表只写一句;brief 只留 + 判定语和具名的数(判定标准、附带说明、SQL 检查的提醒、算不了的原因那句不列,判定语里本来就写着);head 只列编号那一行。 + agent 写的假设不列:它不是结论,只在证据面板显示。 受管级别的正式出具、契约没开「单元格引用」时,Agent 现场做的分析只列判定语和判定标准,写明只能当依据([[see:A1]]), 不列数和表:列出来的数写作者只能照抄成没有出处的数。""" @@ -4070,8 +4080,11 @@ def _analysis_prompt(entry: dict[str, Any], catalog: dict[str, Any], snaps: _Sna result, why = snaps.analysis(alias, entry) if result is None: return [f"{head}({why},不要引用)"] + if level == "head": + return [head] + brief = level == "brief" lines = [head] - if note := _sql_check_note(entry.get("artifact"), snaps, "这次分析的输入查询"): + if not brief and (note := _sql_check_note(entry.get("artifact"), snaps, "这次分析的输入查询")): lines.append(note) phrase = _clean(str(result.get("phrase") or "")) see_only = not cells_allowed and not analysis_card(entry) @@ -4080,13 +4093,15 @@ def _analysis_prompt(entry: dict[str, Any], catalog: dict[str, Any], snaps: _Sna f"[[see:{alias}]] 作依据)") else: lines.append(f" 判定语 [[a:{alias}]] = {phrase}" if phrase else f" (没有判定语,不要写 [[a:{alias}]])") - if result.get("status") == "refused": + if brief: + pass + elif result.get("status") == "refused": lines.append(f" 算不了:{_clean(str(result.get('reason') or '没有写原因'))}。这条结论不能当「异常」「主要来自」" "「呈上升趋势」这类判断的依据,只能照实写算不了的原因") elif criterion := _clean(str(result.get("criterion") or "")): lines.append(f" 判定标准:{criterion}") caveats = [_clean(str(c.get("text") or "")) for c in result.get("caveats") or [] if isinstance(c, dict)] - if caveats := [c for c in caveats if c]: + if not brief and (caveats := [c for c in caveats if c]): lines.append(f" 附带说明:{';'.join(caveats)}") if see_only: return lines @@ -4100,6 +4115,8 @@ def _analysis_prompt(entry: dict[str, Any], catalog: dict[str, Any], snaps: _Sna if shown: more = f",另有 {len(values) - _PROMPT_VALUES} 个" if len(values) > _PROMPT_VALUES else "" lines.append(f" 具名的数(写 [[a:{alias}.键]]):{','.join(shown)}{more}") + if brief: + return lines snapshot, why = snaps.table(alias, entry) if why is not None: return lines @@ -4107,7 +4124,7 @@ def _analysis_prompt(entry: dict[str, Any], catalog: dict[str, Any], snaps: _Sna labels = snapshot.get("labels") or {} rows = snapshot["rows"] named = "、".join(f"{c}({_clean(str(labels[c]))})" if labels.get(c) else c for c in columns) - if not table: + if level != "full": lines.append(f" 结果表共 {len(rows)} 段,列 {named}(这里没列出各段;要列出就单独一行写 [[table:{alias}]])") return lines settings = result.get("settings") if isinstance(result.get("settings"), dict) else {} @@ -4135,12 +4152,13 @@ _TOP_K = 10 @dataclass class _AnalysisLines: - """写作目录里「分析结论」一段:每条结论两种样子——full 连同结果表的前 table_top_k 段,compact 只有判定、判定标准、 - 具名的数和说明。放得下时全按 full;放不下时先按 compact 列出,再按顺序补上结果表,排在查询的行前面(_trimmed_catalog)。""" + """写作目录里「分析结论」一段:每条结论四种样子(ANALYSIS_LEVELS,从全到简)。放得下时全按 full;放不下时这一段最多占 + 预算的 1/_PROMPT_ANALYSIS_SHARE:先按 compact 列出,还超就从后往前降成 brief、再降成 head,然后按顺序补上结果表,排在 + 查询的行前面(_trimmed_catalog)。""" order: list[str] - full: dict[str, list[str]] - compact: dict[str, list[str]] + #: 样子 → {编号: 那几行} + levels: dict[str, dict[str, list[str]]] #: 有 Agent 现场做的分析只能当依据(受管级别的出具、契约没开「单元格引用」) see_only: bool = False @@ -4148,15 +4166,19 @@ class _AnalysisLines: def of(cls, catalog: dict[str, Any], snaps: _Snapshots, cells_allowed: bool) -> "_AnalysisLines": entries = [e for e in catalog.values() if isinstance(e, dict) and e.get("kind") == LEDGER_KIND] return cls([e["alias"] for e in entries], - {e["alias"]: _analysis_prompt(e, catalog, snaps, cells_allowed) for e in entries}, - {e["alias"]: _analysis_prompt(e, catalog, snaps, cells_allowed, table=False) for e in entries}, + {level: {e["alias"]: _analysis_prompt(e, catalog, snaps, cells_allowed, level=level) + for e in entries} for level in ANALYSIS_LEVELS}, see_only=not cells_allowed and any(not analysis_card(e) for e in entries)) + def shape(self, alias: str, level: str) -> list[str]: + return self.levels[level][alias] + def lines(self, tier: dict[str, str] | None = None) -> list[str]: - """这一段的文字:tier 里记成 compact 的按 compact,其余按 full。没有分析结论时是空的(目录和以前一字不差)。""" + """这一段的文字:每条按 tier 里记的样子,没记的按 full。没有分析结论时是空的(目录和以前一字不差)。""" if not self.order: return [] - shapes = [self.compact[a] if (tier or {}).get(a) == "compact" else self.full[a] for a in self.order] + shapes = [self.shape(a, level if (level := (tier or {}).get(a)) in ANALYSIS_LEVELS else "full") + for a in self.order] head = [f"\n{ANALYSIS_PROMPT}", *([ANALYSIS_SEE_ONLY_PROMPT] if self.see_only else [])] return [*head, *(line for shape in shapes for line in shape)] @@ -4176,6 +4198,8 @@ CATALOG_MAX_CHARS = 160_000 _PROMPT_COMPACT_ROWS = 3 #: 放不下时表和字段、知识库检索两节合起来最多占预算的几分之一(剩下的给查询) _PROMPT_TAIL_SHARE = 4 +#: 放不下时分析结论一段最多占预算的几分之一:几十份结论、每份上千段时,不能把查询挤得只剩一行摘要 +_PROMPT_ANALYSIS_SHARE = 2 _PROMPT_CUT = "\n…(目录太长,后面的省略了)" @@ -4451,16 +4475,17 @@ def _trimmed_catalog(catalog: dict[str, Any], head: list[str], first: list[str], pinned: list[str], analyses: _AnalysisLines | None = None) -> WriterCatalog: """放不下时的目录。以前从头截断:查询按编号排,前面的节点占满预算,后面的节点一次都不露面。 - 1. 口径卡指标、运行输入、结构化字段一律列出(第一优先);分析结论的判定、判定标准、具名的数、说明也一律列出 + 1. 口径卡指标、运行输入、结构化字段一律列出(第一优先);分析结论的判定、判定标准、具名的数、说明也列出,只是这一段 + 最多占预算的 1/_PROMPT_ANALYSIS_SHARE:超了就按结论的先后从后往前降,先只留判定语和具名的数,再只列编号 2. 表和字段、知识库检索两节合起来最多占预算的 1/_PROMPT_TAIL_SHARE - 3. 分析结论的结果表(前 table_top_k 段)先补:判定是从原始明细算出来的结论,比原始明细更值得留 + 3. 分析结论的结果表(前 table_top_k 段)先补(不超过那个上限):判定是从原始明细算出来的结论,比原始明细更值得留 4. 每次查询先只占一行摘要;结构化字段引用到的查询先展开(pinned),再按产出它的节点分组、轮流展开: 第一轮每次列前 _PROMPT_COMPACT_ROWS 行,第二轮补到 _PROMPT_ROWS 行,放不下为止 5. 开头写明目录有省略、各来源省略了几次,以及没展开的查询怎么用(OMITTED_RULES) 最后整份量一次:超了就把最后展开的退回去(先退查询,最后才退分析结论的结果表);指标和字段本身就超(极少见) 才从尾部截断。 """ - analyses = analyses or _AnalysisLines([], {}, {}) + analyses = analyses or _AnalysisLines([], {}) if tail and len("\n".join(tail)) > budget // _PROMPT_TAIL_SHARE: tail = _cut_lines(tail, budget // _PROMPT_TAIL_SHARE, "…(表和字段、知识库检索太长,后面的省略了)") tables: dict[str, list[str]] = {} @@ -4493,9 +4518,7 @@ def _trimmed_catalog(catalog: dict[str, Any], head: list[str], first: list[str], notice = (f"注意:证据太多,这份目录有省略(篇幅上限约 {budget} 字)。共 {len(queries)} 次查询:" f"{n['full']} 次列出前 {_PROMPT_ROWS} 行,{n['compact']} 次只列前 {_PROMPT_COMPACT_ROWS} 行," f"{n['omitted']} 次只列编号和摘要。各来源:{per}。") - if bare := sum(1 for alias in analyses.order if tier[alias] == "compact"): - notice += (f"另有 {len(analyses.order)} 条分析结论,判定和具名的数都列出了,其中 {bare} 条没列出结果表的各段" - "(要列出就单独一行写整表引用,系统照结果表生成)。") + notice += _analysis_notice(Counter(tier[alias] for alias in analyses.order), len(analyses.order)) body = [*head, notice, OMITTED_RULES, *first, *analyses.lines(tier), *rules] for r, entries in zip(stats, groups.values()): body.append(f"\n来自「{r['title']}」的查询(共 {r['total']} 次,展开 {r['full'] + r['compact']} 次):") @@ -4510,14 +4533,24 @@ def _trimmed_catalog(catalog: dict[str, Any], head: list[str], first: list[str], def size(lines: list[str]) -> int: return sum(len(line) + 1 for line in lines) + # 分析结论一段最多占预算的 1/_PROMPT_ANALYSIS_SHARE:超了按结论的先后从后往前降,先只留判定语和具名的数,再只列编号 + cap = budget // _PROMPT_ANALYSIS_SHARE + for level in ("brief", "head"): + for alias in reversed(analyses.order): + if size(analyses.lines(tier)) <= cap: + break + tier[alias] = level # 省略情况那句里的数字会变长几位:留一点余量,最后再整份量一次 remaining = budget - len(assemble()) - 8 * (len(groups) + 3) upgraded: list[tuple[str, str]] = [] + listed = size(analyses.lines(tier)) for alias in analyses.order: - # 分析结论的结果表先补,排在查询的行前面;整份量下来超了,最后才退回它们 - if (delta := size(analyses.full[alias]) - size(analyses.compact[alias])) <= remaining: + # 分析结论的结果表先补,排在查询的行前面(这一段照样不超过上限);整份量下来超了,最后才退回它们 + delta = size(analyses.shape(alias, "full")) - size(analyses.shape(alias, "compact")) + if tier[alias] == "compact" and delta <= remaining and listed + delta <= cap: tier[alias] = "full" remaining -= delta + listed += delta upgraded.append((alias, "compact")) for alias in pinned: if tier.get(alias) == "brief" and (delta := size(compact[alias]) - size(brief[alias])) <= remaining: @@ -4551,6 +4584,29 @@ def _trimmed_catalog(catalog: dict[str, Any], head: list[str], first: list[str], omitted=[e["alias"] for e in queries if tier[e["alias"]] == "brief"], empty_sources=empty) +#: 目录有省略、分析结论只是结果表没列各段时,开头那句里接的说明(给模型看) +ANALYSIS_TABLES_OMITTED_PROMPT = ("另有 {total} 条分析结论,判定和具名的数都列出了,其中 {bare} 条没列出结果表的各段" + "(要列出就单独一行写整表引用,系统照结果表生成)。") +#: 分析结论还降到了只留判定语和具名的数、只列编号时接的说明(给模型看):各样子各有几条(ANALYSIS_LEVEL_PROMPT_WORDS), +#: 以及没列出的怎么引用 +ANALYSIS_TRIMMED_PROMPT = ("另有 {total} 条分析结论,这一段最多占篇幅的一半:{parts}。没列出的也能引用:判定语单独写 " + "[[a:编号]],系统照结论渲染;要列出结果表就单独一行写整表引用,系统照结果表生成;没列出的具名的数" + "不要编造。") +ANALYSIS_LEVEL_PROMPT_WORDS = (("full", "列出了结果表的前几段"), ("compact", "没列出结果表的各段"), + ("brief", "只列了判定语和具名的数(判定标准、附带说明没列)"), ("head", "只列了编号")) + + +def _analysis_notice(levels: Counter, total: int) -> str: + """目录有省略时开头那句里分析结论的情况:都列全了的不说;只是结果表没列各段的照旧一句;还降到了只留判定语和具名的数、 + 只列编号的,照实写各有几条,以及没列出的怎么引用。""" + if not total or levels["full"] == total: + return "" + if not levels["brief"] and not levels["head"]: + return ANALYSIS_TABLES_OMITTED_PROMPT.format(total=total, bare=levels["compact"]) + parts = ",".join(f"{levels[level]} 条{word}" for level, word in ANALYSIS_LEVEL_PROMPT_WORDS if levels[level]) + return ANALYSIS_TRIMMED_PROMPT.format(total=total, parts=parts) + + def catalog_prompt(catalog: dict[str, Any], *, budget: int = DEFAULT_CATALOG_BUDGET, cells_allowed: bool = True, loader: Loader | None = None, fields: list[dict[str, Any]] | None = None, titles: dict[str, str] | None = None) -> str: @@ -4748,7 +4804,11 @@ _DECOMPOSE_NUMBERS = ("base", "target", "delta", "pct", "share", "contrib", "fla def _decompose_row(result: dict[str, Any]) -> tuple[str, list[Any]] | None: """变化拆解的结果表里判定点名的那一段:(这一段的叫法, 那一行)。段按维度列(params.by;乘积模式是 factor 一列)对 - verdict.subjects 的第一项,对不上就取第一行(结果表按变化的绝对值从大到小排)。叫法是维度值用「·」连起来。""" + verdict.subjects 的第一项,对不上就取第一行(结果表按变化的绝对值从大到小排)。维度值和叫法都按算子写段的同一个函数 + (app.analysis.phrases.key_text:空值写「(空)」、布尔写「是 / 否」,两列用「·」连):subjects 就是这么写的,拿原值的 + str 去比,空值、布尔维度的段对不上,代入式会退回第一行。""" + from app.analysis.phrases import key_text + table = result.get("table") if isinstance(result.get("table"), dict) else {} rows = [r for r in table.get("rows") or [] if isinstance(r, list)] columns = [str(c) for c in table.get("columns") or []] @@ -4765,8 +4825,8 @@ def _decompose_row(result: dict[str, Any]) -> tuple[str, list[Any]] | None: picked = rows[0] if subjects and at and isinstance(subjects[0], list): wanted = [str(v) for v in subjects[0]] - picked = next((row for row in rows if [str(row[i]) for i in at if i < len(row)] == wanted), rows[0]) - name = "·".join(str(picked[i]) for i in at if i < len(picked) and picked[i] is not None) + picked = next((row for row in rows if [key_text([row[i]]) for i in at if i < len(row)] == wanted), rows[0]) + name = key_text([picked[i] for i in at if i < len(picked)]) if at else "" return name or "这一段", picked diff --git a/backend/app/engine/nodes/metrics.py b/backend/app/engine/nodes/metrics.py index 77aea26e..20004a67 100644 --- a/backend/app/engine/nodes/metrics.py +++ b/backend/app/engine/nodes/metrics.py @@ -1,6 +1,7 @@ from __future__ import annotations import ast +import asyncio import json import math import re @@ -56,8 +57,9 @@ async def run_metrics(state: GraphState, ctx: NodeContext) -> dict[str, Any]: 取数的那条 SQL 对照数据目录查出 error 级问题的(查询快照的 checks,见 data/sqlcheck.py:一对多关联后重复计算、 存量跨期加总),同样值照算,标 sql_check_failed 并写明原因,出具契约据此降档(issuance.sql_check_gaps)。 - 一个指标的输入落在两份不同的查询快照上、两份 SQL 同形只差字面量(本周对上周)时,自动做一次可比性检查(分析算子 - comparable),结论落成工件、记在指标的 comparability 上(_comparability)。做不成不挡口径卡,照实记一条日志。 + 一个指标的输入落在两份不同的查询快照上、两份 SQL 同形只差字面量、不同的字面量里有日期或期次(本周对上周)时,自动做 + 一次可比性检查(分析算子 comparable),结论落成工件、记在指标的 comparability 上(_comparability)。只差地区、门店号这类 + 字面量的(华东对华南)是两段不是两期,不查。做不成不挡口径卡,照实记一条日志。 """ definitions = ctx.cfg("metrics", []) or [] caliber_raw, caliber_version = ctx.cfg("caliber", "") or ctx.node.title, str(ctx.cfg("caliber_version", "") or "v1") @@ -475,11 +477,15 @@ def _rendered(value: Any, definition: dict[str, Any], display: dict[str, Any]) - # 可比性检查(docs/design/analysis-round/00-总方案.md §3.4 第 3 条) # -------------------------------------------------------------------------- -#: 可比性检查没做成(探查失败、算子出错、还没有算子……)时那条日志的 code +#: 可比性检查没做成(快照读不出来、探查失败、算子出错、算不了……)时那条日志的 code COMPARABILITY_SKIPPED = "metric_comparability_skipped" #: 那条日志的说法 COMPARABILITY_SKIPPED_TEXT = "指标「{name}」用到的两次查询没做成可比性检查:{why}。指标照常计算,两期能不能直接比请自行核对" -_DAY = re.compile(r"^(\d{4})-?(\d{2})-?(\d{2})(?:[ T].*)?$") +#: 期次的几种写法(_period_key,排先后用):月 2026-09、202609,周 2026-W41,季 2026Q3、2026-Q3,年 2026 +_MONTH_TEXT = re.compile(r"^(\d{4})-?(0[1-9]|1[0-2])$") +_WEEK_TEXT = re.compile(r"^(\d{4})-?[Ww](\d{1,2})$") +_QUARTER_TEXT = re.compile(r"^(\d{4})-?[Qq]([1-4])$") +_YEAR_TEXT = re.compile(r"^(?:19|20|21)\d{2}$") class _Comparing: @@ -494,13 +500,15 @@ class _Comparing: async def _comparability(metric_id: str, name: str, inputs: list[dict[str, Any]], ctx: NodeContext, comparing: _Comparing, snapshots: dict[str, Any]) -> dict[str, Any] | None: """指标的可比性(contract.MetricComparability:结论工件、判定、判定语、原因)。只在这种时候做:输入里取自查询快照的格 - (cell() 取的格、按位置取的格、Agent 按出处交来的字段)恰好落在两份不同的快照上,两份 SQL 同形、只差字面量,而且查的是 - 同一个数据源——本周对上周。SQL 结构不同的(退款额 / 销售额)不是同一个量的两期,不查。 + (cell() 取的格、按位置取的格、Agent 按出处交来的字段)恰好落在两份不同的快照上,两份 SQL 同形、只差字面量,查的是同一个 + 数据源,而且不同的字面量里至少有一处两边都是日期或期次(_dated_pairs:时间条件的界、像日期的常量)——本周对上周。SQL 结构不同的 + (退款额 / 销售额)不是同一个量的两期,不查;只差地区、门店号这类字面量的(华东对华南、门店 1203 对 1204)是两段,也 + 不查。日期和别的筛选都不同的照查,算子判不可比(两边除了日期,筛选的还不一样)。 - 哪份是基期:两边第一处不同的字面量是日期(或同样写法的期次)的,早的是基期;认不出的,表达式里先出现的那份算本期 - (「(本周 - 上周) / 上周」的写法)。 + 哪份是基期:不同的日期、期次里第一处两边写法相同、排得出先后的(真正的日期,或同一种写法的月、周、季、年),早的是基期; + 排不出的(按运行当天现算的界 date('now', '-7 days')),表达式里先出现的那份算本期(「(本周 - 上周) / 上周」的写法)。 - 做不成(快照读不出来、探查失败、算子出错、算不了、还没有算子)不挡口径卡:照实记一条日志(COMPARABILITY_SKIPPED), + 做不成(快照读不出来、探查失败、算子出错、算不了)不挡口径卡:照实记一条日志(COMPARABILITY_SKIPPED), 这个指标不写 comparability。同一对快照只查一次。""" from app.engine.lineage import SNAPSHOT_VIAS @@ -526,7 +534,8 @@ async def _comparability(metric_id: str, name: str, inputs: list[dict[str, Any]] class _NotPeriods(Exception): - """两份快照不是同一个量的两期(SQL 结构不同、数据源不同、字面量一样、没有 SQL):不查,也不记日志。""" + """两份快照不是同一个量的两期(SQL 结构不同、数据源不同、字面量一样、不同的字面量里没有日期或期次、没有 SQL):不查, + 也不记日志。""" class _Skipped(Exception): @@ -551,7 +560,8 @@ async def _check_comparable(pair: tuple[str, str], inputs: list[dict[str, Any]], for artifact in pair: if artifact not in snapshots: try: - snapshots[artifact] = artifact_store.load(artifact) + # 快照可能有几 MB:读文件、复验哈希放进线程,不占着事件循环(同 tools/analysis.prepare) + snapshots[artifact] = await asyncio.to_thread(artifact_store.load, artifact) except (ValueError, OSError): snapshots[artifact] = None if not isinstance(snapshots[artifact], dict): @@ -565,7 +575,10 @@ async def _check_comparable(pair: tuple[str, str], inputs: list[dict[str, Any]], norms = [normalize_sql(q, dialect) for q in sqls] if not norms[0].same_shape(norms[1]) or norms[0].literals == norms[1].literals: raise _NotPeriods() - first, second = _period_order(norms[0].literals, norms[1].literals) + dated = _dated_pairs(sqls[0], sqls[1], dialect) + if not dated: + raise _NotPeriods() # 不同的字面量里没有日期或期次:比的是两段,不是两期 + first, second = _period_order(dated) order = (pair[first], pair[second]) # (基期, 本期) refs: dict[str, str] = {} for role, artifact in zip(("base", "target"), order): @@ -608,20 +621,55 @@ async def _dialect_of(source: str, comparing: _Comparing) -> str | None: return comparing.kinds[source] -def _period_order(a: tuple[str, ...], b: tuple[str, ...]) -> tuple[int, int]: - """两份同形 SQL 的字面量 → (基期的下标, 本期的下标)。第一处不同、两边都认得出的日期(YYYY-MM-DD、YYYYMMDD,带不带 - 时间)或同样长短的期次写法(2026-W41、202610)定先后,早的是基期;都认不出的,表达式里先出现的那份(下标 0)是本期。""" - for x, y in zip(a, b): - if x == y: - continue - left, right = x.strip("'\""), y.strip("'\"") - dx, dy = _DAY.match(left), _DAY.match(right) - if dx and dy: - if dx.groups() != dy.groups(): - return (0, 1) if dx.groups() < dy.groups() else (1, 0) - continue - if len(left) == len(right) and re.fullmatch(r"[\dW\-/Qq]+", left) and re.fullmatch(r"[\dW\-/Qq]+", right): - return (0, 1) if left < right else (1, 0) +def _dated_pairs(a: str, b: str, dialect: str | None) -> list[tuple[str, str]]: + """两份同形 SQL 里不同的字面量中,两边都是日期或期次的那几对:[(左边的值, 右边的值)](不带引号),按语法树里的先后。 + 一对都没有(只差地区、门店号这类字面量),或者解析不了、两边的字面量对不上的,为空。 + + 日期、期次的认法和可比性算子「同口径」一项同一套(app/analysis/comparable 的 _undated):data/coverage 认出的时间条件的 + 界(整个界的表达式:day BETWEEN '2026-10-12' AND …、month = '2026-09'、yr = 2026、week = '2026-W41',按运行当天现算的 + date('now', '-7 days') 里的字面量也算),加上别处像日期的常量(coverage.literal_bound 认得出的真正的日期)。两边同形, + 语法树一样,字面量按同一个先后一一对上。""" + from sqlglot import exp + + from app.data import coverage + + sides: list[list[tuple[Any, bool]]] = [] + for sql in (a, b): + tree = coverage.parse_sql(sql, dialect) + if tree is None: + return [] + kept = {id(n) for flt in coverage.filters(tree) for _, bound, _ in flt.conds for n in bound.walk()} + sides.append([(lit, id(lit) in kept or coverage.literal_bound(lit)[1] is not None) + for lit in tree.find_all(exp.Literal)]) + if len(sides[0]) != len(sides[1]): + return [] + return [(str(x.this), str(y.this)) for (x, x_dated), (y, y_dated) in zip(*sides) + if x_dated and y_dated and (x.this, x.is_string) != (y.this, y.is_string)] + + +def _period_key(text: str) -> tuple[str, tuple[int, ...]] | None: + """一个日期、期次字面量(去掉引号)排先后用的 (写法, 键):真正的日期(YYYY-MM-DD、YYYYMMDD,带不带时间)按那一天, + 月、周、季、年按 (年, 第几)。认不出的(-7 days 这类偏移量、序号)为 None:只有写法相同的两个才比先后。""" + from app.data import coverage + + day = coverage.as_day(text) + if day is not None: + return "day", (day.day.toordinal(),) + for shape, pattern in (("month", _MONTH_TEXT), ("week", _WEEK_TEXT), ("quarter", _QUARTER_TEXT)): + if found := pattern.match(text): + return shape, (int(found.group(1)), int(found.group(2))) + if _YEAR_TEXT.match(text): + return "year", (int(text),) + return None + + +def _period_order(pairs: list[tuple[str, str]]) -> tuple[int, int]: + """两份同形 SQL 里不同的日期、期次(_dated_pairs)→ (基期的下标, 本期的下标)。第一对两边写法相同、排得出先后的 + (_period_key)定先后,早的是基期;都排不出的,表达式里先出现的那份(下标 0)是本期。""" + for x, y in pairs: + kx, ky = _period_key(x.strip()), _period_key(y.strip()) + if kx is not None and ky is not None and kx[0] == ky[0] and kx[1] != ky[1]: + return (0, 1) if kx[1] < ky[1] else (1, 0) return 1, 0 diff --git a/backend/app/report_export/runtime.js b/backend/app/report_export/runtime.js index a98595ad..1d574e44 100644 --- a/backend/app/report_export/runtime.js +++ b/backend/app/report_export/runtime.js @@ -95,8 +95,10 @@ calc: '计算', dValues: '具名的数', dFormulas: '关键数的代入式', dResultTable: '结果表', resultRows: (shown, total) => `列出 ${shown} 行,共 ${total} 行`, analysisInput: '输入', dRole: '角色', dRef: '调用里写的名字', dColumnsUsed: '用到的列', dRowsUsed: '用到的行', - allRows: '全部行', dHeadline: '总数那一格', dClosure: '闭合核对', + allRows: '全部行', dHeadline: '总数那一格', dHeadlineRatio: '整体比率那一格', dClosure: '闭合核对', closureOk: (h, t) => `总数 ${h} 与各段之和 ${t} 闭合`, closureBad: (h, t) => `总数 ${h} 与各段之和 ${t} 对不上`, + ratioClosureOk: (h, t) => `整体比率 ${h} 与各段分子之和 ÷ 分母之和 ${t} 闭合`, + ratioClosureBad: (h, t) => `整体比率 ${h} 与各段分子之和 ÷ 分母之和 ${t} 对不上`, dParent: '下钻自', recompute: '复算', seal: '封存', inputOfAnalysis: (alias) => `分析结论 ${alias} 的输入`, inLedger: '封存范围内的台账记过这份分析结论', notInLedger: '封存范围内的台账里没有这份分析结论', viaCard: '记在口径卡的指标集里,随卡封存(不进台账)', @@ -541,10 +543,13 @@ return h('section', { class: 'part', 'data-step': 'analysis_calc' }, h('h3', {}, T.calc), h('div', { class: 'part-body' }, ...body)) } - /** 分析结论 · 输入:角色、落到哪个 Q、用到的列和行、总数那一格和闭合核对、下钻的上一层 */ + /** 分析结论 · 输入:角色、落到哪个 Q、用到的列和行、总数(比率拆解是整体比率)那一格和闭合核对、下钻的上一层 */ function analysisInputPart(step) { const rows = Array.isArray(step.rows) ? step.rows.map((r) => `第 ${Number(r) + 1} 行`).join('、') : step.rows === 'all' ? T.allRows : null const closure = isObj(step.closure) ? step.closure : null + const ratio = closure !== null && closure.mode === 'ratio' + const closed = closure && (ratio ? (closure.closed ? T.ratioClosureOk : T.ratioClosureBad) + : (closure.closed ? T.closureOk : T.closureBad)) const body = [kv([ [T.dRole, String(step.role_label || step.role || NONE)], step.ref ? [T.dRef, h('code', {}, String(step.ref))] : null, @@ -553,8 +558,8 @@ step.node_id ? [T.dNode, nodeLabel(step.node_id)] : null, Array.isArray(step.columns) && step.columns.length ? [T.dColumnsUsed, step.columns.join('、')] : null, rows ? [T.dRowsUsed, rows] : null, - 'value' in step ? [T.dHeadline, `${step.column ? `第 ${Number(step.row) + 1} 行「${step.column}」 = ` : ''}${valueText(step.value)}`] : null, - closure ? [T.dClosure, closure.closed ? T.closureOk(closure.headline, closure.total) : T.closureBad(closure.headline, closure.total)] : null, + 'value' in step ? [ratio ? T.dHeadlineRatio : T.dHeadline, `${step.column ? `第 ${Number(step.row) + 1} 行「${step.column}」 = ` : ''}${valueText(step.value)}`] : null, + closure ? [T.dClosure, closed(closure.headline, closure.total)] : null, step.phrase ? [T.dPhrase, String(step.phrase)] : null, ])] const bits = integrity(step) diff --git a/backend/app/tools/analysis.py b/backend/app/tools/analysis.py index 14566090..fc122fdc 100644 --- a/backend/app/tools/analysis.py +++ b/backend/app/tools/analysis.py @@ -67,8 +67,9 @@ class AnalysisScope: queries:编号 Qn → 输入快照(contract.InputSnapshot;记台账时先放交回模型的那份 payload,分析前 prepare 换成工件库里 那份完整快照,补上遮罩列)。results:编号 An → 整份结论;artifacts:An → 结论的工件 id。snapshot:发起时的分析设置快照 - (没有为 None,现读设置)。mask_cache:快照工件 → 遮罩列;source_cache:数据源和表 → 目录、查询用的源;probe_cache: - 数据源名 → coverage.Cache(一次节点执行里各读、各查一次,只在工具执行时用,不影响编号)。""" + (没有为 None,现读设置)。mask_cache:快照工件 → 遮罩列;source_cache:数据源和表 → 目录、查询用的源;tables_cache: + SQL → 它用到的表(每次分析调用都要按本节点全部查询认表,同一条 SQL 只解析一次);probe_cache:数据源名 → + coverage.Cache(一次节点执行里各读、各查一次,只在工具执行时用,不影响编号)。""" node_id: str snapshot: dict[str, Any] | None = None @@ -77,6 +78,7 @@ class AnalysisScope: artifacts: dict[str, str] = field(default_factory=dict) mask_cache: dict[str, list[str]] = field(default_factory=dict) source_cache: dict[str, Any] = field(default_factory=dict) + tables_cache: dict[str, tuple[str, ...]] = field(default_factory=dict) probe_cache: dict[str, Any] = field(default_factory=dict) #: 已经换成工件库里那份完整快照的编号(prepare 每份只取回一次) prepared: set[str] = field(default_factory=set) @@ -198,27 +200,33 @@ async def prepare(scope: AnalysisScope) -> None: # -------------------------------------------------------------------------- -def _tables_of(snapshots: Iterable[Mapping[str, Any]]) -> dict[str, set[str]]: - """数据源名 → 这几份快照的 SQL 里用到的表(evidence.sql_tables,去掉 schema 前缀前的写法)。合并查询不是数据源。""" +def _tables_of(snapshots: Iterable[Mapping[str, Any]], + memo: dict[str, tuple[str, ...]] | None = None) -> dict[str, set[str]]: + """数据源名 → 这几份快照的 SQL 里用到的表(evidence.sql_tables,去掉 schema 前缀前的写法)。合并查询不是数据源。 + memo 是 SQL → 表的缓存(调用方的,同一个节点里分析好几次,每条 SQL 只解析一次)。""" from app.engine.evidence import sql_tables from app.engine.merge_query import MERGE_SOURCE + memo = memo if memo is not None else {} out: dict[str, set[str]] = {} for snap in snapshots: name = str(snap.get("source") or "") if not name or name == MERGE_SOURCE: continue - try: - tables = sql_tables(str(snap.get("sql") or "")) - except Exception: # noqa: BLE001 - 认不出表就只是少给一点目录 - tables = [] - out.setdefault(name, set()).update(str(t) for t in tables) + sql = str(snap.get("sql") or "") + if sql not in memo: + try: + memo[sql] = tuple(str(t) for t in sql_tables(sql)) + except Exception: # noqa: BLE001 - 认不出表就只是少给一点目录 + memo[sql] = () + out.setdefault(name, set()).update(memo[sql]) return out async def context_of(snapshots: Iterable[Mapping[str, Any]], *, tool_ctx: Any, cache: dict[str, Any] | None = None, - probe_cache: dict[str, Any] | None = None) -> contract.Context: + probe_cache: dict[str, Any] | None = None, + tables_cache: dict[str, tuple[str, ...]] | None = None) -> contract.Context: """算子要的运行期依赖(contract.Context)。每次分析调用一份(探查器有总时限,一次调用建一个)。 - today:运行当天,服务端时区(算子据此判本期完不完整) @@ -229,15 +237,16 @@ async def context_of(snapshots: Iterable[Mapping[str, Any]], *, tool_ctx: Any, 遮罩列按数据源设的不探,探不成由算子退回剖析事实;一个源都解析不出来时为 None tool_ctx 是查询用的那个 ToolContext(固定的数据版本、目录视图都从它取)。cache 是调用方的缓存(同一个节点里分析好几次, - 目录、数据源读一次);probe_cache 是数据源名 → coverage.Cache(同一张表的范围一个节点里只查一次)。读不到的源当没有: - 目录、剖析、探查都是附加的,不能挡住分析。""" + 目录、数据源读一次);tables_cache 是 SQL → 用到的表(同一条 SQL 一个节点里只解析一次:agent 每次分析都按本节点全部 + 查询认表);probe_cache 是数据源名 → coverage.Cache(同一张表的范围一个节点里只查一次)。读不到的源当没有:目录、剖析、 + 探查都是附加的,不能挡住分析。""" from app.analysis.probe import profile_facts cache = cache if cache is not None else {} catalogs: dict[str, Any] = {} facts: dict[str, Any] = {} sources: dict[str, Any] = {} - for name, tables in _tables_of(snapshots).items(): + for name, tables in _tables_of(snapshots, tables_cache).items(): key = f"{name}\n" + "\n".join(sorted(tables)) if key not in cache: cache[key] = await _source_context(name, tables, tool_ctx) diff --git a/backend/app/tools/builtin/analysis.py b/backend/app/tools/builtin/analysis.py index 0dfd9dfa..89213a2a 100644 --- a/backend/app/tools/builtin/analysis.py +++ b/backend/app/tools/builtin/analysis.py @@ -194,7 +194,7 @@ async def run_tool(op: str, ctx: ToolContext, args: BaseModel) -> dict[str, Any] request = request_of(op, args) await prepare(scope) context = await context_of(list(scope.queries.values()), tool_ctx=ctx, cache=scope.source_cache, - probe_cache=scope.probe_cache) + probe_cache=scope.probe_cache, tables_cache=scope.tables_cache) result = await analysis_ops.run(request, scope.resolve, settings=settings, context=context, parents=scope.parent) artifact = await store(result, run_id=ctx.run_id, node_id=scope.node_id) return {"analysis_artifact": artifact, "result": result} diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index d4be71bf..88c1d8a8 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -111,6 +111,24 @@ def pytest_collection_modifyitems(config, items) -> None: item.add_marker(skip) +@pytest.fixture +async def scratch_source(): + """测试里建的 SQLite 数据源:source_id = await scratch_source("shop", 库文件路径, options={...})。同名的旧源先删掉(连同 + 它的数据目录、剖析和现查缓存,tests/fixtures/sources.sqlite_source),测试结束时再删一遍、清掉现查缓存(forget_source)。 + 分析相关的用例都建名叫 shop 的源:别的用例留下的数据目录、5 分钟内现查过的表范围会改掉这次的判定。""" + from tests.fixtures.sources import forget_source, sqlite_source + + made: list[str] = [] + + async def make(name: str, path, **kw) -> str: + made.append(await sqlite_source(name, path, **kw)) + return made[-1] + + yield make + for source_id in made: + await forget_source(source_id) + + @pytest.fixture(scope="session") def scenic_db(tmp_path_factory) -> str: """合成的景区业务库(tests/fixtures/catalog/scenic.py):50 张表、真外键与只能靠命名推断的关系并存。 diff --git a/backend/tests/fixtures/analysis_ops.py b/backend/tests/fixtures/analysis_ops.py index a785ed4c..8b787901 100644 --- a/backend/tests/fixtures/analysis_ops.py +++ b/backend/tests/fixtures/analysis_ops.py @@ -1,4 +1,5 @@ -"""测试用的分析算子替身(第 1 波各包并行:W1-a 的 app.analysis.run 落地前,入口的测试不等它)。 +"""测试用的分析算子替身:入口(agent 分析工具、分析卡、口径卡的可比性挂钩)的测试用它断言入口交给算子的东西、算子交回后 +入口做的事,不受算子的判定细节牵连。接真算子的端到端见 tests/test_analysis_e2e*.py。 FakeOps 换掉模块上的 app.analysis.run(入口都按模块取 run),按请求交回 app.analysis.testing.fake_result 的假结论, 并记下每次调用拿到的请求、按名字解析到的输入快照、阈值、运行期依赖、上一层结论——测试据此断言入口交进去的东西。 diff --git a/backend/tests/fixtures/sources.py b/backend/tests/fixtures/sources.py index 81d9b6da..54eb501d 100644 --- a/backend/tests/fixtures/sources.py +++ b/backend/tests/fixtures/sources.py @@ -3,9 +3,15 @@ 数据源表在每个测试进程的测试库里是共用的:留下来的源会出现在助手的数据源清单里(50 张表的合成库还会 触发挑表、多一次模型调用),别的用例的结果就跟着执行顺序变。阶段 1B 就是这样被绊过一次。 删数据源时它的数据目录随外键级联删除(测试库开着 PRAGMA foreign_keys=ON)。 + +分析相关的用例都建名叫 shop 的源(sqlite_source / forget_source,tests/conftest.py 的 scratch_source 夹具按它们建、按它们删): +别的用例留下的数据目录(金额标成了流量,窗口长度不同就从有条件可比变成不可比)、coverage 的现查缓存里 5 分钟内按同一个 +源 id 记下的表范围(另一份库的日期范围),都会改掉这次的判定。所以建之前先把同名的旧源连同这两样删干净,用完再删一遍。 """ from __future__ import annotations +from typing import Any + from sqlalchemy import delete, select from app.data.engine import engines @@ -26,6 +32,42 @@ async def drop_source(source_id: str | None = None, *, name: str | None = None) await engines.invalidate(source_id) +async def sqlite_source(name: str, path: Any, *, options: dict[str, Any] | None = None, schema: bool = False) -> str: + """建一个指向这份 SQLite 库文件的只读数据源(启用),交回 id。同名的旧源先删掉(forget_source:连同它的数据目录、剖析 + 和现查缓存),这次的源是新的 id。schema 为真时顺手探好表结构(要写数据目录的用例用)。用完调 forget_source。""" + await forget_source(name=name) + async with SessionLocal() as session: + row = DataSource(name=name, kind="sqlite", database=str(path), readonly=True, enabled=True, + options=dict(options or {})) + session.add(row) + await session.flush() + if schema: + from app.data.introspect import introspect + + row.schema_cache = await introspect(row) + await session.commit() + source_id = row.id + await engines.invalidate(source_id) + return source_id + + +async def forget_source(source_id: str | None = None, *, name: str | None = None) -> None: + """按 id 或名字删一个数据源(drop_source:数据目录、剖析随外键级联删除,连接池作废),再清掉 coverage 现查缓存里它的 + 表范围(data/coverage 的 _LIVE,按数据源 id 记、5 分钟有效:同一个进程里后面的用例会拿到这份旧范围)。不存在就什么都 + 不做。""" + from app.data import coverage + + if source_id is None: + async with SessionLocal() as session: + source_id = (await session.execute( + select(DataSource.id).where(DataSource.name == name))).scalar_one_or_none() + if source_id is None: + return + await drop_source(source_id) + for key in [k for k in coverage._LIVE if k and k[0] == source_id]: + coverage._LIVE.pop(key, None) + + async def introspect_now(client, source_id: str, *, expect: int = 202) -> dict: """探查结构并等它做完(探查是后台任务:POST /introspect 回 202 和任务,next-round 03 A1),返回数据源此刻的样子 (DataSourceOut)。expect 不是 202 时只发请求、断言状态码,返回响应体。""" diff --git a/backend/tests/test_analysis_bundle.py b/backend/tests/test_analysis_bundle.py index d41f31aa..6572040a 100644 --- a/backend/tests/test_analysis_bundle.py +++ b/backend/tests/test_analysis_bundle.py @@ -106,6 +106,8 @@ def test_the_analysis_and_its_input_travel_with_the_file_and_verify(): # 导出时服务端真的复算了一遍:按封存的输入快照(这份文件里哈希对得上的原字节)重算,逐字段相同 assert analysis_item["recheck"] == {"status": "same", "method": "decompose.additive@1", "text": "复算一致:按封存的输入重新计算,结论逐项相同"} + # 方法的叫法由服务端带上(和导出页同一份),核对页只显示「变化拆解(可加) · decompose.additive@1」 + assert analysis_item["method_label"] == "变化拆解(可加)" [query] = [x for x in bundle["artifacts"] if x["kind"] == "query"] assert "b64" in query @@ -259,7 +261,7 @@ def test_the_comparability_on_a_card_travels_with_the_card_and_is_chained_throug assert "b64" in item(bundle, card_id) comparability = item(bundle, found) assert comparability["kind"] == ARTIFACT_KIND and "b64" in comparability and comparability["node_id"] == "card" - assert comparability["recheck"]["status"] == "same" + assert comparability["recheck"]["status"] == "same" and comparability["method_label"] == "可比性检查" got = verdicts(bundle) assert all(r == "pass" for key in ("events", "manifest", "artifacts", "chain") for r in got[key]), got["why"] assert {w["kind"] for w in got["why"]["chain"]} >= {"metric_set", ARTIFACT_KIND, "query"} diff --git a/backend/tests/test_analysis_card.py b/backend/tests/test_analysis_card.py index 23773219..5732ccf5 100644 --- a/backend/tests/test_analysis_card.py +++ b/backend/tests/test_analysis_card.py @@ -1,7 +1,8 @@ """分析卡(节点类型 analysis,docs/design/analysis-round/00-总方案.md §3.2):画图时的校验、执行、产出的变量、受管规则、 登记处(叫法、变量表、助手查看、目录影响面、升级提醒)。运行对比按项对齐见 tests/test_analysis_compare.py。 -算子(app.analysis.run)换成按请求交回假结论的替身(tests/fixtures/analysis_ops.FakeOps)。库是虚构的零售库 shop。 +算子(app.analysis.run)换成按请求交回假结论的替身(tests/fixtures/analysis_ops.FakeOps);样例里的下钻那一项另用真算子 +跑一遍(替身不核参数)。库是虚构的零售库 shop:orders(region, week, amount, channel),华东、华南、华北两周。 """ from __future__ import annotations @@ -14,9 +15,8 @@ from sqlalchemy import select from app.analysis import contract from app.analysis.contract import CallError from app.core import artifact_store -from app.data.engine import engines from app.db.base import SessionLocal -from app.db.models import DataSource, Run, RunEvent +from app.db.models import Run, RunEvent from app.engine.runner import run_manager from app.engine.schema import GraphSpec, NodeType, validate_graph from tests.fixtures.analysis_ops import FakeOps @@ -24,16 +24,22 @@ from tests.fixtures.analysis_ops import FakeOps TOOL = "db_query__shop" WEEKLY = "SELECT week, SUM(amount) AS gmv FROM orders GROUP BY week ORDER BY week" REGION = "SELECT region, week, SUM(amount) AS gmv FROM orders GROUP BY region, week ORDER BY region, week" -ROWS = [(1, "华东", "2026-W41", 820.0), (2, "华东", "2026-W42", 700.0), (3, "华南", "2026-W41", 700.0), - (4, "华南", "2026-W42", 666.0), (5, "华北", "2026-W41", 600.0), (6, "华北", "2026-W42", 560.0)] +#: 各地区两周的销售额:华东 820 → 700(线上 500 → 420、线下 320 → 280),华南 700 → 666,华北 600 → 560 +ROWS = [(1, "华东", "2026-W41", 500.0, "线上"), (2, "华东", "2026-W41", 320.0, "线下"), (3, "华东", "2026-W42", 420.0, "线上"), + (4, "华东", "2026-W42", 280.0, "线下"), (5, "华南", "2026-W41", 700.0, "线上"), (6, "华南", "2026-W42", 666.0, "线上"), + (7, "华北", "2026-W41", 600.0, "线下"), (8, "华北", "2026-W42", 560.0, "线下")] +#: 华东按渠道(DRILL 的输入):两个渠道之和 820 → 700,和 WHERE 里华东那一段闭合 +EAST = ("SELECT channel, week, SUM(amount) AS gmv FROM orders WHERE region = '华东' GROUP BY channel, week " + "ORDER BY channel, week") WOW = {"id": "wow", "op": "baseline", "from": "weekly", "params": {"period": "week", "measure": "gmv", "label": "销售额"}} WHERE = {"id": "where", "op": "decompose", "mode": "additive", "from": "by_region", "params": {"by": ["region"], "period": "week", "base": "2026-W41", "target": "2026-W42", "measure": "gmv", "headline": {"base": "weekly.r0.gmv", "target": "weekly.r1.gmv"}, "label": "销售额"}, "hypothesis": "下降是不是集中在某个地区"} -DRILL = {"id": "east", "op": "decompose", "mode": "additive", "from": "by_region", - "params": {"by": ["week"], "period": "week", "base": "2026-W41", "target": "2026-W42", "measure": "gmv", +#: 下钻:WHERE 里华东那一段按渠道再拆一层,读华东按渠道的那次查询(drill_graph 里的 east_q) +DRILL = {"id": "east", "op": "decompose", "mode": "additive", "from": "east_q", + "params": {"by": ["channel"], "period": "week", "base": "2026-W41", "target": "2026-W42", "measure": "gmv", "parent": {"of": "where", "key": ["华东"]}}} @@ -44,30 +50,20 @@ async def engine_up(): await run_manager.shutdown() -async def _shop(tmp_path, options=None): +async def _shop(tmp_path, scratch_source, options=None): + """虚构零售库 shop(tests/conftest.py 的 scratch_source:同名的旧源连同数据目录、现查缓存先删掉,测试结束时再删)。""" path = tmp_path / "shop.db" db = sqlite3.connect(path) - db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, region TEXT, week TEXT, amount REAL);") - db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?)", ROWS) + db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, region TEXT, week TEXT, amount REAL, channel TEXT);") + db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?)", ROWS) db.commit() db.close() - async with SessionLocal() as session: - row = (await session.execute(select(DataSource).where(DataSource.name == "shop"))).scalar_one_or_none() - if row is None: - row = DataSource(name="shop", kind="sqlite", database=str(path), readonly=True, enabled=True) - session.add(row) - row.database, row.kind, row.enabled, row.readonly, row.options = str(path), "sqlite", True, True, options or {} - await session.commit() - source_id = row.id - await engines.invalidate(source_id) - return source_id + return await scratch_source("shop", path, options=options) @pytest.fixture -async def shop(tmp_path): - source_id = await _shop(tmp_path) - yield - await engines.invalidate(source_id) +async def shop(tmp_path, scratch_source): + return await _shop(tmp_path, scratch_source) def node(nid, ntype, **config): @@ -90,6 +86,12 @@ def card_graph(items=(WOW, WHERE), *, extra_nodes=(), extra_edges=(), **card): return {"nodes": nodes, "edges": edges} +def drill_graph(items=(WOW, WHERE, DRILL), **card): + """带下钻那一项的分析卡:华东按渠道的那次查询(east_q)接在 by_region 后面、分析卡前面。""" + return card_graph(items=items, extra_nodes=[node("east_q", "tool", tool=TOOL, args={"sql": EAST})], + extra_edges=[edge("by_region", "east_q"), edge("east_q", "card")], **card) + + def card_issues(graph, level="error"): return [i for i in validate_graph(GraphSpec.model_validate(graph)).issues if i.node_id == "card" and (level is None or i.level == level)] @@ -139,7 +141,7 @@ def test_node_type_and_labels(): def test_a_well_formed_card_validates(): - assert card_issues(card_graph(items=(WOW, WHERE, DRILL), overrides={"main_share": 0.6}), level=None) == [] + assert card_issues(drill_graph(overrides={"main_share": 0.6}), level=None) == [] @pytest.mark.parametrize("items, field, code, why", [ @@ -174,9 +176,9 @@ def test_a_well_formed_card_validates(): "items[1].params.parent", "analysis.params_invalid", "没有「parent」"), ]) def test_item_problems(items, field, code, why): - found = [i for i in card_issues(card_graph(items=items)) if i.field == field] + found = [i for i in card_issues(drill_graph(items=items)) if i.field == field] assert found and found[0].code == code and why in found[0].message, [i.message for i in card_issues( - card_graph(items=items))] + drill_graph(items=items))] @pytest.mark.parametrize("extra, why", [ @@ -235,8 +237,7 @@ def test_two_snapshot_items_need_both_nodes(): async def test_the_card_runs_each_item_and_records_it(monkeypatch, shop): ops = FakeOps(monkeypatch, verdicts=["outside_band", "driver", "driver"], peek=["wow", "where", "east"]) - row = await run_graph(card_graph(items=(WOW, WHERE, DRILL), overrides={"main_share": 0.6}), - options={contract.SETTINGS_KEY: {"k_warn": 2.5}}) + row = await run_graph(drill_graph(overrides={"main_share": 0.6}), options={contract.SETTINGS_KEY: {"k_warn": 2.5}}) assert row.status == "succeeded", row.error # 算子拿到的:请求、按上游节点 id 解析的快照(只认台账里的那几份)、阈值(缺省 ← 发起时的快照 ← 卡上的覆盖) @@ -280,6 +281,25 @@ async def test_the_card_runs_each_item_and_records_it(monkeypatch, shop): assert row.output["判定"] == results["wow"]["phrase"] +async def test_the_drill_item_holds_up_with_the_real_operator(shop): + """上面的 DRILL 换真算子也成立(替身不核参数):华东按渠道拆,两个渠道之和和上一项里华东那一段闭合,主要来自线上;上一层 + 按存下的工件认(parent 的哈希就是 where 那一项的结论工件),按封存的工件复算逐字段相同。""" + import app.analysis as analysis_ops + + row = await run_graph(drill_graph(overrides={"main_share": 0.6})) + assert row.status == "succeeded", row.error + _, output = await output_of(row.id, "card") + results = output["results"] + assert results["where"]["verdict"] == "driver" and "主要来自华东" in results["where"]["phrase"] + east = artifact_store.load(results["east"]["artifact"]) + assert east["status"] == "ok" and east["verdict"] == {"code": "driver", "subjects": [["线上"]], "direction": "down"} + assert east["values"]["base_total"] == "820" and east["values"]["target_total"] == "700" + [parent] = [i for i in east["inputs"] if i.get("role") == "parent"] + assert parent["artifact"] == results["where"]["artifact"] + again = analysis_ops.recompute(east, artifact_store.load, load_result=artifact_store.load) + assert contract.same(east, again) + + async def test_a_refused_item_does_not_fail_the_card(monkeypatch, shop): FakeOps(monkeypatch, statuses=["refused", "ok"]) row = await run_graph(card_graph()) @@ -320,15 +340,12 @@ async def test_inputs_that_are_not_query_snapshots_fail_clearly(monkeypatch, sho assert row.status == "failed" and "不是存成快照的查询结果" in row.error -async def test_masked_columns_reach_the_operator(monkeypatch, tmp_path): - source_id = await _shop(tmp_path, {"mask_columns": ["region"]}) - try: - ops = FakeOps(monkeypatch) - row = await run_graph(card_graph(items=[WHERE])) - assert row.status == "succeeded", row.error - assert "region" in ops.calls[0]["snapshots"]["by_region"]["masked"] - finally: - await engines.invalidate(source_id) +async def test_masked_columns_reach_the_operator(monkeypatch, tmp_path, scratch_source): + await _shop(tmp_path, scratch_source, {"mask_columns": ["region"]}) + ops = FakeOps(monkeypatch) + row = await run_graph(card_graph(items=[WHERE])) + assert row.status == "succeeded", row.error + assert "region" in ops.calls[0]["snapshots"]["by_region"]["masked"] # -------------------------------------------------------------------------- diff --git a/backend/tests/test_analysis_compare.py b/backend/tests/test_analysis_compare.py index 33da6a32..1df4aa0e 100644 --- a/backend/tests/test_analysis_compare.py +++ b/backend/tests/test_analysis_compare.py @@ -40,3 +40,24 @@ def test_compare_aligns_analysis_items(): assert wow["change"] == "phrase" # 没有分析卡的两次运行:analyses 是空的 assert compare.compare(week("r5").side(), week("r6").side())["analyses"] == [] + + +def test_a_card_that_failed_half_way_lists_only_what_a_finished_execution_decided(): + """分析卡执行到一半失败:前几项已经发了 analysis.end,后一项让卡失败(node.failed,没有 node.finished)。那次执行的判定 + 不是这次运行的结果:只认后面跟着 node.finished 的执行,失败的那次不列、也不算执行次数。""" + base = _with_card(week("r7"), [("wow", "within_band", "在常规波动范围内")]) + failed = week("r8") + failed.add("node.started", "insight", {"node_type": "analysis"}) + failed.add("analysis.end", "insight", {"item": "wow", "op": "baseline", "mode": None, "status": "ok", + "verdict": "outside_band", "direction": "down", "phrase": "超出常规波动范围", + "analysis_artifact": "wow-failed"}) + failed.add("node.failed", "insight", {"error": "分析项「where」(变化拆解):比率不能按段相加"}) + [wow] = compare.compare(base.side(), failed.side())["analyses"] + assert wow["change"] == "removed" and wow["target"] is None and wow["base"]["verdict"] == "within_band" + + # 失败后重来一次做完了:按做完的那次对比,执行次数只算做完的 + _with_card(failed, [("wow", "far_outside_band", "明显超出常规波动范围")]) + result = compare.compare(base.side(), failed.side()) + [wow] = result["analyses"] + assert wow["change"] == "verdict" and wow["target"]["verdict"] == "far_outside_band" + assert not [n for n in result["notes"] if "执行了" in n] diff --git a/backend/tests/test_analysis_e2e.py b/backend/tests/test_analysis_e2e.py index c7933b95..4fa907bb 100644 --- a/backend/tests/test_analysis_e2e.py +++ b/backend/tests/test_analysis_e2e.py @@ -1,5 +1,6 @@ -"""分析入口接真算子跑一遍(不用替身):agent 分析工具(含下钻)、分析卡(波动基线 + 变化拆解)、口径卡的可比性挂钩、 -审批恢复时的重放。断言工件、台账、交回模型的那一段、节点产出,并用 analysis.recompute 按封存的工件复算、逐字段相同。 +"""分析入口接真算子跑一遍(不用替身):agent 的五个分析工具(变化拆解含下钻,波动基线、趋势、差异检验、可比性检查各一次)、 +分析卡(波动基线 + 变化拆解;输入是合并查询的一次)、口径卡的可比性挂钩、审批恢复时的重放。断言工件、台账、交回模型的那一段、 +节点产出,并用 analysis.recompute 按封存的工件复算、逐字段相同。 虚构零售库 shop:orders(day, region, channel, amount),2026-03-02 起 12 个整周,每周每个地区、渠道两笔(周三、周日), 最后一周华东掉三成。模型换成脚本(MockChatModel._decide),不调用任何模型。 @@ -19,13 +20,13 @@ import app.analysis as analysis_ops from app.analysis import contract from app.core import artifact_store from app.core.artifact_store import canonical_json, content_hash -from app.data.engine import engines from app.db.base import SessionLocal -from app.db.models import DataSource, Run, RunEvent +from app.db.models import Run, RunEvent from app.engine.runner import run_manager TOOL = "db_query__shop" DECOMPOSE = contract.TOOL_NAMES["decompose"] +BASELINE, TREND, DIFFER, COMPARABLE = (contract.TOOL_NAMES[op] for op in ("baseline", "trend", "differ", "comparable")) WEEK = "date(day, '-6 days', 'weekday 1')" BY_REGION = (f"SELECT region, {WEEK} AS week, SUM(amount) AS gmv FROM orders WHERE day >= '2026-05-11' " "GROUP BY region, week ORDER BY region, week") @@ -64,24 +65,15 @@ async def engine_up(): @pytest.fixture(autouse=True) -async def shop(tmp_path): +async def shop(tmp_path, scratch_source): + """虚构零售库 shop(tests/conftest.py 的 scratch_source:同名的旧源连同数据目录、现查缓存先删掉,用完再删)。""" path = tmp_path / "shop.db" db = sqlite3.connect(path) db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, day TEXT, region TEXT, channel TEXT, amount INTEGER);") db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?)", _orders()) db.commit() db.close() - async with SessionLocal() as session: - row = (await session.execute(select(DataSource).where(DataSource.name == "shop"))).scalar_one_or_none() - if row is None: - row = DataSource(name="shop", kind="sqlite", database=str(path), readonly=True, enabled=True) - session.add(row) - row.database, row.kind, row.enabled, row.readonly, row.options = str(path), "sqlite", True, True, {} - await session.commit() - source_id = row.id - await engines.invalidate(source_id) - yield - await engines.invalidate(source_id) + return await scratch_source("shop", path) def node(nid, ntype, **config): @@ -150,9 +142,9 @@ async def events(run_id, etype=None, node_id=None): return [e.data for e in (await session.execute(q.order_by(RunEvent.seq))).scalars()] -def agent_graph(approval="never"): +def agent_graph(approval="never", tools=(TOOL, DECOMPOSE)): return chain(node("start", "input"), - node("bot", "agent", prompt="销售额为什么下降", tools=[TOOL, DECOMPOSE], approval=approval), + node("bot", "agent", prompt="销售额为什么下降", tools=list(tools), approval=approval), node("out", "output", fields=[{"name": "answer", "value": "{{ nodes.bot.text }}"}])) @@ -211,6 +203,77 @@ async def test_an_agent_decomposes_and_drills_down_with_the_real_operator(monkey recomputes(a2) +async def agent_analysis(monkeypatch, tool, calls): + """Agent 依次查库、最后调一次分析工具(真算子):交回 (那份结论, 交回模型的那一段的 JSON)。顺带断言台账记了这一条(带 + call_id)、交回模型的编号是「【分析 A1】」、判定语和结论一字不差,并按封存的工件复算、逐字段相同。""" + script = Script(monkeypatch, calls) + run = await run_manager.start(graph=agent_graph(tools=(TOOL, tool)), input_payload={}) + row = await wait(run.id) + assert row.status == "succeeded", row.error + ends = await events(row.id, "tool.end", "bot") + artifact = ends[-1]["analysis_artifact"] + stored = artifact_store.load(artifact) + [finished] = await events(row.id, "node.finished", "bot") + [entry] = [e for e in finished["evidence"] if e["kind"] == "analysis"] + assert entry == contract.ledger_entry(stored, node_id="bot", exec_no=1, artifact=artifact, + call_id=f"call_{len(calls) - 1}") + assert set(entry["inputs"]) == {e["query_artifact"] for e in ends[:-1]} + head, body = script.replies()[-1].split("\n", 1) + assert head == "【分析 A1】" and json.loads(body)["phrase"] == stored["phrase"] + recomputes(stored) + return stored, json.loads(body) + + +def _live(stored): + """结论里现查的表范围:按这次运行的数据源探 orders.day。""" + return [(p["table"], p["column"], p["basis"], p["max"]) for p in stored["probes"]] + + +async def test_an_agent_puts_the_week_into_its_history_with_the_real_operator(monkeypatch): + stored, body = await agent_analysis(monkeypatch, BASELINE, [ + (TOOL, {"sql": WEEKLY}), + (BASELINE, {"query": "Q1", "period": "week", "measure": "gmv", "label": "销售额", "hypothesis": "这周是不是异常"})]) + assert stored["op"] == "baseline" and stored["status"] == "ok" and stored["hypothesis"] == "这周是不是异常" + assert stored["verdict"] == {"code": "far_outside_band", "direction": "down", "subjects": []} + assert stored["phrase"].startswith("销售额本期变化 -10.6%,明显超出过去 10 周的常规波动范围") + assert "record_low" in {c["code"] for c in stored["caveats"]} and body["verdict"] == "far_outside_band" + assert _live(stored) == [("orders", "day", "live", "2026-05-24")] # 本期完不完整按现查的表范围判 + + +async def test_an_agent_asks_for_a_trend_with_the_real_operator(monkeypatch): + stored, body = await agent_analysis(monkeypatch, TREND, [ + (TOOL, {"sql": WEEKLY}), (TREND, {"query": "Q1", "period": "week", "measure": "gmv", "label": "销售额"})]) + assert stored["op"] == "trend" and stored["status"] == "ok" + assert stored["verdict"] == {"code": "no_trend", "direction": None, "subjects": []} + assert stored["phrase"] == "过去 12 周销售额没有明显的上升或下降趋势" and stored["values"]["n"] == "12" + assert _live(stored) == [("orders", "day", "live", "2026-05-24")] and body["values"] == stored["values"] + + +async def test_an_agent_tests_a_difference_with_the_real_operator(monkeypatch): + rows = "SELECT region, amount FROM orders WHERE region IN ('华东', '华南') ORDER BY region, day" + stored, body = await agent_analysis(monkeypatch, DIFFER, [ + (TOOL, {"sql": rows}), + (DIFFER, {"mode": "rows", "query": "Q1", "by": ["region"], "value": "amount", "label": "单笔金额"})]) + assert stored["op"] == "differ" and stored["mode"] == "rows" and stored["status"] == "ok" + assert stored["verdict"] == {"code": "differs", "direction": "up", "subjects": [["华东"], ["华南"]]} + assert stored["phrase"] == "华东的单笔金额(196.83)高于华南(152),差异站得住" + assert stored["values"]["n_a"] == stored["values"]["n_b"] == "48" and stored["values"]["test"] == "welch" + assert body["subjects"] == [["华东"], ["华南"]] + + +async def test_an_agent_checks_two_weeks_are_comparable_with_the_real_operator(monkeypatch): + stored, body = await agent_analysis(monkeypatch, COMPARABLE, [ + (TOOL, {"sql": _week_sql("2026-05-18", "2026-05-24")}), (TOOL, {"sql": _week_sql("2026-05-11", "2026-05-17")}), + (COMPARABLE, {"base_query": "Q2", "target_query": "Q1", "measure": "gmv"})]) + assert stored["op"] == "comparable" and stored["status"] == "ok" and stored["phrase"] == "两期可比" + assert stored["params"]["base_query"] == "Q2" and stored["params"]["target_query"] == "Q1" + values = stored["values"] + assert (values["base_start"], values["base_end"], values["target_start"], values["target_end"]) == ( + "2026-05-11", "2026-05-17", "2026-05-18", "2026-05-24") + assert {values[k] for k in ("same_query", "window", "complete", "weekdays", "data")} == {"pass"} + assert _live(stored) == [("orders", "day", "live", "2026-05-24")] and body["verdict"] == "comparable" + + async def test_replaying_after_approvals_does_not_recompute(monkeypatch): counted = Counting(monkeypatch) script = Script(monkeypatch, AGENT_CALLS) @@ -275,6 +338,31 @@ async def test_an_analysis_card_runs_the_real_operators(): recomputes(decomposed) +async def test_an_analysis_card_reads_a_merged_query(): + """分析卡的输入是合并查询(merge 节点):按合并查询节点的做法取它的快照,输入写明 sqlite 方言(合并查询在 SQLite 里 + 执行),真算子照常拆、复算逐字段相同。""" + where = {"id": "where", "op": "decompose", "mode": "additive", "from": "merged", + "params": {"by": ["region"], **TWO_WEEKS, "label": "销售额"}} + graph = chain(node("start", "input"), node("by_region", "tool", tool=TOOL, args={"sql": BY_REGION}), + node("merged", "merge", inputs={"a": "by_region"}, sql="SELECT region, week, gmv FROM a"), + node("card", "analysis", items=[where]), + node("out", "output", fields=[{"name": "r", "value": "{{ nodes.card.results.where.phrase }}"}])) + run = await run_manager.start(graph=graph, input_payload={}) + row = await wait(run.id) + assert row.status == "succeeded", row.error + [finished] = await events(row.id, "node.finished", "card") + results = artifact_store.load(finished["artifact"])["results"] + stored = artifact_store.load(results["where"]["artifact"]) + assert stored["verdict"]["code"] == "driver" and stored["verdict"]["subjects"] == [["华东"]] + assert row.output["r"] == stored["phrase"] and "主要来自华东" in stored["phrase"] + [given] = stored["inputs"] + [merged] = await events(row.id, "node.finished", "merged") + assert (given["ref"], given["role"], given["dialect"]) == ("merged", "data", "sqlite") + assert given["artifact"] == merged["evidence"][0]["artifact"] + assert [(e["kind"], e["item"]) for e in finished["evidence"]] == [("analysis", "where")] + recomputes(stored) + + # -------------------------------------------------------------------------- # 口径卡的可比性挂钩:真 comparable # -------------------------------------------------------------------------- @@ -312,3 +400,86 @@ async def test_a_caliber_card_checks_two_periods_with_the_real_operator(this_wee assert any(p.get("table") == "orders" and p.get("basis") == "live" for p in stored["probes"]) recomputes(stored) assert not [e for e in await events(row.id, "log", "card") if e.get("code") == "metric_comparability_skipped"] + + +def _region_sql(region: str, start: str, end: str) -> str: + return f"SELECT SUM(amount) AS gmv FROM orders WHERE region = '{region}' AND day BETWEEN '{start}' AND '{end}'" + + +@pytest.mark.parametrize("south_week, verdict", [ + (("2026-05-18", "2026-05-24"), None), # 同一周、只差地区:是两段不是两期,不查 + (("2026-05-11", "2026-05-17"), "incomparable"), # 日期和地区都不同:两边除了日期,筛选的还不一样 +]) +async def test_a_ratio_between_two_regions_is_not_a_period_comparison(south_week, verdict): + """华东对华南的倍数:两份同形 SQL 只差地区字面量,不做可比性检查,带契约的出口照常正式出具(以前判成两期不可比、降档)。 + 日期和地区都不同的照查,真算子判不可比,出口记缺口降档。""" + metrics = [{"id": "times", "name": "华东对华南的倍数", "decimals": 4, + "expression": "cell(nodes.east, 0, 'gmv') / cell(nodes.south, 0, 'gmv')"}] + graph = chain(node("start", "input"), + node("east", "tool", tool=TOOL, args={"sql": _region_sql("华东", "2026-05-18", "2026-05-24")}), + node("south", "tool", tool=TOOL, args={"sql": _region_sql("华南", *south_week)}), + node("card", "metrics", caliber="周报口径", metrics=metrics), + node("out", "output", fields=[{"name": "r", "value": "{{ nodes.card.text }}"}], + contract={"metrics_from": ["card"], "required": ["times"]})) + run = await run_manager.start(graph=graph, input_payload={}) + row = await wait(run.id) + assert row.status == "succeeded", row.error + [finished] = await events(row.id, "node.finished", "card") + [metric] = artifact_store.load(finished["artifact"])["metrics"] + issuance = row.output["_issuance"] + assert not [e for e in await events(row.id, "log", "card") if e.get("code") == "metric_comparability_skipped"] + if verdict is None: + assert "comparability" not in metric + assert issuance["tier"] == "formal" and not issuance.get("gaps") and not issuance.get("comparability") + return + found = metric["comparability"] + assert found["verdict"] == verdict and found["reasons"] == ["两边除了日期,筛选的还不一样"] + assert issuance["tier"] == "degraded" + assert issuance["gaps"] == ["指标「华东对华南的倍数」两期不可比:两边除了日期,筛选的还不一样"] + recomputes(artifact_store.load(found["artifact"])) + + +# -------------------------------------------------------------------------- +# 测试夹具:复用名叫 shop 的源,前后都删干净 +# -------------------------------------------------------------------------- + + +async def test_a_scratch_source_starts_clean_and_forgets_its_probes(tmp_path): + """分析相关的用例都建名叫 shop 的源(tests/conftest.py 的 scratch_source):别的用例留下的数据目录、coverage 的现查缓存里 + 按同一个源 id 记下的表范围(5 分钟有效)会改掉这次的判定。sqlite_source 建的源是新的 id,同名的旧源连同数据目录、现查 + 缓存先删掉;forget_source 删源(数据目录随外键级联删除)、清掉现查缓存里它的表范围。""" + from app.analysis.probe import make_prober + from app.data import catalog as data_catalog + from app.data import coverage + from app.data.catalog import make_item + from app.db.models import CatalogNote, DataSource + from tests.fixtures.sources import forget_source, sqlite_source + + path = tmp_path / "other.db" + db = sqlite3.connect(path) + db.executescript("CREATE TABLE orders (day TEXT, amount INTEGER); INSERT INTO orders VALUES ('2026-01-05', 1);") + db.close() + + async def leave_traces(source_id): + async with SessionLocal() as session: + await data_catalog.write_entry(session, source_id, "orders", {"columns": {"amount": { + "measure": make_item("flow", "human")}}}, if_version=0, actor=None) + row = await session.get(DataSource, source_id) + probe = await make_prober({"scratch": row})("scratch", "orders", "day") + assert probe["max"] == "2026-01-05" and probe["basis"] == "live" + assert any(k[0] == source_id for k in coverage._LIVE) + + async def traces(source_id): + async with SessionLocal() as session: + notes = (await session.execute(select(CatalogNote).where(CatalogNote.source_id == source_id))).scalars().all() + row = await session.get(DataSource, source_id) + return {"row": row is not None, "catalog": len(notes), "probes": sum(k[0] == source_id for k in coverage._LIVE)} + + old = await sqlite_source("scratch", path) + await leave_traces(old) + new = await sqlite_source("scratch", path) + assert new != old and await traces(old) == {"row": False, "catalog": 0, "probes": 0} + assert await traces(new) == {"row": True, "catalog": 0, "probes": 0} + await leave_traces(new) + await forget_source(new) + assert await traces(new) == {"row": False, "catalog": 0, "probes": 0} diff --git a/backend/tests/test_analysis_e2e_claims.py b/backend/tests/test_analysis_e2e_claims.py index c07b74f3..9c3a98eb 100644 --- a/backend/tests/test_analysis_e2e_claims.py +++ b/backend/tests/test_analysis_e2e_claims.py @@ -26,15 +26,12 @@ from app.analysis import contract from app.core import artifact_store from app.data import catalog as data_catalog from app.data.catalog import make_item -from app.data.engine import engines -from app.data.introspect import introspect from app.db.base import SessionLocal -from app.db.models import DataSource, Run, RunEvent +from app.db.models import Run, RunEvent from app.engine import analysis_claims as ac from app.engine.evidence import iter_segments from app.engine.runner import run_manager from app.main import app -from tests.fixtures.sources import drop_source TOOL = "db_query__shop" DECOMPOSE = contract.TOOL_NAMES["decompose"] @@ -74,7 +71,7 @@ async def engine_up(): @pytest.fixture(autouse=True) -async def shop(tmp_path): +async def shop(tmp_path, scratch_source): path = tmp_path / "shop.db" db = sqlite3.connect(path) db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, week TEXT, region TEXT, channel TEXT, amount INTEGER);") @@ -84,25 +81,14 @@ async def shop(tmp_path): db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?)", [(i, w, r, c, a) for i, (r, w, c, a) in enumerate(rows, 1)]) db.commit() db.close() + # scratch_source 建的是新源(同名的旧源连同数据目录、现查缓存先删掉),测试结束时删掉、数据目录随外键级联删除:同一个 + # 测试进程里别的用例也用名叫 shop 的源,留着这份目录会改掉它们的判定(比如金额标成了流量,窗口长度不同就从有条件可比 + # 变成不可比) + source_id = await scratch_source("shop", path, schema=True) async with SessionLocal() as session: - row = (await session.execute(select(DataSource).where(DataSource.name == "shop"))).scalar_one_or_none() - if row is None: - row = DataSource(name="shop", kind="sqlite", database=str(path), readonly=True, enabled=True) - session.add(row) - row.database, row.kind, row.enabled, row.readonly, row.options = str(path), "sqlite", True, True, {} - await session.commit() - row.schema_cache = await introspect(row) - await session.commit() for table, notes in NOTES.items(): - current = await data_catalog.read_entry(session, row.id, table) - await data_catalog.write_entry(session, row.id, table, notes, if_version=current.version if current else 0, - actor=None) - source_id = row.id - await engines.invalidate(source_id) - yield - # 删掉这个源(数据目录随外键级联删除):同一个测试进程里别的用例也用名叫 shop 的源,留着这份目录会改掉它们的判定 - # (比如金额标成了流量,窗口长度不同就从有条件可比变成不可比) - await drop_source(source_id) + await data_catalog.write_entry(session, source_id, table, notes, if_version=0, actor=None) + return source_id def _node(nid, ntype, **config): diff --git a/backend/tests/test_analysis_evidence.py b/backend/tests/test_analysis_evidence.py index 84c922de..150fdf1a 100644 --- a/backend/tests/test_analysis_evidence.py +++ b/backend/tests/test_analysis_evidence.py @@ -14,6 +14,7 @@ from __future__ import annotations import asyncio +import re from concurrent.futures import ThreadPoolExecutor from datetime import date from typing import Any @@ -26,6 +27,7 @@ from app.analysis.contract import CallError, effective_settings, ledger_entry from app.analysis.testing import fake_result from app.core.artifact_store import canonical_json, content_hash from app.engine.evidence import ( + ANALYSIS_LEVELS, ANALYSIS_CELLS_REASON, StreamRenderer, analysis_formulas, @@ -473,6 +475,65 @@ def test_trimmed_catalogs_keep_the_judgments_and_fill_result_tables_before_query assert "另有 1 条分析结论,判定和具名的数都列出了,其中 1 条没列出结果表的各段" in tight.text +def _analysis_part(text: str) -> str: + """写作目录里「分析结论」一段(从那段开头的说明到查询结果那段之前)。""" + start = text.index("\n分析结论(") + end = text.find("\n查询结果(", start) + return text[start:end if end >= 0 else len(text)] + + +def _many_analyses(n: int, queries: int = 30) -> tuple[Store, dict[str, Any]]: + """n 份分析结论(A1…An,内容各不相同)、1 + queries 次查询的写作目录。""" + store, ledger, q, _ = setup() + for i in range(n - 1): + res = fake_result(inputs=[{"ref": "Q2", "artifact": q, "role": "data", "columns": ["region", "week", "gmv"], + "rows": "all"}], hypothesis=f"假设 {i}") + ledger.append(ledger_entry(res, node_id="an", exec_no=1, artifact=store.put(res), item=f"i{i}")) + many = [{"kind": "query", "node_id": "fetch", "exec": i + 2, "artifact": store.put({**SNAPSHOT, "sql": f"-- {i}"}), + "tool": "db_query__shop", "source": "shop", "columns": SNAPSHOT["columns"], "rows": 8} + for i in range(queries)] + return store, build_catalog(nodes={}, ledger=[*ledger, *many], loader=store.load) + + +@pytest.mark.parametrize("n, first, last, words", [ + # 先截结果表的段,还超就从后往前只留判定语和具名的数 + (20, "compact", "brief", ("条没列出结果表的各段", "条只列了判定语和具名的数")), + # 都只留判定语和具名的数还超:再从后往前只列编号 + (80, "brief", "head", ("条只列了判定语和具名的数", "条只列了编号")), +]) +def test_the_analysis_part_of_a_trimmed_catalog_takes_at_most_half_of_the_budget(n, first, last, words): + """几十份结论时,分析结论一段最多占预算的一半:按结论的先后从后往前降,先截结果表的段、再只留判定语和具名的数、再只列 + 编号,开头照实写各有几条、没列的照样能引用;剩下的篇幅给查询展开。以前这一段没有上限,40 份结论、每份上千段时在 51.2k + 的预算里占了 50.6k,查询只剩一行摘要。""" + store, catalog = _many_analyses(n) + budget = 12_000 + writer = writer_catalog(catalog, loader=store.load, budget=budget) + assert writer.trimmed and len(writer.text) <= budget + part = _analysis_part(writer.text) + assert len(part) <= budget // 2 + blocks = [b.strip("\n") for b in re.split(r"\n(?=- A\d+:)", part)[1:]] + assert len(blocks) == n # 每条结论都还在,编号连着 + + def level(block: str) -> str: + if "\n" not in block: + return "head" + if "判定标准:" not in block: + return "brief" + return "full" if "\n r0:" in block else "compact" + + assert level(blocks[0]) == first and level(blocks[-1]) == last + assert blocks[-1].startswith(f"- A{n}:变化拆解(可加),来自 Q1") + for block in blocks: + if level(block) == "brief": + assert "判定语 [[a:" in block and "具名的数(写 [[a:" in block and "附带说明" not in block + order = [ANALYSIS_LEVELS.index(level(b)) for b in blocks] + assert order == sorted(order) # 越往后越简 + notice = next(line for line in writer.text.split("\n") if line.startswith("注意:")) + assert f"另有 {n} 条分析结论,这一段最多占篇幅的一半:" in notice and all(w in notice for w in words), notice + assert "没列出的也能引用:判定语单独写 [[a:编号]]" in notice + assert writer.queries == 31 and writer.compact > 0 # 查询照样有展开的 + + def test_metric_comparability_is_spoken_in_the_writer_catalog(): card = {"kind": "metric_set", "caliber": "周报", "caliber_version": "v1", "artifact": "c" * 64, "metrics": [ {"id": "wow", "name": "环比", "value": -0.087, "unit": "", "comparability": { @@ -601,6 +662,25 @@ def test_formulas_substitute_the_key_numbers(): assert analysis_formulas(fake_result(values={})) == [] +@pytest.mark.parametrize("keys, named", [((None, "华南", "华北"), "(空)"), ((True, False, None), "是")]) +def test_the_formula_names_the_segment_the_verdict_names(keys, named): + """代入式里点名的那一段按判定的 subjects 找,维度值按算子写段的同一个函数比(空值写「(空)」、布尔写「是 / 否」)。 + 这里判定点名的段不在结果表第一行(第一行是反方向变化最大的那段):以前拿原值的 str 比,对不上就退回第一行,代入式写成了 + 第一行那段。""" + store = Store() + driver, opposite, other = keys # 变化:driver -220、opposite +250、other -180,总共 -150 + snap = {"columns": ["region", "week", "gmv"], "column_types": {"region": "text", "week": "text", "gmv": "number"}, + "rows": [[driver, "W41", 1000], [driver, "W42", 780], [opposite, "W41", 500], [opposite, "W42", 750], + [other, "W41", 600], [other, "W42", 420]], + "sql": "SELECT region, week, SUM(gmv) AS gmv FROM orders GROUP BY region, week", "source": "shop", + "truncated": False} + result = real_result(store.put(snap), snap) + assert result["verdict"]["code"] == "driver" and result["verdict"]["subjects"] == [[named]] + assert [r[0] for r in result["table"]["rows"]] == [opposite, driver, other] # 按变化的绝对值排 + texts = {f["label"]: f["text"] for f in analysis_formulas(result)} + assert texts["占比"] == f"{named}的占比 = {named}的变化 ÷ 总变化 = -220 ÷ (-150) = 146.7%" + + def test_formulas_for_factor_ratio_and_baseline(): store = Store() factor_snap = {"columns": ["week", "gmv", "orders", "aov"], diff --git a/backend/tests/test_analysis_evidence_api.py b/backend/tests/test_analysis_evidence_api.py index 9c2c7568..87e403da 100644 --- a/backend/tests/test_analysis_evidence_api.py +++ b/backend/tests/test_analysis_evidence_api.py @@ -178,6 +178,51 @@ async def test_named_values_and_result_cells_point_at_what_they_cite(client): assert cell["note"] == "该值由系统从分析结论的结果表中取出、按固定规则显示,未经模型转写" +#: 按门店拆的快照:门店码是文本("1203"),渲染时不加千分位 +STORES: dict[str, Any] = { + "columns": ["store", "week", "gmv"], "column_types": {"store": "text", "week": "text", "gmv": "number"}, + "rows": [["1203", "W41", 820], ["1203", "W42", 700], ["1305", "W41", 600], ["1305", "W42", 560], + ["2201", "W41", 700], ["2201", "W42", 666]], + "sql": "SELECT store, week, SUM(gmv) AS gmv FROM orders GROUP BY store, week ORDER BY store, week", + "source": "shop", "truncated": False, "row_count": 6, +} +#: 按时段拆的快照:时段是 ISO 日期时间(带 T),文档按 2 版显示规则写成空格 +SLOTS: dict[str, Any] = { + "columns": ["slot", "week", "gmv"], "column_types": {"slot": "text", "week": "text", "gmv": "number"}, + "rows": [["2026-10-12T08:00:00", "W41", 820], ["2026-10-12T08:00:00", "W42", 700], + ["2026-10-12T20:00:00", "W41", 600], ["2026-10-12T20:00:00", "W42", 560]], + "sql": "SELECT slot, week, SUM(gmv) AS gmv FROM orders GROUP BY slot, week ORDER BY slot, week", + "source": "shop", "truncated": False, "row_count": 4, +} + + +@pytest.mark.parametrize("snapshot, by, shown", [ + (STORES, "store", "1203"), + (SLOTS, "slot", "2026-10-12 08:00:00"), +]) +async def test_a_result_cell_rerenders_the_way_it_was_composed(client, monkeypatch, snapshot, by, shown): + """证据接口按组装时同一套规则重新渲染结果表的格(evidence.resolve_ref,按文档记下的单元格显示规则):文本码值照结果表 + 记的列类型不加千分位,ISO 日期时间的 T 换成空格。以前不看列类型和显示规则,渲染成「1,203」「2026-10-12T08:00:00」, + render_ok 为假,证据面板和导出页报假的「与按结论渲染的结果不一致」。""" + import tests.test_analysis_evidence_api as here + + monkeypatch.setattr(here, "SNAPSHOT", snapshot) + q = await artifact_store.put_json(snapshot, kind="query_snapshot") + result = await analyzed({"Q2": (snapshot, q)}, by=[by]) + ref = f"v:A1.r0.{by}" + made = await sealed_run(f"{by} [[{ref}]] 变化 [[v:A1.r0.delta]]。[[see:A1]]", result=result) + segment = next(s for s in iter_segments(made["doc"]) if s.get("ref") == ref) + assert segment["text"] == shown + body = (await client.get(f"/api/runs/{made['run']}/evidence/segments/{segment['id']}")).json() + judged = body["chain"][0] + assert judged["rendered"] == shown and judged["render_ok"] is True + assert judged["cited"] == {"what": "cell", "row": 0, "column": by, "raw": snapshot["rows"][0][0]} + # 导出页的数据块是同一条链 + data = data_block((await client.get(f"/api/runs/{made['run']}/report.html")).text) + chain = data["chains"][data["refs"][segment["id"]]] + assert chain[0]["step"] == "analysis" and chain[0]["render_ok"] is True + + async def test_the_recompute_gets_the_sealed_inputs_by_hash(client, monkeypatch): """复算的 load 交工件原内容(不加出处、不改列的类型),只认封存范围内交回过的查询;别的一律当取不回。""" seen: dict[str, Any] = {} @@ -242,6 +287,25 @@ async def test_a_tampered_result_is_said_and_not_shown(client): assert seal["hash_ok"] is False and seal["sealed"] is False and body["seal"]["covered"] is False +def test_suspicious_name_hints_do_not_read_the_analysis_results(monkeypatch): + """可疑实体找「最接近的名字」时按封存的台账重建目录(_scope_catalog):分析结论不进,不读盘、不复验哈希(候选只看表和 + 字段,用不到结论的内容)。以前每点开一个可疑的名字,就把这次运行的分析结论全读一遍。""" + from app.api import evidence as api + from app.engine import evidence as engine_evidence + + read: list[str] = [] + monkeypatch.setattr(engine_evidence, "load_artifact", lambda artifact: read.append(artifact)) + q, a = "1" * 64, "2" * 64 + entries = [_query_entry(SNAPSHOT, q), {"kind": "analysis", "node_id": "an", "exec": 1, "artifact": a, + "op": "decompose", "mode": "additive", "item": "drop", "inputs": [q]}] + graph = {**GRAPH, "edges": [{"source": "fetch", "target": "an"}, {"source": "an", "target": "write"}]} + sealed = api._Sealed(run_id="r", graph=graph, seal={"sealed": True, "ok": True}, events=[], entries=entries) + for report in ("write", "nope"): + catalog = api._scope_catalog(sealed, report) + assert "Q1" in catalog and not [e for e in catalog.values() if e.get("kind") == "analysis"] + assert read == [] + + async def test_agent_analyses_are_marked_as_such(client): made = await sealed_run(TEXT, item=None, call_id="call-9") body = (await client.get(f"/api/runs/{made['run']}/evidence/segments/{seg_id(made['doc'], 'a:A1')}")).json() @@ -260,12 +324,51 @@ async def test_headline_inputs_show_the_closure(client): body = (await client.get(f"/api/runs/{made['run']}/evidence/segments/{seg_id(made['doc'], 'a:A1')}")).json() inputs = [s for s in body["chain"] if s["step"] == "analysis_input"] assert [s["role"] for s in inputs] == ["headline_base", "headline_target", "data"] - assert inputs[0]["closure"] == {"total_key": "base_total", "total": "2,220", "headline": "2,220", "closed": True} - assert inputs[1]["closure"] == {"total_key": "target_total", "total": "2,026", "headline": "2,026", "closed": True} + assert inputs[0]["closure"] == {"mode": "additive", "total_key": "base_total", "total": "2,220", "headline": "2,220", + "closed": True} + assert inputs[1]["closure"] == {"mode": "additive", "total_key": "target_total", "total": "2,026", + "headline": "2,026", "closed": True} + assert [s["role_label"] for s in inputs[:2]] == ["基期总数", "本期总数"] assert inputs[1]["query"] == "Q2" and inputs[1]["sealed"] is True and inputs[2]["query"] == "Q1" assert next(s for s in body["chain"] if s["step"] == "analysis_recompute")["status"] == "same" +async def ratio_run() -> dict[str, Any]: + """一次封存的运行:按渠道拆退款率(比率拆解),给了整体比率的两格(headline,0.035 → 0.053),报告引用判定语。""" + rates = {"columns": ["channel", "week", "refunds", "orders"], + "column_types": {"channel": "text", "week": "text", "refunds": "number", "orders": "number"}, + "rows": [["线上", "W41", 50, 1000], ["线上", "W42", 80, 1000], ["线下", "W41", 20, 1000], + ["线下", "W42", 26, 1000]], + "sql": "SELECT channel, week, SUM(refund_flag) AS refunds, COUNT(*) AS orders FROM orders GROUP BY 1, 2", + "source": "shop", "truncated": False, "row_count": 4} + head = {"columns": ["week", "rate"], "column_types": {"week": "text", "rate": "number"}, + "rows": [["W41", "0.035"], ["W42", "0.053"]], + "sql": "SELECT week, SUM(refund_flag) * 1.0 / COUNT(*) AS rate FROM orders GROUP BY week", + "source": "shop", "truncated": False, "row_count": 2} + r, h = (artifact_store.content_hash(artifact_store.canonical_json(c)) for c in (rates, head)) + result = await analyzed({"Q2": (rates, r), "Q5": (head, h)}, {"op": "decompose", "mode": "ratio", "params": {}}, + query="Q2", by=["channel"], period="week", base="W41", target="W42", numerator="refunds", + denominator="orders", label="退款率", headline={"base": "Q5.r0.rate", "target": "Q5.r1.rate"}) + assert result["status"] == "ok" and result["verdict"]["code"] != "unclosed" + return await sealed_run("[[a:A1]]。", result=result, queries=[rates, head]) + + +async def test_a_ratio_headline_closes_against_the_overall_ratio(client): + """比率拆解给了整体比率的两格(headline):闭合核对对着结论里的整体比率,即各段分子之和 ÷ 分母之和(r_base_total、 + r_target_total),角色写「基期整体比率」「本期整体比率」,closure.mode 写明是比率(导出页照它说「各段分子之和 ÷ 分母 + 之和」,不说「各段之和」)。以前只认可加拆解的总数键,比率拆解的 headline 不显示闭合核对。""" + made = await ratio_run() + body = (await client.get(f"/api/runs/{made['run']}/evidence/segments/{seg_id(made['doc'], 'a:A1')}")).json() + inputs = {s["role"]: s for s in body["chain"] if s["step"] == "analysis_input"} + assert inputs["headline_base"]["closure"] == {"mode": "ratio", "total_key": "r_base_total", "total": "0.035", + "headline": "0.035", "closed": True} + assert inputs["headline_target"]["closure"] == {"mode": "ratio", "total_key": "r_target_total", "total": "0.053", + "headline": "0.053", "closed": True} + assert inputs["headline_base"]["role_label"] == "基期整体比率" + assert inputs["headline_target"]["role_label"] == "本期整体比率" + assert next(s for s in body["chain"] if s["step"] == "analysis_recompute")["status"] == "same" + + async def test_a_drill_down_points_at_its_parent_and_recomputes_from_it(client): """下钻(华东按渠道再拆):输入里有上一层结论(parent),落到报告目录的 A1;复算按 sha256 取回台账记过的上一层。""" q = artifact_store.content_hash(artifact_store.canonical_json(SNAPSHOT)) @@ -434,6 +537,36 @@ async def test_the_html_export_carries_the_analysis_chain_and_the_scope(client): assert f'data-seg="{phrase}"' in html and 'data-kind="analysis"' in html +async def test_the_export_recomputes_each_conclusion_once(client, monkeypatch): + """导出页里同一份结论被判定语、具名的数、整表的每一格、句末依据、分析范围各引一处,引用键不同、各接一条链,复算只做 + 一次(按结论工件记在这一次请求里);分析范围的编号和正文的判定语同一个引用键,不再另建一条链。以前每条链各复算一次, + 1000 段的拆解导出一次复算三十多次。""" + calls: list[str] = [] + real = analysis.recompute + + def counting(result, load, *, load_result=None): + calls.append(str(result.get("method"))) + return real(result, load, load_result=load_result) + + monkeypatch.setattr(analysis, "recompute", counting) + text = ("[[a:A1]]。华东占 [[a:A1.top_share|pct]]。[[see:A1]]\n\n" + "[[table:A1 cols=region,delta,share rows=0-3]]") + made = await sealed_run(text, doc_extra=SCOPE) + data = data_block((await client.get(f"/api/runs/{made['run']}/report.html")).text) + assert calls == ["decompose.additive@1"] + cells = [s["id"] for s in iter_segments(made["doc"]) if str(s.get("ref") or "").startswith("v:A1.")] + assert len(cells) == 12 + keys = {data["refs"][sid] for sid in [*cells, seg_id(made["doc"], "a:A1"), seg_id(made["doc"], "a:A1.top_share|pct")]} + assert len(keys) == 14 # 每处引用各一条链 + for key in keys: + [again] = [s for s in data["chains"][key] if s["step"] == "analysis_recompute"] + assert again["status"] == "same" + assert data["refs"]["scope:A1"] == data["refs"][seg_id(made["doc"], "a:A1")] + # 下一次请求重新复算(缓存只活在一次请求里) + await client.get(f"/api/runs/{made['run']}/evidence/segments/{seg_id(made['doc'], 'a:A1')}") + assert len(calls) == 2 + + async def test_docs_without_a_scope_render_no_scope(client): made = await sealed_run(TEXT) html = (await client.get(f"/api/runs/{made['run']}/report.html")).text diff --git a/backend/tests/test_analysis_pages.py b/backend/tests/test_analysis_pages.py index 7d058b76..6d086b2f 100644 --- a/backend/tests/test_analysis_pages.py +++ b/backend/tests/test_analysis_pages.py @@ -20,7 +20,9 @@ import app.bundle import app.report_export from app.main import app as asgi from tests.test_analysis_bundle import bundle_of, card_bundle -from tests.test_analysis_evidence_api import LAST_WEEK, SCOPE, TEXT, THIS_WEEK, comparability_card, sealed_run, seg_id +from tests.test_analysis_evidence_api import ( + LAST_WEEK, SCOPE, TEXT, THIS_WEEK, comparability_card, ratio_run, sealed_run, seg_id, +) NODE = shutil.which("node") pytestmark = pytest.mark.skipif(NODE is None, reason="没有 node,无法执行页面上的脚本") @@ -155,6 +157,23 @@ async def test_the_export_page_opens_the_analysis_steps_and_the_scope_numbers(): assert out["scope:A1"]["steps"] == opened["steps"] +async def test_the_export_page_speaks_of_the_overall_ratio_for_a_ratio_decomposition(): + """比率拆解的总数那一格是整体比率:导出页写「整体比率那一格」「整体比率 x 与各段分子之和 ÷ 分母之和 y 闭合」,不说 + 「各段之和」。可加拆解照旧(上一个用例的页面里没有「整体比率」)。""" + made = await ratio_run() + async with AsyncClient(transport=ASGITransport(app=asgi), base_url="http://test") as client: + html = (await client.get(f"/api/runs/{made['run']}/report.html")).text + data = re.search(r'', html, re.S).group(1) + phrase = seg_id(made["doc"], "a:A1") + out = run_node(RUN_RUNTIME, RUNTIME, {"data": data, "segs": [ + {"seg": phrase, "kind": "analysis", "state": "deterministic", "text": made["result"]["phrase"]}]}) + body = out[phrase]["body"] + for text in ("基期整体比率", "整体比率那一格", "整体比率 0.035 与各段分子之和 ÷ 分母之和 0.035 闭合", + "整体比率 0.053 与各段分子之和 ÷ 分母之和 0.053 闭合"): + assert text in body, text + assert "各段之和 0.0" not in body + + async def test_the_bundle_page_shows_the_analysis_detail_and_the_scope(): bundle, _, _ = bundle_of(doc_extra=SCOPE) out = run_node(RUN_VERIFIER, VERIFIER, {"data": json.dumps(bundle), "seal": bundle["meta"]["manifest_hash"]}) @@ -164,7 +183,8 @@ async def test_the_bundle_page_shows_the_analysis_detail_and_the_scope(): assert "没拆渠道" in scope and "这次做了 2 次差异检验,几次检验之间没有按次数校正" in scope phrase = next(text for text in out["opened"].values() if "出处分析结论 A1 的判定语" in text) assert "核对结果✓ 通过" in phrase and "分析结论里的值「销售额下降 194(-8.7%)" in phrase - assert "判定语销售额下降 194(-8.7%)" in phrase and "方法与版本decompose.additive@1" in phrase + # 方法写服务端带来的叫法和 id(和导出页同一个写法),不只写 id + assert "判定语销售额下降 194(-8.7%)" in phrase and "方法与版本变化拆解(可加) · decompose.additive@1" in phrase assert "输入数据 · Q1" in phrase # 输入的角色写叫法,不写 data assert "复算一致:按封存的输入重新计算,结论逐项相同(导出时服务端按封存的输入复算的结论;本页只核哈希" in phrase value = next(text for text in out["opened"].values() if "具名的数「top_share」" in text) @@ -196,7 +216,7 @@ def test_the_bundle_page_shows_the_comparability_in_the_metric_detail(): out = run_node(RUN_VERIFIER, VERIFIER, {"data": json.dumps(bundle), "seal": bundle["meta"]["manifest_hash"]}) assert out["verified"] == "" [detail] = out["opened"].values() - for text in ("两期可比性", "判定两期可比", "判定标准只差日期", "方法与版本comparable@1", "各项检查", "同口径", + for text in ("两期可比性", "判定两期可比", "判定标准只差日期", "方法与版本可比性检查 · comparable@1", "各项检查", "同口径", "复算一致:按封存的输入重新计算,结论逐项相同(导出时服务端按封存的输入复算的结论"): assert text in detail, text diff --git a/backend/tests/test_analysis_tools.py b/backend/tests/test_analysis_tools.py index 4f22c8a7..96513247 100644 --- a/backend/tests/test_analysis_tools.py +++ b/backend/tests/test_analysis_tools.py @@ -1,8 +1,9 @@ """agent 分析工具(docs/design/analysis-round/00-总方案.md §3.2):参数、方法说明、查询容器、台账、编号、下钻、次数上限、 遮罩列、续跑。 -算子(app.analysis.run)还没落地:换成按请求交回假结论的替身(tests/fixtures/analysis_ops.FakeOps),断言入口交给算子的 -东西、算子交回后入口做的事。模型换成脚本(MockChatModel._decide),不调用任何模型。库是虚构的零售库 shop。 +算子(app.analysis.run)换成按请求交回假结论的替身(tests/fixtures/analysis_ops.FakeOps),断言入口交给算子的东西、算子 +交回后入口做的事;五个工具接真算子的端到端见 tests/test_analysis_e2e.py。模型换成脚本(MockChatModel._decide),不调用任何 +模型。库是虚构的零售库 shop。 """ from __future__ import annotations @@ -18,9 +19,8 @@ from app.analysis import contract from app.analysis.contract import CallError from app.core import artifact_store from app.core.artifact_store import canonical_json, content_hash -from app.data.engine import engines from app.db.base import SessionLocal -from app.db.models import Artifact, DataSource, Run, RunEvent +from app.db.models import Artifact, Run, RunEvent from app.engine.runner import run_manager from app.tools.analysis import CALL_ERROR_HEAD, REFUSED_CODE from app.tools.builtin.analysis import ANALYSIS_METHOD_PROMPT, ARGS, METHOD_GUIDE, NO_SCOPE @@ -45,30 +45,20 @@ async def engine_up(): await run_manager.shutdown() -async def _shop(tmp_path, options=None): +async def _shop(tmp_path, scratch_source, options=None): + """虚构零售库 shop(tests/conftest.py 的 scratch_source:同名的旧源连同数据目录、现查缓存先删掉,测试结束时再删)。""" path = tmp_path / "shop.db" db = sqlite3.connect(path) db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, region TEXT, week TEXT, amount REAL, phone TEXT);") db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?)", ROWS) db.commit() db.close() - async with SessionLocal() as session: - row = (await session.execute(select(DataSource).where(DataSource.name == "shop"))).scalar_one_or_none() - if row is None: - row = DataSource(name="shop", kind="sqlite", database=str(path), readonly=True, enabled=True) - session.add(row) - row.database, row.kind, row.enabled, row.readonly, row.options = str(path), "sqlite", True, True, options or {} - await session.commit() - source_id = row.id - await engines.invalidate(source_id) - return source_id + return await scratch_source("shop", path, options=options) @pytest.fixture -async def shop(tmp_path): - source_id = await _shop(tmp_path) - yield - await engines.invalidate(source_id) +async def shop(tmp_path, scratch_source): + return await _shop(tmp_path, scratch_source) class Script: @@ -230,6 +220,30 @@ def test_input_snapshots_carry_the_payload_as_is(): assert merged["dialect"] == "sqlite" and merged["masked"] == [] +async def test_each_query_is_parsed_once_per_node_for_its_tables(monkeypatch): + """agent 每次分析都按本节点全部查询认表(目录、剖析、探查都按表给,tools/analysis.context_of):同一条 SQL 一个节点里只 + 解析一次(AnalysisScope.tables_cache)。以前每次调用都把本节点的查询从头解析一遍。""" + from app.engine import evidence + from app.tools.analysis import AnalysisScope, SCOPE_KEY, input_snapshot + from app.tools.builtin.analysis import DecomposeArgs, run_tool + from app.tools.registry import ToolContext + + FakeOps(monkeypatch) + parsed: list[str] = [] + real = evidence.sql_tables + monkeypatch.setattr(evidence, "sql_tables", lambda sql: parsed.append(sql) or real(sql)) + scope = AnalysisScope(node_id="bot", snapshot={}) + for alias, sql in (("Q1", SQL), ("Q2", MASKED_SQL)): + content = {"columns": ["region", "week", "gmv"], "rows": [["华东", "2026-W41", 820.0]], "sql": sql, + "source": "nowhere", "truncated": False} + artifact = await artifact_store.put_json(content, kind="query_snapshot") + scope.add_query(alias, input_snapshot(content, ref=alias, artifact=artifact, node_id="bot")) + ctx = ToolContext(run_id="r", node_id="bot", extra={SCOPE_KEY: scope}) + for _ in range(3): + await run_tool("decompose", ctx, DecomposeArgs(**DECOMPOSE_ARGS)) + assert parsed == [SQL, MASKED_SQL] and scope.tables_cache[SQL] == ("orders",) + + # -------------------------------------------------------------------------- # 方法说明:绑了分析工具的 Agent 注入一次,模型调用节点、没绑的 Agent 不注入 # -------------------------------------------------------------------------- @@ -383,31 +397,25 @@ async def test_the_number_of_analyses_per_node_is_capped(monkeypatch, shop): assert ops.calls[0]["settings"]["max_calls"] == 1 -async def test_masked_columns_reach_the_operator(monkeypatch, tmp_path): - source_id = await _shop(tmp_path, {"mask_columns": ["phone"]}) - try: - ops = FakeOps(monkeypatch) - Script(monkeypatch, [(TOOL, {"sql": MASKED_SQL}), (DECOMPOSE, {**DECOMPOSE_ARGS, "by": ["tel"]})]) - await run_graph(agent_graph()) - masked = ops.calls[0]["snapshots"]["Q1"]["masked"] - # 按 SQL 认出改了名字的遮罩列(phone AS tel)排在前面,数据源设的列名也在 - assert masked[0] == "tel" and "phone" in masked - finally: - await engines.invalidate(source_id) +async def test_masked_columns_reach_the_operator(monkeypatch, tmp_path, scratch_source): + await _shop(tmp_path, scratch_source, {"mask_columns": ["phone"]}) + ops = FakeOps(monkeypatch) + Script(monkeypatch, [(TOOL, {"sql": MASKED_SQL}), (DECOMPOSE, {**DECOMPOSE_ARGS, "by": ["tel"]})]) + await run_graph(agent_graph()) + masked = ops.calls[0]["snapshots"]["Q1"]["masked"] + # 按 SQL 认出改了名字的遮罩列(phone AS tel)排在前面,数据源设的列名也在 + assert masked[0] == "tel" and "phone" in masked -async def test_mask_results_runs_hand_over_the_masked_snapshot(monkeypatch, tmp_path): - source_id = await _shop(tmp_path, {"mask_columns": ["phone"]}) - try: - ops = FakeOps(monkeypatch) - Script(monkeypatch, [(TOOL, {"sql": MASKED_SQL}), (DECOMPOSE, {**DECOMPOSE_ARGS, "by": ["region"]})]) - await run_graph(agent_graph(), options={"mask_results": True, "read_only_data": True}) - snap = ops.calls[0]["snapshots"]["Q1"] - assert "tel" in snap["masked"] - tel = snap["columns"].index("tel") - assert {r[tel] for r in snap["rows"]} == {"已遮罩"}, "遮罩运行里算子拿到的是遮过的快照" - finally: - await engines.invalidate(source_id) +async def test_mask_results_runs_hand_over_the_masked_snapshot(monkeypatch, tmp_path, scratch_source): + await _shop(tmp_path, scratch_source, {"mask_columns": ["phone"]}) + ops = FakeOps(monkeypatch) + Script(monkeypatch, [(TOOL, {"sql": MASKED_SQL}), (DECOMPOSE, {**DECOMPOSE_ARGS, "by": ["region"]})]) + await run_graph(agent_graph(), options={"mask_results": True, "read_only_data": True}) + snap = ops.calls[0]["snapshots"]["Q1"] + assert "tel" in snap["masked"] + tel = snap["columns"].index("tel") + assert {r[tel] for r in snap["rows"]} == {"已遮罩"}, "遮罩运行里算子拿到的是遮过的快照" # -------------------------------------------------------------------------- diff --git a/backend/tests/test_contract_analysis_claims.py b/backend/tests/test_contract_analysis_claims.py index d61f6b47..1b5859df 100644 --- a/backend/tests/test_contract_analysis_claims.py +++ b/backend/tests/test_contract_analysis_claims.py @@ -17,9 +17,8 @@ from httpx import ASGITransport, AsyncClient from langchain_core.messages import AIMessage from sqlalchemy import select -from app.data.engine import engines from app.db.base import SessionLocal -from app.db.models import DataSource, Run, RunEvent, Workflow, WorkflowVersion +from app.db.models import Run, RunEvent, Workflow, WorkflowVersion from app.engine import analysis_claims as ac from app.engine import compiler from app.engine.nodes.io import _analysis_claims, _comparability @@ -271,25 +270,15 @@ async def test_a_governed_formal_run_counts_the_untraced_cited_metric(client, mo @pytest.fixture -async def shop(tmp_path): +async def shop(tmp_path, scratch_source): + """虚构零售库 shop(tests/conftest.py 的 scratch_source:同名的旧源连同数据目录、现查缓存先删掉,测试结束时再删)。""" path = tmp_path / "shop.db" db = sqlite3.connect(path) db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, week TEXT, amount REAL);") db.executemany("INSERT INTO orders VALUES (?, ?, ?)", [(1, "2026-W37", 100.5), (2, "2026-W37", 200.0)]) db.commit() db.close() - async with SessionLocal() as session: - row = (await session.execute(select(DataSource).where(DataSource.name == "shop"))).scalar_one_or_none() - if row is None: - row = DataSource(name="shop", kind="sqlite", database=str(path), readonly=True, enabled=True) - session.add(row) - else: - row.database, row.kind, row.enabled, row.readonly, row.options = str(path), "sqlite", True, True, {} - await session.commit() - source_id = row.id - await engines.invalidate(source_id) - yield - await engines.invalidate(source_id) + return await scratch_source("shop", path) async def test_a_metric_read_from_a_query_cell_is_traced(client, monkeypatch, shop): diff --git a/backend/tests/test_metric_comparability.py b/backend/tests/test_metric_comparability.py index 2fa604ea..fa2530b1 100644 --- a/backend/tests/test_metric_comparability.py +++ b/backend/tests/test_metric_comparability.py @@ -4,7 +4,8 @@ - 同形只差日期 → 有 comparability,早的那期是基期(和表达式里谁先写无关);同一对快照只查一次;原因按算子的 analysis.comparable_reasons 拆 - SQL 结构不同(退款额 / 销售额)、两次是同一条 SQL → 不查、不记日志 -- 算子出错、算不了、算子没有实现(NotImplementedError)→ 不挡口径卡,记一条 metric_comparability_skipped +- 不同的字面量里没有日期、期次(只差地区、只差门店号)→ 是两段不是两期,不查、不记日志;日期和地区都不同的照查 +- 算子出错(调用错误、意料之外的异常)、算不了 → 不挡口径卡,记一条 metric_comparability_skipped 这里的算子是替身(tests/fixtures/analysis_ops.FakeOps);真算子跑一遍的见 tests/test_analysis_e2e.py。 """ @@ -18,9 +19,8 @@ from sqlalchemy import select from app.analysis.contract import CallError from app.core import artifact_store -from app.data.engine import engines from app.db.base import SessionLocal -from app.db.models import DataSource, Run, RunEvent +from app.db.models import Run, RunEvent from app.engine.nodes.metrics import COMPARABILITY_SKIPPED from app.engine.runner import run_manager from tests.fixtures.analysis_ops import FakeOps @@ -40,26 +40,17 @@ async def engine_up(): @pytest.fixture(autouse=True) -async def shop(tmp_path): +async def shop(tmp_path, scratch_source): path = tmp_path / "shop.db" db = sqlite3.connect(path) - db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, day TEXT, region TEXT, amount REAL, refund REAL);") - db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?)", [ - (1, "2026-10-06", "华东", 500.0, 0.0), (2, "2026-10-08", "华南", 400.0, 20.0), - (3, "2026-10-13", "华东", 450.0, 0.0), (4, "2026-10-15", "华北", 300.0, 15.0)]) + db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, day TEXT, region TEXT, store INTEGER, amount REAL, " + "refund REAL);") + db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?, ?)", [ + (1, "2026-10-06", "华东", 1203, 500.0, 0.0), (2, "2026-10-08", "华南", 1204, 400.0, 20.0), + (3, "2026-10-13", "华东", 1203, 450.0, 0.0), (4, "2026-10-15", "华北", 1204, 300.0, 15.0)]) db.commit() db.close() - async with SessionLocal() as session: - row = (await session.execute(select(DataSource).where(DataSource.name == "shop"))).scalar_one_or_none() - if row is None: - row = DataSource(name="shop", kind="sqlite", database=str(path), readonly=True, enabled=True) - session.add(row) - row.database, row.kind, row.enabled, row.readonly, row.options = str(path), "sqlite", True, True, {} - await session.commit() - source_id = row.id - await engines.invalidate(source_id) - yield - await engines.invalidate(source_id) + return await scratch_source("shop", path) def node(nid, ntype, **config): @@ -150,9 +141,59 @@ async def test_different_structures_and_the_same_query_are_not_checked(monkeypat assert not [log for log in [*logs, *logs2] if log.get("code") == COMPARABILITY_SKIPPED] +async def test_two_segments_of_the_same_week_are_not_checked(monkeypatch): + """只差地区(华东对华北的倍数)、只差门店号的两份查询是两段,不是两期:不查、不记日志。日期和地区都不同的照查,两期按 + 日期排先后(不同的那一对里第一处排得出的)。""" + ops = FakeOps(monkeypatch) + ratio = "cell(nodes.this_week, 0, 'gmv') / cell(nodes.last_week, 0, 'gmv')" + east, north = (f"SELECT SUM(amount) AS gmv FROM orders WHERE region = '{r}' AND day BETWEEN '2026-10-12' AND " + "'2026-10-18'" for r in ("华东", "华北")) + _, card, logs = await run_card(graph([{"id": "times", "name": "华东对华北的倍数", "expression": ratio}], + this=east, last=north)) + assert "comparability" not in by_id(card)["times"] and ops.calls == [] + one, two = (f"SELECT SUM(amount) AS gmv FROM orders WHERE store = {n}" for n in (1203, 1204)) + _, card, logs2 = await run_card(graph([{"id": "stores", "name": "两家门店之比", "expression": ratio}], + this=one, last=two)) + assert "comparability" not in by_id(card)["stores"] and ops.calls == [] + assert not [log for log in [*logs, *logs2] if log.get("code") == COMPARABILITY_SKIPPED] + + south_last = "SELECT SUM(amount) AS gmv FROM orders WHERE region = '华南' AND day BETWEEN '2026-10-05' AND '2026-10-11'" + _, card, _ = await run_card(graph([{"id": "wow", "name": "销售额环比", "expression": WOW}], this=east, last=south_last)) + [call] = ops.calls + assert call["request"]["params"] == {"base_query": "last_week", "target_query": "this_week"} + assert by_id(card)["wow"]["comparability"]["verdict"] == "comparable" + + +@pytest.mark.parametrize("this, last, pairs, order", [ + ("region = '华东'", "region = '华南'", [], None), + ("store = 1203", "store = 1204", [], None), + ("amount > 100", "amount > 200", [], None), + ("day BETWEEN '2026-10-12' AND '2026-10-18'", "day BETWEEN '2026-10-05' AND '2026-10-11'", + [("2026-10-12", "2026-10-05"), ("2026-10-18", "2026-10-11")], (1, 0)), + ("day BETWEEN '2026-10-05' AND '2026-10-11'", "day BETWEEN '2026-10-12' AND '2026-10-18'", + [("2026-10-05", "2026-10-12"), ("2026-10-11", "2026-10-18")], (0, 1)), + ("region = '华东' AND day >= '2026-10-12'", "region = '华南' AND day >= '2026-10-05'", + [("2026-10-12", "2026-10-05")], (1, 0)), + ("month = '2026-09'", "month = '2026-10'", [("2026-09", "2026-10")], (0, 1)), + ("week = '2026-W41'", "week = '2026-W40'", [("2026-W41", "2026-W40")], (1, 0)), + ("ym = 202609", "ym = 202608", [("202609", "202608")], (1, 0)), + ("yr = 2025", "yr = 2026", [("2025", "2026")], (0, 1)), + # 按运行当天现算的界:排不出先后,表达式里先出现的那份算本期 + ("day >= date('now', '-7 days')", "day >= date('now', '-14 days')", [("-7 days", "-14 days")], (1, 0)), +]) +def test_which_literals_are_periods(this, last, pairs, order): + from app.engine.nodes.metrics import _dated_pairs, _period_order + + found = _dated_pairs(f"SELECT SUM(amount) AS gmv FROM orders WHERE {this}", + f"SELECT SUM(amount) AS gmv FROM orders WHERE {last}", "sqlite") + assert found == pairs + if order is not None: + assert _period_order(found) == order + + async def test_an_operator_that_is_not_there_does_not_block_the_card(monkeypatch): - """算子没有实现(合并前的骨架抛 NotImplementedError):口径卡照常算,照实记一条没做成。""" - FakeOps(monkeypatch, errors=[NotImplementedError("analysis.run 还没有实现")]) + """算子抛出意料之外的异常(这里是 NotImplementedError):口径卡照常算,照实记一条没做成。""" + FakeOps(monkeypatch, errors=[NotImplementedError("这种写法还没有实现")]) _, card, logs = await run_card(graph([{"id": "wow", "name": "销售额环比", "expression": WOW, "decimals": 4}])) wow = by_id(card)["wow"] assert "comparability" not in wow and wow["value"] == pytest.approx((450 + 300 - 900) / 900, abs=1e-4)