From 73e5eab4a401ecd002ba85648a0b089b63522ea9 Mon Sep 17 00:00:00 2001 From: Vladimir Buzalka Date: Mon, 26 Jan 2026 19:46:49 +0100 Subject: [PATCH] notebookVB --- 05 Kontrola pojištěnců/10 Čti OZP.py | 180 ++++++++++++++ 05 Kontrola pojištěnců/20 Kapdetais.py | 47 ++++ .../30 Porovnani kapitace.py | 234 ++++++++++++++++++ knihovny/medicus_db.py | 15 ++ 4 files changed, 476 insertions(+) create mode 100644 05 Kontrola pojištěnců/10 Čti OZP.py create mode 100644 05 Kontrola pojištěnců/20 Kapdetais.py create mode 100644 05 Kontrola pojištěnců/30 Porovnani kapitace.py diff --git a/05 Kontrola pojištěnců/10 Čti OZP.py b/05 Kontrola pojištěnců/10 Čti OZP.py new file mode 100644 index 0000000..4048906 --- /dev/null +++ b/05 Kontrola pojištěnců/10 Čti OZP.py @@ -0,0 +1,180 @@ +import os +from pathlib import Path +from blake3 import blake3 +import pymysql +from datetime import date + +# ========================= +# VÝVOJOVÝ PŘEPÍNAČ +# ========================= + +RESET_DB = True # !!! POZOR: smaže kapitace a pojištěnce !!! + + +# ========================= +# KONFIGURACE +# ========================= + +BASE_PATH = Path( + r"U:\Dropbox\Ordinace\Dokumentace_ke_zpracování\Výpis pojištěnců" +) + +DB_CONFIG = { + "host": "192.168.1.76", + "port": 3307, + "user": "root", + "password": "Vlado9674+", + "database": "ordinace", + "charset": "utf8mb4", + "autocommit": False, +} + +# ========================= +# DB POMOCNÉ FUNKCE +# ========================= +def reset_kapitace_tables(conn): + print("!!! RESET_DB=True – mažu data kapitace !!!") + + with conn.cursor() as cur: + # kvůli FK + cur.execute("SET FOREIGN_KEY_CHECKS = 0") + + cur.execute("TRUNCATE TABLE zp_kapitace_pojistenec") + cur.execute("TRUNCATE TABLE zp_kapitace_header") + + cur.execute("SET FOREIGN_KEY_CHECKS = 1") + + conn.commit() + print("✔ Kapitace resetována") + +def get_conn(): + return pymysql.connect(**DB_CONFIG) + +def blake_exists(cur, blake): + cur.execute( + "SELECT id FROM zp_kapitace_header WHERE file_blake3 = %s", + (blake,), + ) + return cur.fetchone() + +# ========================= +# PARSERY +# ========================= + +def parse_header(line: str): + return { + "icp_lekar": line[1:9].strip(), + "pocet_pojistencu": int(line[9:14]), + "rok": 2000 + int(line[14:16]), + "mesic": int(line[16:18]), + "den": int(line[18:20]), + } + +def parse_pojistenec(line: str): + return { + "poradi_radku": int(line[1:5]), + "vekova_skupina": int(line[5:7]), + "prijmeni": line[7:37].strip(), + "jmeno": line[37:61].strip(), + "cislo_pojistence": line[61:71].strip(), + "kapitace_od": date( + int(line[75:79]), + int(line[73:75]), + int(line[71:73]), + ), + "zp_kod": line[79:82].strip(), + } + +# ========================= +# HLAVNÍ IMPORT +# ========================= + +def import_file(path: Path): + raw = path.read_bytes() + blake = blake3(raw).hexdigest() + + text = raw.decode("cp852") + lines = text.splitlines() + + with get_conn() as conn: + cur = conn.cursor() + + if blake_exists(cur, blake): + print(f"SKIP {path.name} (už existuje)") + return + + # ---- HEADER ---- + h = parse_header(lines[0]) + snapshot_date = date(h["rok"], h["mesic"], h["den"]) + + cur.execute( + """ + INSERT INTO zp_kapitace_header + (source_file, zp_kod, icp_lekar, pocet_pojistencu, + rok, mesic, den, snapshot_date, + file_blake3, file_content, file_size, file_lines) + VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) + """, + ( + path.name, + path.name[1:4], # xxx ze jména Fxxx... + h["icp_lekar"], + h["pocet_pojistencu"], + h["rok"], + h["mesic"], + h["den"], + snapshot_date, + blake, + text, + len(raw), + len(lines), + ), + ) + + header_id = cur.lastrowid + + # ---- POJISTENCI ---- + for line in lines[1:]: + if not line.startswith("I"): + continue + + p = parse_pojistenec(line) + + cur.execute( + """ + INSERT INTO zp_kapitace_pojistenec + (header_id, poradi_radku, vekova_skupina, + prijmeni, jmeno, cislo_pojistence, + kapitace_od, zp_kod) + VALUES (%s,%s,%s,%s,%s,%s,%s,%s) + """, + ( + header_id, + p["poradi_radku"], + p["vekova_skupina"], + p["prijmeni"], + p["jmeno"], + p["cislo_pojistence"], + p["kapitace_od"], + p["zp_kod"], + ), + ) + + conn.commit() + print(f"IMPORTED {path.name}") + +# ========================= +# RUN +# ========================= + +def main(): + with get_conn() as conn: + if RESET_DB: + reset_kapitace_tables(conn) + + for f in sorted(BASE_PATH.glob("F*.???")): + import_file(f) + + +if __name__ == "__main__": + main() diff --git a/05 Kontrola pojištěnců/20 Kapdetais.py b/05 Kontrola pojištěnců/20 Kapdetais.py new file mode 100644 index 0000000..ee7f3b8 --- /dev/null +++ b/05 Kontrola pojištěnců/20 Kapdetais.py @@ -0,0 +1,47 @@ +from knihovny.medicus_db import MedicusDB # nebo odkud třídu importuješ + + +def main(): + poj = input("Zadej kód pojišťovny (např. 207): ").strip() + + db = MedicusDB( + host="192.168.1.4", + db_path=r"z:\medicus 3\data\medicus.fdb" + ) + + sql = """ + SELECT + fak.id, + fak.cisfak, + fak.poj, + fak.kapdetail + FROM fak + WHERE fak.poj = ? + AND fak.kapdetail IS NOT NULL + AND fak.kapdetail <> '' + ORDER BY fak.cisfak DESC + ROWS 1 + """ + + rows = db.query_dict(sql, (poj,)) + + if not rows: + print(f"❌ Nenalezena žádná faktura s kapdetails pro pojišťovnu {poj}") + return + + r = rows[0] + + print("=" * 80) + print(f"FAKTURA ID : {r['id']}") + print(f"ČÍSLO FAKTURY: {r['cisfak']}") + print(f"POJIŠŤOVNA : {r['poj']}") + print("-" * 80) + print("KAPDETAILS:") + print(r["kapdetail"]) + print("=" * 80) + + db.close() + + +if __name__ == "__main__": + main() diff --git a/05 Kontrola pojištěnců/30 Porovnani kapitace.py b/05 Kontrola pojištěnců/30 Porovnani kapitace.py new file mode 100644 index 0000000..bfd97db --- /dev/null +++ b/05 Kontrola pojištěnců/30 Porovnani kapitace.py @@ -0,0 +1,234 @@ +import re +from pathlib import Path +from typing import List, Dict, Set + +import pymysql +from knihovny.medicus_db import MedicusDB + + +# ============================================================ +# KONFIGURACE +# ============================================================ + +# --- Firebird / Medicus --- +FIREBIRD_HOST = "192.168.1.4" +FIREBIRD_DB = r"Z:\medicus 3\data\medicus.fdb" +FIREBIRD_USER = "SYSDBA" +FIREBIRD_PASS = "masterkey" +FIREBIRD_CHARSET = "WIN1250" + +# --- MySQL / Kapitace --- +MYSQL_CFG = { + "host": "192.168.1.76", + "port": 3307, + "user": "root", + "password": "Vlado9674+", + "database": "ordinace", + "charset": "utf8mb4", + "autocommit": True, +} + +MYSQL_TABLE_HEADER = "zp_kapitace_header" +MYSQL_TABLE_PAC = "zp_kapitace_pojistenec" + + +# ============================================================ +# HELPERY +# ============================================================ + +_rc_re = re.compile(r"\D+") + + +def normalize_rc(value: str) -> str: + """Rodné číslo / číslo pojištěnce → jen číslice.""" + if not value: + return "" + return _rc_re.sub("", str(value)) + + +def parse_kapdetail_idpacs(kapdetail: str) -> List[int]: + """ + KAPDETAIL: + =SOUHRN;idpac;vek|castka;idpac;vek|castka;... + + Vrací seznam idpac. + """ + if not kapdetail: + return [] + + s = kapdetail.strip() + + if "=" in s: + s = s.split("=", 1)[1] + + parts = [p.strip() for p in s.split(";") if p.strip()] + + if not parts: + return [] + + # první položka je souhrn (není idpac) + if not parts[0].isdigit(): + parts = parts[1:] + + idpacs: List[int] = [] + + # idpac je vždy na sudém indexu (0,2,4,...) + for i in range(0, len(parts), 2): + if parts[i].isdigit(): + idpacs.append(int(parts[i])) + + return idpacs + + +def chunk_list(lst: List[int], size: int = 500) -> List[List[int]]: + return [lst[i:i + size] for i in range(0, len(lst), size)] + + +# ============================================================ +# FIREBIRD – DATA Z MEDICUSU +# ============================================================ + +def get_latest_fak_for_poj(db: MedicusDB, poj: str) -> Dict: + sql = """ + SELECT + fak.id, + fak.cisfak, + fak.poj, + fak.kapdetail + FROM fak + WHERE fak.poj = ? + AND fak.kapdetail IS NOT NULL + AND fak.kapdetail <> '' + ORDER BY fak.cisfak DESC + ROWS 1 + """ + rows = db.query_dict(sql, (poj,)) + return rows[0] if rows else {} + + +def get_rodcis_for_idpacs(db: MedicusDB, idpacs: List[int]) -> Dict[int, str]: + """ + Vrátí mapu: idpac → rodcis + """ + result: Dict[int, str] = {} + + if not idpacs: + return result + + for batch in chunk_list(idpacs): + placeholders = ",".join("?" for _ in batch) + sql = f""" + SELECT + kar.idpac, + kar.rodcis + FROM kar + WHERE kar.idpac IN ({placeholders}) + AND kar.rodcis IS NOT NULL + AND kar.rodcis <> '' + AND (kar.vyrazen IS NULL OR kar.vyrazen <> 'A') + """ + rows = db.query(sql, tuple(batch)) + for idpac, rodcis in rows: + result[int(idpac)] = str(rodcis) + + return result + + +# ============================================================ +# MYSQL – KAPITACE ZP +# ============================================================ + +def get_mysql_kapitace_rc(zp_kod: str) -> Set[str]: + sql = f""" + SELECT p.cislo_pojistence + FROM {MYSQL_TABLE_PAC} p + JOIN {MYSQL_TABLE_HEADER} h ON h.id = p.header_id + WHERE h.zp_kod = %s + AND h.snapshot_date = ( + SELECT MAX(h2.snapshot_date) + FROM {MYSQL_TABLE_HEADER} h2 + WHERE h2.zp_kod = %s + ) + """ + + with pymysql.connect(**MYSQL_CFG) as conn: + with conn.cursor() as cur: + cur.execute(sql, (zp_kod, zp_kod)) + rows = cur.fetchall() + + return { + normalize_rc(r[0]) + for r in rows + if normalize_rc(r[0]) + } + + +# ============================================================ +# MAIN +# ============================================================ + +def main(): + poj = input("Zadej kód pojišťovny (např. 207): ").strip() + + # --- Medicus --- + mdb = MedicusDB( + host=FIREBIRD_HOST, + db_path=FIREBIRD_DB, + user=FIREBIRD_USER, + password=FIREBIRD_PASS, + charset=FIREBIRD_CHARSET, + ) + + fak = get_latest_fak_for_poj(mdb, poj) + if not fak: + print(f"❌ Nenalezena žádná kapitační faktura pro pojišťovnu {poj}") + mdb.close() + return + + print("\nNalezena faktura:") + print(f" ID : {fak['id']}") + print(f" CISFAK : {fak['cisfak']}") + print(f" POJ : {fak['poj']}") + + idpacs = parse_kapdetail_idpacs(fak["kapdetail"]) + print(f"\nKAPDETAIL → idpac celkem: {len(idpacs)}") + + idpac_to_rc = get_rodcis_for_idpacs(mdb, idpacs) + mdb.close() + + medicus_rc = { + normalize_rc(rc) + for rc in idpac_to_rc.values() + if normalize_rc(rc) + } + + print(f"Medicus (KAR.RODCIS): {len(medicus_rc)}") + + # --- MySQL --- + mysql_rc = get_mysql_kapitace_rc(poj) + print(f"MySQL kapitace ZP: {len(mysql_rc)}") + + # --- Porovnání --- + both = medicus_rc & mysql_rc + only_medicus = medicus_rc - mysql_rc + only_mysql = mysql_rc - medicus_rc + + print("\n" + "=" * 80) + print(f"✅ V obou (OK): {len(both)}") + print(f"❌ Jen v Medicusu: {len(only_medicus)}") + print(f"❌ Jen v MySQL (kapitace): {len(only_mysql)}") + print("=" * 80) + + if only_medicus: + print("\n❌ Jen v Medicusu (fakturováno, ale není v kapitaci ZP):") + for rc in sorted(only_medicus): + print(" ", rc) + + if only_mysql: + print("\n❌ Jen v MySQL (v kapitaci ZP, ale není ve faktuře Medicusu):") + for rc in sorted(only_mysql): + print(" ", rc) + + +if __name__ == "__main__": + main() diff --git a/knihovny/medicus_db.py b/knihovny/medicus_db.py index 45a5cd7..8674e6e 100644 --- a/knihovny/medicus_db.py +++ b/knihovny/medicus_db.py @@ -13,6 +13,21 @@ class MedicusDB: ) self.cur = self.conn.cursor() + def get_fak_kapitace(self, as_dict=False): + sql = """ + SELECT + fak.id, + fak.cisfak, + fak.poj, + fak.kapdetail + FROM fak + WHERE fak.kapdetail IS NOT NULL + AND fak.kapdetail <> '' + """ + if as_dict: + return self.query_dict(sql) + return self.query(sql) + def query(self, sql, params=None): self.cur.execute(sql, params or ()) return self.cur.fetchall()