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

"""采集入库：插件上传文本 + 图片 URL，落库后异步转存。"""
from __future__ import annotations

import re
from uuid import UUID

from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, UploadFile, File, Form
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from db import get_db, get_session_factory
from deps import get_current_user
from models import Product, ProductAsset
from models.enums import AssetStatus, Stage
from schemas.collection import MaterialsRequest, MaterialsResponse, TextMaterial

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


def _parse_number(text: str | None) -> float | None:
    """'1 290 ₽' / '3.5 кг' / '48*18*25' → 1290.0 / 3.5 / 48"""
    if not text:
        return None
    m = re.search(r"(\d+(?:[.,]\d+)?)", text.replace(" ", "").replace(",", "."))
    return float(m.group(1)) if m else None


def _find_param(pairs: list[dict] | None, keys: list[str]) -> str | None:
    for p in pairs or []:
        k = (p.get("key") or "").lower()
        if any(kw in k for kw in keys):
            return p.get("value")
    return None


def _apply_texts(product: Product, texts: list[TextMaterial]) -> None:
    raw = dict(product.raw or {})
    raw_texts: list[dict] = list(raw.get("texts") or [])
    for t in texts:
        raw_texts.append({"kind": t.kind, "content": t.content, "pairs": t.pairs})
        if t.kind == "title" and t.content and not product.name:
            product.name = t.content
            raw["title"] = t.content
        elif t.kind == "price":
            raw["price"] = t.content
            num = _parse_number(t.content)
            if num is not None and (product.price is None or product.price == 0):
                product.price = num
        elif t.kind == "params":
            raw["params"] = t.pairs
            _apply_weight_dims(product, t.pairs)
        elif t.kind == "selling_point":
            raw["sellingPoints"] = t.content
        elif t.kind == "desc":
            raw["desc"] = t.content
            if not product.description:
                product.description = t.content
        elif t.kind == "brand":
            raw["brand"] = t.content
    raw["texts"] = raw_texts
    product.raw = raw


def _apply_weight_dims(product: Product, pairs: list[dict] | None) -> None:
    """从参数表里解析「包装重量 / 包装尺寸(长宽高)」，统一换算成克 / 毫米回填。"""
    weight = _find_param(pairs, ["包装重量", "重量", "вес"])
    if weight is not None:
        num = _parse_number(weight)
        if num is not None:
            is_kg = any(u in weight.lower() for u in ("кг", "kg"))
            product.weight = num * 1000 if is_kg else num  # 统一为克
            product.weight_unit = "g"

    l = _find_param(pairs, ["包装长度", "长度", "длина"])
    w = _find_param(pairs, ["包装宽度", "宽度", "ширина"])
    h = _find_param(pairs, ["包装高度", "高度", "высота"])
    if l or w or h:
        combined = (l or "") + (w or "") + (h or "")
        factor = 1 if any(u in combined.lower() for u in ("мм", "mm")) else 10  # 厘米→毫米
        product.depth = (_parse_number(l) or 0) * factor if l else None
        product.width = (_parse_number(w) or 0) * factor if w else None
        product.height = (_parse_number(h) or 0) * factor if h else None
        product.dimension_unit = "mm"
    else:
        dim = _find_param(pairs, ["包装尺寸", "размер", "габарит", "尺寸"])
        if dim is not None:
            nums = re.findall(r"\d+(?:[.,]\d+)?", dim.replace(",", "."))
            if len(nums) >= 3:
                factor = 1 if any(u in dim.lower() for u in ("мм", "mm")) else 10
                product.depth = float(nums[0]) * factor
                product.width = float(nums[1]) * factor
                product.height = float(nums[2]) * factor
                product.dimension_unit = "mm"


async def _get_or_create_product(db: AsyncSession, req: MaterialsRequest) -> Product:
    if req.product_id:
        product = await db.get(Product, UUID(req.product_id))
        if product is None:
            raise HTTPException(status_code=404, detail="商品不存在")
        return product
    product = Product(
        stage=Stage.collected,
        source_platform=req.source.platform,
        source_item_id=req.source.itemId,
        source_url=req.source.url,
    )
    db.add(product)
    await db.flush()
    return product


