diff --git a/nlb-export/nlb.py b/nlb-export/nlb.py index c2176ec..ad6d48e 100644 --- a/nlb-export/nlb.py +++ b/nlb-export/nlb.py @@ -2,6 +2,7 @@ import csv import re import subprocess import sys +from decimal import Decimal from pathlib import Path # extract transaction from izpisek pdf TRANSACTION_START = re.compile(r"^\d{2}\.\d{2}\.\d{2}\s") @@ -24,6 +25,14 @@ def pdf_to_text(pdf_path): return result.stdout +def to_number(text): + """Statements write 1.234,56 and mark a negative balance with a trailing minus.""" + text = text.strip() + if text.endswith("-"): + text = "-" + text[:-1] + return Decimal(text.replace(".", "").replace(",", ".")) + + def parse_transaction(line): m = TRANSACTION_LINE.match(line.rstrip()) account = "" @@ -33,15 +42,12 @@ def parse_transaction(line): account = f else: desc_parts.append(f) - balance = m.group("balance") - if balance.endswith("-"): - balance = "-" + balance[:-1] return { "date": m.group("date"), "desc": " ".join(desc_parts), "account": account, - "amount": m.group("amount"), - "balance": balance, + "amount": f"{to_number(m.group('amount')):+.2f}", + "balance": f"{to_number(m.group('balance')):.2f}", "currency": "EUR", } @@ -70,13 +76,25 @@ def extract_transactions(pdf_path): return rows +def check_balances(rows, path): + """Every balance must be the previous one plus the transaction amount.""" + for prev, row in zip(rows, rows[1:]): + expected = Decimal(prev["balance"]) + Decimal(row["amount"]) + if expected != Decimal(row["balance"]): + print(f" warning: {path.name} {row['date']} {row['desc'][:40]!r} " + f"balance {row['balance']} != {expected}") + + def main(): pdf_dir = Path(sys.argv[1]) out_path = pdf_dir.with_suffix(".csv") all_rows = [] for pdf_path in sorted(pdf_dir.glob("*.pdf")): - all_rows.extend(extract_transactions(pdf_path)) + rows = extract_transactions(pdf_path) + # consecutive izpiski of one account, so the chain carries over the file boundary + check_balances(all_rows[-1:] + rows, pdf_path) + all_rows.extend(rows) with open(out_path, "w", newline="") as f: writer = csv.DictWriter(f, fieldnames=["date", "desc", "account", "amount", "balance", "currency"]) diff --git a/nlb-export/revolut.py b/nlb-export/revolut.py new file mode 100644 index 0000000..f03b3a2 --- /dev/null +++ b/nlb-export/revolut.py @@ -0,0 +1,95 @@ +import csv +import re +import sys +from collections import Counter +from datetime import datetime +from decimal import Decimal +from pathlib import Path +# convert revolut account statement csv exports into the same shape as nlb.py / trade_republic.py +STATEMENTS = "account-statement*.csv" +IBAN = re.compile(r"\b[A-Z]{2}\d{2}[A-Z0-9]{11,30}\b") +FIELDS = ["date", "desc", "account", "amount", "balance", "currency", "type"] + + +def decimals(text): + return len(text.partition(".")[2]) + + +def scale(source): + """Revolut writes JPY without decimals and EUR with two, so follow the file.""" + places = {} + for r in source: + seen = [decimals(r[c]) for c in ("Amount", "Fee", "Balance") if r[c]] + places[r["Currency"]] = max(seen + [places.get(r["Currency"], 0)]) + return places + + +def extract_transactions(path): + print(path) + with open(path, newline="") as f: + source = list(csv.DictReader(f)) + source.sort(key=lambda r: r["Completed Date"]) + places = scale(source) + + rows = [] + skipped = Counter() + for r in source: + if r["State"] != "COMPLETED" or not r["Balance"]: + skipped[r["State"]] += 1 + continue + digits = places[r["Currency"]] + fee = Decimal(r["Fee"] or 0) + # the fee is deducted alongside the amount rather than booked separately + amount = Decimal(r["Amount"]) - fee + desc = r["Description"] + if fee: + desc += f" (fee {fee:.{digits}f})" + account = IBAN.search(desc) + rows.append({ + "date": datetime.strptime(r["Completed Date"][:10], "%Y-%m-%d").strftime("%d.%m.%y"), + "desc": desc, + "account": account.group() if account else "", + "amount": f"{amount:+.{digits}f}", + "balance": f"{Decimal(r['Balance']):.{digits}f}", + "currency": r["Currency"], + "type": r["Type"], + }) + for state, count in sorted(skipped.items()): + print(f" skipped {count} {state or 'unfinished'} transactions") + return rows + + +def check_balances(rows, path): + """Every balance must be the previous one plus the transaction amount.""" + for currency in dict.fromkeys(row["currency"] for row in rows): + chain = [row for row in rows if row["currency"] == currency] + for prev, row in zip(chain, chain[1:]): + expected = Decimal(prev["balance"]) + Decimal(row["amount"]) + if expected != Decimal(row["balance"]): + print(f" warning: {path.name} {row['date']} {row['desc'][:40]!r} " + f"balance {row['balance']} != {expected}") + + +def main(): + target = Path(sys.argv[1]) + paths = sorted(target.glob(STATEMENTS)) if target.is_dir() else [target] + out_path = target.with_suffix(".csv") + if out_path in paths: + sys.exit(f"{out_path} would overwrite the statement it reads; pass the directory instead") + + all_rows = [] + for path in paths: + rows = extract_transactions(path) + check_balances(rows, path) + all_rows.extend(rows) + + with open(out_path, "w", newline="") as f: + writer = csv.DictWriter(f, fieldnames=FIELDS) + writer.writeheader() + writer.writerows(all_rows) + + print(f"Wrote {len(all_rows)} transactions to {out_path}") + + +if __name__ == "__main__": + main() diff --git a/nlb-export/trade_republic.py b/nlb-export/trade_republic.py index 93a6c44..589ee36 100644 --- a/nlb-export/trade_republic.py +++ b/nlb-export/trade_republic.py @@ -5,7 +5,7 @@ import sys from datetime import datetime from decimal import Decimal from pathlib import Path -# extract transactions from trade republic account statement (pdf or pdftotext -layout txt) +# extract transactions from trade republic account statement pdf HEADER = re.compile(r"^\s*DATE\b.*\bMONEY IN\b.*\bMONEY OUT\b.*\bBALANCE\b") TEXT_COLUMNS = ("DATE", "TYPE", "DESCRIPTION") MONEY_COLUMNS = ("MONEY IN", "MONEY OUT", "BALANCE") @@ -16,14 +16,12 @@ IBAN = re.compile(r"\b[A-Z]{2}\d{2}[A-Z0-9]{11,30}\b") AMOUNT_SLACK = 5 -def read_text(path): - if path.suffix.lower() == ".pdf": - result = subprocess.run( - ["pdftotext", "-layout", str(path), "-"], - capture_output=True, text=True, check=True, - ) - return result.stdout - return path.read_text(encoding="utf-8") +def pdf_to_text(pdf_path): + result = subprocess.run( + ["pdftotext", "-layout", str(pdf_path), "-"], + capture_output=True, text=True, check=True, + ) + return result.stdout def column_starts(header_line): @@ -110,7 +108,7 @@ def blocks(lines): def extract_transactions(path): print(path) rows = [] - for page_text in read_text(path).split("\f"): + for page_text in pdf_to_text(path).split("\f"): page_lines = page_text.split("\n") header = next((i for i, l in enumerate(page_lines) if HEADER.match(l)), None) if header is None: @@ -134,25 +132,14 @@ def check_balances(rows, path): def main(): target = Path(sys.argv[1]) - if target.is_dir(): - paths = sorted(p for p in target.iterdir() if p.suffix.lower() in {".pdf", ".txt"}) - else: - paths = [target] + paths = sorted(target.glob("*.pdf")) if target.is_dir() else [target] out_path = target.with_suffix(".csv") all_rows = [] - seen = set() - duplicates = 0 for path in paths: rows = extract_transactions(path) check_balances(rows, path) - for row in rows: - key = (row["date"], row["desc"], row["amount"], row["balance"]) - if key in seen: - duplicates += 1 - continue - seen.add(key) - all_rows.append(row) + all_rows.extend(rows) with open(out_path, "w", newline="") as f: writer = csv.DictWriter( @@ -161,8 +148,6 @@ def main(): writer.writeheader() writer.writerows(all_rows) - if duplicates: - print(f"Skipped {duplicates} transactions already seen in an earlier statement") print(f"Wrote {len(all_rows)} transactions to {out_path}")