""" Nacte odpoved lekoveho zaznamu (XML) a ulozi ji do MySQL. Schema: zprava / predpis / predpis_slozka / vydej / vydej_slozka Typy a delky presne dle XSD (Cuer2Schema.xsd + CuerSchema.xsd, verze 202501A) Spusteni: python 06UlozitDoMySQL.py nebo: python 06UlozitDoMySQL.py cesta/k/odpoved.xml """ import sys from pathlib import Path import xml.etree.ElementTree as ET import pymysql import pymysql.cursors # ── konfigurace ─────────────────────────────────────────────────────────────── XML_SOUBOR = Path(__file__).parent / "odpoved_lekovy_zaznam.xml" DB = dict( host = "192.168.1.76", user = "root", password = "Vlado9674+", database = "medicus", charset = "utf8mb4", cursorclass = pymysql.cursors.DictCursor, ) NS = "http://www.sukl.cz/erp/201912" # ───────────────────────────────────────────────────────────────────────────── def t(el, tag): """Prvni potomek s danym tagem → text nebo None.""" found = el.find(f"{{{NS}}}{tag}") return found.text.strip() if found is not None and found.text else None def datum(s): return s[:10] if s else None def ts(s): return s[:19].replace("T", " ") if s else None # ── DDL ─────────────────────────────────────────────────────────────────────── DDL_TABULKY = [ # smazat v opacnem poradi kvuli FK "DROP TABLE IF EXISTS predpis_slozka", "DROP TABLE IF EXISTS vydej_slozka", "DROP TABLE IF EXISTS vydej", "DROP TABLE IF EXISTS predpis", "DROP TABLE IF EXISTS zprava", # ── zprava ──────────────────────────────────────────────────────────────── # zprava_odpoved_type + zprava_type: # ID_Zpravy CHAR(36), Verze, Odeslano dateTime # Aplikace(512), ID_Podani CHAR(36), Prijato dateTime # jmeno_osoby_type: Prijmeni(35), Jmena(24) """ CREATE TABLE zprava ( id INT AUTO_INCREMENT PRIMARY KEY, id_zpravy CHAR(36) NOT NULL UNIQUE, verze VARCHAR(20), odeslano DATETIME, aplikace VARCHAR(512), id_podani CHAR(36), prijato DATETIME, pacient_prijmeni VARCHAR(35), pacient_jmena VARCHAR(24), pacient_datum_narozeni DATE, stazeno DATETIME DEFAULT CURRENT_TIMESTAMP, INDEX idx_pacient (pacient_prijmeni, pacient_datum_narozeni) ) ENGINE=InnoDB """, # ── predpis ─────────────────────────────────────────────────────────────── # lz_nacteni_predepsany_lp_erp_type: # ID_LP_Predpis CHAR(36) NOT NULL, KodPredepisujiciho(36) NOT NULL, # DatumVystaveni date NOT NULL, Mnozstvi int(1-9999) NOT NULL, # Navod(80) NOT NULL, Opakovani int?, ModryPruh boolean? # lek — vzdy jen jeden z: HVLPReg / HVLPNereg / IPLP / INN # hvlp_type: Kod CHAR(7)?, ATC(7)?, Nazev(146)!, Forma(27)?, # Sila(24)?, CestaPodani(15)?, Baleni VARCHAR(22)? # inn_predpis_type: Nazev(200) — nejdelsi nazev mezi typy # iplp_predpis_type: PostupPripravy(4000), Nazev(146), CestaPodani(15), Forma(27) """ CREATE TABLE predpis ( id INT AUTO_INCREMENT PRIMARY KEY, zprava_id INT NOT NULL, id_lp_predpis CHAR(36) NOT NULL UNIQUE, kod_predepisujiciho VARCHAR(36) NOT NULL, datum_vystaveni DATE NOT NULL, mnozstvi SMALLINT NOT NULL, navod VARCHAR(80) NOT NULL, opakovani INT, modry_pruh TINYINT(1), typ_leku ENUM('HVLPReg','HVLPNereg','IPLP','INN'), lek_kod CHAR(7), atc VARCHAR(7), nazev VARCHAR(200), forma VARCHAR(27), sila VARCHAR(24), cesta_podani VARCHAR(15), baleni VARCHAR(22), postup_pripravy VARCHAR(4000), FOREIGN KEY (zprava_id) REFERENCES zprava(id) ON DELETE CASCADE, INDEX idx_atc (atc), INDEX idx_datum (datum_vystaveni) ) ENGINE=InnoDB """, # ── predpis_slozka ──────────────────────────────────────────────────────── # slozka_iplp_predpis_type: # Mnozstvi DECIMAL(15,6) NOT NULL, Jednotka ENUM('g','ks') NOT NULL, # Nazev(200) NOT NULL, Surovina CHAR(7)?, HVLPReg CHAR(7)? """ CREATE TABLE predpis_slozka ( id INT AUTO_INCREMENT PRIMARY KEY, predpis_id INT NOT NULL, mnozstvi DECIMAL(15,6) NOT NULL, jednotka ENUM('g','ks') NOT NULL, nazev VARCHAR(200) NOT NULL, surovina CHAR(7), hvlp_reg CHAR(7), FOREIGN KEY (predpis_id) REFERENCES predpis(id) ON DELETE CASCADE, INDEX idx_nazev (nazev) ) ENGINE=InnoDB """, # ── vydej ───────────────────────────────────────────────────────────────── # lz_nacteni_vydany_lp_erp_type: # Mnozstvi DECIMAL(6,2) NOT NULL, Navod(80) NOT NULL, # Sarze(50) NOT NULL, SerioveCislo(20)?, Pozn(1000)? # lek — vzdy jen jeden z: HVLPReg / HVLPNereg / IPLP # iplp_type (vydej): KodVZP CHAR(7)?, PostupPripravy(4000), Nazev(146), CestaPodani(15) """ CREATE TABLE vydej ( id INT AUTO_INCREMENT PRIMARY KEY, zprava_id INT NOT NULL, id_lp_vydej CHAR(36) NOT NULL UNIQUE, id_lp_predpis CHAR(36), kod_vydavajiciho VARCHAR(36) NOT NULL, datum_vydeje DATE NOT NULL, mnozstvi DECIMAL(6,2) NOT NULL, navod VARCHAR(80) NOT NULL, exspirace DATE, sarze VARCHAR(50) NOT NULL, seriove_cislo VARCHAR(20), pozn VARCHAR(1000), typ_leku ENUM('HVLPReg','HVLPNereg','IPLP'), lek_kod CHAR(7), atc VARCHAR(7), nazev VARCHAR(146), forma VARCHAR(27), sila VARCHAR(24), cesta_podani VARCHAR(15), postup_pripravy VARCHAR(4000), FOREIGN KEY (zprava_id) REFERENCES zprava(id) ON DELETE CASCADE, FOREIGN KEY (id_lp_predpis) REFERENCES predpis(id_lp_predpis) ON DELETE SET NULL, INDEX idx_predpis (id_lp_predpis), INDEX idx_atc (atc), INDEX idx_datum (datum_vydeje) ) ENGINE=InnoDB """, # ── vydej_slozka ────────────────────────────────────────────────────────── # slozka_iplp_type: # Mnozstvi DECIMAL(15,6) NOT NULL, Jednotka ENUM('g','ks') NOT NULL, # Nazev(200) NOT NULL, HrazenoZP DECIMAL(9,2)?, # Surovina CHAR(7)?, HVLPReg CHAR(7)? """ CREATE TABLE vydej_slozka ( id INT AUTO_INCREMENT PRIMARY KEY, vydej_id INT NOT NULL, mnozstvi DECIMAL(15,6) NOT NULL, jednotka ENUM('g','ks') NOT NULL, nazev VARCHAR(200) NOT NULL, hrazeno_zp DECIMAL(9,2), surovina CHAR(7), hvlp_reg CHAR(7), FOREIGN KEY (vydej_id) REFERENCES vydej(id) ON DELETE CASCADE, INDEX idx_nazev (nazev) ) ENGINE=InnoDB """, ] def vytvor_schema(conn): with conn.cursor() as cur: for stmt in DDL_TABULKY: stmt = stmt.strip() if stmt: cur.execute(stmt) conn.commit() print("Schema OK (5 tabulek smazano a vytvoreno znovu)") # ── parsovani leku ──────────────────────────────────────────────────────────── def parsuj_slozky_predpis(lek_el): """Ze IPLP elementu vraci seznam slovniku pro predpis_slozka.""" slozky = [] for s in lek_el.findall(f"{{{NS}}}Slozka"): slozky.append(dict( mnozstvi = t(s, "Mnozstvi"), jednotka = t(s, "Jednotka"), nazev = t(s, "Nazev"), surovina = t(s, "Surovina"), hvlp_reg = t(s, "HVLPReg"), )) return slozky def parsuj_slozky_vydej(lek_el): """Ze IPLP elementu vraci seznam slovniku pro vydej_slozka.""" slozky = [] for s in lek_el.findall(f"{{{NS}}}Slozka"): slozky.append(dict( mnozstvi = t(s, "Mnozstvi"), jednotka = t(s, "Jednotka"), nazev = t(s, "Nazev"), hrazeno_zp = t(s, "HrazenoZP"), surovina = t(s, "Surovina"), hvlp_reg = t(s, "HVLPReg"), )) return slozky def parsuj_lek_predpis(p): """Vraci (lek_fields dict, slozky list). Typ leku: HVLPReg / HVLPNereg / IPLP / INN — vzdy jen jeden. """ for typ in ("HVLPReg", "HVLPNereg"): lek = p.find(f"{{{NS}}}{typ}") if lek is not None: return dict( typ_leku = typ, lek_kod = t(lek, "Kod"), atc = t(lek, "ATC"), nazev = t(lek, "Nazev"), forma = t(lek, "Forma"), sila = t(lek, "Sila"), cesta_podani = t(lek, "CestaPodani"), baleni = t(lek, "Baleni"), postup_pripravy = None, ), [] lek = p.find(f"{{{NS}}}IPLP") if lek is not None: return dict( typ_leku = "IPLP", lek_kod = None, atc = None, nazev = t(lek, "Nazev"), forma = t(lek, "Forma"), sila = None, cesta_podani = t(lek, "CestaPodani"), baleni = None, postup_pripravy = t(lek, "PostupPripravy"), ), parsuj_slozky_predpis(lek) lek = p.find(f"{{{NS}}}INN") if lek is not None: return dict( typ_leku = "INN", lek_kod = None, atc = None, nazev = t(lek, "Nazev"), forma = t(lek, "Forma"), sila = t(lek, "Sila"), cesta_podani = t(lek, "CestaPodani"), baleni = t(lek, "Baleni"), postup_pripravy = None, ), [] return dict(typ_leku=None, lek_kod=None, atc=None, nazev=None, forma=None, sila=None, cesta_podani=None, baleni=None, postup_pripravy=None), [] def parsuj_lek_vydej(v): """Vraci (lek_fields dict, slozky list). Typ leku: HVLPReg / HVLPNereg / IPLP — vzdy jen jeden. """ for typ in ("HVLPReg", "HVLPNereg"): lek = v.find(f"{{{NS}}}{typ}") if lek is not None: return dict( typ_leku = typ, lek_kod = t(lek, "Kod"), atc = t(lek, "ATC"), nazev = t(lek, "Nazev"), forma = t(lek, "Forma"), sila = t(lek, "Sila"), cesta_podani = t(lek, "CestaPodani"), postup_pripravy = None, ), [] lek = v.find(f"{{{NS}}}IPLP") if lek is not None: return dict( typ_leku = "IPLP", lek_kod = t(lek, "KodVZP"), atc = None, nazev = t(lek, "Nazev"), forma = None, sila = None, cesta_podani = t(lek, "CestaPodani"), postup_pripravy = t(lek, "PostupPripravy"), ), parsuj_slozky_vydej(lek) return dict(typ_leku=None, lek_kod=None, atc=None, nazev=None, forma=None, sila=None, cesta_podani=None, postup_pripravy=None), [] # ── parsovani XML ───────────────────────────────────────────────────────────── def parsuj_xml(xml_soubor): tree = ET.parse(xml_soubor) root = tree.getroot() odpoved = root[0][0] # Envelope > Body > NacistLekovyZaznamOdpoved # Doklad prvni, Zprava druha — presne dle XSD sequence doklad = odpoved.find(f"{{{NS}}}Doklad") zpr = odpoved.find(f"{{{NS}}}Zprava") # ── Zprava ─────────────────────────────────────────────────────────────── zprava = dict( id_zpravy = t(zpr, "ID_Zpravy"), verze = t(zpr, "Verze"), odeslano = ts(t(zpr, "Odeslano")), aplikace = t(zpr, "Aplikace"), id_podani = t(zpr, "ID_Podani"), prijato = ts(t(zpr, "Prijato")), ) # ── Pacient ─────────────────────────────────────────────────────────────── pac = doklad.find(f"{{{NS}}}Pacient") jmeno = pac.find(f"{{{NS}}}Jmeno") zprava["pacient_prijmeni"] = t(jmeno, "Prijmeni") if jmeno is not None else None zprava["pacient_jmena"] = t(jmeno, "Jmena") if jmeno is not None else None zprava["pacient_datum_narozeni"] = datum(t(pac, "DatumNarozeni")) # ── Predpisy ────────────────────────────────────────────────────────────── predpisy = [] # kazda polozka: (row_dict, slozky_list) predpis_seznam = doklad.find(f"{{{NS}}}PredpisSeznam") if predpis_seznam is not None: for p in predpis_seznam.findall(f"{{{NS}}}Predpis"): row = dict( id_lp_predpis = t(p, "ID_LP_Predpis"), kod_predepisujiciho = t(p, "KodPredepisujiciho"), datum_vystaveni = datum(t(p, "DatumVystaveni")), mnozstvi = t(p, "Mnozstvi"), navod = t(p, "Navod"), opakovani = t(p, "Opakovani"), modry_pruh = t(p, "ModryPruh"), ) lek_fields, slozky = parsuj_lek_predpis(p) row.update(lek_fields) predpisy.append((row, slozky)) # ── Vydeji ──────────────────────────────────────────────────────────────── vydeji = [] # kazda polozka: (row_dict, slozky_list) vydej_seznam = doklad.find(f"{{{NS}}}VydejSeznam") if vydej_seznam is not None: for v in vydej_seznam.findall(f"{{{NS}}}Vydej"): row = dict( id_lp_vydej = t(v, "ID_LP_Vydej"), id_lp_predpis = t(v, "ID_LP_Predpis"), kod_vydavajiciho = t(v, "KodVydavajiciho"), datum_vydeje = datum(t(v, "DatumVydeje")), mnozstvi = t(v, "Mnozstvi"), navod = t(v, "Navod"), exspirace = datum(t(v, "Exspirace")), sarze = t(v, "Sarze"), seriove_cislo = t(v, "SerioveCislo"), pozn = t(v, "Pozn"), ) lek_fields, slozky = parsuj_lek_vydej(v) row.update(lek_fields) vydeji.append((row, slozky)) return zprava, predpisy, vydeji # ── ulozeni do DB ───────────────────────────────────────────────────────────── def _najdi_id(cur, tabulka, sloupec, hodnota): """Pomocna funkce — vrati id radku dle unikatniho sloupce.""" cur.execute(f"SELECT id FROM {tabulka} WHERE {sloupec} = %s", (hodnota,)) row = cur.fetchone() return row["id"] if row else None def uloz(conn, zprava, predpisy, vydeji): iplp_predpisu = 0 iplp_vydejuu = 0 with conn.cursor() as cur: # ── zprava ──────────────────────────────────────────────────────────── cur.execute(""" INSERT INTO zprava (id_zpravy, verze, odeslano, aplikace, id_podani, prijato, pacient_prijmeni, pacient_jmena, pacient_datum_narozeni) VALUES (%(id_zpravy)s, %(verze)s, %(odeslano)s, %(aplikace)s, %(id_podani)s, %(prijato)s, %(pacient_prijmeni)s, %(pacient_jmena)s, %(pacient_datum_narozeni)s) ON DUPLICATE KEY UPDATE prijato = VALUES(prijato), stazeno = CURRENT_TIMESTAMP """, zprava) zprava_id = _najdi_id(cur, "zprava", "id_zpravy", zprava["id_zpravy"]) print(f" zprava id={zprava_id} ({zprava['id_zpravy']})") # ── predpisy + jejich slozky ─────────────────────────────────────────── vlozeno_p = 0 vlozeno_ps = 0 for row, slozky in predpisy: row["zprava_id"] = zprava_id cur.execute(""" INSERT IGNORE INTO predpis (zprava_id, id_lp_predpis, kod_predepisujiciho, datum_vystaveni, mnozstvi, navod, opakovani, modry_pruh, typ_leku, lek_kod, atc, nazev, forma, sila, cesta_podani, baleni, postup_pripravy) VALUES (%(zprava_id)s, %(id_lp_predpis)s, %(kod_predepisujiciho)s, %(datum_vystaveni)s, %(mnozstvi)s, %(navod)s, %(opakovani)s, %(modry_pruh)s, %(typ_leku)s, %(lek_kod)s, %(atc)s, %(nazev)s, %(forma)s, %(sila)s, %(cesta_podani)s, %(baleni)s, %(postup_pripravy)s) """, row) vlozeno_p += cur.rowcount if slozky: iplp_predpisu += 1 predpis_db_id = cur.lastrowid or _najdi_id(cur, "predpis", "id_lp_predpis", row["id_lp_predpis"]) for s in slozky: s["predpis_id"] = predpis_db_id cur.execute(""" INSERT INTO predpis_slozka (predpis_id, mnozstvi, jednotka, nazev, surovina, hvlp_reg) VALUES (%(predpis_id)s, %(mnozstvi)s, %(jednotka)s, %(nazev)s, %(surovina)s, %(hvlp_reg)s) """, s) vlozeno_ps += 1 print(f" predpisy: {vlozeno_p} novych (celkem {len(predpisy)})") print(f" predpis_slozka: {vlozeno_ps} slozek z {iplp_predpisu} IPLP predpisu") # ── vydeji + jejich slozky ──────────────────────────────────────────── vlozeno_v = 0 vlozeno_vs = 0 for row, slozky in vydeji: row["zprava_id"] = zprava_id cur.execute(""" INSERT IGNORE INTO vydej (zprava_id, id_lp_vydej, id_lp_predpis, kod_vydavajiciho, datum_vydeje, mnozstvi, navod, exspirace, sarze, seriove_cislo, pozn, typ_leku, lek_kod, atc, nazev, forma, sila, cesta_podani, postup_pripravy) VALUES (%(zprava_id)s, %(id_lp_vydej)s, %(id_lp_predpis)s, %(kod_vydavajiciho)s, %(datum_vydeje)s, %(mnozstvi)s, %(navod)s, %(exspirace)s, %(sarze)s, %(seriove_cislo)s, %(pozn)s, %(typ_leku)s, %(lek_kod)s, %(atc)s, %(nazev)s, %(forma)s, %(sila)s, %(cesta_podani)s, %(postup_pripravy)s) """, row) vlozeno_v += cur.rowcount if slozky: iplp_vydejuu += 1 vydej_db_id = cur.lastrowid or _najdi_id(cur, "vydej", "id_lp_vydej", row["id_lp_vydej"]) for s in slozky: s["vydej_id"] = vydej_db_id cur.execute(""" INSERT INTO vydej_slozka (vydej_id, mnozstvi, jednotka, nazev, hrazeno_zp, surovina, hvlp_reg) VALUES (%(vydej_id)s, %(mnozstvi)s, %(jednotka)s, %(nazev)s, %(hrazeno_zp)s, %(surovina)s, %(hvlp_reg)s) """, s) vlozeno_vs += 1 print(f" vydeji: {vlozeno_v} novych (celkem {len(vydeji)})") print(f" vydej_slozka: {vlozeno_vs} slozek z {iplp_vydejuu} IPLP vydejuu") conn.commit() # ── main ────────────────────────────────────────────────────────────────────── def main(): xml = Path(sys.argv[1]) if len(sys.argv) > 1 else XML_SOUBOR print(f"XML: {xml} ({xml.stat().st_size // 1024} KB)") print("Parsovani XML ...") zprava, predpisy, vydeji = parsuj_xml(xml) print(f" -> {len(predpisy)} predpisu, {len(vydeji)} vydejuu") print("Pripojeni k MySQL ...") conn = pymysql.connect(**DB) try: vytvor_schema(conn) print("Ukladani ...") uloz(conn, zprava, predpisy, vydeji) print("Hotovo OK") finally: conn.close() if __name__ == "__main__": main()