@router.post("/materials", response_model=MaterialsResponse)
async def create_materials(
    req: MaterialsRequest,
    background: BackgroundTasks,
    db: AsyncSession = Depends(get_db),
    _user: dict = Depends(get_current_user),
) -> MaterialsResponse:
    product = await _get_or_create_product(db, req)
    _apply_texts(product, req.texts)

    # 采集溯源（追加来源）
    if not product.source_url:
        product.source_url = req.source.url
    if not product.source_platform:
        product.source_platform = req.source.platform

    # 去重 + 建素材
    existing = set()
    if req.images:
        rows = (await db.execute(
            select(ProductAsset.dedupe_key).where(
                ProductAsset.product_id == product.id,
                ProductAsset.dedupe_key.isnot(None),
            )
        )).scalars().all()
        existing = {k for k in rows if k}

    queued, skipped = 0, 0
    for img in req.images:
        if img.dedupeKey and img.dedupeKey in existing:
            skipped += 1
            continue
        db.add(ProductAsset(
            product_id=product.id,
            group_key=img.groupKey,
            variant_name=img.variantName,
            sort_order=img.index,
            type=img.type,
            source_url=img.url,
            status=AssetStatus.pending,
            dedupe_key=img.dedupeKey,
        ))
        if img.dedupeKey:
            existing.add(img.dedupeKey)
        queued += 1

    # 更新分组计数
    counts: dict = {}
    for a in await db.scalars(select(ProductAsset).where(ProductAsset.product_id == product.id)):
        counts[a.group_key] = counts.get(a.group_key, 0) + 1
    product.asset_counts = counts
    product.stage = Stage.collected if product.stage == Stage.collected else product.stage

    await db.commit()
    await db.refresh(product)

    if queued:
        background.add_task(process_product_assets, str(product.id))
    return MaterialsResponse(
        product_id=str(product.id),
        stage=product.stage.value,
        assets_queued=queued,
        assets_skipped=skipped,
    )


async def process_product_assets(product_id: str) -> None:
    """后台：下载 pending 素材 → 转存 storage。失败逐张标记，不中断。"""
    from services.storage import get_storage

    storage = get_storage()
    async with get_session_factory()() as db:
        assets = (await db.scalars(
            select(ProductAsset).where(
                ProductAsset.product_id == UUID(product_id),
                ProductAsset.status == AssetStatus.pending,
            )
        )).all()
        for a in assets:
            a.status = AssetStatus.downloading
            await db.commit()
            try:
                stored = await storage.save_from_url(a.source_url, key_prefix="assets")
                a.stored_url = stored
                a.status = AssetStatus.uploaded
            except Exception as exc:  # noqa: BLE001
                a.status = AssetStatus.failed
                a.error = str(exc)[:500]
            await db.commit()


@router.post("/materials/bytes")
async def upload_material_bytes(
    background: BackgroundTasks,
    product_id: str = Form(...),
    group_key: str = Form("main"),
    variant_name: str | None = Form(None),
    sort_order: int = Form(0),
    type: str = Form("img"),
    file: UploadFile = File(...),
    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="商品不存在")
    data = await file.read()
    asset = ProductAsset(
        product_id=product.id,
        group_key=group_key,
        variant_name=variant_name,
        sort_order=sort_order,
        type=type,
        source_url="",
        status=AssetStatus.pending,
    )
    db.add(asset)
    await db.flush()
    # 直接转存字节
    from services.storage import get_storage
    storage = get_storage()
    try:
        asset.stored_url = await storage.save_bytes(data, f"assets/{asset.id}", file.content_type or "")
        asset.status = AssetStatus.uploaded
    except Exception as exc:  # noqa: BLE001
        asset.status = AssetStatus.failed
        asset.error = str(exc)[:500]
    await db.commit()
    return {"asset_id": str(asset.id), "status": asset.status.value}


@router.get("/products/{product_id}/fingerprints")
async def product_fingerprints(
    product_id: str,
    db: AsyncSession = Depends(get_db),
    _user: dict = Depends(get_current_user),
):
    rows = (await db.scalars(
        select(ProductAsset.dedupe_key).where(
            ProductAsset.product_id == UUID(product_id),
            ProductAsset.dedupe_key.isnot(None),
        )
    )).all()
    return {"dedupe_keys": list(rows)}


@router.get("/collected")
async def is_collected(
    platform: str,
    itemId: str,
    db: AsyncSession = Depends(get_db),
    _user: dict = Depends(get_current_user),
):
    rows = (await db.execute(
        select(Product).where(
            Product.source_platform == platform,
            Product.source_item_id == itemId,
        )
    )).scalars().all()
    return {"collected": len(rows) > 0, "count": len(rows)}