#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ seed_tracking.py ================ Naplní MongoDB databázi `ordinace`, kolekci `registrovani_tracking`, výchozím snímkem registrovaných pacientů a jejich OVĚŘENÝM stavem u VZP. Logika "v zakoupeném souboru pacientů": - "Registrovaný v Medicusu" je jen stav v software. - Skutečnost ověřujeme u pojišťovny: kdo je k danému dni registrující praktik (odbornost 001) daného pacienta. * praktik = Buzalková (IČP 09305001) -> v pořádku, v zakoupeném souboru * praktik = někdo jiný / žádný -> NEBYL V ZAKOUPENÉM SOUBORU PACIENTŮ Kategorie (plné podkategorie): OK_BUZALKOVA praktik 001 je Buzalková (IČP 09305001) JINY_PRAKTIK praktik 001 je jiné ZZZ BEZ_PRAKTIKA_VZP pacient má u VZP záznam (jiná odbornost), ale praktika 001 ne BEZ_ZAZNAMU_VZP VZP nevrátila žádný záznam (typicky jiná pojišťovna / neplatné RČ) Schéma dokumentu (1 dokument na pacienta, _id = rodné číslo): { "_id": "8202...", "rc": "...", "prijmeni": "...", "jmeno": "...", "pojistovna": {"kod": "111", "zkratka": "VZP"}, "vychozi_datum": "2025-01-01", "aktualni": { ...snímek... }, "historie": [ { ...snímek..., "zmena": "výchozí snímek" }, ... ], "created_at": ..., "updated_at": ... } Snímek (aktualni i položka historie): { "k_datu", "kategorie", "kategorie_popis", "v_zakoupenem_souboru" (bool), "flag", "praktik_nazev", "praktik_icz", "praktik_icp", "datum_zahajeni", "datum_ukonceni" } Spuštění: python seed_tracking.py # seed k 2025-01-01 python seed_tracking.py 2026-05-02 # aplikuje další snímek (appendne změny do historie) """ import sys from pathlib import Path from datetime import datetime, date, timezone PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent sys.path.insert(0, str(PROJECT_ROOT)) import pymongo from Knihovny.mysql_db import connect_mysql # ── KONFIGURACE ─────────────────────────────────────────────────────────────── MONGO_URI = "mongodb://192.168.1.76:27017" MONGO_DB = "ordinace" MONGO_COLL = "registrovani_tracking" ICP_BUZALKOVA = "09305001" KATEGORIE_POPIS = { "OK_BUZALKOVA": "OK – praktik je Buzalková (IČP 09305001)", "JINY_PRAKTIK": "Registrován u jiného praktika", "BEZ_PRAKTIKA_VZP": "U VZP bez praktika (odb. 001)", "BEZ_ZAZNAMU_VZP": "VZP nevrátila žádný záznam (jiná pojišťovna / neplatné RČ)", } FLAG_MIMO_SOUBOR = "NEBYL V ZAKOUPENÉM SOUBORU PACIENTŮ" def klasifikuj(praktik_001: dict | None, ma_nejaky_zaznam: bool) -> dict: """Vrátí snímek stavu (bez k_datu) na základě 001 záznamu z VZP.""" if praktik_001 and praktik_001.get("ICP") == ICP_BUZALKOVA: kat = "OK_BUZALKOVA" elif praktik_001: kat = "JINY_PRAKTIK" elif ma_nejaky_zaznam: kat = "BEZ_PRAKTIKA_VZP" else: kat = "BEZ_ZAZNAMU_VZP" v_souboru = (kat == "OK_BUZALKOVA") nazev = (praktik_001 or {}).get("nazev_zzz") icz = (praktik_001 or {}).get("ICZ") od = (praktik_001 or {}).get("datum_zahajeni") # Čitelný důvod, proč pacient NENÍ v zakoupeném souboru (kdo + od kdy) if kat == "JINY_PRAKTIK": flag_duvod = f"jiný praktik: {nazev} (IČZ {icz}) od {od}" elif kat == "BEZ_PRAKTIKA_VZP": flag_duvod = "u VZP bez registrujícího praktika (odb. 001)" elif kat == "BEZ_ZAZNAMU_VZP": flag_duvod = "VZP nevrátila žádný záznam (jiná pojišťovna / neplatné RČ / zaniklé pojištění)" else: flag_duvod = "" return { "kategorie": kat, "kategorie_popis": KATEGORIE_POPIS[kat], "v_zakoupenem_souboru": v_souboru, "flag": "" if v_souboru else FLAG_MIMO_SOUBOR, "flag_duvod": flag_duvod, # "kdo" a "od kdy" registrujícího praktika dle VZP "praktik_nazev": nazev, "praktik_icz": icz, "praktik_icp": (praktik_001 or {}).get("ICP"), "praktik_od": od, "datum_zahajeni": od, "datum_ukonceni": (praktik_001 or {}).get("datum_ukonceni"), } def nacti_snimek_z_mysql(mysql, k_datu: str) -> dict: """ Vrátí {rc: {prijmeni, jmeno, pojistovna{}, praktik_001 | None, ma_zaznam}} pro populaci registrovaných dotázaných k danému datu. """ cur = mysql.cursor() # Populace = všechna dotázaná RČ (raw) k tomuto datu cur.execute("SELECT rc FROM vzp_registrace_raw WHERE k_datu = %s", (k_datu,)) populace = [r[0] for r in cur.fetchall()] # Parsované záznamy lékařů k tomuto datu cur.execute(""" SELECT rc, prijmeni, jmeno, kod_odbornosti, ICP, ICZ, nazev_zzz, poj_kod, poj_zkratka, datum_zahajeni, datum_ukonceni FROM vzp_registrace_lekari WHERE k_datu = %s """, (k_datu,)) data: dict[str, dict] = {rc: {"prijmeni": None, "jmeno": None, "pojistovna": {"kod": None, "zkratka": None}, "praktik_001": None, "ma_zaznam": False} for rc in populace} for (rc, prijmeni, jmeno, odb, icp, icz, nazev_zzz, poj_kod, poj_zkr, dat_zah, dat_uk) in cur.fetchall(): d = data.setdefault(rc, {"prijmeni": None, "jmeno": None, "pojistovna": {"kod": None, "zkratka": None}, "praktik_001": None, "ma_zaznam": False}) d["ma_zaznam"] = True if prijmeni and not d["prijmeni"]: d["prijmeni"] = prijmeni if jmeno and not d["jmeno"]: d["jmeno"] = jmeno # Pojišťovnu vezmi z jakéhokoli záznamu (preferuj 001 níže) if poj_kod and not d["pojistovna"]["kod"]: d["pojistovna"] = {"kod": poj_kod, "zkratka": poj_zkr} if odb == "001": d["praktik_001"] = { "ICP": icp, "ICZ": icz, "nazev_zzz": nazev_zzz, "poj_kod": poj_kod, "poj_zkratka": poj_zkr, "datum_zahajeni": str(dat_zah) if dat_zah else None, "datum_ukonceni": str(dat_uk) if dat_uk else None, } # Pojišťovna z 001 má přednost if poj_kod: d["pojistovna"] = {"kod": poj_kod, "zkratka": poj_zkr} return data def apply_snapshot(coll, mysql, k_datu: str) -> dict: """ Klasifikuje populaci k danému datu a upsertne do Mongo. Při změně kategorie/praktika oproti `aktualni` appendne do `historie`. Vrátí statistiku. """ data = nacti_snimek_z_mysql(mysql, k_datu) now = datetime.now(timezone.utc) stats = {"novych": 0, "zmen": 0, "beze_zmeny": 0, "kategorie": {}} for rc, d in data.items(): snimek = klasifikuj(d["praktik_001"], d["ma_zaznam"]) snimek_s_datem = {"k_datu": k_datu, **snimek} stats["kategorie"][snimek["kategorie"]] = stats["kategorie"].get(snimek["kategorie"], 0) + 1 existing = coll.find_one({"_id": rc}) if existing is None: doc = { "_id": rc, "rc": rc, "prijmeni": d["prijmeni"], "jmeno": d["jmeno"], "pojistovna": d["pojistovna"], "vychozi_datum": k_datu, "aktualni": snimek_s_datem, "historie": [{**snimek_s_datem, "zmena": "výchozí snímek"}], "created_at": now, "updated_at": now, } coll.insert_one(doc) stats["novych"] += 1 else: akt = existing.get("aktualni", {}) zmena = (akt.get("kategorie") != snimek["kategorie"] or akt.get("praktik_icp") != snimek["praktik_icp"]) update = {"aktualni": snimek_s_datem, "updated_at": now} if d["prijmeni"]: update["prijmeni"] = d["prijmeni"] if d["jmeno"]: update["jmeno"] = d["jmeno"] ops = {"$set": update} if zmena: popis = (f"{akt.get('kategorie')} → {snimek['kategorie']}") ops["$push"] = {"historie": {**snimek_s_datem, "zmena": popis}} stats["zmen"] += 1 else: stats["beze_zmeny"] += 1 coll.update_one({"_id": rc}, ops) return stats def main(): k_datu = sys.argv[1] if len(sys.argv) > 1 else "2025-01-01" mysql = connect_mysql() client = pymongo.MongoClient(MONGO_URI, serverSelectionTimeoutMS=3000) client.admin.command("ping") coll = client[MONGO_DB][MONGO_COLL] # Indexy pro běžné dotazy coll.create_index("aktualni.kategorie") coll.create_index("aktualni.v_zakoupenem_souboru") coll.create_index("prijmeni") print(f"Aplikuji snímek k {k_datu} do {MONGO_DB}.{MONGO_COLL} ...") stats = apply_snapshot(coll, mysql, k_datu) print(f"\nNových pacientů : {stats['novych']}") print(f"Změn stavu : {stats['zmen']}") print(f"Beze změny : {stats['beze_zmeny']}") print("\nRozpad podle kategorií:") for kat, n in sorted(stats["kategorie"].items(), key=lambda x: -x[1]): print(f" {kat:18s} {n:5d} {KATEGORIE_POPIS[kat]}") celkem = sum(stats["kategorie"].values()) mimo = celkem - stats["kategorie"].get("OK_BUZALKOVA", 0) print(f"\nCelkem v populaci: {celkem}") print(f" v zakoupeném souboru (Buzalková): {stats['kategorie'].get('OK_BUZALKOVA', 0)}") print(f" NEBYL v zakoupeném souboru : {mimo}") mysql.close() client.close() if __name__ == "__main__": main()