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 造数
184 lines
8.1 KiB
Python
184 lines
8.1 KiB
Python
"""报告 PDF 导出的 celery 任务（name=``meida.report.export``，cpu 队列）。
|
||
|
||
只做「取产物 → 渲染 → 落盘 → 回写 rendered_asset_key」，不重算内容。
|
||
配额三段式与报告生成一致：成功 commit、失败 refund（FR-BILL-05），
|
||
预扣凭 ``tk_task.reserve_tx_id`` 找回（复用 pipeline.load_reservation）。
|
||
|
||
渲染是同步 CPU 工作（WeasyPrint 没有 async API），在 --pool=solo 的 cpu
|
||
worker 里直接执行——单份报告渲染耗时为秒级，可接受；未来上量再挪线程池。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from pathlib import Path
|
||
from typing import Any, cast
|
||
|
||
from apps.worker import loop as worker_loop
|
||
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.report import repo as report_repo
|
||
from packages.report.pdf.renderer import ReportPdfMeta, render_report_pdf
|
||
from packages.report.pipeline import load_reservation
|
||
from packages.shared.config import 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.task import repo as task_repo
|
||
from packages.task.service import TaskService
|
||
|
||
logger = get_logger(__name__)
|
||
|
||
|
||
class ExportRejectedError(Exception):
|
||
"""业务性拒绝（如报告不再是 succeeded）：已在任务内结案并退款，重跑也不会好。
|
||
|
||
与流水线的 StageFailureError 同义：celery 包装层据类型区分「业务失败
|
||
不重试」与「基础设施异常值得重试」。
|
||
"""
|
||
|
||
|
||
def _write_pdf(export_dir: str, filename: str, pdf_bytes: bytes) -> Path:
|
||
"""落盘导出文件（同步：pathlib 阻塞方法不进 async 函数，ASYNC240）。
|
||
|
||
返回含导出目录前缀的相对路径——rendered_asset_key 按此入库，下载侧
|
||
按同一 cwd 解析，且整体迁移 var/ 目录时不用改库。
|
||
"""
|
||
directory = Path(export_dir)
|
||
directory.mkdir(parents=True, exist_ok=True)
|
||
path = directory / filename
|
||
path.write_bytes(pdf_bytes)
|
||
return path
|
||
|
||
|
||
async def _export_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:
|
||
# 重复投递/幂等重放的守卫：已结案的任务不再执行，防重复 commit 配额。
|
||
logger.info("report_export_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)
|
||
if report is None:
|
||
raise AppError(ErrorCode.STATE_CONFLICT, message="报告数据不完整")
|
||
|
||
service = TaskService(session)
|
||
# 与报告生成同一纪律：重试入站先归一 QUEUED 再进 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="渲染中")
|
||
await session.commit() # RUNNING 先落库：前端进度读它
|
||
|
||
if report.status != report_repo.STATUS_SUCCEEDED:
|
||
# 发起导出后报告被重生成顶掉/置失败等：业务性拒绝，结案退款不重试。
|
||
await _reject_export(task_id, "报告当前状态不可导出")
|
||
raise ExportRejectedError(f"report {report_id} status={report.status}")
|
||
|
||
pdf_bytes = render_report_pdf(
|
||
report.result or {},
|
||
meta=ReportPdfMeta(
|
||
report_id=report.report_id,
|
||
version=report.version,
|
||
generated_at=report.completed_at or report.created_at,
|
||
),
|
||
)
|
||
filename = f"report_{report.report_id}_v{report.version}.pdf"
|
||
asset_key = _write_pdf(settings.report_export_dir, filename, pdf_bytes).as_posix()
|
||
|
||
report.rendered_asset_key = asset_key
|
||
reservation = await load_reservation(session, task_id)
|
||
if reservation is not None:
|
||
await BillingService(session).commit(reservation)
|
||
await service.transition(
|
||
task_id=task_id,
|
||
target=TaskStatus.SUCCEEDED,
|
||
stage_text="已完成",
|
||
result={
|
||
"report_id": str(report.report_id),
|
||
"version": report.version,
|
||
"asset_key": asset_key,
|
||
},
|
||
)
|
||
await session.commit()
|
||
logger.info(
|
||
"report_export_succeeded",
|
||
report_id=report_id,
|
||
task_id=task_id,
|
||
asset_key=asset_key,
|
||
pdf_bytes=len(pdf_bytes),
|
||
)
|
||
return {"task_id": str(task_id), "report_id": str(report_id), "status": "succeeded"}
|
||
|
||
|
||
async def _reject_export(task_id: int, message: str) -> None:
|
||
"""业务性拒绝的结案：task failed + 退款，提交后才返回（纪律同流水线失败结案）。"""
|
||
factory = get_session_factory()
|
||
async with factory() as session:
|
||
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_export_rejected")
|
||
await session.commit()
|
||
|
||
|
||
async def _fail_export(task_id: int, message: str) -> None:
|
||
"""基础设施异常重试耗尽后的兜底结案：failed + 退款（尽力而为）。"""
|
||
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
|
||
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_export_failed")
|
||
await session.commit()
|
||
except Exception:
|
||
logger.exception("report_export_fail_settle_failed", task_id=task_id)
|
||
|
||
|
||
# Celery 的 task 装饰器没有类型标注，会把被装饰函数擦成 untyped
|
||
@celery_app.task(name="meida.report.export", bind=True, max_retries=1) # type: ignore[untyped-decorator]
|
||
def export_report_pdf(self: Any, task_id: int) -> dict[str, Any]:
|
||
"""cpu 队列的报告 PDF 导出入口。
|
||
|
||
重试语义与报告生成对齐：基础设施类异常重试一次；业务性拒绝
|
||
（ExportRejectedError）已在任务内结案，重跑也不会好，不再重试。
|
||
"""
|
||
bind_context(task_id=task_id)
|
||
try:
|
||
return cast("dict[str, Any]", worker_loop.run(_export_report(task_id)))
|
||
except ExportRejectedError as exc:
|
||
logger.warning("report_export_rejected", task_id=task_id, error=str(exc))
|
||
raise
|
||
except Exception as exc:
|
||
if self.request.retries < self.max_retries:
|
||
logger.warning("report_export_retry", task_id=task_id, error=str(exc))
|
||
raise self.retry(exc=exc, countdown=10) from exc
|
||
logger.exception("report_export_failed_after_retry", task_id=task_id)
|
||
worker_loop.run(_fail_export(task_id, f"{type(exc).__name__}: {exc}"))
|
||
raise
|