#51 feat: servicio de importacion externa de categorias de productos

Issue #51: discriminacion y administracion de categorias de productos.

- CategoryTrytonService.import_from_external: lee categorias asociadas a
  productos salables, hace upsert por external_id (fallback nombre),
  preserva description/alternative_name/active, reporta
  checked/created/updated/untouched/failed
- upsert_category compartido y reutilizado por ProductTrytonService
This commit is contained in:
2026-10-10 16:38:47 -05:00
parent 73fff016a3
commit 74cd9738b4
3 changed files with 351 additions and 29 deletions

View File

@@ -0,0 +1,147 @@
from ...models.products import ProductCategory
def upsert_category(category_data):
"""Crea o actualiza una ProductCategory desde datos de Tryton.
Solo administra name y external_id (fuente Tryton). Preserva los
campos administrados por Don Confiao (description, alternative_name
y active). Devuelve (category, status, changes) donde status es
'created', 'updated' o 'untouched' y changes es la lista de cambios.
"""
external_id = str(category_data.get("id"))
name = category_data.get("name")
category = ProductCategory.objects.filter(external_id=external_id).first()
if category is None and name:
category = ProductCategory.objects.filter(name=name).first()
is_new = category is None
if is_new:
category = ProductCategory()
changes = []
if category.name != name:
changes.append(f"Nombre: {category.name} → {name}")
if category.external_id != external_id:
changes.append(
f"External ID: {category.external_id} → {external_id}"
)
category.name = name
category.external_id = external_id
category.save()
status = "created" if is_new else ("updated" if changes else "untouched")
return category, status, changes
class CategoryTrytonService:
"""Servicio para sincronización de categorías con Tryton"""
def __init__(self, tryton_client):
self.client = tryton_client
def import_from_external(self):
"""Importa desde Tryton las categorías asociadas a productos.
Lee las categorías referenciadas por los productos salables
(sin suponer eliminación de categorías locales). Reporta las
categorías revisadas, creadas, actualizadas, intactas y fallidas.
"""
context = {"company": 1}
product_ids = self._search_salable_products(context)
if not product_ids:
return self._empty_report()
tryton_products = self._get_product_details(product_ids, context)
category_ids = set()
for tryton_product in tryton_products:
category_ids.update(self._extract_category_ids(tryton_product))
if not category_ids:
return self._empty_report()
tryton_categories = self._get_category_details(
sorted(category_ids), context
)
checked = [
{"id": c.get("id"), "name": c.get("name")}
for c in tryton_categories
]
created_categories = []
updated_categories = []
untouched_categories = []
failed_categories = []
for tryton_category in tryton_categories:
try:
category, status, changes = upsert_category(tryton_category)
item = {
"id": category.id,
"name": category.name,
"external_id": category.external_id,
}
if status == "created":
created_categories.append(item)
elif status == "updated":
updated_categories.append(
{**item, "detail": ", ".join(changes)}
)
else:
untouched_categories.append(item)
except Exception as e:
print(
f"Error al importar categoría: {e}"
f"La categoría: {tryton_category}"
)
failed_categories.append(
{
"external_id": tryton_category.get("id"),
"name": tryton_category.get("name"),
"error": f"Error al importar la categoría: {e}",
}
)
return {
"checked_tryton_categories": checked,
"created_categories": created_categories,
"updated_categories": updated_categories,
"untouched_categories": untouched_categories,
"failed_categories": failed_categories,
}
def _search_salable_products(self, context):
method = "model.product.product.search"
params = [
[["salable", "=", True]],
0,
1000,
[["rec_name", "ASC"], ["id", None]],
context,
]
return self.client.call(method, params) or []
def _get_product_details(self, product_ids, context):
tryton_fields = ["id", "name", "template.categories"]
method = "model.product.product.read"
params = (product_ids, tryton_fields, context)
return self.client.call(method, params)
@staticmethod
def _extract_category_ids(tryton_product):
"""Extrae los IDs de categorías de la respuesta RPC de Tryton"""
template = tryton_product.get("template.") or {}
return template.get("categories") or []
def _get_category_details(self, category_ids, context):
method = "model.product.category.read"
params = (category_ids, ["id", "name"], context)
return self.client.call(method, params)
def _empty_report(self):
return {
"checked_tryton_categories": [],
"created_categories": [],
"updated_categories": [],
"untouched_categories": [],
"failed_categories": [],
}

View File

@@ -1,4 +1,5 @@
from ...models.products import Product, ProductCategory
from .categories import upsert_category
class ProductTrytonService:
@@ -121,42 +122,16 @@ class ProductTrytonService:
)
for tryton_category in tryton_categories:
try:
external_id = str(tryton_category.get("id"))
name = tryton_category.get("name")
category = ProductCategory.objects.filter(
external_id=external_id
).first()
if category is None and name:
category = ProductCategory.objects.filter(
name=name
).first()
is_new = category is None
if is_new:
category = ProductCategory()
old_name = category.name
old_external_id = category.external_id
category.name = name
category.external_id = external_id
category.save()
category, status, changes = upsert_category(tryton_category)
categories[tryton_category.get("id")] = category
item = {
"id": category.id,
"name": category.name,
"external_id": category.external_id,
}
if is_new:
if status == "created":
report["created"].append(item)
continue
changes = []
if old_name != category.name:
changes.append(f"Nombre: {old_name} → {category.name}")
if old_external_id != category.external_id:
changes.append(
"External ID: "
f"{old_external_id} → {category.external_id}"
)
if changes:
elif status == "updated":
report["updated"].append(
{**item, "detail": ", ".join(changes)}
)