#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Conciliacion Caja Fibromuebles — ene-mar 2026. Lee general.csv (export Odoo account.move) y propone los movimientos de Caja para la app nueva, cruzando cobros de plata contra documentos de venta/compra. NO escribe en PocketBase. Solo genera CSVs de preview para revisar. """ import csv, os from datetime import date, timedelta from collections import defaultdict, Counter HERE = os.path.dirname(os.path.abspath(__file__)) SRC = os.path.join(HERE, "general.csv") # ---- diarios -> rol ---- TREASURY = { # diario Odoo -> cuenta canonica app (CUENTAS_INICIALES) "Mercado Pago": "Mercadopago", "Efectivo de la mueblería": "Efectivo", "Cuenta DNI": "Cuenta DNI", "Banco Macro": "Banco Macro", "Banco Nación": "Banco Nación", "Cocos": "Cocos", } SALE_J = {"Control X", "Factura Electrónica", "Punto de venta"} BUY_J = "Compras" MISC_J = "Operaciones misceláneas" # ---- proveedor -> categoria egreso (EGRESO_CATEGORIAS) ---- SUP_CAT = { "Meta": "Marketing / publicidad", "Google": "Marketing / publicidad", "ARCA": "Impuestos", "MBB": "Impuestos", "EDES": "Servicios", "Transporte": "Fletes / envíos", "Vaquero Herrajes": "Herrajes", "Ciarrocchi HNOS.": "Materiales", "Ferreteria Corrientes": "Materiales", "MercadoLibre": "Materiales", "Odoo": "Software / sistemas", "Anibal Alonso": "Sueldos / retiros", # Edd: es sueldo } ANON = {"Consumidor Final Anónimo", ""} def serial_to_date(s): return (date(1899, 12, 30) + timedelta(days=int(float(s)))).isoformat() def num(x): x = (x or "").strip() return float(x) if x else 0.0 # ---- cargar ---- rows = [] with open(SRC, encoding="utf-8-sig") as f: for r in csv.DictReader(f): rows.append({ "fecha": serial_to_date(r["Fecha"]), "fserial": int(float(r["Fecha"])), "numero": r["Número"].strip(), "contacto": r["Contacto"].strip(), "ref": r["Referencia"].strip(), "diario": r["Diario"].strip(), "total": num(r["Total firmado"]), "estado": r["Estado"].strip(), }) treasury = [r for r in rows if r["diario"] in TREASURY] sales = [r for r in rows if r["diario"] in SALE_J and r["ref"]] buys = [r for r in rows if r["diario"] == BUY_J] misc = [r for r in rows if r["diario"] == MISC_J] for t in treasury: t["used"] = False t["mov"] = None # se completa abajo # ---- dedup POS: PMP (pago POS de Mercado Pago) duplica la linea MP de la cuenta; me quedo con MP ---- _mp = [t for t in treasury if t["numero"].startswith("MP/")] descartados = [] for t in treasury: if t["numero"].startswith("PMP"): for m in _mp: if m["fserial"] == t["fserial"] and abs(abs(m["total"]) - abs(t["total"])) < 1: t["drop"] = True descartados.append(t) break treasury = [t for t in treasury if not t.get("drop")] def best_match(doc, pool, name_match, amt_abs_tol, amt_pct_tol, day_win): best, bestd = None, None a = abs(doc["total"]) for t in pool: if t["used"]: continue if name_match and t["contacto"] != doc["contacto"]: continue if (not name_match) and (t["contacto"] not in ANON): continue b = abs(t["total"]) if abs(a - b) > amt_abs_tol and (a == 0 or abs(a - b) / max(a, b) > amt_pct_tol): continue dd = abs(t["fserial"] - doc["fserial"]) if dd > day_win: continue if bestd is None or dd < bestd: best, bestd = t, dd return best # ---- pase 1: ventas con cliente nombrado ---- for d in sorted([s for s in sales if s["contacto"] not in ANON], key=lambda x: x["fserial"]): m = best_match(d, treasury, True, 5, 0.02, 45) if m: m["used"] = True m["mov"] = {"tipo": "ingreso", "pedido": d["ref"], "contacto": d["contacto"], "metodo": "venta-nombre", "conf": "alta"} # ---- pase 2: ventas anonimas (monto+fecha estricto) ---- for d in sorted([s for s in sales if s["contacto"] in ANON], key=lambda x: x["fserial"]): m = best_match(d, treasury, False, 1, 0.0, 2) if m: m["used"] = True m["mov"] = {"tipo": "ingreso", "pedido": d["ref"], "contacto": "Consumidor Final", "metodo": "venta-anon", "conf": "media"} # ---- pase 3: compras (egresos) ---- for d in sorted(buys, key=lambda x: x["fserial"]): m = best_match(d, treasury, True, 5, 0.02, 45) if m: m["used"] = True cat = SUP_CAT.get(d["contacto"], "Otros") m["mov"] = {"tipo": "egreso", "pedido": "", "contacto": d["contacto"], "categoria": cat, "metodo": "compra", "conf": "alta", "revisar": "" if d["contacto"] in SUP_CAT else "categoria?"} # ---- pase 4: lo que quedo sin matchear ---- sup_names = {b["contacto"] for b in buys} cli_names = {s["contacto"] for s in sales if s["contacto"] not in ANON} for t in treasury: if t["mov"]: continue c = t["contacto"] if c == "de los Santos, Edelmar": t["mov"] = {"tipo": "egreso", "pedido": "", "contacto": c, "categoria": "Cuenta socio", "metodo": "retiro-socio", "conf": "alta", "detalle": "Retiro socio"} elif c in sup_names: t["mov"] = {"tipo": "egreso", "pedido": "", "contacto": c, "categoria": SUP_CAT.get(c, "Otros"), "metodo": "egreso-sin-doc", "conf": "baja", "revisar": "egreso sin factura?"} elif c in ANON: t["mov"] = {"tipo": "ingreso", "pedido": "", "contacto": "Consumidor Final", "categoria": "Venta directa", "metodo": "ingreso-anon-suelto", "conf": "baja", "revisar": "PENDIENTE archivo Odoo"} else: t["mov"] = {"tipo": "ingreso", "pedido": "", "contacto": c, "categoria": "Venta directa", "metodo": "ingreso-suelto", "conf": "media", "revisar": "venta sin pedido (ok Edd)"} # ---- categoria seña/saldo para ingresos con pedido ---- byped = defaultdict(list) for t in treasury: mv = t["mov"] if mv and mv["tipo"] == "ingreso" and mv["pedido"]: byped[mv["pedido"]].append(t) for ped, ts in byped.items(): ts.sort(key=lambda x: (x["fserial"], abs(x["total"]))) n = len(ts) for i, t in enumerate(ts): if n == 1: t["mov"]["categoria"] = "Venta directa" elif i == 0: t["mov"]["categoria"] = "Seña" elif i == n - 1: t["mov"]["categoria"] = "Saldo final" else: t["mov"]["categoria"] = "Pago parcial" # ---- detectar posibles duplicados (artefacto POS de Odoo) ---- dupkey = lambda t: (t["fecha"], TREASURY[t["diario"]], round(abs(t["total"])), t["contacto"]) dcnt = Counter(dupkey(t) for t in treasury) n_dup = 0 for t in treasury: if dcnt[dupkey(t)] > 1: n_dup += 1 r = t["mov"].get("revisar", "") t["mov"]["revisar"] = (r + " | posible duplicado").strip(" |") # ---- escribir preview ---- cols = ["fecha", "cuenta", "tipo", "monto", "categoria", "numero_pedido", "contacto", "detalle", "metodo", "confianza", "revisar", "numero_odoo"] out = os.path.join(HERE, "preview_movimientos.csv") rev = os.path.join(HERE, "preview_revisar.csv") allmovs = [] for t in sorted(treasury, key=lambda x: (x["fserial"], x["diario"])): mv = t["mov"] allmovs.append({ "fecha": t["fecha"], "cuenta": TREASURY[t["diario"]], "tipo": mv["tipo"], "monto": round(abs(t["total"])), "categoria": mv.get("categoria", ""), "numero_pedido": mv.get("pedido", ""), "contacto": mv.get("contacto", ""), "detalle": mv.get("detalle", ""), "metodo": mv.get("metodo", ""), "confianza": mv.get("conf", ""), "revisar": mv.get("revisar", ""), "numero_odoo": t["numero"], }) with open(out, "w", encoding="utf-8-sig", newline="") as f: w = csv.DictWriter(f, fieldnames=cols); w.writeheader(); w.writerows(allmovs) with open(rev, "w", encoding="utf-8-sig", newline="") as f: w = csv.DictWriter(f, fieldnames=cols); w.writeheader() w.writerows([m for m in allmovs if m["revisar"] or m["confianza"] in ("baja", "media")]) with open(os.path.join(HERE, "preview_descartados.csv"), "w", encoding="utf-8-sig", newline="") as f: w = csv.writer(f); w.writerow(["fecha", "cuenta", "monto", "contacto", "numero_odoo", "motivo"]) for t in descartados: w.writerow([t["fecha"], TREASURY[t["diario"]], round(abs(t["total"])), t["contacto"], t["numero"], "duplicado POS (PMP=MP)"]) # ---- resumen ---- def s(p): return sum(m["monto"] for m in allmovs if p(m)) ingr = s(lambda m: m["tipo"] == "ingreso") egr = s(lambda m: m["tipo"] == "egreso") socio = s(lambda m: m["categoria"] == "Cuenta socio") con_ped = sum(1 for m in allmovs if m["numero_pedido"]) flagged = sum(1 for m in allmovs if m["revisar"] or m["confianza"] in ("baja", "media")) print(f"TOTAL movimientos propuestos: {len(allmovs)} (de {len(treasury)} lineas de plata)") print(f" DESCARTADOS por dup POS (PMP=MP): {len(descartados)} (${sum(abs(t['total']) for t in descartados):,.0f})") print(f" ingresos: {sum(1 for m in allmovs if m['tipo']=='ingreso')} = ${ingr:,.0f}") print(f" egresos : {sum(1 for m in allmovs if m['tipo']=='egreso')} = ${egr:,.0f}") print(f" de los cuales Cuenta socio (retiros): ${socio:,.0f}") print(f" con pedido linkeado: {con_ped}") print(f" PARA REVISAR (baja/media conf o flag): {flagged}") print(f" POSIBLES DUPLICADOS (mismo dia+cuenta+monto+contacto): {n_dup} filas") sale_contacts = {s_['contacto'] for s_ in sales} ns = [t for t in treasury if t['mov'].get('metodo') == 'ingreso-suelto'] ns_nodoc = sum(1 for t in ns if t['contacto'] not in sale_contacts) print(f" ingresos-suelto con nombre que NUNCA aparecen en docs venta: {ns_nodoc}/{len(ns)}") print() print("Por cuenta:") for cta in ["Mercadopago","Efectivo","Cuenta DNI","Banco Macro","Banco Nación","Cocos"]: n_ = sum(1 for m in allmovs if m["cuenta"]==cta) tot = sum(m["monto"] for m in allmovs if m["cuenta"]==cta) print(f" {cta:<14} {n_:>4} mov ${tot:>12,.0f}") print() print("Chequeo conciliacion (vs documentos Odoo):") print(f" docs venta total Odoo : ${sum(abs(s_['total']) for s_ in sales):,.0f}") print(f" ingresos matcheados : ${s(lambda m: m['tipo']=='ingreso' and m['numero_pedido']):,.0f}") print(f" compras total Odoo : ${sum(abs(b['total']) for b in buys):,.0f}") print(f" egresos no-socio : ${egr-socio:,.0f}") print() print("Metodos:") md = defaultdict(int) for m in allmovs: md[m["metodo"]] += 1 for k,v in sorted(md.items(), key=lambda x:-x[1]): print(f" {k:<22} {v}")