From 2d108d116fdd975ba7d81bfb4d49ef7f838a6e50 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Mon, 19 Oct 2026 10:10:38 +0800 Subject: [PATCH] =?UTF-8?q?=E8=A1=A8=E7=BB=93=E6=9E=84=E5=8F=98=E4=BA=86?= =?UTF-8?q?=E4=B9=8B=E5=90=8E=E5=8F=A3=E5=BE=84=E5=87=91=E4=B8=8D=E9=BD=90?= =?UTF-8?q?=E5=8F=82=E6=95=B0=E6=97=B6=E7=85=A7=E5=AE=9E=E8=AE=B0=E4=B8=8B?= =?UTF-8?q?=EF=BC=9A=E4=BB=A5=E5=89=8D=E4=B8=8D=E8=AF=95=E7=AE=97=E3=80=81?= =?UTF-8?q?=E4=B9=9F=E4=B8=8D=E8=AE=B0=E6=97=A5=E5=BF=97=EF=BC=8C=E4=B8=80?= =?UTF-8?q?=E5=A3=B0=E4=B8=8D=E5=93=8D=E5=9C=B0=E8=B7=B3=E8=BF=87=E3=80=82?= =?UTF-8?q?=E7=8E=B0=E5=9C=A8=E8=AE=B0=E4=B8=80=E6=9D=A1=E6=97=A5=E5=BF=97?= =?UTF-8?q?=E5=86=99=E7=BC=BA=E5=93=AA=E5=87=A0=E4=B8=AA=E5=8F=82=E6=95=B0?= =?UTF-8?q?=EF=BC=88=E5=BF=85=E5=A1=AB=E3=80=81=E6=B2=A1=E7=BB=99=E5=80=BC?= =?UTF-8?q?=E4=B9=9F=E6=B2=A1=E6=9C=89=E7=BC=BA=E7=9C=81=E5=80=BC=E7=9A=84?= =?UTF-8?q?=E5=85=A5=E5=8F=A3=E5=AD=97=E6=AE=B5=EF=BC=89=E5=92=8C=E5=8F=82?= =?UTF-8?q?=E6=95=B0=E6=9C=AC=E6=9D=A5=E4=BB=8E=E5=93=AA=E6=9D=A5=EF=BC=8C?= =?UTF-8?q?=E5=90=8C=E6=97=B6=E5=86=99=E8=BF=9B=E5=8F=98=E5=8C=96=E8=AE=B0?= =?UTF-8?q?=E5=BD=95=E7=9A=84=20affects.probes=EF=BC=88=E5=90=8C=E4=B8=80?= =?UTF-8?q?=E7=89=88=E5=8F=AA=E8=AE=B0=E4=B8=80=E6=9D=A1=EF=BC=89=EF=BC=8C?= =?UTF-8?q?=E5=8F=98=E5=8C=96=E9=A1=B5=E7=9A=84=E3=80=8C=E5=BD=B1=E5=93=8D?= =?UTF-8?q?=E3=80=8D=E4=B8=80=E5=9D=97=E5=86=99=E3=80=8C=E3=80=8Cx?= =?UTF-8?q?=E3=80=8Dv3=EF=BC=9A=E7=BC=BA=E5=B0=91=E5=8F=82=E6=95=B0=20star?= =?UTF-8?q?t=E3=80=81end=EF=BC=8C=E6=B2=A1=E6=9C=89=E8=AF=95=E7=AE=97?= =?UTF-8?q?=E3=80=8D=EF=BC=9B=E5=90=8C=E4=B8=80=E6=9D=A1=E5=8F=98=E5=8C=96?= =?UTF-8?q?=E5=86=8D=E8=A2=AB=E6=8E=A2=E6=9F=A5=E5=88=B0=E6=97=B6=E8=BF=99?= =?UTF-8?q?=E4=BB=BD=E8=AE=B0=E5=BD=95=E4=B8=8D=E4=B8=A2=E3=80=82=E5=8F=98?= =?UTF-8?q?=E5=8C=96=E6=8E=A5=E5=8F=A3=E7=BB=99=E7=9A=84=20affects=20?= =?UTF-8?q?=E7=BC=BA=E7=9A=84=E5=87=A0=E9=A1=B9=E8=A1=A5=E4=B8=8A=E7=A9=BA?= =?UTF-8?q?=E7=9A=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/data/source_changes.py | 8 +++- backend/app/ledger/recall.py | 49 ++++++++++++++++++++- backend/tests/test_ledger_recall.py | 41 +++++++++++++++++ frontend/src/lib/termsSource.ts | 5 +++ frontend/src/pages/catalog/ChangeDetail.tsx | 18 +++++++- frontend/src/typesSource.ts | 16 +++++++ scripts/check-catalog-changes.mjs | 5 +++ 7 files changed, 137 insertions(+), 5 deletions(-) diff --git a/backend/app/data/source_changes.py b/backend/app/data/source_changes.py index 5b625baf..cb1cb35e 100644 --- a/backend/app/data/source_changes.py +++ b/backend/app/data/source_changes.py @@ -156,7 +156,10 @@ async def record(session: AsyncSession, source: Any, change: NewChange, detected found.last_seen_at = now found.seen_count = int(found.seen_count or 0) + 1 if found.status == "open": - found.affects, found.before, found.after = change.affects, change.before, change.after + # 影响按这一次刷新；表结构变了之后口径试算的情况（ledger/recall 记的 probes）不是探查算的，留着 + probes = (found.affects or {}).get("probes") if isinstance(found.affects, dict) else None + found.affects = {**(change.affects or {}), "probes": probes} if probes else change.affects + found.before, found.after = change.before, change.after if SEVERITY_RANK[change.severity] > SEVERITY_RANK.get(found.severity, 0): found.severity = change.severity return found, False @@ -227,7 +230,8 @@ def change_out(row: SourceChange) -> dict[str, Any]: "severity": row.severity, "summary": summary(row), "detected_at": _iso(row.detected_at), "last_seen_at": _iso(row.last_seen_at), "seen_count": row.seen_count or 1, "detected_by": dict(row.detected_by or {}), "basis": row.basis, "before": row.before, "after": row.after, - "affects": row.affects or {"catalog": [], "templates": [], "rules_touched": False}, + # 缺的几项补上空的：只记了口径试算情况（ledger/recall 的 probes）的那种没有 catalog、templates + "affects": {"catalog": [], "templates": [], "rules_touched": False, **(row.affects or {})}, "struct_version_before": row.struct_version_before, "struct_version_after": row.struct_version_after, "status": row.status, "resolution": row.resolution, } diff --git a/backend/app/ledger/recall.py b/backend/app/ledger/recall.py index bcb9d7b8..b03425d8 100644 --- a/backend/app/ledger/recall.py +++ b/backend/app/ledger/recall.py @@ -1527,11 +1527,16 @@ async def _probe_calibers(change: Mapping[str, Any]) -> list[str]: if answer is not None: latest, inputs_from = dict(answer.binding), "answer" for card in caliber.cards[:1]: + pinned = None try: pinned = await derive.load_pinned(session, caliber.workflow_id, caliber.version, card) names = {str(f.get("name")) for f in pinned.entry_fields} - inputs = compose.check_inputs(pinned, {k: v for k, v in (latest or {}).items() if k in names}) - except (derive.NotDerivable, compose.ComposeRefused): + given = {k: v for k, v in (latest or {}).items() if k in names} + inputs = compose.check_inputs(pinned, given) + except (derive.NotDerivable, compose.ComposeRefused) as e: + # 凑不齐参数（这一版没有正式运行、也没有等价的问数回答，必填的入口字段又没有缺省值）或者派生不了：不试算， + # 照实记一条缺哪几个参数——日志里一条，变化记录的影响里一条（变化页照它写），不再一声不响地跳过 + await _probe_skipped(change, caliber, pinned, latest, inputs_from, e.message) continue probe = derive.caliber_probe(pinned) origin = {"probe": 1, "workflow_id": caliber.workflow_id, "version": caliber.version, @@ -1551,6 +1556,46 @@ async def _probe_calibers(change: Mapping[str, Any]) -> list[str]: return launched +#: 变化记录 affects 里试算情况的键（data/source_changes 再次探查到同一条变化时保留它） +PROBES_KEY = "probes" + + +def missing_inputs(pinned: Any, given: Mapping[str, Any] | None) -> list[str]: + """口径入口里必填、又没给值也没有缺省值的字段（按入口字段的先后）。""" + out = [] + for f in pinned.entry_fields if pinned is not None else (): + name = str(f.get("name") or "").strip() + if not name or not f.get("required") or f.get("default") not in (None, ""): + continue + if (given or {}).get(name) in (None, ""): + out.append(name) + return out + + +async def _probe_skipped(change: Mapping[str, Any], caliber: Any, pinned: Any, given: Mapping[str, Any] | None, + inputs_from: str, reason: str) -> None: + """表结构变了、这一版口径没能试算：记一条日志，写进变化记录的 affects.probes（同一版只记一条，换成最新的说法）。""" + from app.db.models_sources import SourceChange + + missing = missing_inputs(pinned, given) + name = getattr(pinned, "workflow_name", None) or caliber.workflow_id + said = f"缺少参数 {'、'.join(missing)}" if missing else reason + logger.info("表结构变了之后口径「%s」v%s 没有试算：%s（参数来自%s）", name, caliber.version, said, + {"formal": "最近一次正式运行", "answer": "等价的问数回答", "defaults": "入口的缺省值"}.get(inputs_from, inputs_from)) + note = {"workflow_id": caliber.workflow_id, "version": caliber.version, "name": str(name), "outcome": "skipped", + "missing": missing, "reason": reason, "inputs_from": inputs_from} + async with SessionLocal() as session: + row = (await session.execute(select(SourceChange).where(SourceChange.id == change.get("id")))).scalar_one_or_none() + if row is None: + return + affects = dict(row.affects or {}) + kept = [p for p in affects.get(PROBES_KEY) or [] if isinstance(p, dict) + and (p.get("workflow_id"), p.get("version")) != (caliber.workflow_id, caliber.version)] + affects[PROBES_KEY] = [*kept, note] + row.affects = affects + await session.commit() + + async def _settle_probe(run: Run) -> str: """试算运行结束：算出来了只记结论；没算出来给那一版口径的发布人开 recall 待办（reason=structure）。""" from app.apps import versions as app_versions diff --git a/backend/tests/test_ledger_recall.py b/backend/tests/test_ledger_recall.py index 7e7232c7..751a7ed7 100644 --- a/backend/tests/test_ledger_recall.py +++ b/backend/tests/test_ledger_recall.py @@ -913,3 +913,44 @@ async def test_restart_resumes_the_auto_batch_and_settles_finished_runs(monkeypa await recall.resume() async with SessionLocal() as session: assert len(await recall.recall_runs(session, wf, 2)) == 2, "再重启一次：都算过了，不再发起" + + +async def test_a_probe_that_cannot_assemble_its_inputs_says_which_are_missing(engine_up, caplog): + """表结构变了、口径凑不齐参数（这一版没有正式运行、也没有等价的问数回答，必填的入口字段又没有缺省值）：不试算，照实记一条 + 「缺哪几个参数」——日志里有，变化记录的影响里也有（变化页照它写），不再一声不响地跳过。之后这条变化再被探查到一次，记下的 + 不丢。""" + import logging + + from app.data import source_changes as sc + from app.db.models import DataSource + from app.db.models_sources import SourceChange + + wf = await S.publish() + async with SessionLocal() as session: + source = (await session.execute(select(DataSource).where(DataSource.name == S.CURRENT["source"]))).scalar_one() + change, _ = await sc.record(session, source, sc.NewChange(kind="column_removed", table="orders", column="amount", + severity="action", key="v3"), + sc.Detected(job_kind="introspect")) + await session.commit() + out = sc.change_out(change) + with caplog.at_level(logging.INFO, logger="app.ledger.recall"): + await recall.process_change(out) + async with SessionLocal() as session: + probes = [r for r in (await session.execute(select(Run).where(Run.origin_kind == "recall"))).scalars() + if (r.origin or {}).get("change") == out["id"]] + row = (await session.execute(select(SourceChange).where(SourceChange.id == out["id"]))).scalar_one() + assert probes == [], "凑不齐参数：不试算" + [note] = row.affects["probes"] + assert (note["workflow_id"], note["version"], note["outcome"]) == (wf, 1, "skipped") + assert note["missing"] == ["start", "end"] and "start" in note["reason"], note + assert any("start、end" in r.getMessage() for r in caplog.records), [r.getMessage() for r in caplog.records] + # 变化接口（变化页）照样带着 + assert sc.change_out(row)["affects"]["probes"] == [note] + # 同一条变化再被探查到一次：影响按新的一次刷新，试算的记录不丢 + async with SessionLocal() as session: + source = await session.get(DataSource, row.source_id) + await sc.record(session, source, sc.NewChange(kind="column_removed", table="orders", column="amount", + severity="action", key="v3"), sc.Detected(job_kind="introspect")) + await session.commit() + again = (await session.execute(select(SourceChange).where(SourceChange.id == out["id"]))).scalar_one() + assert again.affects["probes"] == [note] diff --git a/frontend/src/lib/termsSource.ts b/frontend/src/lib/termsSource.ts index cd04b605..c10fc9be 100644 --- a/frontend/src/lib/termsSource.ts +++ b/frontend/src/lib/termsSource.ts @@ -110,6 +110,11 @@ export const SOURCE_CHANGE_TEXT = { impactNeeds: '人工确认过、还说着被删的列', impactRules: '有效记录条件用到了这一列：取值有变时，按它过滤的结果可能漏掉新状态的行', impactTemplates: (n: number) => `用到这张表的已发布模板 ${formatNumber(n)} 个`, + /** 表结构变了之后没能试算的口径：缺参数的写缺哪几个，别的写原因原话 */ + probesTitle: '没有试算的口径', + probeSkipped: (name: string, version: number, missing: string[], reason: string) => + `「${name}」v${version}：${missing.length ? `缺少参数 ${missing.join('、')}，没有试算` : `没有试算（${reason}）`}`, + probeFrom: { formal: '参数按最近一次正式运行取', answer: '参数按等价的问数回答取', defaults: '这一版还没有正式运行，只能用入口的缺省值' } as Record, templateImpact: { direct: '写死的 SQL 直接引用', possible: '可能涉及' } as Record, templateVersion: (v: number) => `v${v}`, breaks: (cols: string[]) => (cols.length diff --git a/frontend/src/pages/catalog/ChangeDetail.tsx b/frontend/src/pages/catalog/ChangeDetail.tsx index 7657db92..979ac9a2 100644 --- a/frontend/src/pages/catalog/ChangeDetail.tsx +++ b/frontend/src/pages/catalog/ChangeDetail.tsx @@ -294,7 +294,8 @@ function Facts({ change: c }: { change: SourceChange }) { /** 影响面：说到这些列的目录项（测到时和现在的状态）、用到这张表的已发布模板（写死的 SQL 会不会坏）、有效记录条件 */ function Impact({ change: c }: { change: SourceChange }) { const { catalog, templates, rules_touched: rules } = c.affects - const empty = !catalog.length && !templates.length && !rules + const probes = c.affects.probes ?? [] + const empty = !catalog.length && !templates.length && !rules && !probes.length return ( {empty &&

{T.impactNone}

} @@ -343,6 +344,21 @@ function Impact({ change: c }: { change: SourceChange }) { )} + {probes.length > 0 && ( +
+

{T.probesTitle}

+
    + {probes.map((p) => ( +
  • + {T.probeSkipped(p.name, p.version, p.missing ?? [], p.reason)} + {p.inputs_from && T.probeFrom[p.inputs_from] && ( + {T.probeFrom[p.inputs_from]} + )} +
    • + ))} +
+
+ )} ) } diff --git a/frontend/src/typesSource.ts b/frontend/src/typesSource.ts index 0317fc2e..8eec3417 100644 --- a/frontend/src/typesSource.ts +++ b/frontend/src/typesSource.ts @@ -54,6 +54,22 @@ export interface SourceChangeAffects { templates: SourceChangeTemplate[] /** 有效记录条件用到了这一列 */ rules_touched: boolean + /** + * 表结构变了之后口径没能试算的那几版（服务端 ledger/recall._probe_skipped）：缺哪几个参数、原因原话、参数本来从哪来。 + * 只记没试算的；试算了的结论在召回待办里 + */ + probes?: SourceChangeProbe[] +} + +export interface SourceChangeProbe { + workflow_id: string + version: number + name: string + outcome: 'skipped' + missing: string[] + reason: string + /** formal 最近一次正式运行 / answer 等价的问数回答 / defaults 入口的缺省值 */ + inputs_from?: string } /** 现在能做的一个处理动作；ok 为假时 reason 写为什么做不了 */ diff --git a/scripts/check-catalog-changes.mjs b/scripts/check-catalog-changes.mjs index 751e7bd7..ff46c5e4 100644 --- a/scripts/check-catalog-changes.mjs +++ b/scripts/check-catalog-changes.mjs @@ -178,6 +178,9 @@ function freshChanges() { templates: [{ workflow_id: 'wf-check-weekly', name: '门店周报', version: 3, level: 'published', impact: 'direct', nodes: [{ node_id: 'q_orders', label: '取订单', type: 'tool', impact: 'direct', breaks: true, unknown_columns: ['status'] }] }], rules_touched: true, + // 表结构变了之后没能试算的口径（服务端 ledger/recall._probe_skipped）：缺哪几个参数 + probes: [{ workflow_id: 'wf-check-weekly', version: 3, name: '门店周报', outcome: 'skipped', missing: ['start', 'end'], + reason: '「start」必填；「end」必填', inputs_from: 'defaults' }], }, actions: [act('clean_catalog'), act('redraft'), act('reprofile', NOT_PROFILED), ...COMMON] }), change({ seq: 14, id: 'chg-codes', table: 'orders', column: 'channel', kind: 'codes_new', category: 'value', severity: 'action', @@ -458,6 +461,8 @@ await section('变化：依据、前后、影响面、动作照服务端给的 check('……写死这张表的已发布模板：「这条 SQL 现在会报列不存在（status）」', (await textOf(d.locator('[data-change-node="q_orders"]'))).includes('这条 SQL 现在会报列不存在（status）') && (await d.locator('[data-change-node="q_orders"]').getAttribute('data-change-breaks')) === '1') check('……模板链到编排页那个节点', (await d.locator('[data-change-template] a').first().getAttribute('href')) === '/studio/wf-check-weekly?focus=q_orders') + check('……没能试算的口径照实写缺哪几个参数', (await textOf(d.locator('[data-change-probe="wf-check-weekly"]'))) + .includes('「门店周报」v3：缺少参数 start、end，没有试算')) check('……有效记录条件用到了这一列', (await textOf(d.locator('[data-change-rules]'))).includes('有效记录条件用到了这一列')) check('动作照服务端给的：清理目录项、重新起草、重新剖析、知道了、误报', JSON.stringify(await attrs(d.locator('[data-change-action]'), 'data-change-action')) === '["clean_catalog","redraft","reprofile","acknowledge","dismiss"]')