C43 to CSV converter

c43_to_csv.py

#!/usr/bin/env python3
"""
Converts a Spanish bank Norma 43 / C43 (Cuaderno 43) file to CSV.

Usage:
    python3 c43_to_csv.py <input.txt> [-o output.csv] [--encoding latin-1] [--sample N]

Changelog:
  v1 - initial: 10 columns, basic parsing
  v2 - merged detalle2/3/4 via raw 76-char block concatenation; concepto_adicional split
  v3 - v2 + collapse runs of >=6 consecutive spaces in detalle (fixes dates artifact)
  v4 - v3 + argparse CLI · Decimal amounts · dataclass · validation with line numbers ·
       oficina_origen / subconcepto / documento / referencia_1 fields
  v5 - v4 + concepto_adicional whitelist: only SEPA scheme codes (CORE, COR1, …) are
       split from the description; all other 2301 text goes into descripcion as-is
  v6 - v5 + cleanup pass after human review:
       · drop documento / referencia_1 (99%+ rows = 0000000000, pure noise)
       · drop signo (redundant with importe sign)
       · drop concepto_propio_desc (2-digit [24:26] does NOT map to CaixaBank CLOP codes)
       · strip leading "00" type-indicator from ref_inline
"""

from __future__ import annotations

import argparse
import csv
import re
import sys
from dataclasses import asdict, dataclass, field
from datetime import datetime
from decimal import Decimal, InvalidOperation
from pathlib import Path
from typing import List, Optional

# ---------------------------------------------------------------------------
# Concepto adicional whitelist (2301 record, positions [4:8])
# Only standardised SEPA payment scheme identifiers qualify.
# All other values at that position are the start of a plain description.
# ---------------------------------------------------------------------------

VALID_CONCEPTO_ADICIONAL = {
    # SEPA Core Direct Debit scheme (COR1 retired but still circulates)
    "CORE", "COR1", "COR2", "COR3",

    # SEPA B2B Direct Debit scheme
    "B2B",
    "B2B0", "B2B1", "B2B2",

    # ISO 20022 sequence types (SeqTp from pain.008 — copied verbatim by most banks)
    "FRST",   # First collection
    "RCUR",   # Recurring collection
    "OOFF",   # One-off collection
    "FNAL",   # Final collection

    # Return / refund management
    "RFDD",   # Recurring First Direct Debit
    "RETN",   # Returned
    "RFDN",   # Refund

    # Generic SEPA infrastructure markers (non-standard but observed in production)
    "SEPA",
    "SDD",
}

# ---------------------------------------------------------------------------
# Code tables
# ---------------------------------------------------------------------------

CONCEPTOS_COMUNES = {
    "01": "TALONES / REINTEGROS",
    "02": "ABONARES / ENTREGAS / INGRESOS",
    "03": "DOMICILIADOS / RECIBOS / LETRAS / PAGOS",
    "04": "GIROS / TRANSFERENCIAS / TRASPASOS / CHEQUES",
    "05": "AMORTIZACIONES PRESTAMOS / CREDITOS",
    "06": "REMESAS EFECTOS",
    "07": "SUSCRIPCIONES / DIV. PASIVOS / CANJES",
    "08": "DIV. CUPONES / PRIMA JUNTA / AMORTIZACIONES",
    "09": "OPERACIONES DE BOLSA / COMPRA-VENTA VALORES",
    "10": "CHEQUES GASOLINA",
    "11": "CAJERO AUTOMATICO",
    "12": "TARJETAS DE CREDITO / DEBITO",
    "13": "OPERACIONES EXTRANJERO",
    "14": "DEVOLUCIONES E IMPAGADOS",
    "15": "NOMINAS / SEGUROS SOCIALES",
    "16": "TIMBRES / CORRETAJE / POLIZA",
    "17": "INTERESES / COMISIONES / CUSTODIA / GASTOS E IMPUESTOS",
    "98": "ANULACIONES / CORRECCIONES ASIENTO",
    "99": "VARIOS",
}

