feat(report): deterministic vision provider and 32-dimension diagnosis assembly

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。
This commit is contained in:
xsl
2026-08-14 22:14:54 +08:00
parent d974526904
commit e7076c9c01
8 changed files with 1947 additions and 0 deletions
+223
View File
@@ -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)
+262
View File
@@ -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
+87
View File
@@ -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}
+142
View File
@@ -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"