338 lines
14 KiB
Python
338 lines
14 KiB
Python
"""报告域服务：配额预扣、版本化与幂等（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)
|