319 lines
9.5 KiB
Python
319 lines
9.5 KiB
Python
#!/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 EXPORT SCRIPT — FINÁLNÍ VERZE S KONTROLNÍM VÝPISEM
|
|
-------------------------------------------
|
|
|
|
Skript nyní vypisuje POČET ŘÁDKŮ NAČTENÝCH Z DATABÁZE pro každý list.
|
|
Tím ověříme, zda se data ztrácí při čtení z DB, nebo až při exportu do Excelu.
|
|
"""
|
|
|
|
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, date as dt_date
|
|
from decimal import Decimal
|
|
import os
|
|
import glob
|
|
import json
|
|
from typing import List, Dict, Any
|
|
|
|
# ======================================================
|
|
# CONFIGURATION
|
|
# ======================================================
|
|
|
|
# MySQL server parameters
|
|
DB_HOST = "192.168.1.76"
|
|
DB_PORT = 3307
|
|
DB_USER = "root"
|
|
DB_PASS = "Vlado9674+"
|
|
DB_NAME = "fio"
|
|
|
|
# Where to save Excel files
|
|
OUTPUT_DIR = r"Z:\Dropbox\!!!Days\Downloads Z230"
|
|
|
|
# JSON file with list of accounts (name + account_number)
|
|
ACCOUNTS_JSON = r"C:\Users\vlado\PycharmProjects\FIO\accounts.json"
|
|
|
|
# VÍCE SLOUPCŮ PŘENASTAVENO NA TEXT
|
|
TEXT_COLUMNS = [
|
|
"cislo_uctu",
|
|
"protiucet",
|
|
"kod_banky",
|
|
"vs",
|
|
"ks",
|
|
"ss",
|
|
"id_operace",
|
|
"id_pokynu"
|
|
]
|
|
|
|
|
|
# ======================================================
|
|
# REMOVE OLD EXPORT FILES (Beze změny)
|
|
# ======================================================
|
|
|
|
def cleanup_old_exports():
|
|
"""
|
|
Deletes older versions of exported XLSX files.
|
|
"""
|
|
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
|
|
|
|
|
|
# ======================================================
|
|
# CORE EXCEL FORMATTING FUNCTION (Oprava konverze typů zachována)
|
|
# ======================================================
|
|
|
|
def format_sheet(ws, rows: List[Dict[str, Any]], headers: List[str]):
|
|
"""
|
|
Applies ALL formatting rules to a worksheet:
|
|
FIX: Explicitní konverze datových typů pro OpenPyXL zachována.
|
|
"""
|
|
|
|
# -------------------------------
|
|
# 1) Format HEADER row
|
|
# -------------------------------
|
|
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")
|
|
|
|
# -------------------------------
|
|
# 2) Write DATA rows (OPRAVA KONVERZE TYPŮ)
|
|
# -------------------------------
|
|
# Klíčová konverze pro Decimal/Date objekty
|
|
for row in rows:
|
|
excel_row = []
|
|
for h in headers:
|
|
val = row[h]
|
|
|
|
# 🛠️ KLÍČOVÁ OPRAVA: Konverze MySQL/Decimal objektů na nativní Python typy
|
|
if isinstance(val, Decimal):
|
|
val = float(val)
|
|
elif isinstance(val, dt_date) and not isinstance(val, datetime):
|
|
val = val.strftime("%Y-%m-%d")
|
|
# -----------------------------------------------------------
|
|
|
|
# Pro text-sensitive sloupce (ID, symboly), zapisuj ="hodnota"
|
|
if h in TEXT_COLUMNS and val is not None:
|
|
val = str(val)
|
|
excel_row.append(f'="{val}"')
|
|
else:
|
|
excel_row.append(val)
|
|
|
|
ws.append(excel_row)
|
|
|
|
# -------------------------------
|
|
# 3) Background coloring by "amount"
|
|
# -------------------------------
|
|
fill_red = PatternFill(start_color="FFFFDDDD", end_color="FFFFDDDD", fill_type="solid")
|
|
fill_green = PatternFill(start_color="FFEEFFEE", end_color="FFEEFFEE", fill_type="solid")
|
|
|
|
try:
|
|
amount_col_index = headers.index("amount") + 1
|
|
except ValueError:
|
|
amount_col_index = -1
|
|
|
|
if amount_col_index != -1:
|
|
for row_idx in range(2, len(rows) + 2):
|
|
cell_amount = ws.cell(row=row_idx, column=amount_col_index)
|
|
|
|
try:
|
|
value = float(str(cell_amount.value).strip('="'))
|
|
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
|
|
|
|
# -------------------------------
|
|
# 4) Fixed column widths (22 sloupců)
|
|
# -------------------------------
|
|
fixed_widths = [
|
|
13, 14, 11, 14, 8, 14, 11, 30, 30, 25,
|
|
13, 13, 13, 35, 30, 15, 13, 30, 20, 13,
|
|
30, 20
|
|
]
|
|
|
|
if len(fixed_widths) < len(headers):
|
|
fixed_widths.extend([15] * (len(headers) - len(fixed_widths)))
|
|
|
|
for i, width in enumerate(fixed_widths, start=1):
|
|
col_letter = chr(64 + i)
|
|
ws.column_dimensions[col_letter].width = width
|
|
|
|
# -------------------------------
|
|
# 5) Add borders + alignment
|
|
# -------------------------------
|
|
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)
|
|
|
|
ALIGN_CENTER_COLS = ["id_operace", "transaction_date", "currency", "kod_banky", "vs", "ks", "ss"]
|
|
center_indices = [headers.index(col) + 1 for col in ALIGN_CENTER_COLS if col in 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
|
|
if col_idx in center_indices:
|
|
cell.alignment = align_center
|
|
|
|
ws.freeze_panes = "A2"
|
|
ws.auto_filter.ref = ws.dimensions
|
|
|
|
|
|
# ======================================================
|
|
# MAIN EXPORT PROCESS
|
|
# ======================================================
|
|
|
|
def export_fio():
|
|
print("Connecting to MySQL...")
|
|
|
|
# Connect to MySQL database
|
|
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
|
|
|
|
# Používáme dictionary=True pro získání dat jako slovník
|
|
cur = conn.cursor(dictionary=True)
|
|
|
|
# -------------------------------
|
|
# Load accounts.json
|
|
# -------------------------------
|
|
with open(ACCOUNTS_JSON, "r", encoding="utf-8") as f:
|
|
accounts = json.load(f)
|
|
|
|
# -------------------------------
|
|
# Define priority first sheets
|
|
# -------------------------------
|
|
preferred_order = [
|
|
"CZK rodina",
|
|
"CZK ordinace",
|
|
"CZK na jídlo",
|
|
"CZK TrialHelp",
|
|
"CZK maminka svojě věci"
|
|
]
|
|
|
|
accounts_sorted = []
|
|
|
|
# Step 1: add priority accounts first
|
|
for pref in preferred_order:
|
|
for acc in accounts:
|
|
if acc["name"] == pref:
|
|
accounts_sorted.append(acc)
|
|
|
|
# Step 2: add remaining accounts afterward
|
|
for acc in accounts:
|
|
if acc not in accounts_sorted:
|
|
accounts_sorted.append(acc)
|
|
|
|
# -------------------------------
|
|
# Create a new Excel workbook
|
|
# -------------------------------
|
|
wb = Workbook()
|
|
wb.remove(wb.active) # remove default empty sheet
|
|
|
|
# -------------------------------
|
|
# FIRST SHEET: ALL TRANSACTIONS
|
|
# -------------------------------
|
|
cur.execute("SELECT * FROM transactions ORDER BY transaction_date DESC")
|
|
all_rows = cur.fetchall()
|
|
|
|
if all_rows:
|
|
headers = list(all_rows[0].keys())
|
|
# Tisk počtu řádků pro "ALL"
|
|
print(f"➡ Sheet: ALL | Řádků z DB: {len(all_rows)}")
|
|
|
|
ws_all = wb.create_sheet(title="ALL")
|
|
ws_all.append(headers)
|
|
format_sheet(ws_all, all_rows, headers)
|
|
|
|
# -------------------------------
|
|
# INDIVIDUAL SHEETS PER ACCOUNT
|
|
# -------------------------------
|
|
for acc in accounts_sorted:
|
|
acc_num = acc["account_number"]
|
|
sheet_name = acc["name"][:31] # Excel sheet name limit
|
|
|
|
print(f"➡ Creating sheet: {sheet_name}", end=' | ') # Tisk názvu listu
|
|
|
|
query = f"""
|
|
SELECT *
|
|
FROM transactions
|
|
WHERE cislo_uctu = '{acc_num}'
|
|
ORDER BY transaction_date DESC
|
|
"""
|
|
|
|
cur.execute(query)
|
|
rows = cur.fetchall()
|
|
|
|
# VÝPIS POČTU ZÁZNAMŮ Z DB
|
|
print(f"Řádků z DB: {len(rows)}")
|
|
|
|
if not rows:
|
|
print(f"⚠ No data for {sheet_name}")
|
|
continue
|
|
|
|
headers = list(rows[0].keys())
|
|
ws = wb.create_sheet(title=sheet_name)
|
|
ws.append(headers)
|
|
|
|
format_sheet(ws, rows, headers)
|
|
|
|
conn.close()
|
|
|
|
# -------------------------------
|
|
# Save Excel file
|
|
# -------------------------------
|
|
|
|
cleanup_old_exports()
|
|
|
|
# File name includes timestamp
|
|
timestamp = datetime.now().strftime("%Y-%m-%d %H-%M-%S")
|
|
filename = f"{timestamp} FIO transactions.xlsx"
|
|
output_file = os.path.join(OUTPUT_DIR, filename)
|
|
|
|
wb.save(output_file)
|
|
|
|
print(f"✅ Export complete:\n{output_file}")
|
|
|
|
|
|
# ======================================================
|
|
# MAIN ENTRY POINT
|
|
# ======================================================
|
|
|
|
if __name__ == "__main__":
|
|
export_fio() |