Python script to convert BCA e-Statement PDFs to CSV and extract all the data

bca_pdf_to_csv.py

#!/usr/bin/env python3
"""Convert a BCA (Bank Central Asia) e-Statement PDF to CSV.

Pure standard library (zlib + re + csv). BCA e-statements embed the whole
transaction table in a single FlateDecode content stream, drawn as positioned
text (Tm / Tj operators) with a monospace font at fixed column x-positions.
This script decompresses that stream, rebuilds rows by their y-coordinate and
columns by their x-coordinate, and writes a CSV in the exact layout the sibling
`bca.yaml` unifier config consumes:

    Account No.,=,'<account>
    Name,=,<name>
    Currency,=,<currency>
    <blank>
    Date,Description,Branch,Amount,,Balance
    '<dd/mm>,<description>,'<branch>,<amount>,<DB|CR|>,<balance>
    ...
    <4-row summary footer>

Values that Excel would coerce (date, branch, account) are apostrophe-prefixed,
and thousands separators are stripped from Amount/Balance, matching the sibling
CSVs. Debits carry the PDF's "DB" marker; credits (an amount with no marker in
the PDF) are written as "CR" so `bca.yaml`'s sign_from (positive_when ["CR"])
resolves them. Set INFER_CR_FOR_CREDITS = False to transcribe the blank marker
verbatim instead.

Statements from ~2025 onward are AES-encrypted (empty user password) and draw
their text through a scaling transform the stdlib parser cannot resolve. For
those, extraction falls back to `pdfplumber` (open with password=""), which
decrypts and applies the transform, yielding the same device-space layout the
older statements use. `pdfplumber` is imported lazily, so converting only older
(unencrypted) statements needs no third-party package.

Before writing the CSV the parsed data is validated: the opening SALDO plus the
signed (DB/CR) transaction amounts must equal the closing SALDO parsed from the
PDF, otherwise a ValueError is raised describing the mismatch.

Usage:
    python3 bca_pdf_to_csv.py "e-Statement 2018-10.pdf" [output.csv]

or with docker:
    docker run --rm -v "$PWD":/data -w /data python:3.12-slim \
        sh -c 'pip install --quiet pdfplumber && \
        for f in *.pdf; do python3 bca_pdf_to_csv.py "$f"; done'

With no output path the CSV is written next to the PDF, named
"<pdf stem>-converted.csv".
"""
import csv
import re
import sys
import zlib
from decimal import Decimal, InvalidOperation
from pathlib import Path

# Column x-dividers observed in the BCA layout (PDF points).
COL_DATE_MAX = 86.71   # x < this  -> TANGGAL (date)
COL_DESC_MAX = 299.92  # ..this    -> KETERANGAN (description, 2 sub-columns)
COL_CBG_MAX = 337.23   # ..this    -> CBG (branch)
COL_AMT_MAX = 465.15   # ..this    -> MUTASI (amount + DB/CR marker); else SALDO
Y_TOLERANCE = 2.5      # fragments within this many points share a row
INFER_CR_FOR_CREDITS = True

# WinAnsiEncoding backslash escapes that appear in PDF literal strings.
_ESCAPES = {ord("n"): 10, ord("r"): 13, ord("t"): 9, ord("b"): 8, ord("f"): 12,
            ord("("): 40, ord(")"): 41, ord("\\"): 92}


def flate_streams(pdf_bytes):
    """Yield every FlateDecode-decompressible stream body in the PDF."""
    for match in re.finditer(rb"stream\r?\n(.*?)\r?\nendstream", pdf_bytes, re.DOTALL):
        try:
            yield zlib.decompress(match.group(1))
        except zlib.error:
            continue


def content_stream(pdf_bytes):
    """Return the largest text-bearing content stream (the page body)."""
    candidates = [s for s in flate_streams(pdf_bytes) if b"Tj" in s or b"TJ" in s]
    if not candidates:
        raise ValueError("no text content stream found; PDF may be scanned/encrypted")
    return max(candidates, key=len)


def stdlib_pages(raw):
    """One page of grouped rows from the plain-PDF stdlib extractor.

    The regex/zlib extractor reads a single content stream, so this covers only
    single-page statements. Multi-page or transformed PDFs fail to reconcile and
    are handled by the pdfplumber fallback in `convert`.
    """
    return [group_rows(extract_fragments(content_stream(raw)))]


