325 lines
14 KiB
Python
Executable File
325 lines
14 KiB
Python
Executable File
import json
|
|
import os
|
|
import shutil
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
from dotenv import load_dotenv
|
|
from lxml import etree
|
|
from jinja2 import Environment, FileSystemLoader
|
|
from logger_manager import get_logger
|
|
|
|
# ============================================================================
|
|
_BASE = Path(__file__).resolve().parent
|
|
# load_dotenv(_BASE / ".env")
|
|
# load_dotenv(_BASE.parent / ".env")
|
|
|
|
load_dotenv(dotenv_path=".env")
|
|
|
|
INPUT_DIR = Path(os.getenv("INPUT_DIR", "data/input"))
|
|
PROCESSED_DIR = Path("data/processed")
|
|
OUTPUT_DIR = Path("data/output")
|
|
HTML_DIR = Path("data/html")
|
|
SAP_B1_DIR = Path("data/sap_b1")
|
|
ERRORS_DIR = Path("data/errors")
|
|
TEMPLATE_FILE = _BASE / "template.html"
|
|
|
|
POST_TO_SAP_B1 = True
|
|
|
|
TEXT_MONTHS = {
|
|
1: "enero", 2: "febrero", 3: "marzo", 4: "abril",
|
|
5: "mayo", 6: "junio", 7: "julio", 8: "agosto",
|
|
9: "septiembre", 10: "octubre", 11: "noviembre", 12: "diciembre",
|
|
}
|
|
# ============================================================================
|
|
log = get_logger()
|
|
|
|
# ============================================================================
|
|
def _format_date(d: str) -> str:
|
|
if not d:
|
|
return ""
|
|
p = d.split("-")
|
|
if len(p) == 3:
|
|
return f"{p[2][:2]} de {TEXT_MONTHS.get(int(p[1]), p[1])} de {p[0]}"
|
|
return d
|
|
|
|
# ============================================================================
|
|
def _format_datetime(dt: str) -> str:
|
|
if not dt:
|
|
return ""
|
|
dp = dt.split("T")[0]
|
|
tp = dt.split("T")[1][:5] if "T" in dt else ""
|
|
fd = _format_date(dp)
|
|
return f"{fd} {tp} hrs." if tp else fd
|
|
|
|
# ============================================================================
|
|
def _event_text(ev: str) -> str:
|
|
return {"DATE_OF_INVOICE": "fecha de factura", "RECEIPT_OF_GOODS": "recepción de mercadería"}.get(ev, ev or "")
|
|
|
|
# ============================================================================
|
|
def apply_allowances(net_price: int, qty: int, allowances: list) -> tuple:
|
|
total = net_price * qty
|
|
applied = []
|
|
for alw in allowances:
|
|
pct = alw.get("percentage", 0)
|
|
if pct > 0:
|
|
dsc = round(total * pct / 100)
|
|
else:
|
|
dsc = round(alw.get("amount", 0))
|
|
applied.append({"type": alw.get("type"), "percentage": pct, "amount": dsc})
|
|
total -= dsc
|
|
return total, applied
|
|
|
|
# ============================================================================
|
|
def _strip_ns(root):
|
|
for el in root.iter():
|
|
if "}" in el.tag:
|
|
el.tag = el.tag.split("}", 1)[1]
|
|
etree.cleanup_namespaces(root)
|
|
return root
|
|
|
|
# ============================================================================
|
|
def _x(node, path, default=None):
|
|
v = node.xpath(path)
|
|
return v[0] if v else default
|
|
|
|
# ============================================================================
|
|
def _t(node, path, default=""):
|
|
v = _x(node, path)
|
|
return v.text.strip() if v is not None and v.text else default
|
|
|
|
# ============================================================================
|
|
def _a(node, path, attr, default=""):
|
|
v = _x(node, path)
|
|
return v.get(attr, default) if v is not None else default
|
|
|
|
# ============================================================================
|
|
def _l(node, path):
|
|
return node.xpath(path)
|
|
|
|
# ============================================================================
|
|
def substring_desde_fin(texto: str, cantidad: int = 10, excluir_ultimos: int = 1) -> str:
|
|
inicio = -(cantidad + excluir_ultimos)
|
|
fin = -excluir_ultimos if excluir_ultimos > 0 else None
|
|
return texto[inicio:fin]
|
|
|
|
# ============================================================================
|
|
def parse_order(xml_bytes: bytes, filename: str = "") -> dict:
|
|
root = _strip_ns(etree.fromstring(xml_bytes))
|
|
order = _x(root, "//order")
|
|
if order is None:
|
|
raise ValueError("Elemento <order> no encontrado")
|
|
|
|
msg = _x(root, "//messageHeader")
|
|
oc_id = _x(
|
|
order,
|
|
'typedEntityIdentification[@entityType="ORDER"]/entityIdentification/uniqueCreatorIdentification',
|
|
)
|
|
oc_num = oc_id.text.strip() if oc_id is not None and oc_id.text else ""
|
|
|
|
cd = order.get("creationDate", "")
|
|
pt = _x(order, "paymentTerms")
|
|
event = pt.get("eventType", "") if pt is not None else ""
|
|
net_days = _t(order, "paymentTerms/netPayment/timePeriodDue")
|
|
|
|
cab = {
|
|
"emisor": _t(msg, "from/gln") if msg is not None else "",
|
|
"receptor": _t(msg, "to/gln") if msg is not None else "",
|
|
"numero_oc": substring_desde_fin(oc_num) if (oc_num is not None and len(oc_num) > 10) else "",
|
|
"fecha_generacion": cd,
|
|
"fecha_generacion_format": _format_datetime(cd),
|
|
"fecha_embarque": _t(order, "movementDate"),
|
|
"fecha_embarque_format": _format_date(_t(order, "movementDate")),
|
|
"fecha_cancelacion": "",
|
|
"fecha_cancelacion_format": "",
|
|
"condiciones_pago": f"NET {net_days} días desde {_event_text(event)}" if net_days else "",
|
|
"condiciones_pago_texto": f"NET {net_days} días desde {_event_text(event)}" if net_days else "",
|
|
"lugar_entrega": _t(order, 'shipParty[@identificationType="SHIP_TO"]/gln'),
|
|
"dpto_ventas": _t(order, "salesDepartamentNumber"),
|
|
"tipo_oc": _t(order, "orderType"),
|
|
"promocion": _t(order, "promotionDealNumber"),
|
|
"numero_proveedor": _t(order, "internalVendorNumber"),
|
|
"observaciones": _t(order, "specialHandling"),
|
|
}
|
|
|
|
carrier = _x(root, "//mainCarrierTransport")
|
|
if carrier is not None:
|
|
cab["etapa_transporte"] = _t(carrier, "mode")
|
|
cab["transportista"] = _t(carrier, "carrierName")
|
|
cab["divisa"] = _t(carrier, "../../currency") or _t(carrier, "../../../currency") or _t(carrier, "../currency")
|
|
|
|
detalle = []
|
|
for item in _l(order, "lineItem"):
|
|
np = int(_t(item, "netPrice/amount", "0"))
|
|
rq = int(_t(item, "requestedQuantity", "0"))
|
|
cu = int(_t(item, "containedUnits", "0"))
|
|
ta = int(_t(item, "totalAmount/amount", "0"))
|
|
currency = _a(item, "netPrice/amount", "currencyISOcode", "CLP")
|
|
|
|
alw_list = []
|
|
for alw in _l(item, "allowanceCharge"):
|
|
alw_list.append({
|
|
"type": alw.get("allowanceOrChargeType", ""),
|
|
"percentage": int(_t(alw, "monetaryAmountOrPercentage/percentage", "0")),
|
|
"amount": int(_t(alw, "monetaryAmountOrPercentage/amount", "0")),
|
|
})
|
|
|
|
calc_total, applied_alw = apply_allowances(np, rq, alw_list)
|
|
color = _t(item, "itemColor")
|
|
size = _t(item, "itemSize")
|
|
|
|
detalle.append({
|
|
"linea": item.get("number", ""),
|
|
"cod_upc": _t(item, "itemIdentification/gtin"),
|
|
"item": _t(item, "itemIdentification/buyerItemNumber"),
|
|
"cod_prov": _t(item, "itemIdentification/vendorItemNumber"),
|
|
"color_talla_um": "/ ".join(filter(None, [color, size])),
|
|
"descripcion": _t(item, 'itemDescription[@language="SP"]/text'),
|
|
"precio_original": np,
|
|
"moneda": currency,
|
|
"allowances": applied_alw,
|
|
"precio_efectivo": round(calc_total / rq) if rq > 0 else 0,
|
|
"empaques": rq,
|
|
"unidad_empaque": _a(item, "requestedQuantity", "UnitType", ""),
|
|
"unid_empaque": cu,
|
|
"cantidad_unid": rq * cu,
|
|
"importe": ta,
|
|
})
|
|
|
|
return {
|
|
"metadata": {"archivo": filename, "procesado": datetime.now().isoformat()},
|
|
"cabecera": cab,
|
|
"detalle": detalle,
|
|
"totales": {
|
|
"importe_total": sum(d["importe"] for d in detalle),
|
|
"moneda": detalle[0]["moneda"] if detalle else "CLP",
|
|
},
|
|
}
|
|
|
|
# ============================================================================
|
|
def write_json(data: dict, path: Path):
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(path, "w", encoding="utf-8") as f:
|
|
json.dump(data, f, ensure_ascii=False, indent=2)
|
|
|
|
# ============================================================================
|
|
def write_html(data: dict, template_path: Path, output_path: Path):
|
|
env = Environment(loader=FileSystemLoader(str(template_path.parent)))
|
|
template = env.get_template(template_path.name)
|
|
output_path.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(output_path, "w", encoding="utf-8") as f:
|
|
f.write(template.render(data=data))
|
|
|
|
# ============================================================================
|
|
def process_file(xml_path: Path):
|
|
with log.task(f"📦 {xml_path.name}"):
|
|
log.info("📄 XML parseado")
|
|
with open(xml_path, "rb") as f:
|
|
data = parse_order(f.read(), filename=xml_path.name)
|
|
|
|
base = xml_path.stem
|
|
write_json(data, OUTPUT_DIR / f"{base}.json")
|
|
log.info("💾 JSON guardado")
|
|
|
|
write_html(data, TEMPLATE_FILE, HTML_DIR / f"{base}.html")
|
|
log.info("💾 HTML generado")
|
|
|
|
if POST_TO_SAP_B1:
|
|
with log.task("🔐 SAP B1"):
|
|
try:
|
|
from sapb1_mapper import to_draft_payload
|
|
from sapb1_manager import create_orders
|
|
from sql_manager import get_client_by_gln, get_item_catalog, get_price_list, get_prices, get_card_name
|
|
from email_reporter import generate_email_html
|
|
|
|
cab = data.get("cabecera", {})
|
|
emisor = cab.get("emisor", "")
|
|
lugar_entrega = cab.get("lugar_entrega", "")
|
|
|
|
result = get_client_by_gln(emisor, lugar_entrega)
|
|
|
|
if result:
|
|
cardcode = result["CardCode"]
|
|
city = result["City"]
|
|
county = result["County"]
|
|
log.info("CardCode={} (desde CRD1)".format(cardcode))
|
|
else:
|
|
cardcode = ""
|
|
log.warning("No se obtuvo Socio de negocios para el {}.{}".format(emisor, lugar_entrega))
|
|
try:
|
|
from mail_sender import send_mail
|
|
send_mail(
|
|
subject=f"OC {cab.get('numero_oc', '')} — Sin socio de negocios",
|
|
text=f"No se obtuvo Socio de negocios para el {emisor}.{lugar_entrega}",
|
|
)
|
|
log.info("Correo de notificación enviado")
|
|
except Exception as e:
|
|
log.warning("Error enviando correo de notificación: {}".format(e))
|
|
ERRORS_DIR.mkdir(parents=True, exist_ok=True)
|
|
shutil.move(str(xml_path), str(ERRORS_DIR / xml_path.name))
|
|
log.info("📁 XML movido a errors/")
|
|
return
|
|
|
|
item_catalog = get_item_catalog(cardcode)
|
|
pricelist = get_price_list(cardcode)
|
|
prices = get_prices(pricelist) if pricelist else {}
|
|
log.info("Catálogo: {} items, precios: {} registros".format(len(item_catalog), len(prices)))
|
|
|
|
cardname = get_card_name(cardcode)
|
|
email_html = generate_email_html(data, cardcode, cardname, item_catalog, prices)
|
|
(HTML_DIR / f"{base}_email.html").write_text(email_html, encoding="utf-8")
|
|
log.info("Email HTML generado")
|
|
|
|
payload = to_draft_payload(data, cardcode=cardcode, city=city, county=county, item_catalog=item_catalog, prices=prices)
|
|
write_json(payload, SAP_B1_DIR / f"{base}_sap.json")
|
|
log.info("Payload guardado en data/sap_b1/")
|
|
|
|
det = data.get("detalle", [])
|
|
price_ok = True
|
|
for d in det:
|
|
upc = d.get("cod_upc", "") or ""
|
|
sap_item = item_catalog.get(upc, "")
|
|
sap_price = prices.get(sap_item) if sap_item else None
|
|
importe = d.get("importe", 0) or 0
|
|
qty = d.get("cantidad_unid", 1) or 1
|
|
precio_xml = round(importe / qty) if qty > 0 else 0
|
|
if not sap_item or sap_price is None or sap_price != precio_xml:
|
|
price_ok = False
|
|
log.warning("⚠️ {}: difiere (XML={} vs SAP={})".format(upc, precio_xml, sap_price))
|
|
|
|
if not price_ok:
|
|
email_path = HTML_DIR / f"{base}_email.html"
|
|
num_oc = cab.get("numero_oc", "") or base
|
|
try:
|
|
from mail_sender import send_validation_email
|
|
send_validation_email(
|
|
email_path,
|
|
subject=f"Validación OC {num_oc} — diferencias precio/catálogo",
|
|
)
|
|
log.info("Notificación enviada a MAIL_TO")
|
|
except Exception as e:
|
|
log.warning("Error enviando correo: {}".format(e))
|
|
|
|
create_orders(payload)
|
|
except ImportError as e:
|
|
log.warning("⚠️ SAP saltado (dependencia faltante: {})".format(e))
|
|
except Exception as e:
|
|
log.warning("❌ SAP error: {}".format(e))
|
|
|
|
PROCESSED_DIR.mkdir(parents=True, exist_ok=True)
|
|
shutil.move(str(xml_path), str(PROCESSED_DIR / xml_path.name))
|
|
log.info("📁 XML movido a processed/")
|
|
|
|
# ============================================================================
|
|
def process_all():
|
|
INPUT_DIR.mkdir(parents=True, exist_ok=True)
|
|
files = sorted(f for f in INPUT_DIR.iterdir() if f.is_file())
|
|
if not files:
|
|
return
|
|
log.info("📦 Archivos pendientes: {}".format(len(files)))
|
|
for f in files:
|
|
try:
|
|
process_file(f)
|
|
except Exception as e:
|
|
log.error("❌ Error con {}: {}".format(f.name, e))
|
|
|
|
# ============================================================================ |