administrator/fio
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

merge into: administrator/fio:main
Branches Tags
administrator/fio:main administrator/fio:backup-original-main
..
pull from: administrator/fio:backup-original-main
Branches Tags
administrator/fio:main administrator/fio:backup-original-main

1 Commits
main .. backup-original-main

Author SHA1 Message Date
administrator fd2a811f38 reporter 2026-01-03 12:21:03 +01:00
17 changed files with 503 additions and 2019 deletions
Expand all files Collapse all files
.gitignore
-21
View File
@@ -1,21 +0,0 @@
# Python
__pycache__/
*.py[cod]
*.pyo
# Virtual environment
.venv/
# PyInstaller build artifacts
build/
dist/
*.spec
# IDE
.idea/
# Claude
.claude/
# Output folders
QRPlatby/
.idea/FIO.iml
Generated
+10
View File
@@ -0,0 +1,10 @@
<?xml version="1.0" encoding="UTF-8"?>
<module type="PYTHON_MODULE" version="4">
<component name="NewModuleRootManager">
<content url="file://$MODULE_DIR$">
<excludeFolder url="file://$MODULE_DIR$/.venv" />
</content>
<orderEntry type="jdk" jdkName="Python 3.13 (FIO)" jdkType="Python SDK" />
<orderEntry type="sourceFolder" forTests="false" />
</component>
</module>
.idea/misc.xml
Generated
+7
View File
@@ -0,0 +1,7 @@
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
<component name="Black">
<option name="sdkName" value="Python 3.12 (FIO)" />
</component>
<component name="ProjectRootManager" version="2" project-jdk-name="Python 3.13 (FIO)" project-jdk-type="Python SDK" />
</project>
2025-11-30 final reporter/20 Jednorázové načtení FIO JSONs.py
-382
View File
@@ -1,382 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
FIO JSON ARCHIVE → MYSQL IMPORTER (ONE-TIME REIMPORT)
=====================================================
CO TOHLE DĚLÁ
-------------
Tento skript je určený pro JEDNORÁZOVÝ import historických transakcí,
které už máš uložené jako RAW JSON soubory na disku (archiv).
✅ NEVOLÁ Fio API (žádné stahování)
✅ Rekurzivně projde celý adresář:
U:\Dropbox\!!!Days\Downloads Z230\Fio\
a najde všechny *.json soubory (typicky ve složkách podle účtu)
✅ Z každého JSONu:
- vytáhne accountId (ID účtu z JSON hlavičky)
- vytáhne transaction list
- mapuje Fio "columnN" hodnoty na sloupce tabulky v MySQL
✅ Poté vkládá data do MySQL tabulky:
fio.transactions
IDEMPOTENTNÍ CHOVÁNÍ (DŮLEŽITÉ)
------------------------------
Tabulka má primární klíč:
PRIMARY KEY (cislo_uctu, id_operace)
Proto je INSERT napsaný jako:
INSERT ... ON DUPLICATE KEY UPDATE ...
To znamená:
- pokud (cislo_uctu, id_operace) ještě v DB není → vloží nový řádek
- pokud už tam je → neudělá duplicitu, ale může aktualizovat vybraná pole
Takže skript můžeš spustit opakovaně bez rizika duplikací.
CO MUSÍ EXISTOVAT
-----------------
1) MySQL DB: fio
2) Tabulka: fio.transactions (podle DDL co jsme skládali z ibd2sdi SDI)
VÝKON
-----
- Používá executemany() a dávky (BATCH_SIZE), aby to bylo rychlé
- Commituje po dávkách
POZNÁMKA K CHYBÁM
-----------------
- Pokud narazí na rozbitý JSON nebo neočekávanou strukturu, soubor přeskočí
a pokračuje dál (aby jednorázový import doběhl celý).
"""
import json
from pathlib import Path
import pymysql
# ======================================================
# CONFIG
# ======================================================
# Základní adresář s archivními JSON soubory z Fio API
# Očekáváme strukturu například:
# U:\Dropbox\!!!Days\Downloads Z230\Fio\2100046291_2010\2025-01-01_to_2025-03-31.json
JSON_BASE_DIR = Path(r"u:\Dropbox\!!!Days\Downloads Z230\Fio")
# Připojení do MySQL (pozor: port 3307 dle tvého prostředí)
DB = {
"host": "192.168.1.76",
"port": 3306,
"user": "root",
"password": "Vlado9674+",
"database": "fio",
"charset": "utf8mb4",
}
# Kolik řádků posílat do DB v jedné dávce (performance)
BATCH_SIZE = 500
# ======================================================
# HELPERS
# ======================================================
def safe_col(tx: dict, n: int):
"""
Bezpečné čtení hodnoty Fio JSON "columnN".
Fio JSON pro jednu transakci vypadá typicky takto:
{
"column0": { "name": "Datum", "value": "2025-02-14+0100" },
"column1": { "name": "Objem", "value": 123.45 },
...
}
Ale:
- některé columnN vůbec nemusí existovat
- nebo mohou být None
Tato funkce tedy:
vrátí tx["columnN"]["value"]
nebo:
None (pokud tam columnN není)
"""
obj = tx.get(f"column{n}")
if not obj:
return None
return obj.get("value")
def clean_date(dt_str):
"""
Fio vrací datum často v podobě:
"YYYY-MM-DD+0100"
MySQL DATE chceme jen:
"YYYY-MM-DD"
Pokud je dt_str None/empty, vrátí None.
"""
if not dt_str:
return None
return str(dt_str)[:10]
def as_list(maybe_list_or_dict):
"""
Fio někdy vrací:
- seznam transakcí (list)
jindy:
- jedinou transakci jako dict
Tady zaručíme, že vždy pracujeme se seznamem.
"""
if maybe_list_or_dict is None:
return []
if isinstance(maybe_list_or_dict, dict):
return [maybe_list_or_dict]
return list(maybe_list_or_dict)
# ======================================================
# SQL (MAPOVÁNÍ NA TVOU TABULKU)
# ======================================================
# Vkládáme do tabulky fio.transactions (sloupce dle DDL z SDI).
# Pozn.: sloupec stazeno_kdy neuvádíme -> vyplní se automaticky DEFAULT CURRENT_TIMESTAMP
SQL_INSERT = """
INSERT INTO transactions
(
id_operace,
cislo_uctu,
transaction_date,
amount,
currency,
protiucet,
kod_banky,
nazev_protiuctu,
nazev_banky,
typ,
vs,
ks,
ss,
zprava_pro_prijemce,
uziv_identifikace,
provedl,
id_pokynu,
komentar,
upr_objem_mena,
api_bic,
reference_platce
)
VALUES
(
%(id_operace)s,
%(cislo_uctu)s,
%(transaction_date)s,
%(amount)s,
%(currency)s,
%(protiucet)s,
%(kod_banky)s,
%(nazev_protiuctu)s,
%(nazev_banky)s,
%(typ)s,
%(vs)s,
%(ks)s,
%(ss)s,
%(zprava_pro_prijemce)s,
%(uziv_identifikace)s,
%(provedl)s,
%(id_pokynu)s,
%(komentar)s,
%(upr_objem_mena)s,
%(api_bic)s,
%(reference_platce)s
)
ON DUPLICATE KEY UPDATE
-- když už transakce existuje, aktualizujeme vybraná pole
transaction_date = VALUES(transaction_date),
amount = VALUES(amount),
currency = VALUES(currency),
protiucet = VALUES(protiucet),
kod_banky = VALUES(kod_banky),
nazev_protiuctu = VALUES(nazev_protiuctu),
nazev_banky = VALUES(nazev_banky),
typ = VALUES(typ),
vs = VALUES(vs),
ks = VALUES(ks),
ss = VALUES(ss),
zprava_pro_prijemce= VALUES(zprava_pro_prijemce),
uziv_identifikace = VALUES(uziv_identifikace),
provedl = VALUES(provedl),
id_pokynu = VALUES(id_pokynu),
komentar = VALUES(komentar),
upr_objem_mena = VALUES(upr_objem_mena),
api_bic = VALUES(api_bic),
reference_platce = VALUES(reference_platce)
"""
# ======================================================
# MAIN
# ======================================================
def main():
"""
Hlavní běh:
1) Najde všechny *.json soubory pod JSON_BASE_DIR (rekurzivně)
2) Každý JSON načte a zkusí vytáhnout:
- accountId
- transactionList.transaction
3) Pro každou transakci složí dict hodnot podle SQL_INSERT
4) Vkládá do DB po dávkách (BATCH_SIZE) přes executemany()
"""
print("=== FIO JSON ARCHIVE → MYSQL IMPORTER ===")
print(f"Base dir: {JSON_BASE_DIR}")
# 1) Najdi všechny JSON soubory rekurzivně
json_files = sorted(JSON_BASE_DIR.rglob("*.json"))
print(f"Nalezeno JSON souborů: {len(json_files)}")
if not json_files:
print("Nic k importu. Končím.")
return
# 2) Připoj se k DB
conn = pymysql.connect(**DB)
cur = conn.cursor()
total_rows_seen = 0 # kolik transakcí jsme celkem naparsovali (všech souborů)
total_rows_sent = 0 # kolik jsme celkem poslali do DB (insert/update attempt)
files_ok = 0
files_skipped = 0
try:
# 3) Zpracuj každý JSON soubor
for idx, jf in enumerate(json_files, start=1):
print(f"\n[{idx}/{len(json_files)}] Soubor: {jf}")
# 3a) načtení JSON souboru
try:
data = json.loads(jf.read_text(encoding="utf-8"))
except Exception as e:
print(f" ❌ Nejde načíst JSON ({e}) → přeskočeno")
files_skipped += 1
continue
# 3b) vytažení listu transakcí a accountId
try:
fio_acc_id = data["accountStatement"]["info"]["accountId"]
t_raw = data["accountStatement"]["transactionList"].get("transaction", [])
except Exception:
print(" ❌ Neočekávaná struktura JSON (chybí accountStatement/info/transactionList) → přeskočeno")
files_skipped += 1
continue
tlist = as_list(t_raw)
print(f" Účet: {fio_acc_id} | transakcí v souboru: {len(tlist)}")
if not tlist:
print(" (prázdný seznam transakcí) → OK, nic nevkládám")
files_ok += 1
continue
# 3c) naparsuj všechny transakce v souboru do listu rows
rows = []
for tx in tlist:
# mapování columnN -> DB sloupce (dle tvé tabulky)
row = {
"id_operace": safe_col(tx, 22),
"cislo_uctu": fio_acc_id,
"transaction_date": clean_date(safe_col(tx, 0)),
"amount": safe_col(tx, 1),
"currency": safe_col(tx, 14),
"protiucet": safe_col(tx, 2),
"kod_banky": safe_col(tx, 3),
"nazev_protiuctu": safe_col(tx, 10),
"nazev_banky": safe_col(tx, 15),
"typ": safe_col(tx, 8),
"vs": safe_col(tx, 5),
"ks": safe_col(tx, 4),
"ss": safe_col(tx, 6),
"zprava_pro_prijemce": safe_col(tx, 16),
# v SDI tabulce existuje sloupec uziv_identifikace (TEXT)
# ve starém importeru se brala hodnota z column12
"uziv_identifikace": safe_col(tx, 12),
# Pozor: ve tvé tabulce je "provedl" (VARCHAR(50)).
# Starý importer bral "provedl" typicky z column9.
"provedl": safe_col(tx, 9),
"id_pokynu": safe_col(tx, 19),
# textová poznámka / komentář (starý importer používal column25 jako "poznamka")
"komentar": safe_col(tx, 25),
# upřesnění objemu/měny (column20 / column21)
# v tabulce máme jen jeden sloupec upr_objem_mena (varchar 255)
# a zvlášť "currency" už je column14, takže sem dáváme column20 (Upřesnění)
"upr_objem_mena": safe_col(tx, 20),
# BIC a reference plátce (dle komentářů v SDI)
"api_bic": safe_col(tx, 26),
"reference_platce": safe_col(tx, 27),
}
# Minimální sanity check:
# PRIMARY KEY je (cislo_uctu, id_operace),
# takže pokud id_operace chybí, nebude to unikátní transakce.
# Takové řádky přeskočíme (jinak by to dělalo problémy).
if not row["id_operace"]:
continue
rows.append(row)
total_rows_seen += len(tlist)
print(f" Naparsováno validních řádků (s id_operace): {len(rows)}")
if not rows:
print(" (po filtraci nic nezbylo) → OK")
files_ok += 1
continue
# 3d) insert do DB po dávkách
inserted_attempt = 0
for i in range(0, len(rows), BATCH_SIZE):
chunk = rows[i:i + BATCH_SIZE]
cur.executemany(SQL_INSERT, chunk)
conn.commit()
inserted_attempt += len(chunk)
total_rows_sent += inserted_attempt
files_ok += 1
print(f" ✓ Odesláno do DB (insert/update attempt): {inserted_attempt}")
finally:
cur.close()
conn.close()
print("\n=== HOTOVO ===")
print(f"Souborů OK: {files_ok}")
print(f"Souborů přeskočeno: {files_skipped}")
print(f"Celkem transakcí nalezeno v JSON (raw): {total_rows_seen}")
print(f"Celkem posláno do DB (insert/update attempt): {total_rows_sent}")
print("Pozn.: 'posláno do DB' neznamená čisté INSERTy — část mohla být UPDATE (ON DUPLICATE KEY).")
if __name__ == "__main__":
main()
2025-11-30 final reporter/21ReadJSONmultipleaccounts.py
-351
View File
@@ -1,351 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import sys
import io
# Force UTF-8 output for Scheduled Tasks
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8', errors='replace')
"""
FIO MULTI–ACCOUNT IMPORTER — FULLY COMMENTED VERSION
====================================================
This script downloads transactions for **multiple Fio bank accounts**
(using their API tokens) and imports them into a MySQL database
(`fio.transactions` table).
It also saves the raw JSON responses into a folder structure
for backup / debugging / later use.
Main features:
• Reads all accounts from accounts.json
• Downloads last N days (default 90)
• Saves JSON files to disk
• Extracts all transactions with safe parsing
• Inserts into MySQL with ON DUPLICATE KEY UPDATE
• Efficient batch insertion (executemany)
"""
import os
import json
import time
from datetime import date, timedelta
from pathlib import Path
import requests # used to call Fio REST API
import pymysql # MySQL driver
# =========================================
# CONFIGURATION
# =========================================
# JSON file containing multiple account configs:
# [
# { "name": "CZK rodina", "account_number": "2100046291", "token": "xxx" },
# ...
# ]
ACCOUNTS_FILE = r"../accounts.json"
# Directory where raw JSON files from Fio API will be stored.
JSON_BASE_DIR = r"u:\Dropbox\!!!Days\Downloads Z230\Fio"
# MySQL connection parameters
DB = {
"host": "192.168.1.76",
"port": 3306,
"user": "root",
"password": "Vlado9674+",
"database": "fio",
"charset": "utf8mb4",
}
# How many transactions insert per batch (performance tuning)
BATCH_SIZE = 500
# How many days back we load from Fio (default = last 90 days)
DAYS_BACK = 90
# =========================================
# HELPERS
# =========================================
def load_accounts(path: str):
"""
Reads accounts.json and does simple validation to ensure
each entry contains: name, account_number, token.
"""
with open(path, "r", encoding="utf-8") as f:
accounts = json.load(f)
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:
"""
Constructs the exact URL for Fio REST API "periods" endpoint.
Example:
https://fioapi.fio.cz/v1/rest/periods/<token>/2025-01-01/2025-01-31/transactions.json
"""
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):
"""
Calls Fio API and fetches JSON.
Handles HTTP errors and JSON decoding errors.
"""
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):
"""
SAFE ACCESSOR for Fio transaction column numbers.
Fio JSON schema example:
"column5": { "name": "VS", "value": "123456" }
But the structure is NOT guaranteed to exist.
So this function prevents KeyError or NoneType errors.
Returns:
t["columnN"]["value"] or None
"""
key = f"column{n}"
val = t.get(key)
if not val:
return None
return val.get("value")
def clean_date(dt_str: str):
"""
Fio returns dates like: "2025-02-14+0100"
We strip timezone → "2025-02-14"
"""
if not dt_str:
return None
return dt_str[:10]
def ensure_dir(path: Path):
"""Creates directory if it doesn’t exist."""
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):
"""
Saves raw JSON to:
<base_dir>/<account_number>/YYYY-MM-DD_to_YYYY-MM-DD.json
Useful for debugging, backups, or re-imports.
"""
acc_num_raw = account_cfg["account_number"]
acc_folder_name = acc_num_raw.replace("/", "_") # sanitize dir name for filesystem
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 LOGIC
# =========================================
def main():
start_all = time.time()
# Calculate time range (last N days)
today = date.today()
d_from = today - timedelta(days=DAYS_BACK)
d_to = today
print("=== Fio multi-account import ===")
print(f"Období: {d_from} až {d_to}")
print("Načítám účty z JSON konfigurace...")
# Load all accounts from accounts.json
accounts = load_accounts(ACCOUNTS_FILE)
print(f" Účtů v konfiguraci: {len(accounts)}\n")
# Connect to database
conn = pymysql.connect(**DB)
cur = conn.cursor()
# SQL INSERT with ON DUPLICATE KEY UPDATE
# This means: if transaction already exists (same unique key), update it.
sql = """
INSERT INTO transactions
(
datum, objem, mena, cislo_uctu, protiucet, kod_banky,
ks, vs, ss, zprava_pro_prijemce, poznamka,
id_operace, id_pokynu,
nazev_banky, nazev_protiuctu,
typ, upresneni_objem, upresneni_mena, 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,
%(nazev_banky)s, %(nazev_protiuctu)s,
%(typ)s, %(upr_objem)s, %(upr_mena)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),
nazev_banky = VALUES(nazev_banky),
nazev_protiuctu = VALUES(nazev_protiuctu),
typ = VALUES(typ),
upresneni_objem = VALUES(upresneni_objem),
upresneni_mena = VALUES(upresneni_mena),
zadal = VALUES(zadal),
id_operace = VALUES(id_operace),
id_pokynu = VALUES(id_pokynu)
"""
total_inserted = 0
# ======================================================
# PROCESS EACH ACCOUNT IN accounts.json
# ======================================================
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()
# --- 1) Download JSON from Fio API
data = fetch_fio_json(token, d_from, d_to)
if data is None:
print(" Přeskakuji, žádná data / chyba API.\n")
continue
# --- 2) Save raw JSON file to disk
json_path = save_json_for_account(JSON_BASE_DIR, acc, data, d_from, d_to)
print(f" JSON uložen do: {json_path}")
# --- 3) Extract transactions from JSON tree
tlist = data["accountStatement"]["transactionList"].get("transaction", [])
# FIO can return single transaction as an object (not list)
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 returns account ID under accountStatement.info.accountId
fio_acc_id = data["accountStatement"]["info"]["accountId"]
# Warn if account ID in JSON doesn't match config (informational only)
if cfg_acc_num and cfg_acc_num.split("/")[0] not in fio_acc_id:
print(f" ⚠ Upozornění: accountId z Fio ({fio_acc_id}) "
f"se neshoduje s account_number v konfiguraci ({cfg_acc_num})")
# --- 4) Build list of MySQL rows
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 pohybu
"id_pokynu": safe_col(t, 19), # ID pokynu
"nazev_banky": safe_col(t, 15),
"nazev_protiuctu": safe_col(t, 10),
"typ": safe_col(t, 8),
"upr_objem": safe_col(t, 20),
"upr_mena": safe_col(t, 21),
"zadal": safe_col(t, 12),
}
rows.append(row)
# --- 5) INSERT rows into MySQL in batches
inserted = 0
for i in range(0, len(rows), BATCH_SIZE):
chunk = rows[i : i + BATCH_SIZE]
cur.executemany(sql, chunk) # fast multi-row insert/update
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")
# Close DB
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 ===")
# ======================================================
# ENTRY POINT
# ======================================================
if __name__ == "__main__":
main()
2025-11-30 final reporter/22ReadJSONmultipleaccounts.py
-259
View File
@@ -1,259 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import sys
import io
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8', errors='replace')
import os
import json
import time
from datetime import date, timedelta
from pathlib import Path
import requests
import pymysql
# =========================================
# CONFIGURATION
# =========================================
ACCOUNTS_FILE = r"../accounts.json"
JSON_BASE_DIR = r"u:\Dropbox\!!!Days\Downloads Z230\Fio"
DB = {
"host": "192.168.1.76",
"port": 3306,
"user": "root",
"password": "Vlado9674+",
"database": "fio",
"charset": "utf8mb4",
}
BATCH_SIZE = 500
DAYS_BACK = 90
# =========================================
# HELPERS
# =========================================
def load_accounts(path):
with open(path, "r", encoding="utf-8") as f:
accounts = json.load(f)
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, d_from, d_to):
return f"https://fioapi.fio.cz/v1/rest/periods/{token}/{d_from:%Y-%m-%d}/{d_to:%Y-%m-%d}/transactions.json"
def fetch_fio_json(token, d_from, d_to):
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")
return None
return resp.json()
def safe_col(t, n):
val = t.get(f"column{n}")
return None if not val else val.get("value")
def clean_date(dt_str):
return None if not dt_str else dt_str[:10]
def ensure_dir(path):
Path(path).mkdir(parents=True, exist_ok=True)
def save_json_for_account(base_dir, account_cfg, data, d_from, d_to):
acc_folder = account_cfg["account_number"].replace("/", "_")
out_dir = Path(base_dir) / acc_folder
ensure_dir(out_dir)
filename = f"{d_from:%Y-%m-%d}_to_{d_to:%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
# =========================================
def main():
today = date.today()
d_from = today - timedelta(days=DAYS_BACK)
d_to = today
print(f"=== Fio import {d_from} → {d_to} ===")
accounts = load_accounts(ACCOUNTS_FILE)
conn = pymysql.connect(**DB)
cur = conn.cursor()
sql = """
INSERT INTO transactions
(
id_operace,
cislo_uctu,
transaction_date,
amount,
currency,
protiucet,
kod_banky,
nazev_protiuctu,
nazev_banky,
typ,
vs,
ks,
ss,
zprava_pro_prijemce,
uziv_identifikace,
provedl,
id_pokynu,
komentar,
upr_objem_mena,
api_bic,
reference_platce
)
VALUES
(
%(id_operace)s,
%(cislo_uctu)s,
%(transaction_date)s,
%(amount)s,
%(currency)s,
%(protiucet)s,
%(kod_banky)s,
%(nazev_protiuctu)s,
%(nazev_banky)s,
%(typ)s,
%(vs)s,
%(ks)s,
%(ss)s,
%(zprava)s,
%(uziv_identifikace)s,
%(provedl)s,
%(id_pokynu)s,
%(komentar)s,
%(upr_objem_mena)s,
%(api_bic)s,
%(reference_platce)s
)
ON DUPLICATE KEY UPDATE
transaction_date = VALUES(transaction_date),
amount = VALUES(amount),
currency = VALUES(currency),
protiucet = VALUES(protiucet),
kod_banky = VALUES(kod_banky),
nazev_protiuctu = VALUES(nazev_protiuctu),
nazev_banky = VALUES(nazev_banky),
typ = VALUES(typ),
vs = VALUES(vs),
ks = VALUES(ks),
ss = VALUES(ss),
zprava_pro_prijemce = VALUES(zprava_pro_prijemce),
uziv_identifikace = VALUES(uziv_identifikace),
provedl = VALUES(provedl),
id_pokynu = VALUES(id_pokynu),
komentar = VALUES(komentar),
upr_objem_mena = VALUES(upr_objem_mena),
api_bic = VALUES(api_bic),
reference_platce = VALUES(reference_platce)
"""
total_inserted = 0
for acc in accounts:
print(f"\n--- {acc['name']} ---")
data = fetch_fio_json(acc["token"], d_from, d_to)
if not data:
continue
save_json_for_account(JSON_BASE_DIR, acc, data, d_from, d_to)
tlist = data["accountStatement"]["transactionList"].get("transaction", [])
if isinstance(tlist, dict):
tlist = [tlist]
fio_acc_id = data["accountStatement"]["info"]["accountId"]
rows = []
for t in tlist:
row = {
"id_operace": safe_col(t, 22),
"cislo_uctu": fio_acc_id,
"transaction_date": clean_date(safe_col(t, 0)),
"amount": safe_col(t, 1),
"currency": safe_col(t, 14),
"protiucet": safe_col(t, 2),
"kod_banky": safe_col(t, 3),
"nazev_protiuctu": safe_col(t, 10),
"nazev_banky": safe_col(t, 15),
"typ": safe_col(t, 8),
"vs": safe_col(t, 5),
"ks": safe_col(t, 4),
"ss": safe_col(t, 6),
"zprava": safe_col(t, 16),
"uziv_identifikace": safe_col(t, 12),
"provedl": safe_col(t, 12),
"id_pokynu": safe_col(t, 19),
"komentar": safe_col(t, 25),
"upr_objem_mena": safe_col(t, 20),
"api_bic": safe_col(t, 26),
"reference_platce": safe_col(t, 27),
}
if row["id_operace"]:
rows.append(row)
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)
print(f"✓ Insert/Update: {inserted}")
total_inserted += inserted
cur.close()
conn.close()
print(f"\n=== DONE → {total_inserted} rows processed ===")
if __name__ == "__main__":
main()
2025-11-30 final reporter/30Report.py
+1 -1
View File
@@ -41,7 +41,7 @@ import json
# MySQL server parameters
DB_HOST = "192.168.1.76"
DB_PORT = 3306
DB_PORT = 3307
DB_USER = "root"
DB_PASS = "Vlado9674+"
DB_NAME = "fio"
2025-11-30 final reporter/31Reportnový.py
-249
View File
@@ -1,249 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import sys
import io
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8', errors='replace')
import mysql.connector
from mysql.connector import Error
from openpyxl import Workbook
from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
from datetime import datetime
import os
import glob
import json
# ======================================================
# CONFIGURATION
# ======================================================
DB_HOST = "192.168.1.76"
DB_PORT = 3306
DB_USER = "root"
DB_PASS = "Vlado9674+"
DB_NAME = "fio"
OUTPUT_DIR = r"u:\Dropbox\!!!Days\Downloads Z230"
ACCOUNTS_JSON = r"accounts.json"
TEXT_COLUMNS = ["cislo_uctu", "protiucet", "kod_banky", "ks", "vs", "ss"]
# ======================================================
# SQL VIEW COMPATIBILITY LAYER
# ======================================================
SELECT_LEGACY = """
SELECT
transaction_date AS datum,
amount AS objem,
currency AS mena,
cislo_uctu,
protiucet,
kod_banky,
ks,
vs,
ss,
zprava_pro_prijemce,
komentar AS poznamka,
id_operace,
id_pokynu,
nazev_banky,
nazev_protiuctu,
typ,
upr_objem_mena AS upresneni_objem,
NULL AS upresneni_mena,
provedl AS zadal,
api_bic,
reference_platce,
stazeno_kdy
FROM transactions
"""
# ======================================================
# REMOVE OLD EXPORT FILES
# ======================================================
def cleanup_old_exports():
patterns = [
os.path.join(OUTPUT_DIR, "*FIO*transaction*.xlsx"),
os.path.join(OUTPUT_DIR, "*FIO*transactions*.xlsx"),
os.path.join(OUTPUT_DIR, "*FIO_transactions*.xlsx"),
]
for pattern in patterns:
for file in glob.glob(pattern):
try:
os.remove(file)
print(f"🗑 Deleted old export: {file}")
except:
pass
# ======================================================
# FORMAT SHEET
# ======================================================
def format_sheet(ws, rows, headers):
for col_idx in range(1, len(headers) + 1):
cell = ws.cell(row=1, column=col_idx)
cell.font = Font(bold=True)
cell.fill = PatternFill(start_color="FFFF00", fill_type="solid")
for row in rows:
excel_row = []
for h in headers:
val = row[h]
if h in TEXT_COLUMNS and val is not None:
excel_row.append(f'="{val}"')
else:
excel_row.append(val)
ws.append(excel_row)
fill_red = PatternFill(start_color="FFFFDDDD", fill_type="solid")
fill_green = PatternFill(start_color="FFEEFFEE", fill_type="solid")
objem_col_index = headers.index("objem") + 1
for row_idx in range(2, len(rows) + 2):
try:
value = float(ws.cell(row=row_idx, column=objem_col_index).value)
except:
value = 0
fill = fill_red if value < 0 else fill_green
for col_idx in range(1, len(headers) + 1):
ws.cell(row=row_idx, column=col_idx).fill = fill
fixed_widths = [
10.29, 10.29, 10.71, 13.29, 13.29, 7.29, 15.14, 12.29,
34.43, 49.29, 13.57, 11.29, 11.29, 8, 30.57, 28.29,
11, 28.29, 15.29, 14.29, 20
]
for i, width in enumerate(fixed_widths, start=1):
ws.column_dimensions[chr(64 + i)].width = width
thin = Side(border_style="thin", color="000000")
border = Border(left=thin, right=thin, top=thin, bottom=thin)
align_center = Alignment(horizontal="center")
total_rows = len(rows) + 1
total_cols = len(headers)
for row_idx in range(1, total_rows + 1):
for col_idx in range(1, total_cols + 1):
cell = ws.cell(row=row_idx, column=col_idx)
cell.border = border
align_left = Alignment(horizontal="left")
if row_idx > 1 and col_idx == 10:
cell.alignment = align_left
elif col_idx <= 9:
cell.alignment = align_center
ws.freeze_panes = "A2"
ws.auto_filter.ref = ws.dimensions
# ======================================================
# EXPORT
# ======================================================
def export_fio():
print("Connecting to MySQL...")
try:
conn = mysql.connector.connect(
host=DB_HOST,
port=DB_PORT,
user=DB_USER,
password=DB_PASS,
database=DB_NAME
)
except Error as e:
print("❌ Failed to connect:", e)
return
cur = conn.cursor(dictionary=True)
with open(ACCOUNTS_JSON, "r", encoding="utf-8") as f:
accounts = json.load(f)
preferred_order = [
"CZK rodina",
"CZK ordinace",
"CZK na jídlo",
"CZK TrialHelp",
"CZK maminka svojě věci"
]
accounts_sorted = []
for pref in preferred_order:
for acc in accounts:
if acc["name"] == pref:
accounts_sorted.append(acc)
for acc in accounts:
if acc not in accounts_sorted:
accounts_sorted.append(acc)
wb = Workbook()
wb.remove(wb.active)
# -------- ALL sheet --------
cur.execute(SELECT_LEGACY + " ORDER BY datum DESC")
all_rows = cur.fetchall()
if all_rows:
headers = list(all_rows[0].keys())
ws_all = wb.create_sheet(title="ALL")
ws_all.append(headers)
format_sheet(ws_all, all_rows, headers)
# -------- per account sheets --------
for acc in accounts_sorted:
acc_num = acc["account_number"]
sheet_name = acc["name"][:31]
print(f"➡ Creating sheet: {sheet_name}")
query = SELECT_LEGACY + f"""
WHERE cislo_uctu = '{acc_num}'
ORDER BY datum DESC
"""
cur.execute(query)
rows = cur.fetchall()
if not rows:
continue
headers = list(rows[0].keys())
ws = wb.create_sheet(title=sheet_name)
ws.append(headers)
format_sheet(ws, rows, headers)
conn.close()
cleanup_old_exports()
timestamp = datetime.now().strftime("%Y-%m-%d %H-%M-%S")
output_file = os.path.join(OUTPUT_DIR, f"{timestamp} FIO transactions.xlsx")
wb.save(output_file)
print(f"✅ Export complete:\n{output_file}")
# ======================================================
if __name__ == "__main__":
export_fio()
2026-01-03 TestScriptů/10 Read.py
+400
View File
@@ -0,0 +1,400 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import sys
import io
import time
from datetime import date, timedelta
from pathlib import Path
import json
import requests
import mysql.connector
from mysql.connector import Error
from typing import Dict, Any, List
import hashlib
# ====================================================================
# A. Vynucení UTF-8 pro správnou diakritiku v plánovaných úlohách
# ====================================================================
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8', errors='replace')
"""
FIO MULTI–ACCOUNT IMPORTER — VERZE S ROBUSTNĚJŠÍM HANDLINGEM PK
===============================================================
- mysql.connector (Oracle) pro stabilní manipulaci s datovými typy
- Bezpečné generování id_operace, pokud chybí Column22
- Správné mapování id_pokynu = Column19
- Detailní logování chybných řádků
"""
# =========================================
# CONFIGURATION
# =========================================
ACCOUNTS_FILE = r"accounts.json"
JSON_BASE_DIR = r"z:\Dropbox\!!!Days\Downloads Z230\Fio"
DB = {
"host": "192.168.1.76",
"port": 3307,
"user": "root",
"password": "Vlado9674+",
"database": "fio",
"charset": "utf8mb4",
}
BATCH_SIZE = 500
DAYS_BACK = 90
# Zapíná detailnější logování při chybách insertu
DEBUG_ON_ERROR = True
# =========================================
# HELPERS
# =========================================
def load_accounts(path: str) -> List[Dict[str, str]]:
"""Reads accounts.json and validates content."""
with open(path, "r", encoding="utf-8") as f:
accounts = json.load(f)
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:
"""Constructs the URL for Fio REST API periods endpoint."""
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) -> Any:
"""Calls Fio API and fetches JSON."""
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}", flush=True)
return None
try:
return resp.json()
except json.JSONDecodeError:
print(" ❌ Cannot decode JSON from Fio response", flush=True)
return None
def safe_col(t: dict, n: int) -> Any:
"""SAFE ACCESSOR for Fio transaction column numbers (ColumnN)."""
key = f"column{n}"
val = t.get(key)
if not val:
return None
return val.get("value")
def clean_date(dt_str: str) -> str:
"""Strips timezone from Fio date string ("YYYY-MM-DD+HH:MM") → "YYYY-MM-DD"."""
if not dt_str:
return None
return str(dt_str)[:10]
def ensure_dir(path: Path):
"""Creates directory if it doesn’t exist."""
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) -> Path:
"""Saves raw JSON to disk."""
acc_num_raw = account_cfg["account_number"]
acc_folder_name = acc_num_raw.replace("/", "_")
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
def generate_fallback_id(fio_acc_id: str, t: dict) -> str:
"""
Vygeneruje deterministický fallback ID, pokud chybí Column22 (id pohybu).
Použije SHA1 hash z několika sloupců a ořízne na 20 znaků, aby se vešlo
do VARCHAR(20) primárního klíče.
"""
raw_date = clean_date(safe_col(t, 0)) or ""
amount = str(safe_col(t, 1) or "")
protiucet = str(safe_col(t, 2) or "")
vs = str(safe_col(t, 5) or "")
source = f"{fio_acc_id}|{raw_date}|{amount}|{protiucet}|{vs}"
digest = hashlib.sha1(source.encode("utf-8")).hexdigest()
return digest[:20]
# =========================================
# MAIN IMPORT LOGIC
# =========================================
def main():
start_all = time.time()
today = date.today()
d_from = today - timedelta(days=DAYS_BACK)
d_to = today
print("=== Fio multi-account import v3 (PK fix, lepší logování) ===", flush=True)
print(f"Období: {d_from} až {d_to}", flush=True)
# Load all accounts from accounts.json
try:
accounts = load_accounts(ACCOUNTS_FILE)
except Exception as e:
print(f"FATÁLNÍ CHYBA při načítání účtů: {e}", flush=True)
return
print(f" Účtů v konfiguraci: {len(accounts)}\n", flush=True)
# Připojení k DB
try:
conn = mysql.connector.connect(
host=DB["host"],
port=DB["port"],
user=DB["user"],
password=DB["password"],
database=DB["database"],
charset=DB["charset"],
)
cur = conn.cursor()
except Error as e:
print(f"FATÁLNÍ CHYBA při připojení k DB: {e}", flush=True)
return
# SQL INSERT dotaz přizpůsobený nové DB struktuře
sql = """
INSERT INTO transactions
(
id_operace, cislo_uctu, transaction_date, amount, currency,
protiucet, kod_banky, nazev_protiuctu, nazev_banky, typ,
vs, ks, ss, uziv_identifikace, zprava_pro_prijemce,
provedl, id_pokynu, komentar, upr_objem_mena, api_bic, reference_platce
)
VALUES
(
%(id_operace)s, %(cislo_uctu)s, %(transaction_date)s, %(amount)s, %(currency)s,
%(protiucet)s, %(kod_banky)s, %(nazev_protiuctu)s, %(nazev_banky)s, %(typ)s,
%(vs)s, %(ks)s, %(ss)s, %(uziv_identifikace)s, %(zprava_pro_prijemce)s,
%(provedl)s, %(id_pokynu)s, %(komentar)s, %(upr_objem_mena)s, %(api_bic)s, %(reference_platce)s
)
ON DUPLICATE KEY UPDATE
cislo_uctu = VALUES(cislo_uctu),
transaction_date = VALUES(transaction_date),
amount = VALUES(amount),
currency = VALUES(currency),
protiucet = VALUES(protiucet),
kod_banky = VALUES(kod_banky),
nazev_protiuctu = VALUES(nazev_protiuctu),
nazev_banky = VALUES(nazev_banky),
typ = VALUES(typ),
vs = VALUES(vs),
ks = VALUES(ks),
ss = VALUES(ss),
uziv_identifikace = VALUES(uziv_identifikace),
zprava_pro_prijemce = VALUES(zprava_pro_prijemce),
provedl = VALUES(provedl),
id_pokynu = VALUES(id_pokynu),
komentar = VALUES(komentar),
upr_objem_mena = VALUES(upr_objem_mena),
api_bic = VALUES(api_bic),
reference_platce = VALUES(reference_platce)
"""
total_inserted = 0
total_skipped_pk = 0
total_skipped_error = 0
# ======================================================
# PROCESS EACH ACCOUNT IN accounts.json
# ======================================================
for acc in accounts:
name = acc["name"]
cfg_acc_num = acc["account_number"]
token = acc["token"]
print(f"--- Účet: {name} ({cfg_acc_num}) ---", flush=True)
t0 = time.time()
# 1) Download JSON from Fio API
data = fetch_fio_json(token, d_from, d_to)
if data is None:
print(" Přeskakuji, žádná data / chyba API.\n", flush=True)
continue
# 2) Save raw JSON file to disk
try:
json_path = save_json_for_account(JSON_BASE_DIR, acc, data, d_from, d_to)
print(f" JSON uložen do: {json_path}", flush=True)
except Exception as e:
print(f" ❌ Chyba při ukládání JSON souboru: {e}", flush=True)
# 3) Extract transactions from JSON tree
tlist = data.get("accountStatement", {}).get("transactionList", {}).get("transaction", [])
if isinstance(tlist, dict):
tlist = [tlist]
print(f" Počet transakcí v období: {len(tlist)}", flush=True)
if not tlist:
print(" Žádné transakce, jdu dál.\n", flush=True)
continue
fio_acc_id = data.get("accountStatement", {}).get("info", {}).get("accountId")
if cfg_acc_num and fio_acc_id and cfg_acc_num.split("/")[0] not in fio_acc_id:
print(
f" ⚠ Upozornění: accountId z Fio ({fio_acc_id}) "
f"se neshoduje s account_number v konfiguraci ({cfg_acc_num})",
flush=True,
)
# 4) Build list of MySQL rows
rows = []
skipped_pk_account = 0
for t in tlist:
# id_operace = Column22 (tvé "id pohybu")
id_operace_val = safe_col(t, 22)
# Pokud chybí, vygenerujeme si stabilní fallback (hash) – vejde se do VARCHAR(20)
if id_operace_val is None:
fallback = generate_fallback_id(fio_acc_id or "", t)
id_operace_val = fallback
# Můžeš odkomentovat, pokud chceš vidět, kde se používá fallback
# print(f" ⚠ Fallback id_operace (hash) pro transakci: {fallback}", flush=True)
# Bez PK nemá smysl zápis – jen pro jistotu, fallback by měl vše pokrýt
if id_operace_val is None:
skipped_pk_account += 1
continue
transaction_date = clean_date(safe_col(t, 0))
if not transaction_date:
# Bez data by insert stejně spadl (NOT NULL), tak to raději přeskočíme
if DEBUG_ON_ERROR:
print(f" ⚠ Přeskakuji transakci bez data, id_operace={id_operace_val}", flush=True)
skipped_pk_account += 1
continue
id_pokynu_val = safe_col(t, 19) # tvé "id pokynu" = Column19
row = {
"id_operace": str(id_operace_val),
"cislo_uctu": fio_acc_id,
"transaction_date": transaction_date,
"amount": safe_col(t, 1),
"currency": safe_col(t, 14),
"typ": safe_col(t, 8),
"provedl": safe_col(t, 9),
"protiucet": safe_col(t, 2),
"kod_banky": safe_col(t, 3),
"nazev_protiuctu": safe_col(t, 10),
"nazev_banky": safe_col(t, 12),
"api_bic": safe_col(t, 26),
"vs": safe_col(t, 5),
"ks": safe_col(t, 4),
"ss": safe_col(t, 6),
"zprava_pro_prijemce": safe_col(t, 16),
"uziv_identifikace": safe_col(t, 7),
"komentar": safe_col(t, 25),
"upr_objem_mena": safe_col(t, 18),
"id_pokynu": str(id_pokynu_val) if id_pokynu_val is not None else None,
"reference_platce": safe_col(t, 27),
}
rows.append(row)
if skipped_pk_account:
print(f" ⚠ Přeskočeno {skipped_pk_account} transakcí kvůli chybějícímu/invalidnímu PK nebo datu.", flush=True)
total_skipped_pk += skipped_pk_account
# 5) INSERT rows into MySQL in batches
inserted = 0
skipped_error_account = 0
for i in range(0, len(rows), BATCH_SIZE):
chunk = rows[i: i + BATCH_SIZE]
try:
cur.executemany(sql, chunk)
conn.commit()
inserted += len(chunk)
except Error as e:
print(f" ❌ Chyba při zápisu batch do DB: {e}", flush=True)
conn.rollback()
if DEBUG_ON_ERROR:
print(" ► Přecházím na per-row insert pro detail chyb...", flush=True)
for row in chunk:
try:
cur.execute(sql, row)
conn.commit()
inserted += 1
except Error as e_row:
skipped_error_account += 1
conn.rollback()
print(
f" ✗ Chybná transakce id_operace={row.get('id_operace')} "
f"datum={row.get('transaction_date')} částka={row.get('amount')} "
f"→ {e_row}",
flush=True,
)
elapsed = time.time() - t0
total_inserted += inserted
total_skipped_error += skipped_error_account
print(
f" ✓ Zapsáno (insert/update): {inserted} řádků do DB "
f"(přeskočeno chybějící PK/dat {skipped_pk_account}, chybou insertu {skipped_error_account}) "
f"za {elapsed:.2f} s\n",
flush=True,
)
# Close DB
cur.close()
conn.close()
total_elapsed = time.time() - start_all
print(
f"=== Hotovo. Celkem zapsáno {total_inserted} transakcí. "
f"Přeskočeno kvůli PK/datům: {total_skipped_pk}, kvůli chybě insertu: {total_skipped_error}. "
f"Celkový čas: {total_elapsed:.2f} s ===",
flush=True,
)
# ======================================================
# ENTRY POINT
# ======================================================
if __name__ == "__main__":
main()
2026-01-03 TestScriptů/accounts.json
+53
View File
@@ -0,0 +1,53 @@
[
{
"name": "EUR tatínek 1",
"account_number": "2100074583",
"token": "GuV2Boaulx56ZiQUqUArgg6P9qdfEVKOoH6wF3PfAZ0fPS01r2WbiNiCsCcIBZ0U"
},
{
"name": "CZK rodina",
"account_number": "2100046291",
"token": "v0GJaAVeefzV1lnx1jPCf2nFF7SuOPzzrL5tobPNsC7oCChXG4hahDYVb8Rdcex0"
},
{
"name": "EUR TrialHelp",
"account_number": "2200787265",
"token": "9yG5g6lHWGS6YU2R2petm5DRYTb9orhJ8VPJ0p7RtTjlIo2vB83ynBlPCMGRIwzy"
},
{
"name": "CZK tatínek",
"account_number": "2400046293",
"token": "j2qmpvWe4RfKtBTBlhwC1VFED7HJlVAe23iPBH1TWis9htEyYe8fRejcMeSxOLqC"
},
{
"name": "CHF tatínek",
"account_number": "2402161017",
"token": "aNfK9iu6qIPlugGCR6gvSJ7NXtTkDfVVj8fBz4X1pORuGKf6VXjWin4wrr9WRjSd"
},
{
"name": "EUR tatínek 2",
"account_number": "2500074582",
"token": "aLsl9ETRUU1IgoYeinAzYWyruIoJvs6UvJKTGRlJcm7HaEc5ojsFdxJizyT9lREO"
},
{
"name": "CZK TrialHelp",
"account_number": "2900046548",
"token": "pKZVHbFDVsbTa8ryEaVc6A2nyrlb4TbT1tCiimieesHvhKFoJmYBRVjCpnvjiUUK"
},
{
"name": "CZK maminka svojě věci",
"account_number": "2003310572",
"token": "TkrRvnMK77OSSYdVulNvZcT6ltWcmjqkp3RN5WYwnBpNTuaKCWO1zHKOlDGAiNyv"
},
{
"name": "CZK na jídlo",
"account_number": "2403310563",
"token": "axRvFxu4VCzsDp5QZXN8LQ0fQUqzV2FEBZrM595x3Rtp10zowRBcGOFs9uNNPb7Q"
},
{
"name": "CZK ordinace",
"account_number": "2800046620",
"token": "Xzdr3eK7se7ZgeE3JujgeidGb0WrB7mGQ6HSOiBJzWi0kPURYKRpkRKB3ZOpt3rq"
}
]
40 Fio 02 diagnostika 2.py
-24
View File
@@ -1,24 +0,0 @@
import pymysql
from pymysql.cursors import DictCursor
conn = pymysql.connect(
host="192.168.1.76",
port=3307,
user="root",
password="Vlado9674+",
database="fio",
charset="utf8mb4",
cursorclass=DictCursor
)
with conn.cursor() as cur:
cur.execute("SHOW TABLES;")
print("📋 Tables:", [r[f"Tables_in_fio"] for r in cur.fetchall()])
cur.execute("SELECT COUNT(*) AS cnt FROM transactions;")
print("🧾 Rows in `transactions`:", cur.fetchone()["cnt"])
cur.execute("SHOW COLUMNS FROM transactions;")
print("\n📊 Columns:")
for r in cur.fetchall():
print(" -", r["Field"])
40 Fio 03 excel.py
-117
View File
@@ -1,117 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Export Fio transactions (from MySQL → Excel)
--------------------------------------------
- Reads only cislo_uctu = '2800046620'
- For OZP (protiucet=2070101041) includes only positive objem
- Each sheet = insurance company (filtered by protiucet)
- First sheet = summary with total amounts and transaction counts
"""
import pandas as pd
import pymysql
from pathlib import Path
from datetime import datetime
# ======== CONFIG ========
MYSQL_CONFIG = {
"host": "192.168.1.76",
"port": 3307,
"user": "root",
"password": "Vlado9674+",
"database": "fio",
"charset": "utf8mb4",
}
REPORTOVAT = {
"VZP": "1114007221",
"VOZP": "2010009091",
"ČPZP": "2054108761",
"OZP": "2070101041",
"ZPŠ": "2090309181",
"ZPMV": "2112108031",
}
EXPORT_PATH = Path(r"u:\Dropbox\!!!Days\Downloads Z230") / f"Fio_report_{datetime.now():%Y-%m-%d_%H-%M-%S}.xlsx"
# ======== LOAD DATA ========
def load_data():
print("🔄 Načítám data z MySQL (účet 2800046620, pro OZP jen kladné objemy)...")
conn = pymysql.connect(**MYSQL_CONFIG)
sql = """
SELECT *
FROM transactions
WHERE cislo_uctu = '2800046620'
AND (
protiucet <> '2070101041'
OR (protiucet = '2070101041' AND objem > 0)
);
"""
df = pd.read_sql(sql, conn)
conn.close()
df.columns = df.columns.str.strip()
print(f"✅ Načteno {len(df)} řádků, {len(df.columns)} sloupců.")
return df
# ======== EXPORT TO EXCEL ========
def export_to_excel(df):
summary_rows = [] # to collect summary per insurer
with pd.ExcelWriter(EXPORT_PATH, engine="openpyxl") as writer:
# --- INDIVIDUAL SHEETS ---
for name, acc in REPORTOVAT.items():
filtered = df[df["protiucet"].astype(str) == acc]
if filtered.empty:
print(f"⚠️ {name}: žádné transakce (účet {acc})")
summary_rows.append({
"Pojišťovna": name,
"Číslo účtu": acc,
"Počet transakcí": 0,
"Součet objemu": 0.0
})
continue
# safe numeric conversion
filtered = filtered.copy()
filtered["objem_num"] = (
filtered["objem"]
.astype(str)
.str.replace("\u00A0", "", regex=False)
.str.replace(",", ".", regex=False)
.astype(float)
)
# --- summary data ---
total_sum = filtered["objem_num"].sum()
total_count = len(filtered)
summary_rows.append({
"Pojišťovna": name,
"Číslo účtu": acc,
"Počet transakcí": total_count,
"Součet objemu": round(total_sum, 2)
})
# --- write detailed sheet ---
filtered.to_excel(writer, index=False, sheet_name=name)
print(f"✅ {name}: {len(filtered)} řádků exportováno, součet {total_sum:,.2f} Kč")
# --- SUMMARY SHEET ---
summary_df = pd.DataFrame(summary_rows)
summary_df["Součet objemu"] = summary_df["Součet objemu"].map("{:,.2f} Kč".format)
summary_df.to_excel(writer, index=False, sheet_name="Přehled")
print("🧾 Přidán přehledový list s celkovými součty.")
print(f"\n📊 Hotovo! Soubor uložen jako:\n{EXPORT_PATH}")
# ======== MAIN ========
if __name__ == "__main__":
df = load_data()
export_to_excel(df)
40 fio 01.py
-168
View File
@@ -1,168 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Fio CSV import → MySQL (dev version)
------------------------------------
- Always drops & recreates `transactions` table
- Uses real CSV headers as seen in "Vyhledane pohyby (3).csv"
- Unique key = (Číslo účtu, ID operace, ID pokynu)
"""
import csv
from pathlib import Path
from datetime import datetime
import pymysql
from pymysql.cursors import DictCursor
import re
# ======== CONFIG ========
CSV_PATH = Path(r"u:\Dropbox\!!!Days\Downloads Z230\Vyhledane pohyby (3).csv")
TABLE_NAME = "transactions"
MYSQL_CONFIG = {
"host": "192.168.1.76",
"port": 3307,
"user": "root",
"password": "Vlado9674+",
"database": "fio",
"charset": "utf8mb4",
"cursorclass": DictCursor,
"autocommit": True,
}
# ======== HELPERS ========
def clean(s: str):
if not s:
return None
return s.strip() or None
def parse_date(raw: str):
raw = (raw or "").strip()
if not raw:
return None
try:
return datetime.strptime(raw, "%d.%m.%Y").date()
except ValueError:
return None
def parse_float(raw: str):
if raw is None:
return None
s = str(raw).strip()
for ch in (" ", "\u00A0", "\u202F", "\u2007"):
s = s.replace(ch, "")
s = s.replace(",", ".")
s = re.sub(r"[^0-9.+-]", "", s)
try:
return float(s)
except ValueError:
return None
# ======== DB ========
def get_mysql_connection():
return pymysql.connect(**MYSQL_CONFIG)
def recreate_table(conn):
"""Drop and recreate table with schema matching CSV structure."""
sql = f"""
DROP TABLE IF EXISTS `{TABLE_NAME}`;
CREATE TABLE `{TABLE_NAME}` (
id INT AUTO_INCREMENT PRIMARY KEY,
datum DATE,
objem DECIMAL(14,2),
mena CHAR(3),
cislo_uctu VARCHAR(40),
protiucet VARCHAR(40),
kod_banky VARCHAR(20),
ks VARCHAR(20),
vs VARCHAR(20),
ss VARCHAR(20),
zprava_pro_prijemce VARCHAR(500),
poznamka VARCHAR(500),
id_operace VARCHAR(50),
id_pokynu VARCHAR(50),
ks_1 VARCHAR(20),
nazev_banky VARCHAR(100),
nazev_protiuctu VARCHAR(200),
ss_1 VARCHAR(20),
typ VARCHAR(100),
upresneni_objem VARCHAR(100),
upresneni_mena VARCHAR(20),
vs_1 VARCHAR(20),
zadal VARCHAR(200),
imported_at DATETIME DEFAULT CURRENT_TIMESTAMP,
UNIQUE KEY uniq_tx (cislo_uctu, id_operace, id_pokynu)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
"""
with conn.cursor() as cur:
for stmt in sql.strip().split(";"):
if stmt.strip():
cur.execute(stmt)
print(f"✅ Tabulka `{TABLE_NAME}` znovu vytvořena podle CSV struktury.")
# ======== IMPORT ========
def import_fio_csv():
with open(CSV_PATH, "r", encoding="utf-8-sig", newline="") as f:
reader = csv.DictReader(f, delimiter=";", quotechar='"')
rows = list(reader)
total = len(rows)
print(f"📄 Načteno {total} řádků ze souboru {CSV_PATH.name}")
with get_mysql_connection() as conn:
recreate_table(conn)
inserted, skipped = 0, 0
for i, row in enumerate(rows, start=1):
data = {
"datum": parse_date(row.get("Datum")),
"objem": parse_float(row.get("Objem")),
"mena": clean(row.get("Měna")),
"cislo_uctu": clean(row.get("Číslo účtu")),
"protiucet": clean(row.get("Protiúčet")),
"kod_banky": clean(row.get("Kód banky")),
"ks": clean(row.get("KS")),
"vs": clean(row.get("VS")),
"ss": clean(row.get("SS")),
"zprava_pro_prijemce": clean(row.get("Zpráva pro příjemce")),
"poznamka": clean(row.get("Poznámka")),
"id_operace": clean(row.get("ID operace")),
"id_pokynu": clean(row.get("ID pokynu")),
"ks_1": clean(row.get("KS.1")),
"nazev_banky": clean(row.get("Název banky")),
"nazev_protiuctu": clean(row.get("Název protiúčtu")),
"ss_1": clean(row.get("SS.1")),
"typ": clean(row.get("Typ")),
"upresneni_objem": clean(row.get("Upřesnění - objem")),
"upresneni_mena": clean(row.get("Upřesnění - měna")),
"vs_1": clean(row.get("VS.1")),
"zadal": clean(row.get("Zadal")),
}
cols = ", ".join(data.keys())
placeholders = ", ".join(["%s"] * len(data))
sql = f"INSERT IGNORE INTO `{TABLE_NAME}` ({cols}) VALUES ({placeholders})"
with conn.cursor() as cur:
affected = cur.execute(sql, list(data.values()))
if affected:
inserted += 1
else:
skipped += 1
if i % 500 == 0 or i == total:
print(f" {i}/{total} zpracováno... ({inserted} vloženo, {skipped} duplicit)")
print(f"\n✅ Import dokončen: {inserted} nových, {skipped} duplicit přeskočeno.")
# ======== MAIN ========
if __name__ == "__main__":
import_fio_csv()
40 fio 02 diagnostika.py
-47
View File
@@ -1,47 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Quick, verified dump of all Fio transactions from MySQL → Excel.
Column names are exactly as in DB.
"""
import pandas as pd
import pymysql
from pymysql.cursors import DictCursor
from pathlib import Path
from datetime import datetime
# ======== CONFIG ========
MYSQL_CONFIG = {
"host": "192.168.1.76",
"port": 3307,
"user": "root",
"password": "Vlado9674+",
"database": "fio",
"charset": "utf8mb4",
}
EXPORT_PATH = Path(r"u:\Dropbox\!!!Days\Downloads Z230") / f"Fio_ALL_{datetime.now():%Y-%m-%d_%H-%M-%S}.xlsx"
# ======== MAIN ========
def dump_all_transactions():
with pymysql.connect(**MYSQL_CONFIG) as conn:
sql = """
SELECT
*
FROM transactions
ORDER BY datum DESC;
"""
df = pd.read_sql(sql, conn)
print(f"✅ Načteno {len(df)} transakcí z MySQL.")
# Save to Excel
df.to_excel(EXPORT_PATH, index=False)
print(f"📊 Excel export hotov:\n{EXPORT_PATH}")
if __name__ == "__main__":
dump_all_transactions()
QRPlatba/QRPlatbaApp.py
-209
View File
@@ -1,209 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import sys
import urllib.parse
import qrcode
import fdb
from pathlib import Path
from datetime import datetime
from PIL import Image, ImageTk
import customtkinter as ctk
from tkinter import messagebox
# ================================
# ⚙️ Default Configuration
# ================================
ACCOUNTS = {
"2100046291/2010": "CZ1720100000002100046291",
"2800046620/2010": "CZ7520100000002800046620",
}
CURRENCY = "CZK"
# sys.executable = cesta k EXE (i po přesunutí), __file__ by v onefile EXE ukazoval do temp
OUTPUT_DIR = Path(sys.executable).parent / "QRPlatby"
OUTPUT_DIR.mkdir(exist_ok=True)
# Firebird připojení
DB_DSN = r'localhost:c:\medicus 3\data\medicus.fdb'
DB_USER = 'SYSDBA'
DB_PASSWORD = 'masterkey'
DB_CHARSET = 'win1250'
# Default values (can be overridden by arguments)
PRIJMENI = "Buzalka"
JMENO = "Vladimír"
RODCIS = "730928104"
# ================================
# 💬 Argument Handling
# ================================
if len(sys.argv) >= 4:
JMENO = sys.argv[1]
PRIJMENI = sys.argv[2]
RODCIS = sys.argv[3]
elif len(sys.argv) == 2 and sys.argv[1] in ("-h", "--help"):
print("Usage: QRPlatbaApp.py JMENO PRIJMENI RODCIS")
sys.exit(0)
# ================================
# 💉 Items to Pay – načteno z Medicusu
# ================================
def nacti_polozky():
"""Načte ceník z Medicusu seřazený podle KOD (pořadového čísla)."""
try:
conn = fdb.connect(dsn=DB_DSN, user=DB_USER, password=DB_PASSWORD, charset=DB_CHARSET)
cur = conn.cursor()
cur.execute("SELECT V.KOD, V.NAZEV, V.CENA FROM VLV_SEL(NULL, NULL, NULL) V ORDER BY V.KOD")
rows = cur.fetchall()
conn.close()
# Vrátí dict: název -> cena (float), seřazený podle KOD
return {row[1].strip(): float(row[2]) for row in rows if row[2] is not None}
except Exception as e:
messagebox.showerror("Chyba databáze", f"Nepodařilo se načíst ceník z Medicusu:\n{e}")
return {}
ITEMS = nacti_polozky()
# ================================
# 🧩 Helper – sestavení SPAYD řetězce
# ================================
def create_spayd(iban, amount, vs, msg, currency="CZK"):
msg_encoded = urllib.parse.quote(msg, safe="$%*+-.:/")
return f"SPD*1.0*ACC:{iban}*AM:{amount:.2f}*CC:{currency}*X-VS:{vs}*MSG:{msg_encoded}"
# ================================
# 🪟 GUI Class
# ================================
class QRPlatbaApp(ctk.CTk):
def __init__(self):
super().__init__()
self.title("QR Platba – Ordinace MUDr. Buzalková")
self.geometry("520x680")
self.minsize(480, 480)
self.resizable(True, True)
ctk.set_appearance_mode("light")
ctk.set_default_color_theme("blue")
frame = ctk.CTkFrame(self, corner_radius=10)
frame.pack(expand=True, fill="both", padx=20, pady=20)
ctk.CTkLabel(frame, text="Generátor QR Platby",
font=("Arial", 20, "bold")).pack(pady=(8, 10))
# 👤 Informace o pacientovi
patient = ctk.CTkFrame(frame, corner_radius=8)
patient.pack(fill="x", pady=(0, 8), padx=10)
for text in [f"Příjmení: {PRIJMENI}",
f"Jméno: {JMENO}",
f"Rodné číslo: {RODCIS}"]:
ctk.CTkLabel(patient, text=text, font=("Arial", 12)).pack(anchor="w", padx=10, pady=1)
# 💰 Výběr položky
pay = ctk.CTkFrame(frame, corner_radius=8)
pay.pack(fill="x", pady=(0, 8), padx=10)
ctk.CTkLabel(pay, text="Vyberte položku k úhradě:",
font=("Arial", 12, "bold")).pack(anchor="w", padx=10, pady=(6, 3))
self.display_items = [f"{name} ({price:.0f} Kč)" for name, price in ITEMS.items()]
self.item_map = {f"{name} ({price:.0f} Kč)": name for name, price in ITEMS.items()}
self.selected_item = ctk.StringVar(value=self.display_items[0])
self.combo = ctk.CTkOptionMenu(
pay,
variable=self.selected_item,
values=self.display_items,
font=("Arial", 12),
command=self.on_change
)
self.combo.pack(fill="x", padx=10)
self.amount_label = ctk.CTkLabel(pay, text="", font=("Arial", 12, "italic"))
self.amount_label.pack(anchor="e", padx=10, pady=(3, 6))
# 🏦 Výběr účtu
acc = ctk.CTkFrame(frame, corner_radius=8)
acc.pack(fill="x", pady=(0, 8), padx=10)
ctk.CTkLabel(acc, text="Číslo účtu:", font=("Arial", 12, "bold")).pack(anchor="w", padx=10, pady=(6, 3))
self.selected_account = ctk.StringVar(value=list(ACCOUNTS.keys())[0])
ctk.CTkOptionMenu(
acc,
variable=self.selected_account,
values=list(ACCOUNTS.keys()),
font=("Arial", 12),
command=self.on_change
).pack(fill="x", padx=10, pady=(0, 6))
ctk.CTkButton(frame, text="Uložit QR kód",
font=("Arial", 13, "bold"),
height=40,
command=self.ulozit_qr).pack(pady=6)
self.qr_label = ctk.CTkLabel(frame, text="")
self.qr_label.pack(pady=6)
ctk.CTkLabel(frame,
text="© Ordinace MUDr. Buzalková | QR Platba dle ČBA v1.2",
font=("Arial", 10),
text_color="#666").pack(side="bottom", pady=(10, 0))
self.center_window()
# QR se vygeneruje automaticky po načtení okna
self.after(100, self.refresh_qr)
# ================================
# 🪟 Vycentrování okna
# ================================
def center_window(self):
self.update_idletasks()
width = self.winfo_width()
height = self.winfo_height()
screen_width = self.winfo_screenwidth()
screen_height = self.winfo_screenheight()
x = int((screen_width / 2) - (width / 2))
y = int((screen_height / 2) - (height / 2))
self.geometry(f"{width}x{height}+{x}+{y}")
# ================================
# 💸 QR logika
# ================================
def _get_current(self):
"""Vrátí (item, iban, spayd) pro aktuální výběr."""
display_item = self.selected_item.get()
item = self.item_map[display_item]
iban = ACCOUNTS[self.selected_account.get()]
spayd = create_spayd(iban, ITEMS[item], RODCIS, f"{PRIJMENI} {JMENO} – {item}", CURRENCY)
return item, iban, spayd
def on_change(self, _=None):
"""Při změně dropdownu – obnoví částku i QR kód."""
self.refresh_qr()
def refresh_qr(self):
"""Zobrazí QR kód pro aktuální výběr (bez uložení na disk)."""
item, _, spayd = self._get_current()
self.amount_label.configure(text=f"Částka: {ITEMS[item]:.2f} Kč")
img = qrcode.make(spayd)
img_resized = img.resize((200, 200), Image.LANCZOS)
qr_tk = ImageTk.PhotoImage(img_resized)
self.qr_label.configure(image=qr_tk)
self.qr_label.image = qr_tk
def ulozit_qr(self):
"""Uloží QR kód do souboru PNG a zobrazí potvrzení."""
item, _, spayd = self._get_current()
img = qrcode.make(spayd)
filename = f"{PRIJMENI}_{JMENO}_{datetime.now():%Y%m%d_%H%M%S}.png"
out_path = OUTPUT_DIR / filename
img.save(out_path)
messagebox.showinfo("Uloženo", f"QR kód uložen:\n{out_path}")
# ================================
# 🚀 Main
# ================================
if __name__ == "__main__":
app = QRPlatbaApp()
app.mainloop()
QRPlatba/README.md
-116
View File
@@ -1,116 +0,0 @@
# QRPlatbaApp – Generátor QR plateb pro ordinaci MUDr. Buzalková
Aplikace pro generování QR kódů plateb dle standardu **SPAYD / ČBA** přímo z prostředí Medicus 3.
Ceník se načítá živě z databáze Medicusu, takže jakákoliv změna cen v Medicusu se okamžitě projeví.
---
## Funkce
- Načítá ceník přímo z databáze Medicus (Firebird) – vždy aktuální ceny
- Pacientské údaje (jméno, příjmení, rodné číslo) přijímá jako argumenty z tlačítka Medicusu
- QR kód se zobrazí automaticky po otevření okna
- QR kód se obnoví při každé změně položky nebo čísla účtu
- Výběr ze dvou bankovních účtů (dropdown)
- Tlačítko **Uložit QR kód** uloží PNG soubor do složky `QRPlatby/`
- Variabilní symbol = rodné číslo pacienta
- Zpráva příjemci = `Příjmení Jméno – Název položky`
---
## Spuštění z Medicusu
Příkaz pro tlačítko v Medicusu:
```
"U:\PycharmProjects\FIO\.venv\Scripts\pythonw.exe" "U:\PycharmProjects\FIO\QRPlatba\QRPlatbaApp.py" "%JMENO%" "%PRIJMENI%" "%RODCISN%"
```
> `pythonw.exe` se používá místo `python.exe`, aby se nezobrazovalo černé konzolové okno.
---
## Struktura projektu
```
QRPlatba/
├── QRPlatbaApp.py # Hlavní aplikace
├── README.md # Tato dokumentace
└── QRPlatby/ # Výstupní složka pro uložené QR kódy (PNG)
```
---
## Závislosti
Nainstalovat přes pip do projektového venv:
```bash
pip install customtkinter pillow qrcode fdb
```
| Balíček | Účel |
|-----------------|-------------------------------------------|
| `customtkinter` | Moderní GUI (tmavý/světlý režim) |
| `pillow` | Práce s obrázky (zobrazení QR v okně) |
| `qrcode` | Generování QR kódů |
| `fdb` | Připojení k Firebird databázi (Medicus) |
---
## Databázové připojení
Aplikace se připojuje k lokální Firebird databázi Medicusu:
| Parametr | Hodnota |
|-----------|------------------------------------------|
| DSN | `localhost:c:\medicus 3\data\medicus.fdb`|
| Uživatel | `SYSDBA` |
| Charset | `win1250` |
| View | `VLV_SEL` – ceník výkonů |
Položky jsou seřazeny podle pole `KOD` (pořadové číslo v ceníku Medicusu).
---
## Formát QR kódu (SPAYD)
QR kód odpovídá standardu **Short Payment Descriptor (SPAYD)** dle specifikace České bankovní asociace.
Příklad vygenerovaného řetězce:
```
SPD*1.0*ACC:CZ1720100000002100046291*AM:600.00*CC:CZK*X-VS:486122443*MSG:Abohamda%20Horia%20%E2%80%93%20O%C4%8Dkov%C3%A1n%C3%AD%20ch%C5%99ipka%20(VAXIGRIP)
```
| Pole | Obsah |
|---------|------------------------------------|
| `ACC` | IBAN cílového účtu |
| `AM` | Částka v Kč |
| `CC` | Měna (vždy CZK) |
| `X-VS` | Variabilní symbol = rodné číslo |
| `MSG` | Příjmení Jméno – Název položky |
---
## Bankovní účty
| Číslo účtu | IBAN |
|------------------|----------------------------|
| 2100046291/2010 | CZ1720100000002100046291 |
| 2800046620/2010 | CZ7520100000002800046620 |
---
## Uložené QR kódy
Soubory se ukládají do složky `QRPlatby/` ve formátu:
```
Příjmení_Jméno_YYYYMMDD_HHMMSS.png
```
Příklad: `Abohamda_Horia_20260325_143022.png`
---
*© Ordinace MUDr. Buzalková | QR Platba dle ČBA v1.2*
QRPlatbaApp.py
+32 -75
View File
@@ -4,7 +4,6 @@
import sys
import urllib.parse
import qrcode
import fdb
from pathlib import Path
from datetime import datetime
from PIL import Image, ImageTk
@@ -15,20 +14,11 @@ from tkinter import messagebox
# ================================
# ⚙️ Default Configuration
# ================================
ACCOUNTS = {
"2100046291/2010": "CZ1720100000002100046291",
"2800046620/2010": "CZ7520100000002800046620",
}
IBAN = "CZ7520100000002800046620"
CURRENCY = "CZK"
OUTPUT_DIR = Path("QRPlatby")
OUTPUT_DIR.mkdir(exist_ok=True)
# Firebird připojení
DB_DSN = r'localhost:c:\medicus 3\data\medicus.fdb'
DB_USER = 'SYSDBA'
DB_PASSWORD = 'masterkey'
DB_CHARSET = 'win1250'
# Default values (can be overridden by arguments)
PRIJMENI = "Buzalka"
JMENO = "Vladimír"
@@ -46,23 +36,12 @@ elif len(sys.argv) == 2 and sys.argv[1] in ("-h", "--help"):
sys.exit(0)
# ================================
# 💉 Items to Pay – načteno z Medicusu
# 💉 Items to Pay
# ================================
def nacti_polozky():
"""Načte ceník z Medicusu seřazený podle KOD (pořadového čísla)."""
try:
conn = fdb.connect(dsn=DB_DSN, user=DB_USER, password=DB_PASSWORD, charset=DB_CHARSET)
cur = conn.cursor()
cur.execute("SELECT V.KOD, V.NAZEV, V.CENA FROM VLV_SEL(NULL, NULL, NULL) V ORDER BY V.KOD")
rows = cur.fetchall()
conn.close()
# Vrátí OrderedDict: název -> cena (float)
return {row[1].strip(): float(row[2]) for row in rows if row[2] is not None}
except Exception as e:
messagebox.showerror("Chyba databáze", f"Nepodařilo se načíst ceník z Medicusu:\n{e}")
return {}
ITEMS = nacti_polozky()
ITEMS = {
"Očkování chřipka Vaxigrip": 600.00,
"Očkování chřipka Efluelda": 1300.00,
}
# ================================
# 🧩 Helper
@@ -79,7 +58,7 @@ class QRPlatbaApp(ctk.CTk):
def __init__(self):
super().__init__()
self.title("QR Platba – Ordinace MUDr. Buzalková")
self.geometry("520x680")
self.geometry("520x520")
self.minsize(480, 480)
self.resizable(True, True)
@@ -90,21 +69,21 @@ class QRPlatbaApp(ctk.CTk):
frame.pack(expand=True, fill="both", padx=20, pady=20)
ctk.CTkLabel(frame, text="Generátor QR Platby",
font=("Arial", 20, "bold")).pack(pady=(8, 10))
font=("Arial", 20, "bold")).pack(pady=(10, 20))
# 👤 Patient Info
patient = ctk.CTkFrame(frame, corner_radius=8)
patient.pack(fill="x", pady=(0, 8), padx=10)
patient.pack(fill="x", pady=(0, 20), padx=10)
for text in [f"Příjmení: {PRIJMENI}",
f"Jméno: {JMENO}",
f"Rodné číslo: {RODCIS}"]:
ctk.CTkLabel(patient, text=text, font=("Arial", 12)).pack(anchor="w", padx=10, pady=1)
ctk.CTkLabel(patient, text=text, font=("Arial", 12)).pack(anchor="w", padx=10, pady=2)
# 💰 Payment Section
pay = ctk.CTkFrame(frame, corner_radius=8)
pay.pack(fill="x", pady=(0, 8), padx=10)
pay.pack(fill="x", pady=(0, 20), padx=10)
ctk.CTkLabel(pay, text="Vyberte položku k úhradě:",
font=("Arial", 12, "bold")).pack(anchor="w", padx=10, pady=(6, 3))
font=("Arial", 12, "bold")).pack(anchor="w", padx=10, pady=(10, 5))
self.display_items = [f"{name} ({price:.0f} Kč)" for name, price in ITEMS.items()]
self.item_map = {f"{name} ({price:.0f} Kč)": name for name, price in ITEMS.items()}
@@ -115,33 +94,21 @@ class QRPlatbaApp(ctk.CTk):
variable=self.selected_item,
values=self.display_items,
font=("Arial", 12),
command=self.on_change
command=self.update_amount
)
self.combo.pack(fill="x", padx=10)
self.amount_label = ctk.CTkLabel(pay, text="", font=("Arial", 12, "italic"))
self.amount_label.pack(anchor="e", padx=10, pady=(3, 6))
self.amount_label.pack(anchor="e", padx=10, pady=(5, 10))
self.update_amount()
# 🏦 Account Selection
acc = ctk.CTkFrame(frame, corner_radius=8)
acc.pack(fill="x", pady=(0, 8), padx=10)
ctk.CTkLabel(acc, text="Číslo účtu:", font=("Arial", 12, "bold")).pack(anchor="w", padx=10, pady=(6, 3))
self.selected_account = ctk.StringVar(value=list(ACCOUNTS.keys())[0])
ctk.CTkOptionMenu(
acc,
variable=self.selected_account,
values=list(ACCOUNTS.keys()),
font=("Arial", 12),
command=self.on_change
).pack(fill="x", padx=10, pady=(0, 6))
ctk.CTkButton(frame, text="Uložit QR kód",
ctk.CTkButton(frame, text="Vytvořit QR Platbu",
font=("Arial", 13, "bold"),
height=40,
command=self.ulozit_qr).pack(pady=6)
command=self.generate_qr).pack(pady=10)
self.qr_label = ctk.CTkLabel(frame, text="")
self.qr_label.pack(pady=6)
self.qr_label.pack(pady=15)
ctk.CTkLabel(frame,
text="© Ordinace MUDr. Buzalková | QR Platba dle ČBA v1.2",
@@ -149,8 +116,6 @@ class QRPlatbaApp(ctk.CTk):
text_color="#666").pack(side="bottom", pady=(10, 0))
self.center_window()
# QR automaticky při startu
self.after(100, self.refresh_qr)
# ================================
# 🪟 Center Window
@@ -168,36 +133,28 @@ class QRPlatbaApp(ctk.CTk):
# ================================
# 💸 Update and Generate
# ================================
def _get_current(self):
"""Vrátí (item, iban, spayd) pro aktuální výběr."""
def update_amount(self, _=None):
display_item = self.selected_item.get()
item = self.item_map[display_item]
iban = ACCOUNTS[self.selected_account.get()]
spayd = create_spayd(iban, ITEMS[item], RODCIS, f"{PRIJMENI} {JMENO} – {item}", CURRENCY)
return item, iban, spayd
def on_change(self, _=None):
"""Při změně dropdownu – aktualizuj částku i QR."""
self.refresh_qr()
def refresh_qr(self):
"""Zobrazí QR kód pro aktuální výběr (bez uložení)."""
item, _, spayd = self._get_current()
self.amount_label.configure(text=f"Částka: {ITEMS[item]:.2f} Kč")
img = qrcode.make(spayd)
img_resized = img.resize((200, 200), Image.LANCZOS)
qr_tk = ImageTk.PhotoImage(img_resized)
self.qr_label.configure(image=qr_tk)
self.qr_label.image = qr_tk
def ulozit_qr(self):
"""Uloží QR kód do souboru a informuje uživatele."""
item, _, spayd = self._get_current()
def generate_qr(self):
display_item = self.selected_item.get()
item = self.item_map[display_item]
spayd = create_spayd(IBAN, ITEMS[item], RODCIS, f"{PRIJMENI} {JMENO} – {item}", CURRENCY)
img = qrcode.make(spayd)
filename = f"{PRIJMENI}_{JMENO}_{datetime.now():%Y%m%d_%H%M%S}.png"
out_path = OUTPUT_DIR / filename
img.save(out_path)
messagebox.showinfo("Uloženo", f"QR kód uložen:\n{out_path}")
img_resized = img.resize((300, 300))
qr_tk = ImageTk.PhotoImage(img_resized)
self.qr_label.configure(image=qr_tk)
self.qr_label.image = qr_tk
self.update_idletasks()
self.geometry(f"{self.winfo_reqwidth()}x{self.winfo_reqheight()}")
self.center_window()
# ================================