def pdfplumber_pages(pdf_path):
    """Grouped rows per page via pdfplumber, which decrypts (empty password) and
    applies the drawing transform, yielding the same device-space layout."""
    try:
        import pdfplumber
    except ImportError as exc:
        raise ValueError(
            "this statement is encrypted / uses a transformed layout and needs "
            "pdfplumber; install it (pip install pdfplumber) and retry"
        ) from exc

    pages = []
    with pdfplumber.open(str(pdf_path), password="") as pdf:
        for page in pdf.pages:
            height = page.height
            fragments = [(height - word["top"], word["x0"], word["text"]) for word in page.extract_words()]
            pages.append(group_rows(fragments))
    return pages


def _read_literal_string(buf, start):
    """Parse a PDF ( ... ) literal string starting at buf[start] == '('."""
    out = bytearray()
    depth = 1
    idx = start + 1
    length = len(buf)
    while idx < length and depth > 0:
        char = buf[idx]
        if char == 0x5C:  # backslash escape
            nxt = buf[idx + 1] if idx + 1 < length else 0x5C
            if 0x30 <= nxt <= 0x37:  # up to 3 octal digits
                octal = chr(nxt)
                idx += 2
                while idx < length and len(octal) < 3 and 0x30 <= buf[idx] <= 0x37:
                    octal += chr(buf[idx])
                    idx += 1
                out.append(int(octal, 8) & 0xFF)
                continue
            out.append(_ESCAPES.get(nxt, nxt))
            idx += 2
            continue
        if char == 0x28:  # (
            depth += 1
            out.append(char)
        elif char == 0x29:  # )
            depth -= 1
            if depth > 0:
                out.append(char)
        else:
            out.append(char)
        idx += 1
    return bytes(out).decode("cp1252", "replace"), idx


def tokenize(buf):
    """Yield ('str', text) | ('num', float) | ('op', name) | ('arr_end', None)."""
    idx = 0
    length = len(buf)
    while idx < length:
        char = buf[idx]
        if char in b" \t\r\n":
            idx += 1
        elif char == 0x28:  # (
            text, idx = _read_literal_string(buf, idx)
            yield ("str", text)
        elif char == 0x5D:  # ]  (end of TJ array)
            yield ("arr_end", None)
            idx += 1
        elif char in b"[<>{}/%":  # delimiters/comments we don't need
            idx += 1
        else:
            end = idx
            while end < length and buf[end] not in b" \t\r\n()<>[]{}/%":
                end += 1
            token = buf[idx:end].decode("latin-1")
            idx = end
            if re.fullmatch(r"-?\d*\.?\d+", token):
                yield ("num", float(token))
            elif token:
                yield ("op", token)


def extract_fragments(content):
    """Return [(y, x, text)] for every non-empty drawn text run."""
    fragments = []
    operands = []
    strings = []
    text_x = text_y = 0.0
    for kind, value in tokenize(content):
        if kind == "num":
            operands.append(value)
        elif kind == "str":
            operands.append(value)
            strings.append(value)
        elif kind == "arr_end":
            pass
        elif kind == "op":
            if value == "Tm" and len(operands) >= 6:
                text_x, text_y = operands[-2], operands[-1]
            elif value in ("Td", "TD") and len(operands) >= 2:
                text_x += operands[-2]
                text_y += operands[-1]
            elif value == "Tj" and strings:
                if strings[-1].strip():
                    fragments.append((text_y, text_x, strings[-1]))
            elif value == "TJ" and strings:
                joined = "".join(strings)
                if joined.strip():
                    fragments.append((text_y, text_x, joined))
            operands = []
            strings = []
    return fragments


def group_rows(fragments):
    """Group fragments into rows by y, returned top-to-bottom, x-sorted."""
    rows = []
    for frag in sorted(fragments, key=lambda item: -item[0]):
        placed = False
        for row in rows:
            if abs(row["y"] - frag[0]) <= Y_TOLERANCE:
                row["frags"].append(frag)
                placed = True
                break
        if not placed:
            rows.append({"y": frag[0], "frags": [frag]})
    for row in rows:
        row["frags"].sort(key=lambda item: item[1])
    return rows


