81 lines
3.2 KiB
Python
81 lines
3.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Side exercise: DTE (factura) <-> payment reconciliation.
|
|
Reads the existing SII-RCV cross-reference CSV and emits web/reconciliation.json
|
|
(normalized rows + summary) for the standalone reconciliation view."""
|
|
import csv, json, os, datetime
|
|
|
|
SRC = 'reference/contabilidad_cruces/dte_pagos_muralla_murallita_rcv_hasta_2026-04_pagos_hasta_2025-12.csv'
|
|
OUT = 'web/reconciliation.json'
|
|
|
|
def toint(v):
|
|
try:
|
|
return int(str(v or '0').replace('.', '').replace(',', '').strip() or 0)
|
|
except Exception:
|
|
return 0
|
|
|
|
rows = list(csv.DictReader(open(SRC, encoding='utf-8-sig'), delimiter=';'))
|
|
|
|
out_rows = []
|
|
for r in rows:
|
|
estado = (r.get('estado_asignacion') or '').strip()
|
|
out_rows.append({
|
|
'empresa': (r.get('empresa') or '').strip(),
|
|
'periodo': (r.get('periodo_rcv') or '').strip(),
|
|
'fecha_docto': (r.get('fecha_docto') or '').strip(),
|
|
'tipo_doc': (r.get('tipo_doc') or '').strip(),
|
|
'folio': (r.get('folio') or '').strip(),
|
|
'rut_proveedor': (r.get('rut_proveedor') or '').strip(),
|
|
'razon_social': (r.get('razon_social') or '').strip(),
|
|
'monto_neto': toint(r.get('monto_neto')),
|
|
'iva': toint(r.get('iva_recuperable')),
|
|
'monto_total': toint(r.get('monto_total')),
|
|
'estado': estado,
|
|
'quien_pago': (r.get('quien_pago') or '').strip(),
|
|
'tipo_pagador': (r.get('tipo_pagador') or '').strip(),
|
|
'origen_fondos': (r.get('origen_fondos') or '').strip(),
|
|
'instrumento_pago': (r.get('instrumento_pago') or '').strip(),
|
|
'banco_cuenta': (r.get('banco_cuenta_origen') or '').strip(),
|
|
'fecha_pago': (r.get('fecha_pago') or '').strip(),
|
|
'monto_pago': toint(r.get('monto_pago')),
|
|
'confianza': (r.get('confianza') or '').strip(),
|
|
'notas': (r.get('notas') or '').strip(),
|
|
})
|
|
|
|
def bucket(predicate):
|
|
rs = [x for x in out_rows if predicate(x)]
|
|
return {'n': len(rs), 'monto': sum(x['monto_total'] for x in rs)}
|
|
|
|
def group(key):
|
|
g = {}
|
|
for x in out_rows:
|
|
k = x[key] or '(sin)'
|
|
g.setdefault(k, {'n': 0, 'monto': 0})
|
|
g[k]['n'] += 1
|
|
g[k]['monto'] += x['monto_total']
|
|
return dict(sorted(g.items(), key=lambda kv: -kv[1]['monto']))
|
|
|
|
summary = {
|
|
'total_dtes': len(out_rows),
|
|
'total_facturado': sum(x['monto_total'] for x in out_rows),
|
|
'iva_total': sum(x['iva'] for x in out_rows),
|
|
'asignado': bucket(lambda x: x['estado'] == 'Asignado'),
|
|
'pendiente': bucket(lambda x: x['estado'] == 'Pendiente'),
|
|
'sin_pago': bucket(lambda x: x['estado'] == 'Sin pago requerido'),
|
|
'con_pago_linked': bucket(lambda x: x['monto_pago'] > 0),
|
|
'by_empresa': group('empresa'),
|
|
'by_estado': group('estado'),
|
|
'by_pagador': group('tipo_pagador'),
|
|
}
|
|
|
|
data = {
|
|
'generated': datetime.date.today().isoformat(),
|
|
'source': os.path.basename(SRC),
|
|
'summary': summary,
|
|
'rows': out_rows,
|
|
}
|
|
json.dump(data, open(OUT, 'w'), ensure_ascii=False)
|
|
print(f'wrote {OUT}: {len(out_rows)} DTEs, ${summary["total_facturado"]:,} facturado')
|
|
print(f' asignado {summary["asignado"]["n"]} (${summary["asignado"]["monto"]:,}) | '
|
|
f'pendiente {summary["pendiente"]["n"]} (${summary["pendiente"]["monto"]:,}) | '
|
|
f'sin pago {summary["sin_pago"]["n"]}')
|