george/descarga_factura_transporte
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
cdcecc2b56ce4a556ad714ae7b492079ff4503ad
descarga_factura_transporte/draft_sap.py
T
GeorgeandClaude Sonnet 5.5 f0ad6aa78e se valida si la factura ya existe en SAP y se avisa cuando SAP está fuera de línea 
- sapb1_manager.buscar_factura_existente busca la factura (proveedor, tipo,
  serie, número) en facturas de compra no anuladas y en drafts; si existe no
  se crea otro draft y se responde "ya está registrada" sin pisar el registro
  de una aceptación anterior.
- SapNoDisponible (subclase de ConnectionError) distingue una caída (timeout,
  sin conexión, HTTP 5xx) de un rechazo normal de SAP (HTTP 4xx). Una caída
  de Service Layer con HANA sin resultados ya no se lee como "guía no existe".
- Con SAP o HANA caídos se avisa una sola vez al proveedor con la plantilla
  "En cola" y el correo se reintenta en los ciclos siguientes (hasta 24), sin
  repetir avisos; al agotarse se avisa al equipo.
- La lógica por factura pasa a _procesar_factura; lo ya resuelto dentro de un
  correo no se repite al reintentarlo.
- seguimiento: columna intentos (con migración), registrar_intento y el
  estado de notificación en_espera.
- database_utilities.DB_NAME pasa a ser absoluto (junto al código).

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-22 12:08:38 -05:00

445 lines
19 KiB
Python
Raw Blame History

