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