Vladimir Buzalka
303 lines
12 KiB
Python
303 lines
12 KiB
Python
"""
|
|
Naparsuje XML soubory z xml_archive a uloží data do MySQL tabulek
|
|
recept_doklad a recept_plp.
|
|
|
|
Pro každý ERP kód zpracuje NEJNOVĚJŠÍ XML soubor (nejvyšší datum v archivu).
|
|
Opakované spuštění je bezpečné — používá UPSERT a INSERT IGNORE.
|
|
|
|
Spuštění:
|
|
python 11_ParseXML.py # celý archiv
|
|
python 11_ParseXML.py --datum 2026-04-14 # jen konkrétní den
|
|
"""
|
|
|
|
import sys
|
|
import argparse
|
|
from datetime import date
|
|
from pathlib import Path
|
|
import xml.etree.ElementTree as ET
|
|
|
|
import pymysql
|
|
import pymysql.cursors
|
|
|
|
if hasattr(sys.stdout, "reconfigure"):
|
|
sys.stdout.reconfigure(errors="replace")
|
|
|
|
# ── Konfigurace ───────────────────────────────────────────────────────────────
|
|
DB = dict(
|
|
host = "192.168.1.76",
|
|
user = "root",
|
|
password = "Vlado9674+",
|
|
database = "medicus",
|
|
charset = "utf8mb4",
|
|
cursorclass = pymysql.cursors.DictCursor,
|
|
)
|
|
|
|
XML_DIR = Path(__file__).parent / "xml_archive"
|
|
NS = "http://www.sukl.cz/erp/201704"
|
|
|
|
# ── Parametry spuštění (uprav zde, nebo nech None = celý archiv) ──────────────
|
|
DATUM_FILTR = None # např. "2026-04-14", nebo None = celý archiv
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def t(el, tag):
|
|
"""Vrátí text prvního potomka s daným tagem, nebo None."""
|
|
found = el.find(f"{{{NS}}}{tag}")
|
|
return found.text.strip() if found is not None and found.text else None
|
|
|
|
|
|
def ts(s):
|
|
"""ISO datetime → MySQL DATETIME string."""
|
|
return s[:19].replace("T", " ") if s else None
|
|
|
|
|
|
def bool_el(el, tag):
|
|
return 1 if t(el, tag) == "true" else 0
|
|
|
|
|
|
def parsuj_xml(xml_text, xml_soubor):
|
|
"""
|
|
Naparsuje XML odpověď NacistPredpis.
|
|
Vrátí (doklad_dict, [plp_dict, ...]) nebo None při chybě.
|
|
"""
|
|
try:
|
|
root = ET.fromstring(xml_text)
|
|
doklad = root.find(f".//{{{NS}}}Doklad")
|
|
if doklad is None:
|
|
return None
|
|
except ET.ParseError:
|
|
return None
|
|
|
|
# ── doklad ────────────────────────────────────────────────────────────────
|
|
id_dokladu = t(doklad, "ID_Dokladu")
|
|
stav = t(doklad, "Stav")
|
|
datum_vystaveni = t(doklad, "DatumVystaveni")
|
|
platnost_do_s = t(doklad, "PlatnostDo")
|
|
vypis_do = t(doklad, "VypisDo")
|
|
akutni = bool_el(doklad, "Akutni")
|
|
rodina = bool_el(doklad, "Rodina")
|
|
opakovani_s = t(doklad, "Opakovani")
|
|
druh_pojisteni = t(doklad, "DruhPojisteni")
|
|
modry_pruh = bool_el(doklad, "ModryPruh")
|
|
pozn = t(doklad, "Pozn")
|
|
zap_s = t(doklad, "ZapocitatelnyDoplatekZbyvaDoLimitu")
|
|
zmena = ts(t(doklad, "Zmena"))
|
|
zalozeni = ts(t(doklad, "Zalozeni"))
|
|
|
|
# stav_terminal
|
|
platnost_date = date.fromisoformat(platnost_do_s) if platnost_do_s else None
|
|
expiroval = platnost_date is not None and platnost_date < date.today()
|
|
stav_terminal = 1 if stav in ("PLNE_VYDANY", "ZRUSENY") or expiroval else 0
|
|
|
|
# ── pacient ───────────────────────────────────────────────────────────────
|
|
pac = doklad.find(f"{{{NS}}}Pacient")
|
|
zp_el = pac.find(f"{{{NS}}}ZP") if pac is not None else None
|
|
cp = t(pac, "CP") if pac is not None else None
|
|
zp_kod = t(zp_el, "Kod") if zp_el is not None else None
|
|
zp_nazev = t(zp_el, "Nazev") if zp_el is not None else None
|
|
pac_telefon = t(pac, "Telefon") if pac is not None else None
|
|
pac_notif = t(pac, "Notifikace") if pac is not None else None
|
|
pac_pohlavi = t(pac, "Pohlavi") if pac is not None else None
|
|
|
|
# ── predepisujici ─────────────────────────────────────────────────────────
|
|
pred = doklad.find(f"{{{NS}}}Predepisujici")
|
|
lekar_el = pred.find(f"{{{NS}}}Lekar") if pred is not None else None
|
|
odb_el = pred.find(f"{{{NS}}}Odbornost") if pred is not None else None
|
|
lekar_kod_raw = t(lekar_el, "Kod") if lekar_el is not None else None
|
|
# "skryto" není UUID → uložíme NULL
|
|
lekar_kod = lekar_kod_raw if lekar_kod_raw and lekar_kod_raw.lower() != "skryto" else None
|
|
odb_kod = t(odb_el, "Kod") if odb_el is not None else None
|
|
odb_nazev = t(odb_el, "Nazev") if odb_el is not None else None
|
|
lekar_email = t(pred, "Email") if pred is not None else None
|
|
|
|
doklad_dict = dict(
|
|
id_dokladu = id_dokladu,
|
|
stav = stav,
|
|
stav_terminal = stav_terminal,
|
|
datum_vystaveni = datum_vystaveni,
|
|
platnost_do = platnost_do_s,
|
|
vypis_do = vypis_do,
|
|
akutni = akutni,
|
|
rodina = rodina,
|
|
opakovani = int(opakovani_s) if opakovani_s else None,
|
|
druh_pojisteni = druh_pojisteni,
|
|
modry_pruh = modry_pruh,
|
|
pozn = pozn,
|
|
zap_doplatek = float(zap_s) if zap_s else None,
|
|
zmena = zmena,
|
|
zalozeni = zalozeni,
|
|
lekar_kod = lekar_kod,
|
|
odbornost_kod = odb_kod,
|
|
odbornost_nazev = odb_nazev,
|
|
lekar_email = lekar_email,
|
|
cp = cp,
|
|
zp_kod = zp_kod,
|
|
zp_nazev = zp_nazev,
|
|
pac_telefon = pac_telefon,
|
|
pac_notifikace = pac_notif,
|
|
pac_pohlavi = pac_pohlavi,
|
|
xml_soubor = str(xml_soubor),
|
|
)
|
|
|
|
# ── PLP položky ───────────────────────────────────────────────────────────
|
|
plp_list = []
|
|
for plp_el in doklad.findall(f"{{{NS}}}PLP"):
|
|
id_lp = t(plp_el, "ID_LP")
|
|
uhrada = t(plp_el, "Uhrada")
|
|
prekroceni = 1 if t(plp_el, "Prekroceni") == "true" else 0
|
|
if id_lp:
|
|
plp_list.append(dict(
|
|
id_lp = id_lp,
|
|
id_dokladu = id_dokladu,
|
|
uhrada = uhrada,
|
|
prekroceni = prekroceni,
|
|
))
|
|
|
|
return doklad_dict, plp_list
|
|
|
|
|
|
def uloz(conn, doklad, plp_list):
|
|
"""UPSERT dokladu a INSERT IGNORE PLP položek."""
|
|
with conn.cursor() as cur:
|
|
# recept_doklad — ON DUPLICATE KEY UPDATE (stav se může změnit)
|
|
cur.execute("""
|
|
INSERT INTO recept_doklad
|
|
(id_dokladu, stav, stav_terminal, datum_vystaveni, platnost_do,
|
|
vypis_do, akutni, rodina, opakovani, druh_pojisteni, modry_pruh,
|
|
pozn, zap_doplatek, zmena, zalozeni,
|
|
lekar_kod, odbornost_kod, odbornost_nazev, lekar_email,
|
|
cp, zp_kod, zp_nazev, pac_telefon, pac_notifikace, pac_pohlavi,
|
|
xml_soubor, stazeno)
|
|
VALUES
|
|
(%(id_dokladu)s, %(stav)s, %(stav_terminal)s, %(datum_vystaveni)s, %(platnost_do)s,
|
|
%(vypis_do)s, %(akutni)s, %(rodina)s, %(opakovani)s, %(druh_pojisteni)s, %(modry_pruh)s,
|
|
%(pozn)s, %(zap_doplatek)s, %(zmena)s, %(zalozeni)s,
|
|
%(lekar_kod)s, %(odbornost_kod)s, %(odbornost_nazev)s, %(lekar_email)s,
|
|
%(cp)s, %(zp_kod)s, %(zp_nazev)s, %(pac_telefon)s, %(pac_notifikace)s, %(pac_pohlavi)s,
|
|
%(xml_soubor)s, NOW())
|
|
ON DUPLICATE KEY UPDATE
|
|
stav = VALUES(stav),
|
|
stav_terminal = VALUES(stav_terminal),
|
|
platnost_do = VALUES(platnost_do),
|
|
druh_pojisteni = VALUES(druh_pojisteni),
|
|
pozn = VALUES(pozn),
|
|
zap_doplatek = VALUES(zap_doplatek),
|
|
zmena = VALUES(zmena),
|
|
zp_kod = VALUES(zp_kod),
|
|
zp_nazev = VALUES(zp_nazev),
|
|
pac_telefon = VALUES(pac_telefon),
|
|
pac_notifikace = VALUES(pac_notifikace),
|
|
xml_soubor = VALUES(xml_soubor),
|
|
stazeno = NOW()
|
|
""", doklad)
|
|
|
|
# recept_plp — INSERT IGNORE (UUID je stabilní, nemění se)
|
|
for plp in plp_list:
|
|
cur.execute("""
|
|
INSERT IGNORE INTO recept_plp (id_lp, id_dokladu, uhrada, prekroceni)
|
|
VALUES (%(id_lp)s, %(id_dokladu)s, %(uhrada)s, %(prekroceni)s)
|
|
""", plp)
|
|
|
|
conn.commit()
|
|
|
|
|
|
def najdi_nejnovejsi_xml(datum_filtr=None):
|
|
"""
|
|
Projde xml_archive, vrátí dict {erp_kod: Path} s nejnovějším XML pro každý kód.
|
|
Přeskočí soubory končící _CHYBA.xml.
|
|
datum_filtr: pokud zadáno (YYYY-MM-DD), zpracuje jen daný den.
|
|
"""
|
|
nejnovejsi = {}
|
|
|
|
if datum_filtr:
|
|
slozky = [XML_DIR / datum_filtr]
|
|
else:
|
|
slozky = sorted(XML_DIR.iterdir()) # seřazeno dle názvu = chronologicky
|
|
|
|
for slozka in slozky:
|
|
if not slozka.is_dir():
|
|
continue
|
|
for xml_file in slozka.glob("*.xml"):
|
|
if xml_file.stem.endswith("_CHYBA"):
|
|
continue
|
|
erp_kod = xml_file.stem
|
|
# pozdější složka přepíše dřívější → nejnovější vyhraje
|
|
nejnovejsi[erp_kod] = xml_file
|
|
|
|
return nejnovejsi
|
|
|
|
|
|
def nacti_zpracovane(conn):
|
|
"""Vrátí dict {id_dokladu: xml_soubor} pro všechny již zpracované záznamy."""
|
|
with conn.cursor() as cur:
|
|
cur.execute("SELECT id_dokladu, xml_soubor FROM recept_doklad WHERE xml_soubor IS NOT NULL")
|
|
return {row["id_dokladu"]: row["xml_soubor"] for row in cur.fetchall()}
|
|
|
|
|
|
def main():
|
|
datum_filtr = DATUM_FILTR
|
|
|
|
xml_mapa = najdi_nejnovejsi_xml(datum_filtr)
|
|
celkem = len(xml_mapa)
|
|
print(f"Nalezeno {celkem} XML souborů v archivu\n")
|
|
|
|
if not celkem:
|
|
print("Žádné soubory.")
|
|
return
|
|
|
|
conn = pymysql.connect(**DB)
|
|
|
|
# Jednorázové opravy schématu (bezpečné opakovat)
|
|
with conn.cursor() as cur:
|
|
# uhrada: původní ENUM neobsahoval PACIENT
|
|
cur.execute("ALTER TABLE recept_plp MODIFY uhrada VARCHAR(20)")
|
|
# pac_pohlavi: XML posílá 'Ž' (ne 'Z')
|
|
cur.execute("ALTER TABLE recept_doklad MODIFY pac_pohlavi VARCHAR(5)")
|
|
# lekar_kod FK: predepisujici se plní z lékového záznamu, ne z detailu
|
|
# → FK by blokoval vložení, zrušíme ji
|
|
try:
|
|
cur.execute("ALTER TABLE recept_doklad DROP FOREIGN KEY recept_doklad_ibfk_1")
|
|
except Exception:
|
|
pass # FK už byl zrušen dříve
|
|
conn.commit()
|
|
|
|
# Načti již zpracované soubory — přeskočíme ty, jejichž cesta se nezměnila
|
|
zpracovane = nacti_zpracovane(conn)
|
|
|
|
ok = chyb = preskoceno = 0
|
|
|
|
for i, (erp_kod, xml_file) in enumerate(xml_mapa.items(), 1):
|
|
rel_path = str(xml_file.relative_to(XML_DIR))
|
|
|
|
if zpracovane.get(erp_kod) == rel_path:
|
|
preskoceno += 1
|
|
continue
|
|
|
|
print(f"[{i:4d}/{celkem}] {erp_kod} ", end="", flush=True)
|
|
|
|
xml_text = xml_file.read_text(encoding="utf-8")
|
|
vysledek = parsuj_xml(xml_text, xml_file.relative_to(XML_DIR))
|
|
|
|
if vysledek is None:
|
|
print("CHYBA parsování")
|
|
chyb += 1
|
|
continue
|
|
|
|
doklad, plp_list = vysledek
|
|
try:
|
|
uloz(conn, doklad, plp_list)
|
|
print(f"{doklad['stav']:20s} terminal={doklad['stav_terminal']} PLP={len(plp_list)}")
|
|
ok += 1
|
|
except Exception as e:
|
|
print(f"CHYBA DB {e}")
|
|
conn.rollback()
|
|
chyb += 1
|
|
|
|
conn.close()
|
|
print(f"\nHotovo: {ok} zpracováno, {preskoceno} přeskočeno (beze změny), {chyb} chyb")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|