diff --git a/Recepty/LékovýZáznamWithClaude/07StahnoutVsechny.py b/Recepty/LékovýZáznamWithClaude/07StahnoutVsechny.py index 9a6b11e..f3fcb6f 100644 --- a/Recepty/LékovýZáznamWithClaude/07StahnoutVsechny.py +++ b/Recepty/LékovýZáznamWithClaude/07StahnoutVsechny.py @@ -37,11 +37,12 @@ from datetime import datetime, timezone, date from pathlib import Path from xml.sax.saxutils import escape as xml_escape -import fdb -import pymysql import pymysql.cursors from requests import Session from requests_pkcs12 import Pkcs12Adapter +from Knihovny.medicus_db import get_medicus_connection +from Knihovny.mysql_db import connect_mysql +from Knihovny.najdi_dropbox import get_dropbox_root # Windows konzole — nahrad neunikatni znaky misto padu if hasattr(sys.stdout, "reconfigure"): @@ -72,26 +73,11 @@ POCET_MESICU_MAX = 60 PAUZA_MIN = 10 # sekund PAUZA_MAX = 20 # sekund -# ── Konfigurace Firebird ────────────────────────────────────────────────────── -FB_DSN = r'localhost:c:\medicus 3\data\medicus.fdb' -FB_USER = 'SYSDBA' -FB_PASS = 'masterkey' -FB_CHARSET = 'win1250' -ICP = '09305001' -ODB = '001' - -# ── Konfigurace MySQL ───────────────────────────────────────────────────────── -DB = dict( - host = "192.168.1.76", - user = "root", - password = "Vlado9674+", - database = "medicus", - charset = "utf8mb4", - cursorclass = pymysql.cursors.DictCursor, -) +ICP = '09305001' +ODB = '001' # ── Adresare ────────────────────────────────────────────────────────────────── -XML_DIR = Path(__file__).parent / "xml_archive" +XML_DIR = Path(get_dropbox_root()) / "Ordinace" / "Dokumentace_ke_zpracování" / "Zúčtovací zprávy" / "LékovýZáznamWithClaude" / "xml_archive" LOGS_DIR = Path(__file__).parent / "Logs" @@ -159,7 +145,7 @@ _SQL_FILTR = """ def nacti_pacienty(prijmeni_filtr=None): - conn = fdb.connect(dsn=FB_DSN, user=FB_USER, password=FB_PASS, charset=FB_CHARSET) + conn = get_medicus_connection() try: cur = conn.cursor() if prijmeni_filtr: @@ -334,7 +320,7 @@ def main(): log.info("Zadni pacienti — konec.") return - conn = pymysql.connect(**DB) + conn = connect_mysql(database="medicus", cursorclass=pymysql.cursors.DictCursor) inicializuj_schema(conn) log.debug("MySQL schema OK") diff --git a/Recepty/LékovýZáznamWithClaude/08TýdenníAktualizaceLékovéhoZáznamu.py b/Recepty/LékovýZáznamWithClaude/08TýdenníAktualizaceLékovéhoZáznamu.py new file mode 100644 index 0000000..202571e --- /dev/null +++ b/Recepty/LékovýZáznamWithClaude/08TýdenníAktualizaceLékovéhoZáznamu.py @@ -0,0 +1,466 @@ +""" +Týdenní aktualizace lékových záznamů všech registrovaných pacientů z eRecept SÚKL. + +Logika počtu měsíců: + - nový pacient (žádná zpráva v DB) → 60 měsíců (maximum) + - pacient s chybou Z002 (poznamka) → 60 měsíců (zkusit znovu, možná se ztotožní) + - pacient s předchozím stažením → ceil(dny od posledního stažení / 30) + 1 + (jednoměsíční překryv, INSERT IGNORE zajistí bez duplikátů) + +Spuštění: + python 08TýdenníAktualizaceLékovéhoZáznamu.py + python 08TýdenníAktualizaceLékovéhoZáznamu.py --prijmeni Buzalka,Buzalkova +""" + +import argparse +import importlib.util +import logging +import math +import random +import sys +import time +import uuid +from datetime import datetime, timezone, date +from pathlib import Path +from xml.sax.saxutils import escape as xml_escape + +import html +import pymysql.cursors +from requests import Session +from requests_pkcs12 import Pkcs12Adapter +from Knihovny.EmailMessagingGraph import send_mail +from Knihovny.medicus_db import get_medicus_connection +from Knihovny.mysql_db import connect_mysql +from Knihovny.najdi_dropbox import get_dropbox_root + +if hasattr(sys.stdout, "reconfigure"): + sys.stdout.reconfigure(errors="replace") + +# ── Import parsovaci logiky z 06 ────────────────────────────────────────────── +_spec = importlib.util.spec_from_file_location( + "m06", Path(__file__).parent / "06UlozitDoMySQL.py" +) +_m06 = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(_m06) + +parsuj_xml = _m06.parsuj_xml +uloz = _m06.uloz +inicializuj_schema = _m06.inicializuj_schema + +# ── Konfigurace eRecept ─────────────────────────────────────────────────────── +PFX_FILE = r"C:\Users\vlado\PycharmProjects\Recepty\AMBSUKL214235369G_31DEC2024.pfx" +PFX_PASS = "Vlado7309208104++" +API_USER = "e08c89c6-2b1a-4eba-8ed9-4e3e63618379" +API_PASS = "Buzalka@Vladimir2025" +UZIVATEL = "E08C89C6-2B1A-4EBA-8ED9-4E3E63618379" +PRACOVISTE = "00214235367" +ENDPOINT = "https://lekar-soap.erecept.sukl.cz/cuer/Lekar2" + +POCET_ZNAKU_ATC = 7 +POCET_MESICU_MAX = 60 +PAUZA_MIN = 2 +PAUZA_MAX = 4 + +VYBRAT_NAHODNE = True # True = testovací běh (10 náhodných pacientů), False = všichni + +EMAIL_PRIJEMCE = "vladimir.buzalka@buzalka.cz" + +ICP = "09305001" +ODB = "001" + +# ── Adresáře ────────────────────────────────────────────────────────────────── +XML_DIR = Path(get_dropbox_root()) / "Ordinace" / "Dokumentace_ke_zpracování" / "Zúčtovací zprávy" / "LékovýZáznamWithClaude" / "xml_archive" +LOGS_DIR = Path(__file__).parent / "Logs" + + +# ── Logging ─────────────────────────────────────────────────────────────────── + +def setup_logging(dnes_str, cas_str): + LOGS_DIR.mkdir(exist_ok=True) + log_soubor = LOGS_DIR / f"{dnes_str}_{cas_str}.log" + + log = logging.getLogger("lz08") + log.setLevel(logging.DEBUG) + + fh = logging.FileHandler(log_soubor, encoding="utf-8") + fh.setLevel(logging.DEBUG) + fh.setFormatter(logging.Formatter("%(asctime)s %(message)s", datefmt="%H:%M:%S")) + + ch = logging.StreamHandler(sys.stdout) + ch.setLevel(logging.INFO) + ch.setFormatter(logging.Formatter("%(message)s")) + + log.addHandler(fh) + log.addHandler(ch) + return log, log_soubor + + +# ── Firebird: načtení registrovaných pacientů ───────────────────────────────── + +_SQL_VSICHNI = """ + SELECT + KAR.IDPAC, + KAR.PRIJMENI, + KAR.JMENO, + KAR.DATNAR + FROM KAR + WHERE (vyrazen = 'N') + AND EXISTS ( + SELECT id FROM registr r + JOIN icp i ON r.idicp = i.idicp + WHERE r.idpac = kar.idpac + AND (r.datum <= ?) + AND (r.datum_zruseni IS NULL OR r.datum_zruseni >= ?) + AND (r.priznak IN ('V','D','A')) + AND (i.icp = ?) + AND (i.odb = ?) + ) + ORDER BY KAR.PRIJMENI_UP, KAR.RODCIS +""" + +_SQL_FILTR = """ + SELECT + KAR.IDPAC, + KAR.PRIJMENI, + KAR.JMENO, + KAR.DATNAR + FROM KAR + WHERE (vyrazen = 'N') + AND KAR.PRIJMENI IN ({ph}) + ORDER BY KAR.PRIJMENI_UP, KAR.RODCIS +""" + + +def nacti_pacienty(prijmeni_filtr=None): + conn = get_medicus_connection() + try: + cur = conn.cursor() + if prijmeni_filtr: + ph = ",".join("?" * len(prijmeni_filtr)) + cur.execute(_SQL_FILTR.format(ph=ph), prijmeni_filtr) + else: + dnes = date.today().isoformat() + cur.execute(_SQL_VSICHNI, (dnes, dnes, ICP, ODB)) + cols = [d[0].lower() for d in cur.description] + return [dict(zip(cols, row)) for row in cur.fetchall()] + finally: + conn.close() + + +# ── MySQL ───────────────────────────────────────────────────────────────────── + +def upsert_pacient(cur, pac): + cur.execute(""" + INSERT INTO pacient (idpac, prijmeni, jmena, datum_narozeni) + VALUES (%s, %s, %s, %s) + ON DUPLICATE KEY UPDATE + prijmeni = VALUES(prijmeni), + jmena = VALUES(jmena) + """, (pac["idpac"], pac["prijmeni"], pac["jmeno"], pac["datnar"])) + cur.execute("SELECT id, poznamka FROM pacient WHERE idpac = %s", (pac["idpac"],)) + return cur.fetchone() + + +def posledni_stazeni(cur, pacient_id): + cur.execute( + "SELECT MAX(stazeno) AS posledni FROM zprava WHERE pacient_id = %s", + (pacient_id,) + ) + row = cur.fetchone() + return row["posledni"] if row and row["posledni"] else None + + +def vypocti_pocet_mesicu(posledni, ma_chybu): + """60 měsíců pro nové pacienty a ty s chybou Z002, jinak diferenciálně.""" + if ma_chybu or posledni is None: + return POCET_MESICU_MAX + delta_dni = (datetime.now() - posledni).days + return min(math.ceil(delta_dni / 30) + 1, POCET_MESICU_MAX) + + +def nacti_chybove_idpac(conn): + """Vrátí množinu idpac pacientů, kteří mají nastavenou poznamku (Z002 apod.).""" + with conn.cursor() as cur: + cur.execute("SELECT idpac FROM pacient WHERE poznamka IS NOT NULL") + return {row["idpac"] for row in cur.fetchall()} + + +def uloz_poznamku(conn, pacient_id, poznamka): + with conn.cursor() as cur: + cur.execute( + "UPDATE pacient SET poznamka = %s WHERE id = %s", + (poznamka, pacient_id) + ) + conn.commit() + + +# ── SOAP volání ─────────────────────────────────────────────────────────────── + +def extrahuj_soap_fault(xml_text): + try: + import xml.etree.ElementTree as ET + NS_SOAP = "http://schemas.xmlsoap.org/soap/envelope/" + NS_SUKL = "http://www.sukl.cz/erp/201912" + root = ET.fromstring(xml_text) + body = root.find(f"{{{NS_SOAP}}}Body") + if body is None: + return "Chybejici SOAP Body" + fault = body.find(f"{{{NS_SOAP}}}Fault") or body.find("Fault") + if fault is not None: + faultstring = (fault.findtext("faultstring") + or fault.findtext("faultcode") + or "Nezname SOAP Fault") + detail = fault.find("detail") + if detail is not None and detail.text: + faultstring = f"{faultstring}: {detail.text.strip()[:200]}" + return faultstring + if body.find(f"{{{NS_SUKL}}}NacistLekovyZaznamOdpoved") is None: + first = list(body) + tag = first[0].tag if first else "prazdne Body" + return f"Neocekavana odpoved: {tag}" + return None + except Exception as e: + return f"Chyba pri parsovani odpovedi: {e}" + + +def nacti_lekovy_zaznam(sess, prijmeni, jmena, datum_narozeni, pocet_mesicu): + id_zpravy = str(uuid.uuid4()) + odeslano = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S+00:00") + soap = ( + '' + '' + '' + f'' + f'' + f'' + f'{UZIVATEL}' + f'{PRACOVISTE}' + f'' + f'{POCET_ZNAKU_ATC}' + f'{pocet_mesicu}' + f'' + f'{xml_escape(prijmeni)}' + f'{xml_escape(jmena)}' + f'{datum_narozeni}' + f'' + f'' + f'' + f'{id_zpravy}' + f'202501A' + f'{odeslano}' + f'MEDICUS_____' + f'' + f'' + '' + '' + ) + headers = { + "Content-Type": 'text/xml; charset="UTF-8"', + "SOAPAction": '"NacistLekovyZaznam"', + "User-Agent": "Medicus", + } + resp = sess.post(ENDPOINT, data=soap.encode("utf-8"), headers=headers, timeout=60) + if resp.status_code != 200: + raise RuntimeError(f"HTTP {resp.status_code}: {resp.text[:300]}") + return resp.text + + +# ── XML archiv ──────────────────────────────────────────────────────────────── + +def uloz_xml_na_disk(xml_text, prijmeni, jmena, datnar_str, dnes_str): + adr = XML_DIR / dnes_str + adr.mkdir(parents=True, exist_ok=True) + nazev = f"{prijmeni}_{jmena}_{datnar_str}.xml".replace(" ", "_") + soubor = adr / nazev + soubor.write_text(xml_text, encoding="utf-8") + return str(soubor) + + +# ── Email ───────────────────────────────────────────────────────────────────── + +def sestav_email(dnes_str, radky, ok, chyby, celkem, testovaci, kriticka_chyba=None): + ma_chybu = chyby > 0 or kriticka_chyba + prefix = "[TEST] " if testovaci else "" + predmet = (f"{prefix}Lékový záznam {dnes_str} — " + f"{'⚠ CHYBA' if ma_chybu else 'OK'} " + f"({ok} OK, {chyby} chyb, {celkem} celkem)") + + css = "font-family:Arial,sans-serif;font-size:14px;color:#222" + th = "padding:4px 10px;background:#f0f0f0;text-align:left;border-bottom:1px solid #ccc" + td = "padding:3px 10px;border-bottom:1px solid #eee" + + def badge(stav): + barva = "#d4edda" if stav == "OK" else "#fdecea" + return f"{stav}" + + radky_html = "".join( + f"" + f"{html.escape(r['jmeno'])}" + f"{badge(r['stav'])}" + f"{html.escape(r['detail'])}" + f"" + for r in radky + ) + + krit = "" + if kriticka_chyba: + krit = ( + f"
" + f"Kritická chyba:" + f"
{kriticka_chyba}
" + ) + + body = ( + f"
" + f"

