consultoria-as/Autoparts-DB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: robust ML publish with pre-flight, preview, validation, async

Browse Source 
- Add /inventory-check endpoint for local pre-flight validation
- Add /listings/validate endpoint using ML /items/validate API
- Add /categories/<id>/attributes endpoint for required attrs
- Add /listings/async + polling for background publishing via Celery
- Editable preview: title (0/60 counter), price, stock per item
- Pre-flight checks: image, stock, price, duplicate detection
- Image upload directly from publish modal (uses existing /items/<id>/image)
- Dynamic required attributes form based on selected ML category
- Frontend: validate button, async polling with progress, detailed error display
- Backend: build_item_payload supports custom_title, extra_attributes
This commit is contained in:
consultoria-as
2026-05-26 04:37:05 +00:00
parent 4866823ba9
commit b314a781a1
8 changed files with 706 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files

177
pos/services/marketplace_external_service.py
View File

@@ -134,9 +134,11 @@ def build_item_payload(
stock: int,
shipping_mode: str = "me2",
listing_type_id: str = "gold_special",
custom_title: str = None,
extra_attributes: list = None,
) -> dict:
"""Convert a Nexus inventory row into a MercadoLibre item payload."""
title = f"{inventory_row['name']} {inventory_row['brand'] or ''} {inventory_row['part_number'] or ''}".strip()
title = custom_title or f"{inventory_row['name']} {inventory_row['brand'] or ''} {inventory_row['part_number'] or ''}".strip()
# ML title limit is 60 chars; truncate smartly
if len(title) > 60:
title = title[:57] + "..."
@@ -185,6 +187,12 @@ def build_item_payload(
{"id": "VEHICLE_MODEL_NAME", "value_name": first["model"]}
)
# Extra attributes from user input (category requirements)
if extra_attributes:
for attr in extra_attributes:
if attr.get("id") and attr.get("value_name"):
payload["attributes"].append(attr)
return payload
@@ -192,12 +200,169 @@ def build_item_payload(
# LISTINGS CRUD
# ═══════════════════════════════════════════════════════════════════════════
def check_inventory_ml_status(tenant_conn, inventory_ids: list[int]) -> dict:
"""Check local pre-flight status for ML publishing.
Returns per-item dict with checks: has_image, has_stock, has_price,
already_published, and the generated title.
"""
cur = tenant_conn.cursor()
cur.execute(
"""
SELECT id, part_number, name, brand, price_1, image_url
FROM inventory
WHERE id = ANY(%s)
""",
(inventory_ids,),
)
rows = {r[0]: r for r in cur.fetchall()}
stock_map = get_stock_bulk(tenant_conn, branch_id=None)
# Check existing active listings
cur.execute(
"""
SELECT inventory_id, external_item_id, external_status, external_permalink
FROM marketplace_listings
WHERE inventory_id = ANY(%s) AND channel = 'mercadolibre' AND is_active = true
""",
(inventory_ids,),
)
listings = {r[0]: {"external_item_id": r[1], "status": r[2], "permalink": r[3]} for r in cur.fetchall()}
cur.close()
results = []
for inv_id in inventory_ids:
row = rows.get(inv_id)
if not row:
results.append({"inventory_id": inv_id, "exists": False})
continue
price = float(row[4]) if row[4] else 0
stock = stock_map.get(inv_id, 0)
image_url = row[5]
title = f"{row[2]} {row[3] or ''} {row[1] or ''}".strip()
if len(title) > 60:
title = title[:57] + "..."
existing = listings.get(inv_id)
results.append({
"inventory_id": inv_id,
"exists": True,
"title": title,
"has_image": bool(image_url),
"has_stock": stock > 0,
"has_price": price > 0,
"price": price,
"stock": stock,
"image_url": image_url,
"already_published": existing is not None,
"existing_listing": existing,
})
return {"items": results}
def validate_items(
tenant_conn,
inventory_ids: list[int],
meli_category_id: str,
listing_type_id: str = "gold_special",
shipping_mode: str = "me2",
custom_data: dict = None,
) -> dict:
"""Validate items against ML /items/validate without creating them.
Returns per-item validation results with ML error details if any.
"""
cfg = get_meli_config(tenant_conn)
svc = _get_meli_service(cfg)
if not svc:
raise ValueError("MercadoLibre not configured")
cur = tenant_conn.cursor()
cur.execute(
"""
SELECT id, part_number, name, brand, price_1, vehicle_compatibility,
image_url, unit, is_active
FROM inventory
WHERE id = ANY(%s) AND is_active = true
""",
(inventory_ids,),
)
rows = {r[0]: r for r in cur.fetchall()}
stock_map = get_stock_bulk(tenant_conn, branch_id=None)
custom_data = custom_data or {}
results = {"valid": [], "invalid": []}
for inv_id in inventory_ids:
row = rows.get(inv_id)
if not row:
results["invalid"].append({"inventory_id": inv_id, "error": "No encontrado o inactivo"})
continue
inv = {
"id": row[0],
"part_number": row[1],
"name": row[2],
"brand": row[3],
"price_1": float(row[4]) if row[4] else 0,
"vehicle_compatibility": row[5],
"image_url": row[6],
"unit": row[7],
}
stock = stock_map.get(inv_id, 0)
if stock <= 0:
results["invalid"].append({"inventory_id": inv_id, "error": "Sin stock disponible"})
continue
if inv["price_1"] <= 0:
results["invalid"].append({"inventory_id": inv_id, "error": "El precio debe ser mayor a 0"})
continue
images = []
if inv.get("image_url"):
images.append(inv["image_url"])
if not images:
results["invalid"].append({"inventory_id": inv_id, "error": "El producto no tiene imagen"})
continue
title = (custom_data.get("titles") or {}).get(str(inv_id))
extra_attrs = (custom_data.get("attributes") or {}).get(str(inv_id))
price = (custom_data.get("prices") or {}).get(str(inv_id), inv["price_1"])
item_stock = (custom_data.get("stocks") or {}).get(str(inv_id), stock)
payload = build_item_payload(
inv, images, meli_category_id, price, item_stock,
shipping_mode=shipping_mode, listing_type_id=listing_type_id,
custom_title=title, extra_attributes=extra_attrs,
)
try:
validation = svc.validate_item(payload)
if validation.get("status") == "valid":
results["valid"].append({"inventory_id": inv_id, "validation": validation})
else:
errors = validation.get("validation_errors", [])
error_msgs = [f"{e.get('field', '')}: {e.get('message', '')}" for e in errors]
results["invalid"].append({"inventory_id": inv_id, "error": " | ".join(error_msgs) or "Validación fallida", "validation": validation})
except MeliError as e:
err_msg = _extract_meli_error(e)
results["invalid"].append({"inventory_id": inv_id, "error": err_msg})
except Exception as e:
results["invalid"].append({"inventory_id": inv_id, "error": str(e)})
cur.close()
return results
def publish_items(
tenant_conn,
inventory_ids: list[int],
meli_category_id: str,
listing_type_id: str = "gold_special",
shipping_mode: str = "me2",
custom_data: dict = None,
) -> dict:
"""Publish one or more inventory items to MercadoLibre.
@@ -225,6 +390,7 @@ def publish_items(
# Batch fetch stock
stock_map = get_stock_bulk(tenant_conn, branch_id=None)
custom_data = custom_data or {}
results = {"success": [], "failed": []}
for inv_id in inventory_ids:
@@ -257,15 +423,20 @@ def publish_items(
images = []
if inv.get("image_url"):
images.append(inv["image_url"])
# TODO: fetch additional images from a separate table if we have gallery support
if not images:
results["failed"].append({"inventory_id": inv_id, "error": "El producto no tiene imagen. ML requiere imagen para publicar."})
continue
title = (custom_data.get("titles") or {}).get(str(inv_id))
extra_attrs = (custom_data.get("attributes") or {}).get(str(inv_id))
price = (custom_data.get("prices") or {}).get(str(inv_id), inv["price_1"])
item_stock = (custom_data.get("stocks") or {}).get(str(inv_id), stock)
payload = build_item_payload(
inv, images, meli_category_id, inv["price_1"], stock,
inv, images, meli_category_id, price, item_stock,
shipping_mode=shipping_mode, listing_type_id=listing_type_id,
custom_title=title, extra_attributes=extra_attrs,
)
try:

4
pos/services/meli_service.py
View File

@@ -145,6 +145,10 @@ class MeliService:
# ─── Items (listings) ────────────────────────────────────────────────
def validate_item(self, payload: dict) -> dict:
"""Validate an item payload without creating it."""
return self._request("POST", "/items/validate", json_payload=payload)
def create_item(self, payload: dict) -> dict:
return self._request("POST", "/items", json_payload=payload)