356 lines
13 KiB
Python
356 lines
13 KiB
Python
import math
|
||
from decimal import Decimal
|
||
from itertools import combinations
|
||
from math import floor
|
||
import sap_services
|
||
import hana_client
|
||
|
||
###################################################
|
||
reglas = {
|
||
"100": lambda m: m > 0,
|
||
"101": lambda m: m < 12,
|
||
"102": lambda m: m >= 12,
|
||
"103": lambda m: m >= 18,
|
||
"104": lambda m: m >= 24,
|
||
"105": lambda m: m >= 30,
|
||
"106": lambda m: m >= 36,
|
||
"107": lambda m: m >= 42,
|
||
"108": lambda m: m >= 48,
|
||
"109": lambda m: m >= 54,
|
||
"110": lambda m: m >= 60
|
||
}
|
||
|
||
vigencias = {
|
||
"100": "meses > 0",
|
||
"101": "meses < 12",
|
||
"102": "meses >= 12",
|
||
"103": "meses >= 18",
|
||
"104": "meses >= 24",
|
||
"105": "meses >= 30",
|
||
"106": "meses >= 36",
|
||
"107": "meses >= 42",
|
||
"108": "meses >= 48",
|
||
"109": "meses >= 54",
|
||
"110": "meses >= 60",
|
||
}
|
||
|
||
presentacion = {
|
||
"100": 1,
|
||
"101": 1,
|
||
"102": 2,
|
||
"103": 4,
|
||
"104": 6,
|
||
"105": 8,
|
||
"106": 10,
|
||
"107": 12,
|
||
"108": 24,
|
||
"109": 36,
|
||
"110": 12,
|
||
"111": 12,
|
||
"112": 24,
|
||
"113": 3
|
||
}
|
||
|
||
# si es venta consignación, se vende por unidad
|
||
# el multiplo = 1
|
||
# vigencia en meses > 0
|
||
ventas_tipo_consignacion = {
|
||
"110": "venta consignacion"
|
||
}
|
||
|
||
###################################################
|
||
###################################################
|
||
def obtener_documento_formato_json(session,registro):
|
||
|
||
objkey = registro['DocEntry']
|
||
endpoint = f"Orders({objkey})"
|
||
|
||
return sap_services.get_by_key(session,endpoint)
|
||
###################################################
|
||
|
||
def actualizar_detalle_documento(session,documento,body):
|
||
objkey = documento['DocEntry']
|
||
endpoint = f"Orders({objkey})"
|
||
|
||
return sap_services.update_document(session,endpoint,body)
|
||
|
||
###################################################
|
||
def seleccionar_mejor_lotes(line,lotes,vigencia,unidades_por_caja,tipo_venta,max_lotes=3):
|
||
'''
|
||
Si el tipo_venta = consignacion
|
||
unidades_por_caja = 1
|
||
'''
|
||
# cantidad total de unidades requeridas
|
||
total_objetivo = line["Quantity"]
|
||
|
||
# si el tipo de venta no es consignacion
|
||
# se debe ver si las cantidades requeridas
|
||
# son multiplos o no de las unidades_por_caja
|
||
if not tipo_venta in ventas_tipo_consignacion:
|
||
# validar si la cantidad solicitada es múltiplo de unidades_por_caja
|
||
es_multiplo = True if total_objetivo % unidades_por_caja == 0 else False
|
||
|
||
# si no es multiplo, entonces obtengo el maximo comun divisor
|
||
if not es_multiplo:
|
||
maximo_comun_divisor = math.gcd(int(total_objetivo), int(unidades_por_caja))
|
||
print(f" ⚠️Se utilizara el MCD que es {maximo_comun_divisor}")
|
||
unidades_por_caja = maximo_comun_divisor
|
||
|
||
# quitar los lotes que no cumplan con los meses minimos requeridos de vencimiento de lotes
|
||
filtrados = [l for l in lotes if reglas[vigencia](l["Months"])]
|
||
# ordenar los lotes por fecha de creacion, el mas antiguo primero
|
||
filtrados = sorted(filtrados, key=lambda x: x["InDate"])
|
||
|
||
# agregar indice para priorizar antiguedad
|
||
for i, lote in enumerate(filtrados):
|
||
lote["indice"] = i
|
||
|
||
mejor_exacto = None
|
||
mejor_parcial = None
|
||
|
||
for r in range(1, max_lotes + 1):
|
||
for combo in combinations(filtrados, r):
|
||
total = 0
|
||
detalle = []
|
||
|
||
# clave: lista ordenada de indices
|
||
prioridad_fifo = sorted(l["indice"] for l in combo)
|
||
|
||
for lote in combo:
|
||
restante = total_objetivo - total
|
||
if restante <= 0:
|
||
break
|
||
|
||
disponible = lote["AvailableQty"]
|
||
|
||
tomar = min(disponible, restante)
|
||
tomar = floor(tomar / unidades_por_caja) * unidades_por_caja
|
||
|
||
if tomar > 0:
|
||
detalle.append({
|
||
"DistNumber": lote["DistNumber"],
|
||
"InDate": lote["InDate"],
|
||
"Quantity": lote["Quantity"],
|
||
"AvailableQty": lote["AvailableQty"],
|
||
"TakeQuantity": tomar,
|
||
"ExpDate": lote["ExpDate"],
|
||
"Months": lote["Months"]
|
||
})
|
||
total += tomar
|
||
|
||
# ignorar combos que no aportan nada
|
||
if total == 0:
|
||
continue
|
||
|
||
# solucion exacta -> priorizar absoluta
|
||
if total == total_objetivo:
|
||
if not mejor_exacto or prioridad_fifo < mejor_exacto[2]:
|
||
mejor_exacto = (detalle, total, prioridad_fifo)
|
||
|
||
# mejor solucion parcial -> mayor total posible
|
||
elif total < total_objetivo:
|
||
if not mejor_parcial or (
|
||
total > mejor_parcial[1] or
|
||
(total == mejor_parcial[1] and prioridad_fifo < mejor_parcial[2])
|
||
):
|
||
mejor_parcial = (detalle, total, prioridad_fifo)
|
||
|
||
if mejor_exacto:
|
||
return mejor_exacto[0], mejor_exacto[1]
|
||
elif mejor_parcial:
|
||
return mejor_parcial[0], mejor_parcial[1]
|
||
return [], 0
|
||
|
||
###################################################
|
||
def se_le_debe_asignar_lote_a_linea(line):
|
||
cantidad = line["Quantity"]
|
||
lotes = [] if "BatchNumbers" not in line else line["BatchNumbers"]
|
||
|
||
linestatus = line["LineStatus"] ## bost_Open
|
||
|
||
if linestatus == "bost_Close":
|
||
return False
|
||
|
||
# no tiene asignado lotes
|
||
if len(lotes) == 0:
|
||
# se deba asignar lotes
|
||
return True
|
||
else:
|
||
# sumar cantidades de lotes
|
||
total_lotes = sum(lote["Quantity"] for lote in lotes)
|
||
if total_lotes == cantidad:
|
||
# no se deba asignar lotes
|
||
return False
|
||
else:
|
||
# se deba asignar lotes
|
||
return True
|
||
|
||
###################################################
|
||
def agregar_columna_available_qty(lotes):
|
||
for lote in lotes:
|
||
lote["AvailableQty"] = (lote["Quantity"] or Decimal(0)) - (lote["CommitQty"] or Decimal(0))
|
||
|
||
return lotes
|
||
|
||
###################################################
|
||
def documento_requiere_desasignar_lotes(documento):
|
||
necesita_actualizar = False
|
||
nuevas_lineas = []
|
||
|
||
for line in documento.get("DocumentLines",[]):
|
||
|
||
nueva_linea = {
|
||
"LineNum": line.get("LineNum"),
|
||
"ItemCode": line.get("ItemCode"),
|
||
"Quantity": line.get("Quantity"),
|
||
"WarehouseCode": line.get("WarehouseCode"),
|
||
"BatchNumbers": []
|
||
}
|
||
|
||
batches = line.get("BatchNumbers",[])
|
||
|
||
if batches:
|
||
total_line = line.get("Quantity",0)
|
||
total_batches = sum(b.get("Quantity",0) for b in batches)
|
||
|
||
if total_batches != total_line:
|
||
necesita_actualizar = True
|
||
for b in batches:
|
||
nueva_linea["BatchNumbers"].append({
|
||
"BaseLineNumber": line.get("LineNum"),
|
||
"BatchNumber": b.get("BatchNumber"),
|
||
"Quantity": 0.00
|
||
})
|
||
else:
|
||
# copiar tal cual
|
||
for b in batches:
|
||
nueva_linea["BatchNumbers"].append(b.copy())
|
||
else:
|
||
# sin lotes -> se deja vacio
|
||
nueva_linea["BatchNumbers"] = []
|
||
|
||
nuevas_lineas.append(nueva_linea)
|
||
|
||
nuevo_json = {
|
||
"DocumentLines": nuevas_lineas
|
||
}
|
||
|
||
return necesita_actualizar, nuevo_json
|
||
|
||
###################################################
|
||
def procesar_registro(registro,session):
|
||
documento = obtener_documento_formato_json(session,registro)
|
||
|
||
necesita_actualizar, nuevo_json = documento_requiere_desasignar_lotes(documento)
|
||
|
||
if necesita_actualizar:
|
||
|
||
print(f" 🧹 Desasignando lotes previos del documento...")
|
||
status_code = actualizar_detalle_documento(session,documento, nuevo_json)
|
||
print(f" ✅ Documento actualizado (Desasignación).")
|
||
|
||
documento = obtener_documento_formato_json(session,registro)
|
||
else:
|
||
# continuar con el documento
|
||
pass
|
||
|
||
maxima_cantidad_lotes = 3
|
||
vigencia = "100"
|
||
tipo_venta = documento["U_SIS_TIPVEN"]
|
||
if tipo_venta in ventas_tipo_consignacion:
|
||
print(f" 🚨 Tipo de venta: {tipo_venta} -> {ventas_tipo_consignacion[tipo_venta]}")
|
||
print(f" 🚨 Presentacion: sera de 1 Unidad para todos los articulos")
|
||
print(f" 🚨 Meses de vigencia > 0 meses")
|
||
|
||
else:
|
||
# de lo contrario se toma la vigencia seleccionada en el documento
|
||
vigencia = documento["U_SIS_VIGREQ"]
|
||
|
||
print(f" 📅 Vigencia requerida(meses de vencimiento): {vigencia} -> {vigencias[vigencia]}")
|
||
|
||
new_lines = []
|
||
# Reservas en memoria: {(itemcode, whscode, dist_number): cantidad_ya_reservada}
|
||
reservas = {}
|
||
|
||
for index, line in enumerate(documento.get("DocumentLines",[]),start=1):
|
||
|
||
treetype = line["TreeType"]
|
||
|
||
new_line = {
|
||
"LineNum": line.get("LineNum"),
|
||
"ItemCode": line.get("ItemCode"),
|
||
"Quantity": line.get("Quantity"),
|
||
"WarehouseCode": line.get("WarehouseCode"),
|
||
"BatchNumbers": []
|
||
}
|
||
|
||
|
||
if treetype == "iNotATree":
|
||
if se_le_debe_asignar_lote_a_linea(line):
|
||
codigo_presentacion = str(hana_client.obtener_presentacion_articulo(line["ItemCode"]))
|
||
|
||
# Presentacion del articulo o Unidaddes por Caja
|
||
unidades_por_caja = presentacion.get(codigo_presentacion, 1)
|
||
|
||
lotes_disponibles = hana_client.obtener_lotes_disponibles(line["ItemCode"],line["WarehouseCode"])
|
||
lotes_disponibles = agregar_columna_available_qty(lotes_disponibles)
|
||
|
||
# Descontar reservas ya asignadas a líneas anteriores del mismo documento
|
||
for lote in lotes_disponibles:
|
||
clave = (line["ItemCode"], line["WarehouseCode"], lote["DistNumber"])
|
||
ya_reservado = reservas.get(clave, 0)
|
||
if ya_reservado > 0:
|
||
lote["AvailableQty"] = max(lote["AvailableQty"] - ya_reservado, 0)
|
||
|
||
# core del proceso de asignacion
|
||
lotes_seleccionados, total_seleccionado = seleccionar_mejor_lotes(line,lotes_disponibles,vigencia
|
||
,1 if tipo_venta in ventas_tipo_consignacion else unidades_por_caja
|
||
,tipo_venta,maxima_cantidad_lotes)
|
||
|
||
### impresion de lotes disponibles
|
||
print(f" ➖ Línea {index} | Artículo: {line['ItemCode']} | Cant. req: {line['Quantity']} | Múltiplo de emp.: {unidades_por_caja} | Almacen: {line['WarehouseCode']}")
|
||
print(f" 📦 Se encontraron {len(lotes_disponibles)} lotes disponibles.")
|
||
print(" 🔍 Detalle lotes disponibles:")
|
||
for a in sorted(lotes_disponibles, key=lambda x: x["InDate"]):
|
||
print(f" - Lote: {a['DistNumber']:10} | F. Fab: {a['InDate']:10} | Disp: {float(a['AvailableQty']):10.2f} | Meses: {a['Months']:4}")
|
||
|
||
### impresion de lotes seleccionados
|
||
print(" 🎯 Lotes seleccionados:")
|
||
|
||
for a in lotes_seleccionados:
|
||
print(f" - Lote: {a['DistNumber']:10} | F. Fab: {a['InDate']:10} | A tomar: {float(a['TakeQuantity']):10.2f} | Meses: {a['Months']:4}")
|
||
|
||
for lote in lotes_seleccionados:
|
||
new_line["BatchNumbers"].append({
|
||
"BaseLineNumber": line.get("LineNum"),
|
||
"BatchNumber": lote.get("DistNumber"),
|
||
"Quantity": lote.get("TakeQuantity")
|
||
})
|
||
# Registrar la reserva en memoria
|
||
clave = (line["ItemCode"], line["WarehouseCode"], lote["DistNumber"])
|
||
reservas[clave] = reservas.get(clave, 0) + lote["TakeQuantity"]
|
||
else:
|
||
# el TreeType de esta Linea es diferente a iNotATree
|
||
pass
|
||
|
||
new_lines.append(new_line)
|
||
|
||
# Obtener líneas que tienen lotes asignados
|
||
lineas_con_lotes = [line for line in new_lines if len(line["BatchNumbers"]) > 0]
|
||
|
||
# Debo enviar todas las lineas del documento, tengan o no lotes asignados
|
||
body = {
|
||
"DocumentLines": new_lines
|
||
}
|
||
|
||
hizo_asignacion = len(lineas_con_lotes) > 0
|
||
|
||
# si no se realizo asignacion de lotes, no tiene sentido actualizar el documento
|
||
if hizo_asignacion:
|
||
print(f" 🔄 Asignando lotes en SAP ({len(lineas_con_lotes)} líneas con lotes)...")
|
||
status_code = actualizar_detalle_documento(session, documento, body)
|
||
print(f" ✅ Documento actualizado con lotes asignados.")
|
||
else:
|
||
print(f" ✅ No es necesario actualizar el documento.")
|
||
|
||
################################################### |