fuj-management/scripts/fio_utils.py

#!/usr/bin/env python3
"""Shared Fio bank fetching utilities."""

import json
import os
import re
import urllib.request
from datetime import datetime
from html.parser import HTMLParser

# ---------------------------------------------------------------------------
# Transaction fetching
# ---------------------------------------------------------------------------

class FioTableParser(HTMLParser):
    """Parse the second <table class="table"> on the Fio transparent page.

    Columns: Datum | Částka | Typ | Název protiúčtu | Zpráva pro příjemce | KS | VS | SS | Poznámka
    Indices:   0       1       2          3                   4              5    6    7      8
    """

    def __init__(self):
        super().__init__()
        self._table_count = 0
        self._in_target_table = False
        self._in_thead = False
        self._in_row = False
        self._in_cell = False
        self._current_row: list[str] = []
        self._rows: list[list[str]] = []
        self._cell_text = ""

    def handle_starttag(self, tag, attrs):
        cls = dict(attrs).get("class", "")
        if tag == "table" and "table" in cls.split():
            self._table_count += 1
            if self._table_count == 2:
                self._in_target_table = True
        if self._in_target_table:
            if tag == "thead":
                self._in_thead = True
            if tag == "tr" and not self._in_thead:
                self._in_row = True
                self._current_row = []
            if self._in_row and tag in ("td", "th"):
                self._in_cell = True
                self._cell_text = ""

    def handle_endtag(self, tag):
        if self._in_cell and tag in ("td", "th"):
            self._in_cell = False
            self._current_row.append(self._cell_text.strip())
        if tag == "thead":
            self._in_thead = False
        if self._in_row and tag == "tr":
            self._in_row = False
            if self._current_row:
                self._rows.append(self._current_row)
        if tag == "table" and self._in_target_table:
            self._in_target_table = False

    def handle_data(self, data):
        if self._in_cell:
            self._cell_text += data

    def get_rows(self) -> list[list[str]]:
        return self._rows


# Fio transparent table column indices
_COL_DATE = 0
_COL_AMOUNT = 1
_COL_SENDER = 3
_COL_MESSAGE = 4
_COL_KS = 5
_COL_VS = 6
_COL_SS = 7
_COL_NOTE = 8


def parse_czech_amount(s: str) -> float | None:
    """Parse '1 500,00 CZK' to float."""
    s = s.replace("\xa0", "").replace(" ", "").replace(",", ".")
    s = re.sub(r"[A-Za-z]+", "", s).strip()
    try:
        return float(s)
    except ValueError:
        return None


def parse_czech_date(s: str) -> str | None:
    """Parse 'DD.MM.YYYY' to 'YYYY-MM-DD'."""
    s = s.strip()
    for fmt in ("%d.%m.%Y", "%d/%m/%Y"):
        try:
            return datetime.strptime(s, fmt).strftime("%Y-%m-%d")
        except ValueError:
            continue
    return None


def fetch_transactions_transparent(
    date_from: str, date_to: str, account_id: str = "2800359168"
) -> list[dict]:
    """Fetch transactions from Fio transparent account HTML page.

    Args:
        date_from: D.M.YYYY format
        date_to: D.M.YYYY format
    """
    url = (
        f"https://ib.fio.cz/ib/transparent?a={account_id}"
        f"&f={date_from}&t={date_to}"
    )
    req = urllib.request.Request(url)
    with urllib.request.urlopen(req) as resp:
        html = resp.read().decode("utf-8")

    parser = FioTableParser()
    parser.feed(html)
    rows = parser.get_rows()

    transactions = []
    for row in rows:
        if len(row) < 5:
            continue

        def col(i):
            return row[i].strip() if i < len(row) else ""

        date_str = parse_czech_date(col(_COL_DATE))
        amount = parse_czech_amount(col(_COL_AMOUNT))

        if date_str is None or amount is None or amount <= 0:
            continue

        transactions.append({
            "date": date_str,
            "amount": amount,
            "sender": col(_COL_SENDER),
            "message": col(_COL_MESSAGE),
            "ks": col(_COL_KS),
            "vs": col(_COL_VS),
            "ss": col(_COL_SS),
            "note": col(_COL_NOTE),
            "bank_id": "", # HTML scraping doesn't give stable ID
        })

    return transactions


def fetch_transactions_api(
    token: str, date_from: str, date_to: str
) -> list[dict]:
    """Fetch transactions via Fio REST API (JSON).

    Args:
        token: Fio API token
        date_from: YYYY-MM-DD format
        date_to: YYYY-MM-DD format
    """
    url = (
        f"https://fioapi.fio.cz/v1/rest/periods/{token}"
        f"/{date_from}/{date_to}/transactions.json"
    )
    req = urllib.request.Request(url)
    with urllib.request.urlopen(req) as resp:
        data = json.loads(resp.read().decode("utf-8"))

    transactions = []
    tx_list = data.get("accountStatement", {}).get("transactionList", {})
    for tx in (tx_list.get("transaction") or []):
        # Each field is {"value": ..., "name": ..., "id": ...} or null
        def val(col_id):
            col = tx.get(f"column{col_id}")
            return col["value"] if col else ""

        amount = float(val(1) or 0)
        if amount <= 0:
            continue  # Skip outgoing

        date_raw = val(0) or ""
        # API returns date as "YYYY-MM-DD+HHMM" or ISO format
        date_str = date_raw[:10] if date_raw else ""

        transactions.append({
            "date": date_str,
            "amount": amount,
            "sender": str(val(10) or ""),     # column10 = sender name
            "message": str(val(16) or ""),     # column16 = message for recipient
            "vs": str(val(5) or ""),           # column5 = VS
            "ks": str(val(4) or ""),           # column4 = KS
            "ss": str(val(6) or ""),           # column6 = SS
            "user_id": str(val(7) or ""),      # column7 = user identification
            "sender_account": str(val(2) or ""),  # column2 = sender account
            "bank_id": str(val(22) or ""),     # column22 = ID operace
            "currency": str(val(14) or "CZK"), # column14 = Currency
        })

    return transactions


def fetch_transactions(date_from: str, date_to: str) -> list[dict]:
    """Fetch transactions, using API if token available, else transparent page."""
    token = os.environ.get("FIO_API_TOKEN", "").strip()
    if token:
        return fetch_transactions_api(token, date_from, date_to)

    # Convert YYYY-MM-DD to DD.MM.YYYY for the transparent page URL
    from_dt = datetime.strptime(date_from, "%Y-%m-%d")
    to_dt = datetime.strptime(date_to, "%Y-%m-%d")
    return fetch_transactions_transparent(
        from_dt.strftime("%d.%m.%Y"),
        to_dt.strftime("%d.%m.%Y"),
    )