Files
meida_backend/apps/api/routers/reports.py
T
xsl e8208d2972 feat(report): orchestration skeleton with staged pipeline and refund-on-failure
POST /reports 骨架:资格校验→终身配额预扣→tk_task→cpu 队列;动态装配
流水线(dotted-path 注册、缺阶段即 CONFIG_ERROR);中间态逐段落库、失败
退款、derived_features 跨重试/重生成复用(T-P3-07/09)。

实现:packages/report 的 pipeline.py(Stage 契约保持不变,新增 assemble_stages/
run_pipeline/StageFailureError,ReportContext 增可选 task_id 参数、向后兼容)、
repo.py、schemas.py、service.py(重放优先于资格校验、reserve+建行+建任务同事务、
提交后才投递、并发同键落败回滚);apps/api/routers/reports.py 填桩(send_task 按
任务名投递,不 import apps.worker);apps/worker/report_tasks.py(worker 侧自建
gateway 装配、状态归一 QUEUED→RUNNING、成功 commit 配额+成本聚合写 model_info、
基础设施异常重试一次、重试耗尽兜底失败退款)。测试:单测 6 个(装配缺失/空配置/
协议违约/顺序执行/失败保留产物),集成 6 场景(预扣 lifetime/PENDING、成功
committed+两阶段键、同 request_key 幂等不重复扣、二次首生成 40920、阶段失败
refund+derived_features 保留、regenerate 走 m: 月配额 version+1 复用视觉中间态)。

偏差说明:
1) apps/worker/celery_app.py 的 include 只含 apps.worker.tasks(禁改文件),真
worker 进程注册 meida.report.generate 需协调者合并时把 include 补上
apps.worker.report_tasks;API 侧已按任务名 send_task,两侧只靠名字衔接,测试
不受影响。2) ReportContext 增加可选 task_id 关键字参数(失败结案要推动
tk_task),Stage 协议/DEFAULT_STAGE_PATHS/stage_names 未动,T16/T17 兼容。
3) JSONB 深变更用 flag_modified 显式标脏(重新赋相等新 dict 会被 flush 的按值
比较跳过,集成测试抓到过该坑)。4) POST 后 tk_task 保持 PENDING(任务文档
§4 场景 1 口径),QUEUED→RUNNING 由 worker 入口归一。
2026-08-14 22:18:14 +08:00

129 lines
4.3 KiB
Python

"""报告路由(T15):触发、查询、重新生成。
投递方式:用 celery ``send_task`` 按任务名发消息,不 import apps.worker ——
apps 互不依赖契约只给 tasks→worker 开了一个例外,报告的投递边靠「任务名
约定」解耦,任务代码由 worker 进程加载(见 apps/worker/report_tasks.py)。
"""
from __future__ import annotations
import uuid
from functools import lru_cache
from typing import Annotated
from celery import Celery
from fastapi import APIRouter, Depends, Query
from apps.api.deps import CurrentUserId, SessionDep, SettingsDep
from packages.report.schemas import (
ReportCreateRequest,
ReportItem,
ReportRegenerateRequest,
ReportSubmitResponse,
)
from packages.report.service import ReportService, ReportSubmission
from packages.shared.config import get_settings
from packages.shared.models import RpReport
from packages.shared.response import ok
router = APIRouter(prefix="/reports", tags=["reports"])
REPORT_TASK_NAME = "meida.report.generate"
@lru_cache(maxsize=1)
def _producer() -> Celery:
"""send_task 专用生产者:只发消息不加载任务代码,不需要 result backend。"""
settings = get_settings()
return Celery("meida-api", broker=settings.celery_broker_url)
def _enqueue_report(task_id: int) -> None:
"""投进 cpu 队列(task_routes 里 meida.report.* → cpu 已配好,显式声明双保险)。"""
_producer().send_task(REPORT_TASK_NAME, args=[task_id], queue="cpu")
def _serialize(report: RpReport) -> dict[str, object]:
result = report.result or {}
if report.status == "failed":
# 失败报告只透错误摘要:半成品是流水线的中间态(给重试/重生成复用),
# 不是给用户看的内容。
result = {"error": result.get("error", {})}
return ReportItem(
report_id=str(report.report_id),
tier=report.tier,
status=report.status,
version=report.version,
result=result,
created_at=report.created_at,
completed_at=report.completed_at,
).model_dump()
def report_service_dep(session: SessionDep, settings: SettingsDep) -> ReportService:
return ReportService(session, settings=settings, enqueue=_enqueue_report)
ReportServiceDep = Annotated[ReportService, Depends(report_service_dep)]
@router.post("", summary="生成第一份美学报告(终身一次,走 report_basic 配额)")
async def create_report(
payload: ReportCreateRequest,
user_id: CurrentUserId,
service: ReportServiceDep,
) -> dict[str, object]:
request_key = payload.request_key or f"report:{user_id}:{uuid.uuid4().hex}"
submission = await service.create_report(
user_id=user_id,
questionnaire=payload.questionnaire,
photo_object_key=payload.photo_object_key,
request_key=request_key,
)
return ok(_submit_payload(submission))
@router.get("", summary="我的报告列表(归属校验)")
async def list_reports(
user_id: CurrentUserId,
service: ReportServiceDep,
limit: int = Query(default=20, ge=1, le=100),
) -> dict[str, object]:
rows = await service.list_reports(user_id=user_id, limit=limit)
return ok([_serialize(row) for row in rows])
@router.get("/{report_id}", summary="报告详情(归属校验)")
async def get_report(
report_id: int,
user_id: CurrentUserId,
service: ReportServiceDep,
) -> dict[str, object]:
report = await service.get_report(user_id=user_id, report_id=report_id)
return ok(_serialize(report))
@router.post(
"/{report_id}/regenerate", summary="重新生成(走 report_regen 月配额,复用照片与视觉结果)"
)
async def regenerate_report(
report_id: int,
payload: ReportRegenerateRequest,
user_id: CurrentUserId,
service: ReportServiceDep,
) -> dict[str, object]:
request_key = payload.request_key or f"report-regen:{user_id}:{report_id}:{uuid.uuid4().hex}"
submission = await service.regenerate(
user_id=user_id, report_id=report_id, request_key=request_key
)
return ok(_submit_payload(submission))
def _submit_payload(submission: ReportSubmission) -> dict[str, object]:
return ReportSubmitResponse(
task_id=str(submission.task_id),
report_id=str(submission.report_id),
status=submission.status,
version=submission.version,
).model_dump()