Vladimir Buzalka
244 lines
9.6 KiB
Python
244 lines
9.6 KiB
Python
#!/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()
|