From d3bd80211a37c796327b0fc71ed2f2470c992f50 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Wed, 14 Oct 2026 10:00:10 +0800 Subject: [PATCH] =?UTF-8?q?=E6=95=B0=E6=8D=AE=E7=9B=AE=E5=BD=95=E9=97=AD?= =?UTF-8?q?=E7=8E=AF=E7=AC=AC=E4=B8=80=E6=AD=A5=EF=BC=88=E5=90=8E=E7=AB=AF?= =?UTF-8?q?=20A=EF=BC=89=E5=A4=8D=E6=A0=B8=E4=BF=AE=E6=AD=A3=EF=BC=9A?= =?UTF-8?q?=E6=B2=A1=E6=94=B6=E4=B8=8B=E7=9A=84=E6=9C=89=E6=95=88=E8=AE=B0?= =?UTF-8?q?=E5=BD=95=E6=9D=A1=E4=BB=B6=E4=B8=8D=E7=AE=97=E5=A4=B1=E8=B4=A5?= =?UTF-8?q?=EF=BC=8C=E5=9C=A8=E8=B5=B7=E8=8D=89=E4=BB=BB=E5=8A=A1=E9=80=90?= =?UTF-8?q?=E8=A1=A8=E7=9A=84=20log=20=E9=87=8C=E8=AE=B0=E4=B8=80=E5=8F=A5?= =?UTF-8?q?=EF=BC=8C=E6=8E=A5=E5=8F=A3=E7=9C=8B=E5=BE=97=E5=88=B0=EF=BC=88?= =?UTF-8?q?=E4=BB=A5=E5=89=8D=E5=8F=AA=E8=BF=9B=E6=9C=8D=E5=8A=A1=E7=AB=AF?= =?UTF-8?q?=E6=97=A5=E5=BF=97=EF=BC=89=E2=80=94=E2=80=94=E6=B2=A1=E9=80=9A?= =?UTF-8?q?=E8=BF=87=E6=A0=A1=E9=AA=8C=E7=9A=84=E5=86=99=E5=8E=9F=E5=9B=A0?= =?UTF-8?q?=EF=BC=88=E3=80=8C=E6=9C=89=E6=95=88=E8=AE=B0=E5=BD=95=E6=9D=A1?= =?UTF-8?q?=E4=BB=B6=E9=87=8C=E4=B8=8D=E8=83=BD=E6=9C=89=E5=AD=90=E6=9F=A5?= =?UTF-8?q?=E8=AF=A2=EF=BC=8C=E6=A8=A1=E5=9E=8B=E6=8F=90=E8=AE=AE=E7=9A=84?= =?UTF-8?q?=E8=BF=99=E4=B8=80=E6=9D=A1=E6=B2=A1=E6=9C=89=E6=94=B6=E4=B8=8B?= =?UTF-8?q?=EF=BC=9A=E2=80=A6=E2=80=A6=E3=80=8D=EF=BC=89=EF=BC=8C=E5=89=96?= =?UTF-8?q?=E6=9E=90=E6=A0=B8=E5=AF=B9=E6=8E=92=E9=99=A4=200%=20=E6=88=96?= =?UTF-8?q?=20100%=20=E7=9A=84=E5=86=99=E6=95=B0=E5=88=B0=E4=BA=86?= =?UTF-8?q?=E5=A4=9A=E5=B0=91=E8=A1=8C=EF=BC=9B=E5=8F=AA=E5=9C=A8=E8=AF=B7?= =?UTF-8?q?=E6=A8=A1=E5=9E=8B=E5=86=99=E8=A1=A8=E7=BA=A7=E7=9A=84=E9=A1=B9?= =?UTF-8?q?=E7=9A=84=E9=82=A3=E6=AE=B5=E9=87=8C=E8=AE=B0=EF=BC=8C=E4=B8=80?= =?UTF-8?q?=E5=BC=A0=E8=A1=A8=E6=9C=80=E5=A4=9A=2010=20=E5=8F=A5=E3=80=82C?= =?UTF-8?q?atalogDraftTask=20=E5=8A=A0=20log=20=E5=88=97=EF=BC=88=E5=AD=98?= =?UTF-8?q?=E9=87=8F=E4=BB=BB=E5=8A=A1=E5=9B=9E=E5=A1=AB=20NULL=EF=BC=8C?= =?UTF-8?q?=E6=8E=A5=E5=8F=A3=E7=BB=99=E7=A9=BA=E5=88=97=E8=A1=A8=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/data/catalog.py | 19 +++++---- backend/app/data/catalog_draft.py | 48 +++++++++++++++++++---- backend/app/data/catalog_jobs.py | 4 +- backend/app/db/base.py | 2 + backend/app/db/models.py | 2 + backend/tests/test_catalog_draft_facts.py | 29 +++++++++++++- backend/tests/test_catalog_draft_jobs.py | 17 ++++++++ 7 files changed, 103 insertions(+), 18 deletions(-) diff --git a/backend/app/data/catalog.py b/backend/app/data/catalog.py index 0157e4bd..a16446ab 100644 --- a/backend/app/data/catalog.py +++ b/backend/app/data/catalog.py @@ -1762,13 +1762,14 @@ 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]: +def _model_facts(entry: Mapping[str, Any], meta: Mapping[str, Any], *, kind: str | None, table: str | None, + dropped: list[tuple[str, str]] | None = None) -> dict[str, Any]: """模型提议的有效行过滤、去重规则（llm / proposed，依据写进 note）。 - 有效行过滤按数据源的方言解析成一个条件、只能用这张表的列（sqlcheck.parse_valid_filter），不过就丢掉、日志记一条， - 不算这张表失败。存的是 sqlcheck.stored_valid_filter 整理过的写法：原文照旧，只把列换成表结构里的真名、不带 - 表名，按数据源的需要加引号——剖析核对的、SQL 检查让模型照着加的都是这一条。去重规则是一句话，同文字项的规矩。""" + 有效行过滤按数据源的方言解析成一个条件、只能用这张表的列（sqlcheck.parse_valid_filter），不过就丢掉、不算这张表 + 失败：服务端日志记一条，dropped 给了就记下 (原因, 模型写的条件)，起草引擎把它写进这张表的日志（起草任务逐表的 + log）。存的是 sqlcheck.stored_valid_filter 整理过的写法：原文照旧，只把列换成表结构里的真名、不带表名，按数据源的 + 需要加引号——剖析核对的、SQL 检查让模型照着加的都是这一条。去重规则是一句话，同文字项的规矩。""" from app.data import sqlcheck out: dict[str, Any] = {} @@ -1789,6 +1790,8 @@ def _model_facts(entry: Mapping[str, Any], meta: Mapping[str, Any], *, kind: str logging.getLogger(__name__).info("起草数据目录：表 %s 的有效行过滤没有通过校验，已丢掉：%s（%s）", table or meta.get("qualified") or entry.get("table"), problem, text[:200]) + if dropped is not None: + dropped.append((problem, text)) 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]) @@ -1796,14 +1799,14 @@ def _model_facts(entry: Mapping[str, Any], meta: Mapping[str, Any], *, kind: str 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]: + table: str | None = None, dropped: list[tuple[str, str]] | None = None) -> dict[str, Any]: """模型给的一张表 → 目录项（llm / proposed）。不合规的值整项丢掉，不硬塞：列名对不上表结构的、 表类型和度量类型不在取值里的、业务主键里有不存在的列的、看起来是计算公式的（_formula_like，公式进口径卡）。 该是文字的地方写成了数组、对象（列名、表类型、业务日期的列、度量类型……）同样丢掉，不抛异常：实测见过把列名 写成数组的，以前在这里抛 TypeError，整张表记出错。导入表格的源，说明已经覆盖的字段也丢掉。 有效行过滤、去重规则见 _model_facts：source_kind 是数据源类型（按它的方言解析条件，不认识的类型不收有效行过滤）， - table 是表名（条件里的列可以带这张表的表名）。""" + table 是表名（条件里的列可以带这张表的表名），dropped 收没通过校验的有效行过滤 (原因, 模型写的条件)。""" 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()) @@ -1825,7 +1828,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)) + notes.update(_model_facts(entry, meta, kind=source_kind, table=table, dropped=dropped)) out_cols: dict[str, dict[str, Any]] = {} listed = entry.get("columns") for col in listed if isinstance(listed, list) else []: diff --git a/backend/app/data/catalog_draft.py b/backend/app/data/catalog_draft.py index 84ba96a0..d972c7fb 100644 --- a/backend/app/data/catalog_draft.py +++ b/backend/app/data/catalog_draft.py @@ -83,7 +83,8 @@ 剖析核对的也是这一条）。开了数据剖析的源（catalog_profile.profile_settings）：目录里记下的码值交给模型做背景（遮罩的列 不给，「已确认的目录项」里遮罩列的码值同样不给），表级那段写完后经剖析的只读守卫数一数有效行过滤排除了多少行 （catalog_profile.check_valid_filter），排除 0% 或 100% 的不收，收下的 note 写上排除比例；没开剖析、连不上的照收， -note 写没有核对的原因。 +note 写没有核对的原因。没收下的（没通过校验、排除 0% 或 100%）不算失败，在这张表的日志里记一句（ModelOutcome.log， +起草任务逐表的 log）。 交给模型的是表结构（列、类型、主键、外键、唯一约束、注释）和使用者补充的背景；开了数据剖析的源另给目录里记下的码值 （剖析取过的取值和含义，遮罩的列不给），别的数据值一概不发。写给模型的文字都在 catalog.py 名字带 _SYSTEM / _PROMPT @@ -149,6 +150,15 @@ REASON_ABANDONED = "这张表接连 {k} 段都没写出来，剩下的 {n} 段 #: 跳过不问的段在「没写出来」记录里的原因（不拼进原因，见 _Run._outcome） _SKIPPED = "\0skipped" +# ---- 这张表起草时记下的几句话（不算失败），经 ModelOutcome.log 交给起草任务逐表的 log（界面看得到） +#: 模型提议的有效记录条件没有收下：why 是原因（没通过校验的说法，或者剖析核对的结果） +LOG_FILTER_DROPPED = "{why}，模型提议的这一条没有收下：{condition}" +#: 剖析核对排除了 0% 或 100% 的行 +LOG_FILTER_COUNTED = "{label}按剖析核对{what}（共 {rows} 行）" +#: 一张表最多记几句；日志里的条件最多写几个字 +LOG_MAX = 10 +_LOG_CONDITION_MAX = 200 + # ---- 提醒（最近几次调用多数没写出来），经进度回调交给任务（CatalogDraftJob.notice） NOTICE = ("最近 {calls} 次调用有 {failed} 次没写出来（{kinds}）。建议在{place}换一个模型、缩小起草范围，" "或者停止这次起草") @@ -178,7 +188,8 @@ class ModelOutcome: （error 写原因）。notes 是写出来的那些（已逐项校验，llm / proposed）；items 是项数。parts_total 分几段起草 （整表段是 1，宽表是 1 + 列分段数），parts_done 已经结束了几段，parts_written 其中写出了内容的段数。blank：模型拿 不准、留空了的列（按表结构的顺序；表的状态不因此变）。chunks_total、chunks_done：这张表实际分成的块数和已结束的 - 块数（段被缩小的上限分开、拆成两半后会变多）。""" + 块数（段被缩小的上限分开、拆成两半后会变多）。log：起草时记下的几句话，不算失败（模型提议的有效记录条件没有收下的 + 原因），最多 LOG_MAX 句。""" status: str notes: dict[str, Any] @@ -190,6 +201,7 @@ class ModelOutcome: parts_written: int = 0 chunks_total: int = 0 chunks_done: int = 0 + log: list[str] = field(default_factory=list) @dataclass(frozen=True) @@ -762,6 +774,8 @@ class _Table: chunks_done: int = 0 #: 有效行过滤核对过了（表级那段写完核对一次） vetted: bool = False + #: 起草时记下的几句话（不算失败）：模型提议的有效记录条件没有收下的原因 + log: list[str] = field(default_factory=list) def whole(self, piece: _Piece) -> bool: return piece.table_level and len(piece.columns) == len(self.columns) @@ -1442,7 +1456,13 @@ class _Run: gain = _Gain() for entry in by_table.get(piece.table, []): gain.named = st.named = True - drafted = C._from_model(entry, st.meta, system=self.system, source_kind=self.kind, table=st.name) + dropped: list[tuple[str, str]] = [] + drafted = C._from_model(entry, st.meta, system=self.system, source_kind=self.kind, table=st.name, + dropped=dropped) + if piece.table_level and "valid_filter" not in st.notes: + # 只在请它写表级的项时记：只写几列的那几段里多给的表级的项本来就不要 + for why, condition in dropped: + self._log(st, why, condition) 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} @@ -1508,7 +1528,7 @@ class _Run: total = len(st.parts) or 1 part_of = st.part_of() progress = {"parts_written": len(st.written_parts), "chunks_total": st.chunks_total, - "chunks_done": st.chunks_done} + "chunks_done": st.chunks_done, "log": list(dict.fromkeys(st.log))[:LOG_MAX]} abandoned = self._abandoned_text(st) if not items: failures = [(why, part_of.get(key, 1)) for key, why in st.failed.items() if why.base != _SKIPPED] @@ -1614,8 +1634,8 @@ class _Run: - 目录里人工确认、驳回过的有效行过滤：并入时不会采用这次的提议，不必再查库； - 没开数据剖析：照收，note 写「没有核对排除比例（数据源没有开数据剖析）」； - - 开了：经剖析的只读守卫数排除比例（catalog_profile.check_valid_filter）。排除 0% 或 100% 的不收（日志记一条， - 不算失败）；收下的 note 在依据后面写上剖析核对的结果，或者没有核对的原因。""" + - 开了：经剖析的只读守卫数排除比例（catalog_profile.check_valid_filter）。排除 0% 或 100% 的不收（这张表的日志 + 记一句，不算失败）；收下的 note 在依据后面写上剖析核对的结果，或者没有核对的原因。""" if st.vetted: return st.vetted = True @@ -1645,9 +1665,22 @@ class _Run: if not found.keep: logger.info("起草数据目录：表 %s 的有效行过滤 %s 没有收下：%s", st.name, item.get("value"), found.dropped) st.notes.pop("valid_filter", None) + why = found.dropped or found.note + if found.rows is not None: + why = LOG_FILTER_COUNTED.format(label=C.TABLE_FIELD_LABEL["valid_filter"], what=found.dropped, + rows=f"{found.rows:,}") + self._log(st, why, str(item.get("value") or "")) return item["note"] = _sentences(item.get("note"), found.note) + @staticmethod + def _log(st: _Table, why: str, condition: str) -> None: + """这张表的日志记一句：模型提议的有效记录条件没有收下（不算失败）。""" + condition = " ".join(condition.split()) + if len(condition) > _LOG_CONDITION_MAX: + condition = condition[:_LOG_CONDITION_MAX] + "…" + st.log.append(LOG_FILTER_DROPPED.format(why=why, condition=condition)) + @staticmethod async def _hook(fn: Callable[..., Awaitable[None]] | None, *args: Any) -> None: if fn is None: @@ -1790,7 +1823,8 @@ def _sentences(*parts: Any) -> str: return "。".join(texts) -__all__ = ["BUDGET_WORDS", "Background", "CallReport", "FAILURE_LABEL", "Hooks", "ModelOutcome", "NOTICE", "PartWrite", +__all__ = ["BUDGET_WORDS", "Background", "CallReport", "FAILURE_LABEL", "Hooks", "LOG_FILTER_COUNTED", + "LOG_FILTER_DROPPED", "LOG_MAX", "ModelOutcome", "NOTICE", "PartWrite", "REASON_ABANDONED", "REASON_CALL", "REASON_CAPPED", "REASON_EXHAUSTED", "REASON_FORMAT", "REASON_HALTED", "REASON_HOLLOW", "REASON_MISSING", "REASON_OMITTED", "REASON_RETRIED", "REASON_SPLIT", "REASON_SPLIT_PARTS", "REASON_TIMEOUT", "REASON_TRUNCATED", "REASON_UNANSWERED", "SEG", "classify_failure", "model_gate", diff --git a/backend/app/data/catalog_jobs.py b/backend/app/data/catalog_jobs.py index a721cf58..48c37056 100644 --- a/backend/app/data/catalog_jobs.py +++ b/backend/app/data/catalog_jobs.py @@ -184,6 +184,8 @@ def task_dict(task: CatalogDraftTask) -> dict[str, Any]: # 模型拿不准、留空了的列（中文名和含义都没给）：几列、最多 5 个列名（按表结构的顺序）。表的状态不因此变 "blank_columns": task.blank_columns or 0, "blank_sample": list(task.blank_sample or []), + # 起草时记下的几句话，不算失败（模型提议的有效记录条件没有收下的原因：没通过校验、剖析核对排除 0% 或 100%） + "log": list(task.log or []), "error": task.error, "model_error": task.model_error, "started_at": _iso(task.started_at), @@ -694,7 +696,7 @@ class _JobRun: values["version"] = result.version if outcome is not None: values.update(parts_done=outcome.parts_done, chunks_total=outcome.chunks_total, - chunks_done=outcome.chunks_done) + chunks_done=outcome.chunks_done, log=list(outcome.log) or None) if not result.error: # 收尾写进去了：写出内容的段都在目录里了。没写进去（一直撞上别人刚改过）的照库里逐段写进去的段数 values["parts_written"] = outcome.parts_written diff --git a/backend/app/db/base.py b/backend/app/db/base.py index 6cd9298b..3bc178cc 100644 --- a/backend/app/db/base.py +++ b/backend/app/db/base.py @@ -216,6 +216,8 @@ _COLUMN_MIGRATIONS: list[tuple[str, str, str]] = [ ("catalog_draft_tasks", "parts_written", "INTEGER DEFAULT 0"), ("catalog_draft_tasks", "chunks_total", "INTEGER"), ("catalog_draft_tasks", "chunks_done", "INTEGER"), + # 复核后：逐表的日志（模型提议的有效记录条件没有收下的原因）。存量任务回填 NULL，接口给空列表 + ("catalog_draft_tasks", "log", "JSON"), ] #: 迁移补上的列要建的索引：(索引名, 表, 列)。create_all 不碰已有的表，表上新列的索引也就不会建。 diff --git a/backend/app/db/models.py b/backend/app/db/models.py index 0671e33b..f341e89a 100644 --- a/backend/app/db/models.py +++ b/backend/app/db/models.py @@ -426,6 +426,8 @@ class CatalogDraftTask(Base): # 模型拿不准、留空了的列（中文名和含义都没给）：几列，最多记 5 个列名。表的状态不因此变 blank_columns: Mapped[int] = mapped_column(Integer, default=0) blank_sample: Mapped[list[Any] | None] = mapped_column(default=None) + # 起草时记下的几句话，不算失败（模型提议的有效记录条件没有收下的原因），最多 10 句 + log: Mapped[list[Any] | None] = mapped_column(default=None) # error：整张表没起草（表结构里没有、写入一直冲突、服务重启中断）；model_error：模型那部分没写出来的原因和办法 error: Mapped[str | None] = mapped_column(Text, default=None) model_error: Mapped[str | None] = mapped_column(Text, default=None) diff --git a/backend/tests/test_catalog_draft_facts.py b/backend/tests/test_catalog_draft_facts.py index 139ba8ca..37f3705b 100644 --- a/backend/tests/test_catalog_draft_facts.py +++ b/backend/tests/test_catalog_draft_facts.py @@ -6,6 +6,7 @@ 存的是整理过的写法：原文照旧，列换成表结构里的真名、不带表名、按方言加引号（剖析核对的也是这一条）； - 开了数据剖析、能连上的源：经剖析的只读守卫数出排除比例，排除 0% 或 100% 的不收，收下的在 note 里写剖析核对的结果； 没开剖析、连不上的照收，note 写没有核对的原因； +- 没收下的（没通过校验、排除 0% 或 100%）不算失败，在这张表的日志里记一句（起草任务逐表的 log）； - 开了剖析的源，目录里记下的码值交给模型做背景（遮罩的列不给；「已确认的目录项」里遮罩列的码值也不给）； - 这两项照旧记模型起草 / 推断；SQL 检查只在人工确认、有确证时才把漏了有效行过滤报成警告，推断的只是提示。 @@ -183,12 +184,14 @@ def test_the_stored_filter_names_its_columns_the_way_the_source_needs(kind, cond def test_a_filter_that_grows_past_the_limit_once_quoted_is_dropped_not_written(): - """整理时加了引号、超过 500 字的不收：目录里这一项最多 500 字，写进去会让整张表写入失败。""" + """整理时加了引号、超过 500 字的不收（记下原因）：目录里这一项最多 500 字，写进去会让整张表写入失败。""" condition = "orderstatus IN (" + ", ".join(f"'{i:03d}'" for i in range(66)) + ") OR updated_at IS NULL" assert len(condition) == 499 + dropped: list[tuple[str, str]] = [] notes = catalog._from_model({"table": "orders", "label": "订单", "valid_filter": condition}, MIXED, system=False, - source_kind="postgresql", table="orders") + source_kind="postgresql", table="orders", dropped=dropped) assert "valid_filter" not in notes and notes["label"]["value"] == "订单" + assert dropped == [("有效记录条件按数据源的写法整理后超过 500 字", condition)] assert not catalog.validate_notes(notes) @@ -436,6 +439,28 @@ async def test_the_engine_checks_the_filter_on_a_profiled_source(scenic, scenic_ assert got["visits"].notes["label"]["value"] == "入园记录" +async def test_a_dropped_filter_is_written_into_the_table_log(scenic, scenic_db): + """没收下的有效记录条件不算失败，在这张表的日志里记一条（起草任务逐表的 log，界面看得到）：没通过校验的写原因， + 剖析核对排除 0% 或 100% 的写数到了多少行。收下的、模型没提的不记。""" + bad = {**REPLY, "valid_filter": {"condition": "status IN (SELECT code FROM bad_codes)", "reason": "猜的"}} + got = await D.run_model_draft(FakeModel({"orders": bad}), _plain({"orders": _orders()}), ["orders"]) + assert got["orders"].status == "ok" and got["orders"].error is None + assert got["orders"].log == [ + "有效记录条件里不能有子查询，模型提议的这一条没有收下：status IN (SELECT code FROM bad_codes)"] + + source = await scenic() + total, _ = _count(scenic_db, "1 = 1") + for condition, what in (("status IN (0, 1)", "没有排除任何一行"), ("status = 7", "所有行都被排除了")): + useless = {"label": "入园记录", "valid_filter": {"condition": condition, "reason": "猜的"}} + got = await D.run_model_draft(FakeModel({"visits": useless}), source, ["visits"]) + assert got["visits"].log == [ + f"有效记录条件按剖析核对{what}（共 {total:,} 行），模型提议的这一条没有收下：{condition}"] + + for reply in (REPLY, {"label": "订单"}): + got = await D.run_model_draft(FakeModel({"orders": reply}), _plain({"orders": _orders()}), ["orders"]) + assert got["orders"].log == [] + + async def test_a_confirmed_filter_is_not_checked_again(scenic, monkeypatch): source = await scenic() called: list[str] = [] diff --git a/backend/tests/test_catalog_draft_jobs.py b/backend/tests/test_catalog_draft_jobs.py index 29bbee8d..81840b34 100644 --- a/backend/tests/test_catalog_draft_jobs.py +++ b/backend/tests/test_catalog_draft_jobs.py @@ -444,6 +444,8 @@ def test_old_job_tables_get_the_new_columns(): # 闭环第一步：逐段落盘的段数、块进度。存量行是 0 和 NULL：升级前没记这些 assert conn.execute(text("SELECT parts_written, chunks_total, chunks_done FROM catalog_draft_tasks")) \ .one() == (0, None, None) + # 复核后：逐表的日志（没收下的有效记录条件）。存量行是 NULL，接口给空列表 + assert conn.execute(text("SELECT log FROM catalog_draft_tasks")).one() == (None,) async def test_a_job_that_cannot_start_is_failed_with_the_reason(client, monkeypatch): @@ -1035,3 +1037,18 @@ async def test_a_wide_table_whose_closing_write_fails_keeps_the_count_of_parts_o assert task["status"] == "failed" and task["error"] == catalog.WRITE_CONFLICT_ERROR assert task["parts_written"] == 3, "最后一段没写进去" assert len((await client.get(f"{_base(sid)}/wide")).json()["notes"]["columns"]) == 64 + + +async def test_a_dropped_filter_is_logged_on_the_task_not_failed(client, monkeypatch): + """模型提议的有效记录条件没通过校验：丢掉、不算失败（表照样完成），逐表的 log 里记一条，接口看得到；没有就是 + 空列表。""" + sid = await _source() + replies = {**REPLIES, "gates": {"label": "闸口", + "valid_filter": {"condition": "gate_code IN (SELECT 1)", "reason": "猜的"}}} + job = await _draft(client, sid, monkeypatch, FakeModel(replies), ["gates", "channels"]) + by = {t["table_name"]: t for t in job["tasks"]} + assert by["gates"]["status"] == "ok" and by["gates"]["model_error"] is None + assert by["gates"]["log"] == ["有效记录条件里不能有子查询，模型提议的这一条没有收下：gate_code IN (SELECT 1)"] + assert by["channels"]["log"] == [] + notes = (await client.get(f"{_base(sid)}/gates")).json()["notes"] + assert "valid_filter" not in notes and notes["label"]["value"] == "闸口"