xsl
cb8a1e3c0f
CI / lint / typecheck / import-linter / tests / secrets (push) Canceled after 0s
FE↔后端联调实测暴露并修复： 1. 问卷键名三处错位（FE height_cm/weight_kg/psych_style vs 文档 height/weight/psyche_style）——vision._bmi 多键取值、diagnosis 维度 装配与规则上下文双键或运算，两侧谁改键名都不再炸 2. celery worker asyncio.run 每投递起新 loop + 模块级引擎缓存 → asyncpg 连接跨 loop（同进程第 3 个报告任务必现 RuntimeError: attached to a different loop；集成测试单循环掩盖）。新增 apps/worker/loop.py 持久 loop，report/ping 两任务接入 E2E 终态： psyche_style/height_weight 恢复、BMI 体型确定值、ok 26/32、 真实 LLM 文案通过校验 degraded=False。
318 lines
14 KiB
Python
318 lines
14 KiB
Python
"""视觉分析提供者：接口位 + 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 _q_num(questionnaire: Mapping[str, Any], *keys: str) -> float | None:
|
||
"""多键取值：FE 发 height_cm/weight_kg，文档口径 height/weight——都认。
|
||
|
||
FE（ReportCreatePage）用带单位的键名，本函数按传入顺序取第一个可解析
|
||
数值，联调两侧谁改键名都不炸。
|
||
"""
|
||
for key in keys:
|
||
value = _num(questionnaire.get(key))
|
||
if value is not None:
|
||
return value
|
||
return None
|
||
|
||
|
||
def _bmi(questionnaire: Mapping[str, Any]) -> float | None:
|
||
height = _q_num(questionnaire, "height_cm", "height")
|
||
weight = _q_num(questionnaire, "weight_kg", "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},
|
||
)
|