270 lines
12 KiB
Python
270 lines
12 KiB
Python
"""MCDT 结果指标：验收节点与关键事实节点的评分卡输入模型。
|
||
|
||
起点事实（照实写清，不得淡化）：
|
||
- 此前 v6 批头 b7ac9d7 全仓 grep MCDT / 关键事实节点 / 验收节点为
|
||
0 命中：此前代码里根本没有 MCDT。
|
||
- 本模块只是评分卡输入模型，不是产品功能。不得声称 MCDT 已是
|
||
产品功能、已接入生产评分、已跑通真实审计循环。
|
||
- 本模块不改任何生产行为：不碰 #238 auto_apply（保持 False）、
|
||
不碰 #247“纠正为 pass 本身不解封”、不改 v1.1 契约、不接线到
|
||
build_scorecard / 服务器端点 / 门禁。本模块的输出只是“每维的
|
||
得分输入”，调用方是否采用、如何采用，须另行拍板。
|
||
|
||
设计（v7 机制第 8 项，只做这一项）：
|
||
- acceptance_node（验收节点）：一项可验收的结果主张，至少含
|
||
id、title、acceptance_criteria（列表）、status、evidence_refs。
|
||
另含 dimension：该节点映射到哪一维（7 维 id 之一）；无 dimension
|
||
或 dimension 非法时该节点不映射到任何维（记入未映射清单），
|
||
绝不靠猜测、按 id 前缀、按其它维借分。
|
||
- key_fact_node（关键事实节点）：一项支撑/证伪验收的关键事实，
|
||
至少含 id、fact、status、source。另含 dimension：与验收节点同
|
||
规则；无 dimension 的关键事实不影响任何维。
|
||
- status 词汇（两类节点共用）：pending / pass / fail / unknown。
|
||
pending = 尚未判定；unknown = 未知/未验证；fail = 已证伪或未达标；
|
||
pass = 已验收通过。其它任何值一律按 unknown 处理（fail-closed，
|
||
不把拼写错误读成通过）。
|
||
|
||
映射到哪几维（写清，不留含糊）：
|
||
- delivery_quality（交付质量）：验收节点证明交付结果达到验收标准。
|
||
- gate_audit（门禁与审计有效性）：验收节点证明门禁真实拦截、
|
||
审计结论可复核。
|
||
- arch_contract（架构与契约一致性）：验收节点证明契约冻结、
|
||
架构 / 目录 / 导入一致。
|
||
- 其余四维 test_coverage、security_ledger、cost_quota、ops_scale：
|
||
本项不预设节点。无对应验收节点时一律记 unknown，并在
|
||
missing_items 里说明；不得用过程指标（测试数、覆盖率）填补，
|
||
不得从其它维借分。注意 test_coverage 尤其如此：覆盖率数字
|
||
本身是过程指标，即使调用方手头有覆盖率，也不能因此把
|
||
test_coverage 从 unknown 改成 pass/fail——只有映射到该维的
|
||
验收节点状态变动才能改分。
|
||
|
||
映射规则（nodes_to_dimension_inputs）：
|
||
(a) 某维无对应验收节点时，该维输入为 unknown（不得用过程指标填补）。
|
||
(b) 过程指标（test_count、coverage）只作为证据字段透传，绝不直接
|
||
进入得分输入：它们只出现在返回值的顶层 process_evidence 里，
|
||
绝不出现在任何一维的 score_input / status / basis 里。只改
|
||
test_count / coverage 而节点状态不变时，每维的得分输入不变；
|
||
只改节点状态（pending -> pass 或 pass -> fail）时，映射输出变化。
|
||
节点 dict 上附带的 test_count / coverage 等额外键同样被忽略，
|
||
只读 id / title / acceptance_criteria / status / evidence_refs /
|
||
dimension（验收节点）与 id / fact / status / source / dimension
|
||
（关键事实节点）。
|
||
(c) 节点状态为 fail 时对应维不得为 pass；节点状态 unknown 时对应维
|
||
为 unknown。具体聚合（fail-closed，优先级 fail > unknown >
|
||
pending > pass 的反向表述见代码）：
|
||
- 任一映射验收节点 fail -> 该维 fail；
|
||
- 任一映射关键事实 fail -> 该维 fail（已证伪的关键事实不得
|
||
让该维通过，即使验收节点全 pass）；
|
||
- 否则任一映射验收节点 unknown 或 pending -> 该维 unknown；
|
||
- 否则任一映射关键事实 unknown 或 pending -> 该维 unknown
|
||
（支撑事实未验证，不得宣称通过）；
|
||
- 否则（映射验收节点全 pass，且映射关键事实全 pass 或无映射
|
||
关键事实）-> 该维 pass。
|
||
关键事实单独存在时不能让维通过：无验收节点时该维仍 unknown。
|
||
|
||
阈值口径（沿用 audit_scorecard.py）：
|
||
- thresholds 参数只用于报告 thresholds_source，绝不改变任何一维
|
||
的状态：传阈值不能把 unknown 变成 pass。
|
||
- 默认 thresholds_source = "test_default_unapproved"（测试默认值、
|
||
未拍板）；只有调用方显式传入且全部条目 source 均为
|
||
"user_approved" 时才记 "user_approved"。本模块绝不把测试默认
|
||
值写成生产已拍板；B / C 维与阈值本身不在本项决定范围。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from dataclasses import dataclass, field
|
||
|
||
# 维顺序与 id 与 cloud/audit_scorecard.py 的 DIMENSIONS 保持一致；
|
||
# 若那边增维、减维、改名，本模块须同步改（测试钉住两者相等）。
|
||
DIMENSION_IDS = [
|
||
"delivery_quality",
|
||
"gate_audit",
|
||
"arch_contract",
|
||
"test_coverage",
|
||
"security_ledger",
|
||
"cost_quota",
|
||
"ops_scale",
|
||
]
|
||
|
||
# 本项有映射语义说明的三维（其余四维无预设节点，见模块 docstring）。
|
||
MAPPED_DIMENSIONS_DOC = {
|
||
"delivery_quality": "验收节点证明交付结果达到验收标准。",
|
||
"gate_audit": "验收节点证明门禁真实拦截、审计结论可复核。",
|
||
"arch_contract": "验收节点证明契约冻结、架构 / 目录 / 导入一致。",
|
||
}
|
||
|
||
VALID_STATUSES = ("pending", "pass", "fail", "unknown")
|
||
|
||
MODEL_NOTE = (
|
||
"MCDT input model only, not a product feature: "
|
||
"v6 head b7ac9d7 had 0 hits for MCDT / acceptance nodes / "
|
||
"key fact nodes; this module only maps node statuses to "
|
||
"per-dimension score inputs and does not wire into scoring, "
|
||
"gates, or production."
|
||
)
|
||
|
||
|
||
@dataclass
|
||
class AcceptanceNode:
|
||
"""验收节点。status 非法值在映射时按 unknown 处理。"""
|
||
|
||
id: str
|
||
title: str
|
||
acceptance_criteria: list = field(default_factory=list)
|
||
status: str = "unknown"
|
||
evidence_refs: list = field(default_factory=list)
|
||
dimension: str | None = None
|
||
|
||
|
||
@dataclass
|
||
class KeyFactNode:
|
||
"""关键事实节点。status 非法值在映射时按 unknown 处理。"""
|
||
|
||
id: str
|
||
fact: str
|
||
status: str = "unknown"
|
||
source: str = ""
|
||
dimension: str | None = None
|
||
|
||
|
||
def _as_dict(node):
|
||
"""接受 dataclass 实例或 dict，一律转成 dict 只读视图。"""
|
||
if isinstance(node, dict):
|
||
return node
|
||
if hasattr(node, "__dataclass_fields__"):
|
||
return {
|
||
f: getattr(node, f) for f in node.__dataclass_fields__ # noqa: B905
|
||
}
|
||
raise TypeError("node must be a dict or dataclass, got %r" % type(node))
|
||
|
||
|
||
def _norm_status(value):
|
||
"""非法状态 -> unknown（fail-closed，不把拼写错误读成通过）。"""
|
||
if value in VALID_STATUSES:
|
||
return value
|
||
return "unknown"
|
||
|
||
|
||
def _thresholds_source(thresholds):
|
||
"""与 audit_scorecard.py 同口径：只有全 user_approved 才算已拍板。"""
|
||
if not thresholds:
|
||
return "test_default_unapproved"
|
||
sources = set()
|
||
for entry in thresholds.values():
|
||
if isinstance(entry, dict):
|
||
sources.add(entry.get("source", "test_default_unapproved"))
|
||
else:
|
||
sources.add("test_default_unapproved")
|
||
if sources == {"user_approved"}:
|
||
return "user_approved"
|
||
return "test_default_unapproved"
|
||
|
||
|
||
def nodes_to_dimension_inputs(acceptance_nodes, key_fact_nodes,
|
||
thresholds=None, process_metrics=None):
|
||
"""把节点状态映射为每维的得分输入。
|
||
|
||
acceptance_nodes / key_fact_nodes：AcceptanceNode / KeyFactNode
|
||
实例或同字段 dict 的列表。process_metrics：过程指标 dict
|
||
（如 {"test_count": ..., "coverage": ...}），只透传到顶层
|
||
process_evidence，绝不进入任何一维的得分输入。
|
||
返回：{"dimensions": {维 id: 维输入}, "dimension_order": [...],
|
||
"process_evidence": {...}, "unmapped_acceptance_node_ids": [...],
|
||
"unmapped_key_fact_node_ids": [...],
|
||
"thresholds_source": ..., "model_note": ...}。
|
||
每维输入含 dimension / status / score_input / acceptance_node_ids /
|
||
key_fact_node_ids / missing_items / basis；score_input 里只有
|
||
节点状态计数，绝无 test_count / coverage 键。
|
||
"""
|
||
acceptance_nodes = list(acceptance_nodes or [])
|
||
key_fact_nodes = list(key_fact_nodes or [])
|
||
|
||
per_dim_acc = {dim: [] for dim in DIMENSION_IDS}
|
||
per_dim_facts = {dim: [] for dim in DIMENSION_IDS}
|
||
unmapped_acc, unmapped_facts = [], []
|
||
|
||
for raw in acceptance_nodes:
|
||
node = _as_dict(raw)
|
||
dim = node.get("dimension")
|
||
if dim in DIMENSION_IDS:
|
||
per_dim_acc[dim].append(node)
|
||
else:
|
||
unmapped_acc.append(node.get("id"))
|
||
|
||
for raw in key_fact_nodes:
|
||
node = _as_dict(raw)
|
||
dim = node.get("dimension")
|
||
if dim in DIMENSION_IDS:
|
||
per_dim_facts[dim].append(node)
|
||
else:
|
||
unmapped_facts.append(node.get("id"))
|
||
|
||
dimensions = {}
|
||
for dim in DIMENSION_IDS:
|
||
acc = per_dim_acc[dim]
|
||
facts = per_dim_facts[dim]
|
||
acc_statuses = [_norm_status(n.get("status")) for n in acc]
|
||
fact_statuses = [_norm_status(n.get("status")) for n in facts]
|
||
missing, basis = [], []
|
||
if not acc:
|
||
missing.append("no_acceptance_node_mapped")
|
||
if dim not in MAPPED_DIMENSIONS_DOC:
|
||
missing.append(
|
||
"no_preset_node_for_dimension:"
|
||
" process metrics must not fill this dimension")
|
||
basis.append("no mapped acceptance node: dimension unknown")
|
||
status = "unknown"
|
||
elif "fail" in acc_statuses:
|
||
status = "fail"
|
||
basis.append("mapped acceptance node fail: dimension not pass")
|
||
elif "fail" in fact_statuses:
|
||
status = "fail"
|
||
basis.append("mapped key fact fail: dimension not pass")
|
||
elif "unknown" in acc_statuses or "pending" in acc_statuses:
|
||
status = "unknown"
|
||
basis.append(
|
||
"mapped acceptance node not yet pass"
|
||
" (unknown/pending): dimension unknown")
|
||
elif "unknown" in fact_statuses or "pending" in fact_statuses:
|
||
status = "unknown"
|
||
basis.append(
|
||
"mapped key fact not yet verified"
|
||
" (unknown/pending): dimension unknown")
|
||
else:
|
||
status = "pass"
|
||
basis.append(
|
||
"all mapped acceptance nodes pass and mapped key facts"
|
||
" pass or absent")
|
||
score_input = {
|
||
"node_status": status,
|
||
"acceptance_pass_count": acc_statuses.count("pass"),
|
||
"acceptance_fail_count": acc_statuses.count("fail"),
|
||
"acceptance_pending_count": acc_statuses.count("pending"),
|
||
"acceptance_unknown_count": acc_statuses.count("unknown"),
|
||
"key_fact_pass_count": fact_statuses.count("pass"),
|
||
"key_fact_fail_count": fact_statuses.count("fail"),
|
||
"key_fact_pending_count": fact_statuses.count("pending"),
|
||
"key_fact_unknown_count": fact_statuses.count("unknown"),
|
||
}
|
||
dimensions[dim] = {
|
||
"dimension": dim,
|
||
"status": status,
|
||
"score_input": score_input,
|
||
"acceptance_node_ids": [n.get("id") for n in acc],
|
||
"key_fact_node_ids": [n.get("id") for n in facts],
|
||
"missing_items": missing,
|
||
"basis": basis,
|
||
}
|
||
|
||
return {
|
||
"dimensions": dimensions,
|
||
"dimension_order": list(DIMENSION_IDS),
|
||
# 过程指标只在这里透传：顶层证据字段，不是得分输入。
|
||
"process_evidence": dict(process_metrics) if process_metrics else {},
|
||
"unmapped_acceptance_node_ids": unmapped_acc,
|
||
"unmapped_key_fact_node_ids": unmapped_facts,
|
||
"thresholds_source": _thresholds_source(thresholds),
|
||
"model_note": MODEL_NOTE,
|
||
}
|