fix:文明

server/app/features/user_profile_scoring/__init__.py

@@ -0,0 +1,9 @@
+"""
+User Profile Scoring（用户画像打分）V1.2
+
+说明：
+- 提供“问卷答案（可跳过）→ 用户画像（可计算、可观测、可版本化）”的服务端实现
+- 规则以 `spec_kit/User Profile Scoring/spec.md`（V1.2）与
+  `设计说明文档/客戶端問卷打分規則.md`（V1.2）为准
+"""
+

server/app/features/user_profile_scoring/scoring.py

@@ -0,0 +1,194 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Optional
+
+from app.features.user_profile_scoring.types import (
+    HardRules,
+    ProfileAnswered,
+    QuestionnaireAnswersV1_2,
+    UserProfileV1_2_Extended,
+    UserStageOneHot,
+)
+
+
+def _clamp(value: float, min_value: float, max_value: float) -> float:
+    if value != value:  # NaN
+        return min_value
+    return max(min_value, min(max_value, value))
+
+
+def normalize_answers(raw: QuestionnaireAnswersV1_2) -> QuestionnaireAnswersV1_2:
+    """
+    归一化答案：
+    - Pydantic 已对枚举做了校验；此处仅统一 None/缺失的语义为“跳过”
+    """
+
+    # 直接返回一份拷贝，保持纯函数语义
+    return QuestionnaireAnswersV1_2.model_validate(raw.model_dump())
+
+
+def compute_profile_answered(answers: QuestionnaireAnswersV1_2) -> ProfileAnswered:
+    return ProfileAnswered(
+        stage=answers.mom_stage is not None,
+        emotion=answers.emotion is not None,
+        context=answers.context is not None,
+        need=answers.need is not None,
+    )
+
+
+def compute_time_confidence(generated_at: datetime, now: datetime) -> float:
+    """
+    时间衰减置信度（conf_time）
+    - 0–7 天：1.0
+    - 7–30 天：线性衰减到 0.7（含第 30 天）
+    - 30 天以上：0.5
+    """
+
+    delta = (now - generated_at).total_seconds()
+    if delta <= 0:
+        return 1.0
+
+    days = delta / (24 * 60 * 60)
+    if days <= 7:
+        return 1.0
+    if days <= 30:
+        t = (days - 7) / (30 - 7)  # 0..1
+        return 1.0 - 0.3 * t
+    return 0.5
+
+
+def compute_profile_confidence(conf_time: float, answered: ProfileAnswered) -> float:
+    """
+    V1.2：profile_confidence（conf_U）
+    conf = clamp(conf_time * (0.5 + 0.5 * completion), 0.2, 1.0)
+    """
+
+    answered_count = sum(
+        [
+            1 if answered.stage else 0,
+            1 if answered.emotion else 0,
+            1 if answered.context else 0,
+            1 if answered.need else 0,
+        ]
+    )
+    completion = answered_count / 4
+    completion_factor = 0.5 + 0.5 * completion
+    return _clamp(float(conf_time) * float(completion_factor), 0.2, 1.0)
+
+
+def _build_stage_one_hot(mom_stage: Optional[str]) -> UserStageOneHot:
+    # V1.2：mom_stage 跳过按安全策略输出 unknown=1
+    if mom_stage is None:
+        return UserStageOneHot(unknown=1)
+
+    return UserStageOneHot(
+        expecting=1 if mom_stage == "expecting" else 0,
+        parenting=1 if mom_stage == "parenting" else 0,
+        unknown=1 if mom_stage == "unknown" else 0,
+    )
+
+
+def _map_emotion_score(emotion: Optional[str]) -> Optional[float]:
+    if emotion is None:
+        return None
+    mapping = {
+        "low": 0.0,
+        "overwhelmed": 0.2,
+        "tired": 0.4,
+        "neutral": 0.6,
+        "calm": 0.8,
+        "joyful": 1.0,
+    }
+    return mapping.get(emotion)
+
+
+def _build_sparse_one_hot(value: Optional[str]) -> dict[str, int]:
+    if value is None:
+        return {}
+    return {value: 1}
+
+
+@dataclass(frozen=True)
+class _RuleOutput:
+    rule_hits: list[str]
+    hard_rules: HardRules
+
+
+def _compute_rule_output(stage: UserStageOneHot, emotion_score: Optional[float]) -> _RuleOutput:
+    rule_hits: list[str] = []
+    forbidden_risk_flags: list[str] = []
+
+    stage_unknown = stage.unknown == 1
+    stage_parenting = stage.parenting == 1
+
+    if stage_unknown:
+        rule_hits.append("unsafe_for_stage_unknown")
+        forbidden_risk_flags.append("unsafe_for_stage_unknown")
+
+    if stage_parenting:
+        rule_hits.append("unsafe_for_stage_parenting")
+        forbidden_risk_flags.append("unsafe_for_stage_parenting")
+
+    if emotion_score is not None and emotion_score <= 0.2:
+        rule_hits.append("unsafe_for_emotion_low")
+        forbidden_risk_flags.append("unsafe_for_emotion_low")
+
+    forbidden_content_predicates = []
+    if stage_unknown:
+        forbidden_content_predicates.append(
+            {
+                "id": "unknown_block_parenting_pressure_personalized",
+                "when_user": {"stage_unknown": True},
+                "forbid_content": {"need": "parenting_pressure", "personalization_power": 1},
+            }
+        )
+
+    return _RuleOutput(
+        rule_hits=rule_hits,
+        hard_rules=HardRules(
+            forbidden_risk_flags=forbidden_risk_flags,
+            forbidden_content_predicates=forbidden_content_predicates,
+        ),
+    )
+
+
+def build_user_profile_from_questionnaire(
+    raw_answers: QuestionnaireAnswersV1_2,
+    *,
+    generated_at: Optional[datetime] = None,
+    now: Optional[datetime] = None,
+) -> UserProfileV1_2_Extended:
+    """
+    主入口：问卷答案（可跳过）→ 用户画像（V1.2）+ 硬规则输出
+    """
+
+    answers = normalize_answers(raw_answers)
+    answered = compute_profile_answered(answers)
+
+    now_dt = now or datetime.now(tz=timezone.utc)
+    gen_dt = generated_at or now_dt
+
+    conf_time = compute_time_confidence(gen_dt, now_dt)
+    conf_u = compute_profile_confidence(conf_time, answered)
+
+    stage = _build_stage_one_hot(answers.mom_stage)
+    emotion_score = _map_emotion_score(answers.emotion)
+    context = _build_sparse_one_hot(answers.context)
+    need = _build_sparse_one_hot(answers.need)
+
+    rule_out = _compute_rule_output(stage, emotion_score)
+
+    return UserProfileV1_2_Extended(
+        profile_generated_at=gen_dt,
+        profile_confidence=conf_u,
+        profile_answered=answered,
+        stage=stage,
+        emotion_score=emotion_score,
+        context=context,  # type: ignore[arg-type]
+        need=need,  # type: ignore[arg-type]
+        rule_hits=rule_out.rule_hits,
+        hard_rules=rule_out.hard_rules,
+    )
+

