finance-app/scripts/split_csv_match.py

#!/usr/bin/env python3
"""Match SplitMyExpenses CSV rows to transactions already in the ledger.

Dry-run by default. It prints what it would do and writes nothing; `--write`
is a separate step (task #9) and is deliberately not implemented here.

Why this exists
---------------
The CSVs are the record of how expenses were actually shared before this app
existed. Importing them is what makes historical *spend* correct: without a
split row, a $200 grocery shop counts as $200 of my spending when half of it
was never mine. The balances are already settled by carryover transaction 2348,
so these splits are imported as `settled = true` and move no balance.

Three things make the matching harder than "same date, same amount":

1.  **Dates are ambiguous across files.** The household file writes D/M/YYYY;
    the four trip files write ISO. 474 rows parse validly under both readings,
    so the format is decided per file, from the file, and never guessed per row.

2.  **The sign convention is not "who paid".** A person's column is their net
    balance impact: positive means they are owed. So the payer is whoever is
    positive, and the other person's share is |their negative| / cost. A row
    reading +cost / -cost therefore means the other party owes 100% -- NOT that
    the expense was unshared, which is the reading that would fake an
    arrangement change.

3.  **A settlement is not an expense.** Rows where one person hands the other
    money must not become split transactions; they are already represented by
    the carryover.

Usage:
    .venv/bin/python scripts/split_csv_match.py [--verbose] [--file NAME]
"""

from __future__ import annotations

import argparse
import csv
import glob
import os
import re
import sys
from collections import Counter
from dataclasses import dataclass, field
from datetime import date, datetime, timedelta

import psycopg2
import psycopg2.extras

DUMP_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "dump")

# Read from the same place the app does rather than hardcoding a container IP,
# which changes on every recreate.
def db_url() -> str:
    url = os.environ.get("DATABASE_URL")
    if url:
        return url
    for envfile in (".env", ".env.test"):
        path = os.path.join(os.path.dirname(DUMP_DIR), envfile)
        if not os.path.exists(path):
            continue
        for line in open(path):
            if line.startswith("DATABASE_URL"):
                return line.split("=", 1)[1].strip().strip('"').strip("'")
    sys.exit("No DATABASE_URL found (env, .env, or .env.test)")


# The two people in these files. The CSV writes full names; the ledger uses
# first names.
CSV_ME = "Siddharth Bose"
CSV_THEM = "Meghalee"
PARTICIPANT_ME = 1
PARTICIPANT_THEM = 4

# A settlement transfers money; it is not a shared cost. These are the
# descriptions SplitMyExpenses uses for them.
SETTLEMENT_PAT = re.compile(
    r"payment|settle|debts? remainder|reimburse|transfer to|paid back", re.I
)


@dataclass
class CsvRow:
    source: str
    line: int
    when: date
    description: str
    category: str
    cost: float
    currency: str
    net_me: float
    net_them: float

    # Filled in by classify()
    mode: str = ""
    payer: int = 0            # participant id who paid
    ower: int = 0             # participant id who owes
    ower_share: float = 0.0   # 0-100

    def __str__(self) -> str:
        return f"{self.source}:{self.line} {self.when} {self.description[:38]!r} ${self.cost:.2f} [{self.mode}]"


@dataclass
class MatchReport:
    rows: list = field(default_factory=list)
    matched: list = field(default_factory=list)
    ambiguous: list = field(default_factory=list)
    unmatched: list = field(default_factory=list)
    skipped: list = field(default_factory=list)


def sniff_date_format(sample: list[str]) -> str:
    """Decide ISO vs D/M/YYYY for a whole file.

    Deciding per row is what produces a ledger where January and February are
    silently swapped for some rows and not others. A file is written by one
    exporter in one format, so the file is the unit of decision.
    """
    if not sample:
        return "%Y-%m-%d"
    slashes = sum(1 for s in sample if "/" in s)
    return "%d/%m/%Y" if slashes > len(sample) / 2 else "%Y-%m-%d"


