mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 03:31:58 +00:00
合入期 5 前置修复 L1：上传表格建的数据源按导入配方里的逻辑类型认日期列，双盲复算的选择能写成业务日期；报告的结论句策略要求核对、正文却写在代码块里（结论句为 0）时，出口契约记一条缺口、不再判完整出具；compose_doc 的说明照实写 source 来自调用方
This commit is contained in:
ThearchyHelios
@@ -9,8 +9,9 @@
|
||||
文本列（SQLite 常把日期存成文字）、整数列（Unix 秒、20260901 这类日期键）、名字带 time 的时长列都可能被说成「按它筛
|
||||
时间」，认不准存的是不是日期，不记。只有时刻的 TIME 也不记：一天里的几点几分（MySQL 还常拿它存时长）不是日期，和
|
||||
catalog.is_date_column（剖析和 SQL 检查共用的日期列判断）同一个口径——记成业务日期，SQL 检查会对按真实日期列统计的
|
||||
查询一直报错误级。上传表格建的数据源，导入时日期一律按文字保存，这一条门槛下它们的业务日期都不由这里记（原因照实说
|
||||
是上传的表格按文字保存，不说成用户声明了 TEXT）
|
||||
查询一直报错误级。上传表格建的数据源，导入时日期一律按文字保存，表结构里总是 TEXT：这样的列看导入时配方里记的逻辑
|
||||
类型（upload_dates，导入确认时界面上标的「日期」就是它），是日期的算日期列；配方读不出来（简单导入没有配方、配方
|
||||
记录坏了）或者配方里不是日期的，照旧不记，原因照实说是上传的表格按文字保存，不说成用户声明了 TEXT
|
||||
- 一边多一个排除无效记录的条件 → 有效记录条件（valid_filter）：被采纳的一边有这个条件才记，记的就是这个条件
|
||||
- 聚合只差去不去重、被采纳的一边去重 → 去重规则（dedup）。采纳了不去重的一边不记：那是这个指标自己的算法（数行），
|
||||
写成表的「不去重」会把别的指标带偏
|
||||
@@ -46,7 +47,8 @@ is_test = 0、is_valid = 1、deleted_at IS NULL）；只涉及一张表。拿不
|
||||
哪一项、本来要写什么，卡上才认得出它和别的指标是不是落到同一项上。同一次裁决里几个指标落到同一项上时，写进去的是
|
||||
合起来的说法（上一段），单个指标的预告说不准：preview_joint 把这几个指标各自选哪一边的每种组合都按真写的办法算一遍
|
||||
（_merged 归拢、_screen 拿掉驳回的、_plan 对目录），卡上按各指标当前的选择取对应的那一种。和真写是同一套判断、同一句
|
||||
原因，只有审批停下期间目录被人改了时才会和最后写的不一样。组合太多的（JOINT_LIMIT）那一项不预告，卡上有条件地说。
|
||||
原因，只有审批停下期间目录被人改了、数据源换了版本时才会和最后写的不一样：真写按写入时的目录、表结构和导入时认作
|
||||
日期的列判断。组合太多的（JOINT_LIMIT）那一项不预告，卡上有条件地说。
|
||||
|
||||
catalog.save_patch 不收备注，这里用它的同一套原料（read_entry、apply_patch、write_entry 的乐观锁）写，只多把备注放进项里。
|
||||
"""
|
||||
@@ -144,11 +146,14 @@ def derive(candidate: Mapping[str, Any], schema_cache: Mapping[str, Any] | None,
|
||||
"""一个候选（审核人在某个指标上选了 a 或 b）→ (要写的目录项, None) 或 (None, 为什么不写)。
|
||||
|
||||
candidate：{metric 指标名, choice "a" | "b", source 数据源名, dialect, sql: {a, b}（比较的两条 SQL）, diff:
|
||||
{equivalent, fallback, items: [FacetDiff.as_dict()]}, scope: {texts, columns}, uploaded}。scope 是这个指标自己的
|
||||
范围：texts 是口径说明、所在分组的取值、本次运行的输入（指标名称总算在内），columns 是分组的列；有效记录条件用它
|
||||
排除指标自己的筛选范围。uploaded 为真是上传表格建的数据源：业务日期不记时，文本列的原因照实说是上传的表格按文字
|
||||
保存日期。schema_cache 是数据源的表结构：认表，业务日期还要看两边的列声明的类型。notes 是这个数据源的目录
|
||||
（{表名: notes}），认码值的意思用。只看差异说明，不重新比 SQL：审核人看到的就是这份说明。
|
||||
{equivalent, fallback, items: [FacetDiff.as_dict()]}, scope: {texts, columns}, uploaded, upload_dates}。scope 是
|
||||
这个指标自己的范围：texts 是口径说明、所在分组的取值、本次运行的输入（指标名称总算在内），columns 是分组的列；有效
|
||||
记录条件用它排除指标自己的筛选范围。uploaded 为真是上传表格建的数据源，upload_dates 是它导入时认作日期的列
|
||||
（{表名: [列名]}，由 upload_dates 从配方读出）：按文字保存的列在这里面的算日期列，不在的业务日期不记、原因照实说
|
||||
是上传的表格按文字保存日期。两项都由节点发起审批时备好、放进候选，审批卡上的预告按它；真写时 record_choices 按
|
||||
写入时的版本重读 upload_dates，和它现读的表结构出自同一个版本。schema_cache 是数据源的表结构：认表，业务日期还要
|
||||
看两边的列声明的类型。notes 是这个数据源的目录（{表名: notes}），认码值的意思用。只看差异说明，不重新比 SQL：
|
||||
审核人看到的就是这份说明。
|
||||
"""
|
||||
name = str(candidate.get("metric") or "")
|
||||
side = candidate.get("choice")
|
||||
@@ -164,7 +169,9 @@ def derive(candidate: Mapping[str, Any], schema_cache: Mapping[str, Any] | None,
|
||||
item = items[0]
|
||||
aspect = str(item.get("aspect") or "")
|
||||
if aspect == "time":
|
||||
fact, why = _business_date(item, side, schema_cache, uploaded=candidate.get("uploaded") is True)
|
||||
dates = candidate.get("upload_dates")
|
||||
fact, why = _business_date(item, side, schema_cache, uploaded=candidate.get("uploaded") is True,
|
||||
dates=dates if isinstance(dates, Mapping) else None)
|
||||
elif aspect == "filter":
|
||||
fact, why = _valid_filter(item, side, candidate, schema_cache, notes or {})
|
||||
elif aspect == "aggregate":
|
||||
@@ -187,11 +194,12 @@ def derive(candidate: Mapping[str, Any], schema_cache: Mapping[str, Any] | None,
|
||||
|
||||
|
||||
def _business_date(item: Mapping[str, Any], side: str, schema_cache: Mapping[str, Any] | None, *,
|
||||
uploaded: bool = False) -> tuple[tuple[str, str, Any] | None, str | None]:
|
||||
uploaded: bool = False, dates: Mapping[str, Any] | None = None
|
||||
) -> tuple[tuple[str, str, Any] | None, str | None]:
|
||||
"""两边按不同的列筛时间、被采纳的一边按哪一列：(表, business_date, {column}) 或 (None, 为什么不记)。两边的列都要在
|
||||
表结构里声明为带日期的类型（_dated_problem），被采纳的一边先看：它有问题说它；没问题再看未被采纳的一边，另说清为什么
|
||||
也要看它（审核人看到的是自己选的那一列，不说就看不懂另一列为什么要紧）。数据源还没有可用的表结构时（没探查、探查
|
||||
失败）只说表结构不可用：那时并不知道有没有这一列，说「不在表结构里」是在误导。"""
|
||||
表结构里声明为带日期的类型、或者是上传表格导入时认作日期的列（_dated_problem），被采纳的一边先看：它有问题说它；没
|
||||
问题再看未被采纳的一边，另说清为什么也要看它（审核人看到的是自己选的那一列，不说就看不懂另一列为什么要紧）。数据源
|
||||
还没有可用的表结构时（没探查、探查失败）只说表结构不可用：那时并不知道有没有这一列，说「不在表结构里」是在误导。"""
|
||||
other = "b" if side == "a" else "a"
|
||||
detail = item.get("detail") if isinstance(item.get("detail"), Mapping) else {}
|
||||
|
||||
@@ -212,7 +220,7 @@ def _business_date(item: Mapping[str, Any], side: str, schema_cache: Mapping[str
|
||||
if not isinstance(tables, Mapping) or not tables:
|
||||
return None, "所在数据源还没有可用的表结构（尚未探查或探查失败），无法确认两边按时间筛的列存的是日期"
|
||||
for adopted, ref in ((True, mine[0]), (False, theirs[0])):
|
||||
if (problem := _dated_problem(tables, ref, uploaded)) is None:
|
||||
if (problem := _dated_problem(tables, ref, uploaded, dates)) is None:
|
||||
continue
|
||||
said = f"按时间筛的列「{ref['table']}.{ref['column']}」{problem}"
|
||||
if adopted:
|
||||
@@ -221,15 +229,20 @@ def _business_date(item: Mapping[str, Any], side: str, schema_cache: Mapping[str
|
||||
return (str(mine[0]["table"]), "business_date", {"column": str(mine[0]["column"])}), None
|
||||
|
||||
|
||||
def _dated_problem(tables: Mapping[str, Any], ref: Mapping[str, Any], uploaded: bool) -> str | None:
|
||||
def _dated_problem(tables: Mapping[str, Any], ref: Mapping[str, Any], uploaded: bool,
|
||||
dates: Mapping[str, Any] | None = None) -> str | None:
|
||||
"""按时间筛的这一列在表结构里是不是声明成了带日期的类型（_DATED_TYPE）：是的为 None，不是的交回它怎么了（查不到、没
|
||||
声明类型、只有时刻、声明成了别的类型）连同结论，接在「按时间筛的列「表.列」」后面读。
|
||||
|
||||
表按 catalog.resolve_table_name 认，列先按原样、再按 name_key 认（差异说明里的名字是规范化过的，SQLite 不加引号的
|
||||
名字一律小写）。按 name_key 认出好几列的说对应好几列：哪一列是它说不准。类型照表结构里记的说（「类型是」，不说
|
||||
「声明为」）：introspect 记的是 SQLAlchemy 反射出来的类型，SQLite 里写成 TIMESTAMPTZ 的列反射出来是 NUMERIC；没写
|
||||
类型的列反射出来是 NullType，记成「NULL」，按没有声明类型说。上传表格建的数据源（uploaded）日期一律存成文字，文字
|
||||
类型不说「类型是 TEXT」：那不是用户声明的，导入确认时这一列标的还是「日期」，表结构也改不了。"""
|
||||
类型的列反射出来是 NullType，记成「NULL」，按没有声明类型说。
|
||||
|
||||
上传表格建的数据源（uploaded）日期一律存成文字，表结构改不了，类型只能看导入时配方里记的（dates，见 upload_dates）：
|
||||
按文字保存的列在里面的就是日期列（按表结构里认出的表名、列名去对，大小写不论）。不在里面的（配方读不出来、配方里
|
||||
不是日期）不说「类型是 TEXT」：那不是用户声明的，说清是上传的表格按文字保存。声明成别的类型的列照常说类型，只认
|
||||
文字类型：配方里的日期列只会存成文字，别的类型的列存的就不是配方里那一列。"""
|
||||
found_table = catalog.resolve_table_name({"tables": tables}, str(ref["table"]))
|
||||
meta = tables.get(found_table) if found_table is not None else None
|
||||
columns = [c for c in meta.get("columns") or [] if isinstance(c, Mapping) and c.get("name")] \
|
||||
@@ -249,10 +262,79 @@ def _dated_problem(tables: Mapping[str, Any], ref: Mapping[str, Any], uploaded:
|
||||
if _TIME_ONLY_TYPE.fullmatch(declared):
|
||||
return f"在表结构里的类型是 {declared}，只有时刻，不含日期"
|
||||
if uploaded and any(word in declared.upper() for word in catalog._TEXT_TYPES):
|
||||
if _imported_as_date(dates, str(found_table), str(found[0]["name"])):
|
||||
return None
|
||||
return "来自上传的表格，表格里的日期一律按文字保存，无法确认存的是日期"
|
||||
return f"在表结构里的类型是 {declared}，无法确认存的是日期"
|
||||
|
||||
|
||||
def _imported_as_date(dates: Mapping[str, Any] | None, table: str, column: str) -> bool:
|
||||
"""表结构里的这一列是不是上传表格导入时认作日期的列（dates：{表名: [列名]}）。表、列都先按原样、再按 name_key 对：
|
||||
名字出自同一份配方，本来一字不差；按 name_key 对上好几个的不算，说不准是哪一个。"""
|
||||
if not isinstance(dates, Mapping):
|
||||
return False
|
||||
|
||||
def one(names: Any, wanted: str) -> str | None:
|
||||
names = [str(n) for n in names if isinstance(n, str)]
|
||||
if wanted in names:
|
||||
return wanted
|
||||
loose = [n for n in names if name_key(n) == name_key(wanted)]
|
||||
return loose[0] if len(loose) == 1 else None
|
||||
|
||||
key = one(list(dates), table)
|
||||
columns = dates.get(key) if key is not None else None
|
||||
return isinstance(columns, (list, tuple)) and one(columns, column) is not None
|
||||
|
||||
|
||||
# ---- 上传表格导入时认作日期的列：配方里记的逻辑类型
|
||||
|
||||
def recipe_dates(recipe: Any) -> dict[str, list[str]]:
|
||||
"""一份配方（recipe_types.Recipe）里逻辑类型是日期的列：{表名: [列名]}，没有日期列的表不列。
|
||||
|
||||
就是导入确认时界面上标「日期」的那些（recipe_notes 给列写日期格式说明认的也是这两种）：交叉表的日期轴（Axis.type 恒为
|
||||
日期），列表块里类型记为 DATE 的列。执行器把它们一律存成 YYYY-MM-DD 的文字，表结构里只剩 TEXT。哪张表有哪几列按
|
||||
recipe_types.derive_tables 推（建表用的同一份）：日期轴写进它各分段的表，列表合计行另存的表只有合计项和数字列。
|
||||
列表的类型按 _DATED_TYPE 认，配方以后添了日期时间也算；只有时刻的不算。几个列表块写同一张表的，这一列每一块都
|
||||
记为日期才算：日期和文字都存成 TEXT，建出的表形状一样，配方校验不拦，可按文字导入的那一块存的是单元格原文。"""
|
||||
from app.data.recipe_types import ListBlock, derive_tables
|
||||
|
||||
tables, _ = derive_tables(recipe)
|
||||
listed: dict[tuple[str, str], bool] = {}
|
||||
for sheet in recipe.sheets:
|
||||
for block in sheet.blocks:
|
||||
if isinstance(block, ListBlock):
|
||||
for column in block.columns:
|
||||
key = (block.table, column.name)
|
||||
listed[key] = listed.get(key, True) and _DATED_TYPE.fullmatch(column.type) is not None
|
||||
out: dict[str, list[str]] = {}
|
||||
for table, columns in tables.items():
|
||||
names = [c.name for c in columns if c.role == "axis" or listed.get((table, c.name), False)]
|
||||
if names:
|
||||
out[table] = list(dict.fromkeys(names))
|
||||
return out
|
||||
|
||||
|
||||
async def upload_dates(session: Any, row: DataSource | None) -> dict[str, list[str]]:
|
||||
"""上传表格建的数据源当前版本里，导入时认作日期的列：{表名: [列名]}（recipe_dates）。只读，不碰导入流程。
|
||||
|
||||
配方取现行的那一版（current_recipe_id）：发布、启用旧版本时它和当前快照、表结构（schema_cache）一起换
|
||||
（table_versions.activate_snapshot），配方和表结构出自同一个版本。读不出来的一律给空——手工登记的源、简单导入
|
||||
（没有配方，构建回执里只有存储类型）、配方记录不在了、配方解析不了：业务日期照旧不记，原因说按文字保存。"""
|
||||
from app.data.recipe_types import Recipe
|
||||
from app.db.models import TableRecipe
|
||||
|
||||
if row is None or (row.origin or "manual") != "upload" or not row.current_recipe_id:
|
||||
return {}
|
||||
record = await session.get(TableRecipe, row.current_recipe_id)
|
||||
if record is None or record.source_id != row.id:
|
||||
return {}
|
||||
try:
|
||||
return recipe_dates(Recipe.model_validate(record.recipe))
|
||||
except ValueError:
|
||||
logger.warning("上传表格的配方记录解析不了，业务日期按文字保存处理：source=%s recipe=%s", row.id, record.id)
|
||||
return {}
|
||||
|
||||
|
||||
def _valid_filter(item: Mapping[str, Any], side: str, candidate: Mapping[str, Any],
|
||||
schema_cache: Mapping[str, Any] | None,
|
||||
notes: Mapping[str, Mapping[str, Any]]) -> tuple[tuple[str, str, Any] | None, str | None]:
|
||||
@@ -732,7 +814,8 @@ async def record_choices(candidates: list[Mapping[str, Any]], *, run_id: str, no
|
||||
进去，都落在 skipped 里照实说。
|
||||
|
||||
先把每个候选认成目录项，再按（数据源、表、项）归拢成一项（_merged：说法一样的、有效记录条件合起来写，说法
|
||||
不同的整组不写），逐项写（_write：按读到的目录先拿掉各自带回驳回说法的指标，其余的照写）。"""
|
||||
不同的整组不写），逐项写（_write：按读到的目录先拿掉各自带回驳回说法的指标，其余的照写）。认目录项用的表结构、
|
||||
目录、上传表格导入时认作日期的列（upload_dates）都按写入时数据源的现状读。"""
|
||||
skipped: list[dict[str, Any]] = []
|
||||
facts: list[Fact] = []
|
||||
for candidate in candidates:
|
||||
@@ -744,6 +827,11 @@ async def record_choices(candidates: list[Mapping[str, Any]], *, run_id: str, no
|
||||
skipped.append({"reason": f"数据源「{name}」已不存在，指标「{candidate.get('metric')}」的选择未写入数据目录"})
|
||||
continue
|
||||
notes = {table: entry.notes for table, entry in (await catalog.read_catalog(session, row.id)).items()}
|
||||
if "upload_dates" in candidate:
|
||||
# 表结构是这里现读的，导入时认作日期的列也按写入时的当前版本重读：审批停着时换了版本（日期列改成了
|
||||
# 文字），拿发起审批时读的那份去配新的表结构，会写出一条哪个版本单独看都不会给出的人工确认。候选里
|
||||
# 没有这一项的是升级前发起的审批，卡上预告的是不写，照旧不认
|
||||
candidate = {**candidate, "upload_dates": await upload_dates(session, row)}
|
||||
fact, why = derive(candidate, row.schema_cache, notes)
|
||||
except Exception: # noqa: BLE001 - 目录读不出来不能毁掉审批的结果：缺口照样按选择记
|
||||
logger.exception("双盲复算的选择核对数据目录失败（数据源 %s）", name)
|
||||
@@ -980,4 +1068,5 @@ async def _write(session: Any, row: DataSource, fact: Fact, *, run_id: str, node
|
||||
return None, [*reasons, f"「{fact.table}」的数据目录正被修改，{label}未能写入，请稍后在数据目录中确认"]
|
||||
|
||||
|
||||
__all__ = ["FACT_ITEMS", "JOINT_LIMIT", "Fact", "derive", "note_of", "preview", "preview_joint", "record_choices"]
|
||||
__all__ = ["FACT_ITEMS", "JOINT_LIMIT", "Fact", "derive", "note_of", "preview", "preview_joint", "recipe_dates",
|
||||
"record_choices", "upload_dates"]
|
||||
|
||||
@@ -2506,7 +2506,7 @@ def compose_doc(
|
||||
|
||||
stats / violations 由 verify_doc 算出，和出口契约复核用的是同一个函数。
|
||||
整表标记先展开成每格一个 [[v:]] 的 Markdown 表，之后和模型自己写的表格走同一条路；
|
||||
doc["source"] 仍是模型的原文。
|
||||
doc["source"] 是调用方交来的原文（报告节点交来前去掉了整篇外层围栏，见 report._unfenced）。
|
||||
"""
|
||||
source = (raw or "").replace("\r\n", "\n").replace("\r", "\n").strip()
|
||||
snaps = _Snapshots(loader, entities=entities)
|
||||
|
||||
@@ -69,7 +69,7 @@ from app.db.models import DataSource
|
||||
from app.engine import deadline, recompute
|
||||
from app.engine.approval import read_decision
|
||||
from app.engine.context import NodeContext, NodeError
|
||||
from app.engine.dual_check_facts import preview, preview_joint, record_choices
|
||||
from app.engine.dual_check_facts import preview, preview_joint, record_choices, upload_dates
|
||||
from app.engine.evidence import build_catalog, is_number, next_exec
|
||||
from app.engine.expressions import ExpressionError, _cell_call, _chain_root, cell_value, column_kind, numeric_text, \
|
||||
parse_expression
|
||||
@@ -1126,6 +1126,7 @@ async def _review(state: GraphState, ctx: NodeContext, checked: dict[str, Any],
|
||||
facts[row["metric_id"]] = {
|
||||
"metric_id": row["metric_id"], "metric": row["name"], "source": first.get("source"),
|
||||
"dialect": info.get("kind"), "uploaded": bool(info.get("uploaded")),
|
||||
"upload_dates": info.get("upload_dates") or {},
|
||||
"sql": {"a": single[0], "b": single[1]} if single else {},
|
||||
"diff": {"equivalent": merged["equivalent"], "fallback": merged["fallback"] or single is None,
|
||||
"items": merged["items"]},
|
||||
@@ -1165,19 +1166,23 @@ async def _review(state: GraphState, ctx: NodeContext, checked: dict[str, Any],
|
||||
|
||||
async def _source_info(sources: _Sources, name: str) -> dict[str, Any]:
|
||||
"""比较两条 SQL、预告目录写入要用的：方言、表结构、列的中文名、整份数据目录，以及是不是上传表格建的数据源
|
||||
（uploaded：导入时日期一律存成文字，业务日期不记时原因照实说，见 dual_check_facts._dated_problem）。数据源没了就
|
||||
都不给，照样比原文。"""
|
||||
（uploaded：导入时日期一律存成文字）和它导入时认作日期的列（upload_dates，按当前版本的配方读，见
|
||||
dual_check_facts.upload_dates）——业务日期据此认上传表格的日期列，认不出的原因照实说（_dated_problem）。两项都进
|
||||
候选：审批卡的预告按发起审批时读的这一份；裁决后的真写按写入时的版本重读 upload_dates，和现读的表结构一起以写入时
|
||||
为准（dual_check_facts.record_choices）。数据源没了就都不给，照样比原文。"""
|
||||
from app.data import catalog
|
||||
from app.data.sql_semantics import catalog_labels
|
||||
|
||||
row = await sources.source(name)
|
||||
if row is None:
|
||||
return {"kind": None, "schema": None, "names": {}, "notes": {}, "uploaded": False}
|
||||
return {"kind": None, "schema": None, "names": {}, "notes": {}, "uploaded": False, "upload_dates": {}}
|
||||
async with SessionLocal() as session:
|
||||
entries = await catalog.read_catalog(session, row.id)
|
||||
dates = await upload_dates(session, row)
|
||||
notes = {table: entry.notes for table, entry in entries.items()}
|
||||
return {"kind": row.kind, "schema": row.schema_cache if isinstance(row.schema_cache, dict) else None,
|
||||
"names": catalog_labels(notes), "notes": notes, "uploaded": (row.origin or "manual") == "upload"}
|
||||
"names": catalog_labels(notes), "notes": notes, "uploaded": (row.origin or "manual") == "upload",
|
||||
"upload_dates": dates}
|
||||
|
||||
|
||||
async def _explain(original: Any, blind: Any, info: dict[str, Any]) -> dict[str, Any]:
|
||||
|
||||
@@ -488,6 +488,224 @@ _ON_UNSUPPORTED = ("degrade", "withhold")
|
||||
#: 要求结论句挂依据的策略：judge 在这一点上和 require_citation 一样严，另外再按裁判的判定判档
|
||||
_CITING = ("require_citation", "judge")
|
||||
|
||||
#: 报告的正文写在代码块里、一句结论都没切出来时的缺口（_fenced_body）。不带核查节点的前缀：界面归在「校验未全部完成」下
|
||||
GAP_FENCED_BODY = "报告「{title}」的正文写在代码块里，结论句未能核对"
|
||||
#: 写的是文字的代码块：信息串为空、markdown、md（报告节点去整篇外层围栏时认的那几个，report._PROSE_INFO），另加 text、
|
||||
#: plaintext、txt、plain 这几种纯文字的写法。sql、python、json 这类是模型真要贴的代码
|
||||
_PROSE_FENCES = frozenset({"", "markdown", "md", "text", "plaintext", "txt", "plain"})
|
||||
_CJK = re.compile(r"[\u4e00-\u9fff]")
|
||||
#: 中文的句读：一行里有汉字、又有它们的，是在写句子（SQL 里的中文列名、别名不带这些）
|
||||
_CJK_PUNCT = re.compile(r"[，。！？；：]")
|
||||
#: 一行里连着六个以上汉字：信息串写明是文字的块里，这就是在写中文。表格的列名、地区名、门店名这类标签多在五个字以内
|
||||
_CJK_RUN = re.compile(r"[\u4e00-\u9fff]{6,}")
|
||||
#: 英文的词串：单词之间只隔空格、逗号。五个以上的单词紧跟句末标点算一句（_en_sentence）
|
||||
_EN_RUN = re.compile(r"[A-Za-z]+(?:[ ,]+[A-Za-z]+)*")
|
||||
_EN_WORD = re.compile(r"[A-Za-z]+")
|
||||
#: 整行是注释（-- // # /* */ 开头），和行尾隔着空白的注释：写着中文句子也是代码的一部分。# 开头的 Markdown 标题本来
|
||||
#: 也不算结论句（evidence._scan_blocks 把它切成 heading）
|
||||
_COMMENT_LINE = re.compile(r"\s*(?:--|//|#|/\*|\*/)")
|
||||
_TRAILING_COMMENT = re.compile(r"\s(?:--|//|#).*")
|
||||
#: 看得出是代码的行里，-- // 紧挨着前面的字也是注释（SELECT SUM(amount)--按区域汇总）：字符串已经先去掉了
|
||||
_CODE_COMMENT = re.compile(r"(?:--|//|\s#).*")
|
||||
#: 写明是文字的块里顶格写的 Markdown 标题：标题前后是两段（_code_lines）
|
||||
_MD_HEADING = re.compile(r"#{1,6}\s")
|
||||
#: 代码里的字符串字面量（'…'、"…"，可带 f、r、b、u、N 这类前缀）：里面的中文标点是数据。引号前面紧挨着字母、数字的
|
||||
#: 不算开头（英文的 it's、don't）
|
||||
_STRING_LITERAL = re.compile(r"(?<![A-Za-z0-9_])[rRbBuUfFnN]{0,2}(?:'[^'\n]*'|\"[^\"\n]*\")")
|
||||
#: 可以跨行的块注释、多行字符串：开头的写法，和它对应的收尾
|
||||
_SPAN_OPEN = re.compile(r"/\*|\"\"\"|'''")
|
||||
_SPAN_CLOSE = {"/*": "*/", '"""': '"""', "'''": "'''"}
|
||||
#: SQL、Python 语句的开头。SQL 的关键字只认全小写、全大写的写法：「From 数据看……」「Update ……」这类首字母大写的是
|
||||
#: 正文里的英文词；def、class 要写成定义的样子（def 名字(、class 名字: ），「class A 门店……」说的是门店的分类
|
||||
_CODE_LINE = re.compile(r"\s*(?:(?:select|with|from|where|join|group\s+by|order\s+by|having|union|insert|update|delete"
|
||||
r"|create|import|return|SELECT|WITH|FROM|WHERE|JOIN|GROUP\s+BY|ORDER\s+BY|HAVING|UNION|INSERT"
|
||||
r"|UPDATE|DELETE|CREATE)\b|(?:def|class)\s+[A-Za-z_]\w*\s*[(:])")
|
||||
#: YAML 的键（「columns:」「- name: 退款占比」），只认小写开头的：正文里的「Note:」「Q1:」是标签。值为空、下面缩进着
|
||||
#: 列表或子键的另记一种（_YAML_NEST）
|
||||
_YAML_KEY = re.compile(r"\s*(?:-\s+)?[a-z_][\w.-]*:(?:\s|$)")
|
||||
_YAML_NEST = re.compile(r"\s*(?:-\s+)?[a-z_][\w.-]*:\s*$")
|
||||
#: 块里再嵌的代码块：~~~ 或三个以上反引号开头。compose_doc 只认整行三个反引号的围栏，那样的一行外层就收了，块里剩下的
|
||||
#: 嵌套写法是 ~~~sql、````sql 这些
|
||||
_NESTED_FENCE = re.compile(r"\s*(`{3,}|~{3,})(.*)")
|
||||
#: 像数据、公式的一行：前面有字的等号（公式、赋值、比较）和箭头（连起的流程），画目录树的制表符，JSON 的对象、数组开头
|
||||
#: （{"、["）和「"键":」，函数调用（print(、groupby(），带扩展名的文件名（周报.xlsx）。行首的「→ 」「=> 」是列表的
|
||||
#: 记号，不算
|
||||
_DATA_LINE = re.compile(r"\S\s*(?:[=＝→⟶⇒]|->)|[\u2500-\u257f]|^\s*[\[{]\s*[\"'\[{]|[\"']\s*:"
|
||||
r"|(?<![\w.])[A-Za-z_][\w.]*\(|(?<=[^\s.])\.[A-Za-z][A-Za-z0-9]{1,4}(?![\w.])")
|
||||
|
||||
|
||||
def _en_sentence(line: str) -> bool:
|
||||
"""一句英文：五个以上的单词（之间只隔空格、逗号）紧跟句末标点，后面是空白或行尾。按词串逐个数，不靠正则回溯：
|
||||
回溯的写法遇上很长、不带句末标点的一行，耗时随行长平方增长，出口复核又是在事件循环上同步执行的。"""
|
||||
for run in _EN_RUN.finditer(line):
|
||||
end = run.end()
|
||||
if line[end:end + 1] in (".", "!", "?") and (end + 1 == len(line) or line[end + 1].isspace()) \
|
||||
and len(_EN_WORD.findall(run.group())) >= 5:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _sentence(line: str) -> bool:
|
||||
"""成句：两个以上汉字又有中文句读，或者是一句英文。"""
|
||||
return (len(_CJK.findall(line)) >= 2 and _CJK_PUNCT.search(line) is not None) or _en_sentence(line)
|
||||
|
||||
|
||||
def _concluding(line: str) -> bool:
|
||||
"""按正文的规矩这一行会算一句结论（evidence._classify 认文字结论的那条）：带方向词、因果词，又不只是在下定义。
|
||||
只认有两个以上汉字的行：drop、rise 这类英文词在代码里也常见。数字不看，代码块里的数字报告节点按裸数字查。"""
|
||||
from app.engine.evidence import _CAUSAL, _DEFINITION, _DIRECTIONAL
|
||||
|
||||
return len(_CJK.findall(line)) >= 2 and _DIRECTIONAL.search(line) is not None \
|
||||
and not (_DEFINITION.search(line) is not None and _CAUSAL.search(line) is None)
|
||||
|
||||
|
||||
def _written(line: str) -> bool:
|
||||
"""信息串写明是文字（markdown、text 这类）的块里，不是代码的这一行在写正文：成句、按正文的规矩会算结论，或者一行里
|
||||
连着六个以上汉字。不看列表标记：标记的写法太多（- + 1. 1、 一、 （1） · ●，也常不带标记），补一种漏一种；也不去
|
||||
字符串、行尾注释，# -- 和半角引号都是正文里的符号。整行是注释、标题的不算。"""
|
||||
if _COMMENT_LINE.match(line):
|
||||
return False
|
||||
return _sentence(line) or _CJK_RUN.search(line) is not None or _concluding(line)
|
||||
|
||||
|
||||
def _prose_line(line: str, *, code: bool) -> bool:
|
||||
"""代码的行（code），或者信息串为空的块里的行，在写正文：去掉字符串字面量和行尾注释之后成句；不是代码的行另认按正文
|
||||
的规矩会算结论的（不带句读的列表项、不带标记的行）。不带句读、也不像结论的不算：YAML 里的列名、diff 里改的口径和
|
||||
列表项没有分别。整行是注释的不算。"""
|
||||
if _COMMENT_LINE.match(line):
|
||||
return False
|
||||
line = (_CODE_COMMENT if code else _TRAILING_COMMENT).sub("", _STRING_LITERAL.sub(" ", line), count=1)
|
||||
return _sentence(line) or (not code and _concluding(line))
|
||||
|
||||
|
||||
def _blank_spans(text: str) -> str:
|
||||
"""块注释（/* … */）、三个引号括起的多行字符串涂成空格，换行留着，块里还是原来那几行：里面写着中文句子也是代码的
|
||||
一部分。没收尾的不动，开头那一行照注释行跳过。"""
|
||||
out: list[str] = []
|
||||
at = 0
|
||||
while (start := _SPAN_OPEN.search(text, at)) is not None:
|
||||
end = text.find(_SPAN_CLOSE[start.group()], start.end())
|
||||
if end == -1:
|
||||
break
|
||||
stop = end + len(_SPAN_CLOSE[start.group()])
|
||||
out += [text[at:start.start()], "\n".join(" " * len(part) for part in text[start.start():stop].split("\n"))]
|
||||
at = stop
|
||||
return "".join(out) + text[at:]
|
||||
|
||||
|
||||
def _diff(lines: list[str], *, labelled: bool) -> bool:
|
||||
"""整块是 diff：有 - 开头的行也有 + 开头的行，其余是空格开头的上下文行、@@ 开头的块头。diff 里一处改动先列删掉的行、
|
||||
再列加上的行，第一个改动行就是 + 的，是 + / - 开头的正负面要点（有 @@ 块头的照旧算 diff）。信息串写明是文字的块里，
|
||||
+ - 后面空一格的是 Markdown 的列表项，只有紧挨着字写的（-按下单日期统计）才是 diff。"""
|
||||
heads = [line[:1] for line in lines if line.strip()]
|
||||
if not {"+", "-"} <= set(heads) <= {"+", "-", " ", "@"}:
|
||||
return False
|
||||
if labelled and any(line[:2] in ("+ ", "- ") for line in lines):
|
||||
return False
|
||||
return "@" in heads or next(h for h in heads if h in ("+", "-")) == "-"
|
||||
|
||||
|
||||
def _data_like(line: str) -> bool:
|
||||
"""像数据、公式的一行（_DATA_LINE），或者逗号隔开的一行表头、数据（两个以上半角逗号，逗号后面都不空格）。"""
|
||||
return _DATA_LINE.search(line) is not None or (line.count(",") >= 2 and ", " not in line)
|
||||
|
||||
|
||||
def _yaml(lines: list[str], *, labelled: bool) -> bool:
|
||||
"""一段是 YAML：第一行是小写开头的键，其余每一行都是键或者缩进的行（键下面的列表项、子键），注释行不看。写明是文字的
|
||||
块里另要有值为空、往下缩进的键：一行一个「键: 一句话」的（「- note: 华南销售额超过华北」）多半是正文里的标签。"""
|
||||
body = [line for line in lines if line.strip() and not _COMMENT_LINE.match(line)]
|
||||
if not body or not _YAML_KEY.match(body[0]):
|
||||
return False
|
||||
if not all(_YAML_KEY.match(line) or line[:1].isspace() for line in body[1:]):
|
||||
return False
|
||||
return not labelled or any(_YAML_NEST.match(line) for line in body)
|
||||
|
||||
|
||||
def _bare_fence(line: str) -> str:
|
||||
"""整行只写着围栏（``` 或 ~~~，不带信息串）的，返回这串围栏，能给块里再嵌的代码块收尾；别的行返回空串。"""
|
||||
fence = _NESTED_FENCE.match(line)
|
||||
return fence.group(1) if fence and not fence.group(2).strip() else ""
|
||||
|
||||
|
||||
def _code_lines(lines: list[str], raw: list[str], *, labelled: bool) -> list[bool]:
|
||||
"""逐行标出哪些是代码：块里再嵌的代码块（~~~sql），SQL、Python 的语句（从开头那一行到这一段结束，续行也算；前面的
|
||||
注释行本来就不算正文），整段是 YAML 的（_yaml），像数据、公式的行（_data_like）。
|
||||
|
||||
lines 是涂掉块注释、多行字符串之后的行，raw 是原样的行：一段到空行为止，只写着注释的行不算空行；写明是文字的块里，
|
||||
顶格的标题也另起一段。不把整块当代码：正文里另贴一段 SQL、嵌一个 ~~~sql，或者有一行像 YAML 的键，别的行照样按正文认。
|
||||
再嵌的代码块要收得了才算（后面有同一种符号、不比开头短的收尾），开了没收的那一行照普通的行看。"""
|
||||
code = [False] * len(lines)
|
||||
paragraph: list[int] = []
|
||||
fence = ""
|
||||
# 每一行后面最长的收尾围栏（整行只有 ``` 或 ~~~），按符号分开记：开了收不了的围栏不算，照普通的行看，免得后面的
|
||||
# 正文都被当成代码
|
||||
reach = {"`": [0] * (len(lines) + 1), "~": [0] * (len(lines) + 1)}
|
||||
for k in range(len(lines) - 1, -1, -1):
|
||||
for after in reach.values():
|
||||
after[k] = after[k + 1]
|
||||
if bare := _bare_fence(lines[k]):
|
||||
reach[bare[0]][k] = max(reach[bare[0]][k], len(bare))
|
||||
|
||||
def close() -> None:
|
||||
statement = _yaml([lines[k] for k in paragraph], labelled=labelled)
|
||||
for k in paragraph:
|
||||
statement = statement or _CODE_LINE.match(lines[k]) is not None
|
||||
code[k] = statement or _data_like(lines[k])
|
||||
paragraph.clear()
|
||||
|
||||
for k, line in enumerate(lines):
|
||||
if fence:
|
||||
code[k] = True
|
||||
closing = _bare_fence(line)
|
||||
if closing[:1] == fence[0] and len(closing) >= len(fence):
|
||||
fence = ""
|
||||
continue
|
||||
if (opening := _NESTED_FENCE.match(line)) and reach[opening.group(1)[0]][k + 1] >= len(opening.group(1)):
|
||||
close()
|
||||
fence, code[k] = opening.group(1), True
|
||||
continue
|
||||
if not raw[k].strip() or (labelled and _MD_HEADING.match(line)):
|
||||
close()
|
||||
continue
|
||||
paragraph.append(k)
|
||||
close()
|
||||
return code
|
||||
|
||||
|
||||
def _fenced_body(doc: dict[str, Any]) -> bool:
|
||||
"""报告的正文写在代码块里：文档里一句结论都没有，却有写文字的代码块（_PROSE_FENCES），里面写着引用（片段里的
|
||||
[[m:…]] 这类标记、句子上挂的 [[see:…]] 依据，代码里不会这样写）或者在写正文。
|
||||
|
||||
报告节点只在整篇正好是一个代码块时去掉外层围栏（report._unfenced），围栏外有前言、只有开头没有收尾、信息串写成
|
||||
text 的原样交给 compose_doc：整段正文落在代码块里，切不出结论句，裁判没有句子可判、反方审稿记 no_claims，没挂依据
|
||||
的、证据相矛盾的结论句都数出 0 句，出具就成了完整出具。有一句结论的不算：正文没被整个藏起来，结论句照常核对。
|
||||
|
||||
在不在写正文逐行看。先标出哪些行是代码（_code_lines；信息串为空、整块是 diff 的全是代码），代码的行只认去掉注释、
|
||||
字符串之后成句的（_prose_line），中文别名、YAML 的值、公式里的「最高」「占比」这类词都不算。其余的行：信息串写明
|
||||
是文字的按 _written，信息串为空的块里也常是代码，去掉注释、字符串之后再看（_prose_line）。一块里只要有一行在写正文
|
||||
就算，所以宁可把拿不准的行当代码：藏起来的正文总有别的行认得出，贴的一段代码、公式里却一行认错就会记错缺口。"""
|
||||
from app.engine.evidence import iter_units
|
||||
|
||||
units = list(iter_units(doc))
|
||||
if any(unit.get("kind") == "claim" for _, unit in units):
|
||||
return False
|
||||
for block, unit in units:
|
||||
lang = str(block.get("lang") or "").lower()
|
||||
if block.get("type") != "code" or lang not in _PROSE_FENCES:
|
||||
continue
|
||||
segments = [s for s in unit.get("segments") or [] if isinstance(s, dict)]
|
||||
if unit.get("see") or any(s.get("ref") for s in segments):
|
||||
return True
|
||||
# 围栏那两行是挂在块上的结构片段，不是块里写的内容：算进来的话，开头那行 ```markdown 会被当成块里再嵌的代码块，
|
||||
# 整块是不是 diff 也看不准
|
||||
text = "".join(str(s.get("text") or "") for s in segments if s.get("kind") != "structural")
|
||||
raw, lines = text.split("\n"), _blank_spans(text).split("\n")
|
||||
code = [True] * len(lines) if _diff(raw, labelled=bool(lang)) else _code_lines(lines, raw, labelled=bool(lang))
|
||||
if any(_written(line) if lang and not is_code else _prose_line(line, code=is_code)
|
||||
for line, is_code in zip(lines, code)):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _claims(out: dict[str, Any], contract: dict[str, Any], payload: dict[str, Any], *, evidence_on: bool,
|
||||
governed: bool = False, uncited: list[dict[str, Any]], doc: dict[str, Any] | None = None,
|
||||
@@ -507,6 +725,9 @@ def _claims(out: dict[str, Any], contract: dict[str, Any], payload: dict[str, An
|
||||
|
||||
数的是这里重新核对出来的 uncited，不信文档里记的 cites。升级前写的报告（节点产出里没有 claims 这个键，
|
||||
包括跨着升级还没跑完的运行）一律不管，契约写了也只记一条缺口：报告生成于升级前，结论句未核对。
|
||||
|
||||
要核对结论句、正文却写在代码块里（一句结论都没切出来，_fenced_body）的，另记一条缺口：数出 0 句不是「核对过、
|
||||
没问题」。
|
||||
"""
|
||||
written = payload.get("claims") if isinstance(payload, dict) else None
|
||||
# 报告节点升级后才在产出里记 claims（off 也记）：没有这个键就是升级前写的报告
|
||||
@@ -526,6 +747,9 @@ def _claims(out: dict[str, Any], contract: dict[str, Any], payload: dict[str, An
|
||||
out["gaps"].append(f"出具契约要求结论句由模型裁判，但报告「{title}」未开启模型裁判，结论句未裁判")
|
||||
if not asks:
|
||||
return
|
||||
if doc is not None and _fenced_body(doc):
|
||||
# 正文在代码块里，一句结论都没切出来：没挂依据的、证据相矛盾的都是 0 句，不能当成「核对过、没问题」
|
||||
out["gaps"].append(GAP_FENCED_BODY.format(title=title))
|
||||
on_uncited = max(asks, key=_ON_UNCITED.index)
|
||||
if governed and on_uncited == "ignore":
|
||||
on_uncited = "degrade"
|
||||
|
||||
@@ -0,0 +1,426 @@
|
||||
"""出口契约：报告的正文仍写在代码块里、一句结论都没切出来时记出具缺口（nodes/io.py 的 _fenced_body）。
|
||||
|
||||
K5 让报告节点去掉整篇外层的 markdown 围栏（report._unfenced），规格有意不去的写法仍会让正文落在代码块里：围栏外有
|
||||
前言、只有开头没有收尾、信息串写成 text。这时结论句是 0 句：裁判没有句子可判、反方审稿记 no_claims，以前正式运行
|
||||
照样完整出具，读的人以为结论都核对过。
|
||||
|
||||
- 结论句策略为 judge 或 require_citation（报告节点的，或出具契约自己写的）、结论句 0 句、又有信息串为空 / markdown /
|
||||
md / text / plaintext / txt / plain 的代码块，里面写着引用（片段里的引用、只挂依据的 [[see:]]）或者在写正文：记一条
|
||||
缺口，降档。缺口不带核查节点的前缀，界面上归在「校验未全部完成」下（run/decode.ts 的 splitIssuanceGaps）
|
||||
- 逐行看，先标出代码的行：块里再嵌的代码块（~~~sql）、SQL / Python 的语句连同续行、整段是 YAML 的（写明是文字的块里
|
||||
要有值为空的键）、像数据的行（公式、JSON、目录树、流程箭头、函数调用、文件名、逗号隔开的表头），信息串为空、整块是
|
||||
diff 的全是代码。正文里另有一段代码、一行像 YAML 键的标签，不妨碍认出别的行
|
||||
- 在写正文：代码的行只认去掉注释、字符串之后成句的；别的行带句读的中文、成句的英文，或者按正文的规矩会算一句结论
|
||||
（方向词、因果词）；信息串写明是文字的，一行里连着六个以上汉字也算。列表不论用什么标记（- + 1. 1、 一、 （1） · ●
|
||||
→），也不论带不带标记，只看字
|
||||
- 行为不变：有正常结论句的；代码块是 sql、python 这类代码的；写文字的块里只有代码、数据、公式的（注释、字符串里的
|
||||
中文句子不算，别名、公式里的方向词也不算）；信息串为空的块里不带句读、也不像结论的标签和列名；结论句策略为 off 的
|
||||
- 取舍：写成 text、markdown 的 diff，+ - 后面又空一格的，和列表分不开，照列表看；信息串为空的块里先列 - 再列 + 的
|
||||
要点、整段都是「小写键: 值」的，和 diff、YAML 分不开，照代码看
|
||||
|
||||
模型一律是 MockChatModel 剧本，数据源是现造的 SQLite 文件，区域、金额都是虚构的。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
|
||||
import pytest
|
||||
from langchain_core.messages import AIMessage
|
||||
|
||||
from app.engine.evidence import compose_doc
|
||||
from app.engine.nodes.io import GAP_FENCED_BODY, _fenced_body
|
||||
from app.providers.mock_model import MockChatModel
|
||||
from tests.test_report_fenced import ( # noqa: F401 - engine_up、shop 是夹具，导入即可用
|
||||
EAST,
|
||||
SOUTH,
|
||||
SQL,
|
||||
TEXT,
|
||||
Models,
|
||||
audited,
|
||||
chain,
|
||||
engine_up,
|
||||
fenced,
|
||||
finish,
|
||||
logs,
|
||||
node,
|
||||
output_of,
|
||||
report_of,
|
||||
shop,
|
||||
)
|
||||
|
||||
#: 写作者把正文包进代码块、又不是整篇只有一个代码块（报告节点不去围栏）的几种写法
|
||||
PREAMBLE = f"以下是本周的销售报告：\n\n{fenced(TEXT)}"
|
||||
UNCLOSED = f"```markdown\n{TEXT}"
|
||||
AS_TEXT = fenced(TEXT, "text")
|
||||
#: 只用 [[see:]] 挂依据的列表：依据记在句子上（unit 的 see），片段里没有引用；列表项也不带句读
|
||||
SEE_LIST = "## 本周结论\n\n- 华东销售额居各区域之首 [[see:Q1]]\n- 华南销售额超过华北 [[see:Q1]]"
|
||||
#: 不带句读、也没挂依据的列表：中文报告的列表项常这样写
|
||||
BARE_LIST = "## 本周结论\n\n- 华东销售额居各区域之首\n- 华南销售额超过华北"
|
||||
#: 同样两句，换成中文报告常见的几种编号、项目符号，或者干脆不带标记。不包代码块时第二句是一句结论（「超过」），
|
||||
#: 第一句是连接性的话：两种句子都要认得出
|
||||
LINES = ("华东销售额居各区域之首", "华南销售额超过华北")
|
||||
#: 不带标记、一行一句
|
||||
UNMARKED = f"## 本周结论\n\n{LINES[0]}\n\n{LINES[1]}"
|
||||
|
||||
|
||||
def listed(first: str, second: str) -> str:
|
||||
return f"## 本周结论\n\n{first}{LINES[0]}\n{second}{LINES[1]}"
|
||||
|
||||
|
||||
#: + / - 开头的正负面要点：不包代码块时两句都是结论（「增长」「回落」）
|
||||
PROS_CONS = "+ 华东销售额增长明显\n- 华北销售额回落"
|
||||
#: 正文后面另贴一段取数的 SQL：块里再嵌一个 ~~~sql 代码块，或者缩进四格
|
||||
WITH_TILDE = f"{BARE_LIST}\n\n## 取数\n\n~~~sql\nSELECT region, SUM(amount) AS gmv FROM orders GROUP BY region\n~~~"
|
||||
WITH_INDENTED = f"{BARE_LIST}\n\n## 取数\n\n select region, sum(amount) as gmv from orders group by region"
|
||||
#: 列表项前面写着小写的英文标签，看着像 YAML 的键
|
||||
WITH_LABEL = "## 本周结论\n\n- 华东销售额居各区域之首\n- note: 华南销售额超过华北"
|
||||
|
||||
|
||||
HIDDEN = [pytest.param(PREAMBLE, id="preamble"), pytest.param(UNCLOSED, id="unclosed"),
|
||||
pytest.param(AS_TEXT, id="text"), pytest.param(fenced(TEXT, "plaintext"), id="plaintext"),
|
||||
pytest.param(f"以下是本周周报：\n\n{fenced(SEE_LIST)}", id="see-only-list"),
|
||||
pytest.param(f"以下是本周周报：\n\n{fenced(BARE_LIST)}", id="list-without-punctuation"),
|
||||
pytest.param(f"以下是本周周报：\n\n{fenced(listed('1、', '2、'))}", id="digit-and-comma"),
|
||||
pytest.param(f"以下是本周周报：\n\n{fenced(listed('一、', '二、'))}", id="chinese-numeral"),
|
||||
pytest.param(f"以下是本周周报：\n\n{fenced(listed('（1）', '（2）'))}", id="fullwidth-parens"),
|
||||
pytest.param(f"以下是本周周报：\n\n{fenced(listed('· ', '· '))}", id="middle-dot"),
|
||||
pytest.param(f"以下是本周周报：\n\n{fenced(listed('● ', '● '))}", id="black-circle"),
|
||||
pytest.param(f"以下是本周周报：\n\n{fenced(UNMARKED)}", id="lines-without-markers"),
|
||||
# 信息串为空：第二句按正文的规矩是一句结论
|
||||
pytest.param(f"以下是本周周报：\n\n{fenced(listed('1、', '2、'), '')}", id="enumerated-no-info"),
|
||||
# + / - 开头的正负面要点：写明是文字的块里是列表；信息串为空的块里先列 + 的也不是 diff
|
||||
pytest.param(f"以下是本周周报：\n\n{fenced(PROS_CONS)}", id="pros-and-cons"),
|
||||
pytest.param(f"以下是本周周报：\n\n{fenced(PROS_CONS, 'text')}", id="pros-and-cons-text"),
|
||||
pytest.param(f"以下是本周周报：\n\n{fenced(PROS_CONS, '')}", id="pros-and-cons-no-info"),
|
||||
# 正文里另有几行代码、像 YAML 键的标签：只有那几行当代码看，正文照样认得出
|
||||
pytest.param(f"以下是本周周报：\n\n{fenced(WITH_TILDE)}", id="tilde-block-inside"),
|
||||
pytest.param(f"以下是本周周报：\n\n{fenced(WITH_INDENTED)}", id="indented-sql-inside"),
|
||||
pytest.param(f"以下是本周周报：\n\n{fenced(WITH_LABEL)}", id="lowercase-label")]
|
||||
|
||||
|
||||
def doc_of(raw: str) -> dict:
|
||||
return compose_doc(raw, {}, node_id="write")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 认得出哪些代码块里写的是正文
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.parametrize("raw", [
|
||||
pytest.param("以下是本周的销售报告：\n\n```markdown\n华东销售额 1500，居各区域之首。\n```", id="preamble"),
|
||||
pytest.param("```markdown\n## 本周概览\n\n华东销售额 1500，居各区域之首。", id="unclosed"),
|
||||
pytest.param("```text\n华东销售额 1500，居各区域之首。\n```", id="text"),
|
||||
pytest.param("```\n华东销售额 1500，居各区域之首。\n```", id="no-info"),
|
||||
pytest.param("```MD\n华东销售额 1500，居各区域之首。\n```", id="upper-md"),
|
||||
pytest.param("```markdown\nEast region sales rose sharply this week.\n```", id="english"),
|
||||
pytest.param("```markdown\nEast region's sales rose sharply this week.\n```", id="english-apostrophe"),
|
||||
# 正文里用半角引号括起一个词：去掉代码字符串的那一步不能把句子吃掉
|
||||
pytest.param('```markdown\n华东的"销售额"居各区域之首，华南次之。\n```', id="quoted-word"),
|
||||
# 纯文字的另几种写法
|
||||
pytest.param("```plaintext\n华东销售额 1500，居各区域之首。\n```", id="plaintext"),
|
||||
pytest.param("```txt\n华东销售额 1500，居各区域之首。\n```", id="txt"),
|
||||
pytest.param("```Plain\n华东销售额 1500，居各区域之首。\n```", id="plain"),
|
||||
# 只有列表、没有句读，可写着报告的引用：代码里不会这样写
|
||||
pytest.param("以下是周报：\n\n```markdown\n- 华东销售额 [[m:gmv]]\n- 华南销售额 [[m:gmv_s]]\n```", id="markers"),
|
||||
# 只挂了依据（[[see:]] 记在句子上，片段里没有引用），列表项短得不成句也认
|
||||
pytest.param("以下是周报：\n\n```markdown\n- East [[see:m:gmv]]\n- South [[see:m:gmv]]\n```", id="see-only"),
|
||||
# 不带句读的列表项：一行里连着六个以上汉字，或者按正文的规矩是一句结论
|
||||
pytest.param("以下是周报：\n\n```markdown\n## 本周结论\n\n- 华东销售额居各区域之首\n- 华南销售额超过华北\n```",
|
||||
id="bullets"),
|
||||
pytest.param("```text\n1. 华东销售额居各区域之首\n2) 华南销售额超过华北\n```", id="numbered"),
|
||||
pytest.param("```\n* 华东销售额居各区域之首\n+ 华南销售额超过华北\n• 华北销售额持平\n```", id="bare-bullets"),
|
||||
# 中文报告常见的编号、项目符号，或者不带标记：不看标记，只看字
|
||||
pytest.param("以下是周报：\n\n```markdown\n1、华东销售额居各区域之首\n2、华南销售额超过华北\n```", id="digit-and-comma"),
|
||||
pytest.param("```markdown\n一、华东销售额居各区域之首\n二、华南销售额超过华北\n```", id="chinese-numeral"),
|
||||
pytest.param("```text\n（1）华东销售额居各区域之首\n(2) 华南销售额超过华北\n```", id="parens"),
|
||||
pytest.param("```markdown\n· 华东销售额居各区域之首\n● 华南销售额超过华北\n▪ 华北销售额持平\n```", id="dots-circles-squares"),
|
||||
pytest.param("```markdown\n华东销售额居各区域之首\n\n华南销售额超过华北\n```", id="no-markers"),
|
||||
pytest.param("```markdown\n> 华东销售额居各区域之首\n```", id="blockquote"),
|
||||
pytest.param("```markdown\n| 区域 | 结论 |\n|---|---|\n| 华东 | 销售额居各区域之首 |\n```", id="table-row"),
|
||||
pytest.param("```markdown\n**华东**销售额居各区域之首\n```", id="bold"),
|
||||
# 写明是文字的块：一行里连着六个以上汉字就是在写中文，不像结论的也算
|
||||
pytest.param("```markdown\n华东销售额居各区域之首\n```", id="six-or-more-in-a-row"),
|
||||
# 不到六个字的短句：按正文的规矩是一句结论（方向词）就算
|
||||
pytest.param("```text\n- 华北回落\n- 华南领先\n```", id="short-conclusions"),
|
||||
# 写明是文字的块不当代码看：# -- 和半角引号都是正文里的符号
|
||||
pytest.param("```markdown\n华东排名 #1，销售额居各区域之首。\n```", id="hash-in-prose"),
|
||||
pytest.param('```markdown\n负责人说"销售额创新高，下周继续推进。"\n```', id="quoted-sentence"),
|
||||
pytest.param("```markdown\n华东领跑 -- 华南次之，华北第三。\n```", id="dash-in-prose"),
|
||||
# 信息串为空、看不出是代码：按正文的规矩会算结论的行也算
|
||||
pytest.param("```\n1、华东销售额居各区域之首\n2、华南销售额超过华北\n```", id="no-info-enumerated"),
|
||||
pytest.param("```\n华北销售额回落\n```", id="no-info-conclusion"),
|
||||
# 首字母大写的「Note:」是正文里的标签，不是 YAML 的键
|
||||
pytest.param("```markdown\nNote: 以下为初步结论\n- 华南销售额超过华北\n```", id="capitalised-label"),
|
||||
# + / - 开头的正负面要点：写明是文字的块里是列表，不论哪个在前；信息串为空的块里先列 + 的不是 diff（diff 里一处
|
||||
# 改动先列删掉的行），+ 下面缩进的子项也一样
|
||||
pytest.param(f"```markdown\n{PROS_CONS}\n```", id="plus-minus-markdown"),
|
||||
pytest.param("```text\n- 华北销售额回落\n+ 华东销售额增长明显\n```", id="minus-plus-text"),
|
||||
# 取舍：写成 text、markdown 的 diff，+ - 后面又空一格，和列表分不开，照列表看（diff 该写 diff 信息串）
|
||||
pytest.param("```text\n- 按下单日期统计\n+ 按支付日期统计\n```", id="spaced-diff-labelled-text"),
|
||||
pytest.param(f"```\n{PROS_CONS}\n```", id="plus-minus-no-info"),
|
||||
pytest.param("```\n+ 华东销售额增长明显\n - 主要来自新店\n- 华北销售额回落\n```", id="plus-minus-nested-no-info"),
|
||||
# 正文里另有几行代码：块里再嵌的 ~~~sql、缩进的 SQL 只是那几行，不把整块都当代码
|
||||
pytest.param(f"```markdown\n{WITH_TILDE}\n```", id="tilde-block-inside"),
|
||||
pytest.param(f"```\n{WITH_TILDE}\n```", id="tilde-block-inside-no-info"),
|
||||
pytest.param(f"```markdown\n{WITH_INDENTED}\n```", id="indented-sql-inside"),
|
||||
pytest.param(f"```\n{WITH_INDENTED}\n```", id="indented-sql-inside-no-info"),
|
||||
pytest.param("```text\n- 华南销售额超过华北\n\nselect region from orders\n```", id="sql-in-another-paragraph"),
|
||||
pytest.param("```markdown\n~~~sql\nSELECT region FROM orders\n~~~\n\n- 华南销售额超过华北\n```",
|
||||
id="prose-after-nested-block"),
|
||||
# 块里再嵌的代码块开了没收：不当围栏，后面的正文照样认
|
||||
pytest.param("```markdown\n## 取数\n\n~~~sql\nSELECT region FROM orders\n\n## 本周结论\n\n- 华南销售额超过华北\n```",
|
||||
id="nested-block-left-open"),
|
||||
pytest.param("```markdown\n~~~~sql\nSELECT region FROM orders\n~~~\n\n- 华南销售额超过华北\n```", id="nested-close-too-short"),
|
||||
# 收尾的围栏比开头长也收得了
|
||||
pytest.param("```markdown\n~~~~sql\nSELECT region FROM orders\n~~~~~\n\n- 华南销售额超过华北\n```", id="nested-close-longer"),
|
||||
# 写明是文字的块里，顶格的标题另起一段：上一段的 SQL 管不到标题下面的列表
|
||||
pytest.param("```markdown\n## 取数\nselect region from orders\n## 本周结论\n- 华南销售额超过华北\n```", id="heading-ends-sql"),
|
||||
# 行首的箭头是列表的记号，不是连起流程的箭头
|
||||
pytest.param("```markdown\n→ 华东销售额居各区域之首\n→ 华南销售额超过华北\n```", id="arrow-bullets"),
|
||||
# 小写英文键开头的行：一段里有不缩进、又不是键的行，不是 YAML；写明是文字的块里，没有值为空的键（下面缩进着
|
||||
# 列表、子键）的「键: 一句话」也是正文里的标签
|
||||
pytest.param(f"```markdown\n{WITH_LABEL}\n```", id="lowercase-label-after"),
|
||||
pytest.param("```\n- note: 华南销售额超过华北\n- 华东销售额居各区域之首\n```", id="lowercase-label-first-no-info"),
|
||||
pytest.param("```text\n- top: 华东销售额居各区域之首\n- note: 华南销售额超过华北\n```", id="lowercase-labels-only"),
|
||||
pytest.param("```markdown\nsummary: 华南销售额超过华北\n```", id="key-and-conclusion"),
|
||||
# 首字母大写的英文词开头是正文里的英文，不是 SQL 的关键字
|
||||
pytest.param("```markdown\nFrom 数据看 华南销售额超过华北\n```", id="capitalised-keyword"),
|
||||
pytest.param("```\nUpdate 华南销售额超过华北\n```", id="capitalised-keyword-no-info"),
|
||||
# def、class 不写成定义的样子，是正文里的词（A 类门店）
|
||||
pytest.param("```markdown\nclass A 门店销售额超过 class B 门店\n```", id="class-of-stores"),
|
||||
])
|
||||
def test_a_report_body_in_a_prose_code_block_is_recognised(raw):
|
||||
doc = doc_of(raw)
|
||||
assert doc["stats"]["claims"] == 0, "背景：正文在代码块里，一句结论都切不出来"
|
||||
assert _fenced_body(doc) is True
|
||||
|
||||
|
||||
@pytest.mark.parametrize("raw", [
|
||||
# 代码：信息串写明了语言
|
||||
pytest.param("```sql\nSELECT region, SUM(amount) AS gmv FROM orders WHERE amount > 100 GROUP BY region\n```",
|
||||
id="sql"),
|
||||
pytest.param("```python\nprint('华东销售额最高，华南次之。')\n```", id="python"),
|
||||
pytest.param('```json\n{"区域": "华东，华南"}\n```', id="json"),
|
||||
# 信息串为空、里面是代码：没有成句的文字；代码注释里的中文句子不算
|
||||
pytest.param("```\nSELECT region, SUM(amount) FROM orders GROUP BY region\n```", id="bare-sql"),
|
||||
pytest.param("```\n-- 按支付日统计，排除退款。\nSELECT SUM(amount) FROM orders\n```", id="sql-comment"),
|
||||
pytest.param("```\n/* 按支付日统计，排除退款。 */\nSELECT SUM(amount) FROM orders\n```", id="block-comment"),
|
||||
pytest.param("```\n/*\n * 按支付日统计，排除退款。\n 已退款的单不计入：见口径说明。\n */\nSELECT SUM(amount) FROM orders\n```",
|
||||
id="multi-line-comment"),
|
||||
# 行尾注释（-- // #）、# 开头的注释行，里面写着中文句子也是代码的一部分
|
||||
pytest.param("```\nSELECT region, SUM(amount) AS gmv FROM orders GROUP BY region -- 按区域汇总，排除退款\n```",
|
||||
id="trailing-sql-comment"),
|
||||
pytest.param("```\n# 按区域汇总，排除退款。\ntotal = sum(rows) # 单位：元，不含税\n```", id="hash-comments"),
|
||||
pytest.param("```\nconst total = sum(rows); // 按区域汇总，排除退款\n```", id="trailing-slash-comment"),
|
||||
# 字符串字面量里的中文标点是数据（带 f、N 这类前缀的也是）
|
||||
pytest.param('```\nprint("销售额：", total)\nprint(f"订单数：{n}，客单价：{avg}")\n```', id="string-literals"),
|
||||
pytest.param("```\nSELECT * FROM orders WHERE note = '已退款，不计入：测试' OR memo = N'作废，重开'\n```",
|
||||
id="sql-strings"),
|
||||
# 信息串为空、看得出是 SQL 的块：以 * + - 开头的是运算的续行，后面的中文列名不算正文
|
||||
pytest.param("```\nSELECT 单价\n * 数量 AS 金额\n + 运费 AS 合计\n - 折扣金额 AS 实付\nFROM 订单\n```",
|
||||
id="sql-continuation"),
|
||||
pytest.param("```text\n区域 销售额\n华东 1500\n华南 900\n```", id="table-of-numbers"),
|
||||
# 写明是文字的块里，表格的列名、门店名这类标签不到六个字，又不像结论
|
||||
pytest.param("```text\n门店名称 累计销售额\n华东一店 1500\n```", id="labels-under-six"),
|
||||
# 标题不算正文：不包代码块时，标题也从不算结论句
|
||||
pytest.param("```markdown\n## 华南销售额超过华北\n```", id="heading-only"),
|
||||
# 信息串为空、不带句读的行：不像结论的不算（列名、改了的口径）
|
||||
pytest.param("## 口径说明\n\n本周报告按支付日期统计。\n\n```\ncolumns:\n - 订单编号\n - 支付日期\n```", id="yaml-list"),
|
||||
pytest.param("## 口径说明\n\n本周报告按支付日期统计。\n\n```\n- 按下单日期统计\n+ 按支付日期统计\n```", id="diff"),
|
||||
pytest.param("```\n- 订单编号\n- 支付日期\n```", id="labels"),
|
||||
# 信息串为空、看得出是代码（SQL、YAML 的键、diff）：值里、别名里有方向词也不是结论
|
||||
pytest.param("```\nSELECT MAX(amount) AS 最高金额, MIN(amount) AS 最低金额 FROM orders\n```", id="sql-aliases"),
|
||||
pytest.param("```\nmetrics:\n - 退款占比\n - 销售额增长率\n```", id="yaml-with-conclusion-words"),
|
||||
pytest.param("```\n- name: 退款占比\n unit: 百分比\n```", id="yaml-mappings"),
|
||||
# 取舍：信息串为空、整段都是「小写键: 值」的照 YAML 看，值里写着结论也一样（写明是文字的块里照正文看）
|
||||
pytest.param("```\n- top: 华东销售额居各区域之首\n- note: 华南销售额超过华北\n```", id="all-keys-no-info"),
|
||||
pytest.param("```\n- 退款率下降\n+ 退款率上升\n```", id="diff-with-conclusion-words"),
|
||||
pytest.param("```\n@@ -1,2 +1,2 @@\n 口径说明\n-销售额按下单日期统计后下降\n+销售额按支付日期统计后上升\n```", id="diff-hunk"),
|
||||
# 写成 text、markdown 的 diff：+ - 紧挨着字写，不是 Markdown 的列表项
|
||||
pytest.param("```text\n-退款率下降时提醒\n+退款率上升时提醒\n```", id="diff-labelled-text"),
|
||||
pytest.param("```markdown\n@@ -1 +1 @@\n-销售额按下单日期统计后下降\n+销售额按支付日期统计后上升\n```",
|
||||
id="diff-hunk-labelled-markdown"),
|
||||
# 有 @@ 块头的，先列加上的行也是 diff
|
||||
pytest.param("```\n@@ -1,2 +1,2 @@\n+销售额按支付日期统计后上升\n-销售额按下单日期统计后下降\n```", id="diff-hunk-plus-first"),
|
||||
# 代码的行里，-- 紧挨着前面的字也是注释
|
||||
pytest.param("```\nSELECT SUM(amount)--按区域汇总，排除退款\nFROM orders\n```", id="sql-comment-without-space"),
|
||||
# 信息串写成 text、markdown，里面却看得出是代码：照代码看，注释里的句子、别名里的方向词不算
|
||||
pytest.param("```text\nSELECT region AS 区域, MAX(amount) AS 最高金额 FROM orders -- 按区域汇总，排除退款\n```",
|
||||
id="sql-labelled-text"),
|
||||
pytest.param("```markdown\nmetrics:\n - 退款占比\n - 销售额增长率\n```", id="yaml-labelled-markdown"),
|
||||
pytest.param("```text\n# 指标\nmetrics:\n # 核心指标\n - 退款占比\n```", id="yaml-with-comments"),
|
||||
pytest.param("```\n# 核心指标\nmetrics:\n - 退款占比\n```", id="yaml-with-comments-no-info"),
|
||||
# 写成 text 的 SQL 跨好几行：以语句开头的一段连同续行都是代码，开头的注释行也是；块注释、多行字符串整段不算，
|
||||
# 只写着块注释的一行也不把一段断开
|
||||
pytest.param("```text\nSELECT region,\n amount AS 销售额增长率\n FROM orders\n```", id="sql-continuation-text"),
|
||||
pytest.param("```text\nSELECT region,\n/* 别名 */\n amount AS 销售额增长率\n FROM orders\n```",
|
||||
id="sql-with-block-comment-text"),
|
||||
pytest.param("```text\n-- 按区域汇总，排除退款。\nSELECT region, amount AS 最高金额 FROM orders\n```", id="sql-leading-comment"),
|
||||
pytest.param('```\ndef summary():\n """\n 按支付日统计，排除退款。\n """\n return 1\n```', id="docstring"),
|
||||
# 像数据、公式的行（等号、JSON、画目录树的制表符、箭头连起的流程、函数调用、文件名、逗号隔开的表头）照代码看：
|
||||
# 去掉字符串之后不成句的不算正文，结论词、六字连写只看写正文的行
|
||||
pytest.param("```\n退款占比 = 退款金额 / 销售额\n```", id="formula-no-info"),
|
||||
pytest.param("```text\n销售额增长率 = (本期销售额 - 上期销售额) / 上期销售额\n```", id="formula-text"),
|
||||
pytest.param('```text\n{"区域": "华东", "说明": "按支付日期统计"}\n```', id="json-text"),
|
||||
pytest.param('```markdown\n "说明": "按支付日期统计，排除退款",\n```', id="json-member"),
|
||||
pytest.param('```text\n["订单编号", "支付日期", "销售额增长率"]\n```', id="json-array"),
|
||||
pytest.param("```text\n报表/\n├── 华东门店销售周报.xlsx\n└── 华南门店销售周报.xlsx\n```", id="tree-text"),
|
||||
pytest.param("```text\n周报\n├── 华东门店销售周报\n│ └── 第四周销售明细\n└── 华南门店销售周报\n```", id="tree-of-folders"),
|
||||
pytest.param("```text\n上传表格导入 → 字段映射校验 → 生成周报\n```", id="flow-text"),
|
||||
pytest.param("```text\nprint('导入完成，请查收')\n```", id="call-with-string-text"),
|
||||
pytest.param("```markdown\nrows.groupby(区域).agg(销售额增长率)\n```", id="call-text"),
|
||||
pytest.param("```markdown\n华东门店销售周报.xlsx\n华南门店销售周报.xlsx\n```", id="file-names"),
|
||||
pytest.param("```text\n订单编号,支付日期,销售额增长率\n```", id="csv-header"),
|
||||
# 块里再嵌的代码块照代码看：注释里的句子、别名里的结论词、列着指标名的 YAML 都不算
|
||||
pytest.param("```markdown\n## 取数\n\n~~~sql\nSELECT MAX(amount) AS 最高金额 FROM orders -- 按区域汇总，排除退款\n~~~\n```",
|
||||
id="tilde-block-only"),
|
||||
pytest.param("```markdown\n## 口径\n\n~~~yaml\n- 退款占比\n- 销售额增长率\n~~~\n```", id="nested-yaml-block"),
|
||||
pytest.param("```markdown\n~~~~\n- 退款占比\n- 销售额增长率\n~~~~~\n```", id="nested-yaml-longer-close"),
|
||||
# 开了没收的那一行照普通的行看，里面的 SQL 照样认得出
|
||||
pytest.param("```markdown\n## 取数\n\n~~~sql\nSELECT MAX(amount) AS 最高金额 FROM orders\n```", id="nested-sql-left-open"),
|
||||
# 有正常的结论句：正文没被藏起来
|
||||
pytest.param("华东销售额超过华南。\n\n```markdown\n华东销售额 1500，居各区域之首。\n```", id="claims-outside"),
|
||||
# 没有代码块
|
||||
pytest.param("以下是本周的销售报告。", id="no-code"),
|
||||
])
|
||||
def test_code_or_a_report_with_claims_is_not_taken_for_a_hidden_body(raw):
|
||||
assert _fenced_body(doc_of(raw)) is False
|
||||
|
||||
|
||||
@pytest.mark.parametrize("lang", ["", "markdown"])
|
||||
@pytest.mark.parametrize("line", [
|
||||
pytest.param("alpha beta " * 4000, id="words"),
|
||||
# 逐行认代码、认数据的几条规则：一长串字母后面没有括号、冒号，一长串空格、横线，一长串汉字、逗号、引号
|
||||
pytest.param("a" * 40000, id="one-long-word"),
|
||||
pytest.param(" " * 20000 + "-" + " " * 20000 + "a", id="spaces-and-a-dash"),
|
||||
pytest.param("华" * 40000, id="cjk"),
|
||||
pytest.param("a," * 20000, id="commas"),
|
||||
pytest.param(' "a' * 13000, id="open-quotes"),
|
||||
pytest.param('"""' * 13000, id="triple-quotes"),
|
||||
# 好多行围栏：收得了的、开了没收的
|
||||
pytest.param("~~~\n" * 13000, id="bare-fences"),
|
||||
pytest.param("~~~sql\n" * 13000, id="fences-left-open"),
|
||||
])
|
||||
def test_a_long_line_without_a_full_stop_takes_linear_time(lang, line):
|
||||
"""很长、不带句末标点的一行：逐串数单词、逐行认代码，耗时随行长线性增长。按正则回溯认英文句子时，四万多字的一行要
|
||||
十几秒（行长翻倍、耗时翻四倍），出口复核在事件循环上同步执行，整个服务都会卡住。"""
|
||||
doc = {"blocks": [{"type": "code", "lang": lang, "units": [
|
||||
{"kind": "code", "see": [], "segments": [{"kind": "text", "text": line}]}]}]}
|
||||
started = time.perf_counter()
|
||||
_fenced_body(doc)
|
||||
assert time.perf_counter() - started < 1.0
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 端到端：正式运行（裁判 + 反方审稿 + 出具契约）
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
def gap(title: str = "写周报") -> str:
|
||||
return GAP_FENCED_BODY.format(title=title)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("raw", HIDDEN)
|
||||
async def test_a_body_left_in_a_code_block_degrades_the_formal_issuance(monkeypatch, shop, raw):
|
||||
"""正文落在代码块里：裁判没有句子可判（不调用）、反方审稿记 no_claims，出口契约不再完整出具，记一条缺口说清原因。"""
|
||||
models = Models(monkeypatch, shop, raw)
|
||||
row = await finish(audited(shop), run_class="formal")
|
||||
assert row.status == "succeeded", row.error
|
||||
_, doc = await report_of(row.id)
|
||||
assert doc["source"] == raw and doc["stats"]["claims"] == 0, "背景：报告节点不去这几种围栏"
|
||||
assert "code" in [b["type"] for b in doc["blocks"]]
|
||||
assert models.judges == [] and models.reviewed == []
|
||||
assert [d["reason"] for d in await logs(row.id, "challenge_skipped")] == ["no_claims"]
|
||||
issuance = row.output["_issuance"]
|
||||
assert issuance["tier"] == "degraded", issuance
|
||||
assert issuance["gaps"] == [gap()], issuance["gaps"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("raw", [
|
||||
pytest.param(TEXT, id="plain"),
|
||||
pytest.param(fenced(TEXT), id="whole-report-unwrapped"),
|
||||
pytest.param(f"华东销售额 [[v:Q1.r0.销售额]]，居各区域之首。[[see:Q1]]\n\n{SQL}\n\n华南销售额超过华北。[[see:Q1]]",
|
||||
id="sql-block"),
|
||||
])
|
||||
async def test_a_report_with_claims_is_issued_as_before(monkeypatch, shop, raw):
|
||||
"""有正常的结论句（整篇外层围栏已由报告节点去掉的也是）、代码块是 SQL 的：判过、审过，照常完整出具，不记这条缺口。"""
|
||||
models = Models(monkeypatch, shop, raw)
|
||||
row = await finish(audited(shop), run_class="formal")
|
||||
assert row.status == "succeeded", row.error
|
||||
assert sorted(models.reviewed) == sorted([EAST, SOUTH])
|
||||
issuance = row.output["_issuance"]
|
||||
assert issuance["tier"] == "formal" and issuance["gaps"] == [], issuance
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 结论句策略：require_citation 同样记，off 不管
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
KPI = "{'gmv': 45678.5, 'orders': 1234}"
|
||||
METRICS = [{"id": "gmv", "name": "销售额", "unit": "元", "decimals": 1, "expression": "vars.kpi.gmv"}]
|
||||
#: 前言 + 代码块包住正文，正文里的引用都解析得了
|
||||
CITED = "以下是本周周报：\n\n```markdown\n本周销售额 [[m:gmv]]，较上周有所增长。[[see:m:gmv]]\n```"
|
||||
|
||||
|
||||
def weekly(*, claims: str | None = None, contract_claims: str | None = None) -> dict:
|
||||
"""输入 → 取数 → 口径卡 → 报告（claims 是报告节点的结论句策略）→ 成果（引用模式的出具契约）。"""
|
||||
return chain(
|
||||
node("start", "input"),
|
||||
node("fetch", "transform", mode="expression", expression=KPI, assign_to="kpi"),
|
||||
node("caliber", "metrics", caliber="周报口径", caliber_version="v1", metrics=METRICS),
|
||||
node("write", "report", "写周报", instructions="写周报", max_repairs=0,
|
||||
**({"claims": claims} if claims else {})),
|
||||
node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}],
|
||||
contract={"report_from": "write", "metrics_from": ["caliber"],
|
||||
**({"claims": contract_claims} if contract_claims else {})}),
|
||||
)
|
||||
|
||||
|
||||
def writer(monkeypatch, text: str) -> None:
|
||||
monkeypatch.setattr(MockChatModel, "_decide", lambda self, messages: AIMessage(content=text))
|
||||
|
||||
|
||||
@pytest.mark.parametrize("claims, contract_claims", [
|
||||
pytest.param("require_citation", None, id="report-node"),
|
||||
pytest.param(None, "require_citation", id="contract"),
|
||||
])
|
||||
async def test_require_citation_records_the_gap_too(monkeypatch, claims, contract_claims):
|
||||
"""要求结论句挂依据（报告节点或出具契约写的）：同样数不出结论句，记同一条缺口。"""
|
||||
writer(monkeypatch, CITED)
|
||||
row = await finish(weekly(claims=claims, contract_claims=contract_claims))
|
||||
assert row.status == "succeeded", row.error
|
||||
issuance = row.output["_issuance"]
|
||||
assert issuance["tier"] == "degraded" and issuance["gaps"] == [gap()], issuance
|
||||
assert issuance["claims"]["uncited_claims"] == 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize("info, code", [
|
||||
pytest.param("", "SELECT region, SUM(amount) AS gmv FROM orders GROUP BY region -- 按区域汇总，排除退款",
|
||||
id="sql-trailing-comment"),
|
||||
pytest.param("", "columns:\n - 订单编号\n - 支付日期", id="yaml-list"),
|
||||
pytest.param("", "- 按下单日期统计\n+ 按支付日期统计", id="diff"),
|
||||
pytest.param("", "退款占比 = 退款金额 / 销售额", id="formula"),
|
||||
pytest.param("text", "销售额增长率 = (本期销售额 - 上期销售额) / 上期销售额", id="formula-text"),
|
||||
pytest.param("text", '{"区域": "华东", "说明": "按支付日期统计"}', id="json-text"),
|
||||
pytest.param("text", "报表/\n├── 华东门店销售周报.xlsx\n└── 华南门店销售周报.xlsx", id="tree-text"),
|
||||
])
|
||||
async def test_code_in_a_report_without_claims_is_issued_as_before(monkeypatch, info, code):
|
||||
"""一句结论都没有的报告（只有口径说明）贴了一段代码、公式或数据：行尾注释里写着中文句子的 SQL、列着中文列名的
|
||||
YAML、改口径的 diff（这三种不带信息串），带「占比」「增长率」的公式，写成 text 的 JSON、目录树。那不是藏起来的
|
||||
正文，照常完整出具，不记这条缺口。"""
|
||||
writer(monkeypatch, f"## 口径说明\n\n本周报告按支付日期统计。\n\n```{info}\n{code}\n```")
|
||||
row = await finish(weekly(claims="require_citation"))
|
||||
assert row.status == "succeeded", row.error
|
||||
issuance = row.output["_issuance"]
|
||||
assert issuance["tier"] == "formal" and issuance["gaps"] == [], issuance
|
||||
|
||||
|
||||
async def test_without_a_claims_policy_nothing_changes(monkeypatch):
|
||||
"""结论句策略为 off：出口不核对结论句，同样的报告照旧完整出具（数字都在代码块里、也都有出处）。"""
|
||||
writer(monkeypatch, CITED)
|
||||
row = await finish(weekly())
|
||||
assert row.status == "succeeded", row.error
|
||||
issuance = row.output["_issuance"]
|
||||
assert issuance["tier"] == "formal" and issuance["gaps"] == [], issuance
|
||||
assert "claims" not in issuance
|
||||
@@ -4,8 +4,9 @@
|
||||
条件，聚合只差去不去重、被采纳的一边去重 → 去重规则；按被采纳那一边的写法记
|
||||
- 业务日期只在两边的列都在表结构里声明为带日期的类型（DATE、DATETIME、TIMESTAMP，带精度、带时区的写法也算）时才写：
|
||||
文本、整数（Unix 秒、20260901 这类日期键）、只有时刻的 TIME、表结构里查不到、没声明类型（SQLite 探出来是 NULL）的
|
||||
不写，原因说清是哪一列、哪一边（未被采纳的那一边另说清为什么也要看它）；上传表格建的数据源说清日期是按文字保存的；
|
||||
数据源还没有可用的表结构时说表结构不可用
|
||||
不写，原因说清是哪一列、哪一边（未被采纳的那一边另说清为什么也要看它）；上传表格建的数据源，按文字保存的列看导入时
|
||||
配方里认作日期的列（候选的 upload_dates）：是的算日期列，不是的、读不出配方的说清日期是按文字保存的；数据源还没有
|
||||
可用的表结构时说表结构不可用
|
||||
- 差了不止一处、有兜底说明、两条 SQL 规范化后一致、差在别的方面、被采纳的一边没有可记的写法、只有一边按时间筛：不写，
|
||||
照实说原因。同一列上的时间范围不同、跨两张表的条件，差异说明只给兜底，同样不写；说明以后真把跨两张表的条件说成一条
|
||||
筛选条件，derive 自己那道「只涉及一张表」的检查也拦得下
|
||||
@@ -367,6 +368,99 @@ def test_an_uploaded_table_says_its_dates_are_stored_as_text():
|
||||
candidate(BY_ORDER_DATE, RIGHT, "b"), texts)[1]
|
||||
|
||||
|
||||
#: 上传表格的列按文字保存、配方里又没认作日期时的原因（同上一个用例）
|
||||
STORED_AS_TEXT = "来自上传的表格，表格里的日期一律按文字保存，无法确认存的是日期"
|
||||
|
||||
|
||||
def uploaded(original: str, blind: str, choice: str, dates: dict | None) -> dict:
|
||||
"""上传表格建的数据源的候选：upload_dates 是节点从当前版本的配方读出的、导入时认作日期的列（{表: [列]}）。"""
|
||||
return {**candidate(original, blind, choice), "uploaded": True,
|
||||
**({"upload_dates": dates} if dates is not None else {})}
|
||||
|
||||
|
||||
def test_an_uploaded_column_the_import_took_as_a_date_is_a_date():
|
||||
"""上传表格建的数据源，表结构里的日期列总是 TEXT（导入时一律按文字保存），可导入时配方里这两列的类型是「日期」（候选的
|
||||
upload_dates）：照常记成业务日期，采纳哪一边记哪一列，预告是同一套判断。"""
|
||||
texts = schema_with(order_date="TEXT", paid_date="TEXT")
|
||||
both = {"orders": ["order_date", "paid_date"]}
|
||||
for choice, column in (("a", "order_date"), ("b", "paid_date")):
|
||||
fact, why = derive(uploaded(BY_ORDER_DATE, RIGHT, choice, both), texts)
|
||||
assert why is None, why
|
||||
assert (fact.table, fact.item, fact.value, fact.choice) == \
|
||||
("orders", "business_date", {"column": column}, choice)
|
||||
said = preview(uploaded(BY_ORDER_DATE, RIGHT, choice, both), texts, {})
|
||||
assert said[choice] == {"write": True, "source": "shop", "table": "orders", "item": "business_date",
|
||||
"value": {"column": column}}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("dates, column, owner", [
|
||||
# 配方读不出来（简单导入没有配方、配方记录坏了）：两列都照旧说按文字保存
|
||||
(None, "orders.paid_date", "b"),
|
||||
({}, "orders.paid_date", "b"),
|
||||
# 配方里只有一列是日期，另一列是文字：没认作日期的那一列照旧说
|
||||
({"orders": ["paid_date"]}, "orders.order_date", "a"),
|
||||
({"orders": ["order_date"]}, "orders.paid_date", "b"),
|
||||
# 认作日期的是别的表的同名列：对不上这一列
|
||||
({"customers": ["order_date", "paid_date"]}, "orders.paid_date", "b"),
|
||||
])
|
||||
def test_an_uploaded_column_the_import_did_not_take_as_a_date_keeps_the_reason(dates, column, owner):
|
||||
"""配方里不是日期、或者读不出配方的列：照现在的原因不写（说按文字保存，不说「类型是 TEXT」），被采纳的一边先说；
|
||||
被采纳的一边是日期、未被采纳的不是的，说未被采纳的那一边、另说清为什么也要看它。"""
|
||||
texts = schema_with(order_date="TEXT", paid_date="TEXT")
|
||||
fact, why = derive(uploaded(BY_ORDER_DATE, RIGHT, "b", dates), texts)
|
||||
assert fact is None
|
||||
if owner == "b":
|
||||
assert why == f"指标「销售额」被采纳的盲算按时间筛的列「{column}」{STORED_AS_TEXT}，不写入数据目录", why
|
||||
else:
|
||||
assert why == (f"指标「销售额」未被采纳的原流程按时间筛的列「{column}」{STORED_AS_TEXT}：{BOTH_DATES}，"
|
||||
"不写入数据目录"), why
|
||||
assert preview(uploaded(BY_ORDER_DATE, RIGHT, "b", dates), texts, {})["b"] == {"write": False, "reason": why}
|
||||
|
||||
|
||||
def test_the_import_dates_only_speak_for_text_columns_of_an_uploaded_source():
|
||||
"""upload_dates 只回答「按文字保存的这一列，导入时是不是日期」：
|
||||
- 不是上传源（手工登记的库）的文本列照旧说类型，候选里带着 upload_dates 也不认
|
||||
- 上传源里声明成别的类型的列（整数）照旧说类型；只有时刻的仍不算日期"""
|
||||
texts = schema_with(order_date="TEXT", paid_date="TEXT")
|
||||
both = {"orders": ["order_date", "paid_date"]}
|
||||
fact, why = derive({**candidate(BY_ORDER_DATE, RIGHT, "b"), "upload_dates": both}, texts)
|
||||
assert fact is None and "「orders.paid_date」在表结构里的类型是 TEXT，无法确认存的是日期" in why
|
||||
fact, why = derive(uploaded(BY_ORDER_DATE, RIGHT, "b", both), schema_with(order_date="INTEGER", paid_date="TEXT"))
|
||||
assert fact is None and "未被采纳的原流程按时间筛的列「orders.order_date」在表结构里的类型是 INTEGER" in why
|
||||
fact, why = derive(uploaded(BY_ORDER_DATE, RIGHT, "b", both), schema_with(order_date="TEXT", paid_date="TIME"))
|
||||
assert fact is None and "「orders.paid_date」在表结构里的类型是 TIME，只有时刻，不含日期" in why
|
||||
|
||||
|
||||
def test_the_import_dates_follow_the_names_in_the_table_structure():
|
||||
"""差异说明里的名字是规范化过的（不加引号的一律小写），配方里、表结构里的写法可能带大写：按表结构认出的那一列去对
|
||||
upload_dates，大小写不同照样对得上。"""
|
||||
upper = {"tables": {"Orders": {"name": "Orders", "columns": [
|
||||
{"name": "Status", "type": "TEXT"}, {"name": "Order_Date", "type": "TEXT"},
|
||||
{"name": "Paid_Date", "type": "TEXT"}, {"name": "Amount", "type": "REAL"}]}}}
|
||||
fact, why = derive(uploaded(BY_ORDER_DATE, RIGHT, "b", {"Orders": ["Order_Date", "Paid_Date"]}), upper)
|
||||
assert why is None and (fact.table, fact.value) == ("Orders", {"column": "paid_date"})
|
||||
fact, why = derive(uploaded(BY_ORDER_DATE, RIGHT, "b", {"orders": ["order_date", "paid_date"]}), upper)
|
||||
assert why is None and (fact.table, fact.value) == ("Orders", {"column": "paid_date"})
|
||||
|
||||
|
||||
def test_the_joint_preview_reads_the_import_dates_too():
|
||||
"""几个指标落到同一张上传表的业务日期上：preview_joint 和单个指标的预告、真写一样从候选认 upload_dates。"""
|
||||
from app.engine.dual_check_facts import preview_joint
|
||||
|
||||
texts = schema_with(order_date="TEXT", paid_date="TEXT")
|
||||
both = {"orders": ["order_date", "paid_date"]}
|
||||
picked = [{**uploaded(BY_ORDER_DATE, RIGHT, "a", both), "metric": "销售额", "metric_id": "gmv"},
|
||||
{**uploaded(BY_ORDER_DATE, RIGHT, "a", both), "metric": "订单金额", "metric_id": "total"}]
|
||||
by = {tuple((m["metric_id"], m["choice"]) for m in e["members"]): e
|
||||
for e in preview_joint(picked, {"shop": (texts, {})})}
|
||||
assert by[(("gmv", "b"), ("total", "b"))]["outcomes"] == {
|
||||
"gmv": {"write": True, "value": {"column": "paid_date"}},
|
||||
"total": {"write": True, "value": {"column": "paid_date"}}}
|
||||
# 配方读不出来的同一组候选：一项都认不出，没有可预告的组合
|
||||
plain = [{**c, "upload_dates": {}} for c in picked]
|
||||
assert preview_joint(plain, {"shop": (texts, {})}) == []
|
||||
|
||||
|
||||
async def test_an_undeclared_column_read_from_a_real_database_is_said_to_have_no_type(tmp_path):
|
||||
"""真探查一个 SQLite 库（introspect 存的是 str(列的类型)）：没写类型的列记成「NULL」（SQLAlchemy 的 NullType），
|
||||
说成没有声明类型，不说「类型是 NULL」。"""
|
||||
|
||||
@@ -0,0 +1,394 @@
|
||||
"""上传表格建的数据源上的双盲复算：业务日期按导入时配方里记的逻辑类型认日期列（engine/dual_check_facts.upload_dates）。
|
||||
|
||||
上传表格导入时日期一律按文字保存，表结构里总是 TEXT，以前这类数据源的业务日期永远写不了。导入时配方里其实记着每一列
|
||||
的逻辑类型（导入确认时界面上标的「日期」）：
|
||||
|
||||
- 配方里是日期的两列（列表块里类型为 DATE 的列、交叉表的日期轴）：审核人选了哪一边，业务日期就记哪一列；审批卡的预告
|
||||
和真写一字不差
|
||||
- 配方里是文字的列不写；配方读不出来（简单导入没有配方、配方记录坏了）的照旧不写，原因都照实说是上传的表格按文字保存；
|
||||
几个列表块写同一张表的，这一列每一块都记为日期才算
|
||||
- 只认数据源当前版本的配方：表结构出自同一个版本。审批停着时数据源换了版本，真写按写入时的版本重读（和表结构一样以
|
||||
写入时为准）；升级前发起的审批（候选里没有 upload_dates）不重读，照它审批卡上的预告不写
|
||||
|
||||
数据源一律经真实的导入接口建（暂存 → 改配方 → 试运行 → 提交，或者简单导入），工作簿是现造的：表名、地区、金额都是
|
||||
虚构的。盲算模型是 MockChatModel 剧本（tests/fixtures/dual_check_shop.Blind）。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime as dt
|
||||
import json
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from httpx import ASGITransport, AsyncClient
|
||||
from openpyxl import Workbook
|
||||
from sqlalchemy import delete, select
|
||||
|
||||
from app.data.recipe_types import Recipe, derive_tables
|
||||
from app.data.sql_semantics import explain_sql_diff
|
||||
from app.db.base import SessionLocal
|
||||
from app.db.models import DataSource, Setting, SourceSnapshot, TableRecipe
|
||||
from app.engine import spend
|
||||
from app.engine.dual_check_facts import recipe_dates, record_choices, upload_dates
|
||||
from app.engine.runner import run_manager
|
||||
from app.main import app
|
||||
from tests.fixtures.dual_check_shop import ROWS, Blind
|
||||
from tests.fixtures.sources import drop_source
|
||||
from tests.fixtures.xlsx import save
|
||||
from tests.test_dual_check_node import decide, events, graph, logs, notes_of, pending, start
|
||||
|
||||
XLSX = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
|
||||
API = "/api/datasources"
|
||||
SHEET = "订单明细"
|
||||
TABLE = "订单"
|
||||
HEADER = ("订单号", "地区", "客户", "金额", "状态", "下单日期", "支付日期")
|
||||
#: 同 dual_check_shop 的五笔单：退款行、下单日和支付日不在同一周的单，只是表名、列名、状态换成中文
|
||||
STATUS = {"paid": "已支付", "refunded": "已退款"}
|
||||
WEEK = "BETWEEN '2026-09-01' AND '2026-09-07'"
|
||||
#: 口径说明要的写法：按支付日、排除退款（1、2、5 号单，330.0）
|
||||
RIGHT = f"SELECT SUM(\"金额\") AS gmv FROM \"订单\" WHERE \"状态\" <> '已退款' AND \"支付日期\" {WEEK}"
|
||||
#: 只错在时间列：按下单日（2、4、5 号单，310.0）
|
||||
BY_ORDER_DATE = f"SELECT SUM(\"金额\") AS gmv FROM \"订单\" WHERE \"状态\" <> '已退款' AND \"下单日期\" {WEEK}"
|
||||
DEFINITION = "已支付订单的金额合计：按支付日期归属到当周，不含退款"
|
||||
#: 配方里不是日期、读不出配方时的原因：照旧说按文字保存
|
||||
STORED_AS_TEXT = "来自上传的表格，表格里的日期一律按文字保存，无法确认存的是日期"
|
||||
FLOW_RECIPE = json.loads((Path(__file__).parent / "fixtures" / "recipes" / "flow_recipe.json").read_text("utf-8"))
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
async def engine_up():
|
||||
await run_manager.setup()
|
||||
async with SessionLocal() as session:
|
||||
await session.execute(delete(Setting).where(Setting.key.in_(["verify", spend.SPEND_KEY])))
|
||||
await session.commit()
|
||||
yield
|
||||
await run_manager.shutdown()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def client():
|
||||
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
|
||||
yield c
|
||||
|
||||
|
||||
def workbook() -> bytes:
|
||||
"""订单明细：第 1 行表头，日期写成 Excel 的日期格（不是文字）。"""
|
||||
wb = Workbook()
|
||||
ws = wb.active
|
||||
ws.title = SHEET
|
||||
ws.append(list(HEADER))
|
||||
for oid, region, customer, amount, status, ordered, paid, *_ in ROWS:
|
||||
ws.append([oid, region, customer, amount, STATUS[status],
|
||||
dt.date.fromisoformat(ordered), dt.date.fromisoformat(paid)])
|
||||
return save(wb)
|
||||
|
||||
|
||||
def list_recipe(*, ordered: str = "DATE", paid: str = "DATE") -> dict:
|
||||
"""列表配方：两个日期列各按给定的类型导入（DATE 是界面上的「日期」，TEXT 是「文字」）。"""
|
||||
types = {"订单号": "INTEGER", "地区": "TEXT", "客户": "INTEGER", "金额": "REAL", "状态": "TEXT",
|
||||
"下单日期": ordered, "支付日期": paid}
|
||||
return {"recipe_format": "agentlab-recipe/2",
|
||||
"sheets": [{"id": "s1", "match": {"name": SHEET}, "blocks": [{
|
||||
"id": "orders", "layout": "list", "table": TABLE,
|
||||
"columns": [{"header": h, "name": h, "type": types[h]} for h in HEADER]}]}],
|
||||
"tables": [{"name": TABLE, "grain": ["订单号"]}]}
|
||||
|
||||
|
||||
async def commit_recipe(client, staging: str, recipe: dict) -> dict:
|
||||
"""暂存区换成给定的配方 → 试运行 → 勾全部确认项提交。返回提交的响应。"""
|
||||
r = await client.put(f"{API}/imports/{staging}/recipe", json={"recipe": recipe})
|
||||
assert r.status_code == 200 and r.json()["recipe_problems"] == [], r.text
|
||||
r = await client.post(f"{API}/imports/{staging}/trial", json={})
|
||||
assert r.status_code == 200, r.text
|
||||
trial = r.json()["trial"]
|
||||
assert trial["status"] in ("passed", "needs_decision"), trial["problems"]
|
||||
r = await client.post(f"{API}/imports/{staging}/commit", json={
|
||||
"trial_id": trial["trial_id"], "confirmations": [i["id"] for i in trial.get("confirm_items") or []]})
|
||||
assert r.status_code == 201, r.text
|
||||
return r.json()
|
||||
|
||||
|
||||
async def recipe_import(client, recipe: dict) -> tuple[str, str]:
|
||||
"""按配方导入：暂存 → 换成给定的配方 → 试运行 → 提交。返回 (数据源名, id)。"""
|
||||
name = f"dc_upload_{uuid.uuid4().hex[:8]}"
|
||||
r = await client.post(f"{API}/imports/stage", files={"file": (f"{SHEET}_2026-09.xlsx", workbook(), XLSX)},
|
||||
data={"name": name})
|
||||
assert r.status_code == 201, r.text
|
||||
out = await commit_recipe(client, r.json()["id"], recipe)
|
||||
return name, out["source"]["id"]
|
||||
|
||||
|
||||
async def simple_import(client) -> tuple[str, str]:
|
||||
"""期 1 的简单导入：按表头行导入，没有配方。"""
|
||||
name = f"dc_simple_{uuid.uuid4().hex[:8]}"
|
||||
r = await client.post(f"{API}/upload", files={"file": (f"{SHEET}.xlsx", workbook(), XLSX)}, data={"name": name})
|
||||
assert r.status_code == 201, r.text
|
||||
async with SessionLocal() as session:
|
||||
source_id = (await session.execute(select(DataSource.id).where(DataSource.name == name))).scalar_one()
|
||||
return name, source_id
|
||||
|
||||
|
||||
async def dates_of(source_id: str) -> dict:
|
||||
async with SessionLocal() as session:
|
||||
return await upload_dates(session, await session.get(DataSource, source_id))
|
||||
|
||||
|
||||
async def break_recipe(source_id: str) -> None:
|
||||
"""把数据源当前版本的配方记录改坏（造「配方读不出来」）。当前快照记的就是这一版。"""
|
||||
async with SessionLocal() as session:
|
||||
row = await session.get(DataSource, source_id)
|
||||
assert (await session.get(SourceSnapshot, row.current_snapshot_id)).recipe_id == row.current_recipe_id
|
||||
record = await session.get(TableRecipe, row.current_recipe_id)
|
||||
record.recipe = {"recipe_format": "agentlab-recipe/0"}
|
||||
await session.commit()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def dated(client):
|
||||
"""两个日期列在配方里都是「日期」的上传源。"""
|
||||
name, source_id = await recipe_import(client, list_recipe())
|
||||
yield name, source_id
|
||||
await drop_source(source_id)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def paid_as_text(client):
|
||||
"""支付日期在配方里是「文字」、下单日期是「日期」的上传源。"""
|
||||
name, source_id = await recipe_import(client, list_recipe(paid="TEXT"))
|
||||
yield name, source_id
|
||||
await drop_source(source_id)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 配方里哪些列是日期
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_the_date_columns_of_a_list_recipe_are_the_ones_typed_as_dates():
|
||||
assert recipe_dates(Recipe.model_validate(list_recipe())) == {TABLE: ["下单日期", "支付日期"]}
|
||||
assert recipe_dates(Recipe.model_validate(list_recipe(paid="TEXT"))) == {TABLE: ["下单日期"]}
|
||||
assert recipe_dates(Recipe.model_validate(list_recipe(ordered="TEXT", paid="TEXT"))) == {}
|
||||
|
||||
|
||||
def test_the_axis_of_a_crosstab_is_a_date_column_of_every_table_it_writes():
|
||||
"""交叉表的日期轴（类型恒为日期）写进各分段的表：宽表、长表、表内合计另存的表都有这一列。"""
|
||||
assert recipe_dates(Recipe.model_validate(FLOW_RECIPE)) == {
|
||||
"日客流": ["日期"], "时段客流": ["日期"], "时段客流_表内合计": ["日期"]}
|
||||
|
||||
|
||||
def test_a_column_two_lists_type_differently_is_not_a_date():
|
||||
"""两个工作表的列表块写同一张表，支付日期一块记为「日期」、一块记为「文字」：两块建出的表形状一样（日期也存成文字），
|
||||
配方校验不拦，可按文字导入的那一块存的是单元格原文（「待定」这类），这一列并不都是日期。写这一列的块都记为日期
|
||||
才算，拿不准就不认；两块都记为日期的下单日期照常认。"""
|
||||
recipe = list_recipe()
|
||||
february = json.loads(json.dumps(recipe["sheets"][0]))
|
||||
february["id"], february["match"] = "s2", {"name": "二月"}
|
||||
february["blocks"][0]["id"] = "orders_feb"
|
||||
for column in february["blocks"][0]["columns"]:
|
||||
if column["name"] == "支付日期":
|
||||
column["type"] = "TEXT"
|
||||
recipe["sheets"].append(february)
|
||||
parsed = Recipe.model_validate(recipe)
|
||||
assert derive_tables(parsed)[1] == [], "背景：两块写成的表形状一样，不报冲突"
|
||||
assert recipe_dates(parsed) == {TABLE: ["下单日期"]}
|
||||
# 两块都记为日期：照常认两列
|
||||
both = list_recipe()
|
||||
both["sheets"].append({**json.loads(json.dumps(both["sheets"][0])), "id": "s2", "match": {"name": "二月"}})
|
||||
both["sheets"][1]["blocks"][0]["id"] = "orders_feb"
|
||||
assert recipe_dates(Recipe.model_validate(both)) == {TABLE: ["下单日期", "支付日期"]}
|
||||
|
||||
|
||||
def test_a_list_total_kept_as_a_table_has_no_date_column():
|
||||
"""列表的合计行另存成表时只有合计项和数字列：日期列不在里面。"""
|
||||
recipe = list_recipe()
|
||||
recipe["sheets"][0]["blocks"][0]["rows"] = {"total_row": {"label_column": "地区", "pick": "合计",
|
||||
"keep_as": "订单合计"}}
|
||||
recipe["tables"].append({"name": "订单合计", "kind": "reported_total"})
|
||||
assert recipe_dates(Recipe.model_validate(recipe)) == {TABLE: ["下单日期", "支付日期"]}
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 读当前版本的配方
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def test_an_upload_source_reads_the_date_columns_from_its_current_recipe(client, dated, paid_as_text):
|
||||
"""真实导入流程建的源：表结构里两列都是 TEXT，配方里是日期的那几列读得出来。"""
|
||||
async with SessionLocal() as session:
|
||||
schema = (await session.get(DataSource, dated[1])).schema_cache
|
||||
types = {c["name"]: c["type"] for c in schema["tables"][TABLE]["columns"]}
|
||||
assert (types["下单日期"], types["支付日期"]) == ("TEXT", "TEXT"), "背景：导入时日期一律按文字保存"
|
||||
assert await dates_of(dated[1]) == {TABLE: ["下单日期", "支付日期"]}
|
||||
assert await dates_of(paid_as_text[1]) == {TABLE: ["下单日期"]}
|
||||
|
||||
|
||||
async def current_snapshot(source_id: str) -> str:
|
||||
async with SessionLocal() as session:
|
||||
return (await session.get(DataSource, source_id)).current_snapshot_id
|
||||
|
||||
|
||||
async def test_the_dates_follow_the_current_version_of_the_source(client, dated):
|
||||
"""只认数据源当前版本的配方（表结构出自同一个版本）：修改配方把支付日期改成「文字」、提交之后只剩下单日期；回滚到
|
||||
旧版本，两列又都是日期。"""
|
||||
_, source_id = dated
|
||||
first = await current_snapshot(source_id)
|
||||
r = await client.post(f"{API}/{source_id}/redraft", json={})
|
||||
assert r.status_code == 201, r.text
|
||||
await commit_recipe(client, r.json()["id"], list_recipe(paid="TEXT"))
|
||||
second = await current_snapshot(source_id)
|
||||
assert second != first
|
||||
assert await dates_of(source_id) == {TABLE: ["下单日期"]}
|
||||
r = await client.post(f"{API}/{source_id}/snapshots/{first}/activate",
|
||||
json={"confirm": True, "expected_current_snapshot_id": second})
|
||||
assert r.status_code == 200, r.text
|
||||
assert await current_snapshot(source_id) == first
|
||||
assert await dates_of(source_id) == {TABLE: ["下单日期", "支付日期"]}
|
||||
|
||||
|
||||
async def test_without_a_readable_recipe_no_column_is_taken_as_a_date(client, dated, tmp_path):
|
||||
"""读不出配方的都不认：简单导入没有配方，配方记录坏了，手工登记的库，不存在的数据源。"""
|
||||
name, source_id = await simple_import(client)
|
||||
try:
|
||||
async with SessionLocal() as session:
|
||||
row = await session.get(DataSource, source_id)
|
||||
assert row.origin == "upload" and row.current_recipe_id is None
|
||||
assert await dates_of(source_id) == {}
|
||||
finally:
|
||||
await drop_source(source_id)
|
||||
await break_recipe(dated[1])
|
||||
assert await dates_of(dated[1]) == {}
|
||||
async with SessionLocal() as session:
|
||||
manual = DataSource(name=f"dc_manual_{uuid.uuid4().hex[:8]}", kind="sqlite", database=str(tmp_path / "x.db"))
|
||||
assert await upload_dates(session, manual) == {}
|
||||
assert await upload_dates(session, None) == {}
|
||||
|
||||
|
||||
async def test_the_node_hands_the_import_dates_over_with_the_source(client, dated):
|
||||
"""节点认数据源来历的那一步（_source_info）带上 upload_dates，候选据此交给 derive。"""
|
||||
from app.engine.nodes.dual_check import _source_info, _Sources
|
||||
|
||||
info = await _source_info(_Sources(), dated[0])
|
||||
assert info["uploaded"] is True and info["upload_dates"] == {TABLE: ["下单日期", "支付日期"]}
|
||||
assert (await _source_info(_Sources(), "dc_no_such_source"))["upload_dates"] == {}
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 双盲复算：审批卡的预告、裁决后的真写
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
def upload_graph(name: str, sql: str) -> dict:
|
||||
return graph(name, sql, dual={"definitions": {"gmv": DEFINITION}})
|
||||
|
||||
|
||||
async def test_dates_the_import_took_as_dates_are_written_as_the_business_date(monkeypatch, client, dated):
|
||||
"""两列在配方里都是「日期」：审批卡预告选哪边记哪一列，审核人选盲算，业务日期记「支付日期」，人工确认、备注写明出处。"""
|
||||
name, source_id = dated
|
||||
Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")})
|
||||
run = await start(upload_graph(name, BY_ORDER_DATE))
|
||||
assert run.status == "interrupted", run.error
|
||||
[item] = (await pending(run.id)).payload["summary"]
|
||||
assert len(item["diff"]) == 1 and "下单日期" in item["diff"][0] and "支付日期" in item["diff"][0]
|
||||
where = {"source": name, "table": TABLE, "item": "business_date"}
|
||||
assert item["catalog"] == {"a": {"write": True, **where, "value": {"column": "下单日期"}},
|
||||
"b": {"write": True, **where, "value": {"column": "支付日期"}}}
|
||||
done = await decide(client, run.id, {"gmv": "b"})
|
||||
assert done.status == "succeeded", done.error
|
||||
[wrote] = logs(await events(run.id), "dual_check_catalog")
|
||||
assert wrote["written"] == [{**where, "value": {"column": "支付日期"}}] and wrote["skipped"] == []
|
||||
entry = await notes_of(source_id, TABLE)
|
||||
stored = entry.notes["business_date"]
|
||||
assert stored["value"] == {"column": "支付日期"}
|
||||
assert (stored["source"], stored["status"]) == ("human", "confirmed") and run.id in stored["note"]
|
||||
|
||||
|
||||
async def test_a_column_the_recipe_kept_as_text_is_not_written(monkeypatch, client, paid_as_text):
|
||||
"""支付日期在配方里是「文字」：选哪边都不记，原因照旧说按文字保存——被采纳的是它就说它，被采纳的下单日期是日期的，说
|
||||
未被采纳的支付日期、另说清为什么也要看它。预告和裁决后跳过的原因一字不差，目录里什么都没写。"""
|
||||
name, source_id = paid_as_text
|
||||
Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")})
|
||||
run = await start(upload_graph(name, BY_ORDER_DATE))
|
||||
assert run.status == "interrupted", run.error
|
||||
said = (await pending(run.id)).payload["summary"][0]["catalog"]
|
||||
column = f"「{TABLE}.支付日期」{STORED_AS_TEXT}"
|
||||
assert said["b"] == {"write": False, "reason": f"指标「销售额」被采纳的盲算按时间筛的列{column}，不写入数据目录"}
|
||||
assert said["a"] == {"write": False, "reason": f"指标「销售额」未被采纳的盲算按时间筛的列{column}："
|
||||
"两边都是日期列时，审核人选的才是业务日期按哪一列，不写入数据目录"}
|
||||
done = await decide(client, run.id, {"gmv": "b"})
|
||||
assert done.status == "succeeded", done.error
|
||||
[wrote] = logs(await events(run.id), "dual_check_catalog")
|
||||
assert wrote["written"] == [] and [s["reason"] for s in wrote["skipped"]] == [said["b"]["reason"]]
|
||||
assert await notes_of(source_id, TABLE) is None
|
||||
|
||||
|
||||
async def test_without_a_readable_recipe_nothing_is_written(monkeypatch, client, dated):
|
||||
"""配方记录坏了、读不出来：两列都照旧说按文字保存，不写。"""
|
||||
name, source_id = dated
|
||||
await break_recipe(source_id)
|
||||
Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")})
|
||||
run = await start(upload_graph(name, BY_ORDER_DATE))
|
||||
assert run.status == "interrupted", run.error
|
||||
said = (await pending(run.id)).payload["summary"][0]["catalog"]
|
||||
for side, label, column in (("a", "原流程", "下单日期"), ("b", "盲算", "支付日期")):
|
||||
assert said[side] == {"write": False, "reason": f"指标「销售额」被采纳的{label}按时间筛的列「{TABLE}.{column}」"
|
||||
f"{STORED_AS_TEXT}，不写入数据目录"}
|
||||
done = await decide(client, run.id, {"gmv": "a"})
|
||||
assert done.status == "succeeded", done.error
|
||||
[wrote] = logs(await events(run.id), "dual_check_catalog")
|
||||
assert wrote["written"] == [] and [s["reason"] for s in wrote["skipped"]] == [said["a"]["reason"]]
|
||||
assert await notes_of(source_id, TABLE) is None
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 审批停着时数据源换了版本：真写按写入时的版本认
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def test_a_new_version_while_waiting_is_what_the_write_reads(monkeypatch, client, dated):
|
||||
"""审批停着时修改配方、把支付日期改成「文字」并提交成新版本：真写时表结构是现读的，导入时认作日期的列也按写入时的
|
||||
当前版本重读，两者出自同一个版本。审批卡上预告会记「支付日期」，裁决后照实说按文字保存、不写——不拿发起审批时读的
|
||||
配方去配新版本的表结构，写出一条哪个版本单独看都不会给出的人工确认。"""
|
||||
name, source_id = dated
|
||||
Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")})
|
||||
run = await start(upload_graph(name, BY_ORDER_DATE))
|
||||
assert run.status == "interrupted", run.error
|
||||
[item] = (await pending(run.id)).payload["summary"]
|
||||
assert item["catalog"]["b"]["write"] is True, "背景：发起审批时两列在配方里都是日期"
|
||||
r = await client.post(f"{API}/{source_id}/redraft", json={})
|
||||
assert r.status_code == 201, r.text
|
||||
await commit_recipe(client, r.json()["id"], list_recipe(paid="TEXT"))
|
||||
assert await dates_of(source_id) == {TABLE: ["下单日期"]}
|
||||
done = await decide(client, run.id, {"gmv": "b"})
|
||||
assert done.status == "succeeded", done.error
|
||||
[wrote] = logs(await events(run.id), "dual_check_catalog")
|
||||
assert wrote["written"] == []
|
||||
assert [s["reason"] for s in wrote["skipped"]] == [
|
||||
f"指标「销售额」被采纳的盲算按时间筛的列「{TABLE}.支付日期」{STORED_AS_TEXT}，不写入数据目录"]
|
||||
assert await notes_of(source_id, TABLE) is None
|
||||
|
||||
|
||||
async def candidate_of(name: str, source_id: str, choice: str) -> dict:
|
||||
"""真写收到的候选（节点从 checkpoint 里取出、加上审核人的选择）：差异说明按数据源的表结构真算。"""
|
||||
async with SessionLocal() as session:
|
||||
schema = (await session.get(DataSource, source_id)).schema_cache
|
||||
diff = explain_sql_diff(BY_ORDER_DATE, RIGHT, "sqlite", schema=schema, labels=("原流程", "盲算")).as_dict()
|
||||
return {"metric_id": "gmv", "metric": "销售额", "source": name, "dialect": "sqlite", "uploaded": True,
|
||||
"sql": {"a": BY_ORDER_DATE, "b": RIGHT}, "diff": diff, "choice": choice}
|
||||
|
||||
|
||||
async def test_a_checkpoint_without_the_dates_is_written_as_its_card_said(client, dated):
|
||||
"""升级前发起的审批，checkpoint 的候选里没有 upload_dates：它的审批卡预告的是按文字保存、不写，真写不重读配方，照旧
|
||||
不写，和那张卡说的一致。候选里带着这一项的（升级后发起的）按写入时的版本重读：发起审批时读出来是空的，写入时配方里
|
||||
两列都是日期，照写——和表结构、目录在审批期间变了一样，以写入时为准。"""
|
||||
name, source_id = dated
|
||||
old = await candidate_of(name, source_id, "b")
|
||||
assert [i["aspect"] for i in old["diff"]["items"]] == ["time"], "背景：两边只差按哪一列筛时间"
|
||||
result = await record_choices([old], run_id="run_old", node_title="复核", actor=None)
|
||||
assert result["written"] == []
|
||||
assert [s["reason"] for s in result["skipped"]] == [
|
||||
f"指标「销售额」被采纳的盲算按时间筛的列「{TABLE}.支付日期」{STORED_AS_TEXT}，不写入数据目录"]
|
||||
result = await record_choices([{**old, "upload_dates": {}}], run_id="run_new", node_title="复核", actor=None)
|
||||
assert result["skipped"] == []
|
||||
assert result["written"] == [
|
||||
{"source": name, "table": TABLE, "item": "business_date", "value": {"column": "支付日期"}}]
|
||||
@@ -2280,6 +2280,17 @@ await section('核查类日志与出具缺口（期 4）：无法复算、未能
|
||||
check('……两路不一致（选了盲算、两边都不对、没作出选择）和被推翻的归「核查未通过」，无法复算的归「核查未完成」',
|
||||
served.failed.length === 4 && served.unfinished.length === 1 && served.unfinished[0].startsWith('双盲复算：指标「订单数」无法复算：')
|
||||
&& served.other.length === 0, JSON.stringify(served))
|
||||
// 出口契约自己报的「正文写在代码块里、结论句未能核对」（nodes/io.py 的 GAP_FENCED_BODY）不是核查节点报的：不带核查
|
||||
// 节点的前缀，横幅归「校验未全部完成」、时间线归「校验不完整」。服务端改了说法、加了前缀，或者这边改了分组，这里先变红
|
||||
const fencedGap = ioPy.match(/^GAP_FENCED_BODY = "([^"]+)"$/m)?.[1]?.replace('{title}', '写周报')
|
||||
const fencedRow = flatten(mod.decodeRun([
|
||||
E(1, 'node.started', 'done', { node_type: 'output', label: '成果' }),
|
||||
E(2, 'issuance', 'done', { tier: 'degraded', missing_required: [], missing_expected: [], unmatched: 0, gaps: [fencedGap] }),
|
||||
E(3, 'node.finished', 'done', { duration_ms: 5 }),
|
||||
])).find((s) => s.kind === 'issuance')
|
||||
check('出口契约报的「正文写在代码块里」：读出了服务端的写法，归「校验不完整」，不进核查的两组', !!fencedGap
|
||||
&& JSON.stringify(mod.splitIssuanceGaps([fencedGap])) === JSON.stringify({ failed: [], unfinished: [], other: [fencedGap] })
|
||||
&& fencedRow?.title === `降档出具：校验不完整：${fencedGap}`, `${fencedGap} / ${fencedRow?.title}`)
|
||||
|
||||
// dual_check_catalog 的 written 带了写进去的值（期 4）：时间线写「…记为「值」」，值的写法和审批卡上的「会记为「值」」
|
||||
// 同一个函数（catalogPatch 的 patchValueText），两处一字不差
|
||||
|
||||
Reference in New Issue
Block a user