server/app/features/user_profile_scoring/types.py

@@ -0,0 +1,89 @@
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any, Literal, Optional
+
+from pydantic import BaseModel, Field
+
+
+MomStageAnswer = Literal["expecting", "parenting", "unknown"]
+EmotionAnswer = Literal["low", "overwhelmed", "tired", "neutral", "calm", "joyful"]
+ContextAnswer = Literal["family", "work", "relationship", "friends", "health"]
+NeedAnswer = Literal[
+    "emotional_support",
+    "parenting_pressure",
+    "self_worth",
+    "anxiety_relief",
+    "rest_balance",
+]
+
+
+class QuestionnaireAnswersV1_2(BaseModel):
+    """
+    V1.2：每题可跳过
+
+    说明：
+    - `None` 表示题目被跳过/无值（与客户端的 `null` 对齐）
+    - 字段缺失（未传）也视为跳过
+    """
+
+    mom_stage: Optional[MomStageAnswer] = None
+    emotion: Optional[EmotionAnswer] = None
+    context: Optional[ContextAnswer] = None
+    need: Optional[NeedAnswer] = None
+
+
+class ProfileAnswered(BaseModel):
+    stage: bool
+    emotion: bool
+    context: bool
+    need: bool
+
+
+class UserStageOneHot(BaseModel):
+    expecting: Optional[Literal[0, 1]] = None
+    parenting: Optional[Literal[0, 1]] = None
+    unknown: Literal[0, 1]
+
+
+class ForbiddenContentPredicate(BaseModel):
+    """
+    用于表达“需要同时看用户与内容字段才能执行”的规则（跨维度规则）。
+    """
+
+    id: str
+    when_user: dict[str, Any] = Field(default_factory=dict)
+    forbid_content: dict[str, Any] = Field(default_factory=dict)
+
+
+class HardRules(BaseModel):
+    forbidden_risk_flags: list[str] = Field(default_factory=list)
+    forbidden_content_predicates: list[ForbiddenContentPredicate] = Field(default_factory=list)
+
+
+class UserProfileV1_2(BaseModel):
+    profile_version: Literal["v1.2"] = "v1.2"
+    profile_source: Literal["questionnaire"] = "questionnaire"
+    profile_generated_at: datetime
+    profile_confidence: float
+    profile_answered: ProfileAnswered
+    stage: UserStageOneHot
+    emotion_score: Optional[float] = None
+    context: dict[str, Literal[1]] = Field(default_factory=dict)
+    need: dict[str, Literal[1]] = Field(default_factory=dict)
+
+
+class UserProfileV1_2_Extended(UserProfileV1_2):
+    rule_hits: list[str] = Field(default_factory=list)
+    hard_rules: HardRules = Field(default_factory=HardRules)
+
+
+class BuildUserProfileRequest(BaseModel):
+    """
+    API 请求体：问卷答案 + 可选时间注入（便于回归测试/服务端批处理）
+    """
+
+    answers: QuestionnaireAnswersV1_2 = Field(default_factory=QuestionnaireAnswersV1_2)
+    generated_at: Optional[datetime] = None
+    now: Optional[datetime] = None
+