""" Nacte odpoved lekoveho zaznamu (XML) a ulozi ji do MySQL. Schema: pacient / zprava / predpis / predpis_slozka / vydej / vydej_slozka / predepisujici / vydavajici Typy a delky presne dle XSD (Cuer2Schema.xsd + CuerSchema.xsd, verze 202501A) Spusteni (jednorazova inicializace + import jednoho XML): python 06UlozitDoMySQL.py nebo: python 06UlozitDoMySQL.py cesta/k/odpoved.xml Pro hromadne stazeni vsech pacientu pouzij 07StahnoutVsechny.py. """ 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 ─────────────────────────────────────────────────────────────────────── # Pouziva se ve vytvor_schema() (DROP + CREATE) pro ciste spusteni DDL_DROP = [ "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", "DROP TABLE IF EXISTS pacient", "DROP TABLE IF EXISTS predepisujici", "DROP TABLE IF EXISTS vydavajici", ] DDL_CREATE = [ # ── pacient ─────────────────────────────────────────────────────────────── # Zrcadlo registrovanych pacientu z Medicusu (Firebird). # idpac = IDPAC z KAR tabulky Medicusu. # poznamka: posledni chyba API (napr. "neztotozneny pacient"); NULL = OK """ CREATE TABLE IF NOT EXISTS pacient ( id INT AUTO_INCREMENT PRIMARY KEY, idpac INT NOT NULL UNIQUE, prijmeni VARCHAR(35) NOT NULL, jmena VARCHAR(24), datum_narozeni DATE NOT NULL, aktivni TINYINT(1) NOT NULL DEFAULT 1, poznamka VARCHAR(500), INDEX idx_prijmeni (prijmeni) ) ENGINE=InnoDB """, # ── 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) # pacient_id: FK na pacient.id (NULL pokud volano z 06 primo) # xml_soubor: relativni cesta k ulozene XML odpovedi """ CREATE TABLE IF NOT EXISTS zprava ( id INT AUTO_INCREMENT PRIMARY KEY, id_zpravy CHAR(36) NOT NULL UNIQUE, pacient_id INT, 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, xml_soubor VARCHAR(255), stazeno DATETIME DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (pacient_id) REFERENCES pacient(id) ON DELETE SET NULL, INDEX idx_pacient_id (pacient_id), 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 IF NOT EXISTS 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 IF NOT EXISTS 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 IF NOT EXISTS 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 IF NOT EXISTS 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 """, # ── predepisujici ───────────────────────────────────────────────────────── # PredepisujiciSeznam > Predepisujici: # Lekar: Kod CHAR(36), Jmeno: Prijmeni(35), Jmena(24) # ICZ CHAR(8)?, ICP CHAR(8)? # PZS: Nazev(200), Adresa: NazevUlice, CisloPopisne, CisloOrientacni, NazevObce, PSC # Telefon(20)? # lekar_kod = predpis.kod_predepisujiciho """ CREATE TABLE IF NOT EXISTS predepisujici ( id INT AUTO_INCREMENT PRIMARY KEY, lekar_kod CHAR(36) NOT NULL UNIQUE, prijmeni VARCHAR(35), jmena VARCHAR(24), icz CHAR(8), icp CHAR(8), pzs_nazev VARCHAR(200), ulice VARCHAR(150), mesto VARCHAR(100), psc CHAR(5), telefon VARCHAR(20), INDEX idx_icp (icp), INDEX idx_icz (icz) ) ENGINE=InnoDB """, # ── vydavajici ──────────────────────────────────────────────────────────── # VydavajiciSeznam > Vydavajici: # Lekarnik: Kod CHAR(36), Jmeno: Prijmeni(35), Jmena(24) # PZS: Nazev(200), Telefon(20)?, Adresa: NazevUlice, CisloPopisne, CisloOrientacni, NazevObce, PSC # lekarnik_kod = vydej.kod_vydavajiciho """ CREATE TABLE IF NOT EXISTS vydavajici ( id INT AUTO_INCREMENT PRIMARY KEY, lekarnik_kod CHAR(36) NOT NULL UNIQUE, prijmeni VARCHAR(35), jmena VARCHAR(24), pzs_nazev VARCHAR(200), ulice VARCHAR(150), mesto VARCHAR(100), psc CHAR(5), telefon VARCHAR(20) ) ENGINE=InnoDB """, ] def vytvor_schema(conn): """DROP + CREATE vsech tabulek. Pouzij pro ciste spusteni / reset dat.""" with conn.cursor() as cur: for stmt in DDL_DROP: cur.execute(stmt) for stmt in DDL_CREATE: stmt = stmt.strip() if stmt: cur.execute(stmt) conn.commit() print("Schema OK — tabulky smazany a vytvoreny znovu") def inicializuj_schema(conn): """CREATE TABLE IF NOT EXISTS — bezpecne pro opakowane spusteni (neznici data).""" with conn.cursor() as cur: for stmt in DDL_CREATE: stmt = stmt.strip() if stmt: cur.execute(stmt) # Zpetna kompatibilita: pridat sloupec poznamka pokud jeste neexistuje cur.execute(""" SELECT COUNT(*) AS cnt FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'pacient' AND COLUMN_NAME = 'poznamka' """) if cur.fetchone()["cnt"] == 0: cur.execute( "ALTER TABLE pacient ADD COLUMN poznamka VARCHAR(500) NULL DEFAULT NULL" ) conn.commit() # ── parsovani predepisujicich a vydavajicich ───────────────────────────────── def _parsuj_adresu(pzs_el): """Ze elementu PZS vraci (ulice, mesto, psc).""" adr = pzs_el.find(f"{{{NS}}}Adresa") if pzs_el is not None else None if adr is None: return None, None, None ulice_parts = [ t(adr, "NazevUlice") or "", t(adr, "CisloPopisne") or "", t(adr, "CisloOrientacni") or "", ] ulice = " ".join(p for p in ulice_parts if p).strip() or None psc = t(adr, "PSC") if psc and len(psc) > 5: psc = psc[:5] return ulice, t(adr, "NazevObce"), psc def parsuj_predepisujici(doklad): """Vraci seznam slovniku pro tabulku predepisujici.""" seznam = [] sez_el = doklad.find(f"{{{NS}}}PredepisujiciSeznam") if sez_el is None: return seznam for el in sez_el.findall(f"{{{NS}}}Predepisujici"): lekar = el.find(f"{{{NS}}}Lekar") if lekar is None: continue jmeno = lekar.find(f"{{{NS}}}Jmeno") pzs = el.find(f"{{{NS}}}PZS") ulice, mesto, psc = _parsuj_adresu(pzs) seznam.append(dict( lekar_kod = t(lekar, "Kod"), prijmeni = t(jmeno, "Prijmeni") if jmeno is not None else None, jmena = t(jmeno, "Jmena") if jmeno is not None else None, icz = t(el, "ICZ"), icp = t(el, "ICP"), pzs_nazev = t(pzs, "Nazev") if pzs is not None else None, ulice = ulice, mesto = mesto, psc = psc, telefon = t(el, "Telefon"), )) return seznam def parsuj_vydavajici(doklad): """Vraci seznam slovniku pro tabulku vydavajici.""" seznam = [] sez_el = doklad.find(f"{{{NS}}}VydavajiciSeznam") if sez_el is None: return seznam for el in sez_el.findall(f"{{{NS}}}Vydavajici"): lekarnik = el.find(f"{{{NS}}}Lekarnik") if lekarnik is None: continue jmeno = lekarnik.find(f"{{{NS}}}Jmeno") pzs = el.find(f"{{{NS}}}PZS") ulice, mesto, psc = _parsuj_adresu(pzs) seznam.append(dict( lekarnik_kod = t(lekarnik, "Kod"), prijmeni = t(jmeno, "Prijmeni") if jmeno is not None else None, jmena = t(jmeno, "Jmena") if jmeno is not None else None, pzs_nazev = t(pzs, "Nazev") if pzs is not None else None, ulice = ulice, mesto = mesto, psc = psc, telefon = t(pzs, "Telefon") if pzs is not None else None, )) return seznam # ── 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)) predepisujici = parsuj_predepisujici(doklad) vydavajici = parsuj_vydavajici(doklad) return zprava, predpisy, vydeji, predepisujici, vydavajici # ── 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, predepisujici, vydavajici, pacient_id=None, xml_soubor=None): """ Ulozi parsovana data do MySQL. pacient_id — FK na tabulku pacient (None pokud volano primo z 06) xml_soubor — relativni cesta k archivnimu XML souboru (None pokud neni archivovano) Vraci dict se statistikami: predpisy_novych, predpisy_celkem, vydeji_novych, vydeji_celkem, predpis_slozka, vydej_slozka """ iplp_predpisu = 0 iplp_vydejuu = 0 with conn.cursor() as cur: # ── zprava ──────────────────────────────────────────────────────────── zprava_row = dict(zprava) zprava_row["pacient_id"] = pacient_id zprava_row["xml_soubor"] = xml_soubor cur.execute(""" INSERT INTO zprava (id_zpravy, pacient_id, verze, odeslano, aplikace, id_podani, prijato, pacient_prijmeni, pacient_jmena, pacient_datum_narozeni, xml_soubor) VALUES (%(id_zpravy)s, %(pacient_id)s, %(verze)s, %(odeslano)s, %(aplikace)s, %(id_podani)s, %(prijato)s, %(pacient_prijmeni)s, %(pacient_jmena)s, %(pacient_datum_narozeni)s, %(xml_soubor)s) ON DUPLICATE KEY UPDATE prijato = VALUES(prijato), xml_soubor = COALESCE(VALUES(xml_soubor), xml_soubor), stazeno = CURRENT_TIMESTAMP """, zprava_row) zprava_id = _najdi_id(cur, "zprava", "id_zpravy", 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 # ── 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 # ── predepisujici ───────────────────────────────────────────────────── for row in predepisujici: if not row.get("lekar_kod"): continue cur.execute(""" INSERT INTO predepisujici (lekar_kod, prijmeni, jmena, icz, icp, pzs_nazev, ulice, mesto, psc, telefon) VALUES (%(lekar_kod)s, %(prijmeni)s, %(jmena)s, %(icz)s, %(icp)s, %(pzs_nazev)s, %(ulice)s, %(mesto)s, %(psc)s, %(telefon)s) ON DUPLICATE KEY UPDATE prijmeni = VALUES(prijmeni), jmena = VALUES(jmena), icz = VALUES(icz), icp = VALUES(icp), pzs_nazev = VALUES(pzs_nazev), ulice = VALUES(ulice), mesto = VALUES(mesto), psc = VALUES(psc), telefon = VALUES(telefon) """, row) # ── vydavajici ──────────────────────────────────────────────────────── for row in vydavajici: if not row.get("lekarnik_kod"): continue cur.execute(""" INSERT INTO vydavajici (lekarnik_kod, prijmeni, jmena, pzs_nazev, ulice, mesto, psc, telefon) VALUES (%(lekarnik_kod)s, %(prijmeni)s, %(jmena)s, %(pzs_nazev)s, %(ulice)s, %(mesto)s, %(psc)s, %(telefon)s) ON DUPLICATE KEY UPDATE prijmeni = VALUES(prijmeni), jmena = VALUES(jmena), pzs_nazev = VALUES(pzs_nazev), ulice = VALUES(ulice), mesto = VALUES(mesto), psc = VALUES(psc), telefon = VALUES(telefon) """, row) conn.commit() return dict( predpisy_novych = vlozeno_p, predpisy_celkem = len(predpisy), vydeji_novych = vlozeno_v, vydeji_celkem = len(vydeji), predpis_slozka = vlozeno_ps, vydej_slozka = vlozeno_vs, ) # ── 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, predepisujici, vydavajici = parsuj_xml(xml) print(f" -> {len(predpisy)} predpisu, {len(vydeji)} vydejuu, " f"{len(predepisujici)} predepisujicich, {len(vydavajici)} vydavajicich") print("Pripojeni k MySQL ...") conn = pymysql.connect(**DB) try: vytvor_schema(conn) print("Ukladani ...") stats = uloz(conn, zprava, predpisy, vydeji, predepisujici, vydavajici) print(f" predpisy: {stats['predpisy_novych']} novych (celkem {stats['predpisy_celkem']})") print(f" vydeji: {stats['vydeji_novych']} novych (celkem {stats['vydeji_celkem']})") print(f" slozky: {stats['predpis_slozka']} predpis / {stats['vydej_slozka']} vydej") print("Hotovo OK") finally: conn.close() if __name__ == "__main__": main()