xsl
cd0ddb131b
- POST /reports/{id}/export：request_key 幂等 + hd_export 日配额三段式（reserve/commit/refund）+ tier2 熔断场景闸（BudgetGuardDep），仅 succeeded 报告可导
- GET /reports/{id}/export/download：归属校验 + 路径收敛（rendered_asset_key 必须落在 report_export_dir 内），FileResponse 携带 UTF-8 文件名
- celery 任务 meida.report.export（cpu 队列）：渲染→落盘 var/exports→回写 rendered_asset_key；业务性拒绝（报告状态变化）结案退款不重试，基础设施异常重试一次后兜底结案退款
- packages/report/pdf：Jinja2 模板 + WeasyPrint 渲染器（url_fetcher 全拒外部资源防 SSRF；维度中文名按 DIMENSION_BY_KEY 回填；色板维度渲染色卡）
- 单测 10 例（资格/幂等/并发同键/下载校验/渲染）；scripts/seed_pdf_e2e.sql 供本地 E2E 造数
209 lines
8.6 KiB
Python
209 lines
8.6 KiB
Python
"""报告 PDF 导出的域服务（Phase A）：资格、幂等、配额预扣与任务投递。
|
|
|
|
与报告生成的差别：导出是「读产物 → 渲染 → 落盘」的纯加工，不重算内容，
|
|
因此不建新报告行，只建 tk_task（type=render）并在成功后回写
|
|
``rp_report.rendered_asset_key``。
|
|
|
|
事务纪律与 ReportService 一致：reserve 与建任务同事务，commit 之后才投递；
|
|
幂等同样是双保险——tk_task 幂等键挡重复任务行，reserve 幂等键挡重复预扣，
|
|
服务层重放快捷路径让同键重试拿到 200。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Callable
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Protocol
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from packages.billing.service import BillingService
|
|
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 RpReport, Task
|
|
from packages.task import repo as task_repo
|
|
from packages.task.service import TaskService
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
#: 导出走 hd_export 日配额（迁移 0003 种子：day / base 1 / hard_cap 3）；
|
|
#: 同一字符串也是熔断档位拦截的场景名（tier2 起被 block，见架构 §13.3）。
|
|
QUOTA_KEY_EXPORT = "hd_export"
|
|
EXPORT_REF_TYPE = "report_export"
|
|
EXPORT_ETA_SECONDS = 30
|
|
|
|
_TERMINAL_TASK_STATUSES = frozenset(
|
|
{TaskStatus.SUCCEEDED.value, TaskStatus.FAILED.value, TaskStatus.CANCELED.value}
|
|
)
|
|
|
|
|
|
class SceneGuard(Protocol):
|
|
"""熔断场景闸的最小接口（DailyBudgetGuard 结构匹配）。
|
|
|
|
声明 Protocol 而非直接依赖 packages.llm.budget：单测可以塞假实现，
|
|
服务单测不必拉起 Redis。
|
|
"""
|
|
|
|
async def ensure_scene_allowed(self, *, scene: str, critical: bool = False) -> None: ...
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ExportSubmission:
|
|
"""一次导出提交的结果；幂等重放返回同值。"""
|
|
|
|
task_id: int
|
|
report_id: int
|
|
status: str
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ExportFile:
|
|
"""下载就绪的导出文件：磁盘路径 + 给客户端的展示文件名。"""
|
|
|
|
path: Path
|
|
filename: str
|
|
|
|
|
|
def _submission_of(task: Task) -> ExportSubmission:
|
|
"""从 tk_task 行重建导出提交结果（幂等重放路径）。"""
|
|
raw_report_id = task.payload.get("report_id")
|
|
if not isinstance(raw_report_id, str | int):
|
|
raise AppError(ErrorCode.IDEMPOTENCY_CONFLICT)
|
|
return ExportSubmission(task_id=task.task_id, report_id=int(raw_report_id), status=task.status)
|
|
|
|
|
|
def _resolve_export_path(export_dir: str, asset_key: str) -> Path:
|
|
"""同步解析导出文件路径（pathlib 的阻塞方法不进 async 函数，ASYNC240）。
|
|
|
|
key 由 worker 写入，越界即数据被篡改——按服务器错误处理而非 404；
|
|
行已回写但文件丢了（磁盘清理等），对用户来说等于还没生成。
|
|
"""
|
|
export_root = Path(export_dir).resolve()
|
|
path = Path(asset_key).resolve()
|
|
if not path.is_relative_to(export_root):
|
|
raise AppError(ErrorCode.INTERNAL_ERROR, message="导出文件路径异常")
|
|
if not path.is_file():
|
|
raise AppError(ErrorCode.REPORT_EXPORT_NOT_READY)
|
|
return path
|
|
|
|
|
|
class ReportExportService:
|
|
def __init__(
|
|
self,
|
|
session: AsyncSession,
|
|
*,
|
|
settings: Settings,
|
|
scene_guard: SceneGuard | None = None,
|
|
enqueue: Callable[[int], None] | None = None,
|
|
) -> None:
|
|
self._session = session
|
|
self._settings = settings
|
|
# scene_guard 为 None 时跳过熔断（单测直调）；API 装配层总是注入真实现。
|
|
self._scene_guard = scene_guard
|
|
self._enqueue = enqueue
|
|
self._billing = BillingService(
|
|
session, tz_offset_hours=settings.llm_budget_timezone_offset_hours
|
|
)
|
|
self._tasks = TaskService(session)
|
|
|
|
# -- 发起导出 ------------------------------------------------------------
|
|
|
|
async def create_export(
|
|
self, *, user_id: int, report_id: int, request_key: str
|
|
) -> ExportSubmission:
|
|
"""预扣 hd_export + 建 render 任务 + commit 后投递；同 request_key 重放返回同一任务。"""
|
|
replay = await self._find_replay(
|
|
request_key=request_key, user_id=user_id, report_id=report_id
|
|
)
|
|
if replay is not None:
|
|
return replay
|
|
|
|
report = await self.get_report(user_id=user_id, report_id=report_id)
|
|
if report.status != repo.STATUS_SUCCEEDED:
|
|
raise AppError(ErrorCode.STATE_CONFLICT, message="只有已完成的报告可以导出 PDF")
|
|
|
|
# 熔断先于预扣：档位拦截时用户额度原样保留（架构 §13.3）。
|
|
if self._scene_guard is not None:
|
|
await self._scene_guard.ensure_scene_allowed(scene=QUOTA_KEY_EXPORT)
|
|
|
|
reservation = await self._billing.reserve(
|
|
user_id=user_id,
|
|
quota_key=QUOTA_KEY_EXPORT,
|
|
request_key=request_key,
|
|
amount=1,
|
|
reason=f"export report v{report.version} pdf",
|
|
ref_type=EXPORT_REF_TYPE,
|
|
ref_id=str(report.report_id),
|
|
)
|
|
task = await self._tasks.submit(
|
|
user_id=user_id,
|
|
task_type=TaskType.RENDER,
|
|
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=EXPORT_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()
|
|
if self._enqueue is not None and task.status not in _TERMINAL_TASK_STATUSES:
|
|
self._enqueue(task.task_id)
|
|
logger.info(
|
|
"report_export_created",
|
|
user_id=user_id,
|
|
report_id=report_id,
|
|
task_id=task.task_id,
|
|
)
|
|
return ExportSubmission(task.task_id, report.report_id, task.status)
|
|
|
|
# -- 下载 -----------------------------------------------------------------
|
|
|
|
async def get_export_file(self, *, user_id: int, report_id: int) -> ExportFile:
|
|
"""已生成校验 + 路径收敛：rendered_asset_key 只允许落在导出目录内。"""
|
|
report = await self.get_report(user_id=user_id, report_id=report_id)
|
|
if not report.rendered_asset_key:
|
|
raise AppError(ErrorCode.REPORT_EXPORT_NOT_READY)
|
|
path = _resolve_export_path(self._settings.report_export_dir, report.rendered_asset_key)
|
|
return ExportFile(path=path, filename=f"美搭形象报告-v{report.version}.pdf")
|
|
|
|
# -- 查询 -----------------------------------------------------------------
|
|
|
|
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 _find_replay(
|
|
self, *, request_key: str, user_id: int, report_id: int
|
|
) -> ExportSubmission | None:
|
|
"""同 request_key 的 RENDER 任务已存在时重建提交结果。"""
|
|
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.RENDER.value:
|
|
# 幂等键全局唯一，撞上了其它用户/其它功能的关键字：按幂等冲突拒绝。
|
|
raise AppError(ErrorCode.IDEMPOTENCY_CONFLICT)
|
|
submission = _submission_of(existing)
|
|
if submission.report_id != report_id:
|
|
# 同键不同报告：不是同一次请求的重放。
|
|
raise AppError(ErrorCode.IDEMPOTENCY_CONFLICT)
|
|
return submission
|