def _value_on_label_row(rows, label, predicate):
    """Find the row whose joined text contains `label`, return the first fragment
    right of the ':' separator that matches predicate.

    Joining the row handles word-level fragments (pdfplumber splits "NO. REKENING"
    into two words); the colon gate stops a label word (e.g. "MATA") from being
    mistaken for the value.
    """
    for row in rows:
        frags = sorted(row["frags"], key=lambda frag: frag[1])
        joined = " ".join(text.strip() for _, _, text in frags)
        if label not in joined:
            continue
        colon_x = next((the_x for _, the_x, text in frags if text.strip() == ":"), None)
        for _, the_x, text in frags:
            value = text.strip()
            if colon_x is not None and the_x <= colon_x:
                continue
            if predicate(value):
                return value
    return ""


def _scan_name(rows):
    """Holder name = topmost left-column line that is not the title, branch, or
    the CATATAN note. Works for both single-fragment (stdlib) and word-level
    (pdfplumber) layouts."""
    for row in sorted(rows, key=lambda item: -item["y"]):
        left = sorted(((the_x, text.strip()) for _, the_x, text in row["frags"]
                       if 20.0 <= the_x < 160.0 and text.strip()), key=lambda pair: pair[0])
        if not left:
            continue
        text = " ".join(word for _, word in left)
        upper = text.upper()
        if upper.startswith(("KCP", "KANTOR", "REKENING", "CATATAN")):
            continue
        if not any(ch.isalpha() for ch in text):
            continue
        return text
    return ""


def extract_metadata(rows):
    """Pull account number, holder name, and currency from the header block."""
    account = _value_on_label_row(rows, "NO. REKENING", lambda t: t.isdigit() and len(t) >= 6)
    currency = _value_on_label_row(rows, "MATA UANG", lambda t: t.isalpha())
    name = _scan_name(rows)
    return account, name, currency


def _num(text):
    """Strip thousands separators from a numeric field."""
    return text.strip().replace(",", "")


def parse_table(pages):
    """Split all pages' rows into (transactions, footer).

    Each page carries its own repeated column header and metadata block; the
    four-row summary footer appears once, after the last page's transactions.
    Rows are processed per page (below that page's TANGGAL header), so a later
    page's metadata `:` rows — which sit above its header — never trip the
    footer detector.
    """
    transactions = []
    footer = []
    header_seen = False
    for rows in pages:
        header_y = None
        for row in rows:
            if any("TANGGAL" in text for _, _, text in row["frags"]):
                header_y = row["y"]
                break
        if header_y is None:
            continue
        header_seen = True

        in_footer = False
        for row in (r for r in rows if r["y"] < header_y - Y_TOLERANCE):
            frags = row["frags"]
            texts = [text.strip() for _, _, text in frags]
            if "bersambung" in " ".join(texts).lower():
                continue  # page-break note ("Bersambung ke/dari halaman ...")
            if "TANGGAL" in texts and any("KETERANGAN" in text for text in texts):
                continue  # repeated column header
            if in_footer or any(text == ":" for text in texts):
                in_footer = True
                colon_x = next((x for _, x, t in frags if t.strip() == ":"), 1e9)
                label = " ".join(
                    t.strip() for _, x, t in sorted(frags, key=lambda frag: frag[1])
                    if x < colon_x and t.strip() not in ("", ":")
                )
                value = next((_num(t) for _, x, t in frags if x > colon_x and any(c.isdigit() for c in t)), "")
                if label:
                    footer.append((label, value))
                continue

            date = cbg = amount = drcr = balance = None
            desc_parts = []
            for _, the_x, text in frags:
                text = text.strip()
                if not text:
                    continue
                if the_x < COL_DATE_MAX:
                    date = text
                elif the_x < COL_DESC_MAX:
                    desc_parts.append(text)
                elif the_x < COL_CBG_MAX:
                    cbg = text
                elif the_x < COL_AMT_MAX:
                    if text in ("DB", "CR"):
                        drcr = text
                    else:
                        amount = _num(text)
                else:
                    balance = _num(text)
            desc = " ".join(desc_parts)

            if date:
                if INFER_CR_FOR_CREDITS and drcr is None and amount:
                    drcr = "CR"
                transactions.append({
                    "date": date, "desc": desc, "cbg": cbg or "",
                    "amount": amount or "", "drcr": drcr or "", "balance": balance or "",
                })
            elif desc and transactions:  # continuation line: fold into prior description
                transactions[-1]["desc"] = (transactions[-1]["desc"] + " " + desc).strip()

    if not header_seen:
        raise ValueError("column header row (TANGGAL) not found")
    return transactions, footer


