From d974526904b3a365ce36e3e36cda8d74e8dcd645 Mon Sep 17 00:00:00 2001 From: xsl Date: Fri, 14 Aug 2026 21:51:05 +0800 Subject: [PATCH] chore(prep): shared groundwork for batch-4 (report pipeline) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 第四批（T15 编排/T16 视觉诊断/T17 文案校验）协调提交： - 迁移 0006：rp_report（result JSONB 按阶段键承载中间态）/rp_input （derived_features 视觉中间态）/rp_share（V1 只建结构） - ORM 三模型；错误码 40420/40920（文案无价格字样） - 配置：report_pipeline_stages（dotted path 动态装配）+ vision/tier - packages/report/pipeline.py：Stage Protocol + ReportContext（三任务共同契约， 动态 import 的动机与代价写在 docstring） - 路由桩 reports.py 已注册 --- .env.example | 7 +++ apps/api/main.py | 14 ++++- apps/api/routers/reports.py | 11 ++++ migrations/versions/0006_report_tables.py | 75 +++++++++++++++++++++++ packages/report/pipeline.py | 72 ++++++++++++++++++++++ packages/shared/config.py | 19 ++++++ packages/shared/errors.py | 4 ++ packages/shared/models.py | 62 +++++++++++++++++++ 8 files changed, 263 insertions(+), 1 deletion(-) create mode 100644 apps/api/routers/reports.py create mode 100644 migrations/versions/0006_report_tables.py create mode 100644 packages/report/pipeline.py diff --git a/.env.example b/.env.example index 2705583..a852da5 100755 --- a/.env.example +++ b/.env.example @@ -152,3 +152,10 @@ ADMIN_TOKEN= RETRIEVAL_CACHE_ENABLED=true RETRIEVAL_CACHE_TTL_SECONDS=60 RETRIEVAL_CACHE_MAX_ENTRIES=512 + +# ── 美学报告流水线（T-P3，第四批）────────────────────────────────────────── +# 阶段 dotted path 列表（importlib 动态装配；缺失即 CONFIG_ERROR） +REPORT_PIPELINE_STAGES_JSON=["packages.report.vision_stage:VisionStage","packages.report.diagnosis_stage:DiagnosisStage","packages.report.copy_stage:CopyStage","packages.report.validation_stage:ValidationStage"] +# deterministic = V1 骨架假实现；真实 RTMPose/BiRefNet 为 Sprint 1 结论后的替换项 +REPORT_VISION_PROVIDER=deterministic +REPORT_TIER=basic diff --git a/apps/api/main.py b/apps/api/main.py index a1e44ee..8de88f8 100644 --- a/apps/api/main.py +++ b/apps/api/main.py @@ -12,7 +12,18 @@ from contextlib import asynccontextmanager from fastapi import FastAPI from apps.api.deps import build_llm_gateway -from apps.api.routers import agent, auth, auth_sms, events, health, kb, quota, tasks, uploads +from apps.api.routers import ( + agent, + auth, + auth_sms, + events, + health, + kb, + quota, + reports, + tasks, + uploads, +) from packages.shared.config import Settings, get_settings from packages.shared.db import dispose_engine from packages.shared.handlers import install_exception_handlers @@ -78,6 +89,7 @@ def create_app(settings: Settings | None = None) -> FastAPI: events.router, uploads.router, kb.router, + reports.router, ) for router in routers: app.include_router(router, prefix=cfg.api_prefix) diff --git a/apps/api/routers/reports.py b/apps/api/routers/reports.py new file mode 100644 index 0000000..a9ae8b8 --- /dev/null +++ b/apps/api/routers/reports.py @@ -0,0 +1,11 @@ +"""报告路由（骨架，T15 填充实现）。 + +POST /reports、GET /reports、GET /reports/{id}、POST /reports/{id}/regenerate； +实现契约见 docs/开发任务包/T15-报告域骨架.md。 +""" + +from __future__ import annotations + +from fastapi import APIRouter + +router = APIRouter(prefix="/reports", tags=["reports"]) diff --git a/migrations/versions/0006_report_tables.py b/migrations/versions/0006_report_tables.py new file mode 100644 index 0000000..880bfc9 --- /dev/null +++ b/migrations/versions/0006_report_tables.py @@ -0,0 +1,75 @@ +"""Report tables for the Sprint-3 (M3) parallel batch (架构文档 §6.2 report 域). + +三张表一次落齐，T15（编排）不再写迁移： + rp_report — result JSONB 承载流水线中间态与最终产物（按 stage 键增量更新） + rp_input — questionnaire / photo_meta / derived_features（视觉中间态，T-P3-09） + rp_share — V1 只建结构（路由属后续任务） + +Reversible: 是。 +""" + +from __future__ import annotations + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects import postgresql + +revision: str = "0006" +down_revision: str | None = "0005" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.create_table( + "rp_report", + sa.Column("report_id", sa.BigInteger(), nullable=False), + sa.Column("user_id", sa.BigInteger(), nullable=False), + sa.Column("tier", sa.String(length=8), nullable=False, server_default=sa.text("'basic'")), + sa.Column("status", sa.String(length=16), nullable=False, server_default=sa.text("'pending'")), + sa.Column("version", sa.Integer(), nullable=False, server_default=sa.text("1")), + sa.Column("result", postgresql.JSONB(), nullable=False), + sa.Column("model_info", postgresql.JSONB(), nullable=False), + sa.Column("content_code", sa.String(length=32), nullable=True), + sa.Column("rendered_asset_key", sa.String(length=255), nullable=True), + sa.Column("cost_cents", sa.Numeric(14, 6), nullable=False, server_default=sa.text("0")), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()), + sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True), + sa.PrimaryKeyConstraint("report_id"), + ) + op.create_index("ix_rp_report_user_created", "rp_report", ["user_id", "created_at"]) + + op.create_table( + "rp_input", + sa.Column("input_id", sa.BigInteger(), nullable=False), + sa.Column("report_id", sa.BigInteger(), nullable=False), + sa.Column("questionnaire", postgresql.JSONB(), nullable=False), + sa.Column("photo_meta", postgresql.JSONB(), nullable=False), + sa.Column("derived_features", postgresql.JSONB(), nullable=False), + sa.PrimaryKeyConstraint("input_id"), + sa.UniqueConstraint("report_id", name="uq_rp_input_report"), + sa.ForeignKeyConstraint(["report_id"], ["rp_report.report_id"], ondelete="CASCADE"), + ) + + op.create_table( + "rp_share", + sa.Column("share_id", sa.BigInteger(), nullable=False), + sa.Column("report_id", sa.BigInteger(), nullable=False), + sa.Column("share_token", sa.String(length=64), nullable=False), + sa.Column("expire_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("view_count", sa.Integer(), nullable=False, server_default=sa.text("0")), + sa.Column("revoked", sa.Boolean(), nullable=False, server_default=sa.text("false")), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()), + sa.PrimaryKeyConstraint("share_id"), + sa.UniqueConstraint("share_token", name="uq_rp_share_token"), + sa.ForeignKeyConstraint(["report_id"], ["rp_report.report_id"], ondelete="CASCADE"), + ) + + +def downgrade() -> None: + op.drop_table("rp_share") + op.drop_table("rp_input") + op.drop_index("ix_rp_report_user_created", table_name="rp_report") + op.drop_table("rp_report") diff --git a/packages/report/pipeline.py b/packages/report/pipeline.py new file mode 100644 index 0000000..4114865 --- /dev/null +++ b/packages/report/pipeline.py @@ -0,0 +1,72 @@ +"""报告流水线的阶段契约（第四批 T15/T16/T17 的共同接口）。 + +本文件由协调提交只定 **接口**；编排实现（assemble_stages / run_pipeline） +归 T15。T16 提供 vision_stage / diagnosis_stage，T17 提供 copy_stage / +validation_stage——三个任务互不改对方文件，靠 Settings.report_pipeline_stages +里的 dotted path 动态装配。 + +设计动机（为什么动态 import）：三个阶段模块由三个并行分支各自创建， +静态 import 会在任一分支缺文件时炸掉编排；按配置动态装配让每个分支 +独立可测，合并后自动咬合。代价是 import-linter 看不到这条边——阶段模块 +自身的依赖仍受契约约束，编排层注入的 session/gateway 都是既有合法依赖。 +""" + +from __future__ import annotations + +from typing import Any, Protocol, runtime_checkable + +from sqlalchemy.ext.asyncio import AsyncSession + +from packages.llm.gateway import LLMGateway +from packages.shared.config import Settings +from packages.shared.models import RpInput, RpReport + + +class ReportContext: + """一个阶段的执行上下文。 + + 不是 frozen：阶段要写 ``ctx.report.result[stage.name]`` 与 + ``ctx.rp_input.derived_features``（中间态落库由编排层 flush）。 + """ + + def __init__( + self, + *, + report: RpReport, + report_input: RpInput, + session: AsyncSession, + settings: Settings, + gateway: LLMGateway | None = None, + ) -> None: + self.report = report + self.report_input = report_input + self.session = session + self.settings = settings + self.gateway = gateway + + +@runtime_checkable +class Stage(Protocol): + """流水线阶段。name 同时是 result JSONB 里的产物键。 + + 约定：阶段把自己的全部产物写进 ``ctx.report.result[self.name]``； + 抛异常即整条流水线失败（编排层负责 tk_task→failed + refund）。 + """ + + name: str + + async def run(self, ctx: ReportContext) -> None: ... + + +#: T15 的 assemble_stages 按此清单 importlib 动态导入；缺失即 CONFIG_ERROR。 +DEFAULT_STAGE_PATHS: tuple[str, ...] = ( + "packages.report.vision_stage:VisionStage", + "packages.report.diagnosis_stage:DiagnosisStage", + "packages.report.copy_stage:CopyStage", + "packages.report.validation_stage:ValidationStage", +) + + +def stage_names(result: dict[str, Any]) -> list[str]: + """result JSONB 里已出现的阶段键（诊断/调试用的小工具）。""" + return [k for k in result if isinstance(result[k], dict)] diff --git a/packages/shared/config.py b/packages/shared/config.py index a0f29c4..78fd657 100644 --- a/packages/shared/config.py +++ b/packages/shared/config.py @@ -213,6 +213,24 @@ class Settings(BaseSettings): retrieval_cache_ttl_seconds: int = 60 retrieval_cache_max_entries: int = 512 + # --- 美学报告流水线（T-P3，第四批）--------------------------------------- + # 阶段按 dotted path 动态装配（importlib）：T15 编排、T16 供 vision/diagnosis、 + # T17 供 copy/validation——三个任务互不改对方文件，靠这份清单衔接。 + # 缺失模块/类在装配时报 CONFIG_ERROR 并列出缺失项，绝不静默跳过。 + report_pipeline_stages: list[str] = Field( + default_factory=lambda: [ + "packages.report.vision_stage:VisionStage", + "packages.report.diagnosis_stage:DiagnosisStage", + "packages.report.copy_stage:CopyStage", + "packages.report.validation_stage:ValidationStage", + ], + alias="REPORT_PIPELINE_STAGES_JSON", + ) + # 视觉提供者：deterministic = V1 骨架的确定性假实现（真实 RTMPose/BiRefNet + # 是 Sprint 1 结论落地后的替换项，接口位见 packages/report/vision.py）。 + report_vision_provider: str = "deterministic" + report_tier: str = "basic" + # --- Observability ----------------------------------------------------- otel_enabled: bool = False otel_service_name: str = "meida-api" @@ -228,6 +246,7 @@ class Settings(BaseSettings): "task_bonus_rules", "upload_scenes", "retrieval_rerank_weights", + "report_pipeline_stages", mode="before", ) @classmethod diff --git a/packages/shared/errors.py b/packages/shared/errors.py index 8ab5d01..923e42f 100644 --- a/packages/shared/errors.py +++ b/packages/shared/errors.py @@ -41,8 +41,10 @@ class ErrorCode(IntEnum): TASK_BONUS_CAP_REACHED = 40310 NOT_FOUND = 40400 TASK_BONUS_UNKNOWN = 40410 + REPORT_NOT_FOUND = 40420 STATE_CONFLICT = 40900 TASK_BONUS_ALREADY_CLAIMED = 40910 + REPORT_ALREADY_EXISTS = 40920 RATE_LIMITED = 42900 SMS_SEND_TOO_FREQUENT = 42910 SMS_DAILY_LIMIT = 42911 @@ -100,10 +102,12 @@ ERROR_META: dict[ErrorCode, ErrorMeta] = { ), ErrorCode.NOT_FOUND: ErrorMeta(404, "内容不存在或已删除"), ErrorCode.TASK_BONUS_UNKNOWN: ErrorMeta(404, "任务不存在或暂未开放"), + ErrorCode.REPORT_NOT_FOUND: ErrorMeta(404, "报告不存在或已删除"), ErrorCode.STATE_CONFLICT: ErrorMeta(409, "当前状态无法执行该操作"), ErrorCode.TASK_BONUS_ALREADY_CLAIMED: ErrorMeta( 409, "这个周期的任务奖励已领取", "下个周期再来" ), + ErrorCode.REPORT_ALREADY_EXISTS: ErrorMeta(409, "本期报告已生成过", "每 30 天可重新生成一次"), ErrorCode.RATE_LIMITED: ErrorMeta(429, "操作太频繁了", "请稍后再试"), ErrorCode.SMS_SEND_TOO_FREQUENT: ErrorMeta(429, "发送太频繁了", "请稍后再试"), ErrorCode.SMS_DAILY_LIMIT: ErrorMeta(429, "今日发送次数已达上限", "明天再试"), diff --git a/packages/shared/models.py b/packages/shared/models.py index 018a74a..ac40194 100644 --- a/packages/shared/models.py +++ b/packages/shared/models.py @@ -479,3 +479,65 @@ class KbGap(Base): ) __table_args__ = (Index("ix_kb_gap_query", "query_text"),) + + +# --------------------------------------------------------------------------- +# report 域（美学报告，架构文档 §6.2 / §7.1） +# --------------------------------------------------------------------------- +class RpReport(Base): + """报告主表。result JSONB 同时承载流水线中间态与最终产物（按 stage 键）—— + 阶段逐段落库是 T-P3-09「失败不需重新上传」的结构基础。""" + + __tablename__ = "rp_report" + + report_id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=False) + user_id: Mapped[int] = mapped_column(BigInteger, nullable=False) + tier: Mapped[str] = mapped_column(String(8), nullable=False, default="basic") + status: Mapped[str] = mapped_column(String(16), nullable=False, default="pending") + version: Mapped[int] = mapped_column(Integer, nullable=False, default=1) + result: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False, default=dict) + model_info: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False, default=dict) + content_code: Mapped[str | None] = mapped_column(String(32)) + rendered_asset_key: Mapped[str | None] = mapped_column(String(255)) + cost_cents: Mapped[Decimal] = mapped_column( + Numeric(14, 6), nullable=False, default=Decimal("0") + ) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + + __table_args__ = (Index("ix_rp_report_user_created", "user_id", "created_at"),) + + +class RpInput(Base): + """报告输入。derived_features 是视觉分析的中间态：重试/重新生成时 + 已存在则跳过视觉阶段（T-P3-09）。只存派生特征与元信息，不存原图（CR-02）。""" + + __tablename__ = "rp_input" + + input_id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=False) + report_id: Mapped[int] = mapped_column(BigInteger, nullable=False) + questionnaire: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False, default=dict) + photo_meta: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False, default=dict) + derived_features: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False, default=dict) + + __table_args__ = (UniqueConstraint("report_id", name="uq_rp_input_report"),) + + +class RpShare(Base): + """分享链接：可失效、可撤回（V1 只建结构，路由后续任务）。""" + + __tablename__ = "rp_share" + + share_id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=False) + report_id: Mapped[int] = mapped_column(BigInteger, nullable=False) + share_token: Mapped[str] = mapped_column(String(64), nullable=False) + expire_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + view_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + revoked: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + + __table_args__ = (UniqueConstraint("share_token", name="uq_rp_share_token"),)