From e7076c9c0155f60825da083345bf6e25988f4db1 Mon Sep 17 00:00:00 2001 From: xsl Date: Fri, 14 Aug 2026 22:14:54 +0800 Subject: [PATCH] feat(report): deterministic vision provider and 32-dimension diagnosis assembly MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit dimensions.py 逐行编码需求文档 §4（32 项四要素、缺失显式 skipped）+ VisionProvider 接口位与确定性实现 + 规则引擎联动 rule_hits（T-P3-01/06）。 实现： - dimensions.py：DimensionSpec 恰 32 项（number 1~32 连续，33/34 永不复用）， basic_coverage 口径机器可读：full 10 / label 3 / partial 8 / none 11， 初级覆盖 21/32（BASIC_FULL_KEYS + BASIC_COVERED_KEYS）。 - vision.py：VisionProvider Protocol（派生特征形状写进 docstring，真实 RTMPose/BiRefNet 实现后续按此替换）+ DeterministicVisionProvider（问卷 BMI/自评驱动体型与肤色，sha256(object_key+canonical问卷) 驱动其余枚举， 同输入同输出）+ build_vision_provider（未知名 CONFIG_ERROR）。 - vision_stage.py：derived_features 已存在 → skipped/reused（T-P3-09）； 否则分析 → 写中间态并 flush；photo_meta.quality_flags 透传进产物（T-P3-01）。 - diagnosis_stage.py：32 维恒全键，缺输入显式 skipped+原因（T-P3-06 红线）； 每维四要素 value/confidence/source/evidence/rule_hits + status；构造 RuleContext → RuleService 载入 published 规则 → evaluate，body_shape/color 类命中回落 #5/#19/#21，全量命中聚合进 #28/#30 的 rule_hits 与 evidence； 色板 #19/#20/#21 从 kb_color 按季型读取，库空 → skipped「色板未录入」。 测试：单测 29 项（32 项完整性/编号连续/确定性/工厂拒绝未知名；四要素齐全 性——故意缺问卷+缺视觉断言 32 键全在且全部 skipped 带原因；规则命中进 rule_hits 用内置最小规则集不连库）；集成 1 项（种 2 条 published 规则 → VisionStage → DiagnosisStage：中间态落库与复用 skipped/reused、32 维全键、 rule_hits 出现、kb_color 空时色板 skipped、coverage basic_full=10/32）。 全绿：ruff check/format、mypy packages/report、lint-imports；单测 536 通过 （3 处既有 skip）、集成 46 通过。 偏差说明： 1. 任务文档写 RulesService，实际类名为 RuleService（packages/rules/service.py），按实际代码调用。 2. kb_color 联表键采用四季代码 spring/summer/autumn/winter（仓库无既定口径），已在 vision.py 注释声明。 3. 任务文档只给 basic_coverage 四枚举，需求表格中的「粗判/8色/3色」归入 partial、「Top1」归入 label。 --- packages/report/diagnosis_stage.py | 737 ++++++++++++++++++++++++ packages/report/dimensions.py | 122 ++++ packages/report/vision.py | 304 ++++++++++ packages/report/vision_stage.py | 70 +++ tests/integration/test_report_stages.py | 223 +++++++ tests/unit/test_report_diagnosis.py | 262 +++++++++ tests/unit/test_report_dimensions.py | 87 +++ tests/unit/test_report_vision.py | 142 +++++ 8 files changed, 1947 insertions(+) create mode 100644 packages/report/diagnosis_stage.py create mode 100644 packages/report/dimensions.py create mode 100644 packages/report/vision.py create mode 100644 packages/report/vision_stage.py create mode 100644 tests/integration/test_report_stages.py create mode 100644 tests/unit/test_report_diagnosis.py create mode 100644 tests/unit/test_report_dimensions.py create mode 100644 tests/unit/test_report_vision.py diff --git a/packages/report/diagnosis_stage.py b/packages/report/diagnosis_stage.py new file mode 100644 index 0000000..9d37702 --- /dev/null +++ b/packages/report/diagnosis_stage.py @@ -0,0 +1,737 @@ +"""诊断装配阶段（name="diagnosis"）：问卷 + 派生特征 + 规则结论 → 32 维画像。 + +T-P3-06 红线：**任何输入组合下 ``result["diagnosis"]["dimensions"]`` 恒 32 项** +（number 1~32 全键在场）。输入不足的维度显式 ``status="skipped"`` 并把原因写进 +evidence——显式缺失优于编造，skipped 语义的本义正在于此。 + +四要素口径（对齐记忆系统 §3.5 的置信度分层）： + +- ``confidence``：问卷自评 0.6（主观）、确定性视觉 0.85（mock 但稳定）、 + 规则推导/知识库映射 0.9（硬结论）；skipped 恒 0.0。 +- ``source``：questionnaire / vision / rule / derived；skipped 用 "none" + （没有数据来源，避免误标成"来自问卷"）。 +- ``evidence``：一句人话依据（将直接出现在报告里）；skipped 时即缺失原因。 +- ``rule_hits``：该维度聚合到的命中规则键（规则引擎硬结论的可解释锚点）。 + +规则引擎联动：构造 ``RuleContext``（body_shape/face_shape/features/goals…）→ +``RuleService`` 载入 published 规则 → ``evaluate()`` → verdict 的 recommend/avoid +聚合进「核心诉求 #28 / 偏好与禁忌 #30」的 rule_hits 与 evidence；体型类 +（category=body_shape）与色彩类（category=color）命中分别回落到 #5/#19/#21。 + +色板维度（#19/#20/#21）：V1 从 ``kb_color`` 按季型代码读 base/accent/avoid； +库空 → 该维度 skipped + 「色板未录入」（显式缺失优于编造）。 + +纯函数 ``evaluate_dimensions`` 与 ``build_rule_context`` 不碰 I/O——单测用 +内置最小规则集（engine.evaluate 直算）即可覆盖四要素齐全性，无需连库。 +""" + +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from dataclasses import asdict, dataclass +from typing import Any + +from packages.kb import repo as kb_repo +from packages.report.dimensions import ( + BASIC_FULL_KEYS, + DIMENSION_BY_KEY, + DimensionSpec, +) +from packages.report.pipeline import ReportContext +from packages.report.vision import SEASON_CODE_CN, STYLE_TYPES +from packages.rules import dsl, engine +from packages.rules.service import RuleService +from packages.shared.logging import get_logger + +logger = get_logger(__name__) + +#: 体型代码 → 中文名（证据文本用；值本身保留字母码，与 kb/记忆系统词汇一致）。 +_BODY_SHAPE_CN: Mapping[str, str] = { + "A": "梨形", + "O": "苹果形", + "X": "沙漏形", + "T": "倒三角", + "H": "矩形", +} + +#: 中性色板的关键词（kb_color.role=base 且 name_cn 命中任一即视为中性色适配款）。 +_NEUTRAL_KEYWORDS: tuple[str, ...] = ("白", "黑", "灰", "米", "驼", "卡其") + +#: 初级口径的色板条数上限（需求文档 §4：最佳色板 8 色、避雷色 3 色）。 +_BEST_COLORS_LIMIT = 8 +_AVOID_COLORS_LIMIT = 3 + + +@dataclass(frozen=True, slots=True) +class DimensionResult: + """单个维度的装配结果（四要素 + status，序列化形状即报告产物）。""" + + number: int + key: str + value: Any # 结论值（枚举/数值/列表/字典）；skipped 时为 None + confidence: float # 0~1，口径见模块 docstring + source: str # questionnaire | vision | rule | derived | none + evidence: str # 一句人话依据；skipped 时为缺失原因 + rule_hits: list[str] # 命中的 rule_key + status: str # "ok" | "skipped" + + +@dataclass(frozen=True, slots=True) +class ColorEntry: + """kb_color 行的纯数据投影（求值纯函数不吃 ORM 对象，测试可手工构造）。""" + + season_code: str + role: str # base / accent / avoid + hex: str # 与 kb_color 列名保持一致 + name_cn: str + + +# ── 小工具 ───────────────────────────────────────────────────────────────── + + +def _str(value: Any) -> str | None: + return value if isinstance(value, str) and value else None + + +def _str_list(value: Any) -> list[str]: + if not isinstance(value, (list, tuple)): + return [] + return [item for item in value if isinstance(item, str) and item] + + +def _num(value: Any) -> float | None: + if isinstance(value, bool) or not isinstance(value, (int, float)): + return None + return float(value) + + +def _fmt(value: float) -> str: + """数值文本化：整数不带小数点（报告里 165cm 而不是 165.0cm）。""" + return str(int(value)) if value == int(value) else str(round(value, 1)) + + +def _rule_keys(verdict: engine.RuleVerdict, *categories: str) -> list[str]: + """命中键提取：给 categories 就是类别过滤，不给就是全量（聚合进 #28/#30）。""" + return [ + hit.rule_key for hit in verdict.rule_hits if not categories or hit.category in categories + ] + + +def _rule_evidence(verdict: engine.RuleVerdict, *categories: str) -> str: + """命中规则的解释文本（≤3 条，报告证据链的人话部分）。""" + explanations = [ + hit.explanation + for hit in verdict.rule_hits + if hit.explanation and (not categories or hit.category in categories) + ] + return "；".join(explanations[:3]) + + +def _ok( + spec: DimensionSpec, + value: Any, + source: str, + confidence: float, + evidence: str, + rule_hits: list[str] | None = None, +) -> DimensionResult: + return DimensionResult( + number=spec.number, + key=spec.key, + value=value, + confidence=confidence, + source=source, + evidence=evidence, + rule_hits=rule_hits or [], + status="ok", + ) + + +def _skip(spec: DimensionSpec, reason: str, rule_hits: list[str] | None = None) -> DimensionResult: + # rule_hits 可选：skipped 维度仍可携带规则命中链（可解释性不因缺数据丢失）。 + return DimensionResult( + number=spec.number, + key=spec.key, + value=None, + confidence=0.0, + source="none", + evidence=reason, + rule_hits=rule_hits or [], + status="skipped", + ) + + +def _spec(key: str) -> DimensionSpec: + return DIMENSION_BY_KEY[key] + + +# ── 纯函数：规则上下文构造与 32 维装配 ──────────────────────────────────── + + +def build_rule_context( + questionnaire: Mapping[str, Any], derived_features: Mapping[str, Any] +) -> dsl.RuleContext: + """问卷 + 派生特征 → 规则引擎画像（视觉侧优先，问卷补场景/风格/禁忌）。 + + occasion 取问卷场景权重中占比最高者（规则按单一主场景匹配；权重并列时 + 按场景名排序取稳定的最大值，保证确定性）。 + """ + occasion: str | None = None + weights = questionnaire.get("occasion_weights") + if isinstance(weights, dict) and weights: + numeric = { + str(k): v for k, v in weights.items() if isinstance(k, str) and _num(v) is not None + } + if numeric: + occasion = max(numeric, key=lambda k: (float(numeric[k]), k)) + + return dsl.RuleContext( + body_shape=_str(derived_features.get("body_shape")), + season=_str(derived_features.get("season_code")), + face_shape=_str(derived_features.get("face_shape")), + features=tuple(_str_list(derived_features.get("body_features"))), + goals=tuple(_str_list(questionnaire.get("core_goals"))), + occasion=occasion, + style=_str(questionnaire.get("psyche_style")), + user_taboos=tuple(_str_list(questionnaire.get("taboos"))), + ) + + +def evaluate_dimensions( + questionnaire: Mapping[str, Any], + derived_features: Mapping[str, Any], + verdict: engine.RuleVerdict, + kb_colors: Sequence[ColorEntry] = (), +) -> list[DimensionResult]: + """32 维装配（纯函数）：按 DIMENSIONS 顺序恒输出 32 项，一个不缺。""" + q, d = questionnaire, derived_features + results: list[DimensionResult] = [] + + # -- 问卷直采（#1~#4） --------------------------------------------------- + gender = _str(q.get("gender")) + if gender: + results.append(_ok(_spec("gender"), gender, "questionnaire", 0.6, f"问卷自评：{gender}")) + else: + results.append(_skip(_spec("gender"), "问卷未填写性别")) + + height, weight = _num(q.get("height")), _num(q.get("weight")) + parts: dict[str, float] = {} + if height is not None: + parts["height"] = height + if weight is not None: + parts["weight"] = weight + if parts: + desc = "、".join( + f"{'身高' if k == 'height' else '体重'} {_fmt(v)}{'cm' if k == 'height' else 'kg'}" + for k, v in parts.items() + ) + results.append( + _ok(_spec("height_weight"), parts, "questionnaire", 0.6, f"问卷自评：{desc}") + ) + else: + results.append(_skip(_spec("height_weight"), "问卷未填写身高/体重")) + + birthday = _str(q.get("birthday")) + if birthday: + results.append( + _ok(_spec("birthday"), birthday, "questionnaire", 0.6, f"问卷填写：{birthday}") + ) + else: + results.append(_skip(_spec("birthday"), "问卷未填写生日")) + + occupation = _str(q.get("occupation")) + if occupation: + results.append( + _ok(_spec("occupation"), occupation, "questionnaire", 0.6, f"问卷填写：{occupation}") + ) + else: + results.append(_skip(_spec("occupation"), "问卷未填写职业（全面报告维度）")) + + # -- 视觉：体型与比例（#5~#7） ------------------------------------------- + body_shape = _str(d.get("body_shape")) + if body_shape: + hits = _rule_keys(verdict, "body_shape") + evidence = ( + f"视觉分析：体型判定为 {body_shape}（{_BODY_SHAPE_CN.get(body_shape, body_shape)}）" + ) + rule_text = _rule_evidence(verdict, "body_shape") + if rule_text: + evidence = f"{evidence}；{rule_text}" + results.append(_ok(_spec("body_shape"), body_shape, "vision", 0.85, evidence, hits)) + else: + results.append(_skip(_spec("body_shape"), "视觉分析未产出体型（无照片或视觉阶段被跳过）")) + + body_features = _str_list(d.get("body_features")) + if d.get("body_features") is not None: + results.append( + _ok( + _spec("body_features"), + body_features, + "vision", + 0.85, + f"视觉分析：身体局部特征 {('、'.join(body_features)) or '无显著特征'}", + ) + ) + else: + results.append(_skip(_spec("body_features"), "视觉分析未产出身体局部特征")) + + body_ratio = _num(d.get("body_ratio")) if d.get("body_ratio") is not None else None + if body_ratio is not None: + results.append( + _ok( + _spec("body_ratio"), + body_ratio, + "vision", + 0.85, + f"视觉测算：上下身比例约 {_fmt(body_ratio)}", + ) + ) + else: + results.append(_skip(_spec("body_ratio"), "视觉分析未产出上下身比例（全面报告维度）")) + + # -- 视觉：面部（#8~#11） ------------------------------------------------- + face_shape = _str(d.get("face_shape")) + if face_shape: + results.append( + _ok(_spec("face_shape"), face_shape, "vision", 0.85, f"视觉分析：脸型为{face_shape}") + ) + else: + results.append(_skip(_spec("face_shape"), "视觉分析未产出脸型")) + + face_proportions = _str(d.get("face_proportions")) + if face_proportions: + results.append( + _ok( + _spec("face_proportions"), + face_proportions, + "vision", + 0.85, + f"视觉分析：三庭五眼{face_proportions}", + ) + ) + else: + results.append(_skip(_spec("face_proportions"), "视觉分析未产出三庭五眼标签")) + + facial_details = d.get("facial_details") + if isinstance(facial_details, dict) and facial_details: + results.append( + _ok( + _spec("facial_details"), + dict(facial_details), + "vision", + 0.85, + "视觉分析：" + "、".join(f"{k}{v}" for k, v in facial_details.items()), + ) + ) + else: + results.append(_skip(_spec("facial_details"), "视觉分析未产出五官细节标签")) + + face_depth = _num(d.get("face_depth")) if d.get("face_depth") is not None else None + if face_depth is not None: + results.append( + _ok( + _spec("face_depth"), + face_depth, + "vision", + 0.85, + f"视觉测算：面部立体度 {_fmt(face_depth)}/10", + ) + ) + else: + results.append(_skip(_spec("face_depth"), "视觉分析未产出面部立体度（全面报告维度）")) + + # -- 色彩：肤档与季型（#12~#18） ------------------------------------------ + skin_tone = _str(d.get("skin_tone")) + if skin_tone: + results.append( + _ok( + _spec("skin_tone"), skin_tone, "vision", 0.85, f"视觉 Lab 采样：肤色档位{skin_tone}" + ) + ) + else: + results.append(_skip(_spec("skin_tone"), "视觉分析未产出肤色档位")) + + for key, label, axis in ( + ("face_hue", "面部色相", "冷←→暖"), + ("face_lightness", "面部亮度", "暗沉←→白皙"), + ("face_chroma", "面部色度", "柔和←→鲜艳"), + ): + axis_value = d.get(key) + if isinstance(axis_value, dict) and axis_value.get("label"): + results.append( + _ok( + _spec(key), + dict(axis_value), + "vision", + 0.85, + f"色彩分析：{label}{axis_value['label']}（{axis} 连续值" + f" {axis_value.get('value')}）", + ) + ) + else: + results.append(_skip(_spec(key), f"色彩分析未产出{label}")) + + face_contrast = _str(d.get("face_contrast")) + if face_contrast: + results.append( + _ok( + _spec("face_contrast"), + face_contrast, + "vision", + 0.85, + f"色彩分析：面部对比度{face_contrast}", + ) + ) + else: + results.append(_skip(_spec("face_contrast"), "色彩分析未产出面部对比度（全面报告维度）")) + + season = _str(d.get("season_code")) + if season: + results.append( + _ok( + _spec("season_code"), + season, + "vision", + 0.85, + f"色彩分析：四季季型为{SEASON_CODE_CN.get(season, season)}（{season}）", + ) + ) + else: + results.append(_skip(_spec("season_code"), "色彩分析未产出四季季型")) + + season_detailed = _str(d.get("season_detailed")) + if season_detailed: + results.append( + _ok( + _spec("season_detailed"), + season_detailed, + "vision", + 0.85, + f"色彩分析：十二季细分为{season_detailed}", + ) + ) + else: + results.append(_skip(_spec("season_detailed"), "色彩分析未产出十二季细分（全面报告维度）")) + + # -- 知识库映射：色板（#19~#21） ------------------------------------------- + color_hits = _rule_keys(verdict, "color") + color_rule_text = _rule_evidence(verdict, "color") + season_colors = [c for c in kb_colors if not season or c.season_code == season] + + best = [c for c in season_colors if c.role in ("base", "accent")][:_BEST_COLORS_LIMIT] + if best: + evidence = ( + f"知识库色板映射（{SEASON_CODE_CN.get(season or '', season or '')}）：共 {len(best)} 色" + ) + if color_rule_text: + evidence = f"{evidence}；{color_rule_text}" + results.append( + _ok( + _spec("best_colors"), + [{"hex": c.hex, "name_cn": c.name_cn, "role": c.role} for c in best], + "derived", + 0.9, + evidence, + color_hits, + ) + ) + else: + results.append( + _skip( + _spec("best_colors"), "色板未录入（kb_color 无该季型数据）", color_hits + ) # skipped 也保留色彩规则命中链（可解释性不因缺数据丢失） + ) + + neutral = [ + c + for c in season_colors + if c.role == "base" and any(k in c.name_cn for k in _NEUTRAL_KEYWORDS) + ] + if neutral: + results.append( + _ok( + _spec("neutral_colors"), + [{"hex": c.hex, "name_cn": c.name_cn} for c in neutral], + "derived", + 0.9, + f"知识库映射：中性色适配款 {('、'.join(c.name_cn for c in neutral))}", + ) + ) + else: + reason = "色板未录入" if not season_colors else "色板中无中性色适配款" + results.append(_skip(_spec("neutral_colors"), f"{reason}（kb_color）")) + + avoid = [c for c in season_colors if c.role == "avoid"][:_AVOID_COLORS_LIMIT] + if avoid: + evidence = f"知识库映射：避雷色 {('、'.join(c.name_cn for c in avoid))}" + if color_rule_text: + evidence = f"{evidence}；{color_rule_text}" + results.append( + _ok( + _spec("avoid_colors"), + [{"hex": c.hex, "name_cn": c.name_cn} for c in avoid], + "derived", + 0.9, + evidence, + color_hits, + ) + ) + else: + results.append( + _skip(_spec("avoid_colors"), "色板未录入（kb_color 无该季型数据）", color_hits) + ) + + # -- 视觉：风格量感（#22~#25） --------------------------------------------- + volume = d.get("volume") + if isinstance(volume, dict) and volume.get("label"): + results.append( + _ok( + _spec("volume"), + dict(volume), + "vision", + 0.85, + f"视觉分析：量感{volume['label']}（总分 {volume.get('score')}）", + ) + ) + else: + results.append(_skip(_spec("volume"), "视觉分析未产出量感粗判")) + + curve = d.get("curve_straight") + if isinstance(curve, dict) and curve.get("label"): + results.append( + _ok( + _spec("curve_straight"), + dict(curve), + "vision", + 0.85, + f"视觉分析：轮廓{curve['label']}（直曲总分 {curve.get('score')}）", + ) + ) + else: + results.append(_skip(_spec("curve_straight"), "视觉分析未产出直曲粗判")) + + motion = _str(d.get("motion")) + if motion: + results.append( + _ok(_spec("motion"), motion, "vision", 0.85, f"视觉分析：动静判定为{motion}") + ) + else: + results.append(_skip(_spec("motion"), "视觉分析未产出动静判定（全面报告维度）")) + + gene_scores = d.get("gene_style_scores") + if isinstance(gene_scores, dict) and gene_scores: + # Top1 并列时按九型固定序取先者——输出必须完全确定。 + top1 = max(STYLE_TYPES, key=lambda s: float(gene_scores.get(s, 0))) + results.append( + _ok( + _spec("gene_style"), + {"top1": top1, "scores": dict(gene_scores)}, + "vision", + 0.85, + f"视觉分析：基因风格 Top1 为{top1}（{gene_scores.get(top1)} 分）", + ) + ) + else: + results.append(_skip(_spec("gene_style"), "视觉分析未产出基因风格评分")) + + # -- 问卷偏好与规则命名（#26~#30） ------------------------------------------ + all_hits = _rule_keys(verdict) + rule_text_all = _rule_evidence(verdict) + + psyche = _str(q.get("psyche_style")) + if psyche: + results.append( + _ok( + _spec("psyche_style"), psyche, "questionnaire", 0.6, f"偏好问卷：心理风格为{psyche}" + ) + ) + else: + results.append(_skip(_spec("psyche_style"), "问卷未完成心理风格测评")) + + gene_top1 = ( + max(STYLE_TYPES, key=lambda s: float(gene_scores.get(s, 0))) + if isinstance(gene_scores, dict) and gene_scores + else None + ) + if gene_top1 and psyche: + results.append( + _ok( + _spec("fusion_style_name"), + f"{gene_top1}底色 × {psyche}倾向", + "rule", + 0.9, + f"规则推导：视觉基因风格{gene_top1}与心理风格{psyche}融合命名", + ) + ) + elif gene_top1 or psyche: + only = gene_top1 or psyche + results.append( + _ok( + _spec("fusion_style_name"), + f"以{only}为主导的风格", + "rule", + 0.9, + f"规则推导：仅有一侧输入，以{only}为主导命名", + ) + ) + else: + results.append( + _skip(_spec("fusion_style_name"), "缺少基因风格与心理风格输入，无法命名融合风格") + ) + + goals = _str_list(q.get("core_goals")) + if goals: + evidence = f"问卷勾选：{('、'.join(goals))}" + if rule_text_all: + evidence = f"{evidence}；规则建议：{rule_text_all}" + # 规则命中链始终聚合（问卷没勾诉求时也不丢可解释性，见 skipped 分支）。 + results.append(_ok(_spec("core_goals"), goals, "questionnaire", 0.6, evidence, all_hits)) + else: + results.append(_skip(_spec("core_goals"), "问卷未勾选核心诉求", all_hits)) + + weights = q.get("occasion_weights") + if isinstance(weights, dict) and weights: + results.append( + _ok( + _spec("occasion_weights"), + dict(weights), + "questionnaire", + 0.6, + "问卷填写：场景权重 " + + "、".join(f"{k} {_fmt(float(v))}%" for k, v in weights.items()), + ) + ) + else: + results.append(_skip(_spec("occasion_weights"), "问卷未填写场景权重（全面报告维度）")) + + preferred_colors = _str_list(q.get("preferred_colors")) + disliked_styles = _str_list(q.get("disliked_styles")) + taboos = _str_list(q.get("taboos")) + rule_avoids = [ + f"回避 {item.slot}（{dim}：{'、'.join(values)}）" + for item in verdict.avoid + for dim, values in item.attrs.items() + ] + if preferred_colors or disliked_styles or taboos: + results.append( + _ok( + _spec("preferences_taboos"), + { + "preferred_colors": preferred_colors, + "disliked_styles": disliked_styles, + "taboos": taboos, + }, + "questionnaire", + 0.6, + "问卷填写：" + + "；".join( + filter( + None, + [ + f"偏好颜色{'、'.join(preferred_colors)}" if preferred_colors else "", + f"不接受风格{'、'.join(disliked_styles)}" if disliked_styles else "", + f"禁忌{'、'.join(taboos)}" if taboos else "", + ], + ) + ), + all_hits, + ) + ) + elif rule_avoids: + hard_keys = [b.rule_key for b in verdict.hard_blocks] + evidence = f"规则推导：{('；'.join(rule_avoids[:3]))}" + if verdict.blocked: + evidence = f"{evidence}（含硬约束 {('、'.join(hard_keys))}）" + results.append( + _ok( + _spec("preferences_taboos"), + {"preferred_colors": [], "disliked_styles": [], "taboos": rule_avoids}, + "rule", + 0.9, + evidence, + all_hits, + ) + ) + else: + results.append(_skip(_spec("preferences_taboos"), "问卷未填偏好与禁忌，且无规则回避结论")) + + # -- 视觉：毛发现状（#31~#32） ---------------------------------------------- + hair = d.get("hair_status") + if isinstance(hair, dict) and hair.get("length"): + results.append( + _ok( + _spec("hair_status"), + dict(hair), + "vision", + 0.85, + f"视觉分析：发长{hair.get('length')}、卷度{hair.get('curl')}", + ) + ) + else: + results.append(_skip(_spec("hair_status"), "视觉分析未产出毛发现状（全面报告维度）")) + + hair_color = d.get("hair_color") + if isinstance(hair_color, dict) and hair_color.get("hex"): + results.append( + _ok( + _spec("hair_color"), + dict(hair_color), + "vision", + 0.85, + f"视觉分析：发色 {hair_color['hex']}（偏{hair_color.get('tone')}）", + ) + ) + else: + results.append(_skip(_spec("hair_color"), "视觉分析未产出发色（全面报告维度）")) + + if len(results) != 32: # T-P3-06 红线：结构缺项是装配 bug，必须响亮失败。 + raise RuntimeError(f"32 维装配缺项：实际 {len(results)} 项") + return results + + +# ── 流水线阶段 ───────────────────────────────────────────────────────────── + + +class DiagnosisStage: + """流水线阶段：诊断装配。由 T15 的编排按 dotted path 动态装配（无参构造）。""" + + name = "diagnosis" + + async def run(self, ctx: ReportContext) -> None: + questionnaire = dict(ctx.report_input.questionnaire) + derived = dict(ctx.report_input.derived_features) + + verdict = await RuleService(ctx.session).evaluate( + build_rule_context(questionnaire, derived) + ) + + season = _str(derived.get("season_code")) + colors: list[ColorEntry] = [] + if season: + rows = await kb_repo.list_colors(ctx.session, season_code=season) + colors = [ + ColorEntry( + season_code=row.season_code, role=row.role, hex=row.hex, name_cn=row.name_cn + ) + for row in rows + ] + + results = evaluate_dimensions(questionnaire, derived, verdict, colors) + ok_basic_full = sum(1 for r in results if r.status == "ok" and r.key in BASIC_FULL_KEYS) + # JSONB 列整体重赋值而非原地改：SQLAlchemy 对 in-place 变更不可见。 + ctx.report.result = { + **(ctx.report.result or {}), + "diagnosis": { + "dimensions": [asdict(r) for r in results], + "coverage": {"basic_full": ok_basic_full, "total": len(results)}, + }, + } + # NFR-SEC-03：诊断结论不进日志，只记计数。 + logger.info( + "diagnosis_stage_done", + report_id=ctx.report.report_id, + total=len(results), + ok=sum(1 for r in results if r.status == "ok"), + basic_full=ok_basic_full, + rule_hits=len(verdict.rule_hits), + ) diff --git a/packages/report/dimensions.py b/packages/report/dimensions.py new file mode 100644 index 0000000..ba0a437 --- /dev/null +++ b/packages/report/dimensions.py @@ -0,0 +1,122 @@ +"""诊断维度字典（需求文档《美学报告-需求文档》§4 的机器可读形态）。 + +32 个维度是报告产出的**事实核心**：diagnosis_stage 逐维装配四要素（T-P3-06）， +任何输入组合下都必须输出全部 32 个维度——缺失只能显式 skipped，不允许缺键。 + +逐行口径（与需求文档 §4 表格一一对应，2026-08-14 版）： + +- ``source`` 按表「数据来源」列归类到六类：questionnaire（问卷）/ vision + （视觉分析）/ color（色彩算法）/ kb（知识库映射）/ rule（规则）/ llm。 + 「视觉 + 问卷」「规则 + LLM」这类复合来源取**主生产方**（视觉、规则）， + 复合事实写进对应维度的 name_cn 与注释。 +- ``basic_coverage`` 按表「初级」列归类：✅→full（完整输出）、标签/Top1→label + （仅标签）、部分/粗判/8 色/3 色→partial（粗判或子集）、❌→none（不输出）。 + +逐行统计（验收「初级 21/32」口径的来源，见需求文档 §4 表格下方统计）： + +- full（10 项）：#1 gender、#2 height_weight、#3 birthday、#5 body_shape、 + #8 face_shape、#12 skin_tone、#17 season_code、#26 psyche_style、 + #27 fusion_style_name、#28 core_goals； +- label（3 项）：#9 face_proportions、#10 facial_details、#25 gene_style； +- partial（8 项）：#6 body_features、#13 face_hue、#14 face_lightness、 + #19 best_colors、#21 avoid_colors、#22 volume、#23 curve_straight、 + #30 preferences_taboos； +- none（11 项）：#4 occupation、#7 body_ratio、#11 face_depth、#15 face_chroma、 + #16 face_contrast、#18 season_detailed、#20 neutral_colors、#24 motion、 + #29 occasion_weights、#31 hair_status、#32 hair_color。 + +合计 10 + 3 + 8 + 11 = 32；初级覆盖（非 none）= 10 + 3 + 8 = **21/32**。 + +编号 33（五行分布）/34（幸运色）已于 2026-08-13 随 TD-31 永久移除，**不复用**： +避免历史数据与文档交叉引用错位。 +""" + +from __future__ import annotations + +from dataclasses import dataclass + +#: basic_coverage 的合法取值（封闭枚举，防止拼写漂移出机器可读口径）。 +BASIC_COVERAGE_VALUES: frozenset[str] = frozenset({"full", "label", "partial", "none"}) + +#: source 的合法取值（按需求文档「数据来源」列归类的六类）。 +DIMENSION_SOURCES: frozenset[str] = frozenset( + {"questionnaire", "vision", "color", "kb", "rule", "llm"} +) + + +@dataclass(frozen=True, slots=True) +class DimensionSpec: + """单个诊断维度的静态定义（字典行）。 + + 只描述「这个维度是什么、数据从哪来、初级报告覆盖到什么程度」； + 具体取值与置信度由 diagnosis_stage 装配成 DimensionResult。 + """ + + number: int # 1~32（33/34 已永久废弃，不复用） + key: str # 稳定英文键，与 kb/记忆系统词汇对齐（body_shape/face_shape/season_code…） + name_cn: str + source: str # questionnaire | vision | color | kb | rule | llm + basic_coverage: str # "full" | "label" | "partial" | "none" + + +#: 32 个维度，按需求文档 §4 表格逐行编码；number 恰好 1~32 连续。 +DIMENSIONS: tuple[DimensionSpec, ...] = ( + # -- 问卷直采（基础信息） -------------------------------------------------- + DimensionSpec(1, "gender", "性别", "questionnaire", "full"), + DimensionSpec(2, "height_weight", "身高/体重", "questionnaire", "full"), + DimensionSpec(3, "birthday", "生日", "questionnaire", "full"), + DimensionSpec(4, "occupation", "职业", "questionnaire", "none"), + # -- 视觉分析（体型与比例） ------------------------------------------------ + DimensionSpec(5, "body_shape", "体型", "vision", "full"), + DimensionSpec(6, "body_features", "身体局部特征", "vision", "partial"), + DimensionSpec(7, "body_ratio", "上下身比例", "vision", "none"), + # -- 视觉分析（面部） ------------------------------------------------------ + DimensionSpec(8, "face_shape", "脸型", "vision", "full"), + DimensionSpec(9, "face_proportions", "三庭五眼", "vision", "label"), + DimensionSpec(10, "facial_details", "五官细节", "vision", "label"), + DimensionSpec(11, "face_depth", "面部立体度", "vision", "none"), + # -- 色彩（肤档与季型，视觉 Lab 采样 + 色彩算法） -------------------------- + DimensionSpec(12, "skin_tone", "肤色档位", "vision", "full"), + DimensionSpec(13, "face_hue", "面部色相", "color", "partial"), + DimensionSpec(14, "face_lightness", "面部亮度", "color", "partial"), + DimensionSpec(15, "face_chroma", "面部色度", "color", "none"), + DimensionSpec(16, "face_contrast", "面部对比度", "color", "none"), + DimensionSpec(17, "season_code", "四季季型", "color", "full"), + DimensionSpec(18, "season_detailed", "十二季细分", "color", "none"), + # -- 知识库映射（色板） ---------------------------------------------------- + DimensionSpec(19, "best_colors", "最佳色板", "kb", "partial"), + DimensionSpec(20, "neutral_colors", "中性色板", "kb", "none"), + DimensionSpec(21, "avoid_colors", "避雷色", "kb", "partial"), + # -- 视觉分析（风格量感） -------------------------------------------------- + DimensionSpec(22, "volume", "量感", "vision", "partial"), + DimensionSpec(23, "curve_straight", "直曲（轮廓）", "vision", "partial"), + DimensionSpec(24, "motion", "动静", "vision", "none"), # 表列「视觉 + 问卷」，主生产方为视觉 + DimensionSpec(25, "gene_style", "基因风格", "vision", "label"), # 初级 Top1 + # -- 问卷偏好（心理侧） ---------------------------------------------------- + DimensionSpec(26, "psyche_style", "心理风格", "questionnaire", "full"), + DimensionSpec(27, "fusion_style_name", "融合风格命名", "rule", "full"), # 表列「规则 + LLM」 + DimensionSpec(28, "core_goals", "核心诉求", "questionnaire", "full"), # 问卷 + 推导 + DimensionSpec(29, "occasion_weights", "场景权重", "questionnaire", "none"), + DimensionSpec(30, "preferences_taboos", "偏好与禁忌", "questionnaire", "partial"), + # -- 视觉分析（毛发现状） -------------------------------------------------- + DimensionSpec(31, "hair_status", "发长与卷度现状", "vision", "none"), + DimensionSpec(32, "hair_color", "发色现状", "vision", "none"), +) + +#: 初级报告**完整输出**（basic_coverage == "full"）的维度键集合，共 10 项。 +#: 这是验收「初级 21/32」口径里 21 = 10 full + 3 label + 8 partial 的机器可读形态。 +BASIC_FULL_KEYS: frozenset[str] = frozenset( + spec.key for spec in DIMENSIONS if spec.basic_coverage == "full" +) + +#: 初级报告**有产出**（full/label/partial，即 basic_coverage != "none"）的维度键集合。 +#: len == 21，对应需求文档 §4 统计表的「初级报告覆盖合计 21/32」。 +BASIC_COVERED_KEYS: frozenset[str] = frozenset( + spec.key for spec in DIMENSIONS if spec.basic_coverage != "none" +) + +#: number → spec（诊断装配按编号稳定排序输出用）。 +DIMENSION_BY_NUMBER: dict[int, DimensionSpec] = {spec.number: spec for spec in DIMENSIONS} + +#: key → spec（视觉/诊断阶段按维度键反查定义用）。 +DIMENSION_BY_KEY: dict[str, DimensionSpec] = {spec.key: spec for spec in DIMENSIONS} diff --git a/packages/report/vision.py b/packages/report/vision.py new file mode 100644 index 0000000..1e7080b --- /dev/null +++ b/packages/report/vision.py @@ -0,0 +1,304 @@ +"""视觉分析提供者：接口位 + V1 确定性（mock）实现。 + +**接口位**（T9 SPIKE 之后的正式契约）：真实 RTMPose/BiRefNet/Lab 采样实现后续 +按 ``VisionProvider`` Protocol 替换，流水线与其余阶段不感知差别。 + +**为什么 mock 要确定性**：黄金集与集成测试必须能对同一输入断言同一输出； +概率性 mock 会让"偶现失败"混进 CI。因此 ``DeterministicVisionProvider`` +不读照片字节（V1 骨架没有照片下载步骤），只从 ``questionnaire`` + +``photo_meta["object_key"]`` 的 sha256 派生特征——同输入恒同输出。 + +派生特征的形状（所有键即诊断维度键，两个例外见下）:: + + { + "body_shape": "A" | "O" | "X" | "T" | "H", # 梨/苹果/沙漏/倒三角/矩形 + "body_features": ["溜肩", ...], # 多标签，可为空列表 + "body_ratio": float, # 上下身比例（0.58~0.68） + "face_shape": "鹅蛋" | "圆" | "方" | "长" | "心形" | "菱形" | "梨形", + "face_proportions": "均匀" | "上庭偏长" | "中庭偏长" | "下庭偏长", + "facial_details": {"眉形": ..., "眼型": ..., "双眼皮": ..., "鼻型": ..., "唇形": ...}, + "face_depth": float, # 面部立体度 0~10 + "skin_lab": {"l": float, "a": float, "b": float}, # 中间量，不是维度键 + "skin_tone": "粉一白" | ... | "黄黑皮", # 六档 + "face_hue": {"value": float, "label": str}, # -1 冷 ←→ 1 暖 + "face_lightness": {"value": float, "label": str}, # -1 暗沉 ←→ 1 白皙 + "face_chroma": {"value": float, "label": str}, # -1 柔和 ←→ 1 鲜艳 + "face_contrast": "低" | "适中" | "高", + "season_code": "spring" | "summer" | "autumn" | "winter", + "season_detailed": "浅春" | ... | "深冬", # 十二季 + "volume": {"label": "小" | "中" | "大", "score": int}, + "curve_straight": {"label": "偏曲" | "适中" | "偏直", "score": float}, + "motion": "动态型" | "动静适中" | "静态型", + "gene_style_scores": {"戏剧": int, ...}, # 9 型评分；例外键：维度键为 gene_style + "hair_status": {"length": "短" | "中" | "长", "curl": "直" | "微卷" | "大卷"}, + "hair_color": {"hex": "#RRGGBB", "tone": "冷" | "暖"}, + } +""" + +from __future__ import annotations + +import hashlib +import json +from collections.abc import Mapping +from typing import Any, Protocol, runtime_checkable + +from packages.report.dimensions import DIMENSION_BY_KEY +from packages.shared.config import Settings +from packages.shared.errors import AppError, ErrorCode + +#: 体型五型（需求文档 §4 #5 取值范围；字母码与表格一致）。 +BODY_SHAPES: tuple[str, ...] = ("A", "O", "X", "T", "H") + +#: 脸型七型（需求文档 §4 #8）。 +FACE_SHAPES: tuple[str, ...] = ("鹅蛋", "圆", "方", "长", "心形", "菱形", "梨形") + +#: 肤色六档（需求文档 §4 #12）。 +SKIN_TONES: tuple[str, ...] = ("粉一白", "粉二白", "粉三白", "黄一白", "黄二白", "黄黑皮") + +#: 四季季型代码。kb_color.season_code 必须用同一套代码（色板维度的联表键）。 +SEASON_CODES: tuple[str, ...] = ("spring", "summer", "autumn", "winter") + +SEASON_CODE_CN: Mapping[str, str] = { + "spring": "春", + "summer": "夏", + "autumn": "秋", + "winter": "冬", +} + +#: 基因/心理风格九型（需求文档 §4 #25/#26）。 +STYLE_TYPES: tuple[str, ...] = ( + "戏剧", + "睿智", + "自然", + "古典", + "优雅", + "浪漫", + "前卫", + "少女", + "少年", +) + +#: 派生特征键 → 诊断维度键的例外映射（其余键与维度键同名）。 +#: skin_lab 是色彩算法的中间量，不对应任何维度。 +FEATURE_TO_DIMENSION: Mapping[str, str] = {"gene_style_scores": "gene_style"} + + +def dimensions_touched(features: Mapping[str, Any]) -> list[str]: + """派生特征覆盖到的诊断维度键（vision 阶段产物的 dimensions_touched）。""" + touched: list[str] = [] + for key in features: + dim = FEATURE_TO_DIMENSION.get(key, key) + if dim in DIMENSION_BY_KEY and dim not in touched: + touched.append(dim) + return touched + + +@runtime_checkable +class VisionProvider(Protocol): + """视觉分析 SPI：照片元信息 + 问卷 → 派生特征 dict（形状见模块 docstring）。 + + 实现必须**确定性**（同输入同输出），否则黄金集断言与失败重放都不成立。 + """ + + async def analyze( + self, photo_meta: dict[str, Any], questionnaire: dict[str, Any] + ) -> dict[str, Any]: ... + + +def _seed(object_key: str, questionnaire: Mapping[str, Any], salt: str) -> bytes: + """确定性种子：canonical JSON 保证 dict 键序无关，salt 分组隔离特征域。""" + canonical = json.dumps(questionnaire, sort_keys=True, ensure_ascii=False, separators=(",", ":")) + return hashlib.sha256(f"{object_key}|{canonical}|{salt}".encode()).digest() + + +def _num(value: Any) -> float | None: + """问卷数值字段的宽容读取：bool 不是数值（是 int 子类，须显式排除）。""" + if isinstance(value, bool) or not isinstance(value, (int, float)): + return None + return float(value) + + +def _bmi(questionnaire: Mapping[str, Any]) -> float | None: + height = _num(questionnaire.get("height")) + weight = _num(questionnaire.get("weight")) + if height is None or weight is None or height <= 0: + return None + return weight / (height / 100.0) ** 2 + + +def _pick(digest: bytes, index: int, options: tuple[str, ...]) -> str: + return options[digest[index] % len(options)] + + +class DeterministicVisionProvider: + """V1 骨架的确定性假实现：真实视觉模型的占位（报告链路先跑通、可回归）。 + + 派生规则（全部可复算，这里就是"模型"本身）： + + - **问卷驱动**： + - 身高/体重 → BMI：``<18.5`` → A/H 二选一；``18.5~24`` → X/H/T 三选一； + ``24~28`` → O/T 二选一；``≥28`` → O。缺失时按哈希在五型里均匀选。 + - 问卷自评（§3.6 降级路径的 body_shape_self / face_shape_self / + skin_tone_self）**优先于哈希**——自评是真实信号，mock 噪声不该盖过它； + skin_tone_self 同时反推 hue/lightness，保证季型/肤色档内部一致。 + - **哈希驱动**（sha256(object_key + canonical 问卷)，face/body 两组独立盐）： + - face 组：d0/d1/d2 → 色相/亮度/色度（-1~1 连续值），季型=色相冷热 × + 亮度深浅（暖+亮→春、冷+柔→夏、暖+深→秋、冷+亮→冬），十二季=季型 × + 色度/亮度细分；d3~d10 → 脸型/三庭五眼/五官/立体度。 + - body 组：b0 → 体型枚举选择；b1~b3 → 身体局部特征多标签； + b4 → 上下身比例（0.58~0.68）；b5/b6 → 量感(30~70)/直曲；b7 → 动静； + b8~b21 → 发长/卷度/发色/九型基因风格评分（3~10 分）。 + """ + + async def analyze( + self, photo_meta: dict[str, Any], questionnaire: dict[str, Any] + ) -> dict[str, Any]: + object_key = str(photo_meta.get("object_key", "")) + face = _seed(object_key, questionnaire, "face") + body = _seed(object_key, questionnaire, "body") + + # -- 色彩三轴：连续值 + 三档标签（粗判即标签口径） -------------------- + hue = round(face[0] / 255 * 2 - 1, 3) + lightness = round(face[1] / 255 * 2 - 1, 3) + chroma = round(face[2] / 255 * 2 - 1, 3) + + # 问卷肤色自评优先：粉系=冷（hue<0）、黄系=暖；白度档位反推亮度。 + skin_tone_self = questionnaire.get("skin_tone_self") + if isinstance(skin_tone_self, str) and skin_tone_self in SKIN_TONES: + skin_tone = skin_tone_self + hue = -abs(hue) - 0.05 if skin_tone.startswith("粉") else abs(hue) + 0.05 + if skin_tone.endswith("一白"): + lightness = 0.6 + elif skin_tone.endswith("二白"): + lightness = 0.2 + elif skin_tone == "粉三白": + lightness = -0.1 + else: # 黄黑皮 + lightness = -0.6 + else: + if lightness > 0.35: + skin_tone = "粉一白" if hue < 0 else "黄一白" + elif lightness > 0.1: + skin_tone = "粉二白" if hue < 0 else "黄二白" + elif lightness > -0.2: + skin_tone = "粉三白" + else: + skin_tone = "黄黑皮" + + warm = hue >= 0 + season_code = ( + "spring" + if warm and lightness >= 0 + else ("autumn" if warm else ("summer" if lightness >= 0 else "winter")) + ) + if season_code == "spring": + season_detailed = "亮春" if chroma > 0.25 else ("浅春" if lightness > 0.5 else "暖春") + elif season_code == "summer": + season_detailed = "浅夏" if lightness > 0.5 else ("柔夏" if chroma < -0.25 else "冷夏") + elif season_code == "autumn": + season_detailed = "柔秋" if chroma < -0.25 else ("深秋" if lightness < -0.5 else "暖秋") + else: + season_detailed = "亮冬" if chroma > 0.25 else ("深冬" if lightness < -0.5 else "冷冬") + + # -- 体型（问卷 BMI 主驱动，自评覆盖） -------------------------------- + body_shape_self = questionnaire.get("body_shape_self") + if isinstance(body_shape_self, str) and body_shape_self in BODY_SHAPES: + body_shape = body_shape_self + else: + bmi = _bmi(questionnaire) + if bmi is None: + body_shape = _pick(body, 0, BODY_SHAPES) + elif bmi < 18.5: + body_shape = _pick(body, 0, ("A", "H")) + elif bmi < 24: + body_shape = _pick(body, 0, ("X", "H", "T")) + elif bmi < 28: + body_shape = _pick(body, 0, ("O", "T")) + else: + body_shape = "O" + + feature_pool = ("溜肩", "平肩", "宽肩", "短颈", "长腰", "短腰", "粗腿") + body_features = [_pick(body, 1, feature_pool)] + extra = _pick(body, 2, feature_pool) + if extra not in body_features and body[3] % 2 == 1: + body_features.append(extra) + + volume_score = 30 + body[5] % 41 + curve_score = round(body[6] / 255 * 2 - 1, 3) + + face_shape_self = questionnaire.get("face_shape_self") + face_shape = ( + face_shape_self + if isinstance(face_shape_self, str) and face_shape_self in FACE_SHAPES + else _pick(face, 3, FACE_SHAPES) + ) + + gene_scores = {style: 3 + body[13 + i] % 8 for i, style in enumerate(STYLE_TYPES)} + hair_hex = f"#{40 + body[10] % 80:02X}{20 + body[11] % 60:02X}{body[12] % 80:02X}" + + return { + "body_shape": body_shape, + "body_features": body_features, + "body_ratio": round(0.58 + body[4] / 255 * 0.10, 3), + "face_shape": face_shape, + "face_proportions": _pick(face, 4, ("均匀", "上庭偏长", "中庭偏长", "下庭偏长")), + "facial_details": { + "眉形": _pick(face, 5, ("平直眉", "弯眉", "挑眉")), + "眼型": _pick(face, 6, ("杏仁眼", "圆眼", "细长眼")), + "双眼皮": _pick(face, 7, ("双眼皮", "内双")), + "鼻型": _pick(face, 8, ("直鼻", "翘鼻", "宽鼻")), + "唇形": _pick(face, 9, ("薄唇", "标准唇", "厚唇")), + }, + "face_depth": round(face[10] % 101 / 10, 1), + "skin_lab": { + "l": round(45 + (lightness + 1) * 20, 1), + "a": round(hue * 12, 1), + "b": round(20 + chroma * 15, 1), + }, + "skin_tone": skin_tone, + "face_hue": { + "value": round(hue, 3), + "label": "偏冷" if hue < -0.33 else ("偏暖" if hue > 0.33 else "中性"), + }, + "face_lightness": { + "value": round(lightness, 3), + "label": "白皙" if lightness > 0.33 else ("暗沉" if lightness < -0.33 else "适中"), + }, + "face_chroma": { + "value": round(chroma, 3), + "label": "鲜艳" if chroma > 0.33 else ("柔和" if chroma < -0.33 else "适中"), + }, + "face_contrast": "高" if chroma > 0.2 else ("低" if chroma < -0.2 else "适中"), + "season_code": season_code, + "season_detailed": season_detailed, + "volume": { + "label": "小" if volume_score < 40 else ("中" if volume_score <= 55 else "大"), + "score": volume_score, + }, + "curve_straight": { + "label": "偏曲" + if curve_score < -0.33 + else ("偏直" if curve_score > 0.33 else "适中"), + "score": curve_score, + }, + "motion": _pick(body, 7, ("动态型", "动静适中", "静态型")), + "gene_style_scores": gene_scores, + "hair_status": { + "length": _pick(body, 8, ("短", "中", "长")), + "curl": _pick(body, 9, ("直", "微卷", "大卷")), + }, + "hair_color": {"hex": hair_hex, "tone": "冷" if hue < 0 else "暖"}, + } + + +def build_vision_provider(settings: Settings) -> VisionProvider: + """按配置装配视觉提供者。未知名直接 CONFIG_ERROR——静默回退到默认实现 + 会让"配置了真实模型却跑着 mock"这种事故无人察觉。""" + name = settings.report_vision_provider + if name == "deterministic": + return DeterministicVisionProvider() + raise AppError( + ErrorCode.CONFIG_ERROR, + message="未知的报告视觉提供者", + data={"report_vision_provider": name}, + ) diff --git a/packages/report/vision_stage.py b/packages/report/vision_stage.py new file mode 100644 index 0000000..5063138 --- /dev/null +++ b/packages/report/vision_stage.py @@ -0,0 +1,70 @@ +"""视觉分析阶段（name="vision"）：照片元信息 + 问卷 → 派生特征中间态。 + +T-P3-09 的实现点：``rp_input.derived_features`` 已存在且非空时**跳过并标记 +reused**——LLM/渲染阶段失败重试时不重跑视觉分析，用户也不需要重新上传照片 +（这是报告链路"分步可恢复"的结构基础，见 packages/report/README.md）。 + +产物形状（``result["vision"]``）:: + + 首跑： {"status": "ok", "dimensions_touched": [...], "quality_flags": [...]} + 复用： {"skipped": true, "reused": true} +""" + +from __future__ import annotations + +from packages.report.pipeline import ReportContext +from packages.report.vision import VisionProvider, build_vision_provider, dimensions_touched +from packages.shared.logging import get_logger + +logger = get_logger(__name__) + + +class VisionStage: + """流水线阶段：视觉分析。由 T15 的编排按 dotted path 动态装配。""" + + name = "vision" + + def __init__(self, provider: VisionProvider | None = None) -> None: + # provider 可注入（单测/FakeProvider，参考 test_vertical_slice 的做法）； + # 留空则每次 run 时按 settings 装配。编排层用无参构造（importlib 动态 + # 导入），因此该参数必须有默认值。 + self._provider = provider + + async def run(self, ctx: ReportContext) -> None: + if ctx.report_input.derived_features: + # 中间态已落库：绝不重算（重算=同照片不同结论，报告可解释性破坏）。 + # JSONB 列整体重赋值而非原地改：SQLAlchemy 对 in-place 变更不可见。 + ctx.report.result = { + **(ctx.report.result or {}), + "vision": {"skipped": True, "reused": True}, + } + logger.info("vision_stage_reused", report_id=ctx.report.report_id) + return + + provider = self._provider or build_vision_provider(ctx.settings) + features = await provider.analyze( + dict(ctx.report_input.photo_meta), dict(ctx.report_input.questionnaire) + ) + + # 中间态立即 flush：后续阶段（或流水线重试）可见，T-P3-09 的落点。 + ctx.report_input.derived_features = features + await ctx.session.flush() + + # 质量标记（T-P3-01）：photo_meta 可选 quality_flags（模糊/背光等）， + # mock 骨架下默认空列表；透传进产物供报告页以"参考"语气呈现。 + quality_flags = list(ctx.report_input.photo_meta.get("quality_flags") or []) + touched = dimensions_touched(features) + ctx.report.result = { + **(ctx.report.result or {}), + "vision": { + "status": "ok", + "dimensions_touched": touched, + "quality_flags": quality_flags, + }, + } + logger.info( + "vision_stage_done", + report_id=ctx.report.report_id, + touched=len(touched), + quality_flags=len(quality_flags), + ) diff --git a/tests/integration/test_report_stages.py b/tests/integration/test_report_stages.py new file mode 100644 index 0000000..85132e7 --- /dev/null +++ b/tests/integration/test_report_stages.py @@ -0,0 +1,223 @@ +"""报告流水线视觉/诊断两阶段集成用例（真实 PostgreSQL）——只测 T16 的两个阶段。 + +覆盖任务契约：种 2 条 published 规则 → VisionStage → DiagnosisStage，断言 +derived_features 落库、32 维全键、规则 rule_hits 出现、kb_color 为空时 +色板维度 skipped。整条流水线的编排（装配/失败退款）归 T15 的集成测试。 + +确定性保证：问卷带 skin_tone_self=黄一白 → 暖+亮 → 季型恒为 spring， +因此色板断言不依赖哈希；body 规则的条件值先用 provider 离线算出再种入。 +""" + +from __future__ import annotations + +import os +import uuid +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager +from typing import Any + +import pytest + + +def _infra_available() -> bool: + return bool(os.getenv("MEIDA_RUN_INTEGRATION")) + + +pytestmark = [ + pytest.mark.integration, + pytest.mark.skipif( + not _infra_available(), + reason="set MEIDA_RUN_INTEGRATION=1 with PostgreSQL and Redis running", + ), +] + +_PREFIX = "R-IT16-" +#: 问卷肤色自评钉死的季型（黄一白 → 暖+亮 → spring），见 vision.py 派生规则。 +_SEASON = "spring" + +_QUESTIONNAIRE: dict[str, Any] = { + "gender": "女", + "height": 165, + "weight": 50, + "birthday": "1998-06-01", + "occupation": "设计师", + "core_goals": ["显高", "显瘦"], + "occasion_weights": {"通勤": 0.7, "休闲": 0.3}, + "psyche_style": "自然", + "preferred_colors": ["白色"], + "taboos": ["不接受深V"], + "skin_tone_self": "黄一白", +} + + +def _key(suffix: str) -> str: + return f"{_PREFIX}{suffix}-{uuid.uuid4().hex[:6]}" + + +@asynccontextmanager +async def _session() -> AsyncIterator[Any]: + """每个用例独立 engine：pytest-asyncio 每用例一个事件循环（同 test_rules_engine）。""" + from sqlalchemy.ext.asyncio import async_sessionmaker + + from packages.shared.config import get_settings + from packages.shared.db import create_engine + + engine = create_engine(get_settings()) + factory = async_sessionmaker(bind=engine, expire_on_commit=False, autoflush=False) + try: + async with factory() as session: + try: + yield session + await session.commit() + except Exception: + await session.rollback() + raise + finally: + await engine.dispose() + + +async def _cleanup(report_id: int, input_id: int) -> None: + from sqlalchemy import delete + + from packages.rules import service as rules_service + from packages.shared.models import KbColor, KbRule, RpInput, RpReport + + rules_service.invalidate_published_cache() + async with _session() as session: + await session.execute(delete(KbRule).where(KbRule.rule_key.like(f"{_PREFIX}%"))) + await session.execute(delete(RpInput).where(RpInput.input_id == input_id)) + await session.execute(delete(RpReport).where(RpReport.report_id == report_id)) + await session.execute(delete(KbColor).where(KbColor.season_code == _SEASON)) + + +def _body_payload(rule_key: str, body_shape: str) -> dict[str, Any]: + return { + "rule_key": rule_key, + "category": "body_shape", + "priority": 90, + "confidence": 0.95, + "conditions": {"body_shape": body_shape}, + "effects": { + "recommend": [{"slot": "top", "attrs": {"shoulder": ["IT16垫肩"]}}], + "explanation": "IT16 体型策略", + }, + } + + +def _color_payload(rule_key: str, season: str) -> dict[str, Any]: + return { + "rule_key": rule_key, + "category": "color", + "priority": 85, + "confidence": 0.9, + "conditions": {"season": season}, + "effects": { + "avoid": [{"slot": "dress", "attrs": {"color": ["IT16荧光绿"]}}], + "explanation": "IT16 季型避雷", + }, + } + + +async def test_vision_then_diagnosis_with_published_rules() -> None: + from packages.domain.ids import next_id + from packages.report.diagnosis_stage import DiagnosisStage + from packages.report.pipeline import ReportContext + from packages.report.vision import DeterministicVisionProvider + from packages.report.vision_stage import VisionStage + from packages.rules.service import RuleService + from packages.shared.config import get_settings + from packages.shared.models import RpInput, RpReport + + report_id, input_id = next_id(), next_id() + await _cleanup(report_id, input_id) + body_key, color_key = _key("BODY"), _key("COLOR") + try: + # 视觉结果是确定性的：先离线算出派生特征，用它的值种规则条件。 + photo_meta = { + "object_key": f"{_PREFIX}photo.jpg", + "quality_flags": ["backlit"], + } + derived = await DeterministicVisionProvider().analyze(photo_meta, _QUESTIONNAIRE) + assert derived["season_code"] == _SEASON # 问卷自评钉死季型，断言不依赖哈希 + + async with _session() as session: + svc = RuleService(session) + await svc.create(_body_payload(body_key, derived["body_shape"]), author="it16") + await svc.create(_color_payload(color_key, derived["season_code"]), author="it16") + await svc.publish(body_key) + await svc.publish(color_key) + + async with _session() as session: + report = RpReport( + report_id=report_id, + user_id=next_id(), + tier="basic", + status="generating", + result={}, + ) + session.add(report) + await session.flush() # rp_input.report_id 有外键，先落主表 + rp_input = RpInput( + input_id=input_id, + report_id=report_id, + questionnaire=_QUESTIONNAIRE, + photo_meta=photo_meta, + derived_features={}, + ) + session.add(rp_input) + await session.flush() + + ctx = ReportContext( + report=report, + report_input=rp_input, + session=session, + settings=get_settings(), + ) + + # -- VisionStage：首跑产出中间态 + 产物 -------------------------- + await VisionStage().run(ctx) + assert rp_input.derived_features == derived + assert report.result["vision"]["status"] == "ok" + assert report.result["vision"]["quality_flags"] == ["backlit"] # T-P3-01 透传 + assert "body_shape" in report.result["vision"]["dimensions_touched"] + + # -- VisionStage：中间态已存在 → skipped/reused（T-P3-09） -------- + await VisionStage().run(ctx) + assert report.result["vision"] == {"skipped": True, "reused": True} + + # -- DiagnosisStage：32 维全键 + 规则联动 + 空色板 skipped -------- + await DiagnosisStage().run(ctx) + diagnosis = report.result["diagnosis"] + dimensions = diagnosis["dimensions"] + assert [d["number"] for d in dimensions] == list(range(1, 33)) + by_key = {d["key"]: d for d in dimensions} + for dim in dimensions: + for field in ("value", "confidence", "source", "evidence", "rule_hits", "status"): + assert field in dim + + # 体型规则 → #5；色彩规则 → #21/#28/#30（聚合口径见 diagnosis_stage） + assert body_key in by_key["body_shape"]["rule_hits"] + assert color_key in by_key["avoid_colors"]["rule_hits"] + for key in ("core_goals", "preferences_taboos"): + assert {body_key, color_key} <= set(by_key[key]["rule_hits"]) + + # kb_color 为空 → 色板维度显式 skipped + 原因（显式缺失优于编造） + assert by_key["best_colors"]["status"] == "skipped" + assert "色板未录入" in by_key["best_colors"]["evidence"] + assert by_key["neutral_colors"]["status"] == "skipped" + + # 本场景问卷完整 + 视觉已跑：初级 full 10/10。 + assert diagnosis["coverage"] == {"basic_full": 10, "total": 32} + + # -- 落库验证：新会话读回中间态与产物 ------------------------------- + async with _session() as session: + saved_input = await session.get(RpInput, input_id) + assert saved_input is not None + assert saved_input.derived_features == derived + saved_report = await session.get(RpReport, report_id) + assert saved_report is not None + assert set(saved_report.result) == {"vision", "diagnosis"} + assert saved_report.result["vision"]["reused"] is True + assert saved_report.result["diagnosis"]["coverage"]["total"] == 32 + finally: + await _cleanup(report_id, input_id) diff --git a/tests/unit/test_report_diagnosis.py b/tests/unit/test_report_diagnosis.py new file mode 100644 index 0000000..f04c301 --- /dev/null +++ b/tests/unit/test_report_diagnosis.py @@ -0,0 +1,262 @@ +"""诊断装配：32 维四要素齐全、缺失显式 skipped、规则命中进 rule_hits（T-P3-06）。 + +规则求值走纯函数 ``engine.evaluate``（内置最小规则集，不连库）—— +``evaluate_dimensions`` 本身不碰 I/O，这正是它能被穷举输入组合的原因。 +""" + +from __future__ import annotations + +import json +from dataclasses import asdict +from typing import Any + +from packages.report.diagnosis_stage import ( + ColorEntry, + build_rule_context, + evaluate_dimensions, +) +from packages.report.dimensions import DIMENSION_BY_KEY, DIMENSIONS +from packages.report.vision import DeterministicVisionProvider +from packages.rules import dsl, engine + +_QUESTIONNAIRE = { + "gender": "女", + "height": 165, + "weight": 50, + "birthday": "1998-06-01 10:00", + "occupation": "设计师", + "core_goals": ["显高", "显瘦", "显白"], + "occasion_weights": {"通勤": 0.6, "休闲": 0.2, "约会": 0.2}, + "psyche_style": "自然", + "preferred_colors": ["白色", "蓝色"], + "disliked_styles": ["前卫"], + "taboos": ["不接受深V"], +} + + +def _kb_colors_for(season: str) -> list[ColorEntry]: + """给定季型的完整色板（base 含中性色命名，覆盖 #19/#20/#21 三个维度）。""" + return [ + ColorEntry(season_code=season, role="base", hex="#FFFFFF", name_cn="柔白"), + ColorEntry(season_code=season, role="base", hex="#2B2B2B", name_cn="炭黑"), + ColorEntry(season_code=season, role="base", hex="#6B6B6B", name_cn="中灰"), + ColorEntry(season_code=season, role="accent", hex="#0F4C81", name_cn="宝蓝"), + ColorEntry(season_code=season, role="avoid", hex="#E4E444", name_cn="芥末黄"), + ] + + +def _rule(rule_key: str, category: str, conditions: dict[str, Any]) -> dsl.Rule: + return dsl.Rule.model_validate( + { + "rule_key": rule_key, + "category": category, + "priority": 90, + "conditions": conditions, + "effects": { + "recommend": [{"slot": "top", "attrs": {"shoulder": ["垫肩"]}}], + "avoid": [{"slot": "dress", "attrs": {"color": ["荧光绿"]}}], + "explanation": f"{rule_key} 命中", + }, + } + ) + + +async def _derived() -> dict[str, Any]: + return await DeterministicVisionProvider().analyze( + {"object_key": "unit/report-vision.jpg"}, _QUESTIONNAIRE + ) + + +async def _verdict(derived: dict[str, Any]) -> engine.RuleVerdict: + rules = [ + _rule("T16-BODY", "body_shape", {"body_shape": derived["body_shape"]}), + _rule("T16-COLOR", "color", {"season": derived["season_code"]}), + ] + return engine.evaluate(rules, build_rule_context(_QUESTIONNAIRE, derived)) + + +def _by_key(results: list[Any]) -> dict[str, Any]: + return {r.key: r for r in results} + + +def _assert_four_elements(result: Any) -> None: + # 四要素 + 定位字段：一个都不能少（T-P3-06 验收原文）。 + for field in ("value", "confidence", "source", "evidence", "rule_hits", "status"): + assert hasattr(result, field) + assert 0.0 <= result.confidence <= 1.0 + assert result.evidence, f"{result.key} 的 evidence 不能为空" + assert isinstance(result.rule_hits, list) + + +# ── 32 键恒在场（红线） ───────────────────────────────────────────────────── + + +async def test_full_inputs_produce_all_32_dimensions() -> None: + derived = await _derived() + results = evaluate_dimensions( + _QUESTIONNAIRE, derived, await _verdict(derived), _kb_colors_for(derived["season_code"]) + ) + assert [r.number for r in results] == list(range(1, 33)) + assert [r.key for r in results] == [spec.key for spec in DIMENSIONS] + for result in results: + _assert_four_elements(result) + assert result.status == "ok", f"{result.key} 应有产出：{result.evidence}" + + +async def test_missing_everything_still_returns_32_keys_all_skipped() -> None: + # 故意缺问卷 + 缺视觉：32 键全在、全部显式 skipped 且带原因（不允许缺键）。 + verdict = engine.evaluate((), dsl.RuleContext()) + results = evaluate_dimensions({}, {}, verdict) + assert len(results) == 32 + for result in results: + _assert_four_elements(result) + assert result.status == "skipped" + assert result.value is None + assert result.confidence == 0.0 + assert result.source == "none" + assert result.evidence # 缺失原因必须显式给出 + + +async def test_questionnaire_only_keeps_32_keys_with_vision_skipped() -> None: + # 问卷版轻画像（§3.6）：视觉维度 skipped，问卷维度照常输出。 + verdict = engine.evaluate((), dsl.RuleContext()) + results = evaluate_dimensions(_QUESTIONNAIRE, {}, verdict) + by_key = _by_key(results) + assert len(results) == 32 + for key in ("body_shape", "face_shape", "skin_tone", "season_code", "hair_color"): + assert by_key[key].status == "skipped" + for key in ("gender", "birthday", "occupation", "psyche_style", "core_goals"): + assert by_key[key].status == "ok" + assert by_key[key].source == "questionnaire" + assert by_key[key].confidence == 0.6 + + +async def test_assembly_is_deterministic() -> None: + derived = await _derived() + verdict = await _verdict(derived) + colors = _kb_colors_for(derived["season_code"]) + first = evaluate_dimensions(_QUESTIONNAIRE, derived, verdict, colors) + second = evaluate_dimensions(_QUESTIONNAIRE, derived, verdict, colors) + assert [asdict(r) for r in first] == [asdict(r) for r in second] + + +# ── 规则引擎联动（rule_hits / evidence） ─────────────────────────────────── + + +async def test_rule_hits_land_on_related_dimensions() -> None: + derived = await _derived() + verdict = await _verdict(derived) + by_key = _by_key(evaluate_dimensions(_QUESTIONNAIRE, derived, verdict)) + + # 体型类命中 → #5 体型维度。 + assert "T16-BODY" in by_key["body_shape"].rule_hits + assert "T16-BODY" in by_key["body_shape"].evidence + + # 核心诉求/偏好与禁忌聚合全部命中（recommend/avoid 服务于诉求与禁忌）。 + for key in ("core_goals", "preferences_taboos"): + assert set(by_key[key].rule_hits) >= {"T16-BODY", "T16-COLOR"} + + +async def test_color_rules_carry_hits_even_when_kb_empty() -> None: + # kb_color 为空 → 色板 skipped，但色彩规则命中链仍在（可解释性不丢）。 + derived = await _derived() + verdict = await _verdict(derived) + by_key = _by_key(evaluate_dimensions(_QUESTIONNAIRE, derived, verdict)) + for key in ("best_colors", "avoid_colors"): + assert by_key[key].status == "skipped" + assert "色板未录入" in by_key[key].evidence + assert "T16-COLOR" in by_key[key].rule_hits + + +async def test_rule_derived_taboos_when_questionnaire_empty() -> None: + # 问卷没有偏好禁忌、但规则有 avoid → #30 以规则为源产出。 + derived = await _derived() + verdict = await _verdict(derived) + by_key = _by_key(evaluate_dimensions({}, derived, verdict)) + prefs = by_key["preferences_taboos"] + assert prefs.status == "ok" + assert prefs.source == "rule" + assert prefs.confidence == 0.9 + assert any("回避" in taboo for taboo in prefs.value["taboos"]) + + +# ── 色板（kb_color）与知识库映射 ─────────────────────────────────────────── + + +async def test_color_palettes_from_kb_entries() -> None: + derived = await _derived() + verdict = await _verdict(derived) + colors = _kb_colors_for(derived["season_code"]) + by_key = _by_key(evaluate_dimensions(_QUESTIONNAIRE, derived, verdict, colors)) + + best = by_key["best_colors"] + assert best.status == "ok" + assert len(best.value) <= 8 # 初级口径：最佳色板 8 色 + assert best.source == "derived" + assert {c.hex for c in colors if c.role != "avoid"} >= {c["hex"] for c in best.value} + + avoid = by_key["avoid_colors"] + assert avoid.status == "ok" + assert len(avoid.value) == 1 # 初级口径：避雷色 3 色为上限 + assert avoid.value[0]["name_cn"] == "芥末黄" + + neutral = by_key["neutral_colors"] + assert neutral.status == "ok" + assert {c["name_cn"] for c in neutral.value} == {"柔白", "炭黑", "中灰"} + + +async def test_other_season_colors_are_not_mixed_in() -> None: + derived = await _derived() + verdict = await _verdict(derived) + # 只录他季色板：本季必须 skipped 而不是混入他季颜色（色板映射按季型联表）。 + other = "summer" if derived["season_code"] != "summer" else "winter" + by_key = _by_key(evaluate_dimensions(_QUESTIONNAIRE, derived, verdict, _kb_colors_for(other))) + assert by_key["best_colors"].status == "skipped" + assert "色板未录入" in by_key["best_colors"].evidence + assert by_key["neutral_colors"].status == "skipped" + + +# ── 融合风格命名与序列化 ─────────────────────────────────────────────────── + + +async def test_fusion_style_naming_degrades_gracefully() -> None: + derived = await _derived() + verdict = await _verdict(derived) + both = _by_key(evaluate_dimensions(_QUESTIONNAIRE, derived, verdict))["fusion_style_name"] + assert both.status == "ok" + assert both.source == "rule" + assert both.confidence == 0.9 + assert "自然" in both.value # 心理风格侧参与了命名 + + only_questionnaire = evaluate_dimensions( + _QUESTIONNAIRE, {}, engine.evaluate((), dsl.RuleContext()) + ) + fused = _by_key(only_questionnaire)["fusion_style_name"] + assert fused.status == "ok" + assert "自然" in fused.value + + +async def test_results_serialize_to_json() -> None: + derived = await _derived() + verdict = await _verdict(derived) + payload = [asdict(r) for r in evaluate_dimensions(_QUESTIONNAIRE, derived, verdict)] + assert len(json.loads(json.dumps(payload, ensure_ascii=False))) == 32 + + +def test_build_rule_context_maps_fields() -> None: + derived = { + "body_shape": "A", + "season_code": "autumn", + "face_shape": "圆", + "body_features": ["溜肩", "短颈"], + } + ctx = build_rule_context(_QUESTIONNAIRE, derived) + assert ctx.body_shape == "A" + assert ctx.season == "autumn" + assert ctx.face_shape == "圆" + assert ctx.features == ("溜肩", "短颈") + assert ctx.goals == ("显高", "显瘦", "显白") + assert ctx.style == "自然" + assert ctx.user_taboos == ("不接受深V",) + assert ctx.occasion == "通勤" # 占比最高的场景 + assert DIMENSION_BY_KEY["body_shape"].number == 5 diff --git a/tests/unit/test_report_dimensions.py b/tests/unit/test_report_dimensions.py new file mode 100644 index 0000000..d901aac --- /dev/null +++ b/tests/unit/test_report_dimensions.py @@ -0,0 +1,87 @@ +"""维度字典：32 项完整性、编号连续、33/34 废弃不复用、初级口径机器可读（T-P3-06）。""" + +from __future__ import annotations + +from packages.report.dimensions import ( + BASIC_COVERAGE_VALUES, + BASIC_COVERED_KEYS, + BASIC_FULL_KEYS, + DIMENSION_BY_KEY, + DIMENSION_BY_NUMBER, + DIMENSION_SOURCES, + DIMENSIONS, +) + +#: 需求文档 §4 逐行统计出的初级 full 覆盖集（10 项）——本文件用硬编码预期 +#: 对照 BASIC_FULL_KEYS，任何一边漂移都会立刻暴露。 +_EXPECTED_BASIC_FULL = { + "gender", + "height_weight", + "birthday", + "body_shape", + "face_shape", + "skin_tone", + "season_code", + "psyche_style", + "fusion_style_name", + "core_goals", +} + +#: 初级有产出（full + label + partial）= 21 项，即验收「初级 21/32」口径。 +_EXPECTED_BASIC_COVERED_COUNT = 21 + + +def test_exactly_32_dimensions_with_consecutive_numbers() -> None: + assert len(DIMENSIONS) == 32 + assert [spec.number for spec in DIMENSIONS] == list(range(1, 33)) + assert len(DIMENSION_BY_NUMBER) == 32 + assert len(DIMENSION_BY_KEY) == 32 + + +def test_keys_unique_and_stable() -> None: + keys = [spec.key for spec in DIMENSIONS] + assert len(set(keys)) == 32 + # 字典键与 kb/记忆系统词汇对齐的锚点抽查。 + for anchor in ("body_shape", "face_shape", "season_code"): + assert anchor in DIMENSION_BY_KEY + + +def test_retired_numbers_33_34_never_reused() -> None: + # TD-31（2026-08-13）移除五行分布/幸运色，编号永久废弃。 + assert 33 not in DIMENSION_BY_NUMBER + assert 34 not in DIMENSION_BY_NUMBER + all_names = "".join(spec.name_cn for spec in DIMENSIONS) + assert "五行" not in all_names + assert "幸运色" not in all_names + + +def test_enums_legal() -> None: + for spec in DIMENSIONS: + assert spec.basic_coverage in BASIC_COVERAGE_VALUES + assert spec.source in DIMENSION_SOURCES + assert spec.name_cn, spec.key + + +def test_basic_full_keys_match_row_by_row_count() -> None: + # 机器可读口径 vs 逐行统计：两边必须同时成立。 + assert BASIC_FULL_KEYS == _EXPECTED_BASIC_FULL + assert ( + frozenset(spec.key for spec in DIMENSIONS if spec.basic_coverage == "full") + == BASIC_FULL_KEYS + ) + assert len(BASIC_FULL_KEYS) == 10 + + +def test_basic_covered_keys_is_21_of_32() -> None: + derived = frozenset(spec.key for spec in DIMENSIONS if spec.basic_coverage != "none") + assert derived == BASIC_COVERED_KEYS + assert len(BASIC_COVERED_KEYS) == _EXPECTED_BASIC_COVERED_COUNT + assert BASIC_FULL_KEYS <= BASIC_COVERED_KEYS + + +def test_coverage_split_matches_requirement_stats() -> None: + # 需求文档 §4 统计表：初级 full 10 / label+partial 11 / none 11，合计 32。 + counts = {"full": 0, "label": 0, "partial": 0, "none": 0} + for spec in DIMENSIONS: + counts[spec.basic_coverage] += 1 + assert counts == {"full": 10, "label": 3, "partial": 8, "none": 11} diff --git a/tests/unit/test_report_vision.py b/tests/unit/test_report_vision.py new file mode 100644 index 0000000..99b7601 --- /dev/null +++ b/tests/unit/test_report_vision.py @@ -0,0 +1,142 @@ +"""确定性视觉提供者：同输入同输出、问卷驱动、枚举合法、工厂拒绝未知名。""" + +from __future__ import annotations + +import pytest + +from packages.report.dimensions import DIMENSION_BY_KEY +from packages.report.vision import ( + BODY_SHAPES, + FACE_SHAPES, + SEASON_CODES, + SKIN_TONES, + STYLE_TYPES, + DeterministicVisionProvider, + VisionProvider, + build_vision_provider, + dimensions_touched, +) +from packages.shared.config import Settings +from packages.shared.errors import AppError, ErrorCode + +_PHOTO = {"object_key": "upload/report/photo-001.jpg"} + +_QUESTIONNAIRE = { + "gender": "女", + "height": 165, + "weight": 50, + "birthday": "1998-06-01", +} + + +async def test_same_input_same_output() -> None: + provider = DeterministicVisionProvider() + first = await provider.analyze(_PHOTO, _QUESTIONNAIRE) + second = await provider.analyze(dict(_PHOTO), dict(_QUESTIONNAIRE)) + assert first == second + + +async def test_satisfies_protocol() -> None: + assert isinstance(DeterministicVisionProvider(), VisionProvider) + + +async def test_shape_contract_and_enum_values() -> None: + features = await DeterministicVisionProvider().analyze(_PHOTO, _QUESTIONNAIRE) + expected_keys = { + "body_shape", + "body_features", + "body_ratio", + "face_shape", + "face_proportions", + "facial_details", + "face_depth", + "skin_lab", + "skin_tone", + "face_hue", + "face_lightness", + "face_chroma", + "face_contrast", + "season_code", + "season_detailed", + "volume", + "curve_straight", + "motion", + "gene_style_scores", + "hair_status", + "hair_color", + } + assert set(features) == expected_keys + assert features["body_shape"] in BODY_SHAPES + assert features["face_shape"] in FACE_SHAPES + assert features["skin_tone"] in SKIN_TONES + assert features["season_code"] in SEASON_CODES + assert set(features["gene_style_scores"]) == set(STYLE_TYPES) + # 色彩轴连续值域与标签三档。 + for axis in ("face_hue", "face_lightness", "face_chroma"): + assert -1 <= features[axis]["value"] <= 1 + assert features[axis]["label"] + assert 0 <= features["face_depth"] <= 10 + assert 0.58 <= features["body_ratio"] <= 0.68 + + +async def test_different_questionnaire_drives_different_body_shape() -> None: + provider = DeterministicVisionProvider() + slim = await provider.analyze(_PHOTO, {"height": 165, "weight": 45}) # BMI≈16.5 + heavy = await provider.analyze(_PHOTO, {"height": 165, "weight": 95}) # BMI≈34.9 + assert slim["body_shape"] in ("A", "H") # 过瘦区间：梨/矩形 + assert heavy["body_shape"] == "O" # 高 BMI：苹果形 + assert slim["body_shape"] != heavy["body_shape"] + + +async def test_questionnaire_self_report_overrides_hash() -> None: + # §3.6 降级路径：问卷自评是真实信号，优先于 mock 哈希。 + provider = DeterministicVisionProvider() + plain = await provider.analyze(_PHOTO, {}) + self_reported = await provider.analyze( + _PHOTO, + {"body_shape_self": "X", "face_shape_self": "圆", "skin_tone_self": "粉一白"}, + ) + assert self_reported["body_shape"] == "X" + assert self_reported["face_shape"] == "圆" + assert self_reported["skin_tone"] == "粉一白" + # 自评肤色反推冷轴：粉系=冷 → 季型落在冷季（夏/冬）。 + assert self_reported["season_code"] in ("summer", "winter") + assert plain["season_code"] in SEASON_CODES + + +async def test_different_photo_changes_features() -> None: + provider = DeterministicVisionProvider() + a = await provider.analyze({"object_key": "a.jpg"}, {}) + b = await provider.analyze({"object_key": "b.jpg"}, {}) + assert a != b + + +async def test_missing_object_key_still_deterministic() -> None: + provider = DeterministicVisionProvider() + empty_meta: dict[str, str] = {} + assert await provider.analyze(empty_meta, {}) == await provider.analyze({}, {}) + + +def test_dimensions_touched_maps_to_dimension_keys() -> None: + features = { + "body_shape": "A", + "skin_lab": {"l": 60.0, "a": 1.0, "b": 20.0}, # 中间量，不是维度 + "gene_style_scores": {"自然": 9}, + } + touched = dimensions_touched(features) + assert touched == ["body_shape", "gene_style"] + assert all(key in DIMENSION_BY_KEY for key in touched) + + +def test_factory_default_is_deterministic() -> None: + settings = Settings(_env_file=None) + provider = build_vision_provider(settings) + assert isinstance(provider, DeterministicVisionProvider) + + +def test_factory_rejects_unknown_provider() -> None: + settings = Settings(_env_file=None, report_vision_provider="rtmpose") + with pytest.raises(AppError) as excinfo: + build_vision_provider(settings) + assert excinfo.value.code == ErrorCode.CONFIG_ERROR + assert excinfo.value.data["report_vision_provider"] == "rtmpose"