#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Conciliacion v2 — fuente autoritativa = diario.csv (extracto bancario nivel linea, con SIGNO real y pedido conciliado en 'Apuntes contables'). Supera a build_preview.py (que usaba general.csv, todo positivo). NO escribe en PocketBase. Genera preview_v2_*.csv. """ import csv, re, os from collections import defaultdict from datetime import date, timedelta HERE = os.path.dirname(os.path.abspath(__file__)) PED = re.compile(r"(? cuenta app "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", } 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", } ANON = {"Consumidor Final Anónimo", "", "VARIOS"} EDD = "de los Santos, Edelmar" def D(s): s = (s or "").strip() return (date(1899,12,30)+timedelta(days=int(float(s)))).isoformat() if s else "" def serial(s): s=(s or "").strip(); return int(float(s)) if s else 0 def num(s): s=(s or "").strip(); return float(s) if s else 0.0 # ---- parsear diario.csv (agrupando sub-filas de apuntes) ---- raw = list(csv.reader(open(os.path.join(HERE,"diario.csv"), encoding="utf-8-sig")))[1:] L = [] cur = None for r in raw: if len(r) < 18: r = r + [""]*(18-len(r)) if r[0].strip(): cur = {"diario": r[6].strip(), "contacto": r[11].strip(), "imp": num(r[12]), "fecha": D(r[14]), "fser": serial(r[14]), "asiento": r[16].strip(), "peds": set(), "used": False} for c in r: for m in PED.findall(c): cur["peds"].add(m) L.append(cur) elif cur is not None: for c in r: for m in PED.findall(c): cur["peds"].add(m) L = [x for x in L if x["diario"] in CUENTA] # ---- detectar transferencias: par signo opuesto, mismo |monto| (±1), <=3 dias, distinta cuenta ---- # SOLO si ambas patas son cuentas propias/anonimas (un tercero nombrado = transaccion real, no transfer) OWN = {EDD} | ANON for a in L: if a["used"] or a["imp"] >= 0: continue # parto desde el egreso (negativo) if a["contacto"] not in OWN: continue for b in L: if b["used"] or b["imp"] <= 0: continue if b["diario"] == a["diario"]: continue if b["contacto"] not in OWN: continue if abs(abs(a["imp"]) - b["imp"]) > 1: continue if abs(a["fser"] - b["fser"]) > 3: continue a["xfer"] = b["diario"]; b["xfer_dup"] = True a["used"] = b["used"] = True break # ---- construir movimientos ---- movs = [] def cat_ingreso_placeholder(): return "Venta directa" for x in L: cta = CUENTA[x["diario"]] monto = round(abs(x["imp"])) ped = ";".join(sorted(x["peds"])) c = x["contacto"] rev = "" if x.get("xfer"): mv = {"tipo":"transferencia","cuenta":cta,"cuenta_destino":CUENTA[x["xfer"]], "categoria":"","pedido":"","contacto":c,"metodo":"transferencia","conf":"media", "revisar":"posible transferencia (confirmar)","detalle":f"{cta}->{CUENTA[x['xfer']]}"} elif x.get("xfer_dup"): continue # el lado positivo del par ya se representa en la transferencia elif c == EDD: if x["imp"] < 0: mv = {"tipo":"egreso","cuenta":cta,"cuenta_destino":"","categoria":"Cuenta socio", "pedido":"","contacto":c,"metodo":"socio","conf":"alta","revisar":"","detalle":"Retiro socio"} else: # ingreso del dueño: "Cuenta socio" es categoria de egreso, uso Otros mv = {"tipo":"ingreso","cuenta":cta,"cuenta_destino":"","categoria":"Otros", "pedido":"","contacto":c,"metodo":"socio","conf":"baja", "revisar":"Edelmar INGRESO: aporte? (revisar)","detalle":"Aporte socio"} elif x["imp"] < 0: # egreso cat = "Sueldos / retiros" if c=="Anibal Alonso" else SUP_CAT.get(c,"Otros") mv = {"tipo":"egreso","cuenta":cta,"cuenta_destino":"","categoria":cat, "pedido":ped,"contacto":c,"metodo":"egreso","conf":"alta", "revisar":"" if (c in SUP_CAT or c==EDD) else "categoria?","detalle":c} else: # ingreso if ped: mv = {"tipo":"ingreso","cuenta":cta,"cuenta_destino":"","categoria":"", "pedido":ped,"contacto":c,"metodo":"venta-link","conf":"alta","revisar":"","detalle":c} else: big = monto >= 150000 and c in ANON mv = {"tipo":"ingreso","cuenta":cta,"cuenta_destino":"","categoria":"Venta directa", "pedido":"","contacto":(c or "Consumidor Final"),"metodo":"ingreso-sin-link", "conf":"baja" if big else "media", "revisar":"PENDIENTE: venta s/pedido (anon grande)" if big else "venta sin pedido","detalle":c} mv.update({"fecha":x["fecha"],"monto":monto,"fser":x["fser"],"odoo":x["asiento"]}) movs.append(mv) # ---- categoria seña/saldo para ingresos con pedido ---- byp = defaultdict(list) for m in movs: if m["tipo"]=="ingreso" and m["pedido"]: byp[m["pedido"].split(";")[0]].append(m) for ped, ms in byp.items(): ms.sort(key=lambda x:(x["fser"], x["monto"])) n=len(ms) for i,m in enumerate(ms): m["categoria"] = "Venta directa" if n==1 else ("Seña" if i==0 else ("Saldo final" if i==n-1 else "Pago parcial")) # ---- merge: PMP de general que NO esten en diario (POS sin linea de cuenta) ---- gen = list(csv.DictReader(open(os.path.join(HERE,"general.csv"), encoding="utf-8-sig"))) pmp = [g for g in gen if g["Diario"]=="Mercado Pago" and g["Número"].strip().startswith("PMP")] mp_diario = [x for x in L if x["diario"]=="Mercado Pago"] descart = [] for g in pmp: gimp = abs(num(g["Total firmado"])); gser = serial(g["Fecha"]) twin = any(abs(abs(x["imp"])-gimp)<1 and abs(x["fser"]-gser)<=2 for x in mp_diario) if twin: descart.append((D(g["Fecha"]),"Mercadopago",round(gimp),g["Contacto"],g["Número"],"dup POS (PMP=linea MP)")) else: movs.append({"tipo":"ingreso","cuenta":"Mercadopago","cuenta_destino":"","categoria":"Venta directa", "pedido":"","contacto":g["Contacto"],"metodo":"PMP-sin-linea","conf":"baja", "revisar":"PMP POS sin linea de cuenta - verificar","detalle":g["Contacto"], "fecha":D(g["Fecha"]),"monto":round(gimp),"fser":gser,"odoo":g["Número"].strip()}) # ---- escribir ---- cols=["fecha","cuenta","cuenta_destino","tipo","monto","categoria","numero_pedido","contacto","detalle","metodo","confianza","revisar","numero_odoo"] def rowify(m): return {"fecha":m["fecha"],"cuenta":m["cuenta"],"cuenta_destino":m.get("cuenta_destino",""), "tipo":m["tipo"],"monto":m["monto"],"categoria":m.get("categoria",""),"numero_pedido":m.get("pedido",""), "contacto":m.get("contacto",""),"detalle":m.get("detalle",""),"metodo":m["metodo"], "confianza":m["conf"],"revisar":m.get("revisar",""),"numero_odoo":m.get("odoo","")} movs.sort(key=lambda x:(x["fser"], x["cuenta"])) with open(os.path.join(HERE,"preview_v2_movimientos.csv"),"w",encoding="utf-8-sig",newline="") as f: w=csv.DictWriter(f,fieldnames=cols); w.writeheader(); w.writerows(rowify(m) for m in movs) with open(os.path.join(HERE,"preview_v2_revisar.csv"),"w",encoding="utf-8-sig",newline="") as f: w=csv.DictWriter(f,fieldnames=cols); w.writeheader() w.writerows(rowify(m) for m in movs if m.get("revisar") or m["conf"] in ("baja",)) with open(os.path.join(HERE,"preview_v2_descartados.csv"),"w",encoding="utf-8-sig",newline="") as f: w=csv.writer(f); w.writerow(["fecha","cuenta","monto","contacto","numero_odoo","motivo"]); w.writerows(descart) # ---- resumen ---- def S(p): return sum(m["monto"] for m in movs if p(m)) ni=sum(1 for m in movs if m["tipo"]=="ingreso"); ne=sum(1 for m in movs if m["tipo"]=="egreso") nx=sum(1 for m in movs if m["tipo"]=="transferencia") print(f"TOTAL movimientos: {len(movs)} (descartados dup POS: {len(descart)})") print(f" INGRESOS {ni:>4} ${S(lambda m:m['tipo']=='ingreso'):>13,.0f}") print(f" EGRESOS {ne:>4} ${S(lambda m:m['tipo']=='egreso'):>13,.0f}") print(f" TRANSFEREN. {nx:>4} ${S(lambda m:m['tipo']=='transferencia'):>13,.0f} (pares internos, no son venta/gasto)") print(f" con pedido linkeado: {sum(1 for m in movs if m['pedido'])}") print(f" Cuenta socio (retiros/aportes): ${S(lambda m:m.get('categoria')=='Cuenta socio'):,.0f}") edd_pos=[m for m in movs if m['contacto']==EDD and m['tipo']=='ingreso'] print(f" Edelmar INGRESO (revisar aporte): {len(edd_pos)} ${sum(m['monto'] for m in edd_pos):,.0f}") anon_big=[m for m in movs if m['metodo']=='ingreso-sin-link' and m['conf']=='baja'] print(f" Ingresos anon GRANDES sin pedido (revisar): {len(anon_big)} ${sum(m['monto'] for m in anon_big):,.0f}") print(f" PARA REVISAR total: {sum(1 for m in movs if m.get('revisar') or m['conf']=='baja')}") print() print("Por cuenta (neto firmado ingreso-egreso, sin transfer):") for cta in ["Mercadopago","Efectivo","Cuenta DNI","Banco Macro","Banco Nación","Cocos"]: neto=sum((m["monto"] if m["tipo"]=="ingreso" else -m["monto"]) for m in movs if m["cuenta"]==cta and m["tipo"]!="transferencia") n=sum(1 for m in movs if m["cuenta"]==cta) print(f" {cta:<13} {n:>4} mov neto ${neto:>12,.0f}") print() md=defaultdict(int) for m in movs: md[m["metodo"]]+=1 print("Metodos:", dict(sorted(md.items(), key=lambda x:-x[1])))