def _dec(text):
    """Parse a comma-stripped numeric string to Decimal, or None if not numeric."""
    try:
        return Decimal(_num(text))
    except (InvalidOperation, ValueError):
        return None


def validate_balances(transactions, footer):
    """Verify opening SALDO +/- signed transactions equals the closing SALDO.

    Raises ValueError describing any mismatch or missing anchor. Returns
    (opening, computed, closing) on success.
    """
    footer_map = {label: value for label, value in footer}

    # Anchor to the summary footer only. Deriving closing from the last parsed
    # transaction would let a truncated read (e.g. only page 1 of a multi-page
    # statement, whose rows are internally consistent) reconcile against itself
    # and hide the missing pages. Requiring the footer forces such a read to
    # fail here so `convert` falls back to the page-aware pdfplumber path.
    opening = _dec(footer_map.get("SALDO AWAL", ""))
    closing = _dec(footer_map.get("SALDO AKHIR", ""))

    if opening is None or closing is None:
        raise ValueError(
            "balance validation: summary footer (SALDO AWAL / SALDO AKHIR) not found; "
            "the statement may be truncated or a multi-page PDF was only partially read"
        )

    total = opening
    for txn in transactions:
        amount = _dec(txn["amount"])
        if amount is None:
            continue  # e.g. the SALDO AWAL row carries no amount
        if txn["drcr"] == "DB":
            total -= amount
        elif txn["drcr"] == "CR":
            total += amount
        else:
            raise ValueError(
                f"balance validation: transaction {txn['date']} '{txn['desc']}' has amount "
                f"{txn['amount']} but no DB/CR direction to sign it"
            )

    if abs(total - closing) > Decimal("0.005"):
        raise ValueError(
            "balance validation FAILED: opening {} +/- transactions = {}, but the PDF's "
            "closing SALDO is {} (difference {}). The parsed transactions do not reconcile."
            .format(opening, total, closing, total - closing)
        )
    return opening, total, closing


def write_csv(path, account, name, currency, transactions, footer):
    with open(path, "w", newline="", encoding="utf-8") as handle:
        writer = csv.writer(handle)
        writer.writerow(["Account No.", "=", "'" + account if account else ""])
        writer.writerow(["Name", "=", name])
        writer.writerow(["Currency", "=", currency])
        writer.writerow([])
        writer.writerow(["Date", "Description", "Branch", "Amount", "", "Balance"])
        for txn in transactions:
            writer.writerow([
                "'" + txn["date"],
                txn["desc"],
                "'" + txn["cbg"] if txn["cbg"] else "",
                txn["amount"],
                txn["drcr"],
                txn["balance"],
            ])
        for label, value in footer:
            writer.writerow([label, "", "", "", "", value])


def convert(pdf_path, csv_path=None):
    """Parse the statement and write the CSV once its balances reconcile.

    The plain-PDF stdlib extractor is tried first (fast, no dependency, keeps
    older single-page statements byte-identical). If its output does not
    validate — encrypted, multi-page, or a transformed layout — parsing falls
    back to pdfplumber. A genuine reconciliation failure surfaces as ValueError.
    """
    pdf_path = Path(pdf_path)
    csv_path = Path(csv_path) if csv_path else pdf_path.with_name(pdf_path.stem + "-converted.csv")
    raw = pdf_path.read_bytes()

    extractors = []
    if b"/Encrypt" not in raw:
        extractors.append(lambda: stdlib_pages(raw))
    extractors.append(lambda: pdfplumber_pages(pdf_path))

    last_error = None
    for get_pages in extractors:
        try:
            pages = get_pages()
            account, name, currency = extract_metadata(pages[0])
            transactions, footer = parse_table(pages)
            validate_balances(transactions, footer)
        except ValueError as error:
            last_error = error
            continue
        write_csv(csv_path, account, name, currency, transactions, footer)
        return csv_path, len(transactions)
    raise last_error


def main(argv):
    if len(argv) < 2:
        print(__doc__)
        return 1
    out_path, count = convert(argv[1], argv[2] if len(argv) > 2 else None)
    print(f"Wrote {count} transactions to {out_path}")
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv))