reporter

2025-12-07 07:11:12 +01:00
parent 49ede2b452
commit 4c57c92332
9 changed files with 2087 additions and 0 deletions
--- a/reports/Trash/30
+++ b/reports/Trash/30
@@ -0,0 +1,222 @@
+#!/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()