diff --git a/backend/app/data/catalog.py b/backend/app/data/catalog.py index 01eb87c1..d6016713 100644 --- a/backend/app/data/catalog.py +++ b/backend/app/data/catalog.py @@ -1516,6 +1516,10 @@ CATALOG_DRAFT_SYSTEM = """你在为一个业务数据库编写数据目录：说 log（日志表）、config（配置表）之一。 - keys：业务主键，即在业务上唯一确定一行的列（可以和数据库主键不同），写列名。 - business_date：业务日期按哪一列算，rule 写一句规则，timezone 写时区（不确定就留空）。 +- valid_filter：有效行过滤。condition 写一段 SQL 条件，筛出业务上算数的行（例如 STATUS NOT IN ('C', 'X')，去掉作废、\ +测试的行）：只用这张表自己的列、不带表名，只写比较、IN、BETWEEN、LIKE、IS NULL，用 AND、OR、NOT 连接，函数只用 UPPER、\ +LOWER、TRIM、COALESCE，不写子查询、聚合、窗口函数和分号；reason 写一句依据（例如哪条注释、哪些码值说明了这一点）。 +- dedup：去重规则，rule 写一句话（例如「同一订单号有多行时只取最后修改的那一行」），reason 写一句依据。 - columns：每一列的 label（中文名）、meaning（含义，一句话）、unit（单位，例如「元」「人」，没有单位就留空）、\ measure（度量类型：flow 流量，可跨期加总；stock 存量，时点值，不能跨期加总；ratio 比率；identifier 标识；\ status 状态；attribute 属性）。 @@ -1523,6 +1527,8 @@ status 状态；attribute 属性）。 规矩： - 只写数据事实，不写计算公式，不写指标怎么算。 - 拿不准的字段留空，不要猜。列名、表名照原样写，不要翻译或改写。 +- 有效行过滤、去重规则只在表结构、注释、码值能看出来时才写（例如状态列的注释写明了哪些取值是作废、测试），拿不准就不写，\ +没有就不写：写错了会把算数的行筛掉、把不该合并的行合并。 - 请你写的每一列都在 columns 里写一项，拿不准的字段可以留空，但不要漏掉列：漏掉的列会被再问一次。 - 标着「系统说明」的内容是导入时按核对结果生成的，以它为准：不要重复，也不要与之矛盾。 - 标着「已确认」的目录项由人工确认过，以它为准。 @@ -1541,6 +1547,7 @@ CATALOG_DRAFT_BACKGROUND_PROMPT = "使用者补充的背景和要求：\n{lines} CATALOG_DRAFT_TEXT_PROMPT = """只输出一个 JSON 对象，不要输出别的文字，格式如下（拿不准的字段不写）： {"tables": [{"table": "表名", "label": "中文名", "grain": "一行代表什么", "kind": "表类型", "keys": ["列名"], \ "business_date": {"column": "列名", "rule": "一句规则", "timezone": "时区"}, \ +"valid_filter": {"condition": "SQL 条件", "reason": "一句依据"}, "dedup": {"rule": "一句话", "reason": "一句依据"}, \ "columns": [{"name": "列名", "label": "中文名", "meaning": "含义", "unit": "单位", "measure": "度量类型"}]}]} 表类型写 fact、dimension、snapshot、log、config 之一；度量类型写 flow、stock、ratio、identifier、status、attribute 之一。""" @@ -1560,11 +1567,14 @@ CATALOG_DRAFT_PROMPT_LABELS = { "unique": "唯一约束：{cols}", "confirmed": "已确认的目录项：", # 分段起草：表名下一行说这次只写哪部分（宽表的表级一段、列分段，拆小重问时的半段） - "only_table": "这次只写表级的项（中文名、粒度、表类型、业务主键、业务日期），不写列", + "only_table": "这次只写表级的项（中文名、粒度、表类型、业务主键、业务日期、有效行过滤、去重规则），不写列", "only_columns": "这次只写下面列出的 {n} 列（这张表共 {total} 列），不写表级的项", "table_and_columns": "这次只写表级的项和这 {n} 列：{cols}", "brief_columns": "列（共 {n} 列，这里只列名字和类型）：", "drafted": "已起草的表级内容：", + # 开了数据剖析的源：目录里记下的码值（剖析看到的取值和已知含义）交给模型，判断有效行过滤用；遮罩的列不给 + "codes": "目录里记下的码值（数据剖析看到的取值和已知含义，可据此判断有效行过滤）：", + "code": "- {name}：{codes}", # 使用者补充的背景和要求 "description": "数据源说明：{text}", "guidance": "起草要求：{text}", @@ -1592,6 +1602,20 @@ CATALOG_DRAFT_PROMPT_SCHEMA: dict[str, Any] = { "timezone": {"type": "string"}}, "required": ["column"], }, + "valid_filter": { + "type": "object", + "description": "有效行过滤：只在表结构、注释、码值能看出来时才写", + "properties": {"condition": {"type": "string", "description": "SQL 条件，只用这张表的列"}, + "reason": {"type": "string", "description": "一句依据"}}, + "required": ["condition"], + }, + "dedup": { + "type": "object", + "description": "去重规则：只在表结构、注释能看出来时才写", + "properties": {"rule": {"type": "string", "description": "一句话"}, + "reason": {"type": "string", "description": "一句依据"}}, + "required": ["rule"], + }, "columns": { "type": "array", "items": { @@ -1712,11 +1736,65 @@ def _clean_text(value: Any, limit: int = _TEXT_MAX) -> str | None: return text if text and len(text) <= limit else None -def _from_model(entry: Mapping[str, Any], meta: Mapping[str, Any], *, system: bool) -> dict[str, Any]: +#: 模型给的依据（有效行过滤、去重规则的 reason）写进这一项的 note +FACT_REASON_NOTE = "依据：{reason}" + + +def _model_fact(raw: Any, key: str) -> tuple[str, str | None] | None: + """模型给的有效行过滤、去重规则 → (取值, 依据)。约定是对象（{condition 或 rule, reason}），只写了一段文字的也认； + 取值不是非空文字的不要。""" + if isinstance(raw, str): + text, reason = raw, None + elif isinstance(raw, Mapping): + text, reason = raw.get(key), raw.get("reason") + if text is None: + text = raw.get("value") + else: + return None + if not isinstance(text, str) or not text.strip(): + return None + return text.strip(), _clean_text(reason) + + +def _fact_item(value: str, reason: str | None) -> dict[str, Any]: + return make_item(value, "llm", note=FACT_REASON_NOTE.format(reason=reason) if reason else None) + + +def _model_facts(entry: Mapping[str, Any], meta: Mapping[str, Any], *, kind: str | None, + table: str | None) -> dict[str, Any]: + """模型提议的有效行过滤、去重规则（llm / proposed，依据写进 note）。 + + 有效行过滤按数据源的方言解析成一个条件、只能用这张表的列（sqlcheck.parse_valid_filter），不过就丢掉、日志记一条， + 不算这张表失败；存的是模型写的原文（只去掉首尾空白，字符串里的空格不动）。去重规则是一句话，同文字项的规矩。""" + from app.data import sqlcheck + + out: dict[str, Any] = {} + found = _model_fact(entry.get("valid_filter"), "condition") + if found is not None: + text, reason = found + problem = sqlcheck.valid_filter_problem(text, meta, kind=kind, table=table) + if problem is None: + out["valid_filter"] = _fact_item(text, reason) + else: + import logging + + logging.getLogger(__name__).info("起草数据目录：表 %s 的有效行过滤没有通过校验，已丢掉：%s（%s）", + table or meta.get("qualified") or entry.get("table"), problem, text[:200]) + found = _model_fact(entry.get("dedup"), "rule") + if found is not None and (text := _clean_text(found[0])) is not None: + out["dedup"] = _fact_item(text, found[1]) + return out + + +def _from_model(entry: Mapping[str, Any], meta: Mapping[str, Any], *, system: bool, source_kind: str | None = None, + table: str | None = None) -> dict[str, Any]: """模型给的一张表 → 目录项（llm / proposed）。不合规的值整项丢掉，不硬塞：列名对不上表结构的、 表类型和度量类型不在取值里的、业务主键里有不存在的列的、看起来是计算公式的（_formula_like，公式进口径卡）。 该是文字的地方写成了数组、对象（列名、表类型、业务日期的列、度量类型……）同样丢掉，不抛异常：实测见过把列名 - 写成数组的，以前在这里抛 TypeError，整张表记出错。导入表格的源，说明已经覆盖的字段也丢掉。""" + 写成数组的，以前在这里抛 TypeError，整张表记出错。导入表格的源，说明已经覆盖的字段也丢掉。 + + 有效行过滤、去重规则见 _model_facts：source_kind 是数据源类型（按它的方言解析条件，不认识的类型不收有效行过滤）， + table 是表名（条件里的列可以带这张表的表名）。""" columns = {c["name"]: c for c in meta.get("columns") or []} notes: dict[str, Any] = {} table_has_system_note = system and bool((meta.get("comment") or "").strip()) @@ -1738,6 +1816,7 @@ def _from_model(entry: Mapping[str, Any], meta: Mapping[str, Any], *, system: bo if (text := _clean_text(date.get(k))) is not None: value[k] = text notes["business_date"] = make_item(value, "llm") + notes.update(_model_facts(entry, meta, kind=source_kind, table=table)) out_cols: dict[str, dict[str, Any]] = {} listed = entry.get("columns") for col in listed if isinstance(listed, list) else []: @@ -2330,7 +2409,7 @@ __all__ = [ "CatalogEntry", "CatalogInvalid", "CatalogModelUnavailable", "CatalogPathError", "DRAFT_DEFAULT_TABLES", "DraftModel", "DraftReport", "INFERRED_MARK", "ITEM_SOURCES", "ITEM_STATUSES", "JoinEdge", "KIND_LABEL", "MEASURE_LABEL", "MODEL_BATCH_COLUMNS", "MODEL_BATCH_TABLES", "MODEL_CONCURRENCY", "MODEL_EMPTY_REASON", - "MODEL_FALLBACK_MAX_TOKENS", "MODEL_HALT_ON", "MODEL_HALT_WINDOW", "MODEL_MAX_TOKENS", + "FACT_REASON_NOTE", "MODEL_FALLBACK_MAX_TOKENS", "MODEL_HALT_ON", "MODEL_HALT_WINDOW", "MODEL_MAX_TOKENS", "MODEL_MIN_PART_COLUMNS", "MODEL_NOTICE_OFF", "MODEL_NOTICE_ON", "MODEL_NOTICE_WINDOW", "MODEL_PART_MAX_CALLS", "MODEL_RETRIES", "MODEL_RETRY_DELAYS_S", "MODEL_SHRINK_AFTER", "MODEL_SPLIT_DEPTH", "MODEL_SPLIT_MIN_COLUMNS", "MODEL_TABLE_GIVE_UP_AFTER", "MODEL_TIMEOUT_S", "MergeStats", "NEW_RELATION_PATH", diff --git a/backend/app/data/catalog_draft.py b/backend/app/data/catalog_draft.py index 49e059b1..07e83877 100644 --- a/backend/app/data/catalog_draft.py +++ b/backend/app/data/catalog_draft.py @@ -75,8 +75,15 @@ 一张表的模型部分全部结束。后台起草任务据此落库（data/catalog_jobs.py）。块：这张表实际分成几块来问——一开始是段数， 一段被缩小的上限分开成几块、被拆成两半，块数就多几块；一块问完（写没写出来都算）算结束一块。 -交给模型的只有表结构（列、类型、主键、外键、唯一约束、注释）和使用者补充的背景，不发任何数据值。写给模型的文字都在 -catalog.py 名字带 _SYSTEM / _PROMPT 的常量里；这里的 REASON_* 是给人看的（拼进这张表的 model_error）。 +**防错事实。** 表级的项里模型可以提议有效行过滤（valid_filter）和去重规则（dedup），各附一句依据（写进这一项的 note）， +校验见 catalog._from_model（有效行过滤按方言解析，不过就丢掉）。开了数据剖析的源（catalog_profile.profile_settings）： +目录里记下的码值交给模型做背景（遮罩的列不给），表级那段写完后经剖析的只读守卫数一数有效行过滤排除了多少行 +（catalog_profile.check_valid_filter），排除 0% 或 100% 的不收，收下的 note 写上排除比例；没开剖析、连不上的照收， +note 写没有核对的原因。 + +交给模型的是表结构（列、类型、主键、外键、唯一约束、注释）和使用者补充的背景；开了数据剖析的源另给目录里记下的码值 +（剖析取过的取值和含义，遮罩的列不给），别的数据值一概不发。写给模型的文字都在 catalog.py 名字带 _SYSTEM / _PROMPT +的常量里；这里的 REASON_* 是给人看的（拼进这张表的 model_error）。 """ from __future__ import annotations @@ -749,6 +756,8 @@ class _Table: #: 块进度：实际分成几块、结束了几块 chunks_total: int = 0 chunks_done: int = 0 + #: 有效行过滤核对过了（表级那段写完核对一次） + vetted: bool = False def whole(self, piece: _Piece) -> bool: return piece.table_level and len(piece.columns) == len(self.columns) @@ -863,6 +872,15 @@ class _Run: self.was_halted = False #: 这次起草里结构化输出只给空壳时换纯文本问过没有（止损期间只问这一次） self.text_probed = False + #: 数据源：开了数据剖析的源把码值交给模型、核对有效行过滤的排除比例（遮罩的列都不碰） + from app.data import catalog_profile + from app.data.engine import masked_columns + + self.source = source + self.kind = getattr(source, "kind", None) + options = getattr(source, "options", None) + self.profiling = catalog_profile.profile_settings(options if isinstance(options, dict) else None) + self.masked = {c.lower() for c in masked_columns(options if isinstance(options, dict) else None)} def _halted(self) -> bool: """止损：最近 MODEL_HALT_WINDOW 次调用里有 MODEL_HALT_ON 次以上没写出来。之后的段不再拆小、每块只问一次，直到 @@ -949,15 +967,18 @@ class _Run: """起草一段；它之后的段（宽表的列分段）排到队尾；一张表的段都结束了，这张表的模型部分就结束了。 这一段（原始段：计划里的整表段、表级那段或一个列分段）连同按缩小后的上限分开的各块、拆出来的、补问的、重试的， - 一共最多调用 MODEL_PART_MAX_CALLS 次（各块共用一个 _Budget）。接连几段没写出来的表（_tally），剩下的段跳过不问。 - 宽表写出了一段、后面还有段没结束，就把这张表到这时拿到的内容交给调用方并入目录（part_written）；最后一段结束由 - table_done 收尾。""" + 一共最多调用 MODEL_PART_MAX_CALLS 次（各块共用一个 _Budget）。表级那段写完先核对模型提议的有效行过滤（_vet）， + 再交出去。接连几段没写出来的表（_tally），剩下的段跳过不问。宽表写出了一段、后面还有段没结束，就把这张表到这时 + 拿到的内容交给调用方并入目录（part_written）；最后一段结束由 table_done 收尾。""" asked = [p for p in work.pieces if not self.tables[p.table].abandoned] for piece in work.pieces: if self.tables[piece.table].abandoned: await self._skip_part(piece) if asked: await self._start(asked, _Budget(max(1, int(C.MODEL_PART_MAX_CALLS)))) + for piece in asked: + if piece.table_level: + await self._vet(self.tables[piece.table]) self._tally(asked) for piece in work.pieces: await self._part_done(piece.table) @@ -1408,7 +1429,8 @@ class _Run: gain = _Gain() for entry in by_table.get(piece.table, []): gain.named = st.named = True - notes = _restrict(C._from_model(entry, st.meta, system=self.system), piece) + drafted = C._from_model(entry, st.meta, system=self.system, source_kind=self.kind, table=st.name) + notes = _restrict(drafted, piece) cols = entry.get("columns") if isinstance(entry.get("columns"), list) else [] gain.answered |= {c.get("name") for c in cols if isinstance(c, dict) and c.get("name") in piece.columns} slots = C._slots(notes) @@ -1572,6 +1594,47 @@ class _Run: st = self.tables[name] await self._hook(self.hooks.part_written, name, PartWrite(copy.deepcopy(st.notes), len(st.written_parts))) + # ---------------------------------------------------------------- 防错事实：核对有效行过滤 + + async def _vet(self, st: _Table) -> None: + """表级那段写完，核对模型提议的有效行过滤（只核对一次）。 + + - 目录里人工确认、驳回过的有效行过滤：并入时不会采用这次的提议，不必再查库； + - 没开数据剖析：照收，note 写「没有核对排除比例（数据源没有开数据剖析）」； + - 开了：经剖析的只读守卫数排除比例（catalog_profile.check_valid_filter）。排除 0% 或 100% 的不收（日志记一条， + 不算失败）；收下的 note 在依据后面写上剖析核对的结果，或者没有核对的原因。""" + if st.vetted: + return + st.vetted = True + item = st.notes.get("valid_filter") + if not isinstance(item, dict): + return + held = (st.existing or {}).get("valid_filter") + if isinstance(held, Mapping) and held.get("status") in ("confirmed", "rejected"): + return + from app.data import catalog_profile + + if not self.profiling.enabled: + found = catalog_profile.FilterCheck(True, catalog_profile.FILTER_UNCHECKED.format( + why=catalog_profile.FILTER_NOT_ENABLED)) + else: + try: + found = await catalog_profile.check_valid_filter(self.source, st.name, dict(st.meta), + str(item.get("value") or ""), settings=self.profiling) + except asyncio.CancelledError: + raise + except Exception as e: # noqa: BLE001 - 核对是附加的：出了意外照收，写明没有核对 + from app.core.errors import describe_exception, first_line + + logger.exception("起草数据目录：核对表 %s 的有效行过滤时出错", st.name) + found = catalog_profile.FilterCheck(True, catalog_profile.FILTER_UNCHECKED.format( + why=f"核对时出错：{first_line(e) or describe_exception(e)}")) + if not found.keep: + logger.info("起草数据目录：表 %s 的有效行过滤 %s 没有收下：%s", st.name, item.get("value"), found.dropped) + st.notes.pop("valid_filter", None) + return + item["note"] = _sentences(item.get("note"), found.note) + @staticmethod async def _hook(fn: Callable[..., Awaitable[None]] | None, *args: Any) -> None: if fn is None: @@ -1650,7 +1713,12 @@ class _Run: for group in meta.get("unique") or []: if isinstance(group, list) and group: lines.append(labels["unique"].format(cols="、".join(group))) - confirmed = self._confirmed_lines(st, piece) + shown = self._confirmed_for(st, piece) + codes = self._codes_lines(st, piece, shown) + if codes: + lines.append(labels["codes"]) + lines += codes + confirmed = C._notes_lines(shown, meta=st.meta, system_notes=False) if confirmed: lines.append(labels["confirmed"]) lines += confirmed @@ -1671,7 +1739,7 @@ class _Run: return line + ("，" + "，".join(marks) if marks else "") @staticmethod - def _confirmed_lines(st: _Table, piece: _Piece) -> list[str]: + def _confirmed_for(st: _Table, piece: _Piece) -> dict[str, Any]: """已确认、有确证的项（以此为准的背景）：整表全给；其余给表级的项和这一段的列，关系只在要写表级的项时给。""" confirmed = C._confirmed_only(st.existing) if not st.whole(piece): @@ -1679,7 +1747,32 @@ class _Run: kept = {key: item for key, item in C._slots(confirmed).items() if key[0] == "t" or (key[0] == "c" and key[1] in wanted) or (key[0] == "r" and piece.table_level)} confirmed = C._assemble(kept) - return C._notes_lines(confirmed, meta=st.meta, system_notes=False) + return confirmed + + def _codes_lines(self, st: _Table, piece: _Piece, shown: Mapping[str, Any]) -> list[str]: + """开了数据剖析的源：目录里记下的码值（剖析取过的取值和已知含义）交给模型，判断有效行过滤、列的含义用。要写表级 + 的项的给全部列的，只写几列的给这几列的；遮罩的列、驳回的码值不给；「已确认的目录项」里已经有的不重复。""" + if not self.profiling.enabled: + return [] + labels = C.CATALOG_DRAFT_PROMPT_LABELS + wanted = set(st.columns) if piece.table_level else set(piece.columns) + known = C.visible_notes(st.existing).get("columns") or {} + given = (shown.get("columns") or {}) if isinstance(shown.get("columns"), Mapping) else {} + out: list[str] = [] + for col in st.columns: + item = (known.get(col) or {}).get("codes") + if col not in wanted or col.lower() in self.masked or not isinstance(item, Mapping): + continue + if "codes" in (given.get(col) or {}) or not isinstance(item.get("value"), Mapping) or not item["value"]: + continue + out.append(labels["code"].format(name=col, codes=C._codes_text(item["value"]))) + return out + + +def _sentences(*parts: Any) -> str: + """几句话连成一段：各句去掉句末的句号，用句号隔开（「依据：……。剖析核对：……」）。""" + texts = [str(p).strip().rstrip("。") for p in parts if isinstance(p, str) and p.strip()] + return "。".join(texts) __all__ = ["BUDGET_WORDS", "Background", "CallReport", "FAILURE_LABEL", "Hooks", "ModelOutcome", "NOTICE", "PartWrite", diff --git a/backend/tests/test_catalog_draft_facts.py b/backend/tests/test_catalog_draft_facts.py index dba6db48..c7c2cabd 100644 --- a/backend/tests/test_catalog_draft_facts.py +++ b/backend/tests/test_catalog_draft_facts.py @@ -154,6 +154,124 @@ def test_the_condition_is_rendered_with_the_columns_quoted_like_the_source(): assert rendered == "`isTest` = 0 AND status <> 'C'" +# -------------------------------------------------------------------------- +# 模型给的两项：校验、依据 +# -------------------------------------------------------------------------- + + +def test_the_model_fields_become_llm_items_with_their_reason(): + entry = {"table": "orders", "label": "订单", + "valid_filter": {"condition": "status NOT IN ('C', 'X')", "reason": "状态注释写明 C 作废、X 测试单"}, + "dedup": {"rule": "同一订单号有多行时只取最后修改的一行", "reason": "订单号可能重复导出"}} + notes = catalog._from_model(entry, META, system=False, source_kind="sqlite", table="orders") + vf, dd = notes["valid_filter"], notes["dedup"] + assert (vf["value"], vf["source"], vf["status"]) == ("status NOT IN ('C', 'X')", "llm", "proposed") + assert vf["note"] == "依据：状态注释写明 C 作废、X 测试单" + assert (dd["value"], dd["source"], dd["status"]) == ("同一订单号有多行时只取最后修改的一行", "llm", "proposed") + assert dd["note"] == "依据：订单号可能重复导出" + assert not catalog.validate_notes(notes) + + +def test_plain_strings_are_accepted_and_bad_filters_dropped(caplog): + entry = {"table": "orders", "label": "订单", "valid_filter": "status <> 'C'", "dedup": "按订单号去重"} + notes = catalog._from_model(entry, META, system=False, source_kind="sqlite", table="orders") + assert notes["valid_filter"]["value"] == "status <> 'C'" and "note" not in notes["valid_filter"] + assert notes["dedup"]["value"] == "按订单号去重" + + with caplog.at_level(logging.INFO, logger="app.data.catalog"): + notes = catalog._from_model({"table": "orders", "label": "订单", + "valid_filter": {"condition": "status IN (SELECT 1)", "reason": "猜的"}}, + META, system=False, source_kind="sqlite", table="orders") + assert "valid_filter" not in notes and notes["label"]["value"] == "订单", "丢掉这一项，别的照收" + assert any("有效行过滤" in r.getMessage() and "orders" in r.getMessage() for r in caplog.records), "日志记一条" + for junk in ({"condition": ""}, {"reason": "只有依据"}, ["status = 1"], 3): + notes = catalog._from_model({"table": "orders", "valid_filter": junk, "dedup": junk}, META, system=False, + source_kind="sqlite", table="orders") + assert "valid_filter" not in notes and "dedup" not in notes, junk + + +def test_the_prompt_asks_for_the_two_facts_only_when_evident(): + system = catalog.CATALOG_DRAFT_SYSTEM + assert "valid_filter" in system and "dedup" in system + assert "拿不准就不写" in system and "reason" in system + assert "UPPER、LOWER、TRIM、COALESCE" in system + props = catalog.CATALOG_DRAFT_PROMPT_SCHEMA["properties"]["tables"]["items"]["properties"] + assert set(props["valid_filter"]["properties"]) == {"condition", "reason"} + assert set(props["dedup"]["properties"]) == {"rule", "reason"} + assert '"valid_filter": {"condition"' in catalog.CATALOG_DRAFT_TEXT_PROMPT + assert '"dedup": {"rule"' in catalog.CATALOG_DRAFT_TEXT_PROMPT + assert "有效行过滤" in catalog.CATALOG_DRAFT_PROMPT_LABELS["only_table"] + + +# -------------------------------------------------------------------------- +# 引擎：没开剖析照收、写明没有核对；开了剖析的源把码值交给模型 +# -------------------------------------------------------------------------- + +REPLY = {"label": "订单", "grain": "每笔订单一行", + "valid_filter": {"condition": "status NOT IN ('C', 'X')", "reason": "状态注释写明 C 作废、X 测试单"}, + "dedup": {"rule": "同一订单号只取最后一行", "reason": "订单号可能重复"}, + "columns": [{"name": "status", "label": "订单状态"}]} + + +async def test_without_profiling_the_filter_is_kept_and_says_it_was_not_checked(): + model = FakeModel({"orders": REPLY}) + got = await D.run_model_draft(model, _plain({"orders": _orders()}), ["orders"]) + o = got["orders"] + assert o.status == "ok" + vf = o.notes["valid_filter"] + assert (vf["source"], vf["status"]) == ("llm", "proposed") + assert vf["note"] == "依据：状态注释写明 C 作废、X 测试单。没有核对排除比例（数据源没有开数据剖析）" + assert o.notes["dedup"]["note"] == "依据：订单号可能重复" + + +async def test_an_invalid_filter_from_the_model_is_dropped_not_failed(): + reply = {**REPLY, "valid_filter": {"condition": "status IN (SELECT code FROM bad_codes)", "reason": "猜的"}} + got = await D.run_model_draft(FakeModel({"orders": reply}), _plain({"orders": _orders()}), ["orders"]) + o = got["orders"] + assert o.status == "ok" and o.error is None, "不算失败" + assert "valid_filter" not in o.notes and o.notes["dedup"]["value"] == "同一订单号只取最后一行" + + +def _codes_existing() -> dict: + return {"orders": {"columns": { + "status": {"codes": make_item({"C": "作废", "X": "", "P": "已支付"}, "profile", "proposed")}, + "order_no": {"codes": make_item({"A1": "甲"}, "profile", "proposed")}, + "amount": {"codes": make_item({"0": "零"}, "profile", "rejected")}, + }}} + + +async def test_codes_go_to_the_model_when_profiling_is_on_but_not_masked_ones(): + on = {PROFILE_OPTION: {"enabled": True}, MASK_COLUMNS_OPTION: ["ORDER_NO"]} + model = FakeModel({"orders": REPLY}) + await D.run_model_draft(model, _plain({"orders": _orders()}, options=on), ["orders"], existing=_codes_existing()) + text = model.calls[0] + head = catalog.CATALOG_DRAFT_PROMPT_LABELS["codes"] + assert head in text + block = text[text.index(head):] + assert "- status：C=作废、P=已支付，X 的含义待填写" in block + assert "A1" not in text, "遮罩的列不给码值" + assert "零" not in text, "驳回的码值不给" + + plain = FakeModel({"orders": REPLY}) + await D.run_model_draft(plain, _plain({"orders": _orders()}), ["orders"], existing=_codes_existing()) + assert head not in plain.calls[0] and "已支付" not in plain.calls[0], "没开剖析不给码值" + + +async def test_codes_reach_the_table_part_of_a_wide_table_and_only_their_column_part(): + cols = [_col(f"c{i:03d}") for i in range(1, 46)] + [_col("state", "TEXT")] + wide = {"qualified": "wide", "schema": None, "is_view": False, "primary_key": ["c001"], "foreign_keys": [], + "comment": None, "columns": cols} + existing = {"wide": {"columns": {"state": {"codes": make_item({"9": "作废"}, "profile", "proposed")}}}} + on = {PROFILE_OPTION: {"enabled": True}} + model = FakeModel({"wide": {"label": "宽表"}}) + await D.run_model_draft(model, _plain({"wide": wide}, options=on), ["wide"], existing=existing) + head = catalog.CATALOG_DRAFT_PROMPT_LABELS["codes"] + with_codes = [requested(c)["wide"] for c in model.calls if "9=作废" in c] + assert ("table", None) in with_codes, "表级那段要提议有效行过滤：给全部码值" + assert all(part == "table" or "state" in (cols or []) for part, cols in with_codes) + assert any(head not in c for c in model.calls), "不含这一列的列分段不给" + + # -------------------------------------------------------------------------- # 剖析核对排除比例 # -------------------------------------------------------------------------- @@ -213,3 +331,55 @@ async def test_unchecked_filters_are_kept_with_the_reason(scenic, tmp_path): gone = await scenic(database=str(tmp_path / "missing" / "nope.db")) found = await catalog_profile.check_valid_filter(gone, "visits", meta_of(gone), "status = 1") assert found.keep and found.note.startswith("没有核对排除比例（查询失败：") + + +async def test_the_engine_checks_the_filter_on_a_profiled_source(scenic, scenic_db): + source = await scenic() + good = {"label": "入园记录", "valid_filter": {"condition": "status = 1", "reason": "0 是作废的检票"}} + got = await D.run_model_draft(FakeModel({"visits": good}), source, ["visits"]) + vf = got["visits"].notes["valid_filter"] + total, excluded = _count(scenic_db, "status = 1") + share = catalog_profile.excluded_share(excluded, total) + assert vf["note"] == f"依据：0 是作废的检票。剖析核对：排除 {share} 的行（共 {total:,} 行）" + + useless = {"label": "入园记录", "valid_filter": {"condition": "status IN (0, 1)", "reason": "猜的"}} + got = await D.run_model_draft(FakeModel({"visits": useless}), source, ["visits"]) + assert got["visits"].status == "ok" and "valid_filter" not in got["visits"].notes, "排除 0% 的不收" + assert got["visits"].notes["label"]["value"] == "入园记录" + + +async def test_a_confirmed_filter_is_not_checked_again(scenic, monkeypatch): + source = await scenic() + called: list[str] = [] + + async def counting(*args, **kw): + called.append("x") + return catalog_profile.FilterCheck(True, "不该用到") + + monkeypatch.setattr(catalog_profile, "check_valid_filter", counting) + existing = {"visits": {"valid_filter": make_item("status = 1", "human", "confirmed")}} + reply = {"label": "入园记录", "valid_filter": {"condition": "status <> 0", "reason": "0 是作废"}} + await D.run_model_draft(FakeModel({"visits": reply}), source, ["visits"], existing=existing) + assert called == [], "人工确认过的有效行过滤，起草的提议并入时不会采用，不必再查库" + + +# -------------------------------------------------------------------------- +# SQL 检查：推断的有效行过滤只是提示，确认后才是警告 +# -------------------------------------------------------------------------- + + +async def test_a_drafted_filter_only_hints_until_someone_confirms_it(scenic): + source = await scenic(options={}) + entry = {"table": "visits", "valid_filter": {"condition": "status = 1", "reason": "0 是作废"}} + meta = source.schema_cache["tables"]["visits"] + notes = catalog._from_model(entry, meta, system=False, source_kind="sqlite", table="visits") + sql = "SELECT COUNT(*) FROM visits" + + def levels(table_notes: dict) -> list[tuple[str, str]]: + checker = sqlcheck.SqlChecker(kind="sqlite", schema_cache=source.schema_cache, notes={"visits": table_notes}) + return [(c.code, c.level) for c in checker.check(sql)] + + assert levels(notes) == [("missing_valid_filter", "info")] + confirmed = catalog.review_item(notes, "valid_filter", "confirm") + assert levels(confirmed) == [("missing_valid_filter", "warning")] + assert levels(catalog.review_item(notes, "valid_filter", "reject")) == []