diff --git a/backend/app/engine/dual_check_facts.py b/backend/app/engine/dual_check_facts.py index ecf2b1eb..6bde1842 100644 --- a/backend/app/engine/dual_check_facts.py +++ b/backend/app/engine/dual_check_facts.py @@ -23,17 +23,23 @@ is_test = 0、is_valid = 1、deleted_at IS NULL);只涉及一张表。拿不 **怎么写。** 照对话提案的规矩核对(plan_patch:列在不在表结构里、取值格式),记为人工填写、已确认,备注写明来自哪次 运行、哪个节点、哪个指标。目录里这一项已经是人工确认过的另一种说法时不覆盖,照实交回原因;已经是同样的说法就什么都 -不写(再写一遍也不升版本)。有效记录条件是一整句条件:目录里已有的条件被采纳的一边(几个指标并成一项时是每一个) -也按它筛了的,和新条件合起来写(已确认的不改,照实说);有一个没按它筛的,两者的关系拿不准,不动。推断的去重规则 -说的未必是同一件事,也不替换。整个写入由节点放在 ask() 之后的一个 @task 里:审批恢复重放、接着运行时不会再写一遍。 +不写(再写一遍也不升版本);这一项被人工驳回过、要写的又带回了驳回的说法(有效记录条件按段比:驳回的那句每一段都在 +要写的条件里),两次人工判断对不上,不写。有效记录条件是一整句条件:目录里已有的条件被采纳的一边(几个指标并成一项 +时是每一个)也按它筛了的,和新条件合起来写,已有的段只写一次(已确认的不改,照实说多出了哪几段);有一个没按它筛 +的,两者的关系拿不准,不动。推断的去重规则说的未必是同一件事,也不替换。整个写入由节点放在 ask() 之后的一个 @task +里:审批恢复重放、接着运行时不会再写一遍。 **同一次裁决里几个指标落到同一项上。** 说法一样的写一次,备注里列出每个指标。有效记录条件是能用 AND 叠加的排除条件 (这里只收排除无效记录的),几个指标各自采纳的条件都有人的选择撑着,合成一句写——「排除退款」和「排除测试单」并不 -矛盾。业务日期一张表只有一个,去重规则说的未必是同一件事,说法不同就一个都不写。 +矛盾。业务日期一张表只有一个,去重规则说的未必是同一件事,说法不同就一个都不写。合起来之前先按每个指标自己的说法 +过一遍驳回检查:自己采纳的说法曾被人工驳回的指标拿出来不写、照实说原因,其余的照写;合起来以后又包含了驳回的整句的, +也不写。 **预告。** 审批卡上要让审核人看得出选了之后会往目录里写什么(preview):按发起审批时的目录,选原流程、选盲算各会 -写什么、不写的为什么。和真写是同一套判断(derive 认事实、_plan 对目录),所以只在两处可能和最后写的不一样:审批停下 -期间目录被人改了,或者同一次裁决里几个指标落到同一项上(上一段)。 +写什么、不写的为什么。和真写是同一套判断(derive 认事实、_plan 对目录、_revives 认驳回),所以只在两处可能和最后写 +的不一样:审批停下期间目录被人改了;同一次裁决里几个指标落到同一项上(上一段)——写进去的是合起来的说法,预告说会 +写的也可能因为说法互相矛盾、和已有条件的关系拿不准、合起来又成了驳回过的说法而不写。反过来不会:预告说不写的, +不会因为和别的指标并成一项被写进去。 catalog.save_patch 不收备注,这里用它的同一套原料(read_entry、apply_patch、write_entry 的乐观锁)写,只多把备注放进项里。 """ @@ -41,7 +47,7 @@ from __future__ import annotations import logging import re -from collections.abc import Mapping +from collections.abc import Mapping, Sequence from dataclasses import dataclass, replace from typing import Any @@ -86,8 +92,9 @@ _SCOPE_DIGITS = 2 @dataclass(frozen=True) class Fact: """要写进目录的一项:哪个数据源的哪张表、哪一项、写什么。metric、choice 只用来写备注和说明;sql、dialect 是被采纳 - 那一边的写法和方言,有效记录条件和目录里已有的条件合并时要看它还按哪些条件筛了。also 是同一次裁决里落到同一项上、 - 并进这一项的其余几个指标(各自的指标、选择、写法),value 是合起来之后的说法。""" + 那一边的写法和方言,有效记录条件和目录里已有的条件合并时要看它还按哪些条件筛了。同一次裁决里几个指标落到同一项上 + 时并成一项:members 是并进来的每个指标各自的那一项(各带自己的说法,审批卡上各自的预告就是按它算的),value 是 + 合起来之后的说法;单个指标的 members 为空。""" source: str table: str @@ -97,17 +104,17 @@ class Fact: choice: str sql: str = "" dialect: str | None = None - also: tuple[Fact, ...] = () + members: tuple[Fact, ...] = () def _members(fact: Fact) -> tuple[Fact, ...]: - """这一项背后的每个指标的选择:并过的连同并进来的。""" - return (fact, *fact.also) + """这一项背后的每个指标的选择:并成一项的是并进来的每一个(各带自己的说法),单个指标的是它自己。""" + return fact.members or (fact,) -def _names(fact: Fact) -> str: - """说明里的指标叫法:「销售额」,几个指标并成一项时「销售额」、「订单金额」。""" - names = list(dict.fromkeys(m.metric for m in _members(fact))) +def _names(members: Sequence[Fact]) -> str: + """说明里的指标叫法:「销售额」,几个指标时「销售额」、「订单金额」。""" + names = list(dict.fromkeys(m.metric for m in members)) return "、".join(f"「{n}」" for n in names[:4]) + (f"等 {len(names)} 个" if len(names) > 4 else "") @@ -422,39 +429,57 @@ def _conditions_of(sql: str, read: str | None) -> set[str] | None: return {_normalized(part, read) for where in tree.find_all(exp.Where) for part in _and_parts(where.this)} -def _and_join(base: str | None, values: list[str], dialect: str | None) -> str | None: - """几句条件用 AND 接成一句:已经有的(base 里的、前面接过的,按 _normalized 比)不再接;最外层是 OR 的加括号, - 不然后面接的条件只并到 OR 的后一半上。每句保留原来的写法。有一句解析不了返回 None。""" +def _keys(condition: Any, read: str | None) -> set[str] | None: + """一句条件按 AND 拆开后每一段的比较用写法;不是文字、解析不了返回 None。""" + tree = _parse(condition, read) if isinstance(condition, str) and condition.strip() else None + return {_normalized(part, read) for part in _and_parts(tree)} if tree is not None else None + + +def _and_join(base: str | None, values: list[str], dialect: str | None, *, keep_base: bool = True) -> str | None: + """几句条件用 AND 接成一句:已经有的段(base 里的、前面接过的,按 _normalized 比)不再接。一句里每一段都是新的, + 原样接上(保留原来的写法);只有几段是新的,只接那几段(按方言重新写)——整句接上会把已有的段再写一遍。 + keep_base 为假时 base 只用来认哪些段已经有了、不写进结果(说「另外多出哪些条件」用),什么都没多出时为空串。 + 不止一段时,最外层是 OR 的加括号,不然后面接的条件只并到 OR 的后一半上。有一句解析不了返回 None。""" from sqlglot import exp read = _read(dialect) - texts: list[str] = [] + pieces: list[tuple[str, bool]] = [] # (写法, 最外层是不是 OR) seen: set[str] = set() - for text in [*([base] if base else []), *values]: + for i, text in enumerate([*([base] if base else []), *values]): tree = _parse(str(text), read) if tree is None: return None - keys = {_normalized(p, read) for p in _and_parts(tree)} - if keys <= seen: + parts = _and_parts(tree) + fresh = [] + for part in parts: + key = _normalized(part, read) + if key not in seen: + seen.add(key) + fresh.append(part) + if not fresh or (base and i == 0 and not keep_base): continue - seen |= keys - texts.append(f"({text})" if isinstance(tree, exp.Or) else str(text)) - return " AND ".join(texts) + if len(fresh) == len(parts): + pieces.append((str(text).strip(), isinstance(tree, exp.Or))) + else: + pieces += [(part.sql(dialect=read), isinstance(part, exp.Or)) for part in fresh] + if len(pieces) == 1: + return pieces[0][0] + return " AND ".join(f"({text})" if is_or else text for text, is_or in pieces) def _combined_filter(existing: str, fact: Fact) -> tuple[str | None, str | None]: """目录里已有的有效记录条件和这次的新条件合起来:(合起来的整句条件, None) 或 (None, 为什么不动)。 有效记录条件是一整句(copilot 提案也要求「原有条件也要带上」)。只写新条件会把两边都按着筛的原有条件静默替换掉; - 原有条件有一个被采纳的一边没按它筛,审核人的选择说明不了它对不对(合起来写就等于替人确认了它),也不能替人删掉。""" + 原有条件有一个被采纳的一边没按它筛,审核人的选择说明不了它对不对(合起来写就等于替人确认了它),也不能替人删掉。 + 接的是每个指标各自的条件(不是并好的整句):哪个指标的条件和已有的重了,那一段都只写一次。""" label = catalog.TABLE_FIELD_LABEL["valid_filter"] read = _read(fact.dialect) - have_tree = _parse(existing, read) + have = _keys(existing, read) combined = _and_join(existing, [str(m.value) for m in _members(fact)], fact.dialect) \ - if have_tree is not None else None - if combined is None: + if have is not None else None + if have is None or combined is None: return None, f"「{fact.table}」已有的{label}「{existing}」无法解析,与这次的条件合不起来,未改动" - have = {_normalized(p, read) for p in _and_parts(have_tree)} for member in _members(fact): if not have <= (_conditions_of(member.sql, read) or set()): return None, (f"「{fact.table}」已有的{label}「{existing}」,指标「{member.metric}」被采纳的" @@ -560,28 +585,82 @@ def _differing(facts: list[Fact]) -> str: def _merged(members: list[Fact]) -> tuple[Fact | None, str | None]: - """同一次裁决里落到同一项上的几个说法合成一项:(合起来的那一项, None) 或 (None, 为什么都不写)。 + """同一次裁决里落到同一项上的几个说法合成一项:(合起来的那一项, None) 或 (None, 为什么都不写)。只有一个时就是它。 说法都一样的写一次。有效记录条件用 AND 合成一句(derive 只收排除无效记录的条件,几句叠起来不会互相打架)。 - 业务日期一张表只有一个、去重规则说的未必是同一件事,说法不同就都不写。""" - first, rest = members[0], tuple(members[1:]) - if not rest: + 业务日期一张表只有一个、去重规则说的未必是同一件事,说法不同就都不写。每个指标各自的那一项原样留在 members 里: + 驳回检查(_screen)、和目录里已有条件的合并(_combined_filter)都要按各自的说法算。""" + first = members[0] + if len(members) == 1: return first, None if first.item == "valid_filter": value = _and_join(None, [str(m.value) for m in members], first.dialect) - return (replace(first, value=value, also=rest), None) if value is not None else (None, _differing(members)) - if all(_same_fact(first.item, m.value, first.value) for m in rest): - return replace(first, also=rest), None + return (replace(first, value=value, members=tuple(members)), None) if value is not None \ + else (None, _differing(members)) + if all(_same_fact(first.item, m.value, first.value) for m in members[1:]): + return replace(first, members=tuple(members)), None return None, _differing(members) +def _rejected(notes: Mapping[str, Any], item: str) -> tuple[bool, Any]: + """目录里这一项是不是被人工驳回的:(是不是, 驳回的说法)。""" + stored = notes.get(item) + if isinstance(stored, Mapping) and stored.get("status") == "rejected": + return True, stored.get("value") + return False, None + + +def _revives(fact: Fact, rejected: Any) -> bool: + """要写的说法是不是又带回了曾被人工驳回的说法。业务日期、去重规则:同一个说法。有效记录条件:驳回的那句的每一段 + 都在要写的条件里(按 _normalized 比,!= 和 <>、列名大小写、先后不同都认得出;有一边解析不了按原文比)——驳回的 + 那句只是要写的一部分也算,写进去就等于替人把驳回过的条件又确认了一遍。""" + if fact.item != "valid_filter": + return _same_fact(fact.item, rejected, fact.value) + read = _read(fact.dialect) + mine, theirs = _keys(fact.value, read), _keys(rejected, read) + if mine is None or theirs is None: + return _same_fact(fact.item, rejected, fact.value) + return bool(theirs) and theirs <= mine + + +def _rejected_text(fact: Fact, rejected: Any, members: Sequence[Fact]) -> str: + """驳回检查拦下时的说法:驳回的是哪一句(照目录里记的原文),哪几个指标的选择又带回了它。审批卡上的预告和真写 + 拦下时用的是同一句话。""" + label = catalog.TABLE_FIELD_LABEL.get(fact.item, fact.item) + return (f"「{fact.table}」的{label}「{_value_text(fact.item, rejected)}」曾被人工驳回,与这次指标{_names(members)}" + "的选择相反,未写入。如需更改,请到数据目录中修改") + + +def _screen(notes: Mapping[str, Any], fact: Fact) -> tuple[Fact | None, list[str]]: + """几个指标并成的一项,先拿掉自己采纳的说法曾被人工驳回的指标:(剩下的重新并成的一项, 拿掉的为什么不写),一个 + 都不剩时前者为 None。单个指标的、这一项没被驳回过的原样返回,由 _plan 照常判断。 + + 审批卡上每个指标的预告是按它自己的说法单独算的(preview → _plan),在那里被驳回检查拦下、卡上写着「不写」的, + 并进来以后也得拦下。只拿合起来的整句去比,「status <> 'refunded'」和别的条件 AND 在一起就不再等于驳回的那句, + 会被写成人工确认,和卡上说的正相反。拿掉的指标照实说原因(和它的预告同一句话),其余的照写,也和各自的预告一致。""" + rejected, said = _rejected(notes, fact.item) + if not rejected or not fact.members: + return fact, [] + out: list[Fact] = [] + kept: list[Fact] = [] + for member in fact.members: + (out if _revives(member, said) else kept).append(member) + if not out: + return fact, [] + reasons = [_rejected_text(fact, said, out)] + if not kept: + return None, reasons + narrowed, why = _merged(kept) + return narrowed, reasons if why is None else [*reasons, why] + + async def record_choices(candidates: list[Mapping[str, Any]], *, run_id: str, node_title: str, actor: str | None) -> dict[str, list[dict[str, Any]]]: """把审核人的选择写进数据目录:{written: [{source, table, item}], skipped: [{reason}]}。永远不抛:数据源没了、 目录写不进去,都落在 skipped 里照实说。 先把每个候选认成目录项,再按(数据源、表、项)归拢成一项(_merged:说法一样的、有效记录条件合起来写,说法 - 不同的整组不写),逐项写。""" + 不同的整组不写),逐项写(_write:按读到的目录先拿掉各自带回驳回说法的指标,其余的照写)。""" skipped: list[dict[str, Any]] = [] facts: list[Fact] = [] for candidate in candidates: @@ -611,39 +690,39 @@ async def record_choices(candidates: list[Mapping[str, Any]], *, run_id: str, no if fact is None: skipped.append({"reason": why}) continue - done, why = await _record(fact, run_id=run_id, node_title=node_title, actor=actor) + done, reasons = await _record(fact, run_id=run_id, node_title=node_title, actor=actor) if done: written.append({"source": fact.source, "table": fact.table, "item": fact.item}) - else: - skipped.append({"reason": why}) + skipped += [{"reason": r} for r in reasons] return {"written": written, "skipped": skipped} -async def _record(fact: Fact, *, run_id: str, node_title: str, actor: str | None) -> tuple[bool, str | None]: +async def _record(fact: Fact, *, run_id: str, node_title: str, actor: str | None) -> tuple[bool, list[str]]: + """写一项:(写了没有, 没写的原因)。几个指标并成的一项可能写了、同时有几个指标被拿出来没写,原因照样交回。""" try: async with SessionLocal() as session: row = (await session.execute(select(DataSource).where(DataSource.name == fact.source))).scalar_one_or_none() if row is None: - return False, f"数据源「{fact.source}」已不存在,指标{_names(fact)}的选择未写入数据目录" - return await _write(session, row, fact, note=note_of(fact, run_id=run_id, node_title=node_title), - actor=actor) + return False, [f"数据源「{fact.source}」已不存在,指标{_names(_members(fact))}的选择未写入数据目录"] + return await _write(session, row, fact, run_id=run_id, node_title=node_title, actor=actor) except Exception: # noqa: BLE001 - 目录写不进去不能毁掉审批的结果:缺口照样按选择记 logger.exception("双盲复算的选择写入数据目录失败(数据源 %s)", fact.source) - return False, f"指标{_names(fact)}的选择写入数据目录时出错,未写入" + return False, [f"指标{_names(_members(fact))}的选择写入数据目录时出错,未写入"] def _plan(notes: Mapping[str, Any], fact: Fact, tables: Mapping[str, Any] | None) -> tuple[Any, str | None]: """按目录里这张表现在的说法,算出这一项要写成什么:(要写的取值, None) 或 (None, 为什么不写)。不读库、不写库: - 真写(_write)和审批卡上的预告(preview)用的是这同一套判断。""" + 真写(_write)和审批卡上的预告(preview)用的是这同一套判断。几个指标并成的一项,先由 _screen 拿掉各自带回驳回 + 说法的指标再交到这里。""" label = catalog.TABLE_FIELD_LABEL.get(fact.item, fact.item) - names = _names(fact) + names = _names(_members(fact)) stored = notes.get(fact.item) if isinstance(notes.get(fact.item), dict) else None - if stored is not None and stored.get("status") == "rejected" and _same_fact(fact.item, stored.get("value"), - fact.value): - # 同一个说法以前被人明确驳回过:两次人工判断对不上,不替人拿主意 - return None, (f"「{fact.table}」的{label}「{_value_text(fact.item, fact.value)}」曾被人工驳回," - f"与这次指标{names}的选择相反,未写入。如需更改,请到数据目录中修改") - live = stored if stored is not None and stored.get("status") != "rejected" else None + rejected, said = _rejected(notes, fact.item) + if rejected and _revives(fact, said): + # 同一个说法以前被人明确驳回过:两次人工判断对不上,不替人拿主意。并成的一项各自的说法都不是驳回的那句、 + # 合起来却又包含了它,也在这里拦下 + return None, _rejected_text(fact, said, _members(fact)) + live = stored if stored is not None and not rejected else None value = fact.value if fact.item == "business_date" and live is not None and isinstance(live.get("value"), dict) \ and live["value"].get("column") == value.get("column"): @@ -655,9 +734,13 @@ def _plan(notes: Mapping[str, Any], fact: Fact, tables: Mapping[str, Any] | None if combined is None: return None, why if live.get("status") == "confirmed" and not _same_fact(fact.item, combined, live["value"]): - adopted = f"被采纳的{SIDE_LABEL.get(fact.choice, fact.choice)}" if not fact.also else "这次几个指标被采纳的写法" + # 只说已确认的说法里没有的那几段:已有的段算成「多出」,审核人会以为连它也被改了 + extra = _and_join(live["value"].strip(), [str(m.value) for m in _members(fact)], fact.dialect, + keep_base=False) + adopted = f"被采纳的{SIDE_LABEL.get(fact.choice, fact.choice)}" if not fact.members \ + else "这次几个指标被采纳的写法" return None, (f"「{fact.table}」的{label}已有人工确认过的说法「{live['value'].strip()}」,{adopted}" - f"另外多出条件「{fact.value}」,未改动。如需补上,请到数据目录中修改") + f"另外多出条件「{extra or fact.value}」,未改动。如需补上,请到数据目录中修改") value = combined elif fact.item == "dedup" and isinstance(live, dict) and live.get("status") != "confirmed" \ and not _same_fact(fact.item, live.get("value"), value): @@ -702,34 +785,41 @@ def preview(candidate: Mapping[str, Any], schema_cache: Mapping[str, Any] | None return out -async def _write(session: Any, row: DataSource, fact: Fact, *, note: str, actor: str | None) -> tuple[bool, str | None]: - """写一项。(写了没有, 没写的原因)。写什么、写不写由 _plan 按读到的目录定;版本对不上(别人刚改过)重读再来。""" +async def _write(session: Any, row: DataSource, fact: Fact, *, run_id: str, node_title: str, + actor: str | None) -> tuple[bool, list[str]]: + """写一项:(写了没有, 没写的原因)。几个指标并成的一项先按读到的目录拿掉各自带回驳回说法的指标(_screen),写什么、 + 写不写再由 _plan 定;备注只列真写进去的指标。版本对不上(别人刚改过)重读再来。""" label = catalog.TABLE_FIELD_LABEL.get(fact.item, fact.item) tables = (row.schema_cache or {}).get("tables") if isinstance(row.schema_cache, dict) else None + reasons: list[str] = [] for _ in range(_RETRIES): entry = await catalog.read_entry(session, row.id, fact.table) notes = entry.notes if entry else {} version = entry.version if entry else 0 - value, why = _plan(notes, fact, tables) + kept, reasons = _screen(notes, fact) + if kept is None: + return False, reasons + value, why = _plan(notes, kept, tables) if why is not None: - return False, why + return False, [*reasons, why] + note = note_of(kept, run_id=run_id, node_title=node_title) at = catalog.now_iso() try: - merged = catalog.apply_patch(notes, [{"path": fact.item, "value": value, "reason": note}], - table=fact.table, tables=tables, at=at) + merged = catalog.apply_patch(notes, [{"path": kept.item, "value": value, "reason": note}], + table=kept.table, tables=tables, at=at) except catalog.CatalogInvalid as e: - return False, f"「{fact.table}」的{label}未写入数据目录:{e}" + return False, [*reasons, f"「{kept.table}」的{label}未写入数据目录:{e}"] # 不论原来是谁提的:这一项现在出自人的选择,记为人工填写、已确认,备注写明出处 - merged[fact.item] = catalog.make_item(value, "human", "confirmed", note=note, at=at) + merged[kept.item] = catalog.make_item(value, "human", "confirmed", note=note, at=at) try: - await catalog.write_entry(session, row.id, fact.table, merged, if_version=version, actor=actor) + await catalog.write_entry(session, row.id, kept.table, merged, if_version=version, actor=actor) except catalog.CatalogConflict: await session.rollback() continue except catalog.CatalogInvalid as e: - return False, f"「{fact.table}」的{label}未写入数据目录:{e}" - return True, None - return False, f"「{fact.table}」的数据目录正被修改,{label}未能写入,请稍后在数据目录中确认" + return False, [*reasons, f"「{kept.table}」的{label}未写入数据目录:{e}"] + return True, reasons + return False, [*reasons, f"「{fact.table}」的数据目录正被修改,{label}未能写入,请稍后在数据目录中确认"] __all__ = ["FACT_ITEMS", "Fact", "derive", "note_of", "preview", "record_choices"] diff --git a/backend/tests/test_dual_check_facts.py b/backend/tests/test_dual_check_facts.py index d75bafe6..6f8067f8 100644 --- a/backend/tests/test_dual_check_facts.py +++ b/backend/tests/test_dual_check_facts.py @@ -7,9 +7,10 @@ - 指标自己的筛选范围(只看华东、只看线上)不是有效记录条件:取值出现在指标名称、口径说明、所在分组、运行输入里的, 条件用到分组的列的,不是在排除退款、作废、测试这类无效记录的,都不写 - 写的是人工填写、已确认,备注写明运行、节点、指标;同样的说法已确认过就不再写(不升版本),确认过的不同说法不覆盖; - 目录里已有的有效记录条件被采纳的一边也用着的,合起来写,没用着的不动 + 目录里已有的有效记录条件被采纳的一边也用着的,合起来写(已有的段只写一次),没用着的不动;驳回过的说法不写回 + (!= 和 <>、列名大小写、先后不同也认得出) - 同一次裁决里几个指标落到同一项上:说法一样的写一次;有效记录条件(都在排除无效记录)用 AND 合成一句写;业务日期 - 说法不同的都不写 + 说法不同的都不写;自己的说法驳回过的指标拿出来不写、其余照写(和各自的预告一致),合起来又成了驳回的整句也不写 - 审批卡上的预告(preview):选 a、选 b 各会写什么、不写的为什么,按目录里已有的说法算,和真写同一套判断 差异条目一律由 sql_semantics.explain_sql_diff 真算出来,和节点交给审核人看的是同一份。表名、数字都是虚构的。 @@ -422,6 +423,110 @@ async def test_combined_filters_extend_an_inferred_condition_every_adopted_side_ assert (await entry_of(source_id)).notes["valid_filter"]["value"] == "channel <> 'test'" +async def test_a_rejected_condition_is_not_written_back_through_a_combined_decision(shop): + """目录里「status <> 'refunded'」被人工驳回过。同一次裁决里「销售额」「订单数」采纳了排除退款、「订单金额」采纳了 + 排除测试单:审批卡上预告前两个不写(曾被人工驳回)、后一个写「is_test = 0」,真写也照这样。驳回过的条件不能借着 + 合并又写成人工确认,拿出来的指标照实说原因。""" + name, source_id = shop + rejected = {"valid_filter": catalog.make_item("status <> 'refunded'", "llm", "rejected")} + await seed(source_id, rejected) + refunds = candidate(SUM_WEEK, extra("status <> 'refunded'"), "b", metric="销售额", source=name) + tests = candidate(SUM_WEEK, extra("is_test = 0"), "b", metric="订单金额", source=name) + counted = candidate(COUNT_WEEK, extra("status <> 'refunded'", COUNT_WEEK), "b", metric="订单数", source=name) + notes = {"orders": (await entry_of(source_id)).notes} + assert preview(refunds, SCHEMA, notes)["b"]["write"] is False + assert preview(tests, SCHEMA, notes)["b"]["value"] == "is_test = 0" + out = await record_choices([refunds, tests, counted], run_id="run-1", node_title="复核", actor=None) + assert out["written"] == [{"source": name, "table": "orders", "item": "valid_filter"}], out + [skipped] = out["skipped"] + assert "「status <> 'refunded'」曾被人工驳回" in skipped["reason"] and "「销售额」、「订单数」" in skipped["reason"] + assert "订单金额" not in skipped["reason"] + item = (await entry_of(source_id)).notes["valid_filter"] + assert item["value"] == "is_test = 0" and (item["source"], item["status"]) == ("human", "confirmed") + assert "「订单金额」" in item["note"] and "销售额" not in item["note"] and "订单数" not in item["note"] + + # 并成一项的每个指标采纳的都是驳回过的说法:一个都不写,一条原因说清是哪几个指标 + await seed_again(source_id, rejected) + out = await record_choices([refunds, counted], run_id="run-2", node_title="复核", actor=None) + assert out["written"] == [] and len(out["skipped"]) == 1, out + assert "曾被人工驳回" in out["skipped"][0]["reason"] and "「销售额」、「订单数」" in out["skipped"][0]["reason"] + assert (await entry_of(source_id)).notes["valid_filter"]["status"] == "rejected" + + +async def test_a_rejected_statement_is_not_rebuilt_from_several_metrics(shop): + """驳回的是「is_test = 0 AND status != 'refunded'」整句:单看每个指标都不是这句(预告各写各的),合起来却正是这句 + (!= 和 <>、先后不同也是同一个说法)。两次人工判断对不上,都不写;再并进一个别的条件,驳回的整句仍在里面,照样不写。""" + name, source_id = shop + rejected = catalog.make_item("is_test = 0 AND status != 'refunded'", "llm", "rejected") + await seed(source_id, {"valid_filter": rejected}) + refunds = candidate(SUM_WEEK, extra("status <> 'refunded'"), "b", metric="销售额", source=name) + tests = candidate(SUM_WEEK, extra("is_test = 0"), "b", metric="订单金额", source=name) + channel = candidate(COUNT_WEEK, extra("channel <> 'test'", COUNT_WEEK), "b", metric="订单数", source=name) + notes = {"orders": (await entry_of(source_id)).notes} + assert all(preview(c, SCHEMA, notes)["b"]["write"] is True for c in (refunds, tests, channel)) + for picked, names in (([refunds, tests], "「销售额」、「订单金额」"), + ([refunds, tests, channel], "「销售额」、「订单金额」、「订单数」")): + out = await record_choices(picked, run_id="run-1", node_title="复核", actor=None) + assert out["written"] == [], out + [skipped] = out["skipped"] + assert "曾被人工驳回" in skipped["reason"] and names in skipped["reason"] + entry = await entry_of(source_id) + assert entry.version == 1 and entry.notes["valid_filter"]["status"] == "rejected" + + +@pytest.mark.parametrize("rejected", ["status != 'refunded'", "STATUS <> 'refunded'"]) +async def test_a_rejected_condition_is_recognised_however_it_was_written(shop, rejected): + """驳回时写的是 != 、列名大写:和这次要写的「status <> 'refunded'」是同一个说法,预告和真写都不写。""" + name, source_id = shop + await seed(source_id, {"valid_filter": catalog.make_item(rejected, "llm", "rejected")}) + picked = candidate(WITH_REFUNDS, RIGHT, "b", source=name) + said = preview(picked, SCHEMA, {"orders": (await entry_of(source_id)).notes}) + assert said["b"]["write"] is False and f"「{rejected}」曾被人工驳回" in said["b"]["reason"] + out = await record_choices([picked], run_id="run-1", node_title="复核", actor=None) + assert out == {"written": [], "skipped": [{"reason": said["b"]["reason"]}]} + assert (await entry_of(source_id)).version == 1 + + +async def test_combined_filters_do_not_repeat_a_condition_already_in_the_catalog(shop): + """目录里推断的「status <> 'refunded'」:「销售额」采纳了带着它的盲算,「订单金额」两边都按它筛、盲算另外排除了 + 测试单。合起来写时已有的条件只出现一次(正是「订单金额」的预告);已有的条件是人工确认过的,原因里只说另外多出 + 的那一段。""" + name, source_id = shop + await seed(source_id, {"valid_filter": catalog.make_item("status <> 'refunded'", "llm")}) + base = extra("status <> 'refunded'") + picked = [candidate(SUM_WEEK, base, "b", metric="销售额", source=name), + candidate(base, extra("is_test = 0", base), "b", metric="订单金额", source=name)] + said = preview(picked[1], SCHEMA, {"orders": (await entry_of(source_id)).notes}) + out = await record_choices(picked, run_id="run-1", node_title="复核", actor=None) + assert out == {"written": [{"source": name, "table": "orders", "item": "valid_filter"}], "skipped": []} + item = (await entry_of(source_id)).notes["valid_filter"] + assert item["value"] == said["b"]["value"] == "status <> 'refunded' AND is_test = 0" + + await seed_again(source_id, {"valid_filter": catalog.make_item("status <> 'refunded'", "human", "confirmed")}) + out = await record_choices(picked, run_id="run-2", node_title="复核", actor=None) + assert out["written"] == [] + reason = out["skipped"][0]["reason"] + assert "已有人工确认过的说法「status <> 'refunded'」" in reason and "另外多出条件「is_test = 0」" in reason + assert (await entry_of(source_id)).notes["valid_filter"]["value"] == "status <> 'refunded'" + + +def test_conditions_are_joined_without_repeating_a_part(): + """几句条件接成一句:已经有的段不再接(整句接上会把已有的段再写一遍);只说多出的时不带原有的;最外层是 OR 的 + 只在接了别的条件时加括号。""" + from app.engine.dual_check_facts import _and_join + + assert _and_join("status <> 'refunded'", ["status <> 'refunded' AND is_test = 0"], "sqlite") == \ + "status <> 'refunded' AND is_test = 0" + assert _and_join("status <> 'refunded'", ["STATUS != 'refunded'", "is_test = 0", "is_test = 0"], "sqlite") == \ + "status <> 'refunded' AND is_test = 0" + assert _and_join("status <> 'refunded'", ["status <> 'refunded' AND is_test = 0"], "sqlite", + keep_base=False) == "is_test = 0" + assert _and_join("status = 'void' OR is_test = 1", ["(status = 'void' OR is_test = 1)"], "sqlite") == \ + "status = 'void' OR is_test = 1" + assert _and_join("status = 'void' OR is_test = 1", ["is_test = 0"], "sqlite") == \ + "(status = 'void' OR is_test = 1) AND is_test = 0" + + async def test_the_same_choice_from_two_metrics_is_written_once(shop): name, source_id = shop out = await record_choices([candidate(BY_ORDER_DATE, RIGHT, "b", metric="销售额", source=name), diff --git a/backend/tests/test_dual_check_node.py b/backend/tests/test_dual_check_node.py index 5ed4c487..1aa09683 100644 --- a/backend/tests/test_dual_check_node.py +++ b/backend/tests/test_dual_check_node.py @@ -7,9 +7,10 @@ - 不一致:挂起,审批 payload 和对照工件按约定,payload 另带实际生效的 on_reject、摘要里每个指标带选 a、选 b 各会往 数据目录里写什么的预告;a / b / neither 三种选择的产出和缺口,没选的照实记,驳回、on_reject 为 fail 时失败 - 差异说明只比这一格所在的那一列(一条查询算好几列、分组后取一行的,收窄了再比,HAVING 一并去掉);分组的列设了 - 遮罩时不收窄,遮罩列的取值不进审批卡、对照工件、产出和事件 + 遮罩时(改了名再分组的也算)不收窄,遮罩列的取值不进审批卡、对照工件、产出和事件 - 只差一个方面时写数据目录(业务日期、有效记录条件),差了不止一处不写,人工确认过的不同说法不覆盖,指标自己的 - 筛选范围(取值在运行输入里、用到分组的列)不写,两条 SQL 规范化后一致的不写 + 筛选范围(取值在运行输入里、用到分组的列)不写,两条 SQL 规范化后一致的不写;人工驳回过的条件不会借着和别的 + 指标并成一项写回去,真写的和审批卡上的预告一致 - 全局关闭直通、节点覆盖;设置按发起时的快照;重放时盲算不再执行、事件不重复;报告的写作目录不含盲算的查询 - 触顶、分组结果的维度(只给取值、遮罩列不给)、带着取舍的列名不交给盲算、两个指标用到同一格只盲算一次、遮罩列上的 盲算结果不采用 @@ -588,6 +589,37 @@ async def test_a_confirmed_catalog_entry_is_not_overwritten(monkeypatch, shop, c assert entry.version == 1 and entry.notes["business_date"]["value"] == {"column": "order_date"} +async def test_a_rejected_filter_is_not_written_back_along_with_another_metric(monkeypatch, shop, client, tmp_path): + """目录里「status <> 'refunded'」被人工驳回过。同一次裁决里「销售额」采纳了排除退款的盲算、「订单金额」采纳了排除 + 测试单的盲算:审批卡上预告前者不写、后者写「is_test = 0」。真写的和预告一字不差——驳回过的条件不会借着两个指标 + 并成一项又写成人工确认。""" + name, source_id = shop + db = sqlite3.connect(tmp_path / "shop.db") + db.execute("INSERT INTO orders VALUES (6, '华东', 105, 40.0, 'paid', '2026-09-03', '2026-09-03', 'online', 1)") + db.commit() + db.close() + async with SessionLocal() as session: + await catalog.write_entry(session, source_id, "orders", { + "valid_filter": catalog.make_item("status <> 'refunded'", "human", "rejected")}, if_version=0, actor="数据组") + original = f"SELECT SUM(amount) AS gmv, SUM(amount) AS total FROM orders WHERE paid_date {WEEK}" + no_test = f"SELECT SUM(amount) AS total FROM orders WHERE is_test = 0 AND paid_date {WEEK}" + total = {**GMV, "id": "total", "name": "订单金额", "expression": "cell(nodes.fetch, 0, 'total')"} + Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv"), "订单金额": (no_test, "total")}) + run = await start(graph(name, original, metrics=[GMV, total], dual={"definitions": {}})) + assert run.status == "interrupted", run.error + said = {s["metric_id"]: s["catalog"]["b"] for s in (await pending(run.id)).payload["summary"]} + assert said["gmv"]["write"] is False and "曾被人工驳回" in said["gmv"]["reason"] + assert said["total"]["write"] is True and said["total"]["value"] == "is_test = 0" + done = await decide(client, run.id, {"gmv": "b", "total": "b"}) + assert done.status == "succeeded", done.error + [wrote] = logs(await events(run.id), "dual_check_catalog") + assert wrote["written"] == [{"source": name, "table": "orders", "item": "valid_filter"}] + assert [s["reason"] for s in wrote["skipped"]] == [said["gmv"]["reason"]] + item = (await notes_of(source_id)).notes["valid_filter"] + assert item["value"] == said["total"]["value"] and (item["source"], item["status"]) == ("human", "confirmed") + assert "「订单金额」" in item["note"] and "「销售额」" not in item["note"] + + async def test_a_metric_scope_filter_is_not_written_as_a_valid_filter(monkeypatch, shop, client): """「华东销售额」原流程按指标名只看华东,盲算漏了地区:审核人选原流程,但 region = '华东' 是这个指标自己的筛选 范围,写成表的有效记录条件,以后所有查订单表的工作流(包括下一次盲算)都会只看华东。""" @@ -845,6 +877,26 @@ async def test_masked_group_values_stay_out_of_the_card_and_the_catalog_log(monk assert wrote["written"] == [] and "不止差一个方面" in wrote["skipped"][0]["reason"] +async def test_a_renamed_masked_group_column_is_not_narrowed_either(monkeypatch, masked_shop, client): + """原流程把遮罩列改了名再分组(region AS area):结果列名 area 不在遮罩里,分组值却还是 region 的取值。收窄时看的 + 是分组键用到的列,一样不拿这一行的分组值补等值条件,「华东」不进审批卡、对照工件、产出和事件。""" + name, _ = masked_shop + renamed = (f"SELECT region AS area, SUM(amount) AS gmv FROM orders WHERE status <> 'refunded' " + f"AND paid_date {WEEK} GROUP BY region ORDER BY region") + first = {**GMV, "id": "first", "name": "首组销售额", "expression": "cell(nodes.fetch, 0, 'gmv')"} + Blind(monkeypatch, name, {"首组销售额": (RIGHT, "gmv")}) + run = await start(graph(name, renamed, metrics=[first], dual={"definitions": {}})) + assert run.status == "interrupted", run.error + payload = (await pending(run.id)).payload + assert payload["summary"][0]["original_rendered"].startswith("300") + card = artifact_store.load(payload["card_artifact"]) + done = await decide(client, run.id, {"first": "a"}) + assert done.status == "succeeded", done.error + _no_masked_value(payload, card, done, await events(run.id)) + # 照整条比:原流程分组、盲算没分组都列出来 + assert len(payload["summary"][0]["diff"]) > 1 + + async def test_a_masked_group_column_of_the_blind_query_is_not_narrowed_either(monkeypatch, masked_shop, client): """盲算自己按遮罩列分组、交了华东那一行:同样不拿这一行的分组值改写盲算的 SQL。""" name, _ = masked_shop