se agrega fallback OWTR y calculo de fecha de vencimiento por condicion de pago
- 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>
This commit is contained in:
1 parent
6b0ed6f519
commit
15e3b5079c
231 files changed
+21421
-475
No files matched your search
@@ -0,0 +1,76 @@
|
||||
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()
|
||||
Reference in new issue
Block a user