109 lines
3.3 KiB
Python
109 lines
3.3 KiB
Python
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")
|
|
TRANSACTION_LINE = re.compile(
|
|
r"^(?P<date>\d{2}\.\d{2}\.\d{2})\s+"
|
|
r"(?P<middle>.*?)\s+"
|
|
r"(?P<amount>[+-][\d.,]+)\s+"
|
|
r"(?P<balance>-?[\d.,]+-?)\s*$"
|
|
)
|
|
CONTINUATION_INDENT = 15
|
|
FIELD_SPLIT = re.compile(r"\s{4,}")
|
|
ACCOUNT = re.compile(r"^SI\d{2}(?:\s?\d{4}){3}\s?\d{3}$")
|
|
|
|
|
|
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 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 = ""
|
|
desc_parts = []
|
|
for f in FIELD_SPLIT.split(m.group("middle").strip()):
|
|
if ACCOUNT.match(f):
|
|
account = f
|
|
else:
|
|
desc_parts.append(f)
|
|
return {
|
|
"date": m.group("date"),
|
|
"desc": " ".join(desc_parts),
|
|
"account": account,
|
|
"amount": f"{to_number(m.group('amount')):+.2f}",
|
|
"balance": f"{to_number(m.group('balance')):.2f}",
|
|
"currency": "EUR",
|
|
}
|
|
|
|
|
|
def extract_transactions(pdf_path):
|
|
print(pdf_path)
|
|
rows = []
|
|
full_text = pdf_to_text(pdf_path)
|
|
for page_text in full_text.split("\f"):
|
|
page_start = len(rows)
|
|
for line in page_text.split("\n"):
|
|
if not line.strip():
|
|
continue
|
|
if TRANSACTION_START.match(line):
|
|
rows.append(parse_transaction(line))
|
|
continue
|
|
if len(rows) <= page_start:
|
|
continue
|
|
indent = len(line) - len(line.lstrip(" "))
|
|
if indent < CONTINUATION_INDENT:
|
|
continue
|
|
parts = FIELD_SPLIT.split(line.strip())
|
|
rows[-1]["desc"] += " " + parts[0]
|
|
if len(parts) > 1:
|
|
rows[-1]["account"] = (rows[-1]["account"] + " " + parts[1]).strip()
|
|
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")):
|
|
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"])
|
|
writer.writeheader()
|
|
writer.writerows(all_rows)
|
|
|
|
print(f"Wrote {len(all_rows)} transactions to {out_path}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|