georgeandClaude Sonnet 5.5
901dd8763d
Pipeline de integración de órdenes de Shopify a SAP Service Layer:
Invoice de reserva, pago con tarjeta o transferencia por transacción,
solicitud de traslado, traslado de stock y entrega.
- Invoice igual a lo cobrado; fulfillments limitados a lo facturado
- Estados terminales integrated, timeout, ignored y cancelled
- Idempotencia contra SAP: adopta documentos existentes en vez de duplicar
- Reintentos de red en el cliente de Shopify
- Secretos fuera de los JSON de tienda (${VARIABLE} resuelta desde .env)
- Herramientas de consulta, integración puntual y carga de pruebas
- Documentación en docs/ (flujo, casos especiales, mapeo, configuración)
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
364 lines
17 KiB
Python
364 lines
17 KiB
Python
import json
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List
|
|
|
|
from core.config import cargar_config_tienda
|
|
from core.db import (
|
|
ensure_order_document,
|
|
ensure_order_fulfillment,
|
|
get_doc_entries,
|
|
get_last_shopify_order_id,
|
|
get_order_document,
|
|
get_pending_orders,
|
|
insert_order_header,
|
|
mark_document_created,
|
|
mark_document_error,
|
|
mark_orders_timeout,
|
|
set_order_card_code,
|
|
set_order_status,
|
|
update_order_snapshot,
|
|
)
|
|
from core.errors import SAPServiceLayerError
|
|
from core.logger_manager import get_logger, order_context, log_event
|
|
from core.sap import SAPServiceLayerClient
|
|
from core.shopify_client import ShopifyClient
|
|
from domain.business_partner import buscar_rut_en_order, resolver_card_code
|
|
from domain.delivery import construir_delivery
|
|
from domain.documents import buscar_documento_existente, crear_documento_sap
|
|
from domain.invoice import agregar_referencia_documento, construir_invoice, crear_invoice, fulfillments_facturados
|
|
from domain.inventory_transfer import construir_inventory_transfer_request, es_tienda, resolver_location
|
|
from domain.payment import PaymentConfigError, construir_incoming_payment
|
|
from domain.rut import formatear_rut
|
|
from domain.stock_transfer import construir_stock_transfer
|
|
|
|
logger = get_logger()
|
|
|
|
STORES_DIR = Path(__file__).resolve().parent.parent / "config" / "stores"
|
|
|
|
|
|
def _registrar_adoptado(document_sys_id: int, entidad: str, doc: Dict[str, Any]) -> None:
|
|
"""
|
|
Un documento que ya existía en SAP para esta orden (la BD local no lo tenía
|
|
registrado) se adopta en vez de crear uno nuevo, para no duplicarlo.
|
|
"""
|
|
mark_document_created(document_sys_id, doc["DocEntry"], doc["DocNum"])
|
|
mensaje = f"{entidad} DocNum {doc['DocNum']} ya existía en SAP; se adopta y no se crea otro"
|
|
logger.warning(mensaje)
|
|
log_event("documento_adoptado", level="warning", message=mensaje, tipo=entidad, doc_entry=doc["DocEntry"])
|
|
|
|
# Una orden pagada y despachada se integra aunque después se haya reembolsado
|
|
# (total o parcialmente): el reembolso es un tratamiento interno de comercial,
|
|
# no se refleja en SAP. Las anuladas (cancelled_at) se descartan antes.
|
|
ESTADOS_FINANCIEROS_FACTURABLES = ("paid", "partially_refunded", "refunded")
|
|
|
|
|
|
# ============================================================================
|
|
def cargar_tiendas() -> List[Dict[str, Any]]:
|
|
"""Lee todos los JSON de configuración en config/stores/."""
|
|
tiendas = []
|
|
for path in sorted(STORES_DIR.glob("*.json")):
|
|
if path.name.startswith("_"):
|
|
continue
|
|
tiendas.append(cargar_config_tienda(path))
|
|
return tiendas
|
|
|
|
|
|
# ============================================================================
|
|
def descargar_cabeceras_nuevas(store_config: Dict[str, Any]) -> int:
|
|
"""
|
|
Descarga cabeceras nuevas desde Shopify (since_id = última cabecera
|
|
guardada para esta tienda) y las guarda en la tabla orders.
|
|
Devuelve la cantidad de cabeceras nuevas insertadas.
|
|
"""
|
|
store_code = store_config["store_code"]
|
|
shopify_config = store_config["shopify"]
|
|
|
|
# El punto de corte de la configuración actúa como piso: en la primera
|
|
# corrida (tabla vacía) evita traer todo el historial desde el id=0, y si
|
|
# se sube después, salta las órdenes anteriores aunque ya haya cabeceras
|
|
# más antiguas guardadas en la BD.
|
|
since_id = max(get_last_shopify_order_id(store_code), int(shopify_config.get("initial_since_id") or 0))
|
|
|
|
page_limit = shopify_config.get("page_limit", 50)
|
|
|
|
shopify = ShopifyClient.from_store_config(shopify_config)
|
|
ordenes = shopify.obtener_cabeceras_ordenes(since_id=since_id, limit=page_limit)
|
|
|
|
nuevas = 0
|
|
for orden in ordenes:
|
|
header = {
|
|
"shopify_order_id": orden["id"],
|
|
"order_number": orden.get("name"),
|
|
"tags": orden.get("tags"),
|
|
"shopify_created_at": orden.get("created_at"),
|
|
"financial_status": orden.get("financial_status"),
|
|
"fulfillment_status": orden.get("fulfillment_status"),
|
|
}
|
|
if insert_order_header(store_code, header):
|
|
nuevas += 1
|
|
|
|
logger.info(f"[{store_code}] Cabeceras nuevas descargadas: {nuevas} (since_id={since_id}, recibidas={len(ordenes)})")
|
|
log_event("headers_downloaded", store_code=store_code, since_id=since_id, count=nuevas)
|
|
return nuevas
|
|
|
|
|
|
# ============================================================================
|
|
def obtener_ordenes_pendientes(store_config: Dict[str, Any]) -> List[Dict[str, Any]]:
|
|
"""
|
|
Marca como timeout las órdenes vencidas de esta tienda y devuelve las
|
|
que aún no llegaron a un estado terminal (integrated/timeout).
|
|
"""
|
|
store_code = store_config["store_code"]
|
|
timeout_days = store_config.get("timeout_days", 30)
|
|
|
|
marcadas = mark_orders_timeout(store_code, timeout_days)
|
|
if marcadas:
|
|
logger.warning(f"[{store_code}] Órdenes marcadas como timeout: {marcadas}")
|
|
log_event("orders_timeout", store_code=store_code, count=marcadas)
|
|
|
|
return get_pending_orders(store_code)
|
|
|
|
|
|
# ============================================================================
|
|
def analizar_y_procesar_orden(store_config: Dict[str, Any], orden_row: Dict[str, Any]) -> bool:
|
|
"""
|
|
Descarga el detalle completo de la orden, valida paid+fulfillments,
|
|
resuelve el BusinessPartner, y crea/continúa (por sys_id de documento,
|
|
así se retoma donde quedó) Invoice -> Payment -> por cada fulfillment:
|
|
ITR -> (si es tienda) StockTransfer -> Delivery.
|
|
"""
|
|
store_code = store_config["store_code"]
|
|
order_sys_id = orden_row["sys_id"]
|
|
|
|
shopify = ShopifyClient.from_store_config(store_config["shopify"])
|
|
sap = SAPServiceLayerClient.from_store_config(store_config["sap_connection"])
|
|
|
|
detalle = shopify.obtener_orden(orden_row["shopify_order_id"])
|
|
update_order_snapshot(order_sys_id, detalle)
|
|
|
|
if detalle.get("cancelled_at"):
|
|
# Orden anulada en Shopify: nunca se integra, aunque siga "paid" y con fulfillments.
|
|
motivo = f"Anulada en Shopify el {detalle['cancelled_at'][:10]}: {detalle.get('cancel_reason') or 'sin motivo'}"
|
|
set_order_status(order_sys_id, "cancelled", motivo)
|
|
log_event("order_cancelled", store_code=store_code, shopify_order_id=orden_row["shopify_order_id"], message=motivo)
|
|
return False
|
|
|
|
if detalle.get("financial_status") not in ESTADOS_FINANCIEROS_FACTURABLES or not detalle.get("fulfillments"):
|
|
set_order_status(order_sys_id, "not_eligible")
|
|
return False
|
|
|
|
envio = float(((detalle.get("total_shipping_price_set") or {}).get("shop_money") or {}).get("amount", 0) or 0)
|
|
if envio > 0:
|
|
mensaje = f"Orden tiene costo de envío ({envio}) y ese mapeo aún no está soportado"
|
|
logger.error(mensaje)
|
|
log_event("envio_no_soportado", level="error", message=mensaje, monto=envio)
|
|
set_order_status(order_sys_id, "error", mensaje)
|
|
return False
|
|
|
|
rut = buscar_rut_en_order(detalle)
|
|
rut_sap = formatear_rut(rut) if rut else None
|
|
|
|
card_code = orden_row.get("card_code")
|
|
if not card_code:
|
|
resuelto = resolver_card_code(sap, store_config, rut, order=detalle)
|
|
card_code = resuelto["card_code"]
|
|
set_order_card_code(order_sys_id, card_code)
|
|
|
|
# ---- Invoice ----
|
|
invoice_doc_sys_id = ensure_order_document(order_sys_id, "INVOICE", 10)
|
|
invoice_row = get_order_document(invoice_doc_sys_id)
|
|
if invoice_row["status"] == "created":
|
|
invoice = sap.forward("GET", f"Invoices({invoice_row['sap_doc_entry']})").json()
|
|
else:
|
|
transacciones = shopify.obtener_transacciones(orden_row["shopify_order_id"])
|
|
payload = construir_invoice(detalle, store_config, card_code, federal_tax_id=rut_sap, transactions=transacciones)
|
|
if not payload["DocumentLines"]:
|
|
mensaje = "Orden sin líneas vigentes (todas canceladas antes de la venta)"
|
|
logger.warning(mensaje)
|
|
set_order_status(order_sys_id, "not_eligible", mensaje)
|
|
return False
|
|
try:
|
|
invoice = buscar_documento_existente(sap, "Invoices", store_code, detalle["id"], "DocumentLines", None, solo_vigentes=True)
|
|
if invoice:
|
|
_registrar_adoptado(invoice_doc_sys_id, "Invoices", invoice)
|
|
else:
|
|
invoice = crear_invoice(sap, payload)
|
|
mark_document_created(invoice_doc_sys_id, invoice["DocEntry"], invoice["DocNum"])
|
|
except SAPServiceLayerError as exc:
|
|
mark_document_error(invoice_doc_sys_id, exc.detail)
|
|
log_event("invoice_error", level="error", message=exc.detail, payload=payload)
|
|
return False
|
|
|
|
# ---- Payment ----
|
|
payment_doc_sys_id = ensure_order_document(order_sys_id, "INCOMING_PAYMENT", 20)
|
|
payment_row = get_order_document(payment_doc_sys_id)
|
|
if payment_row["status"] != "created" and invoice.get("DocTotal") and invoice.get("PaidToDate", 0) >= invoice["DocTotal"] - 0.5:
|
|
# La Invoice ya figura pagada en SAP (el pago existe pero la BD local no lo registró): no se duplica.
|
|
mark_document_created(payment_doc_sys_id, None, None)
|
|
log_event("documento_adoptado", level="warning", message=f"Invoice {invoice['DocNum']} ya pagada en SAP; no se crea el pago", tipo="IncomingPayments")
|
|
elif payment_row["status"] != "created":
|
|
total = sum(line["Quantity"] * line["PriceAfterVAT"] for line in invoice["DocumentLines"])
|
|
try:
|
|
transacciones = shopify.obtener_transacciones(orden_row["shopify_order_id"])
|
|
payload = construir_incoming_payment(detalle, transacciones, store_config, invoice, invoice["DocEntry"], total)
|
|
except PaymentConfigError as exc:
|
|
mark_document_error(payment_doc_sys_id, str(exc))
|
|
log_event("payment_error", level="error", message=str(exc))
|
|
return False
|
|
try:
|
|
data = crear_documento_sap(sap, "IncomingPayments", payload)
|
|
mark_document_created(payment_doc_sys_id, data["DocEntry"], data["DocNum"])
|
|
except SAPServiceLayerError as exc:
|
|
mark_document_error(payment_doc_sys_id, exc.detail)
|
|
log_event("payment_error", level="error", message=exc.detail, payload=payload)
|
|
return False
|
|
|
|
# ---- Por cada fulfillment ----
|
|
todo_ok = True
|
|
for fulfillment in fulfillments_facturados(detalle["fulfillments"], invoice):
|
|
location = resolver_location(store_config, fulfillment["location_id"])
|
|
if not location:
|
|
mensaje = f"location_id {fulfillment['location_id']} sin mapeo en config"
|
|
logger.error(mensaje)
|
|
log_event("location_sin_mapeo", level="error", message=mensaje, location_id=fulfillment["location_id"])
|
|
set_order_status(order_sys_id, "error", mensaje)
|
|
todo_ok = False
|
|
continue
|
|
|
|
fulfillment_sys_id = ensure_order_fulfillment(
|
|
order_sys_id,
|
|
fulfillment["id"],
|
|
str(fulfillment["location_id"]),
|
|
location["is_store"],
|
|
location["warehouse_code"],
|
|
)
|
|
|
|
# ITR (siempre)
|
|
itr_doc_sys_id = ensure_order_document(order_sys_id, "INVENTORY_TRANSFER_REQUEST", 30, fulfillment_sys_id)
|
|
itr_row = get_order_document(itr_doc_sys_id)
|
|
if itr_row["status"] == "created":
|
|
itr = sap.forward("GET", f"InventoryTransferRequests({itr_row['sap_doc_entry']})").json()
|
|
else:
|
|
payload = construir_inventory_transfer_request(detalle, fulfillment, store_config, location["warehouse_code"], invoice)
|
|
try:
|
|
itr = buscar_documento_existente(
|
|
sap, "InventoryTransferRequests", store_code, detalle["id"], "StockTransferLines", payload["StockTransferLines"],
|
|
excluir=get_doc_entries(order_sys_id, "INVENTORY_TRANSFER_REQUEST"),
|
|
)
|
|
if itr:
|
|
_registrar_adoptado(itr_doc_sys_id, "InventoryTransferRequests", itr)
|
|
else:
|
|
item_codes = [line["ItemCode"] for line in payload["StockTransferLines"]]
|
|
data = crear_documento_sap(sap, "InventoryTransferRequests", payload, item_codes=item_codes)
|
|
mark_document_created(itr_doc_sys_id, data["DocEntry"], data["DocNum"])
|
|
agregar_referencia_documento(sap, invoice["DocEntry"], data["DocEntry"], data["DocNum"], "rot_InventoryTransferRequest")
|
|
itr = sap.forward("GET", f"InventoryTransferRequests({data['DocEntry']})").json()
|
|
except SAPServiceLayerError as exc:
|
|
mark_document_error(itr_doc_sys_id, exc.detail)
|
|
log_event("itr_error", level="error", message=exc.detail, payload=payload)
|
|
todo_ok = False
|
|
continue
|
|
|
|
if not location["is_store"]:
|
|
continue # CD: no sigue a StockTransfer/Delivery
|
|
|
|
# StockTransfer (solo tienda)
|
|
st_doc_sys_id = ensure_order_document(order_sys_id, "STOCK_TRANSFER", 40, fulfillment_sys_id)
|
|
st_row = get_order_document(st_doc_sys_id)
|
|
if st_row["status"] == "created":
|
|
stock_transfer = sap.forward("GET", f"StockTransfers({st_row['sap_doc_entry']})").json()
|
|
else:
|
|
payload = construir_stock_transfer(itr, store_config)
|
|
try:
|
|
stock_transfer = buscar_documento_existente(
|
|
sap, "StockTransfers", store_code, detalle["id"], "StockTransferLines", payload["StockTransferLines"],
|
|
excluir=get_doc_entries(order_sys_id, "STOCK_TRANSFER"),
|
|
)
|
|
if stock_transfer:
|
|
_registrar_adoptado(st_doc_sys_id, "StockTransfers", stock_transfer)
|
|
else:
|
|
item_codes = [line["ItemCode"] for line in payload["StockTransferLines"]]
|
|
data = crear_documento_sap(sap, "StockTransfers", payload, item_codes=item_codes)
|
|
mark_document_created(st_doc_sys_id, data["DocEntry"], data["DocNum"])
|
|
agregar_referencia_documento(sap, invoice["DocEntry"], data["DocEntry"], data["DocNum"], "rot_InventoryTransfer")
|
|
stock_transfer = sap.forward("GET", f"StockTransfers({data['DocEntry']})").json()
|
|
except SAPServiceLayerError as exc:
|
|
mark_document_error(st_doc_sys_id, exc.detail)
|
|
log_event("stock_transfer_error", level="error", message=exc.detail, payload=payload)
|
|
todo_ok = False
|
|
continue
|
|
|
|
# Delivery (solo tienda, base la Invoice)
|
|
dn_doc_sys_id = ensure_order_document(order_sys_id, "DELIVERY", 50, fulfillment_sys_id)
|
|
dn_row = get_order_document(dn_doc_sys_id)
|
|
if dn_row["status"] != "created":
|
|
payload = construir_delivery(invoice, stock_transfer, store_config)
|
|
try:
|
|
entrega = buscar_documento_existente(
|
|
sap, "DeliveryNotes", store_code, detalle["id"], "DocumentLines", payload["DocumentLines"],
|
|
excluir=get_doc_entries(order_sys_id, "DELIVERY"),
|
|
)
|
|
if entrega:
|
|
_registrar_adoptado(dn_doc_sys_id, "DeliveryNotes", entrega)
|
|
else:
|
|
item_codes = [line["ItemCode"] for line in payload["DocumentLines"]]
|
|
data = crear_documento_sap(sap, "DeliveryNotes", payload, item_codes=item_codes)
|
|
mark_document_created(dn_doc_sys_id, data["DocEntry"], data["DocNum"])
|
|
agregar_referencia_documento(sap, invoice["DocEntry"], data["DocEntry"], data["DocNum"], "rot_DeliveryNotes")
|
|
except SAPServiceLayerError as exc:
|
|
mark_document_error(dn_doc_sys_id, exc.detail)
|
|
log_event("delivery_error", level="error", message=exc.detail, payload=payload)
|
|
todo_ok = False
|
|
continue
|
|
|
|
if not todo_ok:
|
|
return False
|
|
|
|
if detalle.get("fulfillment_status") == "fulfilled":
|
|
set_order_status(order_sys_id, "integrated")
|
|
else:
|
|
set_order_status(order_sys_id, "in_progress")
|
|
|
|
log_event("order_processed", store_code=store_code, shopify_order_id=orden_row["shopify_order_id"])
|
|
return True
|
|
|
|
|
|
# ============================================================================
|
|
def procesar_tienda(store_config: Dict[str, Any], modo: str) -> bool:
|
|
store_code = store_config.get("store_code", "?")
|
|
logger.info(f"Procesando tienda {store_code} [{modo}]")
|
|
log_event("store_cycle_start", store_code=store_code, modo=modo)
|
|
|
|
descargar_cabeceras_nuevas(store_config)
|
|
pendientes = obtener_ordenes_pendientes(store_config)
|
|
|
|
max_ordenes = store_config.get("max_orders_per_cycle")
|
|
if max_ordenes:
|
|
pendientes = pendientes[:max_ordenes]
|
|
|
|
for orden_row in pendientes:
|
|
with order_context(store_code, orden_row.get("shopify_order_id")):
|
|
try:
|
|
analizar_y_procesar_orden(store_config, orden_row)
|
|
except Exception as exc:
|
|
logger.error(f"Error inesperado procesando orden {orden_row.get('shopify_order_id')}: {exc}")
|
|
|
|
return True
|
|
|
|
|
|
# ============================================================================
|
|
def ejecutar_ciclo(modo: str) -> bool:
|
|
"""Punto de entrada que main.py llama en cada tick del scheduler."""
|
|
tiendas = cargar_tiendas()
|
|
if not tiendas:
|
|
logger.warning("No hay tiendas configuradas en config/stores/")
|
|
return True
|
|
|
|
for store_config in tiendas:
|
|
try:
|
|
procesar_tienda(store_config, modo)
|
|
except Exception as exc:
|
|
logger.error(f"Error procesando tienda {store_config.get('store_code', '?')}: {exc}")
|
|
|
|
return True
|