270 lines
7.8 KiB
Python
270 lines
7.8 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
import os
|
|
import json
|
|
import time
|
|
from datetime import date, timedelta
|
|
from pathlib import Path
|
|
|
|
import requests
|
|
import pymysql
|
|
|
|
# =========================================
|
|
# CONFIG
|
|
# =========================================
|
|
ACCOUNTS_FILE = r"u:\PycharmProjects\FIO\accounts.json"
|
|
JSON_BASE_DIR = r"u:\Dropbox\!!!Days\Downloads Z230\Fio" # kam se budou ukládat JSONy
|
|
|
|
DB = {
|
|
"host": "192.168.1.76",
|
|
"port": 3307,
|
|
"user": "root",
|
|
"password": "Vlado9674+", # uprav podle sebe / dej do .env
|
|
"database": "fio",
|
|
"charset": "utf8mb4",
|
|
}
|
|
|
|
BATCH_SIZE = 500
|
|
DAYS_BACK = 90
|
|
|
|
|
|
# =========================================
|
|
# HELPERS
|
|
# =========================================
|
|
def load_accounts(path: str):
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
accounts = json.load(f)
|
|
|
|
# jednoduchá validace
|
|
for acc in accounts:
|
|
for key in ("name", "account_number", "token"):
|
|
if key not in acc:
|
|
raise ValueError(f"Missing '{key}' in account config: {acc}")
|
|
|
|
return accounts
|
|
|
|
|
|
def fio_url_for_period(token: str, d_from: date, d_to: date) -> str:
|
|
from_str = d_from.strftime("%Y-%m-%d")
|
|
to_str = d_to.strftime("%Y-%m-%d")
|
|
return f"https://fioapi.fio.cz/v1/rest/periods/{token}/{from_str}/{to_str}/transactions.json"
|
|
|
|
|
|
def fetch_fio_json(token: str, d_from: date, d_to: date):
|
|
url = fio_url_for_period(token, d_from, d_to)
|
|
resp = requests.get(url, timeout=30)
|
|
if resp.status_code != 200:
|
|
print(f" ❌ HTTP {resp.status_code} from Fio: {url}")
|
|
return None
|
|
try:
|
|
return resp.json()
|
|
except json.JSONDecodeError:
|
|
print(" ❌ Cannot decode JSON from Fio response")
|
|
return None
|
|
|
|
|
|
def safe_col(t: dict, n: int):
|
|
"""
|
|
Safely read t['columnN']['value'], i.e. Fio column.
|
|
Handles:
|
|
- missing columnN
|
|
- columnN is None
|
|
- missing 'value'
|
|
"""
|
|
key = f"column{n}"
|
|
val = t.get(key)
|
|
if not val:
|
|
return None
|
|
return val.get("value")
|
|
|
|
|
|
def clean_date(dt_str: str):
|
|
"""
|
|
Convert Fio date '2025-10-26+0200' -> '2025-10-26'
|
|
Fio spec: date is always rrrr-mm-dd+GMT.
|
|
"""
|
|
if not dt_str:
|
|
return None
|
|
return dt_str[:10]
|
|
|
|
|
|
def ensure_dir(path: Path):
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
def save_json_for_account(base_dir: str, account_cfg: dict, data: dict, d_from: date, d_to: date):
|
|
"""
|
|
Uloží JSON do podsložky dle čísla účtu, název souboru podle období.
|
|
"""
|
|
acc_num_raw = account_cfg["account_number"]
|
|
acc_folder_name = acc_num_raw.replace("/", "_") # 2101234567_2700
|
|
|
|
out_dir = Path(base_dir) / acc_folder_name
|
|
ensure_dir(out_dir)
|
|
|
|
filename = f"{d_from.strftime('%Y-%m-%d')}_to_{d_to.strftime('%Y-%m-%d')}.json"
|
|
out_path = out_dir / filename
|
|
|
|
with open(out_path, "w", encoding="utf-8") as f:
|
|
json.dump(data, f, ensure_ascii=False, indent=2)
|
|
|
|
return out_path
|
|
|
|
|
|
# =========================================
|
|
# MAIN IMPORT
|
|
# =========================================
|
|
def main():
|
|
start_all = time.time()
|
|
|
|
# období posledních 90 dní
|
|
today = date.today()
|
|
d_from = today - timedelta(days=DAYS_BACK)
|
|
d_to = today
|
|
|
|
print(f"=== Fio multi-account import ===")
|
|
print(f"Období: {d_from} až {d_to}")
|
|
print("Načítám účty z JSON konfigurace...")
|
|
|
|
accounts = load_accounts(ACCOUNTS_FILE)
|
|
print(f" Účtů v konfiguraci: {len(accounts)}\n")
|
|
|
|
# Připojení do DB
|
|
conn = pymysql.connect(**DB)
|
|
cur = conn.cursor()
|
|
|
|
# SQL s ON DUPLICATE KEY UPDATE
|
|
sql = """
|
|
INSERT INTO transactions
|
|
(
|
|
datum, objem, mena, cislo_uctu, protiucet, kod_banky,
|
|
ks, vs, ss, zprava_pro_prijemce, poznamka,
|
|
id_operace, id_pokynu, ks_1, nazev_banky, nazev_protiuctu,
|
|
ss_1, typ, upresneni_objem, upresneni_mena, vs_1, zadal
|
|
)
|
|
VALUES
|
|
(
|
|
%(datum)s, %(objem)s, %(mena)s, %(cislo_uctu)s, %(protiucet)s, %(kod_banky)s,
|
|
%(ks)s, %(vs)s, %(ss)s, %(zprava)s, %(poznamka)s,
|
|
%(id_operace)s, %(id_pokynu)s, %(ks1)s, %(nazev_banky)s, %(nazev_protiuctu)s,
|
|
%(ss1)s, %(typ)s, %(upr_objem)s, %(upr_mena)s, %(vs1)s, %(zadal)s
|
|
)
|
|
ON DUPLICATE KEY UPDATE
|
|
datum = VALUES(datum),
|
|
objem = VALUES(objem),
|
|
mena = VALUES(mena),
|
|
protiucet = VALUES(protiucet),
|
|
kod_banky = VALUES(kod_banky),
|
|
ks = VALUES(ks),
|
|
vs = VALUES(vs),
|
|
ss = VALUES(ss),
|
|
zprava_pro_prijemce = VALUES(zprava_pro_prijemce),
|
|
poznamka = VALUES(poznamka),
|
|
ks_1 = VALUES(ks_1),
|
|
nazev_banky = VALUES(nazev_banky),
|
|
nazev_protiuctu = VALUES(nazev_protiuctu),
|
|
ss_1 = VALUES(ss_1),
|
|
typ = VALUES(typ),
|
|
upresneni_objem = VALUES(upresneni_objem),
|
|
upresneni_mena = VALUES(upresneni_mena),
|
|
vs_1 = VALUES(vs_1),
|
|
zadal = VALUES(zadal)
|
|
"""
|
|
|
|
total_inserted = 0
|
|
|
|
for acc in accounts:
|
|
name = acc["name"]
|
|
cfg_acc_num = acc["account_number"]
|
|
token = acc["token"]
|
|
|
|
print(f"--- Účet: {name} ({cfg_acc_num}) ---")
|
|
t0 = time.time()
|
|
|
|
data = fetch_fio_json(token, d_from, d_to)
|
|
if data is None:
|
|
print(" Přeskakuji, žádná data / chyba API.\n")
|
|
continue
|
|
|
|
# volitelné uložení JSON
|
|
json_path = save_json_for_account(JSON_BASE_DIR, acc, data, d_from, d_to)
|
|
print(f" JSON uložen do: {json_path}")
|
|
|
|
# extrakce transakcí
|
|
tlist = data["accountStatement"]["transactionList"].get("transaction", [])
|
|
if isinstance(tlist, dict):
|
|
tlist = [tlist]
|
|
|
|
print(f" Počet transakcí v období: {len(tlist)}")
|
|
|
|
if not tlist:
|
|
print(" Žádné transakce, jdu dál.\n")
|
|
continue
|
|
|
|
fio_acc_id = data["accountStatement"]["info"]["accountId"]
|
|
if cfg_acc_num and cfg_acc_num.split("/")[0] not in fio_acc_id:
|
|
# jen varování, ne fatální chyba
|
|
print(f" ⚠ Upozornění: accountId z Fio ({fio_acc_id}) "
|
|
f"se neshoduje s account_number v konfiguraci ({cfg_acc_num})")
|
|
|
|
# připravit řádky pro batch insert
|
|
rows = []
|
|
for t in tlist:
|
|
row = {
|
|
"datum": clean_date(safe_col(t, 0)),
|
|
"objem": safe_col(t, 1),
|
|
"mena": safe_col(t, 14),
|
|
|
|
"cislo_uctu": fio_acc_id,
|
|
"protiucet": safe_col(t, 2),
|
|
"kod_banky": safe_col(t, 3),
|
|
|
|
"ks": safe_col(t, 4),
|
|
"vs": safe_col(t, 5),
|
|
"ss": safe_col(t, 6),
|
|
|
|
"zprava": safe_col(t, 16),
|
|
"poznamka": safe_col(t, 25),
|
|
|
|
"id_operace": safe_col(t, 22),
|
|
"id_pokynu": safe_col(t, 24),
|
|
|
|
"ks1": safe_col(t, 18),
|
|
"nazev_banky": safe_col(t, 15),
|
|
"nazev_protiuctu": safe_col(t, 10),
|
|
|
|
"ss1": safe_col(t, 19),
|
|
"typ": safe_col(t, 8),
|
|
|
|
"upr_objem": safe_col(t, 20),
|
|
"upr_mena": safe_col(t, 21),
|
|
"vs1": safe_col(t, 17),
|
|
|
|
"zadal": safe_col(t, 12),
|
|
}
|
|
rows.append(row)
|
|
|
|
# batch insert
|
|
inserted = 0
|
|
for i in range(0, len(rows), BATCH_SIZE):
|
|
chunk = rows[i:i + BATCH_SIZE]
|
|
cur.executemany(sql, chunk)
|
|
conn.commit()
|
|
inserted += len(chunk)
|
|
|
|
elapsed = time.time() - t0
|
|
total_inserted += inserted
|
|
print(f" ✓ Zapsáno (insert/update): {inserted} řádků do DB za {elapsed:.2f} s\n")
|
|
|
|
cur.close()
|
|
conn.close()
|
|
|
|
total_elapsed = time.time() - start_all
|
|
print(f"=== Hotovo. Celkem zapsáno {total_inserted} transakcí. "
|
|
f"Celkový čas: {total_elapsed:.2f} s ===")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|