269 lines
7.6 KiB
Python
269 lines
7.6 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
|
||
import sys
|
||
import io
|
||
|
||
# UTF-8 console 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')
|
||
|
||
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
|
||
from pathlib import Path
|
||
# ==============================
|
||
# DELETE OLD REPORTS (OPTION C)
|
||
# ==============================
|
||
|
||
def delete_all_old_reports(directory: Path):
|
||
"""Deletes all previously generated ordinace expense reports."""
|
||
pattern = "*fio ordinace transactions.xlsx"
|
||
deleted = 0
|
||
|
||
for f in directory.glob(pattern):
|
||
try:
|
||
f.unlink()
|
||
deleted += 1
|
||
print(f"🗑 Deleted old report: {f.name}")
|
||
except Exception as e:
|
||
print(f"❌ Could not delete {f.name}: {e}")
|
||
|
||
if deleted == 0:
|
||
print("ℹ No old reports to delete.")
|
||
else:
|
||
print(f"✓ Deleted {deleted} old reports.")
|
||
|
||
|
||
# ======================================================
|
||
# CONFIG
|
||
# ======================================================
|
||
|
||
DB = {
|
||
"host": "192.168.1.76",
|
||
"port": 3307,
|
||
"user": "root",
|
||
"password": "Vlado9674+",
|
||
"database": "fio",
|
||
}
|
||
|
||
ORDINACE_ACCOUNT = "2800046620"
|
||
|
||
REPORTOVAT = {
|
||
"VZP": "1114007221",
|
||
"VOZP": "2010009091",
|
||
"ČPZP": "2054108761",
|
||
"OZP": "2070101041",
|
||
"ZPŠ": "2090309181",
|
||
"ZPMV": "2112108031",
|
||
}
|
||
|
||
OUTPUT_DIR = Path(r"z:\Dropbox\Ordinace\Reporty")
|
||
|
||
TEXT_COLUMNS = [
|
||
"cislo_uctu", "protiucet", "kod_banky",
|
||
"vs", "ks", "ss",
|
||
"id_operace", "id_pokynu"
|
||
]
|
||
|
||
|
||
# ======================================================
|
||
# FORMAT APPLYING (copied from main report)
|
||
# ======================================================
|
||
|
||
def format_sheet(ws, rows, headers):
|
||
|
||
# ---------------------- HEADER -----------------------
|
||
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")
|
||
|
||
# ---------------------- DATA ROWS --------------------
|
||
for row in rows:
|
||
excel_row = []
|
||
for h in headers:
|
||
val = row[h]
|
||
|
||
# Convert MySQL data types
|
||
if isinstance(val, Decimal):
|
||
val = float(val)
|
||
elif isinstance(val, dt_date):
|
||
val = val.strftime("%Y-%m-%d")
|
||
|
||
# For certain columns, force ="text"
|
||
if h in TEXT_COLUMNS and val is not None:
|
||
excel_row.append(f'="{val}"')
|
||
else:
|
||
excel_row.append(val)
|
||
|
||
ws.append(excel_row)
|
||
|
||
# ---------------------- COLORING ---------------------
|
||
fill_red = PatternFill(start_color="FFFFDDDD", fill_type="solid")
|
||
fill_green = PatternFill(start_color="FFEEFFEE", fill_type="solid")
|
||
|
||
try:
|
||
amount_col = headers.index("amount") + 1
|
||
except ValueError:
|
||
amount_col = -1
|
||
|
||
if amount_col != -1:
|
||
for r in range(2, len(rows) + 2):
|
||
cell = ws.cell(row=r, column=amount_col)
|
||
try:
|
||
value = float(str(cell.value).strip('="'))
|
||
except:
|
||
value = 0
|
||
|
||
fill = fill_red if value < 0 else fill_green
|
||
for c in range(1, len(headers) + 1):
|
||
ws.cell(row=r, column=c).fill = fill
|
||
|
||
# ---------------------- COLUMN WIDTHS -----------------
|
||
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):
|
||
letter = chr(64 + i)
|
||
ws.column_dimensions[letter].width = width
|
||
|
||
# ---------------------- BORDERS & ALIGNMENT ----------
|
||
thin = Side(border_style="thin", color="000000")
|
||
border = Border(left=thin, right=thin, top=thin, bottom=thin)
|
||
align_center = Alignment(horizontal="center")
|
||
|
||
center_cols = ["id_operace", "transaction_date", "currency", "kod_banky", "vs", "ks", "ss"]
|
||
center_indices = [headers.index(c) + 1 for c in center_cols if c in headers]
|
||
|
||
total_rows = len(rows) + 1
|
||
total_cols = len(headers)
|
||
|
||
for r in range(1, total_rows + 1):
|
||
for c in range(1, total_cols + 1):
|
||
cell = ws.cell(row=r, column=c)
|
||
cell.border = border
|
||
if c in center_indices:
|
||
cell.alignment = align_center
|
||
|
||
ws.freeze_panes = "A2"
|
||
ws.auto_filter.ref = ws.dimensions
|
||
|
||
|
||
# ======================================================
|
||
# EXPORT
|
||
# ======================================================
|
||
|
||
def export_ordinace():
|
||
print("Connecting MySQL...")
|
||
conn = mysql.connector.connect(**DB)
|
||
cur = conn.cursor(dictionary=True)
|
||
|
||
# ============================
|
||
# Load ALL transactions for ordinace
|
||
# ============================
|
||
sql_all = f"""
|
||
SELECT *
|
||
FROM transactions
|
||
WHERE cislo_uctu = '{ORDINACE_ACCOUNT}'
|
||
ORDER BY transaction_date DESC;
|
||
"""
|
||
cur.execute(sql_all)
|
||
all_rows = cur.fetchall()
|
||
|
||
if not all_rows:
|
||
print("❌ No transactions found for ordinace account.")
|
||
return
|
||
|
||
headers = list(all_rows[0].keys())
|
||
|
||
# Workbook
|
||
wb = Workbook()
|
||
wb.remove(wb.active)
|
||
|
||
# --------------------- ALL sheet ---------------------
|
||
ws_all = wb.create_sheet("ALL ordinace")
|
||
ws_all.append(headers)
|
||
format_sheet(ws_all, all_rows, headers)
|
||
|
||
print(f"➡ ALL ordinace rows: {len(all_rows)}")
|
||
|
||
# --------------------- INSURANCE sheets ---------------
|
||
summary = []
|
||
|
||
for name, acc in REPORTOVAT.items():
|
||
print(f"➡ Pojišťovna {name} ({acc})")
|
||
|
||
sql = f"""
|
||
SELECT *
|
||
FROM transactions
|
||
WHERE cislo_uctu = '{ORDINACE_ACCOUNT}'
|
||
AND (
|
||
protiucet <> '2070101041'
|
||
OR (protiucet = '2070101041' AND amount > 0)
|
||
)
|
||
AND protiucet = '{acc}'
|
||
ORDER BY transaction_date DESC;
|
||
"""
|
||
|
||
cur.execute(sql)
|
||
rows = cur.fetchall()
|
||
|
||
count = len(rows)
|
||
summa = sum(float(r["amount"]) for r in rows) if rows else 0
|
||
|
||
summary.append({
|
||
"Pojišťovna": name,
|
||
"Účet": acc,
|
||
"Počet transakcí": count,
|
||
"Součet": summa
|
||
})
|
||
|
||
if not rows:
|
||
print(f" ⚠ No rows")
|
||
continue
|
||
|
||
ws = wb.create_sheet(name)
|
||
ws.append(headers)
|
||
format_sheet(ws, rows, headers)
|
||
|
||
print(f" ✓ {count} rows, sum {summa:.2f} Kč")
|
||
|
||
# --------------------- SUMMARY sheet -----------------
|
||
ws_s = wb.create_sheet("Přehled")
|
||
ws_s.append(["Pojišťovna", "Účet", "Počet transakcí", "Součet Kč"])
|
||
|
||
for row in summary:
|
||
ws_s.append([
|
||
row["Pojišťovna"],
|
||
row["Účet"],
|
||
row["Počet transakcí"],
|
||
f"{row['Součet']:.2f}"
|
||
])
|
||
|
||
# ===========================
|
||
# Save Excel
|
||
# ===========================
|
||
timestamp = datetime.now().strftime("%Y-%m-%d %H-%M-%S")
|
||
out_file = OUTPUT_DIR / f"{timestamp} FIO ordinace transactions.xlsx"
|
||
|
||
wb.save(out_file)
|
||
print(f"\n✅ Export hotový:\n{out_file}")
|
||
|
||
|
||
# ======================================================
|
||
# MAIN
|
||
# ======================================================
|
||
|
||
if __name__ == "__main__":
|
||
delete_all_old_reports(OUTPUT_DIR)
|
||
export_ordinace()
|