222 lines
7.9 KiB
Python
222 lines
7.9 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
|
||
import sys
|
||
import io
|
||
import time
|
||
import json
|
||
# ZMĚNA: Importujeme oficiální konektor
|
||
import mysql.connector
|
||
from mysql.connector import Error
|
||
from typing import Dict, Any, List
|
||
from pathlib import Path
|
||
from datetime import date
|
||
|
||
# ====================================================================
|
||
# A. PONECHÁNO: Vynucení UTF-8 pro správnou diakritiku
|
||
# ====================================================================
|
||
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
|
||
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8', errors='replace')
|
||
|
||
# =========================================
|
||
# KONFIGURACE PRO TEST
|
||
# =========================================
|
||
|
||
TEST_JSON_PATH = r"z:\Dropbox\!!!Days\Downloads Z230\Fio\2100074583\2025-09-06_to_2025-12-05.json"
|
||
TEST_ACCOUNT_ID = "2100074583"
|
||
|
||
# MySQL connection parameters
|
||
DB = {
|
||
"host": "192.168.1.76",
|
||
"port": 3307,
|
||
"user": "root",
|
||
"password": "Vlado9674+",
|
||
"database": "fio",
|
||
"charset": "utf8", # mysql-connector používá 'utf8' nebo 'utf8mb4'
|
||
}
|
||
|
||
|
||
# =========================================
|
||
# POMOCNÉ FUNKCE (Z vašeho původního kódu)
|
||
# =========================================
|
||
|
||
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 load_test_data(path: str) -> Any:
|
||
"""Načte data přímo ze souboru JSON."""
|
||
try:
|
||
with open(path, "r", encoding="utf-8") as f:
|
||
return json.load(f)
|
||
except Exception as e:
|
||
print(f"FATÁLNÍ CHYBA: Nelze načíst JSON soubor z {path}: {e}", flush=True)
|
||
return None
|
||
|
||
|
||
# =========================================
|
||
# HLAVNÍ TESTOVACÍ LOGIKA
|
||
# =========================================
|
||
|
||
def run_test():
|
||
start_all = time.time()
|
||
|
||
print("=== TEST VKLÁDÁNÍ DAT Z JSON SOUBORU (OPRAVA S MYSQL-CONNECTOR) ===", flush=True)
|
||
print(f"Zdroj dat: {TEST_JSON_PATH}", flush=True)
|
||
|
||
# KROK 1: Načtení dat přímo z JSON souboru
|
||
data = load_test_data(TEST_JSON_PATH)
|
||
if data is None:
|
||
return
|
||
|
||
# KROK 2: Extrakce transakcí
|
||
tlist = data.get("accountStatement", {}).get("transactionList", {}).get("transaction", [])
|
||
if isinstance(tlist, dict):
|
||
tlist = [tlist]
|
||
|
||
fio_acc_id = data.get("accountStatement", {}).get("info", {}).get("accountId")
|
||
expected_count = len(tlist)
|
||
|
||
print(f"Očekávaný počet transakcí v JSON: {expected_count}", flush=True)
|
||
|
||
if expected_count == 0:
|
||
print("Test přeskočen, JSON je prázdný.", flush=True)
|
||
return
|
||
|
||
# KROK 3: Připojení k DB
|
||
try:
|
||
# ZMĚNA: Používáme mysql.connector
|
||
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()
|
||
# ZMĚNA: Používáme dictionary=True, což je zde podporováno
|
||
cur_dict = conn.cursor(dictionary=True)
|
||
except Error as e:
|
||
print(f"FATÁLNÍ CHYBA: Nelze se připojit k DB: {e}", flush=True)
|
||
return
|
||
|
||
# KROK 4: Sestavení SQL dotazu a řádků (Konverze ID na string zůstává pro bezpečnost)
|
||
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) \
|
||
"""
|
||
|
||
rows = []
|
||
for t in tlist:
|
||
id_operace_val = safe_col(t, 22)
|
||
id_pokynu_val = safe_col(t, 17)
|
||
|
||
row = {
|
||
# Klíčová konverze ID na string (VARCHAR)
|
||
"id_operace": str(id_operace_val) if id_operace_val is not None else None,
|
||
"cislo_uctu": fio_acc_id,
|
||
"transaction_date": clean_date(safe_col(t, 0)),
|
||
"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)
|
||
|
||
# KROK 5: Vložení dat do DB
|
||
inserted = 0
|
||
try:
|
||
# Používáme executemany s připravenými daty
|
||
cur.executemany(sql, rows)
|
||
conn.commit()
|
||
inserted = len(rows)
|
||
except Error as e:
|
||
print(f" ❌ Chyba při VKLÁDÁNÍ do DB: {e}", flush=True)
|
||
conn.rollback()
|
||
|
||
# KROK 6: Kontrola výsledku v DB (používá Dictionary=True)
|
||
check_query = f"SELECT count(*) AS count FROM transactions WHERE cislo_uctu = '{fio_acc_id}'"
|
||
cur_dict.execute(check_query)
|
||
current_db_count = cur_dict.fetchone()['count']
|
||
|
||
conn.close()
|
||
|
||
print(f"\n--- SHRNUTÍ TESTU ---", flush=True)
|
||
print(f"Očekávalo se vložení/aktualizace řádků: {expected_count}", flush=True)
|
||
print(f"Počet řádků zpracovaných skriptem: {inserted}", flush=True)
|
||
print(f"Aktuální počet záznamů pro účet {fio_acc_id} v DB: {current_db_count}", flush=True)
|
||
print(f"Celkový čas: {time.time() - start_all:.2f} s", flush=True)
|
||
|
||
if inserted == expected_count and current_db_count >= expected_count:
|
||
print("✅ TEST ÚSPĚŠNÝ: Všechny transakce byly vloženy/aktualizovány, nebo DB obsahuje očekávaný počet.",
|
||
flush=True)
|
||
else:
|
||
print("🔥 TEST SELHAL: Existuje nesoulad mezi očekávaným a skutečným počtem záznamů.", flush=True)
|
||
|
||
|
||
# ======================================================
|
||
# ENTRY POINT
|
||
# ======================================================
|
||
|
||
if __name__ == "__main__":
|
||
run_test() |