def parse_rows(path: str) -> list[CsvRow]:
    with open(path, newline="", encoding="utf-8-sig") as fh:
        reader = list(csv.DictReader(fh))
    if not reader:
        return []

    # Header names vary in quoting between exports.
    def col(row: dict, *names: str):
        for n in names:
            for k in row:
                if k.strip().strip('"') == n:
                    return row[k]
        return None

    fmt = sniff_date_format([col(r, "Date") or "" for r in reader[:40]])
    out: list[CsvRow] = []
    for i, r in enumerate(reader, start=2):
        raw_date = (col(r, "Date") or "").strip()
        try:
            when = datetime.strptime(raw_date, fmt).date()
        except ValueError:
            continue
        try:
            cost = float(col(r, "Cost") or 0)
            net_me = float(col(r, CSV_ME) or 0)
            net_them = float(col(r, CSV_THEM) or 0)
        except ValueError:
            continue
        out.append(
            CsvRow(
                source=os.path.basename(path),
                line=i,
                when=when,
                description=(col(r, "Description") or "").strip(),
                category=(col(r, "Category") or "").strip(),
                cost=cost,
                currency=(col(r, "Currency") or "AUD").strip(),
                net_me=net_me,
                net_them=net_them,
            )
        )
    return out


def classify(row: CsvRow) -> CsvRow:
    """Work out who paid and what share the other person owes.

    A person's column is their net balance impact, not their share: positive
    means they are owed money. So the payer is whoever is positive.
    """
    if row.cost == 0:
        row.mode = "zero-cost"
        return row

    if SETTLEMENT_PAT.search(row.description):
        row.mode = "settlement"
        return row

    # Both zero against a real cost: recorded but not shared.
    if abs(row.net_me) < 0.005 and abs(row.net_them) < 0.005:
        row.mode = "unshared"
        return row

    if row.net_me > 0:
        row.payer, row.ower, owed = PARTICIPANT_ME, PARTICIPANT_THEM, abs(row.net_them)
    else:
        row.payer, row.ower, owed = PARTICIPANT_THEM, PARTICIPANT_ME, abs(row.net_me)

    row.ower_share = round(owed / row.cost * 100, 2)
    if abs(row.ower_share - 50) < 0.6:
        row.mode = "50/50"
    elif abs(row.ower_share - 100) < 0.6:
        row.mode = "other-owes-all"
    else:
        row.mode = f"uneven-{row.ower_share:.0f}"
    return row


def norm(s: str) -> set[str]:
    return {w for w in re.split(r"[^a-z0-9]+", (s or "").lower()) if len(w) > 2}


