xsl/meida_backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

merge: T16 确定性视觉与 32 维诊断装配（T-P3-01/06）

Browse Source
This commit is contained in:
xsl
2026-08-14 22:15:37 +08:00
parent 597e29786c e7076c9c01
commit fc3393887f
8 changed files with 1947 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/report/diagnosis_stage.py
+737
View File
@@ -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),
)
packages/report/dimensions.py
+122
View File
@@ -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}
packages/report/vision.py
+304
View File
@@ -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},
)
packages/report/vision_stage.py
+70
View File
@@ -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),
)