Copiar, pegar en Notepad, guardar como UTF-8. Orden: python analizar_llave.py dia1.csv dia2.csv --salida reporte_llave y luego python generar_config.py reporte_llave.json. Ninguno de los dos saca valores de la data.
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
analizar_llave.py
=================
Analisis de llave candidata para reportes acumulativos (GL-JE, Trial Balance).
QUE HACE
1. Perfila cada columna: nulos, cardinalidad, tipo inferido, formato.
2. Busca combinaciones de columnas que identifiquen la fila de forma unica.
3. Si le das DOS archivos de dias consecutivos, prueba si esa llave es ESTABLE
entre descargas y separa empiricamente columnas de IDENTIDAD vs ATRIBUTO.
4. Recomienda: llave natural, o esquema de doble hash.
PRIVACIDAD
El reporte NO contiene valores de la data. Solo nombres de columna, conteos,
porcentajes y firmas de formato (9 = digito, A = letra). Revisa el output
antes de compartirlo. Opciones:
--sin-patrones omite las firmas de formato
--anonimizar-columnas reemplaza nombres por COL_01, COL_02...
(guarda el mapeo en un archivo local aparte)
USO
python analizar_llave.py dia1.csv
python analizar_llave.py dia1.csv dia2.csv
python analizar_llave.py dia1.xlsx dia2.xlsx --salida reporte.json --max-combo 4
REQUISITOS
Python 3.8+. Solo stdlib para CSV.
Para Excel usa pandas u openpyxl si estan disponibles.
"""
from __future__ import annotations
import argparse
import csv
import hashlib
import json
import os
import re
import sys
from collections import Counter
from datetime import datetime
from decimal import Decimal, InvalidOperation
from itertools import combinations
SEP = "\x1f"
NULO = "<NULL>"
MESES = {
"JAN": 1, "FEB": 2, "MAR": 3, "APR": 4, "MAY": 5, "JUN": 6,
"JUL": 7, "AUG": 8, "SEP": 9, "OCT": 10, "NOV": 11, "DEC": 12,
"ENE": 1, "ABR": 4, "AGO": 8, "DIC": 12,
}
FORMATOS_FECHA = ("%Y-%m-%d", "%m/%d/%Y", "%d/%m/%Y", "%Y/%m/%d", "%Y%m%d", "%m-%d-%Y")
# ---------------------------------------------------------------- utilidades
def es_vacio(v) -> bool:
return v is None or str(v).strip() == ""
def como_decimal(s: str):
t = str(s).strip().replace(",", "").replace("$", "")
neg = False
if t.startswith("(") and t.endswith(")"):
neg, t = True, t[1:-1]
if t.endswith("-"):
neg, t = True, t[:-1]
try:
d = Decimal(t)
except (InvalidOperation, ValueError):
return None
return -d if neg else d
def como_fecha(s: str):
t = str(s).strip().upper()
if not t:
return None
t = t.split(" ")[0].split("T")[0]
m = re.match(r"^(\d{1,2})[-/]([A-Z]{3})[-/](\d{2,4})$", t) # 15-AUG-2026
if m and m.group(2) in MESES:
d, mes, a = int(m.group(1)), MESES[m.group(2)], int(m.group(3))
a = a + (2000 if a < 100 else 0)
try:
return datetime(a, mes, d).date()
except ValueError:
return None
for f in FORMATOS_FECHA:
try:
return datetime.strptime(t, f).date()
except ValueError:
continue
return None
def canon(v, tipo: str) -> str:
"""Valor canonico para comparar y hashear. Debe ser identico entre descargas."""
if es_vacio(v):
return NULO
s = str(v).strip()
if tipo == "num":
d = como_decimal(s)
return str(d.quantize(Decimal("0.01"))) if d is not None else " ".join(s.upper().split())
if tipo == "fecha":
f = como_fecha(s)
return f.isoformat() if f else " ".join(s.upper().split())
return " ".join(s.upper().split())
def firma_formato(s: str) -> str:
"""'ABC-123' -> 'AAA-999'. Colapsa repeticiones: 'AAAA999' -> 'A{4}9{3}'."""
if es_vacio(s):
return NULO
bruto = "".join("9" if c.isdigit() else "A" if c.isalpha() else c for c in str(s).strip())
out, i = [], 0
while i < len(bruto):
j = i
while j < len(bruto) and bruto[j] == bruto[i]:
j += 1
n = j - i
out.append(bruto[i] if n == 1 else "%s{%d}" % (bruto[i], n))
i = j
return "".join(out)[:60]
def sha(texto: str) -> str:
return hashlib.sha256(texto.encode("utf-8")).hexdigest()
# ---------------------------------------------------------------- carga
def cargar(path: str, max_filas=None):
"""Devuelve (columnas, filas) donde filas es lista de listas de str."""
ext = os.path.splitext(path)[1].lower()
if ext in (".xlsx", ".xlsm", ".xls"):
return _cargar_excel(path, max_filas)
return _cargar_csv(path, max_filas)
def _cargar_csv(path, max_filas):
with open(path, "r", encoding="utf-8-sig", newline="", errors="replace") as fh:
muestra = fh.read(64 * 1024)
fh.seek(0)
try:
dialecto = csv.Sniffer().sniff(muestra, delimiters=",;\t|")
except csv.Error:
dialecto = csv.excel
lector = csv.reader(fh, dialecto)
try:
cols = [c.strip() for c in next(lector)]
except StopIteration:
return [], []
filas = []
ancho = len(cols)
for i, fila in enumerate(lector):
if max_filas and i >= max_filas:
break
if len(fila) < ancho:
fila = fila + [""] * (ancho - len(fila))
filas.append([str(x) for x in fila[:ancho]])
return cols, filas
def _cargar_excel(path, max_filas):
try:
import pandas as pd
df = pd.read_excel(path, dtype=str, nrows=max_filas)
cols = [str(c).strip() for c in df.columns]
filas = [["" if v is None or (isinstance(v, float) and v != v) else str(v)
for v in fila] for fila in df.itertuples(index=False, name=None)]
return cols, filas
except ImportError:
pass
try:
from openpyxl import load_workbook
except ImportError:
sys.exit("ERROR: para leer Excel necesitas pandas u openpyxl. "
"Alternativa: exporta el reporte a CSV.")
wb = load_workbook(path, read_only=True, data_only=True)
ws = wb[wb.sheetnames[0]]
it = ws.iter_rows(values_only=True)
cols = [str(c).strip() if c is not None else "" for c in next(it)]
filas = []
for i, fila in enumerate(it):
if max_filas and i >= max_filas:
break
filas.append(["" if v is None else str(v) for v in fila][:len(cols)])
wb.close()
return cols, filas
# ---------------------------------------------------------------- perfilado
def inferir_tipo(valores) -> str:
muestra = [v for v in valores if not es_vacio(v)][:2000]
if not muestra:
return "vacio"
n_num = sum(1 for v in muestra if como_decimal(v) is not None)
n_fec = sum(1 for v in muestra if como_fecha(v) is not None)
total = len(muestra)
if n_fec / total >= 0.95:
return "fecha"
if n_num / total >= 0.95:
return "num"
return "texto"
def perfilar(cols, filas, incluir_patrones=True):
n = len(filas)
perfil = []
columnas_canon = {}
for idx, c in enumerate(cols):
crudos = [f[idx] for f in filas]
tipo = inferir_tipo(crudos)
canonicos = [canon(v, tipo) for v in crudos]
columnas_canon[c] = canonicos
distintos = len(set(canonicos))
nulos = sum(1 for v in canonicos if v == NULO)
largos = [len(str(v)) for v in crudos if not es_vacio(v)]
info = {
"columna": c,
"posicion": idx,
"tipo_inferido": tipo,
"nulos": nulos,
"pct_nulos": round(100.0 * nulos / n, 3) if n else 0.0,
"distintos": distintos,
"cardinalidad": round(distintos / n, 6) if n else 0.0,
"largo_min": min(largos) if largos else 0,
"largo_max": max(largos) if largos else 0,
"constante": distintos <= 1,
"unica": distintos == n and nulos == 0,
}
if incluir_patrones:
pats = Counter(firma_formato(v) for v in crudos[:5000])
info["formatos_top"] = [{"formato": p, "n": k} for p, k in pats.most_common(3)]
perfil.append(info)
return perfil, columnas_canon
# ---------------------------------------------------------------- llaves
def evaluar_combo(columnas_canon, combo, n):
tuplas = list(zip(*[columnas_canon[c] for c in combo]))
cuentas = Counter(tuplas)
distintos = len(cuentas)
dups = n - distintos
tam = Counter(cuentas.values())
return {
"columnas": list(combo),
"grupos": distintos,
"filas_duplicadas": dups,
"pct_duplicadas": round(100.0 * dups / n, 4) if n else 0.0,
"es_unica": dups == 0,
"dist_tam_grupo": {str(k): v for k, v in sorted(tam.items())[:8]},
}
def buscar_llaves(perfil, columnas_canon, n, max_combo=4, top_cols=14,
umbral_dup=0.5, permitir_nulos=False, muestra=60000):
"""Busca combinaciones minimas que identifiquen la fila."""
elegibles = [
p["columna"] for p in perfil
if not p["constante"] and p["tipo_inferido"] != "vacio"
and (permitir_nulos or p["pct_nulos"] == 0.0)
]
elegibles.sort(key=lambda c: -len(set(columnas_canon[c])))
elegibles = elegibles[:top_cols]
# Filtro rapido sobre una muestra, verificacion final sobre el total.
usar_muestra = n > muestra
paso = max(1, n // muestra) if usar_muestra else 1
canon_muestra = ({c: v[::paso] for c, v in columnas_canon.items()}
if usar_muestra else columnas_canon)
n_muestra = len(next(iter(canon_muestra.values()))) if canon_muestra else 0
llaves, casi, minimas = [], [], []
for tam in range(1, max_combo + 1):
for combo in combinations(elegibles, tam):
if any(set(k).issubset(combo) for k in minimas):
continue # superset de una llave ya encontrada
r_m = evaluar_combo(canon_muestra, combo, n_muestra)
if r_m["pct_duplicadas"] > umbral_dup:
continue
r = evaluar_combo(columnas_canon, combo, n) if usar_muestra else r_m
if r["es_unica"]:
minimas.append(set(combo))
llaves.append(r)
elif r["pct_duplicadas"] <= umbral_dup:
casi.append(r)
if len(llaves) >= 12:
break
casi.sort(key=lambda r: (r["pct_duplicadas"], len(r["columnas"])))
return {
"columnas_evaluadas": elegibles,
"llaves_unicas": llaves[:12],
"casi_llaves": casi[:12],
}
# ---------------------------------------------------------------- comparacion
def comparar(cols1, canon1, n1, cols2, canon2, n2, combo, todas_cols):
"""Estabilidad de la llave entre dos snapshots + split identidad/atributo."""
comunes = [c for c in todas_cols if c in canon1 and c in canon2]
k1 = list(zip(*[canon1[c] for c in combo]))
k2 = list(zip(*[canon2[c] for c in combo]))
set1, set2 = set(k1), set(k2)
inter = set1 & set2
# indice de la primera aparicion de cada llave (para comparar atributos)
idx1, idx2 = {}, {}
for i, k in enumerate(k1):
idx1.setdefault(k, i)
for i, k in enumerate(k2):
idx2.setdefault(k, i)
cambios = Counter()
filas_con_cambio = 0
for k in inter:
i, j = idx1[k], idx2[k]
hubo = False
for c in comunes:
if canon1[c][i] != canon2[c][j]:
cambios[c] += 1
hubo = True
if hubo:
filas_con_cambio += 1
n_inter = len(inter) or 1
volatilidad = {c: round(100.0 * cambios.get(c, 0) / n_inter, 3) for c in comunes}
identidad = sorted([c for c in comunes if volatilidad[c] == 0.0])
atributo = sorted([c for c in comunes if volatilidad[c] > 0.0],
key=lambda c: -volatilidad[c])
return {
"llave": list(combo),
"llaves_dia1": len(set1),
"llaves_dia2": len(set2),
"persisten": len(inter),
"solo_dia1_DESAPARECIDAS": len(set1 - set2),
"solo_dia2_NUEVAS": len(set2 - set1),
"pct_persistencia": round(100.0 * len(inter) / len(set1), 3) if set1 else 0.0,
"llaves_con_algun_cambio": filas_con_cambio,
"pct_llaves_cambiadas": round(100.0 * filas_con_cambio / n_inter, 3),
"volatilidad_por_columna_pct": dict(sorted(volatilidad.items(), key=lambda x: -x[1])),
"columnas_IDENTIDAD_sugeridas": identidad,
"columnas_ATRIBUTO_sugeridas": atributo,
"acumulativo_confirmado": len(set1 - set2) == 0,
}
# ---------------------------------------------------------------- reporte
def texto(rep):
L = []
add = L.append
add("=" * 78)
add("ANALISIS DE LLAVE CANDIDATA")
add("=" * 78)
add("Este reporte no contiene valores de la data.")
add("")
for arch in rep["archivos"]:
add("ARCHIVO: %s" % arch["nombre"])
add(" filas: %-12s columnas: %s" % (arch["filas"], arch["columnas"]))
add("")
add(" %-34s %-7s %8s %10s %7s" % ("COLUMNA", "TIPO", "%NULOS", "DISTINTOS", "CARD"))
add(" " + "-" * 72)
for p in arch["perfil"]:
marca = " *UNICA*" if p["unica"] else (" (constante)" if p["constante"] else "")
add(" %-34s %-7s %7.2f%% %10d %7.4f%s" % (
p["columna"][:34], p["tipo_inferido"], p["pct_nulos"],
p["distintos"], p["cardinalidad"], marca))
add("")
b = rep["busqueda_llave"]
add("-" * 78)
add("LLAVES CANDIDATAS (sobre el primer archivo)")
add("-" * 78)
if b["llaves_unicas"]:
add(" UNICAS (100% de las filas identificadas):")
for r in b["llaves_unicas"]:
add(" + %s" % " + ".join(r["columnas"]))
else:
add(" NO se encontro ninguna combinacion 100% unica.")
add("")
if b["casi_llaves"]:
add(" CASI-LLAVES (duplicados < umbral; revisar si son transacciones")
add(" genuinamente identicas o si falta una columna):")
for r in b["casi_llaves"][:8]:
add(" ~ %-52s dup=%.3f%% tam_grupo=%s" % (
" + ".join(r["columnas"])[:52], r["pct_duplicadas"], r["dist_tam_grupo"]))
add("")
if rep.get("comparacion"):
add("-" * 78)
add("ESTABILIDAD ENTRE LOS DOS SNAPSHOTS")
add("-" * 78)
for cmp_ in rep["comparacion"]:
add(" Llave: %s" % " + ".join(cmp_["llave"]))
add(" dia1=%d dia2=%d persisten=%d (%.2f%%)" % (
cmp_["llaves_dia1"], cmp_["llaves_dia2"],
cmp_["persisten"], cmp_["pct_persistencia"]))
add(" NUEVAS en dia2: %d" % cmp_["solo_dia2_NUEVAS"])
add(" DESAPARECIDAS: %d %s" % (
cmp_["solo_dia1_DESAPARECIDAS"],
"<-- OK, acumulativo" if cmp_["acumulativo_confirmado"]
else "<-- ALERTA: la llave no es estable"))
add(" llaves con cambios: %d (%.2f%%)" % (
cmp_["llaves_con_algun_cambio"], cmp_["pct_llaves_cambiadas"]))
add("")
add(" IDENTIDAD sugerida (0% de cambio entre snapshots):")
add(" %s" % (", ".join(cmp_["columnas_IDENTIDAD_sugeridas"]) or "(ninguna)"))
add(" ATRIBUTO sugerida (cambian; van en attr_hash):")
for c, v in list(cmp_["volatilidad_por_columna_pct"].items()):
if v > 0:
add(" %-40s %6.2f%%" % (c[:40], v))
add("")
add("-" * 78)
add("VEREDICTO")
add("-" * 78)
for l in rep["veredicto"]:
add(" " + l)
add("")
return "\n".join(L)
def veredicto(rep):
v = []
b = rep["busqueda_llave"]
cmps = rep.get("comparacion") or []
estables = [c for c in cmps if c["acumulativo_confirmado"] and c["pct_persistencia"] >= 99.9]
if b["llaves_unicas"] and estables:
v.append("HAY LLAVE NATURAL. Usar: %s" % " + ".join(estables[0]["llave"]))
v.append("Es unica dentro del archivo y estable entre descargas.")
v.append("No hace falta hash de identidad; si el attr_hash para detectar CHANGED.")
elif b["llaves_unicas"] and cmps and not estables:
v.append("Hay combinacion unica dentro de UN archivo, pero NO es estable entre")
v.append("descargas (desaparecen llaves). No sirve como identidad.")
v.append("Ir al esquema de doble hash sobre las columnas de identidad sugeridas.")
elif b["llaves_unicas"]:
v.append("Hay combinacion unica en este archivo. FALTA VALIDAR con un segundo")
v.append("archivo de otro dia. Vuelve a correr con los dos archivos.")
else:
v.append("NO hay llave natural. Ir al esquema de doble hash:")
v.append(" identity_hash = sha256 de las columnas de IDENTIDAD (normalizadas)")
v.append(" attr_hash = sha256 de las columnas de ATRIBUTO")
if b["casi_llaves"]:
r = b["casi_llaves"][0]
v.append("Mejor base: %s (dup %.3f%%)" % (" + ".join(r["columnas"]), r["pct_duplicadas"]))
v.append("Los duplicados residuales se manejan por CONTEO, no por desempate.")
if cmps:
c = cmps[0]
if c["pct_llaves_cambiadas"] > 5:
v.append("")
v.append("ATENCION: %.1f%% de las llaves cambio de un dia a otro. En un GL eso"
% c["pct_llaves_cambiadas"])
v.append("es demasiado (el GL reversa, no edita). Revisa si el reporte se esta")
v.append("renderizando distinto entre descargas o si falta normalizar formatos.")
return v
# ---------------------------------------------------------------- main
def main():
ap = argparse.ArgumentParser(description="Analisis de llave candidata (no expone data).")
ap.add_argument("archivo1")
ap.add_argument("archivo2", nargs="?", help="Mismo reporte, otro dia. Muy recomendado.")
ap.add_argument("--salida", default="reporte_llave", help="Base para .txt y .json")
ap.add_argument("--max-combo", type=int, default=4)
ap.add_argument("--top-cols", type=int, default=14, help="Columnas de mayor cardinalidad a combinar")
ap.add_argument("--umbral-dup", type=float, default=0.5, help="%% max de duplicados en casi-llave")
ap.add_argument("--max-filas", type=int, default=None)
ap.add_argument("--permitir-nulos", action="store_true", help="Deja columnas con nulos como candidatas")
ap.add_argument("--sin-patrones", action="store_true", help="Omite firmas de formato")
ap.add_argument("--anonimizar-columnas", action="store_true")
args = ap.parse_args()
incluir_pat = not args.sin_patrones
rep = {"archivos": [], "generado": datetime.now().strftime("%Y-%m-%d %H:%M")}
print("Leyendo %s ..." % args.archivo1, file=sys.stderr)
cols1, filas1 = cargar(args.archivo1, args.max_filas)
if not filas1:
sys.exit("ERROR: archivo vacio o ilegible.")
mapa = {}
if args.anonimizar_columnas:
mapa = {c: "COL_%02d" % (i + 1) for i, c in enumerate(cols1)}
cols1 = [mapa[c] for c in cols1]
perfil1, canon1 = perfilar(cols1, filas1, incluir_pat)
rep["archivos"].append({"nombre": os.path.basename(args.archivo1),
"filas": len(filas1), "columnas": len(cols1), "perfil": perfil1})
print("Buscando llaves candidatas ...", file=sys.stderr)
rep["busqueda_llave"] = buscar_llaves(
perfil1, canon1, len(filas1), args.max_combo, args.top_cols,
args.umbral_dup, args.permitir_nulos)
if args.archivo2:
print("Leyendo %s ..." % args.archivo2, file=sys.stderr)
cols2, filas2 = cargar(args.archivo2, args.max_filas)
if args.anonimizar_columnas:
cols2 = [mapa.get(c, c) for c in cols2]
perfil2, canon2 = perfilar(cols2, filas2, incluir_pat)
rep["archivos"].append({"nombre": os.path.basename(args.archivo2),
"filas": len(filas2), "columnas": len(cols2), "perfil": perfil2})
rep["schema_drift"] = {
"solo_en_dia1": sorted(set(cols1) - set(cols2)),
"solo_en_dia2": sorted(set(cols2) - set(cols1)),
"orden_identico": cols1 == cols2,
}
b = rep["busqueda_llave"]
candidatos = ([r["columnas"] for r in b["llaves_unicas"][:3]] +
[r["columnas"] for r in b["casi_llaves"][:3]])
vistos, cmps = set(), []
print("Comparando snapshots ...", file=sys.stderr)
for combo in candidatos:
t = tuple(combo)
if t in vistos or not all(c in canon2 for c in combo):
continue
vistos.add(t)
cmps.append(comparar(cols1, canon1, len(filas1), cols2, canon2,
len(filas2), combo, cols1))
rep["comparacion"] = cmps
rep["veredicto"] = veredicto(rep)
txt = texto(rep)
with open(args.salida + ".txt", "w", encoding="utf-8") as fh:
fh.write(txt)
with open(args.salida + ".json", "w", encoding="utf-8") as fh:
json.dump(rep, fh, indent=2, ensure_ascii=False)
if mapa:
with open(args.salida + "_mapeo_columnas_LOCAL.json", "w", encoding="utf-8") as fh:
json.dump(mapa, fh, indent=2, ensure_ascii=False)
print("\n[!] Mapeo de columnas en %s_mapeo_columnas_LOCAL.json -- NO compartir."
% args.salida, file=sys.stderr)
print(txt)
print("\nGuardado: %s.txt y %s.json" % (args.salida, args.salida), file=sys.stderr)
if __name__ == "__main__":
main()
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
generar_config.py
=================
Convierte el reporte de analizar_llave.py en config/columnas.json.
Lee reporte_llave.json, escoge la mejor llave candidata, separa las columnas
en IDENTIDAD y ATRIBUTO segun la volatilidad medida entre los dos snapshots,
y mapea los nombres de columna del archivo a los campos semanticos internos.
USO
python generar_config.py reporte_llave.json
python generar_config.py reporte_llave.json --salida ../app/config/columnas.json
python generar_config.py reporte_llave.json --revisar (solo muestra, no escribe)
NO NECESITA LA DATA. Solo el JSON del reporte, que no contiene valores.
REGLA DE SEGURIDAD IMPORTANTE
Los montos (debit/credit) y los campos de negocio mutables NUNCA entran en
identity_hash, aunque el reporte diga que no cambiaron. Con una muestra de
dos dias es facil que un campo mutable salga con 0% de volatilidad por pura
casualidad. Si un monto entrara en la identidad, un cambio de monto se veria
como "desaparecio una fila y nacio otra" en vez de CHANGED, y el roll-forward
se rompe con excepciones VANISHED.
"""
from __future__ import annotations
import argparse
import json
import os
import re
import sys
# Campos semanticos que entiende la app (nco/config.py CAMPOS_SEMANTICOS)
SINONIMOS = {
"journal_batch": ["journalbatchname", "journalbatch", "batchname", "jebatch",
"batch", "lotename", "lote"],
"journal_name": ["journalname", "jename", "journalheadername", "headername",
"journal", "asiento"],
"line_number": ["linenumber", "lineno", "jelinenum", "jelinenumber",
"line", "seqnum", "linea", "numlinea"],
"source_sys": ["source", "jesource", "journalsource", "sourcename", "fuente"],
"category": ["category", "jecategory", "journalcategory", "categoria"],
"period": ["period", "accountingperiod", "periodname", "glperiod", "periodo"],
"accounting_date": ["accountingdate", "gldate", "effectivedate", "transactiondate",
"postdate", "posteddate", "fechacontable", "accountingdt"],
"created_date": ["createddate", "creationdate", "datecreated", "enteredon",
"fechacreacion", "createdon"],
"entity": ["entity", "legalentity", "company", "companycode", "entidad",
"ledger", "bu", "businessunit"],
"region": ["region", "area", "district", "zona", "distrito"],
"product": ["product", "productcode", "prodcode", "producto", "linea negocio"],
"cost_center": ["costcenter", "costcentre", "cc", "centrocosto", "department",
"dept", "centrodecosto"],
"account": ["account", "glaccount", "accountnumber", "naturalaccount",
"accountcode", "cuenta", "codigocuenta", "accountsegment"],
"description": ["description", "linedescription", "jelinedescription", "memo",
"narrative", "descripcion", "concepto", "reference", "referencia"],
"debit": ["entereddebit", "accounteddebit", "debit", "dr", "debitamount",
"debito", "debe", "montodebito"],
"credit": ["enteredcredit", "accountedcredit", "credit", "cr", "creditamount",
"credito", "haber", "montocredito"],
"currency": ["currency", "currencycode", "curr", "moneda", "divisa"],
}
# Nunca van en identity_hash, aunque midan 0% de volatilidad.
SIEMPRE_ATRIBUTO = {"debit", "credit", "description", "cost_center", "region", "product"}
# Tipos para la canonizacion (nco/canon.py)
TIPOS = {"debit": "num", "credit": "num",
"accounting_date": "fecha", "created_date": "fecha"}
def norm(s: str) -> str:
return re.sub(r"[^a-z0-9]", "", str(s).lower())
def mapear(columnas):
"""columna del archivo -> campo semantico. Devuelve (mapeo, sin_mapear, dudosos)."""
mapeo, usados, dudosos = {}, set(), []
normalizadas = {c: norm(c) for c in columnas}
# 1) coincidencia exacta con un sinonimo
for c, n in normalizadas.items():
for campo, syns in SINONIMOS.items():
if campo in usados:
continue
if n in [norm(s) for s in syns]:
mapeo[c] = campo; usados.add(campo); break
# 2) contencion (p.ej. "GL_ACCOUNTING_DATE" contiene "accountingdate")
for c, n in normalizadas.items():
if c in mapeo:
continue
mejor, largo = None, 0
for campo, syns in SINONIMOS.items():
if campo in usados:
continue
for s in syns:
ns = norm(s)
if len(ns) >= 4 and (ns in n or n in ns) and len(ns) > largo:
mejor, largo = campo, len(ns)
if mejor:
mapeo[c] = mejor; usados.add(mejor); dudosos.append((c, mejor))
sin_mapear = [c for c in columnas if c not in mapeo]
faltantes = [f for f in SINONIMOS if f not in usados]
return mapeo, sin_mapear, dudosos, faltantes
def escoger_llave(rep):
"""Devuelve (columnas_llave, entrada_de_comparacion, motivo)."""
b = rep.get("busqueda_llave", {})
cmps = rep.get("comparacion") or []
por_llave = {tuple(c["llave"]): c for c in cmps}
def estable(c):
return c and c.get("acumulativo_confirmado") and c.get("pct_persistencia", 0) >= 99.9
for r in b.get("llaves_unicas", []):
c = por_llave.get(tuple(r["columnas"]))
if estable(c):
return r["columnas"], c, "llave unica y estable"
for r in b.get("casi_llaves", []):
c = por_llave.get(tuple(r["columnas"]))
if estable(c):
return r["columnas"], c, ("casi-llave estable (dup %.3f%%, los duplicados "
"se manejan por occ_seq)" % r["pct_duplicadas"])
if cmps:
c = min(cmps, key=lambda x: x.get("solo_dia1_DESAPARECIDAS", 9e9))
return c["llave"], c, "NINGUNA candidata es estable -- revisar a mano"
if b.get("casi_llaves"):
return b["casi_llaves"][0]["columnas"], None, "sin segundo archivo: no se pudo validar"
return [], None, "no hay candidatas"
def main():
ap = argparse.ArgumentParser(description="Genera columnas.json desde el reporte de llave.")
ap.add_argument("reporte", help="reporte_llave.json")
ap.add_argument("--salida", default="../app/config/columnas.json")
ap.add_argument("--key-version", default="v1")
ap.add_argument("--revisar", action="store_true", help="Muestra el resultado sin escribir")
a = ap.parse_args()
with open(a.reporte, "r", encoding="utf-8") as fh:
rep = json.load(fh)
columnas = [p["columna"] for p in rep["archivos"][0]["perfil"]]
mapeo, sin_mapear, dudosos, faltantes = mapear(columnas)
llave, cmp_, motivo = escoger_llave(rep)
print("=" * 74)
print("GENERACION DE config/columnas.json")
print("=" * 74)
# ---- chequeos criticos --------------------------------------------
print("\nCHEQUEOS CRITICOS")
alertas = []
if cmp_:
des = cmp_.get("solo_dia1_DESAPARECIDAS", 0)
pct = cmp_.get("pct_llaves_cambiadas", 0.0)
print(" desaparecidas entre snapshots : %d %s"
% (des, "OK" if des == 0 else "<-- ALERTA"))
print(" llaves cambiadas : %.2f%% %s"
% (pct, "OK" if pct <= 5 else "<-- ALERTA"))
print(" persistencia : %.2f%%" % cmp_.get("pct_persistencia", 0))
if des:
alertas.append("Hay %d llaves que desaparecieron. Esa llave NO es estable: "
"el reporte no es acumulativo puro o faltan columnas de "
"identidad." % des)
if pct > 5:
alertas.append("%.1f%% de las llaves cambio de un dia a otro. En un GL eso es "
"demasiado (el GL reversa, no edita). Revisa normalizacion de "
"formatos o el set de columnas de identidad." % pct)
else:
alertas.append("No hubo segundo archivo: la estabilidad de la llave NO se valido. "
"Corre analizar_llave.py con dos dias consecutivos.")
print("\nLLAVE ESCOGIDA")
print(" %s" % (" + ".join(llave) if llave else "(ninguna)"))
print(" motivo: %s" % motivo)
# ---- identidad vs atributo ----------------------------------------
ident_rep = set(cmp_.get("columnas_IDENTIDAD_sugeridas", [])) if cmp_ else set(llave)
vol = cmp_.get("volatilidad_por_columna_pct", {}) if cmp_ else {}
identidad, atributo, forzados = [], [], []
for col, campo in mapeo.items():
if campo in SIEMPRE_ATRIBUTO:
atributo.append(campo)
if col in ident_rep:
forzados.append((campo, col))
elif col in ident_rep:
identidad.append(campo)
else:
atributo.append(campo)
# las columnas de la llave siempre deben estar en identidad
for col in llave:
campo = mapeo.get(col)
if campo and campo not in identidad and campo not in SIEMPRE_ATRIBUTO:
identidad.append(campo)
if campo in atributo:
atributo.remove(campo)
identidad = sorted(set(identidad))
atributo = sorted(set(atributo) - set(identidad))
print("\nIDENTIDAD (%d campos -> identity_hash)" % len(identidad))
for c in identidad:
print(" %-18s volatilidad %.2f%%"
% (c, vol.get(next((k for k, v in mapeo.items() if v == c), ""), 0.0)))
print("\nATRIBUTO (%d campos -> attr_hash)" % len(atributo))
for c in atributo:
col = next((k for k, v in mapeo.items() if v == c), "")
print(" %-18s volatilidad %.2f%%" % (c, vol.get(col, 0.0)))
if forzados:
print("\n Forzados a ATRIBUTO por regla de seguridad (el reporte los daba como")
print(" identidad, pero con 2 dias de muestra eso no es confiable):")
for campo, col in forzados:
print(" %-18s <- %s" % (campo, col))
# ---- mapeo ---------------------------------------------------------
if dudosos:
print("\nMAPEOS POR APROXIMACION -- VERIFICA ESTOS A MANO")
for c, campo in dudosos:
print(" %-34s -> %s" % (c, campo))
if sin_mapear:
print("\nCOLUMNAS SIN MAPEAR (quedan fuera del hash; anadelas si alguna importa)")
for c in sin_mapear:
print(" %s" % c)
if faltantes:
print("\nCAMPOS SEMANTICOS SIN FUENTE (revisa si existen con otro nombre)")
print(" %s" % ", ".join(faltantes))
cfg = {
"key_version": a.key_version,
"_generado_por": "generar_config.py desde %s" % os.path.basename(a.reporte),
"_llave_analizada": llave,
"_motivo": motivo,
"natural_key": [],
"mapeo": mapeo,
"tipos": {k: v for k, v in TIPOS.items() if k in mapeo.values()},
"identidad": identidad,
"atributo": atributo,
"segmentos_cuenta": {"columna": "account", "separador": "-", "nombres": []},
"signo": {"_nota": "CONFIRMAR con Contabilidad antes de produccion.",
"charge_offs": "debit", "recoveries": "credit",
"formula": "net = SUM(debit) - SUM(credit)"},
}
print("\n" + "=" * 74)
if alertas:
print("ALERTAS")
for x in alertas:
print(" ! %s" % x)
print()
if not identidad:
print("ERROR: no quedo ninguna columna de identidad. Revisa el mapeo a mano.")
sys.exit(1)
if a.revisar:
print("--revisar: no se escribio nada.\n")
print(json.dumps(cfg, indent=2, ensure_ascii=False))
return
os.makedirs(os.path.dirname(os.path.abspath(a.salida)), exist_ok=True)
with open(a.salida, "w", encoding="utf-8") as fh:
json.dump(cfg, fh, indent=2, ensure_ascii=False)
print("Escrito: %s" % a.salida)
print("\nSIGUIENTE PASO")
print(" 1. Abre el archivo y verifica el 'mapeo' columna por columna.")
print(" 2. cd ../app && python -c \"from nco import config; config.cargar(); "
"print('config OK')\"")
print(" 3. python -m nco init")
print(" 4. python -m nco correr data/incoming/<archivo dia 1>.csv\n")
if __name__ == "__main__":
main()
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
generar_data_prueba.py
Genera dos CSV sinteticos que imitan un GL-JE acumulativo de dos dias
consecutivos, para probar analizar_llave.py sin usar data real.
dia2 = dia1 + nuevas + backdated + algunas modificadas
Incluye transacciones duplicadas legitimas a proposito.
python generar_data_prueba.py -> dia1.csv, dia2.csv
python generar_data_prueba.py --sin-llave (elimina la columna unica)
"""
import argparse, csv, random
from datetime import date, timedelta
random.seed(42)
ENTIDADES = ["FB-PR", "FB-FL", "FB-VI"]
REGIONES = ["METRO", "NORTE", "SUR", "OESTE"]
PRODUCTOS = ["AUTO", "MORTG", "COMM", "CCARD", "PERS"]
CC = ["1010", "1020", "2030", "3040", "5050"]
CUENTAS = ["4100-100-2200", "4100-100-2210", "4200-300-2200", "4200-300-5100"]
FUENTES = ["MANUAL", "RECEIVABLES", "LOANS", "PAYABLES"]
CATEG = ["CHARGE OFF", "RECOVERY", "RECLASS", "ADJUSTMENT"]
COLS = ["JournalBatchName","JournalName","LineNumber","Source","Category",
"Period","AccountingDate","CreatedDate","Entity","Region","Product",
"CostCenter","Account","Description","EnteredDebit","EnteredCredit",
"Currency","Status","JeLineId"]
BASE = date(2026, 8, 1)
def mk(i, dia_neg, backdated=False):
fecha = BASE + timedelta(days=random.randint(0, max(0, dia_neg - 8)) if backdated
else random.randint(0, dia_neg))
creado = fecha + timedelta(days=random.randint(0, 2))
es_debito = random.random() < 0.7
monto = round(random.uniform(50, 90000), 2)
lote = "NCO %s %s" % (fecha.strftime("%b").upper(), random.randint(1, 40))
return {
"JournalBatchName": lote,
"JournalName": "%s-JE%04d" % (random.choice(FUENTES)[:4], random.randint(1, 900)),
"LineNumber": random.randint(1, 60),
"Source": random.choice(FUENTES),
"Category": random.choice(CATEG),
"Period": fecha.strftime("%b-%y").upper(),
"AccountingDate": fecha.strftime("%d-%b-%Y").upper(),
"CreatedDate": creado.strftime("%d-%b-%Y").upper(),
"Entity": random.choice(ENTIDADES),
"Region": random.choice(REGIONES),
"Product": random.choice(PRODUCTOS),
"CostCenter": random.choice(CC),
"Account": random.choice(CUENTAS),
"Description": "NET CHARGE OFF BATCH %d" % random.randint(1, 300),
"EnteredDebit": ("%.2f" % monto) if es_debito else "",
"EnteredCredit": "" if es_debito else ("%.2f" % monto),
"Currency": "USD",
"Status": "POSTED",
"JeLineId": str(8000000 + i),
}
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--filas", type=int, default=4000)
ap.add_argument("--sin-llave", action="store_true",
help="Elimina JeLineId para simular el caso sin identificador")
args = ap.parse_args()
cols = [c for c in COLS if not (args.sin_llave and c == "JeLineId")]
dia1 = [mk(i, 15) for i in range(args.filas)]
# duplicados legitimos: misma transaccion exacta dos veces
for _ in range(12):
d = dict(random.choice(dia1))
d["JeLineId"] = str(9500000 + random.randint(1, 9999))
dia1.append(d)
dia2 = [dict(r) for r in dia1] # acumulativo
dia2 += [mk(100000 + i, 16) for i in range(320)] # nuevas
dia2 += [mk(200000 + i, 16, backdated=True) for i in range(18)] # backdated
for r in random.sample(dia2[:args.filas], 25): # modificadas
r["CostCenter"] = random.choice(CC)
if r["EnteredDebit"]:
r["EnteredDebit"] = "%.2f" % (float(r["EnteredDebit"]) + 10)
random.shuffle(dia2) # Oracle reordena
for nombre, filas in (("dia1.csv", dia1), ("dia2.csv", dia2)):
with open(nombre, "w", encoding="utf-8", newline="") as fh:
w = csv.DictWriter(fh, fieldnames=cols, extrasaction="ignore")
w.writeheader()
w.writerows(filas)
print("%s: %d filas" % (nombre, len(filas)))
if __name__ == "__main__":
main()