diff --git a/backend/contracts/prescreen.py b/backend/contracts/prescreen.py index 0793323..cfaaa9b 100644 --- a/backend/contracts/prescreen.py +++ b/backend/contracts/prescreen.py @@ -527,3 +527,45 @@ class KnowledgeCard(Contract): if self.effective_at and self.expires_at and self.expires_at <= self.effective_at: raise ValueError("有效期必须晚于生效时间") return self + + +# ---------------------------------------------------------------- 评审方案维度(T06) +class ScaleLevel(Contract): + """刻度的一级。刻度由版本化标准驱动,**不由模型任意打分**(14 §5)。""" + + value: int + label: str = Field(min_length=1, max_length=80) + criterion: str = Field(min_length=1, max_length=DESC_MAX) + + +class DimensionSpec(Contract): + """核查维度定义(docs/v2/03-detailed-design.md §9.2「方案维度解析」)。 + + 维度是 Skills/Rules 的**元数据**,不新增"维度卡"(14 §5)。同 ID 必须标签、 + 量纲与刻度一致;不一致时**拒绝合并**而不是让模型临时猜(BR-28 / E-20)。 + """ + + id: Identifier + label: str = Field(min_length=1, max_length=80) + order: int = Field(ge=0) + scale_id: str = Field(min_length=1, max_length=ID_MAX) + scale_version: str = Field(min_length=1, max_length=70) + levels: list[ScaleLevel] = Field(min_length=2, max_length=10) + component_refs: list[Ref] = Field(default_factory=list, max_length=64) + required: bool = False + + @model_validator(mode="after") + def _check_scale(self) -> DimensionSpec: + values = [level.value for level in self.levels] + if len(set(values)) != len(values): + raise ValueError("同一刻度的层级取值不得重复") + if values != sorted(values): + raise ValueError("刻度层级必须按取值升序声明,避免展示时语义倒置") + ref_ids = [ref.id for ref in self.component_refs] + if len(set(ref_ids)) != len(ref_ids): + raise ValueError("同一维度引用的组件不得重复") + return self + + def scale_key(self) -> tuple[str, str]: + """跨维度可比性的唯一判据:刻度标识 + 版本。""" + return (self.scale_id, self.scale_version) diff --git a/backend/services/review_plan.py b/backend/services/review_plan.py new file mode 100644 index 0000000..a5de7d8 --- /dev/null +++ b/backend/services/review_plan.py @@ -0,0 +1,113 @@ +"""T06-a 方案维度解析:去重、刻度冲突与呈现判定。 + +依据:docs/v2/14-experience-loop.md §5、03-detailed-design.md §9.2; +BR-28 / E-20 / AC-29 / AC-44。 + +三条不可妥协的规则: +1. **同 ID 必须语义一致**:标签、量纲(scale_id/version)、刻度层级任一不同即 + **拒绝合并**并报冲突,绝不让模型临时猜一个(BR-28)。相同定义只保留一份。 +2. **异刻度不强行归一化**:不同刻度的维度不得共用雷达,改为分面展示(14 §5)。 +3. **不默默丢失关键项**:呈现判定返回的 radar / listed / hard_block 三组并集 + 必须等于全部维度;任何维度都要出现在某一组里(AC-44)。 +""" + +from __future__ import annotations + +from contracts.prescreen import DimensionSpec + +RADAR_MIN, RADAR_MAX = 3, 8 + +EMPTY_REASON = "当前方案尚未配置核查维度" +FACETED_REASON = "维度刻度不一致,分面展示;不强行归一化,避免误导读者比较不可比的量" +RADAR_REASON = "同一刻度且 3–8 维,可用雷达并附等价文字列表" +FEW_REASON = "少于 3 维,用状态条/列表比雷达更清楚" +MANY_REASON = "多于 8 维,雷达会不可读,改为分组/列表且全部维度仍可查看" + + +def _definition(spec: dict) -> tuple: + """判据:标签 + 刻度标识/版本 + 层级内容。任一不同即视为不同定义。""" + return ( + str(spec.get("label", "")), + str(spec.get("scale_id", "")), + str(spec.get("scale_version", "")), + tuple((level.get("value"), level.get("label"), level.get("criterion")) + for level in spec.get("levels", [])), + ) + + +def _validate(spec: dict) -> dict: + try: + return DimensionSpec(**spec).model_dump(mode="json") + except Exception as exc: + raise ValueError(f"维度不合法:{exc}") from exc + + +def resolve_dimensions(specs) -> list[dict]: + """按 dimension_id 合并;同 ID 不同定义 → 冲突拒绝(AC-29 / E-20)。 + + 返回按 order 稳定排序后的维度列表,**同一定义只出现一次**。 + """ + merged: dict[str, dict] = {} + for raw in specs or (): + spec = _validate(raw) + existing = merged.get(spec["id"]) + if existing is None: + merged[spec["id"]] = spec + continue + if _definition(existing) != _definition(spec): + raise ValueError( + f"维度 {spec['id']} 存在不一致定义(标签/刻度/层级不同),请统一后保存方案;" + "不得由程序或模型临时选择其一" + ) + # 完全相同的定义:合并引用,避免重复计数(14 §5) + references = {ref["id"]: ref for ref in existing["component_refs"]} + for ref in spec["component_refs"]: + references.setdefault(ref["id"], ref) + existing["component_refs"] = list(references.values()) + existing["required"] = existing["required"] or spec["required"] + return sorted(merged.values(), key=lambda item: (item["order"], item["id"])) + + +def is_single_scale(dimensions) -> bool: + """全部维度是否共用同一可解释刻度(可比性前提)。""" + return len({(d["scale_id"], d["scale_version"]) for d in dimensions}) <= 1 + + +def presentation(dimensions, hard_block_ids=()) -> dict: + """AC-44 的呈现判定:雷达 / 列表 / 分组 / 分面 / 空。 + + 硬阻断维度**单列**,不参与雷达面积(14 §5:不被平均数或面积覆盖)。 + """ + ids = [item["id"] for item in dimensions] + if not dimensions: + return {"layout": "empty", "reason": EMPTY_REASON, + "radar_ids": [], "listed_ids": [], "hard_block_ids": []} + + blocks = [item_id for item_id in ids if item_id in set(hard_block_ids)] + remaining = [item_id for item_id in ids if item_id not in set(blocks)] + + if not is_single_scale(dimensions): + return {"layout": "faceted", "reason": FACETED_REASON, + "radar_ids": [], "listed_ids": remaining, "hard_block_ids": blocks} + + count = len(remaining) + if count < RADAR_MIN: + layout, reason, radar = "list", FEW_REASON, [] + elif count > RADAR_MAX: + layout, reason, radar = "grouped", MANY_REASON, [] + else: + layout, reason, radar = "radar", RADAR_REASON, remaining + listed = [item_id for item_id in remaining if item_id not in set(radar)] + return {"layout": layout, "reason": reason, + "radar_ids": radar, "listed_ids": listed, "hard_block_ids": blocks} + + +def assert_nothing_dropped(result: dict, dimensions) -> None: + """呈现判定不得丢失任何维度(AC-44:不默默丢失关键项)。""" + shown = set(result["radar_ids"]) | set(result["listed_ids"]) | set(result["hard_block_ids"]) + expected = {item["id"] for item in dimensions} + missing = expected - shown + if missing: + raise ValueError(f"呈现判定丢失了维度:{sorted(missing)}") + if shown - expected: + raise ValueError(f"呈现判定出现了未知维度:{sorted(shown - expected)}") diff --git a/backend/test_review_plan.py b/backend/test_review_plan.py new file mode 100644 index 0000000..6a565b0 --- /dev/null +++ b/backend/test_review_plan.py @@ -0,0 +1,149 @@ +"""T06-a 方案维度解析测试:AC-29 / AC-44。 + +维度是 Skills/Rules 的元数据,不新增"维度卡"(14 §5)。本文件锁定三件事: +- 同 ID 不同定义必须拒绝合并(不让模型或程序临时挑一个) +- 相同定义只出现一次,引用被合并而不是重复计数 +- 呈现判定覆盖 0 / 2 / 6 / 9 维与异刻度,且**不丢失任何维度** +""" + +import unittest + +from services import review_plan as plan + +SCALE = "evidence-stage" +LEVELS = [ + {"value": 0, "label": "未提供", "criterion": "材料未提供"}, + {"value": 1, "label": "仅有说明", "criterion": "仅客户经理口头或文字说明"}, + {"value": 2, "label": "材料待核", "criterion": "提供了材料但未交叉核对"}, + {"value": 3, "label": "交叉核验", "criterion": "两个独立来源一致"}, + {"value": 4, "label": "人工复核", "criterion": "评审员已人工核实"}, +] + + +def spec(dimension_id="d1", label="回款依据", order=0, scale_id=SCALE, scale_version="v1", + levels=None, required=False, refs=None): + return {"id": dimension_id, "label": label, "order": order, + "scale_id": scale_id, "scale_version": scale_version, + "levels": levels or LEVELS, "component_refs": refs or [], "required": required} + + +def many(count, **over): + return [spec(f"d{index}", f"维度{index}", index, **over) for index in range(count)] + + +class DeduplicationTests(unittest.TestCase): + def test_identical_definitions_merge_into_one(self): + # AC-29:相同定义只出现一次 + merged = plan.resolve_dimensions([ + spec("d1", refs=[{"id": "skill.a", "version": "1"}]), + spec("d1", refs=[{"id": "skill.b", "version": "1"}]), + ]) + self.assertEqual(len(merged), 1) + self.assertEqual(sorted(ref["id"] for ref in merged[0]["component_refs"]), + ["skill.a", "skill.b"]) + + def test_same_id_different_label_is_a_conflict(self): + with self.assertRaises(ValueError) as ctx: + plan.resolve_dimensions([spec("d1"), spec("d1", label="另一个名字")]) + self.assertIn("不一致定义", str(ctx.exception)) + self.assertIn("不得由程序或模型临时选择其一", str(ctx.exception)) + + def test_same_id_different_scale_id_is_a_conflict(self): + with self.assertRaises(ValueError): + plan.resolve_dimensions([spec("d1"), spec("d1", scale_id="other-scale")]) + + def test_same_scale_id_different_version_is_a_conflict(self): + # 刻度改版后语义可能变化,必须视为不同定义 + with self.assertRaises(ValueError): + plan.resolve_dimensions([spec("d1"), spec("d1", scale_version="v2")]) + + def test_same_id_different_levels_is_a_conflict(self): + changed = [dict(level) for level in LEVELS] + changed[3] = {"value": 3, "label": "两来源一致", "criterion": "改了判据"} + with self.assertRaises(ValueError): + plan.resolve_dimensions([spec("d1"), spec("d1", levels=changed)]) + + def test_required_flag_is_unioned_on_merge(self): + merged = plan.resolve_dimensions([spec("d1"), spec("d1", required=True)]) + self.assertTrue(merged[0]["required"]) + + def test_order_is_stable_and_sorted(self): + merged = plan.resolve_dimensions([spec("db", order=5), spec("da", order=1)]) + self.assertEqual([item["id"] for item in merged], ["da", "db"]) + + def test_invalid_dimension_is_rejected(self): + broken = spec() + broken["levels"] = [{"value": 0, "label": "只有一个层级", "criterion": "c"}] + with self.assertRaises(ValueError) as ctx: + plan.resolve_dimensions([broken]) + self.assertIn("维度不合法", str(ctx.exception)) + + +class PresentationTests(unittest.TestCase): + def check(self, dimensions, hard=()): + result = plan.presentation(dimensions, hard) + plan.assert_nothing_dropped(result, dimensions) # 不丢项是每次都要过的硬断言 + return result + + def test_empty_plan_says_so_without_placeholder_axes(self): + # UI Spec:空维度显示"尚未配置",不补六轴占位 + result = self.check([]) + self.assertEqual(result["layout"], "empty") + self.assertIn("尚未配置核查维度", result["reason"]) + + def test_two_dimensions_use_list_not_radar(self): + result = self.check(many(2)) + self.assertEqual(result["layout"], "list") + self.assertEqual(result["radar_ids"], []) + + def test_six_dimensions_use_radar_plus_equivalent_list(self): + result = self.check(many(6)) + self.assertEqual(result["layout"], "radar") + self.assertEqual(len(result["radar_ids"]), 6) + self.assertIn("等价文字列表", result["reason"]) + + def test_nine_dimensions_are_grouped_not_radar(self): + result = self.check(many(9)) + self.assertEqual(result["layout"], "grouped") + self.assertEqual(result["radar_ids"], []) + self.assertEqual(len(result["listed_ids"]), 9, "全部维度仍可查看") + + def test_three_and_eight_are_the_radar_boundaries(self): + self.assertEqual(self.check(many(3))["layout"], "radar") + self.assertEqual(self.check(many(8))["layout"], "radar") + + def test_mixed_scales_force_faceted_display(self): + # 异刻度不强行归一化 + dimensions = [spec("d1"), spec("d2", "负债率", 1, scale_id="ratio-scale")] + result = self.check(dimensions) + self.assertEqual(result["layout"], "faceted") + self.assertEqual(result["radar_ids"], []) + + def test_hard_blocks_are_listed_separately_and_never_in_the_radar(self): + # 14 §5:硬阻断单列,不被平均数或面积覆盖 + dimensions = many(6) + result = self.check(dimensions, hard={"d0"}) + self.assertEqual(result["hard_block_ids"], ["d0"]) + self.assertNotIn("d0", result["radar_ids"]) + self.assertEqual(len(result["radar_ids"]), 5) + self.assertEqual(result["layout"], "radar") + + def test_hard_blocks_are_separated_even_when_faceted(self): + dimensions = [spec("d1"), spec("d2", "负债率", 1, scale_id="ratio-scale")] + result = self.check(dimensions, hard={"d1"}) + self.assertEqual(result["hard_block_ids"], ["d1"]) + self.assertIn("d2", result["listed_ids"]) + + def test_dropping_a_dimension_would_be_caught(self): + # 守卫本身必须真的会报错,否则"不丢项"只是口号 + with self.assertRaises(ValueError): + plan.assert_nothing_dropped( + {"radar_ids": ["d0"], "listed_ids": [], "hard_block_ids": []}, many(2)) + + def test_unknown_scale_versions_never_share_a_radar(self): + dimensions = [spec("d1"), spec("d2", "同刻度不同版本", 1, scale_version="v2")] + self.assertEqual(self.check(dimensions)["layout"], "faceted") + + +if __name__ == "__main__": + unittest.main()