Joey
273 lines
9.5 KiB
Python
273 lines
9.5 KiB
Python
"""采集入库：插件上传文本 + 图片 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)}
|