ozon-seller-kit/server/api/publish.py

"""发布端点：提交 ImportProductsV3 + 后台轮询回填。"""
from __future__ import annotations

import asyncio
from datetime import datetime, timezone
from uuid import UUID

from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException
from pydantic import BaseModel
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from core.security import decrypt_secret
from db import get_db, get_session_factory
from deps import get_current_user
from models import Product, PublishTask, Shop
from models.enums import PublishStatus, Stage
from services.ozon_client import OzonClient, OzonAPIError
from services.publish import build_import_item, validate_ready

router = APIRouter(prefix="/api", tags=["publish"])


class PublishRequest(BaseModel):
    shop_id: str


def _client(shop: Shop) -> OzonClient:
    return OzonClient(decrypt_secret(shop.client_id_enc), decrypt_secret(shop.api_key_enc))


@router.post("/products/{product_id}/publish")
async def publish_product(
    product_id: str,
    body: PublishRequest,
    background: BackgroundTasks,
    db: AsyncSession = Depends(get_db),
    _user: dict = Depends(get_current_user),
):
    product = await db.get(Product, UUID(product_id))
    if product is None:
        raise HTTPException(status_code=404, detail="商品不存在")
    shop = await db.get(Shop, UUID(body.shop_id))
    if shop is None:
        raise HTTPException(status_code=404, detail="店铺不存在")

    missing = validate_ready(product)
    if missing:
        raise HTTPException(status_code=422, detail=f"缺少必填项：{'、'.join(missing)}")

    item = build_import_item(product)
    client = _client(shop)
    try:
        result = await client.post("/v3/product/import", {"items": [item]})
    except OzonAPIError as exc:
        raise HTTPException(status_code=502, detail=exc.detail)

    task_id = (result.get("result") or {}).get("task_id")
    if not task_id:
        raise HTTPException(status_code=502, detail=f"Ozon 未返回 task_id：{result}")

    task = PublishTask(
        product_id=product.id,
        shop_id=shop.id,
        ozon_task_id=int(task_id),
        status=PublishStatus.pending,
        request_payload=item,
    )
    db.add(task)
    product.stage = Stage.publishing
    await db.commit()
    await db.refresh(task)

    background.add_task(_poll, str(task.id))
    return {"task_id": str(task.id), "ozon_task_id": task.ozon_task_id}


async def _poll(task_id: str) -> None:
    """后台轮询 import/info，直到 imported / failed 或超时（约 40s）。"""
    async with get_session_factory()() as db:
        task = await db.get(PublishTask, UUID(task_id))
        if task is None:
            return
        shop = await db.get(Shop, task.shop_id)
        product = await db.get(Product, task.product_id)
        if shop is None or product is None:
            return
        client = _client(shop)

        for attempt in range(8):
            try:
                result = await client.post("/v1/product/import/info", {"task_id": task.ozon_task_id})
            except OzonAPIError as exc:
                task.status = PublishStatus.failed
                task.errors = [{"error": exc.detail}]
                task.completed_at = datetime.now(timezone.utc)
                product.stage = Stage.failed
                await db.commit()
                return

            items = (result.get("result") or {}).get("items") or []
            item = items[0] if items else {}
            status = item.get("status", "")
            product_id = item.get("product_id")
            errors = item.get("errors") or []

            if status == "imported":
                task.status = PublishStatus.imported
                task.response = item
                task.completed_at = datetime.now(timezone.utc)
                if product_id:
                    product.ozon_product_id = int(product_id)
                product.stage = Stage.published
                product.published_at = datetime.now(timezone.utc)
                await db.commit()
                return

            if status == "failed":
                task.status = PublishStatus.failed
                task.errors = errors
                task.response = item
                task.completed_at = datetime.now(timezone.utc)
                product.stage = Stage.failed
                await db.commit()
                return

            # pending / moderation → 继续等
            task.status = PublishStatus.moderation if status in ("moderating", "moderation") else PublishStatus.processing
            if product_id:
                product.ozon_product_id = int(product_id)
            await db.commit()
            await asyncio.sleep(5 * (attempt + 1))

        # 超时未定：保留 processing，前端可刷新
        task.status = PublishStatus.moderation
        task.response = item
        await db.commit()


@router.get("/publish/{task_id}")
async def get_publish_task(
    task_id: str,
    db: AsyncSession = Depends(get_db),
    _user: dict = Depends(get_current_user),
):
    task = await db.get(PublishTask, UUID(task_id))
    if task is None:
        raise HTTPException(status_code=404, detail="发布任务不存在")
    return {
        "id": str(task.id),
        "product_id": str(task.product_id),
        "shop_id": str(task.shop_id),
        "ozon_task_id": task.ozon_task_id,
        "status": task.status.value,
        "errors": task.errors,
        "response": task.response,
        "created_at": task.created_at,
        "completed_at": task.completed_at,
    }


@router.get("/products/{product_id}/publish-history")
async def publish_history(
    product_id: str,
    db: AsyncSession = Depends(get_db),
    _user: dict = Depends(get_current_user),
):
    rows = (await db.scalars(
        select(PublishTask)
        .where(PublishTask.product_id == UUID(product_id))
        .order_by(PublishTask.created_at.desc())
    )).all()
    return [
        {
            "id": str(t.id),
            "ozon_task_id": t.ozon_task_id,
            "status": t.status.value,
            "errors": t.errors,
            "created_at": t.created_at,
            "completed_at": t.completed_at,
        }
        for t in rows
    ]