"""catalog_transfer Admin API：挂载到 /api/admin/catalog-transfer。 安全：系统加载 + 租户启用（require_plugin）、商品权限（require_permission）、 任务/下载再次按租户校验、单租户同类任务行锁串行。 """ from __future__ import annotations import os from fastapi import APIRouter, Depends, HTTPException, UploadFile, File, Query from fastapi.responses import FileResponse from sqlalchemy import select, text from sqlalchemy.ext.asyncio import AsyncSession from app.api.deps import get_db, get_admin_user, require_permission, require_any_permission from app.core.models.user import User from app.core.models.tenant import Tenant from app.core.models.task import AsyncTaskLog from app.core.services.plugin_helper import require_plugin from . import exporter, tasks from .schemas import ExportRequest, EstimateResult, TaskResponse router = APIRouter(prefix="/catalog-transfer", tags=["catalog-transfer"]) PLUGIN = "catalog_transfer" _ACTIVE = ("pending", "running") STALE_MINUTES = 30 # 超过此时长的活跃任务视为僵尸，不再阻塞新任务 MAX_UPLOAD = 200 * 1024 * 1024 # 上传前压缩大小上限，防止 Web 进程内存耗尽 def _dt(v): return v.isoformat() if v else None def _serialize(row: AsyncTaskLog) -> dict: return { "task_id": row.id, "task_type": row.task_type, "status": row.status, "payload": {k: v for k, v in (row.payload or {}).items() if k != "upload_path"}, "result": {k: v for k, v in (row.result or {}).items() if k != "download_path"}, "created_at": _dt(getattr(row, "created_at", None)), "completed_at": _dt(getattr(row, "completed_at", None)), } async def _create_task_locked(db: AsyncSession, tenant_id: int, task_type: str, payload: dict) -> int: """租户行锁内检查同类活跃任务并插入，避免“先查再插”竞态。 仅最近 STALE_MINUTES 内的活跃任务才阻塞；更久的 pending/running 视为僵尸 （worker 崩溃/未消费），不再永久锁死功能。用 DB 自身时钟比较，规避时区歧义。 ponytail: MySQL 语法；换库需调整 INTERVAL 表达式。 """ await db.execute(select(Tenant.id).where(Tenant.id == tenant_id).with_for_update()) existing = (await db.execute( select(AsyncTaskLog.id).where( AsyncTaskLog.tenant_id == tenant_id, AsyncTaskLog.task_type == task_type, AsyncTaskLog.status.in_(_ACTIVE), AsyncTaskLog.created_at > text(f"NOW() - INTERVAL {STALE_MINUTES} MINUTE"), ).limit(1) )).scalar_one_or_none() if existing is not None: await db.rollback() raise HTTPException(409, {"error": "同类任务已在运行", "task_id": existing}) row = AsyncTaskLog(tenant_id=tenant_id, task_type=task_type, status="pending", payload=payload) db.add(row) await db.commit() await db.refresh(row) return row.id async def _dispatch(db: AsyncSession, task_id: int, celery_task, **kwargs) -> None: """投递 Celery（task_id 同时用于失败标记与转发给任务，避免与关键字撞名）； 失败时把已提交的 pending 任务标 failed，避免永久占用 409。""" try: celery_task.delay(task_id=task_id, **kwargs) except Exception as exc: row = await db.get(AsyncTaskLog, task_id) if row is not None: row.status = "failed" row.result = {"error": f"celery dispatch failed: {str(exc)[:200]}"} await db.commit() raise HTTPException(503, str(exc)[:300]) from exc async def _read_capped(file: UploadFile) -> bytes: """分块读取上传文件，超过 MAX_UPLOAD 立即拒绝，避免整体灌入内存。""" buf = bytearray() while True: chunk = await file.read(1024 * 1024) if not chunk: break buf.extend(chunk) if len(buf) > MAX_UPLOAD: raise HTTPException(413, "上传文件过大") return bytes(buf) # ── 导出 ──────────────────────────────────────────────────────────── @router.post("/exports/estimate", response_model=EstimateResult, summary="导出估算") async def estimate(body: ExportRequest, db: AsyncSession = Depends(get_db), current_user: User = Depends(require_permission("products.view"))): await require_plugin(PLUGIN, db, current_user.tenant_id) return await exporter.estimate(db, current_user.tenant_id, body) @router.post("/exports", response_model=TaskResponse, summary="创建导出任务") async def create_export(body: ExportRequest, db: AsyncSession = Depends(get_db), current_user: User = Depends(require_permission("products.view"))): await require_plugin(PLUGIN, db, current_user.tenant_id) tasks.cleanup_expired() task_id = await _create_task_locked( db, current_user.tenant_id, "catalog_transfer.export", {"user_id": current_user.id, "scope": body.scope, "rows_per_file": body.rows_per_file}, ) await _dispatch(db, task_id, tasks.export_task, tenant_id=current_user.tenant_id, request=body.model_dump()) return {"task_id": task_id} # ── 回导 ──────────────────────────────────────────────────────────── @router.post("/imports", response_model=TaskResponse, summary="上传并创建回导预检任务") async def create_import(file: UploadFile = File(...), db: AsyncSession = Depends(get_db), current_user: User = Depends(require_any_permission(["products.create", "products.update"]))): await require_plugin(PLUGIN, db, current_user.tenant_id) name = (file.filename or "").lower() if not (name.endswith(".xlsx") or name.endswith(".zip")): raise HTTPException(422, "仅支持插件生成的 .xlsx 或 .zip") data = await _read_capped(file) tasks.cleanup_expired() task_id = await _create_task_locked( db, current_user.tenant_id, "catalog_transfer.import_preflight", {"user_id": current_user.id, "filename": file.filename}, ) work_dir = tasks.task_dir(task_id) os.makedirs(work_dir, exist_ok=True) upload_path = os.path.join(work_dir, "input" + (".zip" if name.endswith(".zip") else ".xlsx")) with open(upload_path, "wb") as f: f.write(data) row = await db.get(AsyncTaskLog, task_id) row.payload = {**(row.payload or {}), "upload_path": upload_path} await db.commit() await _dispatch(db, task_id, tasks.preflight_task, tenant_id=current_user.tenant_id) return {"task_id": task_id} @router.post( "/imports/{task_id}/commit", response_model=TaskResponse, summary="确认后创建回导写入任务", # 回导会同时创建与更新商品，须同时具备两种权限，避免单一权限越权 dependencies=[Depends(require_permission("products.create")), Depends(require_permission("products.update"))], ) async def commit_import(task_id: int, create_missing: bool = Query(False), db: AsyncSession = Depends(get_db), current_user: User = Depends(get_admin_user)): await require_plugin(PLUGIN, db, current_user.tenant_id) pre = await db.get(AsyncTaskLog, task_id) if pre is None or pre.tenant_id != current_user.tenant_id or pre.task_type != "catalog_transfer.import_preflight": raise HTTPException(404, "预检任务不存在") report = (pre.result or {}).get("preflight") or {} if not report.get("can_commit"): raise HTTPException(409, "预检未通过，不能提交") # 存在缺失分类/品牌且未勾选“创建缺失项”时拒绝，避免静默清空现有品牌/分类 if not create_missing and (report.get("missing_categories") or report.get("missing_brands")): raise HTTPException(409, { "error": "存在缺失分类/品牌，需勾选“创建缺失项”后才能提交", "missing_categories": report.get("missing_categories") or [], "missing_brands": report.get("missing_brands") or [], }) payload = pre.payload or {} commit_id = await _create_task_locked( db, current_user.tenant_id, "catalog_transfer.import_commit", {"user_id": current_user.id, "filename": payload.get("filename"), "upload_path": payload.get("upload_path"), "source_preflight": task_id}, ) await _dispatch(db, commit_id, tasks.commit_task, tenant_id=current_user.tenant_id, user_id=current_user.id, create_missing=create_missing) return {"task_id": commit_id} # ── 任务查询与下载 ─────────────────────────────────────────────────── @router.get("/tasks", summary="列出当前租户任务") async def list_tasks(limit: int = Query(20, ge=1, le=100), db: AsyncSession = Depends(get_db), current_user: User = Depends(get_admin_user)): await require_plugin(PLUGIN, db, current_user.tenant_id) rows = (await db.execute( select(AsyncTaskLog).where( AsyncTaskLog.tenant_id == current_user.tenant_id, AsyncTaskLog.task_type.like("catalog_transfer.%"), ).order_by(AsyncTaskLog.id.desc()).limit(limit) )).scalars().all() return {"items": [_serialize(r) for r in rows]} @router.get("/tasks/{task_id}", summary="查询任务状态") async def get_task(task_id: int, db: AsyncSession = Depends(get_db), current_user: User = Depends(get_admin_user)): await require_plugin(PLUGIN, db, current_user.tenant_id) row = await db.get(AsyncTaskLog, task_id) if row is None or row.tenant_id != current_user.tenant_id or not row.task_type.startswith("catalog_transfer."): raise HTTPException(404, "任务不存在") return _serialize(row) @router.get("/tasks/{task_id}/download", summary="下载导出结果") async def download_task(task_id: int, db: AsyncSession = Depends(get_db), current_user: User = Depends(require_permission("products.view"))): await require_plugin(PLUGIN, db, current_user.tenant_id) tasks.cleanup_expired() row = await db.get(AsyncTaskLog, task_id) if row is None or row.tenant_id != current_user.tenant_id or row.task_type != "catalog_transfer.export": raise HTTPException(404, "任务不存在") if row.status != "success": raise HTTPException(409, "任务尚未完成") result = row.result or {} path = result.get("download_path") if not path or not os.path.exists(path): raise HTTPException(410, "文件已过期或不存在") return FileResponse(path, filename=result.get("download_name") or os.path.basename(path), media_type="application/octet-stream")