notebook vb
This commit is contained in:
@@ -0,0 +1,184 @@
|
||||
"""
|
||||
002_detail_NES_NP_tabulky.py
|
||||
============================
|
||||
Průzkum klíčových tabulek neschopenky:
|
||||
- NES → hlavní tabulka pracovní neschopnosti
|
||||
- NP → pravděpodobně neschopenka podání
|
||||
- SOC_NEPRITOMNOST → sociální nepřítomnost
|
||||
|
||||
Spustit na Windows:
|
||||
python "002_detail_NES_NP_tabulky.py"
|
||||
"""
|
||||
import fdb
|
||||
import datetime
|
||||
|
||||
conn = fdb.connect(
|
||||
dsn=r'localhost:c:\medicus 3\data\medicus.fdb',
|
||||
user='SYSDBA',
|
||||
password='masterkey',
|
||||
charset='win1250'
|
||||
)
|
||||
cur = conn.cursor()
|
||||
|
||||
|
||||
def serialize(v):
|
||||
if isinstance(v, (datetime.date, datetime.datetime)):
|
||||
return v.isoformat()
|
||||
if isinstance(v, datetime.time):
|
||||
return v.isoformat()
|
||||
if isinstance(v, bytes):
|
||||
# Pro BLOB zkusíme dekódovat jako text
|
||||
try:
|
||||
return v.decode('cp1250')[:80]
|
||||
except Exception:
|
||||
return f'<blob {len(v)}B>'
|
||||
if v is None:
|
||||
return 'NULL'
|
||||
return str(v)
|
||||
|
||||
|
||||
def inspect_table(table):
|
||||
print(f"\n{'=' * 70}")
|
||||
print(f" TABULKA: {table}")
|
||||
print(f"{'=' * 70}")
|
||||
|
||||
# Sloupce
|
||||
cur.execute("""
|
||||
SELECT TRIM(rf.rdb$field_name),
|
||||
f.rdb$field_type,
|
||||
f.rdb$field_length
|
||||
FROM rdb$relation_fields rf
|
||||
JOIN rdb$fields f ON rf.rdb$field_source = f.rdb$field_name
|
||||
WHERE TRIM(rf.rdb$relation_name) = ?
|
||||
ORDER BY rf.rdb$field_position
|
||||
""", (table,))
|
||||
cols = cur.fetchall()
|
||||
type_map = {
|
||||
7: 'SMALLINT', 8: 'INTEGER', 10: 'FLOAT', 12: 'DATE',
|
||||
13: 'TIME', 14: 'CHAR', 16: 'BIGINT', 27: 'DOUBLE',
|
||||
35: 'TIMESTAMP', 37: 'VARCHAR', 261: 'BLOB',
|
||||
}
|
||||
col_names = [c[0] for c in cols]
|
||||
print("\nSloupce:")
|
||||
for c in cols:
|
||||
name, ftype, flen = c
|
||||
type_str = type_map.get(ftype, f'type#{ftype}')
|
||||
if flen and ftype in (14, 37):
|
||||
type_str += f'({flen})'
|
||||
print(f" {name:<30} {type_str}")
|
||||
|
||||
# Počet řádků
|
||||
cur.execute(f'SELECT COUNT(*) FROM {table}')
|
||||
pocet = cur.fetchone()[0]
|
||||
print(f"\nCelkem řádků: {pocet}")
|
||||
|
||||
if pocet == 0:
|
||||
return col_names
|
||||
|
||||
# Ukázka 5 nejnovějších (pokud má datum sloupec)
|
||||
date_cols = [c[0] for c in cols if c[1] in (12, 35)] # DATE nebo TIMESTAMP
|
||||
order = f'ORDER BY {date_cols[0]} DESC' if date_cols else ''
|
||||
|
||||
try:
|
||||
cur.execute(f'SELECT FIRST 5 * FROM {table} {order}')
|
||||
rows = cur.fetchall()
|
||||
print(f"\nUkázka (5 {'nejnovějších' if order else 'prvních'}):")
|
||||
for row in rows:
|
||||
print()
|
||||
for name, val in zip(col_names, row):
|
||||
s = serialize(val)
|
||||
if s != 'NULL':
|
||||
print(f" {name:<30} {s}")
|
||||
except Exception as e:
|
||||
print(f" Ukázka selhala: {e}")
|
||||
|
||||
return col_names
|
||||
|
||||
|
||||
# ── Prozkoumáme klíčové tabulky ───────────────────────────────────────────────
|
||||
for tbl in ['NES', 'NP', 'SOC_NEPRITOMNOST']:
|
||||
inspect_table(tbl)
|
||||
|
||||
# ── Speciálně: aktivní PN ─────────────────────────────────────────────────────
|
||||
print(f"\n\n{'=' * 70}")
|
||||
print(" POKUS: aktivní záznamy v NES (konec NULL nebo v budoucnosti)")
|
||||
print(f"{'=' * 70}")
|
||||
|
||||
dnes = datetime.date.today()
|
||||
|
||||
# Nejdřív zjistíme sloupce NES
|
||||
cur.execute("""
|
||||
SELECT TRIM(rf.rdb$field_name), f.rdb$field_type
|
||||
FROM rdb$relation_fields rf
|
||||
JOIN rdb$fields f ON rf.rdb$field_source = f.rdb$field_name
|
||||
WHERE TRIM(rf.rdb$relation_name) = 'NES'
|
||||
ORDER BY rf.rdb$field_position
|
||||
""")
|
||||
nes_cols = [(r[0], r[1]) for r in cur.fetchall()]
|
||||
nes_col_names = [c[0] for c in nes_cols]
|
||||
|
||||
print(f"\nVšechny sloupce NES: {', '.join(nes_col_names)}")
|
||||
|
||||
# Hledáme sloupce s datumem konce PN
|
||||
date_cols_nes = [c[0] for c in nes_cols if c[1] in (12, 35)]
|
||||
print(f"Datumové sloupce: {date_cols_nes}")
|
||||
|
||||
# Zkusíme různé kandidátní sloupce pro "datum do"
|
||||
candidates_do = [c for c in nes_col_names if 'DO' in c or 'KONEC' in c or 'END' in c.upper()]
|
||||
candidates_od = [c for c in nes_col_names if 'OD' in c or 'ZACATEK' in c or 'VZNIK' in c or 'START' in c.upper()]
|
||||
candidates_pac = [c for c in nes_col_names if 'PAC' in c or 'IDPAC' in c]
|
||||
print(f"Kandidáti DATUM_DO: {candidates_do}")
|
||||
print(f"Kandidáti DATUM_OD: {candidates_od}")
|
||||
print(f"Kandidáti IDPAC: {candidates_pac}")
|
||||
|
||||
# Ukázka posledních 10 záznamů NES seřazená dle prvního datumového sloupce
|
||||
if date_cols_nes:
|
||||
try:
|
||||
cur.execute(f'SELECT FIRST 10 * FROM NES ORDER BY {date_cols_nes[0]} DESC')
|
||||
rows = cur.fetchall()
|
||||
print(f"\nPosledních 10 záznamů NES (dle {date_cols_nes[0]} DESC):")
|
||||
for row in rows:
|
||||
print()
|
||||
for name, val in zip(nes_col_names, row):
|
||||
s = serialize(val)
|
||||
if s != 'NULL':
|
||||
print(f" {name:<30} {s}")
|
||||
except Exception as e:
|
||||
print(f"Selhalo: {e}")
|
||||
|
||||
# ── NP tabulka – stejný rozbor ────────────────────────────────────────────────
|
||||
print(f"\n\n{'=' * 70}")
|
||||
print(" POKUS: aktivní záznamy v NP")
|
||||
print(f"{'=' * 70}")
|
||||
|
||||
cur.execute("""
|
||||
SELECT TRIM(rf.rdb$field_name), f.rdb$field_type
|
||||
FROM rdb$relation_fields rf
|
||||
JOIN rdb$fields f ON rf.rdb$field_source = f.rdb$field_name
|
||||
WHERE TRIM(rf.rdb$relation_name) = 'NP'
|
||||
ORDER BY rf.rdb$field_position
|
||||
""")
|
||||
np_cols = [(r[0], r[1]) for r in cur.fetchall()]
|
||||
np_col_names = [c[0] for c in np_cols]
|
||||
print(f"\nVšechny sloupce NP: {', '.join(np_col_names)}")
|
||||
date_cols_np = [c[0] for c in np_cols if c[1] in (12, 35)]
|
||||
print(f"Datumové sloupce: {date_cols_np}")
|
||||
|
||||
if date_cols_np:
|
||||
try:
|
||||
cur.execute(f'SELECT FIRST 10 * FROM NP ORDER BY {date_cols_np[0]} DESC')
|
||||
rows = cur.fetchall()
|
||||
print(f"\nPosledních 10 záznamů NP (dle {date_cols_np[0]} DESC):")
|
||||
for row in rows:
|
||||
print()
|
||||
for name, val in zip(np_col_names, row):
|
||||
s = serialize(val)
|
||||
if s != 'NULL':
|
||||
print(f" {name:<30} {s}")
|
||||
except Exception as e:
|
||||
print(f"Selhalo: {e}")
|
||||
|
||||
cur.close()
|
||||
conn.close()
|
||||
print(f"\n{'=' * 70}")
|
||||
print("Hotovo.")
|
||||
Reference in New Issue
Block a user