CODIGOS_PROPIOS = {
    "00": "IMPOSICION EN EFECTIVO",
    "01": "COMPANIA DE AGUA",
    "02": "TRANSFERENCIA",
    "05": "COMPANIA DE ELECTRICIDAD",
    "06": "COMPANIA DE SERVICIOS TELEFONICOS",
    "07": "COMPANIA DE GAS",
    "12": "DEVOLUCION CHEQUE",
    "16": "IMPUESTOS/TRIBUTOS/EMBARGOS/RETENCION FISCAL",
    "17": "INTERESES",
    "18": "PRESTAMO/LEASING/FINANCIACION/CUENTA DE CREDITO",
    "19": "SEGURIDAD SOCIAL/AUTONOMOS",
    "20": "CORREO",
    "25": "VALORES",
    "29": "FINCAS / ALQUILERES",
    "30": "FACTORING",
    "32": "OPERACIONES EN DIVISAS",
    "33": "CUOTA DE ASOCIADO",
    "34": "ENSENANZA",
    "35": "ENTIDAD PREVISION / ENTIDAD FINANCIACION",
    "36": "PRECIO SERVICIOS",
    "38": "CARGO DE RECIBOS/PAGARES/PAGOS DOMICILIADOS",
    "39": "PUBLICACIONES",
    "40": "COMERCIOS/CAJEROS/TARJETAS",
    "41": "NOMINA",
    "44": "TARJETA DE CREDITO",
    "46": "COMPRA EN COMERCIO",
    "48": "ABONOS DIVERSOS",
    "49": "CUOTA DE ENTIDAD",
    "51": "OPERATIVA DE SEGUROS",
    "54": "FONDO DE INVERSION MOBILIARIA",
    "55": "ADEUDOS DOMICILIADOS / GESTION COBRO",
    "56": "AVALES",
    "61": "AHORRO A PLAZO",
    "64": "CHEQUE",
    "65": "LIQUIDACION DE FACTURAS",
    "67": "TRASPASO / TRANSFERENCIA URGENTE / B. DIGITAL",
    "68": "EMBARGO",
    "72": "PAGOS A TERCEROS / SERVICAIXA",
    "73": "SERVICIO DE PAGOS",
    "74": "SERVICIO CENTRALIZACION DE FONDOS",
    "76": "ANTICIPO ADEUDOS",
    "79": "CONFIRMING",
    "81": "INGRESO CHEQUE",
    "83": "RECOGIDA/ENTREGA DE FONDOS A DOMICILIO",
    "85": "CANCELACION",
}

# ---------------------------------------------------------------------------
# Errors
# ---------------------------------------------------------------------------

class ParseError(ValueError):
    """Raised when a line cannot be parsed safely."""


# ---------------------------------------------------------------------------
# Data model
# ---------------------------------------------------------------------------

@dataclass
class Movement:
    fecha_operacion:     str
    fecha_valor:         str
    importe:             Decimal
    oficina_origen:      str
    concepto_comun:      str
    concepto_comun_desc: str
    concepto_propio:     str        # raw 2-digit AEB sub-concept code from record 22 [24:26]
    subconcepto:         str
    ref_inline:          str
    concepto_adicional:  str = ""   # 4-char code from 2301[4:8]
    descripcion:         str = ""   # free text from 2301[8:]
    detalle:             str = ""   # merged 2302+2303+2304
    referencia:          str = ""   # from 2305

    def to_csv_row(self) -> dict:
        d = asdict(self)
        d["importe"] = f"{self.importe:.2f}"
        return d


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def parse_date(value: str, line_no: int, field: str) -> str:
    if len(value) != 6 or not value.isdigit():
        raise ParseError(f"Line {line_no}: {field} invalid {value!r} — expected 6-digit YYMMDD")
    yy = int(value[:2])
    year = 2000 + yy if yy < 60 else 1900 + yy
    try:
        dt = datetime(year, int(value[2:4]), int(value[4:6]))
    except ValueError as exc:
        raise ParseError(f"Line {line_no}: {field} invalid {value!r}: {exc}") from exc
    return dt.strftime("%Y-%m-%d")


def parse_amount(raw: str, sign: str, line_no: int) -> Decimal:
    if len(raw) != 14 or not raw.isdigit():
        raise ParseError(f"Line {line_no}: amount invalid {raw!r} — expected 14 digits")
    if sign not in {"1", "2"}:
        raise ParseError(f"Line {line_no}: sign invalid {sign!r} — expected '1' (debit) or '2' (credit)")
    amount = Decimal(raw) / Decimal("100")
    return -amount if sign == "1" else amount


def normalize_line(raw: str, line_no: int) -> str:
    line = raw.rstrip("\r\n")
    if len(line) > 80:
        raise ParseError(f"Line {line_no}: length {len(line)} > 80 — not a valid Norma 43 record")
    return line.ljust(80)


def join_raw_blocks(blocks: list[str]) -> str:
    """
    Concatenate fixed-width 76-char blocks by direct character concatenation
    (no separator), then apply two cleanup passes:

    1. Remove runs of EXACTLY 6 spaces surrounded by non-space characters.
       These are fixed-width field-padding artifacts where a value (e.g. a year
       digit) was split across two 76-char lines: '201      9' → '2019'.
       Longer runs (e.g. 19-space IBAN/reference separators) are untouched because
       the lookahead lands on another space, not a non-space character.

    2. Strip leading/trailing whitespace from the final result.
    """
    combined = "".join(blocks)
    # (?<=[^\s]) {6}(?=[^\s])  — exactly 6 spaces between two non-space chars
    combined = re.sub(r"(?<=[^\s]) {6}(?=[^\s])", "", combined)
    return combined.strip()


