- hana_manager.get_delivery ahora busca en ODLN y, si no encuentra, en OWTR - sapb1_manager resuelve StockTransfers -> Orders (via U_IDC_NUMOV) para validar precios de flete cuando la guia es un traslado de inventario - se corrige el orden de parametros tipo/serie/numero (sin cruces) en get_delivery, get_delivery_notes y _get_delivery_entries - se agrega sapb1_manager.obtener_condicion_pago (BusinessPartners + PaymentTermsTypes) y pipeline._calcular_fecha_vencimiento para que DocDueDate del draft sea fecha de contabilizacion + dias de la condicion de pago del proveedor - se agrega variante de regex para "VALOR REFER" truncado en sunat_schema.py - se actualiza documentacion (docs/) reflejando estos cambios Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
77 lines
2.3 KiB
Python
77 lines
2.3 KiB
Python
import json
|
|
import os
|
|
from glob import glob
|
|
|
|
CARPETA_BASE = os.path.join(os.path.dirname(__file__), "correos_prueba")
|
|
|
|
|
|
def _doc_key(d):
|
|
return (d.get("tipo"), d.get("serie"), d.get("numero"))
|
|
|
|
|
|
def completar_guias_desde_pdf(xml_path: str, pdf_path: str) -> int:
|
|
with open(xml_path, encoding="utf-8") as f:
|
|
xml_data = json.load(f)
|
|
try:
|
|
with open(pdf_path, encoding="utf-8") as f:
|
|
pdf_data = json.load(f)
|
|
except FileNotFoundError:
|
|
return 0
|
|
|
|
pdf_by_key = {_doc_key(d): d for d in pdf_data}
|
|
total_copiadas = 0
|
|
|
|
for xd in xml_data:
|
|
k = _doc_key(xd)
|
|
|
|
has_all_guias = all(
|
|
len(linea.get("guias_remision") or []) > 0
|
|
for linea in xd.get("lineas", [])
|
|
)
|
|
if has_all_guias:
|
|
continue
|
|
|
|
pd = pdf_by_key.get(k)
|
|
if not pd:
|
|
continue
|
|
|
|
for li, xl in enumerate(xd.get("lineas", [])):
|
|
if len(xl.get("guias_remision") or []) > 0:
|
|
continue
|
|
pl = pd["lineas"][li] if li < len(pd.get("lineas", [])) else {}
|
|
p_gr = pl.get("guias_remision") or []
|
|
if p_gr:
|
|
xl["guias_remision"] = list(p_gr)
|
|
total_copiadas += len(p_gr)
|
|
# También copiar guia_transportista si XML no tiene
|
|
if not xl.get("guia_transportista") and pl.get("guia_transportista"):
|
|
xl["guia_transportista"] = dict(pl["guia_transportista"])
|
|
|
|
if total_copiadas > 0:
|
|
with open(xml_path, "w", encoding="utf-8") as f:
|
|
json.dump(xml_data, f, ensure_ascii=False, indent=2)
|
|
|
|
return total_copiadas
|
|
|
|
|
|
def main():
|
|
carpetas = sorted(glob(os.path.join(CARPETA_BASE, "[0-9][0-9][0-9]_correo")))
|
|
total_global = 0
|
|
for fp in carpetas:
|
|
folder = os.path.basename(fp)
|
|
idx = folder[:3]
|
|
xml_path = os.path.join(fp, f"xml_resultado_{idx}.json")
|
|
pdf_path = os.path.join(fp, f"pdf_resultado_{idx}.json")
|
|
if not os.path.exists(xml_path):
|
|
continue
|
|
n = completar_guias_desde_pdf(xml_path, pdf_path)
|
|
if n > 0:
|
|
print(f" {folder}: {n} guías copiadas")
|
|
total_global += n
|
|
|
|
print(f"\nTotal guías copiadas: {total_global}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|