"""Validación de precios contra SAP y construcción del draft SAP B1."""
import re
import logging
import math
from datetime import datetime, timedelta
import sapb1_manager
import hana_manager
from pipeline_utils import log
# ============================================================================
def hay_error_de_consulta(resultados_val: list[dict]) -> str | None:
"""
Devuelve el detalle del primer error de consulta a SAP que dejó ``validate_prices``
(una guía cuya consulta lanzó excepción), o None si no hubo ninguno. Un error de
consulta es una falla nuestra: no debe confundirse con una guía inexistente.
"""
for resultado in resultados_val:
for item in resultado.get("guias_lt", []):
if item.get("error"):
return item["error"]
return None
def hana_disponible() -> bool:
"""
True si se puede abrir una conexión a SAP HANA. ``get_delivery_notes`` devuelve ``[]``
tanto si una guía no existe como si HANA está caído; antes de decirle a un proveedor
que una guía "no se encontró" conviene descartar que haya sido una caída.
"""
conexion = hana_manager.get_connection()
if conexion is None:
return False
try:
conexion.close()
except Exception:
pass
return True
def calcular_fecha_vencimiento(fecha_base: str, condicion_pago: dict | None) -> str:
"""
Calcula la fecha de vencimiento sumando a ``fecha_base`` los días de
la condición de pago (PaymentTermsTypes) del proveedor.
Usa ``NumberOfAdditionalDays`` como días a sumar; si ese valor es 0,
usa ``NumberOfAdditionalMonths * 30`` en su lugar.
Args:
fecha_base: Fecha de contabilización, formato "YYYY-MM-DD".
condicion_pago: Resultado de ``sapb1_manager.obtener_condicion_pago``
(campos ``NumberOfAdditionalDays``, ``NumberOfAdditionalMonths``),
o None si el proveedor no tiene condición de pago configurada.
Returns:
str — Fecha de vencimiento, formato "YYYY-MM-DD". Si
``condicion_pago`` es None, retorna ``fecha_base`` sin cambios.
"""
fecha = datetime.strptime(fecha_base, "%Y-%m-%d")
if not condicion_pago:
return fecha.strftime("%Y-%m-%d")
dias = int(condicion_pago.get("NumberOfAdditionalDays") or 0)
if dias == 0:
dias = int(condicion_pago.get("NumberOfAdditionalMonths") or 0) * 30
fecha += timedelta(days=dias)
return fecha.strftime("%Y-%m-%d")
# ============================================================================
def build_sap_draft(factura: dict, resultados_val: list[dict]) -> dict:
"""
Construye el payload JSON para crear un Draft en SAP B1 Service Layer
(endpoint ``POST /Drafts``).
Flujo:
1. Toma los datos de cabecera de la factura (RUCS, fechas, totales).
2. Determina el ``ItemCode`` según la fecha de emisión:
``C103002-A`` si es de un mes anterior, ``C103002`` si es del
mes actual.
3. Prepara las cuotas de detracción (``DocumentInstallments``):
si la factura tiene detracción, genera dos cuotas (detracción y
neto); si no, una sola cuota por el total.
4. Por cada línea de factura, itera sobre las guías validadas
(``guias_lt[]`` de ``resultados_val``) y genera una línea de
documento (``DocumentLine``) por cada guía con:
- ``ItemCode``, ``Quantity``, ``PriceAfterVAT``,
- ``WarehouseCode``, ``U_IDC_GRFLETE`` y campos fijos.
Cuando la validación es ``ok=True`` usa los valores de SAP
(``draft_quantity``, ``draft_price``); cuando no, usa los del XML.
5. Calcula ``U_IDC_VALREF`` como la suma de ``valor_referencial``
(extraído del XML SUNAT por ``sunat_schema``) de todas las
líneas de la factura.
Args:
factura: Diccionario con los datos de la factura extraídos del
XML (claves: ``serie``, ``numero``, ``tipo``, ``lineas[]``,
``guias_remision[]``, ``tiene_detraccion``, etc.).
resultados_val: Lista de resultados de ``validate_prices``,
uno por línea de factura. Cada elemento contiene
``guias_lt[]``, ``ok``, etc.
Returns:
Dict con la estructura completa del draft SAP B1 lista para
enviar a ``POST /Drafts``.
Example:
Factura F002-486 con guía T001-127773 y SAP devuelve
``FL10007 qty=350 pav=1.3396``::
draft = build_sap_draft(factura, resultados_val)
# → {
# "DocType": "dDocument_Items",
# "CardCode": "PL20537929520",
# "DocumentLines": [{
# "ItemCode": "C103002",
# "Quantity": 350,
# "PriceAfterVAT": 1.3396,
# "U_IDC_GRFLETE": "09T001-127773",
# ...
# }],
# "DocumentInstallments": [...],
# "U_IDC_VALREF": 31181.5,
# ...
# }
"""
serie = factura.get("serie", "")
numero = factura.get("numero", "")
tipo = factura.get("tipo", "01")
total = factura.get("total_factura", 0) or 0
tiene_detraccion = factura.get("tiene_detraccion", False)
codigo_detraccion = factura.get("codigo_detraccion", "")
porcentaje_detraccion = factura.get("porcentaje_detraccion", 0) or 0
monto_detraccion = factura.get("monto_detraccion", 0) or 0
fecha_emision = factura.get("fecha_emision", "")
ruc_emisor = factura.get("ruc_emisor", "")
razon_social_emisor = factura.get("razon_social_emisor", "")
card_code = f"PL{ruc_emisor}"
fecha_contabilizacion = datetime.now().strftime("%Y-%m-%d")
emision = datetime.strptime(fecha_emision, "%Y-%m-%d")
item_code = "C103002-A" if (emision.year, emision.month) < (datetime.now().year, datetime.now().month) else "C103002"
condicion_pago = sapb1_manager.obtener_condicion_pago(card_code)
fecha_vencimiento = calcular_fecha_vencimiento(fecha_contabilizacion, condicion_pago)
if tiene_detraccion:
# neto = round(total - monto_detraccion, 2)
monto_detraccion = math.ceil(monto_detraccion)
neto = total - monto_detraccion
installments = [
{"DueDate": fecha_vencimiento, "Total": monto_detraccion, "InstallmentId": 1, "U_VS_TIPOCUOTA": "D", "U_VS_TIPOSERVICIO": f"{codigo_detraccion}02"},
{"DueDate": fecha_vencimiento, "Total": round(neto,2), "InstallmentId": 2, "U_VS_TIPOCUOTA": "N", "U_VS_TIPOSERVICIO": ""},
]
else:
installments = [
{"DueDate": fecha_vencimiento, "Total": round(total, 2), "InstallmentId": 1, "U_VS_TIPOCUOTA": "N", "U_VS_TIPOSERVICIO": ""},
]
doc_lines = []
guia_referencia = ""
for i, linea in enumerate(factura.get("lineas", [])):
resultado_validacion = resultados_val[i] if i < len(resultados_val) else None
guias_lt = (resultado_validacion or {}).get("guias_lt", [])
if not guias_lt:
continue
guia_referencia = f"{guias_lt[0]['guia']['tipo']}{guias_lt[0]['guia']['serie']}-{guias_lt[0]['guia']['numero']}"
for item in guias_lt:
codigo_guia = f"{item['guia']['tipo']}{item['guia']['serie']}-{item['guia']['numero']}"
lt_val = math.trunc(item["draft_price"] * 1_000_000) / 1_000_000
qty = item.get("draft_quantity", int(item.get("quantity", 1)))
doc_lines.append({
"ItemCode": item_code,
"Quantity": qty,
"PriceAfterVAT": lt_val,
"WarehouseCode": item.get("warehouse_code", "A-CENTRA"),
"CostingCode": "PRO",
"CostingCode2": "PRO03",
"TaxCode": "I18",
"FreeText": linea.get("descripcion", "")[:100],
"MeasureUnit": "UND",
"U_BPP_CTAD": "9231101",
"U_BPP_OPER": "A",
"U_tipoOpT12": "02",
"U_VS_ONEROSO": "1",
"U_VS_TIPAFE": "10",
"U_IDC_GRFLETE": codigo_guia,
})
valor_referencial = sum(l.get("valor_referencial") or 0 for l in factura.get("lineas", []))
return {
"DocType": "dDocument_Items",
"DocDate": fecha_contabilizacion,
"DocDueDate": fecha_vencimiento,
"TaxDate": fecha_emision,
"CardCode": card_code,
"NumAtCard": f"{tipo}{serie}-{numero}",
"U_IDC_NUMGR": "" if not guia_referencia else guia_referencia,
"DocCurrency": "SOL",
"Comments": f"Factura {serie}-{numero} / {razon_social_emisor}"[:254],
"JournalMemo": razon_social_emisor[:254],
"DocObjectCode": "oPurchaseInvoices",
"U_BPV_SERI": f"{serie}-01",
"U_VS_NATNUM": "T",
"U_BPV_NCON2": numero,
"U_BPP_MDTD": tipo,
"U_BPP_MDBI": "A",
"U_BPP_MDCD": numero,
"U_BPP_MDSD": serie,
"U_VS_CODSER": f"{codigo_detraccion}02" if codigo_detraccion else "",
"U_VS_AFEDET": "Y" if tiene_detraccion else "N",
"U_VS_PORDET": porcentaje_detraccion,
"U_VS_APLANT": "Y",
"U_BPP_MDSD2": serie,
"U_IDC_FLETE": "02",
"U_IDC_LINEANE": "01",
"U_IDC_VALREF": valor_referencial,
"U_VS_ANEXO_PDF": factura.get("ruta_red_pdf", ""),
"U_VS_ANEXO_XML": factura.get("ruta_red_xml", ""),
"DocumentLines": doc_lines,
"DocumentInstallments": installments,
}
# ============================================================================
def validate_prices(factura: dict) -> tuple[bool, list[dict]]:
"""
Valida los precios de cada guía contra DeliveryNotes en SAP y
prepara los datos necesarios para construir el draft.
Flujo:
1. Obtiene el maestro de zonas SAP (``U_IDC_ZONAS``) para resolver
el nombre de zona a partir de ``U_IDC_DESTINO``.
2. Por cada línea de factura:
a. Itera sobre sus ``guias_remision[]``.
b. Para cada guía consulta SAP via
``sapb1_manager.get_delivery_notes``.
c. Dentro de cada DeliveryNote, suma los ``LineTotal``,
``PriceAfterVAT`` y obtiene la máxima ``Quantity`` de las
líneas cuyo ``ItemCode`` coincide con ``FL\\d``.
d. Acumula los resultados en ``guias_lt[]`` (uno por guía) con
``line_total``, ``quantity``, ``price_after_vat_sum``,
``warehouse_code`` y ``zona``.
3. Determina si los precios coinciden (``ok``):
``abs(total_linea - sap_total) < 0.01``.
4. Fija ``draft_quantity`` y ``draft_price`` de cada guía:
si ``ok``, usa SAP (``int(quantity)`` y ``price_after_vat_sum``);
si no, usa el XML (``int(cantidad)`` y ``total_linea / cantidad``).
Args:
factura: Diccionario con los datos de la factura (debe contener
``lineas[]``, y cada línea debe tener ``guias_remision[]``
y ``total_linea``).
Returns:
tuple[bool, list[dict]]:
- ``todas_coinciden``: True si TODAS las líneas tienen
``ok=True``.
- ``resultados``: Lista de resultados por línea, cada uno
con claves: ``grflete``, ``ok``, ``draft_price``,
``sap_price``, ``label``, ``warehouse_code``, ``guias_lt[]``
(y ``draft_cantidad`` en las líneas sin validar).
Example:
Factura con guía ``T001-127773`` y SAP devuelve
``FL10007 qty=350 pav=1.3396, FL20007 qty=350 pav=0``::
ok, resultados = validate_prices(factura)
# → ok = False (397.34 != 633.47)
# → resultados[0]["guias_lt"][0] = {
# "guia": {"tipo":"09","serie":"T001","numero":"127773"},
# "line_total": 397.34,
# "quantity": 350.0,
# "price_after_vat_sum": 1.3396,
# "zona": "CAÑETE",
# "warehouse_code": "A-CENTRA"
# }
"""
resultados = []
factura_lines = factura.get("lineas", [])
# listado de zonas para obtener su descripción
zonas_raw = sapb1_manager.get_zonas()
zonas = {z["Code"]: z["Name"] for z in zonas_raw}
for fl in factura_lines:
# por cada linea de factura se obtiene
# cantidad del xml de la factura
cantidad = float(fl.get("cantidad", 0))
# total de linea antes de impuesto del xml de la factura
total_linea = float(fl.get("total_linea", 0) or 0)
# lista de guias asociadas a la linea de factura de xml o pdf
guias = fl.get("guias_remision", [])
# dificil que la cantidad sea 0, pero esos casos los descartamos
if cantidad == 0:
resultados.append({"grflete": "-", "ok": False, "draft_price": None, "sap_price": None, "label": "Cantidad cero, sin validar", "warehouse_code": "-", "draft_cantidad": cantidad, "guias_lt": []})
continue
# si no hay guia asociada, no se puede validar, lo descartamos
if not guias:
resultados.append({"grflete": "-", "ok": False, "draft_price": None, "sap_price": None, "label": "Sin guía de referencia", "warehouse_code": "-", "draft_cantidad": cantidad, "guias_lt": []})
continue
guias_lt = []
all_wh = set()
all_ok = True
# una linea de la factura puede tener varias guias de remision
# hay que validar cada una de ellas
for guia in guias:
codigo_guia = f"{guia['tipo']}{guia['serie']}-{guia['numero']}"
try:
delivery_notes = sapb1_manager.get_delivery_notes(
tipo=guia["tipo"], serie=guia["serie"], numero=guia["numero"]
)
wh_set = set()
suma_line_total = 0.0
max_qty = 0.0
sum_pav = 0.0
weight1 = 0.0
zona = ""
for dn in delivery_notes:
# obtenemos la zona de destino del documento
zona = zonas.get(dn.get("U_IDC_DESTINO",""),"")
# obtenemos el almacen de la primera linea
first_line = dn.get("DocumentLines", [None])[0]
if first_line:
wh = first_line.get("WarehouseCode", "")
if wh:
wh_set.add(wh)
# obtenemos el peso de la primera linea de la guia de remision
# que representa el peso de la guia de remisión del articulo inventariable
# y lo vamos acumulando para obtener el peso total de la linea de la factura
weight1 += float(first_line.get("Weight1", 0) or 0)
# recorremos cada linea del documento
for linea_detalle_dict in dn.get("DocumentLines", []):
# solo procesamos lineas que empiecen con FL\d
if not re.match(r"FL\d", linea_detalle_dict.get("ItemCode", "")):
continue
# obtenemos el total de linea
line_total = float(linea_detalle_dict.get("LineTotal", 0) or 0)
# obtenemos el precio después de impuesto
price_after_vat = float(linea_detalle_dict.get("PriceAfterVAT", 0) or 0)
# obtenemos la cantidad
quantity = float(linea_detalle_dict.get("Quantity", 0) or 0)
# sumamos el total de linea
suma_line_total += line_total
# obtenemos la cantidad maxima
max_qty = max(max_qty, quantity)
sum_pav += price_after_vat
log(f"└─ {dn.get('Series', '?')}-{dn.get('DocNum', '?')} | ItemCode: {linea_detalle_dict.get('ItemCode', '')} | PriceAfterVAT: {price_after_vat:.6f} | Quantity: {quantity:.6f} | LineTotal: {line_total:.6f}", 4)
guias_lt.append({"guia": guia, "line_total": suma_line_total, "warehouse_code": ", ".join(sorted(wh_set)) if wh_set else "A-CENTRA", "quantity": max_qty, "price_after_vat_sum": sum_pav, "zona":zona, "weight1": weight1, "draft_quantity": 0, "draft_price": 0.0})
all_wh.update(wh_set)
except sapb1_manager.SapNoDisponible:
raise # una caída de SAP no es un problema de la guía: lo decide quien llama
except Exception as ex:
log(f"⚠️ Guía {codigo_guia}: error al consultar SAP - {ex}", 2, severity=logging.WARNING)
guias_lt.append({"guia": guia, "line_total": 0.0, "warehouse_code": "A-CENTRA", "weight1": 0.0, "quantity": 0.0, "price_after_vat_sum": 0.0, "zona": "", "draft_quantity": 0, "draft_price": 0.0, "error": str(ex)})
all_ok = False
sap_total = sum(item["line_total"] for item in guias_lt)
sin_sap = all(item["line_total"] == 0.0 and item["quantity"] == 0.0 for item in guias_lt)
ok = all_ok and abs(total_linea - sap_total) < 0.01
label = "Sin guías SAP" if sin_sap else "LineTotal (FL*) vs total_linea"
grflete_str = "; ".join(f"{item['guia']['tipo']}{item['guia']['serie']}-{item['guia']['numero']}" for item in guias_lt)
wh_str = ", ".join(sorted(all_wh)) if all_wh else "A-CENTRA"
for item in guias_lt:
if ok:
item["draft_quantity"] = int(item["quantity"])
item["draft_price"] = item["price_after_vat_sum"]
else:
item["draft_quantity"] = int(cantidad)
item["draft_price"] = float(total_linea) / max(float(cantidad), 1)
resultados.append({
"grflete": grflete_str,
"ok": ok,
"draft_price": total_linea,
"sap_price": sap_total,
"label": label,
"warehouse_code": wh_str,
"guias_lt": guias_lt,
})
if sin_sap:
log("⏭️ Sin guías en SAP — línea saltada", 3)
else:
log(f"Cantidad: {cantidad} | Total LineTotal (SAP): {sap_total:.6f} vs total_linea (XML): {total_linea:.6f}", 3)
if ok:
log(f"✅ Coincide: {total_linea:.6f} ≈ {sap_total:.6f}", 3)
else:
log(f"❌ NO coincide — XML: {total_linea:.6f} vs SAP: {sap_total:.6f}", 3, severity=logging.WARNING)
todas_coinciden = all(r["ok"] for r in resultados)
return todas_coinciden, resultados
# ============================================================================
def obtener_valor_referencial_sap(factura: dict) -> list[dict]:
"""
Ejecuta ``Sp_ObtenerValorReferencial`` en SAP HANA por cada guía de
remisión (tipo 09) única de la factura y acumula el resultado.
Una factura puede referenciar la misma guía desde más de una línea;
en ese caso solo se consulta una vez por guía.
Args:
factura: Diccionario de la factura, con ``lineas[]`` y cada
línea con ``guias_remision[]`` (tipo 09).
Returns:
list[dict] — uno por guía única, con ``serie``, ``numero`` y
``valor_sap``.
"""
vistos = set()
resultados = []
for linea in factura.get("lineas", []):
for guia in linea.get("guias_remision", []):
clave = (guia.get("serie"), guia.get("numero"))
if clave in vistos:
continue
vistos.add(clave)
valor_sap = hana_manager.obtener_valor_referencial(guia.get("serie", ""), guia.get("numero", ""))
resultados.append({"serie": guia.get("serie", ""), "numero": guia.get("numero", ""), "valor_sap": valor_sap})
log(f"└─ Guía 09{guia.get('serie')}-{guia.get('numero')} | Valor Referencial SAP: {valor_sap:.2f}", 3)
return resultados
Reference in new issue View Git Blame Copy Permalink