数据目录闭环第一步(后端 A)复核修正:没收下的有效记录条件不算失败,在起草任务逐表的 log 里记一句,接口看得到(以前只进服务端日志)——没通过校验的写原因(「有效记录条件里不能有子查询,模型提议的这一条没有收下:……」),剖析核对排除 0% 或 100% 的写数到了多少行;只在请模型写表级的项的那段里记,一张表最多 10 句。CatalogDraftTask 加 log 列(存量任务回填 NULL,接口给空列表)

This commit is contained in:
ThearchyHelios
2026-10-14 10:00:10 +08:00
parent 7383877ee7
commit d3bd80211a
7 changed files with 103 additions and 18 deletions
+11 -8
View File
@@ -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 []:
+41 -7
View File
@@ -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",
+3 -1
View File
@@ -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
+2
View File
@@ -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 不碰已有的表,表上新列的索引也就不会建。
+2
View File
@@ -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)
+27 -2
View File
@@ -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] = []
+17
View File
@@ -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"] == "闸口"