notebook
This commit is contained in:
@@ -0,0 +1,249 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user