feat: implement automated payment inference and sync to Google Sheets
This commit is contained in:
Jan Novak
215
scripts/fio_utils.py
Normal file
215
scripts/fio_utils.py
Normal file
@@ -0,0 +1,215 @@
|
||||
#!/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"),
|
||||
)
|
||||
Reference in New Issue
Block a user