diff --git a/apps/api/routers/reports.py b/apps/api/routers/reports.py index a9ae8b8..5df2fb3 100644 --- a/apps/api/routers/reports.py +++ b/apps/api/routers/reports.py @@ -1,11 +1,128 @@ -"""报告路由（骨架，T15 填充实现）。 +"""报告路由（T15）：触发、查询、重新生成。 -POST /reports、GET /reports、GET /reports/{id}、POST /reports/{id}/regenerate； -实现契约见 docs/开发任务包/T15-报告域骨架.md。 +投递方式：用 celery ``send_task`` 按任务名发消息，不 import apps.worker —— +apps 互不依赖契约只给 tasks→worker 开了一个例外，报告的投递边靠「任务名 +约定」解耦，任务代码由 worker 进程加载（见 apps/worker/report_tasks.py）。 """ from __future__ import annotations -from fastapi import APIRouter +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() diff --git a/apps/worker/report_tasks.py b/apps/worker/report_tasks.py new file mode 100644 index 0000000..a9f14cd --- /dev/null +++ b/apps/worker/report_tasks.py @@ -0,0 +1,297 @@ +"""报告生成的 celery 任务（name=``meida.report.generate``，cpu 队列，T-P3-07/09）。 + +注册说明：``apps/worker/celery_app.py`` 的 include 目前只含 apps.worker.tasks +（本批禁改文件），worker 进程要注册本模块需要把 include 补上——由协调者在 +合并阶段处理；API 侧用 send_task 按任务名投递，两侧只靠名字衔接，互不 import。 + +worker 侧 LLM 网关装配：apps.api.deps 属 HTTP 层，worker 不得 import apps.api +（apps 互不依赖契约），这里按同一配方自建；改网关装配时两处同步。 + +成本归因约定：阶段经 gateway 调用时以 ``ref_type="report"``、 +``ref_id=str(report_id)`` 记账——成功结案按这个口径聚合写 +``rp_report.model_info`` / ``cost_cents``（ob_ai_call 是唯一数据源，架构 §13）。 +""" + +from __future__ import annotations + +import asyncio +from datetime import UTC, datetime +from decimal import Decimal +from typing import Any + +from sqlalchemy import func, select +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm.attributes import flag_modified + +from apps.worker.celery_app import celery_app +from packages.billing.service import BillingService +from packages.domain.task.state import TERMINAL_STATUSES, TaskStatus +from packages.llm.budget import DailyBudgetGuard, RedisCostCounter +from packages.llm.circuit import CircuitBreakerRegistry +from packages.llm.cost import ( + CompositeAICallRecorder, + ModelRate, + PeakAwareRateTable, + RateTable, +) +from packages.llm.gateway import LLMGateway +from packages.llm.prompts import PromptRegistry +from packages.llm.providers.base import LLMProvider +from packages.llm.providers.deepseek import DeepSeekProvider +from packages.llm.providers.zhipu import ZhipuGLMProvider +from packages.llm.recording import PrometheusAICallRecorder, SqlAlchemyAICallRecorder +from packages.llm.routing import SceneRouter, build_scene_configs +from packages.report import repo as report_repo +from packages.report.pipeline import ( + ReportContext, + StageFailureError, + assemble_stages, + load_reservation, + run_pipeline, +) +from packages.shared.config import Settings, get_settings +from packages.shared.context import bind_context +from packages.shared.db import get_session_factory +from packages.shared.errors import AppError, ErrorCode +from packages.shared.logging import get_logger +from packages.shared.models import AiCall, RpReport, Task +from packages.shared.redis_client import get_redis +from packages.task import repo as task_repo +from packages.task.service import TaskService + +logger = get_logger(__name__) + + +def build_llm_gateway(settings: Settings) -> LLMGateway: + """worker 侧装配，与 apps.api.deps.build_llm_gateway 同构（见模块 docstring）。""" + providers: dict[str, LLMProvider] = { + "deepseek": DeepSeekProvider( + api_key=settings.deepseek_api_key, + base_url=settings.deepseek_base_url, + default_model=settings.deepseek_model, + ), + "zhipu": ZhipuGLMProvider( + api_key=settings.zhipu_api_key, + base_url=settings.zhipu_base_url, + default_model=settings.zhipu_model, + ), + } + default_rate = ModelRate.from_mapping(settings.llm_default_cost_rate) + rates = PeakAwareRateTable( + RateTable( + { + model: ModelRate.from_mapping(rate) + for model, rate in settings.llm_cost_rates.items() + }, + default=default_rate, + ), + RateTable( + { + model: ModelRate.from_mapping(rate) + for model, rate in settings.llm_cost_rates_peak.items() + }, + default=default_rate, + ) + if settings.llm_cost_rates_peak + else None, + enabled=settings.llm_peak_pricing_enabled, + ) + return LLMGateway( + providers=providers, + router=SceneRouter( + build_scene_configs( + primary_provider=settings.llm_primary_provider, + backup_provider=settings.llm_backup_provider, + structured_max_retries=settings.llm_structured_max_retries, + overrides=settings.llm_scene_overrides, + ) + ), + prompts=PromptRegistry( + settings.prompts_dir, cache_prefix_mode=settings.llm_cache_prefix_mode + ), + rates=rates, + recorder=CompositeAICallRecorder( + SqlAlchemyAICallRecorder(get_session_factory()), + PrometheusAICallRecorder(), + ), + budget=DailyBudgetGuard( + RedisCostCounter(get_redis()), + daily_budget_cents=Decimal(settings.llm_daily_budget_cents), + enabled=settings.llm_daily_budget_enabled, + tz_offset_hours=settings.llm_budget_timezone_offset_hours, + tiers=list(settings.llm_budget_tiers), + ), + breakers=CircuitBreakerRegistry( + failure_threshold=settings.llm_circuit_failure_threshold, + cooldown_seconds=settings.llm_circuit_cooldown_seconds, + ), + ) + + +async def _aggregate_cost(session: AsyncSession, report_id: int) -> tuple[Decimal, int]: + """聚合该报告的 AI 成本与调用次数（口径见模块 docstring 的归因约定）。""" + result = await session.execute( + select(func.coalesce(func.sum(AiCall.cost_cents), 0), func.count()).where( + AiCall.ref_type == "report", AiCall.ref_id == str(report_id) + ) + ) + row = result.one() + return Decimal(row[0]), int(row[1]) + + +async def _settle_success( + session: AsyncSession, + task: Task, + report: RpReport, + settings: Settings, + stage_names: list[str], +) -> None: + """成功结案：commit 配额（幂等）+ succeeded + completed_at + 成本聚合。""" + cost_cents, ai_calls = await _aggregate_cost(session, report.report_id) + report.status = "succeeded" + report.completed_at = datetime.now(UTC) + report.cost_cents = cost_cents + report.model_info = { + "stages": stage_names, + "vision_provider": settings.report_vision_provider, + "ai_calls": ai_calls, + # Decimal 不能直接进 JSONB；字符串保留精度，展示层再格式化。 + "cost_cents": str(cost_cents), + } + reservation = await load_reservation(session, task.task_id) + if reservation is not None: + await BillingService(session).commit(reservation) + # 任务中心视图同步收口（前端轮询的就是 tk_task，不是 rp_report）。 + await TaskService(session).transition( + task_id=task.task_id, + target=TaskStatus.SUCCEEDED, + stage_text="已完成", + result={"report_id": str(report.report_id), "version": report.version}, + ) + await session.commit() + logger.info( + "report_succeeded", + report_id=report.report_id, + task_id=task.task_id, + cost_cents=str(cost_cents), + ai_calls=ai_calls, + ) + + +async def _generate_report(task_id: int) -> dict[str, Any]: + """执行一条报告流水线（celery 包装层的内芯；集成测试直调的就是它）。""" + settings = get_settings() + factory = get_session_factory() + async with factory() as session: + task = await task_repo.get_for_update(session, task_id) + if task is None: + raise AppError(ErrorCode.NOT_FOUND, message=f"任务不存在：{task_id}") + current = TaskStatus(task.status) + if current in TERMINAL_STATUSES: + # 重复投递/幂等重放的守卫：已结案的任务不再执行，防重复扣配额。 + logger.info("report_task_replay_skipped", task_id=task_id, status=task.status) + return {"task_id": str(task_id), "status": task.status, "skipped": True} + + report_id_raw = task.payload.get("report_id") + if not isinstance(report_id_raw, str | int): + raise AppError(ErrorCode.STATE_CONFLICT, message="任务缺少 report_id") + report_id = int(report_id_raw) + report = await report_repo.get_report(session, report_id) + report_input = await report_repo.get_input(session, report_id) + if report is None or report_input is None: + raise AppError(ErrorCode.STATE_CONFLICT, message="报告数据不完整") + + service = TaskService(session) + # 首跑是 PENDING；重试入站可能是 RUNNING（worker 崩溃重投）或 FAILED + # （上一轮失败后重试），先归一到 QUEUED（状态机允许，且会维护 + # retry_count）再进 RUNNING。 + if current is not TaskStatus.QUEUED: + await service.transition(task_id=task_id, target=TaskStatus.QUEUED, stage_text="排队中") + await service.transition(task_id=task_id, target=TaskStatus.RUNNING, stage_text="分析中") + report.status = "running" + await session.commit() # RUNNING 先落库：前端进度与崩溃恢复都读它 + + gateway = build_llm_gateway(settings) + try: + stages = assemble_stages(settings) + ctx = ReportContext( + report=report, + report_input=report_input, + session=session, + settings=settings, + gateway=gateway, + task_id=task_id, + ) + await run_pipeline(ctx, stages) + await _settle_success(session, task, report, settings, [s.name for s in stages]) + finally: + await gateway.aclose() + return {"task_id": str(task_id), "report_id": str(report_id), "status": "succeeded"} + + +async def _fail_report(task_id: int, message: str) -> None: + """基础设施异常重试耗尽后的兜底结案：failed + 退款（尽力而为）。 + + 与 run_pipeline 的失败结案不同，这里没有阶段名——异常发生在流水线之外 + （装配、DB 连接等），错误摘要以 worker 为责任方落库。结案失败只记日志： + 此时额度停在 reserved，靠流水对账兜底。 + """ + factory = get_session_factory() + try: + async with factory() as session: + task = await task_repo.get_for_update(session, task_id) + if task is None or TaskStatus(task.status) in TERMINAL_STATUSES: + return + report_id_raw = task.payload.get("report_id") + if isinstance(report_id_raw, str | int): + report = await report_repo.get_report(session, int(report_id_raw)) + if report is not None and report.status != "succeeded": + report.status = "failed" + report.result["error"] = { + "stage": "worker", + "type": "infrastructure", + "message": message[:200], + } + # JSONB 深变更要显式标脏，否则 flush 按值比较会跳过 UPDATE。 + flag_modified(report, "result") + await TaskService(session).transition( + task_id=task_id, + target=TaskStatus.FAILED, + stage_text="执行失败", + error_message=message, + ) + reservation = await load_reservation(session, task_id) + if reservation is not None: + await BillingService(session).refund(reservation, reason="report_worker_failed") + await session.commit() + except Exception: + logger.exception("report_fail_settle_failed", task_id=task_id) + + +# Celery 的 task 装饰器没有类型标注，会把被装饰函数擦成 untyped +@celery_app.task(name="meida.report.generate", bind=True, max_retries=1) # type: ignore[untyped-decorator] +def generate_report(self: Any, task_id: int) -> dict[str, Any]: + """cpu 队列的报告生成入口。 + + 重试语义：网络/基础设施类异常重试一次——重跑时 ``derived_features`` + 已落库，视觉阶段自然跳过（T-P3-09）；阶段业务失败（StageFailureError） + 已在流水线内结案（退款已提交），重跑也不会好，不再重试。 + """ + bind_context(task_id=task_id) + try: + return asyncio.run(_generate_report(task_id)) + except StageFailureError as exc: + logger.warning( + "report_stage_failed", + task_id=task_id, + stage=exc.stage_name, + error=f"{type(exc.original).__name__}: {exc.original}", + ) + raise + except Exception as exc: + if self.request.retries < self.max_retries: + logger.warning("report_generate_retry", task_id=task_id, error=str(exc)) + raise self.retry(exc=exc, countdown=10) from exc + logger.exception("report_generate_failed_after_retry", task_id=task_id) + asyncio.run(_fail_report(task_id, f"{type(exc).__name__}: {exc}")) + raise diff --git a/packages/report/__init__.py b/packages/report/__init__.py index d2b4915..4ae0176 100644 --- a/packages/report/__init__.py +++ b/packages/report/__init__.py @@ -1 +1 @@ -"""report 域：报告触发判定、生成编排、版本管理、分享。占位。""" +"""report 域：报告触发、流水线编排、版本管理与查询（分享 rp_share 归后续任务）。""" diff --git a/packages/report/pipeline.py b/packages/report/pipeline.py index 4114865..4443147 100644 --- a/packages/report/pipeline.py +++ b/packages/report/pipeline.py @@ -1,9 +1,10 @@ -"""报告流水线的阶段契约（第四批 T15/T16/T17 的共同接口）。 +"""报告流水线的阶段契约与编排（第四批 T15/T16/T17 的共同接口）。 -本文件由协调提交只定 **接口**；编排实现（assemble_stages / run_pipeline） -归 T15。T16 提供 vision_stage / diagnosis_stage，T17 提供 copy_stage / -validation_stage——三个任务互不改对方文件，靠 Settings.report_pipeline_stages -里的 dotted path 动态装配。 +阶段契约（``ReportContext`` / ``Stage`` / ``DEFAULT_STAGE_PATHS``）由协调提交 +定死，T15 实现编排（``assemble_stages`` / ``run_pipeline``）。T16 提供 +vision_stage / diagnosis_stage，T17 提供 copy_stage / validation_stage—— +三个任务互不改对方文件，靠 Settings.report_pipeline_stages 里的 dotted +path 动态装配。 设计动机（为什么动态 import）：三个阶段模块由三个并行分支各自创建， 静态 import 会在任一分支缺文件时炸掉编排；按配置动态装配让每个分支 @@ -13,13 +14,24 @@ validation_stage——三个任务互不改对方文件，靠 Settings.report_pi from __future__ import annotations +import importlib from typing import Any, Protocol, runtime_checkable from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm.attributes import flag_modified +from packages.billing import repo as billing_repo +from packages.billing.service import BillingService, Reservation +from packages.domain.task.state import TaskStatus from packages.llm.gateway import LLMGateway from packages.shared.config import Settings -from packages.shared.models import RpInput, RpReport +from packages.shared.errors import AppError, ErrorCode +from packages.shared.logging import get_logger +from packages.shared.models import Ledger, RpInput, RpReport +from packages.task import repo as task_repo +from packages.task.service import TaskService + +logger = get_logger(__name__) class ReportContext: @@ -37,12 +49,16 @@ class ReportContext: session: AsyncSession, settings: Settings, gateway: LLMGateway | None = None, + task_id: int | None = None, ) -> None: self.report = report self.report_input = report_input self.session = session self.settings = settings self.gateway = gateway + # 编排层专用：失败结案要推动 tk_task。阶段实现不需要感知它， + # 保持可选以保证旧构造调用（T16/T17 的测试）不受影响。 + self.task_id = task_id @runtime_checkable @@ -70,3 +86,162 @@ DEFAULT_STAGE_PATHS: tuple[str, ...] = ( def stage_names(result: dict[str, Any]) -> list[str]: """result JSONB 里已出现的阶段键（诊断/调试用的小工具）。""" return [k for k in result if isinstance(result[k], dict)] + + +class StageFailureError(Exception): + """阶段异常且已按失败结案（rp_report failed + tk_task failed + 退款已提交）。 + + celery 包装层据此区分两类异常：``StageFailureError`` 是业务性失败，重跑 + 也不会好，不再重试；其它异常（DB/网络/装配）发生在结算之外，值得用 + 重试换恢复。 + """ + + def __init__(self, stage_name: str, original: BaseException) -> None: + self.stage_name = stage_name + self.original = original + super().__init__(f"stage={stage_name} error={type(original).__name__}: {original}") + + +def assemble_stages(settings: Settings) -> list[Stage]: + """按 ``settings.report_pipeline_stages`` 的 dotted path 动态装配阶段列表。 + + 动态 import 是刻意的插件式装配（见模块 docstring）：T16/T17 各自建阶段 + 模块，不碰本文件；import-linter 看不到这条边，阶段模块自身的依赖仍受 + 分层契约约束，这里只做装配不做业务。 + + 模块/类不存在、实例化失败、或不符合 Stage 协议时，收集**全部**缺失项 + 抛 ``CONFIG_ERROR`` —— 绝不静默跳过：静默等于流水线空跑还扣用户配额。 + """ + if not settings.report_pipeline_stages: + raise AppError( + ErrorCode.CONFIG_ERROR, + message="报告流水线阶段配置为空", + data={"missing": list(DEFAULT_STAGE_PATHS)}, + ) + + missing: list[str] = [] + stages: list[Stage] = [] + for dotted in settings.report_pipeline_stages: + module_path, sep, attr = dotted.partition(":") + stage: Any = None + if sep and module_path and attr: + try: + # 阶段构造约定无参。__init__ 抛错视同配置坏了（比如阶段自身 + # 校验它的依赖），归入 CONFIG_ERROR 而不是让 worker 崩溃。 + stage = getattr(importlib.import_module(module_path), attr)() + except Exception: + stage = None + if stage is None or not isinstance(stage, Stage): + missing.append(dotted) + else: + stages.append(stage) + + if missing: + raise AppError( + ErrorCode.CONFIG_ERROR, + message="报告流水线阶段装配失败：缺失或不可用", + data={"missing": missing}, + ) + return stages + + +def reservation_from_ledger(ledger: Ledger) -> Reservation: + """从 reserved 流水重建 Reservation（worker 成功 commit / 失败 refund 都要它）。""" + return Reservation( + tx_id=ledger.tx_id, + user_id=ledger.user_id, + quota_key=ledger.quota_key or "", + period_key=ledger.period_key or "", + amount=abs(ledger.delta), + remaining=ledger.balance_after, + idempotency_key=ledger.idempotency_key, + ) + + +async def load_reservation(session: AsyncSession, task_id: int) -> Reservation | None: + """按 ``tk_task.reserve_tx_id`` 取回预扣；无预扣或流水缺失返回 None。""" + task = await task_repo.get(session, task_id) + if task is None or task.reserve_tx_id is None: + return None + ledger = await billing_repo.get_ledger(session, task.reserve_tx_id) + if ledger is None: + return None + return reservation_from_ledger(ledger) + + +async def run_pipeline(ctx: ReportContext, stages: list[Stage]) -> None: + """逐阶段执行，每阶段产物即刻 flush 落库（T-P3-09 中间态持久化）。 + + 中间态的意义：阶段半程失败或 worker 崩溃后重试，已完成阶段（尤其 + ``rp_input.derived_features`` 里的视觉结果）不必重算——「失败不需重新 + 上传照片」的结构保证就落在这几次 flush 里。 + + 阶段异常 → ``_settle_stage_failure``（事务纪律见其 docstring）后包成 + :class:`StageFailureError` 上抛；正常返回即全部阶段成功，成功结案归调用方 + （worker：commit 配额 + succeeded + 成本聚合）。 + """ + # 防御：default=dict 只在 INSERT 时生效，测试/手工构造的瞬态对象可能为 None。 + if ctx.report.result is None: + ctx.report.result = {} + if ctx.report_input.derived_features is None: + ctx.report_input.derived_features = {} + + for stage in stages: + try: + await stage.run(ctx) + except Exception as exc: + await _settle_stage_failure(ctx, stage.name, exc) + raise StageFailureError(stage.name, exc) from exc + # JSONB 列没有 Mutable 追踪：阶段对 dict 的深变更 ORM 看不见。且不能 + # 靠「重新赋一个相等的新 dict」——flush 按值比较会认为没变而跳过 + # UPDATE。flag_modified 强制该列进入下一条 UPDATE（中间态逐段落库 + # 的关键）。derived_features 同理。 + ctx.report.result.setdefault(stage.name, {}) + flag_modified(ctx.report, "result") + flag_modified(ctx.report_input, "derived_features") + await ctx.session.flush() + logger.info( + "report_stage_done", + report_id=ctx.report.report_id, + task_id=ctx.task_id, + stage=stage.name, + ) + + +async def _settle_stage_failure(ctx: ReportContext, stage_name: str, exc: Exception) -> None: + """阶段失败的结案：report failed + tk_task failed + 退款，提交后才返回。 + + 事务纪律（对齐 AgentService）：退款必须先于异常上抛提交，否则外层回滚 + 会把退款一起卷走（FR-BILL-05）。结案本身失败只记日志、不吞原始异常—— + 此时额度挂在 reserved 状态，靠流水对账兜底。 + """ + summary = {"stage": stage_name, "type": type(exc).__name__, "message": str(exc)[:200]} + try: + ctx.report.status = "failed" + ctx.report.result["error"] = summary + flag_modified(ctx.report, "result") # 深变更必须显式标脏，见 run_pipeline 的注释 + if ctx.task_id is not None: + await TaskService(ctx.session).transition( + task_id=ctx.task_id, + target=TaskStatus.FAILED, + stage_text=f"阶段失败：{stage_name}", + error_message=f"{type(exc).__name__}: {exc}", + ) + reservation = await load_reservation(ctx.session, ctx.task_id) + if reservation is not None: + await BillingService(ctx.session).refund( + reservation, reason=f"report_stage_failed:{stage_name}" + ) + await ctx.session.commit() + logger.warning( + "report_stage_failed", + report_id=ctx.report.report_id, + task_id=ctx.task_id, + stage=stage_name, + error_type=type(exc).__name__, + ) + except Exception: + await ctx.session.rollback() + logger.exception( + "report_failure_settle_failed", report_id=ctx.report.report_id, stage=stage_name + ) diff --git a/packages/report/repo.py b/packages/report/repo.py new file mode 100644 index 0000000..4dcecef --- /dev/null +++ b/packages/report/repo.py @@ -0,0 +1,47 @@ +"""rp_report / rp_input 的数据访问。**只有本模块可以触这两张表**（rp_share 归后续任务）。""" + +from __future__ import annotations + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from packages.shared.models import RpInput, RpReport + +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) + .where(RpReport.user_id == user_id) + .order_by(RpReport.created_at.desc(), RpReport.report_id.desc()) + .limit(limit) + ) + return list(result.scalars().all()) + + +async def get_input(session: AsyncSession, report_id: int) -> RpInput | None: + result = await session.execute(select(RpInput).where(RpInput.report_id == report_id)) + return result.scalar_one_or_none() diff --git a/packages/report/schemas.py b/packages/report/schemas.py new file mode 100644 index 0000000..cac88ff --- /dev/null +++ b/packages/report/schemas.py @@ -0,0 +1,48 @@ +"""报告域的请求/响应模型。 + +Wire 响应统一走 ok() 信封（架构文档 §9.2），这里的模型只描述 ``data`` 部分； +雪花 ID 一律以字符串返回，避免 JS 精度丢失（NFR-MAINT-01 契约先行）。 +""" + +from __future__ import annotations + +from datetime import datetime + +from pydantic import BaseModel, Field + + +class ReportCreateRequest(BaseModel): + questionnaire: dict[str, object] = Field( + default_factory=dict, description="问卷答案；字段结构由产品定稿后在路由层收紧" + ) + photo_object_key: str = Field( + min_length=1, max_length=255, description="临时桶里已上传照片的对象键（report_photo 场景）" + ) + request_key: str | None = Field( + default=None, max_length=128, description="幂等键；相同键返回同一任务且不重复扣配额" + ) + + +class ReportRegenerateRequest(BaseModel): + request_key: str | None = Field( + default=None, max_length=128, description="幂等键；相同键返回同一任务且不重复扣配额" + ) + + +class ReportSubmitResponse(BaseModel): + task_id: str + report_id: str + status: str = Field(description="tk_task 状态；报告本体状态与其同步推进") + version: int + + +class ReportItem(BaseModel): + report_id: str + tier: str + status: str + version: int + result: dict[str, object] = Field( + description="按阶段键组织的产物；failed 报告只透错误摘要（中间态留给流水线复用）" + ) + created_at: datetime + completed_at: datetime | None diff --git a/packages/report/service.py b/packages/report/service.py new file mode 100644 index 0000000..7384214 --- /dev/null +++ b/packages/report/service.py @@ -0,0 +1,266 @@ +"""报告域服务：触发资格、配额预扣、版本化与幂等（T-P3-07 / T-P3-09）。 + +事务纪律：reserve 与建报告行、建 tk_task 在**同一事务**；commit 之后才投递 +队列——队列消息不能早于数据落库，否则 worker 可能读到一个不存在的报告。 +幂等是双保险：tk_task 的幂等键（= request_key）挡重复任务行，reserve 的 +幂等键挡重复预扣；服务层再加一次「重放优先」快捷路径，让首单成功后的 +同键重试拿到 200 而不是 40920。 +""" + +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass + +from sqlalchemy.ext.asyncio import AsyncSession + +from packages.billing.service import BillingService +from packages.domain.ids import next_id +from packages.domain.task.state import TaskStatus, TaskType +from packages.report import repo +from packages.shared.config import Settings +from packages.shared.errors import AppError, ErrorCode, ForbiddenError +from packages.shared.logging import get_logger +from packages.shared.models import RpInput, RpReport, Task +from packages.task import repo as task_repo +from packages.task.service import TaskService + +logger = get_logger(__name__) + +#: 首次生成走终身配额（迁移 0003：lifetime / base 1），重新生成走月配额。 +QUOTA_KEY_BASIC = "report_basic" +QUOTA_KEY_REGEN = "report_regen" +REPORT_REF_TYPE = "report" +REPORT_ETA_SECONDS = 60 + +_TERMINAL_TASK_STATUSES = frozenset( + {TaskStatus.SUCCEEDED.value, TaskStatus.FAILED.value, TaskStatus.CANCELED.value} +) + + +@dataclass(frozen=True, slots=True) +class ReportSubmission: + """一次提交（新建或重新生成）的结果；幂等重放返回同值。""" + + task_id: int + report_id: int + status: str + version: int + + +def _submission_of(task: Task) -> ReportSubmission: + """从 tk_task 行重建提交结果（幂等重放与并发落败两条路径共用）。""" + raw_report_id = task.payload.get("report_id") + if not isinstance(raw_report_id, str) or not raw_report_id.isdigit(): + # 同键任务却没有 report_id：不是本域写进去的，按幂等冲突拒绝。 + raise AppError(ErrorCode.IDEMPOTENCY_CONFLICT) + return ReportSubmission( + task_id=task.task_id, + report_id=int(raw_report_id), + status=task.status, + version=int(task.payload.get("version", 1)), + ) + + +class ReportService: + def __init__( + self, + session: AsyncSession, + *, + settings: Settings, + enqueue: Callable[[int], None] | None = None, + ) -> None: + self._session = session + self._settings = settings + # enqueue 注入投递动作（apps 层用 celery send_task）；None 便于测试 + # 与未来非队列场景直调。投递时机由本服务统一保证在 commit 之后。 + self._enqueue = enqueue + self._billing = BillingService( + session, tz_offset_hours=settings.llm_budget_timezone_offset_hours + ) + self._tasks = TaskService(session) + + # -- 首次生成 ----------------------------------------------------------- + + async def create_report( + self, + *, + user_id: int, + questionnaire: dict[str, object], + photo_object_key: str, + request_key: str, + ) -> ReportSubmission: + # 幂等重放最先判：首单成功后用同一 request_key 重试不是「再生成一次」， + # 必须先于终身资格校验按原任务返回（否则会把幂等重试误判成 40920）。 + 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) + + report = RpReport( + report_id=next_id(), + user_id=user_id, + tier=self._settings.report_tier, + status="pending", + version=1, + ) + report_input = RpInput( + input_id=next_id(), + report_id=report.report_id, + questionnaire=dict(questionnaire), + photo_meta={"object_key": photo_object_key}, + derived_features={}, + ) + submission = await self._submit_and_deliver( + user_id=user_id, + report=report, + report_input=report_input, + quota_key=QUOTA_KEY_BASIC, + request_key=request_key, + reason="first basic report", + ) + logger.info( + "report_created", + user_id=user_id, + report_id=submission.report_id, + task_id=submission.task_id, + ) + return submission + + # -- 重新生成 ----------------------------------------------------------- + + async def regenerate( + self, *, user_id: int, report_id: int, request_key: str + ) -> ReportSubmission: + 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.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=source.tier, + status="pending", + version=source.version + 1, + ) + # 复用输入：questionnaire / photo_meta 原样带过来；derived_features 一并 + # 拷贝——新一轮流水线的视觉阶段看到非空即跳过（T-P3-09），这就是 + # 「重新生成 / 失败重试不需要重新上传照片」。 + 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_REGEN, + request_key=request_key, + reason=f"regenerate report v{source.version}", + ) + logger.info( + "report_regenerated", + user_id=user_id, + source_report_id=report_id, + report_id=submission.report_id, + task_id=submission.task_id, + version=submission.version, + ) + return submission + + # -- 查询 --------------------------------------------------------------- + + async def get_report(self, *, user_id: int, report_id: int) -> RpReport: + report = await repo.get_report(self._session, report_id) + if report is None: + raise AppError(ErrorCode.REPORT_NOT_FOUND) + if report.user_id != user_id: + # 归属校验（架构 §11.1）：不能只靠 ID 不可猜，必须显式比对。 + raise ForbiddenError() + return report + + async def list_reports(self, *, user_id: int, limit: int = 20) -> list[RpReport]: + return await repo.list_reports(self._session, user_id=user_id, limit=limit) + + # -- 内部 --------------------------------------------------------------- + + async def _find_replay(self, *, request_key: str, user_id: int) -> ReportSubmission | None: + """同 request_key 的 REPORT 任务已存在时重建提交结果。""" + existing = await task_repo.find_by_idempotency_key(self._session, request_key) + if existing is None: + return None + if existing.user_id != user_id or existing.task_type != TaskType.REPORT.value: + # 幂等键全局唯一，撞上了其它用户/其它功能的关键字：按幂等冲突拒绝， + # 绝不把别人的任务当重放结果返回。 + raise AppError(ErrorCode.IDEMPOTENCY_CONFLICT) + return _submission_of(existing) + + async def _submit_and_deliver( + self, + *, + user_id: int, + report: RpReport, + report_input: RpInput, + quota_key: str, + request_key: str, + reason: str, + ) -> ReportSubmission: + """预扣 + 建行 + 建任务 + 提交后投递——create 与 regenerate 的共用尾段。 + + 并发同 request_key：reserve 幂等命中、tk_task 幂等键唯一，后到者回滚 + 自己多建的报告行并按先到者的任务返回（不重复预扣、不留孤儿行）。 + """ + reservation = await self._billing.reserve( + user_id=user_id, + quota_key=quota_key, + request_key=request_key, + amount=1, + reason=reason, + ref_type=REPORT_REF_TYPE, + ) + # rp_input.report_id 有 FK 指向 rp_report，但两表之间没有 ORM + # relationship，unit of work 不知道先后——必须显式先落报告行。 + self._session.add(report) + await self._session.flush() + self._session.add(report_input) + task = await self._tasks.submit( + user_id=user_id, + task_type=TaskType.REPORT, + idempotency_key=request_key, + payload={"report_id": str(report.report_id), "version": report.version}, + quota_reserved=reservation.amount, + reserve_tx_id=reservation.tx_id, + eta_seconds=REPORT_ETA_SECONDS, + ) + if task.payload.get("report_id") != str(report.report_id): + # 另一个同键请求抢先建了任务（tk_task 唯一键），本请求的报告行作废。 + await self._session.rollback() + winner = await task_repo.find_by_idempotency_key(self._session, request_key) + if winner is None: # pragma: no cover - 唯一键刚被对手占住，不可达兜底 + raise AppError(ErrorCode.IDEMPOTENCY_CONFLICT) + return _submission_of(winner) + + # 先提交再投递：worker 收到消息时报告行必须已经可见。 + await self._session.commit() + self._deliver(task) + return ReportSubmission(task.task_id, report.report_id, task.status, report.version) + + def _deliver(self, task: Task) -> None: + """事务提交后投递（见类 docstring）；终态任务不重复投递。""" + if self._enqueue is None or task.status in _TERMINAL_TASK_STATUSES: + return + self._enqueue(task.task_id) diff --git a/tests/integration/test_report_flow.py b/tests/integration/test_report_flow.py new file mode 100644 index 0000000..c44b92b --- /dev/null +++ b/tests/integration/test_report_flow.py @@ -0,0 +1,425 @@ +"""报告域端到端（T-P3-07/09）：触发→配额→流水线→失败退款→重生成。 + +Fake 阶段定义在本文件里，通过 ``REPORT_PIPELINE_STAGES_JSON`` 让 +assemble_stages 按 dotted path 动态装配——与 T16/T17 的真实阶段走完全相同 +的装配路径。不启真 celery：worker 内芯函数直调；API 投递钩子替换为 no-op， +避免测试消息进真 Redis 队列被开发环境的 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", + ), +] + +# Fake 阶段走 importlib 动态装配，dotted path 必须可导入（本文件自身）。 +_MODULE = "tests.integration.test_report_flow" +_SUCCESS_STAGES = json.dumps([f"{_MODULE}:FakeVisionStage", f"{_MODULE}:FakeDiagnosisStage"]) +_FAILING_STAGES = json.dumps([f"{_MODULE}:FakeVisionStage", f"{_MODULE}:FakeBoomStage"]) + + +class FakeVisionStage: + """视觉阶段替身：计算次数写进 derived_features，用例据此断言「已存在即跳过」。""" + + name = "vision" + + async def run(self, ctx: Any) -> None: + features = ctx.report_input.derived_features + if not features: + # 与 T16 真实视觉阶段同一契约（T-P3-09）：中间态非空则不重算。 + 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"] = {"summary": "肩宽腰细", "source": "vision"} + + +class FakeBoomStage: + name = "boom" + + async def run(self, ctx: Any) -> None: + raise RuntimeError("爆了：模拟阶段失败") + + +# --- fixtures --------------------------------------------------------------- + + +@pytest.fixture +def no_enqueue(monkeypatch: pytest.MonkeyPatch) -> None: + """不向真 Redis 投递：worker 由用例直调（celery 注册/消费不在本用例范围）。""" + monkeypatch.setattr("apps.api.routers.reports._enqueue_report", lambda task_id: None) + + +@contextmanager +def _stage_config(monkeypatch: pytest.MonkeyPatch, payload: str) -> Iterator[None]: + """把阶段配置换成 Fake 阶段（setenv + cache_clear，结束时恢复）。 + + API 进程与 worker 内芯读的是同一个 get_settings 单例，cache_clear 之后 + 两边都会按 env 重建。 + """ + from packages.shared.config import get_settings + + monkeypatch.setenv("REPORT_PIPELINE_STAGES_JSON", payload) + 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, _SUCCESS_STAGES): + yield + + +@pytest.fixture +def failing_stages(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]: + with _stage_config(monkeypatch, _FAILING_STAGES): + 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]: + """dev-token：每个用例独立 device_key，拿到的就是全新用户。""" + resp = await client.post( + "/api/v1/auth/dev-token", json={"device_key": f"report-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 _create_report( + client: Any, headers: dict[str, str], *, request_key: str | None = None +) -> dict[str, Any]: + resp = await client.post( + "/api/v1/reports", + headers=headers, + json={ + "questionnaire": {"style": "通勤", "height_cm": 168}, + "photo_object_key": "report_photo/u1.jpg", + "request_key": request_key or f"rk-{uuid.uuid4().hex}", + }, + ) + assert resp.status_code == 200 + body = resp.json()["data"] + assert body["status"] == "pending" + return body + + +async def _fetch_one(statement: Any) -> Any: + from packages.shared.db import session_scope + + async with session_scope() as session: + result = await session.execute(statement) + return result.scalar_one_or_none() + + +async def _fetch_report(report_id: int) -> Any: + from sqlalchemy import select + + from packages.shared.models import RpReport + + return await _fetch_one(select(RpReport).where(RpReport.report_id == report_id)) + + +async def _fetch_input(report_id: int) -> Any: + from sqlalchemy import select + + from packages.shared.models import RpInput + + return await _fetch_one(select(RpInput).where(RpInput.report_id == report_id)) + + +async def _fetch_task(task_id: int) -> Any: + from packages.shared.db import session_scope + from packages.task import repo as task_repo + + async with session_scope() as session: + task = await task_repo.get(session, task_id) + assert task is not None + # 脱离会话前把要断言的 JSONB 读出来（expire_on_commit=False，够用）。 + return task + + +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：触发与预扣 ------------------------------------------------------- + + +async def test_create_reserves_lifetime_quota_and_creates_pending_task( + fake_stages: None, client: Any +) -> None: + headers, user_id = await _login(client) + + data = await _create_report(client, headers) + task_id, report_id = int(data["task_id"]), int(data["report_id"]) + assert data["version"] == 1 + + task = await _fetch_task(task_id) + assert task.status == "pending" + assert task.task_type == "report" + assert task.payload["report_id"] == str(report_id) + assert task.quota_reserved == 1 + assert task.reserve_tx_id is not None + + report = await _fetch_report(report_id) + assert report.status == "pending" + rp_input = await _fetch_input(report_id) + assert rp_input.photo_meta == {"object_key": "report_photo/u1.jpg"} + assert rp_input.derived_features == {} + + rows = await _ledger_rows(user_id, "report_basic") + assert len(rows) == 1 + assert rows[0].tx_type == "consume" + assert rows[0].status == "reserved" + assert rows[0].period_key == "life" # report_basic 是终身配额（迁移 0003） + + +# --- 场景 2：worker 成功 ------------------------------------------------------ + + +async def test_worker_success_commits_quota_and_writes_stage_keys( + fake_stages: None, client: Any +) -> None: + from apps.worker.report_tasks import _generate_report + + headers, user_id = await _login(client) + data = await _create_report(client, headers) + task_id, report_id = int(data["task_id"]), int(data["report_id"]) + + result = await _generate_report(task_id) + assert result["status"] == "succeeded" + + report = await _fetch_report(report_id) + assert report.status == "succeeded" + assert report.completed_at is not None + assert set(report.result) == {"vision", "diagnosis"} + assert report.result["vision"] == {"features": ["vision_runs"]} + assert "stages" in report.model_info + + task = await _fetch_task(task_id) + assert task.status == "succeeded" + assert task.progress == 100 + + rows = await _ledger_rows(user_id, "report_basic") + assert [row.status for row in rows] == ["committed"] + + # 列表与详情视图透出同样的产物。 + listing = (await client.get("/api/v1/reports", headers=headers)).json()["data"] + assert [item["report_id"] for item in listing] == [str(report_id)] + assert listing[0]["result"]["diagnosis"]["summary"] == "肩宽腰细" + + +# --- 场景 3：幂等重提交 ------------------------------------------------------- + + +async def test_duplicate_request_key_returns_same_task_without_second_reserve( + fake_stages: None, client: Any +) -> None: + from apps.worker.report_tasks import _generate_report + + headers, user_id = await _login(client) + request_key = f"rk-{uuid.uuid4().hex}" + first = await _create_report(client, headers, request_key=request_key) + await _generate_report(int(first["task_id"])) # 先让首单成功 + + again = await client.post( + "/api/v1/reports", + headers=headers, + json={ + "questionnaire": {"style": "通勤"}, + "photo_object_key": "report_photo/u1.jpg", + "request_key": request_key, + }, + ) + assert again.status_code == 200 + replay = again.json()["data"] + assert replay["task_id"] == first["task_id"] + assert replay["report_id"] == first["report_id"] + assert replay["status"] == "succeeded" + + # tk_task 幂等键 + reserve 幂等键双保险：只有一笔 reserve。 + rows = await _ledger_rows(user_id, "report_basic") + assert len(rows) == 1 + assert rows[0].status == "committed" + assert await _count_user_reports(user_id) == 1 # 重放不建新报告行 + + +async def _count_user_reports(user_id: int) -> int: + from sqlalchemy import func, 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(func.count()).select_from(RpReport).where(RpReport.user_id == user_id) + ) + return int(result.scalar_one()) + + +# --- 场景 4：首次资格 40920 --------------------------------------------------- + + +async def test_second_first_report_is_rejected_with_40920(fake_stages: None, client: Any) -> None: + from apps.worker.report_tasks import _generate_report + + headers, _ = await _login(client) + data = await _create_report(client, headers) + await _generate_report(int(data["task_id"])) # 首单成功，终身资格已用掉 + + resp = await client.post( + "/api/v1/reports", + headers=headers, + json={ + "questionnaire": {}, + "photo_object_key": "report_photo/u2.jpg", + "request_key": f"rk-{uuid.uuid4().hex}", # 新键：不是幂等重放 + }, + ) + assert resp.status_code == 409 + body = resp.json() + assert body["code"] == 40920 + assert "30" in (body.get("hint") or "") + + +# --- 场景 5：失败退款 + derived_features 保留 ---------------------------------- + + +async def test_stage_failure_refunds_and_preserves_derived_features( + failing_stages: None, client: Any +) -> None: + from apps.worker.report_tasks import _generate_report + from packages.report.pipeline import StageFailureError + + headers, user_id = await _login(client) + data = await _create_report(client, headers) + task_id, report_id = int(data["task_id"]), int(data["report_id"]) + + with pytest.raises(StageFailureError): + await _generate_report(task_id) + + report = await _fetch_report(report_id) + assert report.status == "failed" + # 已完成阶段的产物仍在 + 错误摘要写进 result。 + assert report.result["vision"] == {"features": ["vision_runs"]} + assert report.result["error"]["stage"] == "boom" + + # T-P3-09 核心：视觉中间态保留——重试/重生成不需要重新上传照片。 + rp_input = await _fetch_input(report_id) + assert rp_input.derived_features == {"vision_runs": 1} + + task = await _fetch_task(task_id) + assert task.status == "failed" + + # 退款：consume 流水转 refunded，反向 refund 流水 committed。 + rows = await _ledger_rows(user_id, "report_basic") + assert [(row.tx_type, row.status) for row in rows] == [ + ("consume", "refunded"), + ("refund", "committed"), + ] + + # failed 报告的详情只透错误摘要，不透半成品。 + detail = (await client.get(f"/api/v1/reports/{report_id}", headers=headers)).json()["data"] + assert set(detail["result"]) == {"error"} + assert detail["result"]["error"]["stage"] == "boom" + + +# --- 场景 6：regenerate 走月配额并复用视觉中间态 ------------------------------- + + +async def test_regenerate_uses_month_quota_and_reuses_derived_features( + fake_stages: None, client: Any +) -> None: + from apps.worker.report_tasks import _generate_report + + headers, user_id = await _login(client) + first = await _create_report(client, headers) + await _generate_report(int(first["task_id"])) + + regen = await client.post( + f"/api/v1/reports/{first['report_id']}/regenerate", + headers=headers, + json={"request_key": f"rk-regen-{uuid.uuid4().hex}"}, + ) + assert regen.status_code == 200 + data = regen.json()["data"] + assert data["version"] == 2 + new_report_id = int(data["report_id"]) + assert new_report_id != int(first["report_id"]) + + # 重新生成走 report_regen 月配额（m:YYYY-MM），不复用终身额度。 + rows = await _ledger_rows(user_id, "report_regen") + assert len(rows) == 1 + assert rows[0].status == "reserved" + assert rows[0].period_key.startswith("m:") + + # 新输入行直接带着上一版的视觉中间态（无需重新上传照片）。 + new_input = await _fetch_input(new_report_id) + assert new_input.derived_features == {"vision_runs": 1} + + await _generate_report(int(data["task_id"])) + report_v2 = await _fetch_report(new_report_id) + assert report_v2.status == "succeeded" + assert report_v2.version == 2 + # 成功后视觉仍未重算：vision_runs 计数保持 1（Fake 阶段非空即跳过）。 + input_after = await _fetch_input(new_report_id) + assert input_after.derived_features == {"vision_runs": 1} diff --git a/tests/unit/test_report_pipeline.py b/tests/unit/test_report_pipeline.py new file mode 100644 index 0000000..6cf1239 --- /dev/null +++ b/tests/unit/test_report_pipeline.py @@ -0,0 +1,175 @@ +"""report 流水线编排的纯逻辑单测（不连库）。 + +覆盖任务文档 §4 的两条：assemble_stages 缺失即 CONFIG_ERROR 且列出缺失项； +run_pipeline 顺序执行、产物按键写入、第二阶段抛错 → failed 且已完成阶段 +产物仍在。 +""" + +from __future__ import annotations + +import json + +import pytest + +from packages.report.pipeline import ( + ReportContext, + StageFailureError, + assemble_stages, + run_pipeline, +) +from packages.shared.config import Settings +from packages.shared.errors import AppError, ErrorCode +from packages.shared.models import RpInput, RpReport + + +class FakeSession: + """只满足 flush/commit/rollback 的最小会话——单测不连库。""" + + def __init__(self) -> None: + self.flush_count = 0 + self.commit_count = 0 + + async def flush(self) -> None: + self.flush_count += 1 + + async def commit(self) -> None: + self.commit_count += 1 + + async def rollback(self) -> None: ... + + +def _stage_settings(*paths: str) -> Settings: + """按别名构造阶段配置——字段名带 alias 且未开 populate_by_name， + 用字段名构造会被静默忽略、落回默认值（踩过坑，写下来防再踩）。""" + return Settings(REPORT_PIPELINE_STAGES_JSON=json.dumps(list(paths))) + + +def _make_ctx() -> tuple[ReportContext, FakeSession]: + # 瞬态 ORM 对象：default=dict 只在 INSERT 时生效，这里必须显式给 JSONB 列初值。 + report = RpReport( + report_id=101, + user_id=1, + status="running", + result={}, + model_info={}, + ) + report_input = RpInput( + input_id=11, + report_id=101, + questionnaire={}, + photo_meta={}, + derived_features={}, + ) + session = FakeSession() + ctx = ReportContext( + report=report, + report_input=report_input, + session=session, # type: ignore[arg-type] + settings=Settings(), + ) + return ctx, session + + +class AlphaStage: + name = "alpha" + + async def run(self, ctx: ReportContext) -> None: + ctx.report.result["alpha"] = {"value": 1} + ctx.report_input.derived_features["pose"] = {"kps": 3} + + +class BetaStage: + name = "beta" + + async def run(self, ctx: ReportContext) -> None: + # 顺序契约：后一阶段能读到前一阶段的产物。 + assert ctx.report.result.get("alpha") == {"value": 1} + ctx.report.result["beta"] = {"value": ctx.report.result["alpha"]["value"] + 1} + + +class BoomStage: + name = "boom" + + async def run(self, ctx: ReportContext) -> None: + raise RuntimeError("爆了：模拟阶段失败") + + +async def test_run_pipeline_executes_stages_in_order_and_flushes_each() -> None: + ctx, session = _make_ctx() + + await run_pipeline(ctx, [AlphaStage(), BetaStage()]) + + assert ctx.report.result == {"alpha": {"value": 1}, "beta": {"value": 2}} + assert ctx.report_input.derived_features == {"pose": {"kps": 3}} + # 每个阶段一次 flush：中间态逐段落库（T-P3-09）的结构保证。 + assert session.flush_count == 2 + + +async def test_run_pipeline_failure_keeps_completed_stage_and_marks_failed() -> None: + ctx, session = _make_ctx() + + with pytest.raises(StageFailureError) as excinfo: + await run_pipeline(ctx, [AlphaStage(), BoomStage()]) + + assert excinfo.value.stage_name == "boom" + assert ctx.report.status == "failed" + # 已完成阶段的产物与视觉中间态都还在——重试/重生成不重跑的结构基础。 + assert ctx.report.result["alpha"] == {"value": 1} + assert ctx.report_input.derived_features == {"pose": {"kps": 3}} + # 错误摘要写进 result，且失败结案在事务里先行提交（退款事务纪律）。 + assert ctx.report.result["error"]["stage"] == "boom" + assert ctx.report.result["error"]["type"] == "RuntimeError" + assert session.commit_count == 1 + + +def test_assemble_stages_lists_all_missing_paths() -> None: + settings = _stage_settings( + "packages.report.no_such_module:Nope", + "packages.report.pipeline:NotAClass", + ) + + with pytest.raises(AppError) as excinfo: + assemble_stages(settings) + + assert excinfo.value.code == ErrorCode.CONFIG_ERROR + assert excinfo.value.data == { + "missing": [ + "packages.report.no_such_module:Nope", + "packages.report.pipeline:NotAClass", + ] + } + + +def test_assemble_stages_rejects_empty_config() -> None: + settings = _stage_settings() + + with pytest.raises(AppError) as excinfo: + assemble_stages(settings) + + assert excinfo.value.code == ErrorCode.CONFIG_ERROR + + +class _NotAStage: + """存在但不满足 Stage 协议（没有 run）——必须被当成配置错误拦下。""" + + name = "broken" + + +def test_assemble_stages_rejects_protocol_violations() -> None: + settings = _stage_settings( + "tests.unit.test_report_pipeline:AlphaStage", + "tests.unit.test_report_pipeline:_NotAStage", + ) + + with pytest.raises(AppError) as excinfo: + assemble_stages(settings) + + assert excinfo.value.code == ErrorCode.CONFIG_ERROR + assert excinfo.value.data == {"missing": ["tests.unit.test_report_pipeline:_NotAStage"]} + + +def test_assemble_stages_imports_from_any_module_path() -> None: + """dotted path 装配对本模块的 Fake 阶段同样生效（importlib 动态导入路径）。""" + settings = _stage_settings("tests.unit.test_report_pipeline:AlphaStage") + stages = assemble_stages(settings) + assert [s.name for s in stages] == ["alpha"]