def score(row: CsvRow, tx: dict) -> float:
    """How well a ledger row matches a CSV row. Amount and date gate it;
    description only ranks among survivors."""
    days = abs((tx["transaction_date"] - row.when).days)
    s = 100.0 - days * 4
    overlap = norm(row.description) & (norm(tx["description"]) | norm(tx["merchant_normalized"]))
    s += 12 * len(overlap)
    return s


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--verbose", action="store_true")
    ap.add_argument("--file", help="only this CSV (substring match)")
    ap.add_argument("--window", type=int, default=5, help="date tolerance in days")
    ap.add_argument(
        "--write", action="store_true",
        help="actually insert the splits (settled=true). Without this, nothing is written.",
    )
    args = ap.parse_args()

    paths = sorted(glob.glob(os.path.join(DUMP_DIR, "*SplitMyExpenses*.csv")))
    if args.file:
        paths = [p for p in paths if args.file in os.path.basename(p)]
    if not paths:
        sys.exit("No SplitMyExpenses CSVs found in dump/")

    conn = psycopg2.connect(db_url())
    cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
    # Superseded rows are duplicates; matching against them would attach a split
    # to a row nothing else counts.
    cur.execute(
        """
        SELECT t.id, t.transaction_date, COALESCE(t.description,'') AS description,
               COALESCE(t.merchant_normalized,'') AS merchant_normalized,
               COALESCE(t.amount_aud, t.amount)::float AS amount,
               t.superseded_by_id,
               EXISTS(SELECT 1 FROM transaction_splits x WHERE x.transaction_id=t.id) AS has_split
        FROM transactions t
        LEFT JOIN statements s ON s.id = t.statement_id
        WHERE COALESCE(t.owner_id, s.owner_id) IN (%s, %s)
          AND t.superseded_by_id IS NULL
          AND NOT (t.statement_id IS NULL AND t.reconciled_with_id IS NOT NULL)
          AND t.transaction_type IN ('debit','fee','interest')
        """,
        (PARTICIPANT_ME, PARTICIPANT_THEM),
    )
    txs = cur.fetchall()

    # Index by rounded amount: the amount must agree, so it is the only cheap
    # gate that never needs fuzzy comparison.
    by_amount: dict[float, list[dict]] = {}
    for t in txs:
        by_amount.setdefault(round(t["amount"], 2), []).append(t)

    rep = MatchReport()
    modes: Counter = Counter()
    candidates: list = []
    for path in paths:
        for row in parse_rows(path):
            classify(row)
            rep.rows.append(row)
            modes[row.mode] += 1

            if row.mode in ("settlement", "zero-cost", "unshared"):
                rep.skipped.append(row)
                continue
            if row.currency != "AUD":
                rep.skipped.append(row)
                continue

            cands = [
                t for t in by_amount.get(round(row.cost, 2), [])
                if abs((t["transaction_date"] - row.when).days) <= args.window
            ]
            if not cands:
                rep.unmatched.append(row)
            else:
                candidates.append((row, cands))

    # Assign one-to-one, best pair first.
    #
    # Without this a ledger row can be claimed by several CSV rows. That is not
    # hypothetical: the NZ trip has two identical $10.16 Uber trips on one day
    # and three matching ledger rows, and four PayMyPark rows in the same shape.
    # Attaching a split twice is harmless (the unique key absorbs it) but it
    # leaves the second CSV row silently unrepresented while looking matched,
    # which is a lie in the report rather than a defect in the data.
    scored = sorted(
        ((score(row, t), row, t) for row, cands in candidates for t in cands),
        key=lambda x: -x[0],
    )
    taken_tx: set[int] = set()
    taken_row: set[int] = set()
    for s, row, t in scored:
        if id(row) in taken_row or t["id"] in taken_tx:
            continue
        taken_row.add(id(row))
        taken_tx.add(t["id"])
        rep.matched.append((row, t))
    for row, cands in candidates:
        if id(row) not in taken_row:
            rep.ambiguous.append((row, cands))

    total = len(rep.rows)
    considered = total - len(rep.skipped)
    print(f"CSV rows              {total}")
    print(f"  skipped             {len(rep.skipped)}  (settlements, zero-cost, unshared, non-AUD)")
    print(f"  considered          {considered}")
    print(f"    matched           {len(rep.matched)}  ({len(rep.matched)/max(considered,1)*100:.1f}%)")
    print(f"    ambiguous         {len(rep.ambiguous)}")
    print(f"    unmatched         {len(rep.unmatched)}")
    print()
    print("Row modes:")
    for m, n in modes.most_common():
        print(f"  {m:<18} {n}")

    already = sum(1 for _, t in rep.matched if t["has_split"])
    print()
    print(f"Of the matched, {already} already carry a split and would be left alone;")
    print(f"{len(rep.matched) - already} would gain one.")

    if args.write:
        # Imported as settled: these obligations were discharged on a platform
        # we no longer run, and the residual is already carried by transaction
        # 2348. Writing them unsettled would re-open ~$40k of debts that were
        # paid years ago. See ACTIVE_OBLIGATION in analytics-sql.ts -- settled
        # rows stay in spend and leave every owed figure, which is exactly the
        # point: this import exists to correct historical SPEND.
        SETTLED_ON = "2026-01-09"  # the carryover's date
        written = 0
        for row, tx in rep.matched:
            if tx["has_split"]:
                continue
            payer_share = round(100 - row.ower_share, 2)
            pairs = [(row.ower, row.ower_share)]
            if payer_share > 0:
                pairs.append((row.payer, payer_share))
            for pid, share in pairs:
                cur.execute(
                    """
                    INSERT INTO transaction_splits
                      (transaction_id, participant_id, share_percent, settled, settled_at)
                    VALUES (%s, %s, %s, true, %s)
                    ON CONFLICT (transaction_id, participant_id) DO NOTHING
                    """,
                    (tx["id"], pid, share, SETTLED_ON),
                )
                written += cur.rowcount
        conn.commit()
        print(f"\nWROTE {written} split rows (settled=true, settled_at={SETTLED_ON}).")

    if args.verbose:
        print("\n--- ambiguous ---")
        for row, cands in rep.ambiguous[:40]:
            print(f"  {row}")
            for t in cands[:3]:
                print(f"      -> #{t['id']} {t['transaction_date']} {t['description'][:44]!r}")
        print("\n--- unmatched ---")
        for row in rep.unmatched[:60]:
            print(f"  {row}")

    conn.close()
    return 0


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