"""采集入库:插件上传文本 + 图片 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)}