"""报告域服务：配额预扣、版本化与幂等（T-P3-07 / T-P3-09）。 事务纪律：reserve 与建报告行、建 tk_task 在**同一事务**；commit 之后才投递 队列——队列消息不能早于数据落库，否则 worker 可能读到一个不存在的报告。 幂等是双保险：tk_task 的幂等键（= request_key）挡重复任务行，reserve 的 幂等键挡重复预扣；服务层再加一次「重放优先」快捷路径，让成功后的同键 重试拿到 200 而不是重新创建。 已生成过报告的用户允许再次生成（产品决策变更）：不再按「已有 succeeded 报告」硬拒（原 40920 终身锁死），只由 report_basic 终身配额兜底——额度 耗尽时报 40301 而非 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.storage.local import require_local_upload 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" #: 深度诊断（全面报告）走日配额 full_report（迁移 0003 SSOT）。 QUOTA_KEY_FULL = "full_report" TIER_BASIC = "basic" TIER_FULL = "full" 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 重试不是「再生成一次」， # 必须先于任何校验按原任务返回（否则会把幂等重试误判成新请求）。 replay = await self._find_replay(request_key=request_key, user_id=user_id) if replay is not None: return replay # 不再按「已有 succeeded 报告」终身锁死（产品决策变更）：重复生成由 # report_basic 终身配额兜底，额度耗尽由 BillingService 报 40301。 # 照片存在性校验（本地磁盘降级路线）：key 必须落在 # upload_dir/report_photo/<本人>/ 下且文件真实存在——模拟 key 与越界 # key 一律 404。校验在配额预扣之前，不合格的照片不消耗额度。 # regenerate 不复核：它复用的是上一版已验过的 photo_meta（T-P3-09）。 require_local_upload( self._settings, user_id=user_id, scene="report_photo", object_key=photo_object_key ) 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 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: 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)