表结构变了之后口径凑不齐参数时照实记下:以前不试算、也不记日志,一声不响地跳过。现在记一条日志写缺哪几个参数(必填、没给值也没有缺省值的入口字段)和参数本来从哪来,同时写进变化记录的 affects.probes(同一版只记一条),变化页的「影响」一块写「「x」v3:缺少参数 start、end,没有试算」;同一条变化再被探查到时这份记录不丢。变化接口给的 affects 缺的几项补上空的

This commit is contained in:
ThearchyHelios
2026-10-19 10:10:38 +08:00
parent a534af6ea1
commit 2d108d116f
7 changed files with 137 additions and 5 deletions
+6 -2
View File
@@ -156,7 +156,10 @@ async def record(session: AsyncSession, source: Any, change: NewChange, detected
found.last_seen_at = now found.last_seen_at = now
found.seen_count = int(found.seen_count or 0) + 1 found.seen_count = int(found.seen_count or 0) + 1
if found.status == "open": 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): if SEVERITY_RANK[change.severity] > SEVERITY_RANK.get(found.severity, 0):
found.severity = change.severity found.severity = change.severity
return found, False 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), "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, "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, "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, "struct_version_before": row.struct_version_before, "struct_version_after": row.struct_version_after,
"status": row.status, "resolution": row.resolution, "status": row.status, "resolution": row.resolution,
} }
+47 -2
View File
@@ -1527,11 +1527,16 @@ async def _probe_calibers(change: Mapping[str, Any]) -> list[str]:
if answer is not None: if answer is not None:
latest, inputs_from = dict(answer.binding), "answer" latest, inputs_from = dict(answer.binding), "answer"
for card in caliber.cards[:1]: for card in caliber.cards[:1]:
pinned = None
try: try:
pinned = await derive.load_pinned(session, caliber.workflow_id, caliber.version, card) pinned = await derive.load_pinned(session, caliber.workflow_id, caliber.version, card)
names = {str(f.get("name")) for f in pinned.entry_fields} 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}) given = {k: v for k, v in (latest or {}).items() if k in names}
except (derive.NotDerivable, compose.ComposeRefused): 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 continue
probe = derive.caliber_probe(pinned) probe = derive.caliber_probe(pinned)
origin = {"probe": 1, "workflow_id": caliber.workflow_id, "version": caliber.version, 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 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: async def _settle_probe(run: Run) -> str:
"""试算运行结束:算出来了只记结论;没算出来给那一版口径的发布人开 recall 待办(reason=structure)。""" """试算运行结束:算出来了只记结论;没算出来给那一版口径的发布人开 recall 待办(reason=structure)。"""
from app.apps import versions as app_versions from app.apps import versions as app_versions
+41
View File
@@ -913,3 +913,44 @@ async def test_restart_resumes_the_auto_batch_and_settles_finished_runs(monkeypa
await recall.resume() await recall.resume()
async with SessionLocal() as session: async with SessionLocal() as session:
assert len(await recall.recall_runs(session, wf, 2)) == 2, "再重启一次:都算过了,不再发起" 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]
+5
View File
@@ -110,6 +110,11 @@ export const SOURCE_CHANGE_TEXT = {
impactNeeds: '人工确认过、还说着被删的列', impactNeeds: '人工确认过、还说着被删的列',
impactRules: '有效记录条件用到了这一列:取值有变时,按它过滤的结果可能漏掉新状态的行', impactRules: '有效记录条件用到了这一列:取值有变时,按它过滤的结果可能漏掉新状态的行',
impactTemplates: (n: number) => `用到这张表的已发布模板 ${formatNumber(n)} 个`, 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<string, string>,
templateImpact: { direct: '写死的 SQL 直接引用', possible: '可能涉及' } as Record<string, string>, templateImpact: { direct: '写死的 SQL 直接引用', possible: '可能涉及' } as Record<string, string>,
templateVersion: (v: number) => `v${v}`, templateVersion: (v: number) => `v${v}`,
breaks: (cols: string[]) => (cols.length breaks: (cols: string[]) => (cols.length
+17 -1
View File
@@ -294,7 +294,8 @@ function Facts({ change: c }: { change: SourceChange }) {
/** 影响面:说到这些列的目录项(测到时和现在的状态)、用到这张表的已发布模板(写死的 SQL 会不会坏)、有效记录条件 */ /** 影响面:说到这些列的目录项(测到时和现在的状态)、用到这张表的已发布模板(写死的 SQL 会不会坏)、有效记录条件 */
function Impact({ change: c }: { change: SourceChange }) { function Impact({ change: c }: { change: SourceChange }) {
const { catalog, templates, rules_touched: rules } = c.affects 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 ( return (
<Block title={T.impactTitle} attr="impact"> <Block title={T.impactTitle} attr="impact">
{empty && <p className="text-xs text-faint" data-change-impact-none="">{T.impactNone}</p>} {empty && <p className="text-xs text-faint" data-change-impact-none="">{T.impactNone}</p>}
@@ -343,6 +344,21 @@ function Impact({ change: c }: { change: SourceChange }) {
</ul> </ul>
</div> </div>
)} )}
{probes.length > 0 && (
<div data-change-probes={probes.length}>
<p className="text-2xs font-medium text-dim">{T.probesTitle}</p>
<ul className="mt-1 space-y-0.5">
{probes.map((p) => (
<li key={`${p.workflow_id}:${p.version}`} className="text-xs text-[var(--warn)]" data-change-probe={p.workflow_id}>
{T.probeSkipped(p.name, p.version, p.missing ?? [], p.reason)}
{p.inputs_from && T.probeFrom[p.inputs_from] && (
<span className="ml-1.5 text-2xs text-faint">{T.probeFrom[p.inputs_from]}</span>
)}
</li>
))}
</ul>
</div>
)}
</Block> </Block>
) )
} }
+16
View File
@@ -54,6 +54,22 @@ export interface SourceChangeAffects {
templates: SourceChangeTemplate[] templates: SourceChangeTemplate[]
/** 有效记录条件用到了这一列 */ /** 有效记录条件用到了这一列 */
rules_touched: boolean 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 写为什么做不了 */ /** 现在能做的一个处理动作;ok 为假时 reason 写为什么做不了 */
+5
View File
@@ -178,6 +178,9 @@ function freshChanges() {
templates: [{ workflow_id: 'wf-check-weekly', name: '门店周报', version: 3, level: 'published', impact: 'direct', 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'] }] }], nodes: [{ node_id: 'q_orders', label: '取订单', type: 'tool', impact: 'direct', breaks: true, unknown_columns: ['status'] }] }],
rules_touched: true, 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] }), 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', 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)') 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') && (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 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('……有效记录条件用到了这一列', (await textOf(d.locator('[data-change-rules]'))).includes('有效记录条件用到了这一列'))
check('动作照服务端给的:清理目录项、重新起草、重新剖析、知道了、误报', JSON.stringify(await attrs(d.locator('[data-change-action]'), 'data-change-action')) check('动作照服务端给的:清理目录项、重新起草、重新剖析、知道了、误报', JSON.stringify(await attrs(d.locator('[data-change-action]'), 'data-change-action'))
=== '["clean_catalog","redraft","reprofile","acknowledge","dismiss"]') === '["clean_catalog","redraft","reprofile","acknowledge","dismiss"]')