diff --git a/apps/api/routers/reports.py b/apps/api/routers/reports.py index 2313ff8..ce06af0 100644 --- a/apps/api/routers/reports.py +++ b/apps/api/routers/reports.py @@ -197,6 +197,23 @@ async def regenerate_report( return ok(_submit_payload(submission)) +@router.post( + "/{report_id}/full", + summary="生成深度诊断报告(基于该基础报告派生 tier=full,走 full_report 日配额)", +) +async def derive_full_report( + report_id: int, + payload: ReportRegenerateRequest, + user_id: CurrentUserId, + service: ReportServiceDep, +) -> dict[str, object]: + request_key = payload.request_key or f"report-full:{user_id}:{report_id}:{uuid.uuid4().hex}" + submission = await service.derive_full_report( + user_id=user_id, report_id=report_id, request_key=request_key + ) + return ok(_submit_payload(submission)) + + @router.post( "/{report_id}/export", summary="导出形象报告 PDF(走 hd_export 日配额,tier2 熔断可拦)" ) diff --git a/apps/worker/report_tasks.py b/apps/worker/report_tasks.py index ab68203..7a545e9 100644 --- a/apps/worker/report_tasks.py +++ b/apps/worker/report_tasks.py @@ -213,7 +213,12 @@ async def _generate_report(task_id: int) -> dict[str, Any]: gateway = build_llm_gateway(settings) try: - stages = assemble_stages(settings) + if report.tier == "full": + # 深度诊断:复用 basic 报告的 derived_features,无视觉阶段, + # 文案/校验换 full 专用阶段(产出 result.full 四大块)。 + stages = assemble_stages(settings, paths=settings.report_full_pipeline_stages) + else: + stages = assemble_stages(settings) ctx = ReportContext( report=report, report_input=report_input, diff --git a/migrations/versions/0003_align_quota_with_product_ssot.py b/migrations/versions/0003_align_quota_with_product_ssot.py index c00d613..71c30cc 100644 --- a/migrations/versions/0003_align_quota_with_product_ssot.py +++ b/migrations/versions/0003_align_quota_with_product_ssot.py @@ -88,6 +88,16 @@ _SSOT = [ "hard_cap": 2, "description": "报告重新生成(每 30 天;月硬上限 2 即「每日 ≤ 2」)", }, + { + "quota_key": "full_report", + "period_type": "day", + "base_limit": 1, + "task_bonus_max": 0, + "hard_cap": 2, + # 深度诊断(全面报告)随「免费功能」上线、无支付链路;数值为内测暂定, + # 正式口径待产品方案补档后回填(改数字只改产品方案,再回来改这一份)。 + "description": "形象深度诊断·全面报告(每日;产品方案待补,内测暂定)", + }, ] _LEGACY = [ diff --git a/packages/llm/routing.py b/packages/llm/routing.py index 67b83ea..a1023c3 100755 --- a/packages/llm/routing.py +++ b/packages/llm/routing.py @@ -62,6 +62,17 @@ _DEFAULTS: dict[Scene, dict[str, Any]] = { # 等黄金集能量化"思考带来的质量提升"再按场景打开(改配置即可)。 "thinking": False, }, + Scene.FULL_REPORT_COPY: { + "temperature": 0.6, + # 深度诊断四大块(总评+章节+季型解析+4 场景+30 天计划)一次产出, + # 输出预算按 REPORT_COPY 翻倍;异步任务,延迟宽松。 + "max_tokens": 6000, + "timeout_seconds": 180.0, + "prompt_id": "full_report_copy", + "prompt_version": "v1", + "critical": False, + "thinking": False, + }, } diff --git a/packages/llm/types.py b/packages/llm/types.py index b45b6a5..0fb24b3 100755 --- a/packages/llm/types.py +++ b/packages/llm/types.py @@ -24,6 +24,7 @@ class Scene(StrEnum): INTENT_CLASSIFY = "intent_classify" PREFERENCE_EXTRACT = "preference_extract" REPORT_COPY = "report_copy" + FULL_REPORT_COPY = "full_report_copy" class Message(BaseModel): diff --git a/packages/report/full_copy.py b/packages/report/full_copy.py new file mode 100644 index 0000000..21e6ce2 --- /dev/null +++ b/packages/report/full_copy.py @@ -0,0 +1,458 @@ +"""深度诊断(全面报告)文案的纯逻辑:十二季色板规则映射、结构化输出 schema 与降级模板。 + +与 basic 报告共用同一条「LLM 不可信任」纪律,分工如下: + +- **色板不出 LLM**:季型 → 色板的映射是本模块的预定义常量(规则引擎口径, + 与 kb_color 同源的确定性数据),LLM 只写季型解析文案。色值交给模型编 + 是幻觉高发区,校验器拦得住但每次都多花一次重试——结构上不让它碰更省。 +- **LLM 产出四块结构化内容**:总评 summary + 正文章节(与 basic 同构,保证 + ReportResultPage / PDF 导出的现有渲染零改动)、季型解析、4 场景穿搭、 + 30 天计划。 +- **降级模板**:持续违规时按规则直出拼全量内容(含场景与计划的兜底版本), + 对齐 validation_stage 的「任何出口要么 passed 要么抛错」。 +""" + +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from typing import Any + +from pydantic import BaseModel + +from packages.report.copy import CopySection, ReportCopy, dimension_label + +# ── 十二季色板规则映射(常量,色值不交给 LLM) ───────────────────────────── + +#: 十二季自研描述性命名(TD-18:对外不得使用「四季色彩 / CMB」商标名)。 +TWELVE_SEASONS: tuple[str, ...] = ( + "浅春", + "暖春", + "亮春", + "浅夏", + "柔夏", + "冷夏", + "柔秋", + "暖秋", + "深秋", + "亮冬", + "冷冬", + "深冬", +) + +#: 四季粗判(season_code)→ 十二季默认细分:视觉缺 season_detailed 时的兜底, +#: 取该季最中位的细分型,宁保守不冒进。 +SEASON_CODE_FALLBACK: Mapping[str, str] = { + "spring": "暖春", + "summer": "柔夏", + "autumn": "暖秋", + "winter": "冷冬", +} + +#: 中性色按冷暖倾向分两档,各季直接引用(白/黑/灰/米/驼/卡其口径,需求 §2.3 篇章二)。 +_NEUTRAL_WARM: tuple[tuple[str, str], ...] = ( + ("米白", "#F5EFE0"), + ("象牙白", "#F8F0DC"), + ("燕麦色", "#D9C9A8"), + ("浅驼", "#C8A97E"), + ("卡其", "#B89B6A"), + ("暖灰", "#A89F8D"), + ("深棕", "#5C4633"), +) + +_NEUTRAL_COOL: tuple[tuple[str, str], ...] = ( + ("纯白", "#FFFFFF"), + ("浅灰", "#D9D9D9"), + ("中灰", "#9AA0A6"), + ("炭灰", "#4A4A4A"), + ("藏蓝", "#1F2A4A"), + ("灰米", "#CFC9BD"), + ("蓝黑", "#16213A"), +) + +#: 季型 → (最佳色≥12, 避雷色≥6),中性色按季型冷暖引用上面两档。 +#: 色值为内测期规则引擎录入口径,后续统一迁至 kb_color 后此处仅作兜底。 +_SEASON_COLORS: Mapping[str, tuple[tuple[tuple[str, str], ...], tuple[tuple[str, str], ...]]] = { + "浅春": ( + ( + ("蜜桃粉", "#FFC8B4"), ("浅杏色", "#FFE0C7"), ("薄荷绿", "#B8E6C8"), + ("浅水蓝", "#AFDDE9"), ("鹅黄", "#FFE9A8"), ("浅珊瑚", "#FFB59E"), + ("嫩草绿", "#CFE8A9"), ("浅金棕", "#E8C79A"), ("樱花粉", "#FFD1DC"), + ("浅松石", "#A3E0D4"), ("淡奶油", "#FFF3D6"), ("浅橙", "#FFCF9E"), + ), + ( + ("纯黑", "#1A1A1A"), ("炭灰", "#3A3A3A"), ("藏青", "#14213D"), + ("酒红", "#6E1423"), ("深紫", "#3B1F47"), ("墨绿", "#1F3D2B"), + ), + ), + "暖春": ( + ( + ("珊瑚橙", "#FF8C69"), ("番茄红", "#E8553D"), ("金盏菊黄", "#FFB92E"), + ("橄榄绿", "#7A8B3D"), ("暖松石", "#2EB8A6"), ("焦糖", "#B9713B"), + ("杏色", "#F2B880"), ("苹果绿", "#8FC45A"), ("暖粉", "#F49AC1"), + ("蜂蜜金", "#D9A441"), ("铜色", "#B0653A"), ("暖米白", "#F7E7CE"), + ), + ( + ("冷灰蓝", "#708090"), ("粉紫", "#C8A2C8"), ("冰粉", "#F2D7E0"), + ("银灰", "#C0C0C0"), ("藏蓝", "#1B2A4A"), ("冷白", "#F4F8FB"), + ), + ), + "亮春": ( + ( + ("明黄", "#FFD93B"), ("亮橙", "#FF7A1A"), ("西瓜红", "#FF4D5E"), + ("翠绿", "#1FB573"), ("湖蓝", "#1FA8C9"), ("亮松石", "#14C4B2"), + ("亮粉", "#FF5E9C"), ("紫罗兰", "#7A4FD0"), ("柠檬黄", "#F5E642"), + ("亮苹果绿", "#7AC74F"), ("朱红", "#E63946"), ("宝蓝", "#2B5FCE"), + ), + ( + ("暗驼", "#A68A64"), ("灰绿", "#8A9A5B"), ("暗玫瑰", "#A76A73"), + ("深棕", "#4A3525"), ("灰蓝", "#6B7B8C"), ("暗紫", "#5A4A6B"), + ), + ), + "浅夏": ( + ( + ("雾粉", "#F3D5DC"), ("浅薰衣草", "#D8CBE8"), ("粉蓝", "#BFD7EA"), + ("浅玫瑰", "#E8B4C4"), ("珍珠白", "#F2F0EB"), ("浅灰绿", "#C9D8C5"), + ("淡紫", "#CDBFE0"), ("浅青", "#BFE0E0"), ("雾蓝", "#C3D2E0"), + ("浅桃粉", "#F6D2D8"), ("灰粉", "#E3C6CE"), ("浅蓝紫", "#BFC6E8"), + ), + ( + ("亮橙", "#FF7A1A"), ("明黄", "#FFD93B"), ("番茄红", "#E8553D"), + ("翠绿", "#1FB573"), ("焦糖", "#B9713B"), ("纯黑", "#1A1A1A"), + ), + ), + "柔夏": ( + ( + ("雾霾蓝", "#8FA8BF"), ("豆沙粉", "#C4909A"), ("灰玫瑰", "#B08E96"), + ("鼠尾草绿", "#9CAF88"), ("灰紫", "#9B8AA5"), ("燕麦", "#C7BBA8"), + ("灰松石", "#7FA8A0"), ("烟粉", "#CFA9B2"), ("蓝灰", "#7C93A8"), + ("藕荷", "#BFA4BC"), ("灰卡其", "#A8A18C"), ("深灰玫瑰", "#96757F"), + ), + ( + ("明黄", "#FFD93B"), ("亮橙", "#FF7A1A"), ("西瓜红", "#FF4D5E"), + ("翠绿", "#1FB573"), ("纯黑", "#1A1A1A"), ("亮粉", "#FF5E9C"), + ), + ), + "冷夏": ( + ( + ("玫瑰粉", "#D46A92"), ("蓝调正红", "#C23B4E"), ("覆盆子", "#A62651"), + ("天蓝", "#5B9BD5"), ("皇家蓝", "#3B5FC0"), ("冷薄荷", "#9FD8D8"), + ("薰衣草", "#A78BC9"), ("松绿", "#2E7D6B"), ("玉兰白", "#F6F4F0"), + ("冷粉", "#E08AAF"), ("钢蓝", "#4F7396"), ("深玫瑰", "#8E3A5B"), + ), + ( + ("珊瑚橙", "#FF8C69"), ("金盏菊黄", "#FFB92E"), ("焦糖", "#B9713B"), + ("橄榄绿", "#7A8B3D"), ("番茄红", "#E8553D"), ("暖驼", "#C19A6B"), + ), + ), + "柔秋": ( + ( + ("陶土", "#B4714F"), ("橄榄", "#8A8B4D"), ("芥末黄", "#C9A227"), + ("砖红", "#A65B43"), ("灰绿", "#8FA382"), ("驼棕", "#A98A63"), + ("暗松石", "#5F8B7E"), ("南瓜", "#C07840"), ("栗棕", "#7C5540"), + ("暗玫瑰", "#A97E7E"), ("麦色", "#CBB37A"), ("苔绿", "#75804F"), + ), + ( + ("冰粉", "#F2D7E0"), ("亮蓝", "#2B5FCE"), ("纯黑", "#1A1A1A"), + ("亮绿", "#35D07F"), ("冰紫", "#D8CBE8"), ("亮粉", "#FF5E9C"), + ), + ), + "暖秋": ( + ( + ("铁锈红", "#A6472E"), ("南瓜橙", "#C77B2E"), ("深橄榄", "#6B7038"), + ("芥末", "#B9932B"), ("深驼", "#8F6B45"), ("番茄棕", "#9C4A32"), + ("森林绿", "#3E5E3E"), ("蜂蜜", "#C99A3C"), ("赤陶", "#B2593D"), + ("暖棕", "#7A5236"), ("暗金", "#A8842F"), ("深松石", "#3F7A6B"), + ), + ( + ("冰蓝", "#BFE3F0"), ("冷粉", "#E08AAF"), ("银灰", "#C0C0C0"), + ("薰衣草", "#A78BC9"), ("冰白", "#F4F8FB"), ("紫罗兰", "#7A4FD0"), + ), + ), + "深秋": ( + ( + ("深棕", "#4A3525"), ("黑橄榄", "#4A4A26"), ("深铁锈", "#7E3B26"), + ("墨绿", "#274435"), ("深驼", "#6B4E30"), ("深橙棕", "#8A4A24"), + ("深酒红", "#5E2432"), ("深松绿", "#2E4A3B"), ("焦糖棕", "#7A4A28"), + ("深芥末", "#8A7220"), ("深铜", "#6E452A"), ("黑棕", "#33261C"), + ), + ( + ("冰粉", "#F2D7E0"), ("浅蓝", "#BFD7EA"), ("浅薰衣草", "#D8CBE8"), + ("珍珠白", "#F2F0EB"), ("浅灰绿", "#C9D8C5"), ("浅桃粉", "#F6D2D8"), + ), + ), + "亮冬": ( + ( + ("洋红", "#D01A7A"), ("宝蓝", "#2040C8"), ("翠绿", "#0E9F6E"), + ("正红", "#DE1A3A"), ("亮松石", "#0FA8B8"), ("柠檬黄", "#F2E13C"), + ("紫红", "#A8126B"), ("亮蓝", "#2B7DE0"), ("纯黑", "#1A1A1A"), + ("纯白", "#FFFFFF"), ("银灰", "#C0C0C0"), ("冰粉", "#F2A7C8"), + ), + ( + ("驼色", "#C19A6B"), ("芥末黄", "#C9A227"), ("橄榄绿", "#7A8B3D"), + ("陶土", "#B4714F"), ("南瓜橙", "#C77B2E"), ("暖棕", "#7A5236"), + ), + ), + "冷冬": ( + ( + ("纯黑", "#1A1A1A"), ("纯白", "#FFFFFF"), ("正红", "#C8102E"), + ("宝石蓝", "#1D4ED8"), ("深紫红", "#7A1245"), ("冰蓝", "#A8D8F0"), + ("冷粉", "#D86AA8"), ("翡翠绿", "#0E7A5F"), ("藏蓝", "#14264A"), + ("银灰", "#C0C0C0"), ("紫罗兰", "#6D28A8"), ("樱桃红", "#A8123B"), + ), + ( + ("珊瑚橙", "#FF8C69"), ("金盏菊黄", "#FFB92E"), ("驼色", "#C19A6B"), + ("焦糖", "#B9713B"), ("橄榄绿", "#7A8B3D"), ("暖杏", "#F2B880"), + ), + ), + "深冬": ( + ( + ("黑棕", "#2B201A"), ("深酒红", "#571C2B"), ("深宝石蓝", "#1A2E6E"), + ("深紫", "#3F1D5E"), ("墨绿", "#1E3A2E"), ("炭黑", "#141414"), + ("深樱桃", "#6E1230"), ("深松绿", "#1F4438"), ("深灰", "#2E2E2E"), + ("深钴蓝", "#16295C"), ("深紫红", "#4A0E33"), ("深橄榄黑", "#2E2E1A"), + ), + ( + ("浅杏色", "#FFE0C7"), ("鹅黄", "#FFE9A8"), ("蜜桃粉", "#FFC8B4"), + ("浅珊瑚", "#FFB59E"), ("薄荷绿", "#B8E6C8"), ("浅金棕", "#E8C79A"), + ), + ), +} + +#: 暖调季型集合(决定中性色引用哪一档)。 +_WARM_SEASONS: frozenset[str] = frozenset( + {"浅春", "暖春", "亮春", "柔秋", "暖秋", "深秋"} +) + + +def resolve_season(derived_features: Mapping[str, Any]) -> str | None: + """派生特征 → 十二季细分名;无细分值时按四季粗判取默认细分(见 SEASON_CODE_FALLBACK)。""" + detailed = derived_features.get("season_detailed") + if isinstance(detailed, str) and detailed in TWELVE_SEASONS: + return detailed + code = derived_features.get("season_code") + if isinstance(code, str): + return SEASON_CODE_FALLBACK.get(code) + return None + + +def palette_for(season_detailed: str) -> dict[str, list[dict[str, str]]]: + """季型 → 结构化色板(best/neutral/avoid,每色 name+hex)。未收录季型给空板。""" + if season_detailed not in _SEASON_COLORS: + return {"best": [], "neutral": [], "avoid": []} + best, avoid = _SEASON_COLORS[season_detailed] + neutral = _NEUTRAL_WARM if season_detailed in _WARM_SEASONS else _NEUTRAL_COOL + + def entries(pairs: Sequence[tuple[str, str]]) -> list[dict[str, str]]: + return [{"name": name, "hex": hex_value} for name, hex_value in pairs] + + return {"best": entries(best), "neutral": entries(neutral), "avoid": entries(avoid)} + + +# ── LLM 结构化输出 schema ────────────────────────────────────────────────── + +#: 场景固定四个(需求 §2.3 篇章六:通勤职场/休闲日常/约会聚会/正式场合)。 +SCENE_DEFINITIONS: tuple[tuple[str, str], ...] = ( + ("commute", "日常通勤"), + ("weekend", "休闲周末"), + ("formal", "正式场合"), + ("date", "约会聚会"), +) + + +class FullOutfit(BaseModel): + """一套搭配:名称 + 要点列表(上装/下装/鞋/配饰、配色、材质都在要点里)。""" + + name: str + points: list[str] + + +class FullScene(BaseModel): + """一个场景的穿搭方案包:2~3 套搭配。""" + + key: str + title: str + outfits: list[FullOutfit] + + +class PlanWeek(BaseModel): + """30 天计划的一周:3~5 条可执行任务。""" + + week: int + title: str + tasks: list[str] + + +class FullReportCopy(BaseModel): + """FULL_REPORT_COPY 场景的结构化输出。 + + summary/sections 与 basic 的 ReportCopy 同构——ReportResultPage 与 PDF + 导出按 result.copy 渲染的部分对两种 tier 零改动;深度诊断的增量内容全在 + 后三个字段,落库时进 result.full(色板由规则映射补齐,不经 LLM)。 + """ + + summary: str + sections: list[CopySection] + season_analysis: str + scenes: list[FullScene] + plan30: list[PlanWeek] + + def as_report_copy(self) -> ReportCopy: + """折成 ReportCopy 视图供生成后校验:全部自由文本都进 full_text 扫描范围。""" + sections = list(self.sections) + sections.append(CopySection(title="季型解析", paragraphs=[self.season_analysis])) + for scene in self.scenes: + paragraphs = [scene.title] + for outfit in scene.outfits: + paragraphs.append(outfit.name) + paragraphs.extend(outfit.points) + sections.append(CopySection(title=scene.title, paragraphs=paragraphs)) + for week in self.plan30: + sections.append(CopySection(title=week.title, paragraphs=list(week.tasks))) + return ReportCopy(summary=self.summary, sections=sections) + + +def build_full_result( + copy: FullReportCopy, + *, + season_detailed: str | None, + season_code: str | None, + palette: Mapping[str, list[dict[str, str]]] | None, +) -> dict[str, Any]: + """组装 result["full"]:LLM 内容 + 规则色板(App 按此结构渲染三个新区块)。""" + season: dict[str, Any] = { + "code": season_code, + "detailed": season_detailed, + "name_cn": f"{season_detailed}型" if season_detailed else None, + "analysis": copy.season_analysis, + } + return { + "season": season, + "palette": dict(palette) if palette else {"best": [], "neutral": [], "avoid": []}, + "scenes": [scene.model_dump() for scene in copy.scenes], + "plan30": [week.model_dump() for week in copy.plan30], + } + + +# ── 降级模板(持续违规时的零 LLM 出口) ──────────────────────────────────── + +_FALLBACK_FULL_SUMMARY = ( + "以下为深度诊断内容,根据本次分析的结构化结论整理," + "每条结论都来自规则引擎对诊断值的直接匹配。" +) + +#: 各场景的兜底搭配要点模板:{best}=最佳色名前三个、{neutral}=中性色名前两个。 +#: 措辞刻意避开「建议/推荐」句式(校验规则③要求建议句有规则命中支撑, +#: 降级模板没有命中链,干脆不用建议句式——受控拼接不做超出数据的断言)。 +_FALLBACK_SCENE_POINTS: Mapping[str, tuple[str, ...]] = { + "commute": ( + "上装选{best}之一的衬衫或针织衫,下装配{neutral}直筒裤", + "整体以{neutral}为基调,{best}之一用于内搭提亮", + "鞋包选{neutral},保持通勤场合的利落感", + ), + "weekend": ( + "上装选{best}之一的卫衣或休闲衬衫,下装配{neutral}休闲裤", + "可用{best}之一的单品做主色,其余用{neutral}压住", + "鞋选{neutral}休闲鞋,包袋同色系呼应", + ), + "formal": ( + "外套选{neutral}的西装或大衣,内搭{best}之一的衬衫", + "下装与外套同色系,形成纵向线条", + "配饰控制在{neutral}范围内,点缀可用{best}之一", + ), + "date": ( + "上装选{best}之一的针织衫或连衣裙,突出柔和气色", + "下装配{neutral}半身裙或长裤,鞋选同色系", + "配饰以小巧精致为先,颜色落在{best}范围内", + ), +} + +_FALLBACK_PLAN: tuple[tuple[str, tuple[str, ...]], ...] = ( + ("第 1 周:整理衣柜", ( + "把衣柜里的单品按适合色、中性色、避雷色分成三堆", + "挑 2 件中性色基础款作为本周固定搭配底版", + "记录本周每天穿的配色,对照色板找偏差", + )), + ("第 2 周:引入适合色", ( + "用适合色中的一件上衣替换惯常穿的避雷色上衣", + "尝试「中性色为主、适合色点缀」的配色方式 3 次", + "对照镜子确认提亮效果,记下表现出色的一件", + )), + ("第 3 周:场景演练", ( + "按报告里的通勤方案完整穿一次,记录感受", + "按报告里的休闲方案穿一次周末出行搭配", + "从配饰入手,为一套固定搭配加一件色板内的配饰", + )), + ("第 4 周:固化习惯", ( + "把前 3 周效果好的 3 套搭配拍照存档", + "补齐衣柜里缺位的中性色基础款清单", + "对照报告复盘整月变化,圈出下个月想尝试的新色", + )), +) + + +def _fallback_outfit_points(key: str, palette: Mapping[str, list[dict[str, str]]]) -> list[str]: + best = "、".join(c["name"] for c in palette.get("best", [])[:3]) or "适合色" + neutral = "、".join(c["name"] for c in palette.get("neutral", [])[:2]) or "中性色" + return [ + point.format(best=best, neutral=neutral) + for point in _FALLBACK_SCENE_POINTS.get(key, ()) + ] + + +def render_fallback_full_copy( + dimensions: Sequence[Any], + *, + season_detailed: str | None, + palette: Mapping[str, list[dict[str, str]]] | None, +) -> FullReportCopy: + """深度诊断降级模板:全量内容规则直出(含场景与 30 天计划),零 LLM。 + + 措辞纪律与 render_fallback_copy 相同——只做受控拼接、逐字复述诊断值, + 天然通过三重校验;FullValidationStage 仍会再校验一遍做 fail-close 自证。 + """ + ok_dims = [d for d in dimensions if isinstance(d, Mapping) and d.get("status") == "ok"] + board = dict(palette) if palette else {"best": [], "neutral": [], "avoid": []} + + conclusion_paragraphs = [] + for dim in ok_dims: + value = dim.get("value") or "" + if isinstance(value, (list, dict)): + continue # 复合值不复述进段落(色板等结构化值由 full.palette 呈现) + conclusion_paragraphs.append(f"你的{dimension_label(dim)}结论为「{value}」。") + sections: list[CopySection] = [] + if conclusion_paragraphs: + sections.append(CopySection(title="诊断结论", paragraphs=conclusion_paragraphs)) + else: + sections.append(CopySection(title="说明", paragraphs=["本次分析暂无可用结论。"])) + + if season_detailed: + season_analysis = ( + f"你的十二季色彩定位为「{season_detailed}型」。" + f"适合色板共 {len(board.get('best', []))} 色," + f"中性色 {len(board.get('neutral', []))} 色," + f"避雷色 {len(board.get('avoid', []))} 色,完整色板见下方色卡区。" + ) + else: + season_analysis = "本次分析未产出十二季定位,色板区暂无内容。" + + scenes = [ + FullScene( + key=key, + title=title, + outfits=[ + FullOutfit(name=f"{title} · 色板方案", points=_fallback_outfit_points(key, board)) + ], + ) + for key, title in SCENE_DEFINITIONS + ] + plan30 = [ + PlanWeek(week=index + 1, title=title, tasks=list(tasks)) + for index, (title, tasks) in enumerate(_FALLBACK_PLAN) + ] + return FullReportCopy( + summary=_FALLBACK_FULL_SUMMARY, + sections=sections, + season_analysis=season_analysis, + scenes=scenes, + plan30=plan30, + ) diff --git a/packages/report/full_copy_stage.py b/packages/report/full_copy_stage.py new file mode 100644 index 0000000..8ed88d3 --- /dev/null +++ b/packages/report/full_copy_stage.py @@ -0,0 +1,117 @@ +"""流水线阶段:深度诊断(tier=full)的 LLM 文案生成。 + +与 CopyStage 同位(name="copy"):result.copy 保持 summary/sections 同构, +App 与 PDF 导出的现有渲染零改动;深度诊断的增量内容落 result["full"] +(色板/场景/30 天计划,结构见 full_copy.build_full_result)。 + +色板纪律:季型 → 色板是 full_copy 里的规则映射常量,prompt 只给色名不给 +色值——LLM 复述色名写解析,hex 由规则侧组装,结构上杜绝色值幻觉。 +""" + +from __future__ import annotations + +import json +from typing import Any + +from packages.llm.types import Scene +from packages.report.copy import compose_prompt_material +from packages.report.full_copy import FullReportCopy, build_full_result, palette_for, resolve_season +from packages.report.pipeline import ReportContext +from packages.shared.errors import AppError, ErrorCode +from packages.shared.logging import get_logger + +logger = get_logger(__name__) + + +def _coerce_full_copy(parsed: Any) -> FullReportCopy: + """同 copy.coerce_parsed_copy 的契约闸门,换成 FullReportCopy 版本。""" + if not isinstance(parsed, FullReportCopy): + msg = "FULL_REPORT_COPY 场景的 completion 未携带 FullReportCopy 结构" + raise AppError(ErrorCode.LLM_STRUCTURED_OUTPUT_FAILED, message=msg) + return parsed + + +def _season_prompt_material( + derived_features: dict[str, Any], + season_detailed: str | None, + palette: dict[str, list[dict[str, str]]], +) -> str: + """{season_json} 载荷:季型结论 + 色名清单(只给色名,不给 hex——见模块 docstring)。 + + 顺带带上面部色彩三维度,LLM 写季型解析时可直接复述这些刻度值。 + """ + names = {role: [c["name"] for c in colors] for role, colors in palette.items()} + axes = { + key: derived_features.get(key) + for key in ("face_hue", "face_lightness", "face_chroma", "face_contrast", "skin_tone") + if derived_features.get(key) is not None + } + payload = { + "season_code": derived_features.get("season_code"), + "season_detailed": season_detailed, + "color_axes": axes, + "palette_names": names, + } + return json.dumps(payload, ensure_ascii=False, separators=(",", ":")) + + +class FullCopyStage: + """name="copy"(与 basic 同位):调 FULL_REPORT_COPY 场景产出深度诊断全量内容。 + + 产物双写:result["copy"](summary/sections,同构 basic)与 + result["full"](season/palette/scenes/plan30)。result["full"] 也复制 + 一份进本阶段键外的独立键——pipeline 按阶段键 setdefault 不冲突。 + """ + + name = "copy" + + async def run(self, ctx: ReportContext) -> None: + diagnosis = ctx.report.result.get("diagnosis") + if not isinstance(diagnosis, dict): + raise AppError( + ErrorCode.TASK_STATE_INVALID, + message="copy 阶段缺少 diagnosis 产物(流水线阶段顺序错误)", + ) + if ctx.gateway is None: + raise AppError(ErrorCode.CONFIG_ERROR, message="报告流水线缺少 LLM 网关") + + derived = dict(ctx.report_input.derived_features or {}) + season_detailed = resolve_season(derived) + palette = palette_for(season_detailed) if season_detailed else None + + material, dropped_keys = compose_prompt_material(diagnosis) + completion = await ctx.gateway.complete( + scene=Scene.FULL_REPORT_COPY, + prompt_vars={ + "dimensions_json": material, + "season_json": _season_prompt_material(derived, season_detailed, palette or {}), + }, + schema=FullReportCopy, + user_id=ctx.report.user_id, + ref_type="report", + ref_id=str(ctx.report.report_id), + ) + full_copy = _coerce_full_copy(completion.parsed) + + ctx.report.result["copy"] = { + "summary": full_copy.summary, + "sections": [section.model_dump() for section in full_copy.sections], + "model": completion.model, + "prompt_version": completion.prompt_version, + "dropped_keys": dropped_keys, + } + ctx.report.result["full"] = build_full_result( + full_copy, + season_detailed=season_detailed, + season_code=derived.get("season_code") + if isinstance(derived.get("season_code"), str) + else None, + palette=palette, + ) + logger.info( + "report_full_copy_generated", + report_id=ctx.report.report_id, + model=completion.model, + season=season_detailed, + dropped_dimensions=len(dropped_keys), + ) diff --git a/packages/report/full_validation_stage.py b/packages/report/full_validation_stage.py new file mode 100644 index 0000000..e894f22 --- /dev/null +++ b/packages/report/full_validation_stage.py @@ -0,0 +1,214 @@ +"""流水线阶段:深度诊断(tier=full)的生成后校验。 + +与 ValidationStage 同一套纪律(三重校验 → 语义重试一次 → 降级模板, +任何出口要么 passed 要么抛错),差别只在两处: + +- 校验对象用 ``FullReportCopy.as_report_copy()`` 折成的全文视图——季型解析、 + 场景穿搭、30 天计划的自由文本全部进入贬损词/一致性/绝对化三重扫描, + 不是只查 summary+sections 就放行。 +- 降级模板换 render_fallback_full_copy:场景与 30 天计划也有零 LLM 的 + 规则直出版本,降级不等于「深度诊断区块消失」(result.full 恒有结构)。 +""" + +from __future__ import annotations + +from typing import Any + +from pydantic import ValidationError + +from packages.llm.types import Completion, Message, Scene +from packages.report.copy import compose_prompt_material +from packages.report.full_copy import ( + FullReportCopy, + build_full_result, + palette_for, + render_fallback_full_copy, + resolve_season, +) +from packages.report.full_copy_stage import _season_prompt_material +from packages.report.pipeline import ReportContext +from packages.report.validation import validate_copy +from packages.shared.errors import AppError, ErrorCode +from packages.shared.logging import get_logger + +logger = get_logger(__name__) + +_MAX_SEMANTIC_RETRIES = 1 +_FALLBACK_MODEL = "fallback-template" + + +def _clean_checks() -> dict[str, int]: + return {"pejorative": 0, "consistency": 0, "absolute": 0} + + +def _full_copy_from_result(result: dict[str, Any]) -> FullReportCopy: + """从 result.copy + result.full 重建 FullReportCopy(校验与降级换写都走它)。""" + copy_payload = result.get("copy") + full_payload = result.get("full") + if not isinstance(copy_payload, dict) or not isinstance(full_payload, dict): + raise AppError( + ErrorCode.TASK_STATE_INVALID, + message="validation 阶段缺少 copy/full 产物(流水线阶段顺序错误)", + ) + season_raw = full_payload.get("season") + season: dict[str, Any] = dict(season_raw) if isinstance(season_raw, dict) else {} + try: + return FullReportCopy( + summary=str(copy_payload.get("summary") or ""), + sections=copy_payload.get("sections") or [], + season_analysis=str(season.get("analysis") or ""), + scenes=full_payload.get("scenes") or [], + plan30=full_payload.get("plan30") or [], + ) + except ValidationError as exc: + raise AppError( + ErrorCode.TASK_STATE_INVALID, + message=f"full 产物结构不合法:{exc.error_count()} 处校验错误", + ) from exc + + +def _write_full_result( + ctx: ReportContext, + copy: FullReportCopy, + *, + model: str, + prompt_version: str | None, +) -> None: + """把一份 FullReportCopy 落回 result.copy + result.full(色板规则映射重算)。""" + derived = dict(ctx.report_input.derived_features or {}) + season_detailed = resolve_season(derived) + palette = palette_for(season_detailed) if season_detailed else None + base_raw = ctx.report.result.get("copy") + base: dict[str, Any] = dict(base_raw) if isinstance(base_raw, dict) else {} + ctx.report.result["copy"] = { + **base, + "summary": copy.summary, + "sections": [section.model_dump() for section in copy.sections], + "model": model, + "prompt_version": prompt_version, + } + season_code_raw = derived.get("season_code") + ctx.report.result["full"] = build_full_result( + copy, + season_detailed=season_detailed, + season_code=season_code_raw if isinstance(season_code_raw, str) else None, + palette=palette, + ) + + +class FullValidationStage: + """name="validation":深度诊断文案的三重校验与降级(语义对齐 ValidationStage)。""" + + name = "validation" + + async def run(self, ctx: ReportContext) -> None: + result = ctx.report.result + diagnosis = result.get("diagnosis") + if not isinstance(diagnosis, dict): + raise AppError( + ErrorCode.TASK_STATE_INVALID, + message="validation 阶段缺少 diagnosis 产物(流水线阶段顺序错误)", + ) + full_copy = _full_copy_from_result(result) + + dimensions = [d for d in (diagnosis.get("dimensions") or []) if isinstance(d, dict)] + rule_hits = [h for h in (diagnosis.get("rule_hits") or []) if isinstance(h, dict)] + + violations = validate_copy(full_copy.as_report_copy(), dimensions, rule_hits=rule_hits) + if not violations: + result["validation"] = {"passed": True, "checks": _clean_checks()} + return + + logger.warning( + "report_full_copy_violations_found", + report_id=ctx.report.report_id, + violations=violations, + ) + retry = await self._retry_with_fixes(ctx, diagnosis, violations) + if retry is not None: + retry_copy, retry_completion = retry + retry_violations = validate_copy( + retry_copy.as_report_copy(), dimensions, rule_hits=rule_hits + ) + if not retry_violations: + _write_full_result( + ctx, + retry_copy, + model=retry_completion.model, + prompt_version=retry_completion.prompt_version, + ) + result["validation"] = { + "passed": True, + "retried": True, + "violations_first_pass": violations, + "checks": _clean_checks(), + } + return + logger.warning( + "report_full_copy_retry_still_violating", + report_id=ctx.report.report_id, + violations=retry_violations, + ) + + derived = dict(ctx.report_input.derived_features or {}) + season_detailed = resolve_season(derived) + palette = palette_for(season_detailed) if season_detailed else None + fallback = render_fallback_full_copy( + dimensions, season_detailed=season_detailed, palette=palette + ) + residual = validate_copy(fallback.as_report_copy(), dimensions, rule_hits=rule_hits) + if residual: + raise AppError( + ErrorCode.INTERNAL_ERROR, + message=f"降级模板文案未通过校验,拒绝放行:{residual}", + ) + _write_full_result(ctx, fallback, model=_FALLBACK_MODEL, prompt_version=None) + result["validation"] = { + "passed": True, + "degraded": True, + "violations_first_pass": violations, + "checks": _clean_checks(), + } + logger.warning( + "report_full_copy_degraded_to_template", + report_id=ctx.report.report_id, + first_pass_violations=len(violations), + ) + + async def _retry_with_fixes( + self, + ctx: ReportContext, + diagnosis: dict[str, Any], + violations: list[str], + ) -> tuple[FullReportCopy, Completion] | None: + """带修正指令重新生成一次(语义对齐 ValidationStage._retry_with_fixes)。""" + if ctx.gateway is None: + raise AppError(ErrorCode.CONFIG_ERROR, message="报告流水线缺少 LLM 网关") + material, _dropped = compose_prompt_material(diagnosis) + derived = dict(ctx.report_input.derived_features or {}) + season_detailed = resolve_season(derived) + season_material = _season_prompt_material( + derived, season_detailed, palette_for(season_detailed) if season_detailed else {} + ) + correction = ( + "上一版深度诊断文案未通过内容校验,违规清单如下:\n" + + "\n".join(f"- {item}" for item in violations) + + "\n请严格遵守系统提示词的内容红线,重新生成整份文案:只复述给定诊断值与" + "给定色板色名,删除全部违规表述,不要解释修改过程。" + ) + completion = await ctx.gateway.complete( + scene=Scene.FULL_REPORT_COPY, + messages=[Message(role="user", content=correction)], + prompt_vars={"dimensions_json": material, "season_json": season_material}, + schema=FullReportCopy, + user_id=ctx.report.user_id, + ref_type="report", + ref_id=str(ctx.report.report_id), + ) + parsed = completion.parsed + if not isinstance(parsed, FullReportCopy): + raise AppError( + ErrorCode.LLM_STRUCTURED_OUTPUT_FAILED, + message="FULL_REPORT_COPY 场景的 completion 未携带 FullReportCopy 结构", + ) + return parsed, completion diff --git a/packages/report/pipeline.py b/packages/report/pipeline.py index 4443147..b72c9b6 100644 --- a/packages/report/pipeline.py +++ b/packages/report/pipeline.py @@ -15,6 +15,7 @@ path 动态装配。 from __future__ import annotations import importlib +from collections.abc import Sequence from typing import Any, Protocol, runtime_checkable from sqlalchemy.ext.asyncio import AsyncSession @@ -102,8 +103,12 @@ class StageFailureError(Exception): super().__init__(f"stage={stage_name} error={type(original).__name__}: {original}") -def assemble_stages(settings: Settings) -> list[Stage]: - """按 ``settings.report_pipeline_stages`` 的 dotted path 动态装配阶段列表。 +def assemble_stages(settings: Settings, *, paths: Sequence[str] | None = None) -> list[Stage]: + """按阶段配置(dotted path 清单)动态装配阶段列表。 + + 默认读 ``settings.report_pipeline_stages``(basic 报告);tier=full 的 + 深度诊断由调用方显式传 ``settings.report_full_pipeline_stages``——两条 + 流水线共用装配与失败语义,只是阶段清单不同。 动态 import 是刻意的插件式装配(见模块 docstring):T16/T17 各自建阶段 模块,不碰本文件;import-linter 看不到这条边,阶段模块自身的依赖仍受 @@ -112,7 +117,8 @@ def assemble_stages(settings: Settings) -> list[Stage]: 模块/类不存在、实例化失败、或不符合 Stage 协议时,收集**全部**缺失项 抛 ``CONFIG_ERROR`` —— 绝不静默跳过:静默等于流水线空跑还扣用户配额。 """ - if not settings.report_pipeline_stages: + stage_paths = list(paths) if paths is not None else list(settings.report_pipeline_stages) + if not stage_paths: raise AppError( ErrorCode.CONFIG_ERROR, message="报告流水线阶段配置为空", @@ -121,7 +127,7 @@ def assemble_stages(settings: Settings) -> list[Stage]: missing: list[str] = [] stages: list[Stage] = [] - for dotted in settings.report_pipeline_stages: + for dotted in stage_paths: module_path, sep, attr = dotted.partition(":") stage: Any = None if sep and module_path and attr: diff --git a/packages/report/repo.py b/packages/report/repo.py index cda8612..6bd1585 100644 --- a/packages/report/repo.py +++ b/packages/report/repo.py @@ -11,30 +11,12 @@ from sqlalchemy.ext.asyncio import AsyncSession from packages.shared.models import RpFeedback, RpInput, RpReport, RpReservation STATUS_SUCCEEDED = "succeeded" -TIER_BASIC = "basic" async def get_report(session: AsyncSession, report_id: int) -> RpReport | None: return await session.get(RpReport, report_id) -async def find_succeeded_basic(session: AsyncSession, *, user_id: int) -> int | None: - """首次终身资格:该用户是否已有 succeeded 的 basic 报告(需求 §2.1)。 - - 只取 report_id 不取整行——资格判定不需要报告内容。 - """ - result = await session.execute( - select(RpReport.report_id) - .where( - RpReport.user_id == user_id, - RpReport.tier == TIER_BASIC, - RpReport.status == STATUS_SUCCEEDED, - ) - .limit(1) - ) - return result.scalar_one_or_none() - - async def list_reports(session: AsyncSession, *, user_id: int, limit: int = 20) -> list[RpReport]: result = await session.execute( select(RpReport) diff --git a/packages/report/service.py b/packages/report/service.py index 088b299..c9a0df9 100644 --- a/packages/report/service.py +++ b/packages/report/service.py @@ -1,10 +1,14 @@ -"""报告域服务:触发资格、配额预扣、版本化与幂等(T-P3-07 / T-P3-09)。 +"""报告域服务:配额预扣、版本化与幂等(T-P3-07 / T-P3-09)。 事务纪律:reserve 与建报告行、建 tk_task 在**同一事务**;commit 之后才投递 队列——队列消息不能早于数据落库,否则 worker 可能读到一个不存在的报告。 幂等是双保险:tk_task 的幂等键(= request_key)挡重复任务行,reserve 的 -幂等键挡重复预扣;服务层再加一次「重放优先」快捷路径,让首单成功后的 -同键重试拿到 200 而不是 40920。 +幂等键挡重复预扣;服务层再加一次「重放优先」快捷路径,让成功后的同键 +重试拿到 200 而不是重新创建。 + +已生成过报告的用户允许再次生成(产品决策变更):不再按「已有 succeeded +报告」硬拒(原 40920 终身锁死),只由 report_basic 终身配额兜底——额度 +耗尽时报 40301 而非 40920。 """ from __future__ import annotations @@ -31,6 +35,10 @@ logger = get_logger(__name__) #: 首次生成走终身配额(迁移 0003:lifetime / base 1),重新生成走月配额。 QUOTA_KEY_BASIC = "report_basic" QUOTA_KEY_REGEN = "report_regen" +#: 深度诊断(全面报告)走日配额 full_report(迁移 0003 SSOT)。 +QUOTA_KEY_FULL = "full_report" +TIER_BASIC = "basic" +TIER_FULL = "full" REPORT_REF_TYPE = "report" REPORT_ETA_SECONDS = 60 @@ -91,16 +99,14 @@ class ReportService: photo_object_key: str, request_key: str, ) -> ReportSubmission: - # 幂等重放最先判:首单成功后用同一 request_key 重试不是「再生成一次」, - # 必须先于终身资格校验按原任务返回(否则会把幂等重试误判成 40920)。 + # 幂等重放最先判:成功后用同一 request_key 重试不是「再生成一次」, + # 必须先于任何校验按原任务返回(否则会把幂等重试误判成新请求)。 replay = await self._find_replay(request_key=request_key, user_id=user_id) if replay is not None: return replay - # 首次资格(终身 1 次):已有 succeeded 的 basic 报告 → 40920。 - # 文案与 hint(「每 30 天可重新生成一次」)见 ErrorCode 定义,不含价格字样。 - if await repo.find_succeeded_basic(self._session, user_id=user_id) is not None: - raise AppError(ErrorCode.REPORT_ALREADY_EXISTS) + # 不再按「已有 succeeded 报告」终身锁死(产品决策变更):重复生成由 + # report_basic 终身配额兜底,额度耗尽由 BillingService 报 40301。 # 照片存在性校验(本地磁盘降级路线):key 必须落在 # upload_dir/report_photo/<本人>/ 下且文件真实存在——模拟 key 与越界 @@ -191,6 +197,62 @@ class ReportService: ) return submission + # -- 深度诊断(全面报告) ------------------------------------------------ + + async def derive_full_report( + self, *, user_id: int, report_id: int, request_key: str + ) -> ReportSubmission: + """基于一份 SUCCEEDED 的 basic 报告派生 tier=full 的深度诊断报告。 + + 与 regenerate 同一派生模式(拷贝 rp_input 三件套,视觉阶段因 + derived_features 非空而天然缺席——full 流水线本就不含视觉阶段); + 差别在 tier 与配额键。full 报告是独立谱系,version 从 1 起算。 + """ + replay = await self._find_replay(request_key=request_key, user_id=user_id) + if replay is not None: + return replay + + source = await self.get_report(user_id=user_id, report_id=report_id) + if source.tier != TIER_BASIC: + # full→full 套娃直接拒绝:深度诊断只能以基础报告为源。 + raise AppError(ErrorCode.STATE_CONFLICT, message="深度诊断报告请以基础报告为源生成") + if source.status != repo.STATUS_SUCCEEDED: + raise AppError(ErrorCode.STATE_CONFLICT, message="只有已完成的基础报告可以生成深度诊断") + source_input = await repo.get_input(self._session, report_id) + if source_input is None: + raise AppError(ErrorCode.STATE_CONFLICT, message="报告输入缺失,无法生成深度诊断") + + report = RpReport( + report_id=next_id(), + user_id=user_id, + tier=TIER_FULL, + status="pending", + version=1, + ) + report_input = RpInput( + input_id=next_id(), + report_id=report.report_id, + questionnaire=source_input.questionnaire, + photo_meta=source_input.photo_meta, + derived_features=source_input.derived_features, + ) + submission = await self._submit_and_deliver( + user_id=user_id, + report=report, + report_input=report_input, + quota_key=QUOTA_KEY_FULL, + request_key=request_key, + reason=f"full report from basic {report_id}", + ) + logger.info( + "full_report_derived", + user_id=user_id, + source_report_id=report_id, + report_id=submission.report_id, + task_id=submission.task_id, + ) + return submission + # -- 查询 --------------------------------------------------------------- async def get_report(self, *, user_id: int, report_id: int) -> RpReport: diff --git a/packages/report/validation.py b/packages/report/validation.py index 681648c..77f4c23 100644 --- a/packages/report/validation.py +++ b/packages/report/validation.py @@ -193,14 +193,15 @@ def check_absolute_claims(text: str) -> list[str]: return [term for term in ABSOLUTE_CLAIM_TERMS if term in text] -def _find_dimension( +def _find_dimensions( dimensions: Sequence[Mapping[str, Any]], key_tokens: Sequence[str] -) -> Mapping[str, Any] | None: - for dim in dimensions: - key = str(dim.get("key") or "") - if any(token in key for token in key_tokens): - return dim - return None +) -> list[Mapping[str, Any]]: + """按 key 识别串收齐全部匹配维度(季型有 season_code 与 season_detailed 两个)。""" + return [ + dim + for dim in dimensions + if any(token in str(dim.get("key") or "") for token in key_tokens) + ] def _check_enum_consistency( @@ -211,34 +212,48 @@ def _check_enum_consistency( 维度 skipped/无值时,文案出现该类任何枚举词都算编造——LLM 没看到 这个维度却写出了结论,正是素材组装要防的幻觉从校验侧兜底。 每类只报第一条命中,避免同一问题刷屏冲淡修正指令。 + + 两类放行:诊断值的「X型」写法(如「柔夏」→「柔夏型」,深度诊断的 + 十二季文案必经此路径);词表 token 是某个放行词的子串时不再单独追责 + (「夏型」⊂「柔夏型」——十二季细分命名无法全量枚举,见 TD-18)。 """ issues: list[str] = [] for category in _ENUM_CATEGORIES: - dim = _find_dimension(dimensions, category.key_tokens) - if dim is None: + matched = _find_dimensions(dimensions, category.key_tokens) + if not matched: continue - value = dim.get("value") - if dim.get("status") == "ok" and value: - canonical = str(value) - allowed = {_normalize_for_match(canonical)} - for alias in category.aliases.get(canonical, ()): # 值本身可能就是别名键 - allowed.add(_normalize_for_match(alias)) - # 诊断值也可能是别名列里的写法(如存了「沙漏型」),反查一次 - for canonical_key, alias_values in category.aliases.items(): - if _normalize_for_match(canonical) == _normalize_for_match(canonical_key) or any( - _normalize_for_match(canonical) == _normalize_for_match(a) for a in alias_values - ): - allowed.add(_normalize_for_match(canonical_key)) - allowed.update(_normalize_for_match(a) for a in alias_values) - else: - allowed = set() + allowed: set[str] = set() + for dim in matched: + value = dim.get("value") + if dim.get("status") == "ok" and isinstance(value, str) and value: + canonical = value + candidates = {canonical, f"{canonical}型"} + candidates.update(category.aliases.get(canonical, ())) # 值本身可能就是别名键 + # 诊断值也可能是别名列里的写法(如存了「沙漏型」),反查一次 + for canonical_key, alias_values in category.aliases.items(): + if _normalize_for_match(canonical) == _normalize_for_match( + canonical_key + ) or any( + _normalize_for_match(canonical) == _normalize_for_match(a) + for a in alias_values + ): + candidates.add(canonical_key) + candidates.update(alias_values) + allowed.update(_normalize_for_match(c) for c in candidates) for token in category.vocabulary: normalized_token = _normalize_for_match(token) - if normalized_token in copy_text_normalized and normalized_token not in allowed: - issues.append( - f"{category.label}用词「{token}」与诊断值「{value or '未诊断'}」不一致" - ) - break + if normalized_token not in copy_text_normalized: + continue + if normalized_token in allowed: + continue + if any(normalized_token in entry for entry in allowed): + continue + observed = next( + (str(d.get("value")) for d in matched if d.get("status") == "ok"), + "未诊断", + ) + issues.append(f"{category.label}用词「{token}」与诊断值「{observed}」不一致") + break return issues diff --git a/prompts/full_report_copy/v1.txt b/prompts/full_report_copy/v1.txt new file mode 100644 index 0000000..1cfae67 --- /dev/null +++ b/prompts/full_report_copy/v1.txt @@ -0,0 +1,35 @@ +[system] +你是「美搭」的资深形象顾问,为用户的深度诊断报告撰写文案。输入是结构化诊断结论、规则命中结果与色彩定位素材,输出是给用户阅读的深度报告正文。 + +内容红线(违反任何一条,整篇文案即作废重写): +- 结论只能复述输入中给出的诊断值与色板色名,不得新增、改写或推断任何维度结论;输入未提供的维度一律不提。 +- 引用颜色只用输入色板中出现的色名,不得自造色名,不得写出任何十六进制色值。 +- 描述而非评判:客观陈述特征与搭配路径,不评判外貌。 +- 不出现任何贬损性词汇,例如胖、丑、矮、土气、显老、缺陷一类表述。 +- 不使用绝对化用语,例如最好、第一、彻底、百分百。 +- 不做医疗、减重、整形、命理相关的结论或暗示。 +- 不出现价格、费用、购买、会员、升级、付费相关的任何表述。 +- 置信度低的维度使用不确定表述,不得断言。 + +输出结构(JSON,字段一个不能少): +- summary:整份深度诊断的总评,3~5 句,比初级报告更完整,串起风格、色彩与体型三条主线。 +- sections:正文章节数组,每章 title + paragraphs;按「个人风格核心诊断」「身材分析与穿搭指南」两条主线展开,每章 2~4 段。 +- season_analysis:季型解析文案 2~4 句,围绕输入给出的十二季定位,说明明度、饱和度、冷暖倾向的特征,并解释适合色与避雷色的取舍逻辑。 +- scenes:恰好 4 个场景,key 依次为 commute(日常通勤)、weekend(休闲周末)、formal(正式场合)、date(约会聚会);每场景 2~3 套搭配(outfits),每套有 name 与 points(3~5 条要点,覆盖上装/下装/鞋/配饰与配色逻辑),搭配用色从输入色板色名中选取。 +- plan30:恰好 4 周(week 1~4),每周 title + tasks(3~5 条可执行任务),节奏按「第 1 周替换基础单品、第 2 周引入配饰、第 3 周尝试色彩进阶、第 4 周固化习惯」推进。 + +写作要求: +- 简体中文,平实具体,避免营销腔。 +- 每个结论后面跟一句「为什么」,依据来自输入的规则命中说明。 +- 专业术语首次出现时补白话解释。 + +[user] +以下是结构化诊断素材(仅含已成功诊断的维度与规则命中结果): + +{dimensions_json} + +以下是色彩定位素材(season_detailed 是十二季定位结论,palette_names 是各色板的可用色名): + +{season_json} + +请按系统提示词规定的结构生成深度诊断报告文案。 diff --git a/tests/integration/test_full_report_flow.py b/tests/integration/test_full_report_flow.py new file mode 100644 index 0000000..cc99c0f --- /dev/null +++ b/tests/integration/test_full_report_flow.py @@ -0,0 +1,356 @@ +"""深度诊断(tier=full)链路端到端:派生 → full_report 日配额 → full 流水线 → 拒绝路径。 + +与 test_report_flow 同一套基建:Fake 阶段经 ``REPORT_PIPELINE_STAGES_JSON`` / +``REPORT_FULL_PIPELINE_STAGES_JSON`` 动态装配(worker 按 report.tier 分流两 +条阶段清单,两份 env 要同时给);不启真 celery,worker 内芯直调。 +""" + +from __future__ import annotations + +import json +import os +import uuid +from collections.abc import AsyncIterator, Iterator +from contextlib import contextmanager +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", + ), +] + +_MODULE = "tests.integration.test_full_report_flow" +_BASIC_STAGES = json.dumps([f"{_MODULE}:FakeVisionStage", f"{_MODULE}:FakeDiagnosisStage"]) +_FULL_STAGES = json.dumps([f"{_MODULE}:FakeFullDiagnosisStage", f"{_MODULE}:FakeFullCopyStage"]) + + +class FakeVisionStage: + """与 test_report_flow 同一契约:derived_features 非空即跳过。""" + + name = "vision" + + async def run(self, ctx: Any) -> None: + features = ctx.report_input.derived_features + if not features: + features["vision_runs"] = 1 + ctx.report.result["vision"] = {"features": sorted(features)} + + +class FakeDiagnosisStage: + name = "diagnosis" + + async def run(self, ctx: Any) -> None: + assert "vision" in ctx.report.result + ctx.report.result["diagnosis"] = {"dimensions": [], "coverage": {"total": 32}} + + +class FakeFullDiagnosisStage: + """full 流水线的诊断阶段替身:前面没有 vision 阶段(full 不含视觉)。""" + + name = "diagnosis" + + async def run(self, ctx: Any) -> None: + assert "vision" not in ctx.report.result + ctx.report.result["diagnosis"] = {"dimensions": [], "coverage": {"total": 32}} + + +class FakeFullCopyStage: + """full 文案阶段替身:copy 与 full 双写(结构与真实 FullCopyStage 一致)。""" + + name = "copy" + + async def run(self, ctx: Any) -> None: + assert "diagnosis" in ctx.report.result + ctx.report.result["copy"] = { + "summary": "深度总评", + "sections": [{"title": "风格", "paragraphs": ["段落"]}], + "model": "fake", + "prompt_version": "v1", + "dropped_keys": [], + } + ctx.report.result["full"] = { + "season": { + "code": "summer", + "detailed": "柔夏", + "name_cn": "柔夏型", + "analysis": "解析文案", + }, + "palette": { + "best": [{"name": "雾霾蓝", "hex": "#8FA8BF"}], + "neutral": [{"name": "浅灰", "hex": "#D9D9D9"}], + "avoid": [{"name": "明黄", "hex": "#FFD93B"}], + }, + "scenes": [ + { + "key": "commute", + "title": "日常通勤", + "outfits": [{"name": "通勤 A", "points": ["要点一"]}], + } + ], + "plan30": [{"week": 1, "title": "第 1 周", "tasks": ["任务一"]}], + } + + +# --- fixtures --------------------------------------------------------------- + + +@pytest.fixture +def no_enqueue(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr("apps.api.routers.reports._enqueue_report", lambda task_id: None) + + +@contextmanager +def _stage_config(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]: + from packages.shared.config import get_settings + + monkeypatch.setenv("REPORT_PIPELINE_STAGES_JSON", _BASIC_STAGES) + monkeypatch.setenv("REPORT_FULL_PIPELINE_STAGES_JSON", _FULL_STAGES) + get_settings.cache_clear() + try: + yield + finally: + monkeypatch.undo() + get_settings.cache_clear() + + +@pytest.fixture +def fake_stages(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]: + with _stage_config(monkeypatch): + yield + + +@pytest.fixture +async def client(no_enqueue: None) -> AsyncIterator[Any]: + import httpx + + from apps.api.main import create_app + from packages.shared.config import get_settings + + app = create_app(get_settings()) + transport = httpx.ASGITransport(app=app) + async with ( + httpx.AsyncClient(transport=transport, base_url="http://test") as http, + app.router.lifespan_context(app), + ): + yield http + + +# --- helpers ---------------------------------------------------------------- + + +async def _login(client: Any) -> tuple[dict[str, str], int]: + resp = await client.post( + "/api/v1/auth/dev-token", json={"device_key": f"full-it-{uuid.uuid4().hex[:8]}"} + ) + assert resp.json()["code"] == 0 + data = resp.json()["data"] + return {"Authorization": f"Bearer {data['access_token']}"}, int(data["user_id"]) + + +async def _upload_photo(client: Any, headers: dict[str, str]) -> str: + resp = await client.post( + "/api/v1/uploads/direct?scene=report_photo", + headers={**headers, "Content-Type": "image/jpeg"}, + content=b"\xff\xd8\xff\xe0" + b"\x00" * 64, + ) + assert resp.status_code == 200 + return str(resp.json()["data"]["object_key"]) + + +async def _create_succeeded_basic(client: Any, headers: dict[str, str]) -> dict[str, Any]: + from apps.worker.report_tasks import _generate_report + + resp = await client.post( + "/api/v1/reports", + headers=headers, + json={ + "questionnaire": {"style": "通勤", "height_cm": 168}, + "photo_object_key": await _upload_photo(client, headers), + "request_key": f"rk-{uuid.uuid4().hex}", + }, + ) + assert resp.status_code == 200 + body = resp.json()["data"] + await _generate_report(int(body["task_id"])) + return body + + +async def _derive_full( + client: Any, headers: dict[str, str], report_id: str, *, request_key: str | None = None +) -> Any: + return await client.post( + f"/api/v1/reports/{report_id}/full", + headers=headers, + json={"request_key": request_key or f"rk-full-{uuid.uuid4().hex}"}, + ) + + +async def _fetch_report(report_id: int) -> Any: + from sqlalchemy import select + + from packages.shared.db import session_scope + from packages.shared.models import RpReport + + async with session_scope() as session: + result = await session.execute(select(RpReport).where(RpReport.report_id == report_id)) + return result.scalar_one_or_none() + + +async def _fetch_input(report_id: int) -> Any: + from sqlalchemy import select + + from packages.shared.db import session_scope + from packages.shared.models import RpInput + + async with session_scope() as session: + result = await session.execute(select(RpInput).where(RpInput.report_id == report_id)) + return result.scalar_one_or_none() + + +async def _ledger_rows(user_id: int, quota_key: str) -> list[Any]: + from sqlalchemy import select + + from packages.shared.db import session_scope + from packages.shared.models import Ledger + + async with session_scope() as session: + result = await session.execute( + select(Ledger) + .where(Ledger.user_id == user_id, Ledger.quota_key == quota_key) + .order_by(Ledger.created_at.asc(), Ledger.tx_id.asc()) + ) + return list(result.scalars().all()) + + +# --- 场景 1:派生成功 + full 流水线 + 幂等重放 ---------------------------------- + + +async def test_derive_full_success_runs_full_pipeline(fake_stages: None, client: Any) -> None: + from apps.worker.report_tasks import _generate_report + + headers, user_id = await _login(client) + basic = await _create_succeeded_basic(client, headers) + + request_key = f"rk-full-{uuid.uuid4().hex}" + resp = await _derive_full(client, headers, basic["report_id"], request_key=request_key) + assert resp.status_code == 200 + data = resp.json()["data"] + full_report_id = int(data["report_id"]) + assert full_report_id != int(basic["report_id"]) + assert data["version"] == 1 # full 是独立谱系,版本从 1 起算 + + # full_report 日配额预扣(period_key d:YYYY-MM-DD)。 + rows = await _ledger_rows(user_id, "full_report") + assert len(rows) == 1 + assert rows[0].status == "reserved" + assert rows[0].period_key.startswith("d:") + + # 输入三件套复用(derived_features 带着 basic 的视觉中间态,无视觉阶段)。 + full_input = await _fetch_input(full_report_id) + assert full_input.derived_features == {"vision_runs": 1} + + await _generate_report(int(data["task_id"])) + report = await _fetch_report(full_report_id) + assert report.status == "succeeded" + assert report.tier == "full" + # full 流水线:无 vision 键,copy + full + diagnosis 齐全。 + assert set(report.result) == {"diagnosis", "copy", "full"} + full = report.result["full"] + assert full["season"]["name_cn"] == "柔夏型" + assert full["palette"]["best"][0]["hex"] == "#8FA8BF" + assert full["scenes"][0]["outfits"][0]["points"] + assert full["plan30"][0]["week"] == 1 + + rows = await _ledger_rows(user_id, "full_report") + assert [row.status for row in rows] == ["committed"] + + # 详情视图透出 tier 与 full 结构(App 凭 tier/full 渲染深度区块)。 + detail = (await client.get(f"/api/v1/reports/{full_report_id}", headers=headers)).json()["data"] + assert detail["tier"] == "full" + assert detail["result"]["full"]["season"]["detailed"] == "柔夏" + + # 幂等重放:同 request_key 返回同任务同报告,不二次预扣。 + replay = await _derive_full(client, headers, basic["report_id"], request_key=request_key) + assert replay.status_code == 200 + assert replay.json()["data"]["report_id"] == str(full_report_id) + assert len(await _ledger_rows(user_id, "full_report")) == 1 + + +# --- 场景 2:非 SUCCEEDED 源报告拒绝 -------------------------------------------- + + +async def test_derive_full_rejects_pending_source(fake_stages: None, client: Any) -> None: + headers, _user_id = await _login(client) + resp = await client.post( + "/api/v1/reports", + headers=headers, + json={ + "questionnaire": {"style": "通勤"}, + "photo_object_key": await _upload_photo(client, headers), + "request_key": f"rk-{uuid.uuid4().hex}", + }, + ) + assert resp.status_code == 200 + basic = resp.json()["data"] # pending,不跑 worker + + derived = await _derive_full(client, headers, basic["report_id"]) + assert derived.status_code == 409 + assert derived.json()["code"] == 40900 # STATE_CONFLICT + + +# --- 场景 3:full 不能再派生 full(防套娃) -------------------------------------- + + +async def test_derive_full_rejects_full_source(fake_stages: None, client: Any) -> None: + from apps.worker.report_tasks import _generate_report + + headers, _user_id = await _login(client) + basic = await _create_succeeded_basic(client, headers) + derived = await _derive_full(client, headers, basic["report_id"]) + assert derived.status_code == 200 + full = derived.json()["data"] + await _generate_report(int(full["task_id"])) + + nested = await _derive_full(client, headers, full["report_id"]) + assert nested.status_code == 409 + assert nested.json()["code"] == 40900 + + +# --- 场景 4:full_report 配额耗尽 ------------------------------------------------ + + +async def test_derive_full_blocked_when_quota_exhausted(fake_stages: None, client: Any) -> None: + """开发库 full_report 规则行是测试期放宽值(base_limit=10);把本用例用户的 + bl_quota 快照 limit_value 收敛到已用量,确定性断言 40301(同 test_report_flow + 场景 4 的做法,只动本用户的快照行,不动全局规则)。""" + from sqlalchemy import update + + from packages.shared.db import session_scope + from packages.shared.models import Quota + + headers, user_id = await _login(client) + basic = await _create_succeeded_basic(client, headers) + + first = await _derive_full(client, headers, basic["report_id"]) + assert first.status_code == 200 # 首次派生占掉 1 次 + + async with session_scope() as session: + await session.execute( + update(Quota) + .where(Quota.user_id == user_id, Quota.quota_key == "full_report") + .values(limit_value=1) + ) + + second = await _derive_full(client, headers, basic["report_id"]) + assert second.status_code == 403 + assert second.json()["code"] == 40301 # QUOTA_EXHAUSTED diff --git a/tests/integration/test_report_flow.py b/tests/integration/test_report_flow.py index f2c52a7..ee60e04 100644 --- a/tests/integration/test_report_flow.py +++ b/tests/integration/test_report_flow.py @@ -331,29 +331,50 @@ async def _count_user_reports(user_id: int) -> int: return int(result.scalar_one()) -# --- 场景 4:首次资格 40920 --------------------------------------------------- +# --- 场景 4:重复生成不再 40920,由 report_basic 终身配额兜底 --------------------- -async def test_second_first_report_is_rejected_with_40920(fake_stages: None, client: Any) -> None: +async def test_second_report_create_is_blocked_by_lifetime_quota( + fake_stages: None, client: Any +) -> None: + """产品决策变更:已生成过报告仍可再次提交,不再抛 40920(终身资格硬校验已删)。 + + 重复生成由 report_basic 终身配额兜底(迁移 0003 SSOT:lifetime base_limit=1)。 + 本地开发库的规则行被人为放宽过(base_limit=10),这里把本用例用户的 + bl_quota 快照 limit_value 收敛回 SSOT 值 1,确定性断言额度耗尽: + 新 request_key 的第二次创建应返回 403 / QUOTA_EXHAUSTED 40301, + 而不是旧的 40920 REPORT_ALREADY_EXISTS。 + """ + from sqlalchemy import update + from apps.worker.report_tasks import _generate_report + from packages.shared.db import session_scope + from packages.shared.models import Quota - headers, _ = await _login(client) + headers, user_id = await _login(client) data = await _create_report(client, headers) - await _generate_report(int(data["task_id"])) # 首单成功,终身资格已用掉 + await _generate_report(int(data["task_id"])) # 首单成功,终身额度已用 1 次 + + # 收敛该用户的终身额度快照到 SSOT base_limit=1(限本用例用户,不动全局规则行)。 + async with session_scope() as session: + await session.execute( + update(Quota) + .where(Quota.user_id == user_id, Quota.quota_key == "report_basic") + .values(limit_value=1) + ) resp = await client.post( "/api/v1/reports", headers=headers, json={ "questionnaire": {}, - "photo_object_key": "report_photo/u2.jpg", + "photo_object_key": await _upload_photo(client, headers), # 真实 key,先过存在性校验 "request_key": f"rk-{uuid.uuid4().hex}", # 新键:不是幂等重放 }, ) - assert resp.status_code == 409 + assert resp.status_code == 403 body = resp.json() - assert body["code"] == 40920 - assert "30" in (body.get("hint") or "") + assert body["code"] == 40301 # QUOTA_EXHAUSTED,而非 40920 # --- 场景 5:失败退款 + derived_features 保留 ---------------------------------- diff --git a/tests/unit/test_full_palette.py b/tests/unit/test_full_palette.py new file mode 100644 index 0000000..a6d454c --- /dev/null +++ b/tests/unit/test_full_palette.py @@ -0,0 +1,112 @@ +"""深度诊断色板规则映射与降级模板的纯函数单测(无 IO、无 LLM)。 + +红线:色板是规则映射常量——12 季型全覆盖、三档色数达标、hex 格式合法; +降级模板必须自带「通过三重校验」的性质(fail-close 的最后一道出口)。 +""" + +from __future__ import annotations + +import re + +from packages.report.full_copy import ( + TWELVE_SEASONS, + FullReportCopy, + build_full_result, + palette_for, + render_fallback_full_copy, + resolve_season, +) +from packages.report.validation import validate_copy + +_HEX_RE = re.compile(r"^#[0-9A-Fa-f]{6}$") + + +def test_all_twelve_seasons_have_complete_palettes() -> None: + assert len(TWELVE_SEASONS) == 12 + for season in TWELVE_SEASONS: + palette = palette_for(season) + assert len(palette["best"]) >= 12, f"{season} 最佳色不足 12 色" + assert len(palette["neutral"]) >= 6, f"{season} 中性色不足 6 色" + assert len(palette["avoid"]) >= 6, f"{season} 避雷色不足 6 色" + for role in ("best", "neutral", "avoid"): + for color in palette[role]: + assert color["name"], f"{season}/{role} 缺色名" + assert _HEX_RE.match(color["hex"]), f"{season}/{role} 色值非法:{color['hex']}" + + +def test_unknown_season_gets_empty_palette() -> None: + assert palette_for("不存在的季型") == {"best": [], "neutral": [], "avoid": []} + + +def test_resolve_season_prefers_detailed_value() -> None: + assert resolve_season({"season_detailed": "柔夏", "season_code": "summer"}) == "柔夏" + + +def test_resolve_season_falls_back_to_season_code() -> None: + assert resolve_season({"season_code": "spring"}) == "暖春" + assert resolve_season({"season_code": "winter"}) == "冷冬" + assert resolve_season({"season_detailed": "编造季型", "season_code": "autumn"}) == "暖秋" + + +def test_resolve_season_returns_none_without_any_input() -> None: + assert resolve_season({}) is None + assert resolve_season({"season_code": "unknown"}) is None + + +def _minimal_dimensions() -> list[dict[str, object]]: + return [ + {"key": "season_code", "value": "summer", "status": "ok"}, + {"key": "season_detailed", "value": "柔夏", "status": "ok"}, + ] + + +def test_fallback_full_copy_passes_validation() -> None: + """降级模板自带合规性:含「柔夏型」等十二季文案也不过三重校验的误杀线。""" + dimensions = _minimal_dimensions() + fallback = render_fallback_full_copy( + dimensions, season_detailed="柔夏", palette=palette_for("柔夏") + ) + violations = validate_copy(fallback.as_report_copy(), dimensions, rule_hits=[]) + assert violations == [] + + +def test_fallback_full_copy_covers_all_scenes_and_four_weeks() -> None: + fallback = render_fallback_full_copy( + _minimal_dimensions(), season_detailed="柔夏", palette=palette_for("柔夏") + ) + assert [s.key for s in fallback.scenes] == ["commute", "weekend", "formal", "date"] + assert all(len(scene.outfits) >= 1 for scene in fallback.scenes) + assert [w.week for w in fallback.plan30] == [1, 2, 3, 4] + assert all(3 <= len(week.tasks) <= 5 for week in fallback.plan30) + # 兜底文案不得出现色值(hex 只能走 full.palette 结构化字段)。 + assert "#" not in fallback.as_report_copy().full_text() + + +def test_build_full_result_shape() -> None: + copy = render_fallback_full_copy( + _minimal_dimensions(), season_detailed="柔夏", palette=palette_for("柔夏") + ) + result = build_full_result( + copy, + season_detailed="柔夏", + season_code="summer", + palette=palette_for("柔夏"), + ) + assert result["season"]["name_cn"] == "柔夏型" + assert result["season"]["analysis"] == copy.season_analysis + assert len(result["palette"]["best"]) >= 12 + assert result["scenes"][0]["outfits"][0]["points"] + assert result["plan30"][3]["week"] == 4 + + +def test_as_report_copy_folds_all_free_text_into_scan_scope() -> None: + """校验视图必须覆盖季型解析/场景/计划的全部自由文本,不是只查 summary。""" + copy = FullReportCopy( + summary="总评", + sections=[], + season_analysis="独一无二的季型解析", + scenes=[], + plan30=[], + ) + violations = validate_copy(copy.as_report_copy(), [], rule_hits=[]) + assert any("独一无二" in v for v in violations)