Lékový záznam — týdenní souhrn {dnes_str}

" + + (f"

Testovací běh — náhodný vzorek {celkem} pacientů

" if testovaci else "") + + krit + + f"

{ok} OK  |  {chyby} chyb  |  {celkem} celkem

" + f"" + f"" + f"{radky_html}" + f"
PacientStavDetail
" + f"
" + ) + + return predmet, body + + +# ── Hlavní smyčka ───────────────────────────────────────────────────────────── + +def main(): + ap = argparse.ArgumentParser(description="Týdenní aktualizace lékových záznamů z eReceptu") + ap.add_argument("--prijmeni", default=None, + help="Filtr příjmení oddělená čárkou (pro testování), např: Buzalka,Buzalkova") + args = ap.parse_args() + + dnes_str = date.today().isoformat() + cas_str = datetime.now().strftime("%H-%M-%S") + log, log_soubor = setup_logging(dnes_str, cas_str) + + prijmeni_filtr = None + if args.prijmeni: + prijmeni_filtr = [p.strip() for p in args.prijmeni.split(",")] + log.info(f"Filtr příjmení: {prijmeni_filtr}") + + log.info("Načítám pacienty z Medicusu...") + pacienti = nacti_pacienty(prijmeni_filtr) + + if VYBRAT_NAHODNE and not prijmeni_filtr: + import random as _random + pacienti = _random.sample(pacienti, min(10, len(pacienti))) + log.info(f"TESTOVACÍ BĚH — náhodný vzorek {len(pacienti)} pacientů") + + celkem = len(pacienti) + log.info(f"Pacientů ke zpracování: {celkem} | log: {log_soubor.name}") + + if not celkem: + log.info("Žádní pacienti — konec.") + return + + conn = connect_mysql(database="medicus", cursorclass=pymysql.cursors.DictCursor) + inicializuj_schema(conn) + log.debug("MySQL schema OK") + + # Seřadit: nejdřív pacienti bez chyby, Z002 na konec + chybove_idpac = nacti_chybove_idpac(conn) + pacienti.sort(key=lambda p: 1 if p["idpac"] in chybove_idpac else 0) + z002_pocet = len(chybove_idpac & {p["idpac"] for p in pacienti}) + log.info(f"Pořadí: {celkem - z002_pocet} bez chyby, pak {z002_pocet} Z002") + + sess = Session() + sess.mount("https://", Pkcs12Adapter(pkcs12_filename=PFX_FILE, pkcs12_password=PFX_PASS)) + sess.auth = (API_USER, API_PASS) + + ok = chyby = 0 + email_radky = [] + kriticka_chyba = None + + try: + for i, pac in enumerate(pacienti, 1): + prijmeni = pac["prijmeni"] + jmena = pac["jmeno"] + datnar = pac["datnar"] + datnar_str = datnar.isoformat() if hasattr(datnar, "isoformat") else str(datnar) + jmeno_str = f"{prijmeni} {jmena}" + + with conn.cursor() as cur: + row = upsert_pacient(cur, pac) + pacient_id = row["id"] + ma_chybu = row["poznamka"] is not None + posledni = posledni_stazeni(cur, pacient_id) + conn.commit() + + pocet_mesicu = vypocti_pocet_mesicu(posledni, ma_chybu) + log.debug(f"[{i:4}/{celkem}] {jmeno_str} (*{datnar_str}) " + f"{pocet_mesicu}m " + f"({'Z002' if ma_chybu else posledni.strftime('%Y-%m-%d') if posledni else 'nový'})") + + try: + xml_text = nacti_lekovy_zaznam(sess, prijmeni, jmena, datnar_str, pocet_mesicu) + except Exception as e: + zprava_chyby = str(e)[:400] + log.info(f"[{i:4}/{celkem}] {jmeno_str:<30} CHYBA {zprava_chyby[:60]}") + uloz_poznamku(conn, pacient_id, zprava_chyby) + email_radky.append({"jmeno": jmeno_str, "stav": "CHYBA", "detail": zprava_chyby[:100]}) + chyby += 1 + continue + + soap_fault = extrahuj_soap_fault(xml_text) + if soap_fault: + log.info(f"[{i:4}/{celkem}] {jmeno_str:<30} CHYBA {soap_fault[:60]}") + uloz_poznamku(conn, pacient_id, soap_fault[:400]) + email_radky.append({"jmeno": jmeno_str, "stav": "CHYBA", "detail": soap_fault[:100]}) + chyby += 1 + continue + + xml_soubor = uloz_xml_na_disk(xml_text, prijmeni, jmena, datnar_str, dnes_str) + kb = Path(xml_soubor).stat().st_size // 1024 + + try: + zprava_d, predpisy, vydeji, predepisujici, vydavajici = parsuj_xml(Path(xml_soubor)) + stats = uloz(conn, zprava_d, predpisy, vydeji, predepisujici, vydavajici, + pacient_id=pacient_id, xml_soubor=xml_soubor) + uloz_poznamku(conn, pacient_id, None) + detail = f"{stats['predpisy_celkem']}p {stats['vydeji_celkem']}v {kb} KB" + log.info(f"[{i:4}/{celkem}] {jmeno_str:<30} OK {detail}") + email_radky.append({"jmeno": jmeno_str, "stav": "OK", "detail": detail}) + ok += 1 + except Exception as e: + zprava_chyby = str(e)[:400] + log.info(f"[{i:4}/{celkem}] {jmeno_str:<30} CHYBA {zprava_chyby[:60]}") + uloz_poznamku(conn, pacient_id, zprava_chyby) + email_radky.append({"jmeno": jmeno_str, "stav": "CHYBA", "detail": zprava_chyby[:100]}) + chyby += 1 + + if i < celkem: + pauza = random.randint(PAUZA_MIN, PAUZA_MAX) + log.debug(f" čekám {pauza}s ...") + time.sleep(pauza) + + except Exception: + import traceback + kriticka_chyba = traceback.format_exc() + log.error(f"KRITICKÁ CHYBA:\n{kriticka_chyba}") + + finally: + conn.close() + sess.close() + + log.info("=" * 55) + log.info(f"Hotovo: {ok} OK | {chyby} chyb | celkem {celkem} pacientů") + + # ── Email se souhrnem a logem jako přílohou — odesílá se vždy ──────────── + try: + predmet, body = sestav_email(dnes_str, email_radky, ok, chyby, celkem, + VYBRAT_NAHODNE, kriticka_chyba) + send_mail(to=EMAIL_PRIJEMCE, subject=predmet, body=body, html=True, + attachments=log_soubor) + log.info(f"Email odeslán: {EMAIL_PRIJEMCE}") + except Exception as e: + log.warning(f"CHYBA odeslání emailu: {e}") + + +if __name__ == "__main__": + main() diff --git a/Recepty/LékovýZáznamWithClaude/reimport_z_xml.py b/Recepty/LékovýZáznamWithClaude/reimport_z_xml.py index 82168ad..fdb84dd 100644 --- a/Recepty/LékovýZáznamWithClaude/reimport_z_xml.py +++ b/Recepty/LékovýZáznamWithClaude/reimport_z_xml.py @@ -11,32 +11,17 @@ import sys import importlib.util from pathlib import Path from datetime import date -import fdb -import pymysql import pymysql.cursors +from Knihovny.medicus_db import get_medicus_connection +from Knihovny.mysql_db import connect_mysql +from Knihovny.najdi_dropbox import get_dropbox_root # Windows konzole if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(errors="replace") # ── Konfigurace ─────────────────────────────────────────────────────────────── -XML_ADRESAR = Path(__file__).parent / "xml_archive" / "2026-04-11" - -FB = dict( - dsn = r"localhost:c:\medicus 3\data\medicus.fdb", - user = "SYSDBA", - password = "masterkey", - charset = "win1250", -) - -DB = dict( - host = "192.168.1.76", - user = "root", - password = "Vlado9674+", - database = "medicus", - charset = "utf8mb4", - cursorclass = pymysql.cursors.DictCursor, -) +XML_ADRESAR = Path(get_dropbox_root()) / "Ordinace" / "Dokumentace_ke_zpracování" / "Zúčtovací zprávy" / "LékovýZáznamWithClaude" / "xml_archive" ICP = "09305001" ODB = "001" @@ -56,7 +41,7 @@ inicializuj_schema = _m06.inicializuj_schema def nacti_pacienty_z_fb(): """Vrati slovnik {(prijmeni_upper, datnar): idpac} ze vsech pacientu v Medicusu.""" - conn = fdb.connect(**FB) + conn = get_medicus_connection() try: cur = conn.cursor() dnes = date.today().isoformat() @@ -114,7 +99,7 @@ def main(): fb_pacienti = nacti_pacienty_z_fb() # Pripoj se k MySQL a inicializuj schema - conn = pymysql.connect(**DB) + conn = connect_mysql(database="medicus", cursorclass=pymysql.cursors.DictCursor) try: inicializuj_schema(conn)