notebook
This commit is contained in:
157
30 Laboratoř2025.py
Normal file
157
30 Laboratoř2025.py
Normal file
@@ -0,0 +1,157 @@
|
||||
|
||||
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
from pathlib import Path
|
||||
import time
|
||||
import fdb
|
||||
import pandas as pd
|
||||
from openpyxl import load_workbook
|
||||
from openpyxl.worksheet.table import Table, TableStyleInfo
|
||||
from openpyxl.styles import Font, PatternFill, Alignment
|
||||
from openpyxl.utils import get_column_letter
|
||||
|
||||
# ================== Výstupní cesta ==================
|
||||
BASE_DIR = Path(r"..") # uprav dle potřeby
|
||||
timestamp = time.strftime("%Y-%m-%d %H-%M-%S")
|
||||
xlsx_name = f"Laboratoře {timestamp}.xlsx"
|
||||
xlsx_path = BASE_DIR / xlsx_name
|
||||
|
||||
# ================== Připojení k DB ==================
|
||||
con = fdb.connect(
|
||||
host='192.168.1.4',
|
||||
database=r'z:\\MEDICUS 3\\data\\medicus.FDB',
|
||||
user='sysdba',
|
||||
password='masterkey',
|
||||
charset='WIN1250'
|
||||
)
|
||||
|
||||
# ================== SQL dotaz ==================
|
||||
# Pozn.: CAST(ockzaz.datum AS DATE) pro přesné párování na lecd.datose (DATE).
|
||||
sql = """
|
||||
select labvh.idpacient,
|
||||
kar.rodcis,
|
||||
upper(KAR.PRIJMENI) || ', ' || KAR.JMENO AS JMENO_CELE,
|
||||
labvh.datum,
|
||||
labvd.idvh,
|
||||
labvd.poradi,
|
||||
labvd.idmetod,
|
||||
labmetod.nazev,
|
||||
labvd.vysl,
|
||||
labvd.idhodn,
|
||||
labhodn.zkr,
|
||||
labvd.idjedn,
|
||||
labjedn.jedn,
|
||||
labvd.TYP
|
||||
from labvd
|
||||
join labvh on labvh.idvh=labvd.idvh
|
||||
join labmetod on labvd.idmetod=labmetod.idmetod
|
||||
join labjedn on labjedn.idjedn=labvd.idjedn
|
||||
join labhodn on labvd.idhodn=labhodn.idhodn
|
||||
join kar on kar.idpac=labvh.idpacient
|
||||
where EXTRACT(YEAR FROM labvh.datum)=2025
|
||||
order by labvh.datum, kar.rodcis, labvd.poradi
|
||||
"""
|
||||
#and nazev containing 'chřipka'
|
||||
# ================== Načtení do DataFrame ==================
|
||||
df = pd.read_sql(sql, con)
|
||||
con.close()
|
||||
|
||||
# Textové sloupce jako string
|
||||
for col in ["idpacient", "datum", "idvh", "poradi", "idmetod", "vysl", "idhodn", "idjedn","typ"]:
|
||||
if col in df.columns:
|
||||
df[col] = df[col].astype("string")
|
||||
|
||||
# Převod dat na datetime
|
||||
for dcol in ["datum", "Expirace"]:
|
||||
if dcol in df.columns:
|
||||
df[dcol] = pd.to_datetime(df[dcol], errors="coerce")
|
||||
|
||||
# ================== Uložení do Excelu ==================
|
||||
with pd.ExcelWriter(xlsx_path, engine="openpyxl") as writer:
|
||||
df.to_excel(writer, index=False, sheet_name="Laboratoře")
|
||||
|
||||
# ================== Formátování ==================
|
||||
wb = load_workbook(xlsx_path)
|
||||
|
||||
def hide_columns(ws, column_names):
|
||||
"""Hide columns in Excel by header name (case-insensitive)."""
|
||||
header = [str(c.value).strip().upper() if c.value else "" for c in ws[1]]
|
||||
for name in column_names:
|
||||
if name.strip().upper() in header:
|
||||
col_idx = header.index(name.strip().upper()) + 1
|
||||
col_letter = get_column_letter(col_idx)
|
||||
ws.column_dimensions[col_letter].hidden = True
|
||||
|
||||
def center_columns(ws, column_names):
|
||||
"""Center-align given columns (case-insensitive header match)."""
|
||||
header_values = [str(c.value).strip().upper() if c.value else "" for c in ws[1]]
|
||||
index_map = {h: i + 1 for i, h in enumerate(header_values)}
|
||||
|
||||
for name in column_names:
|
||||
idx = index_map.get(name.strip().upper())
|
||||
if not idx:
|
||||
continue
|
||||
# Center the header cell
|
||||
ws.cell(row=1, column=idx).alignment = Alignment(horizontal="center", vertical="center")
|
||||
# Center all data cells in the column
|
||||
for row in ws.iter_rows(min_row=2, min_col=idx, max_col=idx, max_row=ws.max_row):
|
||||
row[0].alignment = Alignment(horizontal="center", vertical="center")
|
||||
|
||||
def autosize_columns(ws):
|
||||
for col_idx in range(1, ws.max_column + 1):
|
||||
col_letter = get_column_letter(col_idx)
|
||||
max_len = 0
|
||||
for cell in ws[col_letter]:
|
||||
val = "" if cell.value is None else str(cell.value)
|
||||
if len(val) > max_len:
|
||||
max_len = len(val)
|
||||
ws.column_dimensions[col_letter].width = min(max(12, max_len + 2), 60)
|
||||
|
||||
def style_table(ws, table_name: str):
|
||||
max_row = ws.max_row
|
||||
max_col = ws.max_column
|
||||
if max_col == 0:
|
||||
return
|
||||
|
||||
# Hlavička vzhled
|
||||
header_fill = PatternFill("solid", fgColor="D9E1F2")
|
||||
for cell in ws[1]:
|
||||
cell.font = Font(bold=True)
|
||||
cell.fill = header_fill
|
||||
cell.alignment = Alignment(vertical="center")
|
||||
|
||||
# Freeze top row
|
||||
ws.freeze_panes = "A2"
|
||||
|
||||
# Tabulka jen pokud jsou data
|
||||
if max_row < 2:
|
||||
autosize_columns(ws)
|
||||
return
|
||||
|
||||
ref = f"A1:{get_column_letter(max_col)}{max_row}"
|
||||
tbl = Table(displayName=table_name, ref=ref)
|
||||
tbl.tableStyleInfo = TableStyleInfo(
|
||||
name="TableStyleMedium9", showRowStripes=True, showColumnStripes=False
|
||||
)
|
||||
ws.add_table(tbl)
|
||||
autosize_columns(ws)
|
||||
|
||||
def format_dates(ws, columns_names):
|
||||
header = [c.value for c in ws[1]]
|
||||
date_cols = [header.index(name) + 1 for name in columns_names if name in header]
|
||||
for col_idx in date_cols:
|
||||
for row in ws.iter_rows(min_row=2, min_col=col_idx, max_col=col_idx, max_row=ws.max_row):
|
||||
row[0].number_format = "DD.MM.YYYY"
|
||||
|
||||
ws_main = wb["Laboratoře"]
|
||||
style_table(ws_main, "Laboratoře")
|
||||
format_dates(ws_main, ["datum", "Expirace"])
|
||||
center_columns(ws_main, ["ZKR","JEDN","idpacient", "datum", "idvh", "poradi", "idmetod", "vysl", "idhodn", "idjedn","typ"])
|
||||
hide_columns(ws_main, ["IDVH", "IDMETOD", "IDJEDN", "IDHODN", "TYP","IDPACIENT","PORADI"])
|
||||
|
||||
|
||||
wb.save(xlsx_path)
|
||||
|
||||
print(f"Hotovo. Uloženo do: {xlsx_path.resolve()}")
|
||||
13
40 TowerLibrary.py
Normal file
13
40 TowerLibrary.py
Normal file
@@ -0,0 +1,13 @@
|
||||
import os
|
||||
from pathlib import Path
|
||||
cesta=r"\\tower\library"
|
||||
|
||||
for dirpath,dirnames,filenames in os.walk(cesta):
|
||||
|
||||
for ddir in dirnames:
|
||||
# print(ddir)
|
||||
p=Path(dirpath, ddir)
|
||||
if not any(p.iterdir()):
|
||||
print(f"Directory is empty {p}")
|
||||
p.rmdir()
|
||||
print(f"Directory was deleted {p}")
|
||||
161
drop_all_empty_tables_aggressive.py
Normal file
161
drop_all_empty_tables_aggressive.py
Normal file
@@ -0,0 +1,161 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
import fdb
|
||||
from pathlib import Path
|
||||
|
||||
# ================== CONFIG ==================
|
||||
DB_PATH = r"u:\MEDICUS 3\data\medicus.FDB"
|
||||
HOST = "localhost"
|
||||
USER = "sysdba"
|
||||
PASSWORD = "masterkey"
|
||||
LOG_FILE = Path("dropped_tables_full.txt")
|
||||
# ============================================
|
||||
|
||||
con = fdb.connect(
|
||||
host=HOST,
|
||||
database=DB_PATH,
|
||||
user=USER,
|
||||
password=PASSWORD,
|
||||
charset="WIN1250",
|
||||
)
|
||||
cur = con.cursor()
|
||||
|
||||
print("💣 Aggressively removing all empty tables and their dependencies...")
|
||||
|
||||
# ---------------------------------------------------------------------
|
||||
# Helper functions
|
||||
# ---------------------------------------------------------------------
|
||||
def commit_safe():
|
||||
try:
|
||||
con.commit()
|
||||
except Exception as e:
|
||||
print(f" ⚠️ Commit failed: {e}")
|
||||
con.rollback()
|
||||
|
||||
def drop_objects(sql_query: str, drop_sql_template: str, obj_type: str, table_name: str):
|
||||
"""Generic dropper for dependent objects (views, triggers, procedures)."""
|
||||
cur.execute(sql_query)
|
||||
objs = [r[0].strip() for r in cur.fetchall()]
|
||||
for name in objs:
|
||||
stmt = drop_sql_template.format(name=name)
|
||||
try:
|
||||
print(f" 🔧 Dropping {obj_type}: {name}")
|
||||
con.execute_immediate(stmt)
|
||||
commit_safe()
|
||||
except Exception as e:
|
||||
print(f" ⚠️ Could not drop {obj_type} {name}: {e}")
|
||||
con.rollback()
|
||||
|
||||
def drop_fk_constraints_referring_to(table_name: str):
|
||||
cur.execute(f"""
|
||||
SELECT TRIM(rc.rdb$relation_name), TRIM(rc.rdb$constraint_name)
|
||||
FROM rdb$relation_constraints rc
|
||||
JOIN rdb$ref_constraints ref ON ref.rdb$constraint_name = rc.rdb$constraint_name
|
||||
JOIN rdb$relation_constraints pk ON pk.rdb$constraint_name = ref.rdb$const_name_uq
|
||||
WHERE rc.rdb$constraint_type = 'FOREIGN KEY'
|
||||
AND UPPER(pk.rdb$relation_name) = '{table_name.upper()}'
|
||||
""")
|
||||
for fk_table, fk_name in cur.fetchall():
|
||||
stmt = f'ALTER TABLE {fk_table} DROP CONSTRAINT {fk_name}'
|
||||
try:
|
||||
print(f" 🔧 Dropping FK {fk_name} in {fk_table}")
|
||||
con.execute_immediate(stmt)
|
||||
commit_safe()
|
||||
except Exception as e:
|
||||
print(f" ⚠️ Could not drop FK {fk_name} in {fk_table}: {e}")
|
||||
con.rollback()
|
||||
|
||||
# ---------------------------------------------------------------------
|
||||
# Collect all user tables
|
||||
# ---------------------------------------------------------------------
|
||||
cur.execute("""
|
||||
SELECT TRIM(rdb$relation_name)
|
||||
FROM rdb$relations
|
||||
WHERE rdb$system_flag = 0
|
||||
ORDER BY 1
|
||||
""")
|
||||
tables = [r[0] for r in cur.fetchall()]
|
||||
|
||||
dropped, skipped = [], []
|
||||
|
||||
# ---------------------------------------------------------------------
|
||||
# Main loop
|
||||
# ---------------------------------------------------------------------
|
||||
for t in tables:
|
||||
try:
|
||||
cur.execute(f"SELECT COUNT(*) FROM {t}")
|
||||
count = cur.fetchone()[0]
|
||||
except Exception as e:
|
||||
print(f"⚠️ Skipping {t}: {e}")
|
||||
continue
|
||||
|
||||
if count > 0:
|
||||
continue
|
||||
|
||||
print(f"\n🧹 {t} → empty, removing dependencies and dropping...")
|
||||
|
||||
# 1️⃣ Drop foreign keys referencing this table
|
||||
drop_fk_constraints_referring_to(t)
|
||||
|
||||
# 2️⃣ Drop views referencing this table
|
||||
drop_objects(
|
||||
sql_query=f"""
|
||||
SELECT TRIM(rdb$view_name)
|
||||
FROM rdb$view_relations
|
||||
WHERE UPPER(rdb$relation_name) = '{t.upper()}'
|
||||
""",
|
||||
drop_sql_template="DROP VIEW {name}",
|
||||
obj_type="view",
|
||||
table_name=t
|
||||
)
|
||||
|
||||
# 3️⃣ Drop triggers whose source mentions this table
|
||||
drop_objects(
|
||||
sql_query=f"""
|
||||
SELECT TRIM(rdb$trigger_name)
|
||||
FROM rdb$triggers
|
||||
WHERE rdb$trigger_source CONTAINING '{t.upper()}'
|
||||
""",
|
||||
drop_sql_template="DROP TRIGGER {name}",
|
||||
obj_type="trigger",
|
||||
table_name=t
|
||||
)
|
||||
|
||||
# 4️⃣ Drop stored procedures referencing this table
|
||||
drop_objects(
|
||||
sql_query=f"""
|
||||
SELECT TRIM(rdb$procedure_name)
|
||||
FROM rdb$procedures
|
||||
WHERE rdb$procedure_source CONTAINING '{t.upper()}'
|
||||
""",
|
||||
drop_sql_template="DROP PROCEDURE {name}",
|
||||
obj_type="procedure",
|
||||
table_name=t
|
||||
)
|
||||
|
||||
# 5️⃣ Finally, drop the table itself
|
||||
try:
|
||||
con.execute_immediate(f"DROP TABLE {t}")
|
||||
commit_safe()
|
||||
dropped.append(t)
|
||||
print(f" ✅ Dropped table {t}")
|
||||
except Exception as e:
|
||||
print(f" ⚠️ Could not drop {t}: {e}")
|
||||
con.rollback()
|
||||
skipped.append(t)
|
||||
|
||||
con.close()
|
||||
|
||||
# ---------------------------------------------------------------------
|
||||
# Log results
|
||||
# ---------------------------------------------------------------------
|
||||
LOG_FILE.write_text(
|
||||
"=== DROPPED TABLES ===\n" + "\n".join(dropped) +
|
||||
"\n\n=== SKIPPED TABLES ===\n" + "\n".join(skipped),
|
||||
encoding="utf-8"
|
||||
)
|
||||
print("\n✅ Done.")
|
||||
print(f"Total dropped: {len(dropped)}")
|
||||
print(f"Skipped: {len(skipped)}")
|
||||
print(f"Log saved to {LOG_FILE.resolve()}")
|
||||
91
find_empty_tables.py
Normal file
91
find_empty_tables.py
Normal file
@@ -0,0 +1,91 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
import fdb
|
||||
from pathlib import Path
|
||||
|
||||
# ================== CONFIG ==================
|
||||
DB_PATH = r"u:\MEDICUS 3\data\medicus.FDB"
|
||||
HOST = "localhost"
|
||||
USER = "sysdba"
|
||||
PASSWORD = "masterkey"
|
||||
LOG_FILE = Path("dropped_tables_with_deps.txt")
|
||||
# ============================================
|
||||
|
||||
con = fdb.connect(
|
||||
host=HOST,
|
||||
database=DB_PATH,
|
||||
user=USER,
|
||||
password=PASSWORD,
|
||||
charset="WIN1250",
|
||||
)
|
||||
cur = con.cursor()
|
||||
|
||||
print("🔍 Checking tables for emptiness and dependencies...")
|
||||
|
||||
# All user tables
|
||||
cur.execute("""
|
||||
SELECT TRIM(rdb$relation_name)
|
||||
FROM rdb$relations
|
||||
WHERE rdb$system_flag = 0
|
||||
ORDER BY 1
|
||||
""")
|
||||
tables = [r[0] for r in cur.fetchall()]
|
||||
|
||||
dropped = []
|
||||
skipped = []
|
||||
|
||||
def drop_fk_constraints_referring_to(table_name: str):
|
||||
# Find all FKs that reference this table (PK or UNIQUE)
|
||||
cur.execute(f"""
|
||||
SELECT TRIM(rc.rdb$relation_name), TRIM(rc.rdb$constraint_name)
|
||||
FROM rdb$relation_constraints rc
|
||||
JOIN rdb$ref_constraints ref ON ref.rdb$constraint_name = rc.rdb$constraint_name
|
||||
JOIN rdb$relation_constraints pk ON pk.rdb$constraint_name = ref.rdb$const_name_uq
|
||||
WHERE rc.rdb$constraint_type = 'FOREIGN KEY'
|
||||
AND UPPER(pk.rdb$relation_name) = '{table_name.upper()}'
|
||||
""")
|
||||
for fk_table, fk_name in cur.fetchall():
|
||||
stmt = f'ALTER TABLE {fk_table} DROP CONSTRAINT {fk_name}'
|
||||
try:
|
||||
print(f" 🔧 Dropping FK {fk_name} in {fk_table}")
|
||||
con.execute_immediate(stmt)
|
||||
con.commit() # commit each FK drop
|
||||
except Exception as e:
|
||||
print(f" ⚠️ Could not drop FK {fk_name} in {fk_table}: {e}")
|
||||
con.rollback()
|
||||
|
||||
for t in tables:
|
||||
try:
|
||||
# Count rows
|
||||
cur.execute(f"SELECT COUNT(*) FROM {t}")
|
||||
count = cur.fetchone()[0]
|
||||
if count > 0:
|
||||
continue
|
||||
|
||||
print(f"\n🧹 {t} → empty")
|
||||
|
||||
# 1) Drop FKs that reference this table
|
||||
drop_fk_constraints_referring_to(t)
|
||||
|
||||
# 2) Try to drop the table itself
|
||||
try:
|
||||
con.execute_immediate(f"DROP TABLE {t}")
|
||||
con.commit() # commit this table drop
|
||||
dropped.append(t)
|
||||
print(f" ✅ Dropped table {t}")
|
||||
except Exception as e:
|
||||
print(f" ⚠️ Could not drop {t}: {e}")
|
||||
con.rollback()
|
||||
skipped.append(t)
|
||||
|
||||
except Exception as e:
|
||||
print(f"⚠️ Skipping {t}: {e}")
|
||||
|
||||
con.close()
|
||||
|
||||
LOG_FILE.write_text("\n".join(dropped), encoding="utf-8")
|
||||
print("\n✅ Done.")
|
||||
print(f"Total dropped: {len(dropped)}")
|
||||
print(f"Skipped: {len(skipped)}")
|
||||
print(f"List saved to {LOG_FILE.resolve()}")
|
||||
Reference in New Issue
Block a user