# ---------------------------------------------------------------------------
# Parser
# ---------------------------------------------------------------------------

def parse_c43(path: Path, encoding: str = "latin-1") -> List[Movement]:
    movements: List[Movement] = []
    current: Optional[Movement] = None
    raw_blocks: List[str] = []          # accumulates 76-char detalle lines

    def finalize(m: Movement, blocks: list) -> Movement:
        m.detalle = join_raw_blocks(blocks)
        return m

    with path.open("r", encoding=encoding) as fh:
        for line_no, raw in enumerate(fh, start=1):
            line = normalize_line(raw, line_no)
            rec = line[0:2]

            if rec == "22":
                if current is not None:
                    movements.append(finalize(current, raw_blocks))
                raw_blocks = []

                raw_ref = line[62:80].strip()
                current = Movement(
                    fecha_operacion      = parse_date(line[10:16], line_no, "fecha_operacion"),
                    fecha_valor          = parse_date(line[16:22], line_no, "fecha_valor"),
                    importe              = parse_amount(line[28:42], line[27], line_no),
                    oficina_origen       = line[6:10].strip(),
                    concepto_comun       = line[22:24],
                    concepto_comun_desc  = CONCEPTOS_COMUNES.get(line[22:24], ""),
                    concepto_propio      = line[24:26],
                    subconcepto          = line[26].strip(),
                    ref_inline           = raw_ref[2:] if raw_ref.startswith("00") else raw_ref,
                )

            elif rec == "23" and current is not None:
                sub    = line[2:4]
                raw76  = line[4:80]         # raw 76 chars — do NOT strip yet
                content = raw76.strip()

                if sub == "01":
                    code = line[4:8]
                    if code in VALID_CONCEPTO_ADICIONAL:
                        # Genuine SEPA scheme code: split code from description
                        current.concepto_adicional = code
                        current.descripcion        = line[8:80].strip()
                    else:
                        # Plain description starting at position 4 (no code)
                        current.concepto_adicional = ""
                        current.descripcion        = line[4:80].strip()
                elif sub == "02":
                    raw_blocks.append(raw76)
                elif sub == "03":
                    raw_blocks.append(raw76)
                elif sub == "04":
                    raw_blocks.append(raw76)
                elif sub == "05":
                    current.referencia = content
                # unknown sub-types (bank-specific extensions) are silently ignored

            elif rec in {"11", "33", "88", "99"}:
                pass    # structural records — not extracted to CSV

            elif rec.strip() == "":
                pass    # blank padding line

            # unknown record types are silently skipped (tolerant parsing)

    if current is not None:
        movements.append(finalize(current, raw_blocks))

    return movements


# ---------------------------------------------------------------------------
# Output
# ---------------------------------------------------------------------------

FIELDNAMES = [
    "fecha_operacion", "fecha_valor", "importe",
    "oficina_origen",
    "concepto_comun", "concepto_comun_desc",
    "concepto_propio",
    "subconcepto",
    "concepto_adicional", "descripcion",
    "detalle",
    "referencia", "ref_inline",
]


def write_csv(path: Path, movements: List[Movement]) -> None:
    with path.open("w", newline="", encoding="utf-8-sig") as fh:
        writer = csv.DictWriter(fh, fieldnames=FIELDNAMES)
        writer.writeheader()
        writer.writerows(m.to_csv_row() for m in movements)


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------

def build_arg_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(
        description="Convierte un fichero Norma 43 / Cuaderno 43 (AEB) a CSV"
    )
    p.add_argument("input", type=Path, help="Fichero de entrada (.txt / .n43 / .c43)")
    p.add_argument("-o", "--output", type=Path,
                   help="CSV de salida (defecto: <input>.csv)")
    p.add_argument("--encoding", default="latin-1",
                   help="Codificación del fichero de entrada (defecto: latin-1)")
    p.add_argument("--sample", type=int, default=8,
                   help="Filas de muestra a imprimir (defecto: 8)")
    return p


def main() -> int:
    args = build_arg_parser().parse_args()

    input_path: Path = args.input
    if not input_path.exists():
        print(f"ERROR: no existe el fichero {input_path}", file=sys.stderr)
        return 1

    output_path: Path = args.output or input_path.with_suffix(".csv")

    try:
        movements = parse_c43(input_path, encoding=args.encoding)
    except ParseError as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 1

    write_csv(output_path, movements)
    print(f"✓ {len(movements)} movimientos → {output_path}")

    if args.sample > 0:
        print(f"\nMuestra ({args.sample} filas):")
        for m in movements[:args.sample]:
            concept = m.concepto_comun_desc or m.concepto_comun
            print(f"  {m.fecha_operacion}  {m.importe:>12}  "
                  f"{concept[:30]:<30}  {m.descripcion[:50]}")

    return 0


if __name__ == "__main__":
    raise SystemExit(main())