421 lines
15 KiB
Python
421 lines
15 KiB
Python
"""
|
|
Hromadne stazeni lekovych zaznamu z eReceptu pro registrovane pacienty Medicusu.
|
|
|
|
Spusteni:
|
|
# pouze rodina (testovaci run)
|
|
python 07StahnoutVsechny.py --prijmeni Buzalka,Buzalkova,Kusinova
|
|
|
|
# vsichni registrovani pacienti
|
|
python 07StahnoutVsechny.py
|
|
|
|
Logika poctu mesicu:
|
|
- prvni stazeni pacienta → 60 mesicu (maximum)
|
|
- opakowane stazeni → ceil(pocet_dni_od_posledniho / 30) + 1
|
|
(prekryv 1 mesic pro jistotu, INSERT IGNORE zajisti bez duplikatu)
|
|
|
|
XML archiv:
|
|
xml_archive/YYYY-MM-DD/{Prijmeni}_{Jmena}_{datnar}.xml
|
|
Cesta ulozena take v zprava.xml_soubor pro snadne dohledani.
|
|
"""
|
|
|
|
import argparse
|
|
import importlib.util
|
|
import math
|
|
import sys
|
|
import time
|
|
import uuid
|
|
|
|
# Windows konzole — nahrad neunikatni znaky misto padu
|
|
if hasattr(sys.stdout, "reconfigure"):
|
|
sys.stdout.reconfigure(errors="replace")
|
|
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
|
|
|
|
# ── 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_MEZI_VOLANIMI = 15 # 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,
|
|
)
|
|
|
|
# ── XML archiv ────────────────────────────────────────────────────────────────
|
|
XML_DIR = Path(__file__).parent / "xml_archive"
|
|
|
|
|
|
# ── Firebird: nacteni registrovanych pacientu ─────────────────────────────────
|
|
|
|
_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):
|
|
"""
|
|
Vraci seznam dict {idpac, prijmeni, jmeno, datnar}.
|
|
prijmeni_filtr: list prijmeni (napr. ['Buzalka', 'Buzalkova']) nebo None = vsichni.
|
|
"""
|
|
conn = fdb.connect(dsn=FB_DSN, user=FB_USER, password=FB_PASS, charset=FB_CHARSET)
|
|
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: pacient UPSERT ─────────────────────────────────────────────────────
|
|
|
|
def upsert_pacient(cur, pac):
|
|
"""
|
|
Vlozi nebo aktualizuje pacienta v tabulce pacient.
|
|
Vraci MySQL id radku.
|
|
"""
|
|
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 FROM pacient WHERE idpac = %s", (pac["idpac"],))
|
|
return cur.fetchone()["id"]
|
|
|
|
|
|
def posledni_stazeni(cur, pacient_id):
|
|
"""Vraci datetime posledniho stazeni, nebo None pro noveho pacienta."""
|
|
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):
|
|
"""60 pro prvni stazeni, jinak delta v mesicich + 1 (prekryv)."""
|
|
if posledni is None:
|
|
return POCET_MESICU_MAX
|
|
delta_dni = (datetime.now() - posledni).days
|
|
return min(math.ceil(delta_dni / 30) + 1, POCET_MESICU_MAX)
|
|
|
|
|
|
# ── SOAP volani ───────────────────────────────────────────────────────────────
|
|
|
|
def extrahuj_soap_fault(xml_text):
|
|
"""
|
|
Pokud XML obsahuje SOAP Fault, vraci text chyby (str).
|
|
Pokud je odpoved v poradku, vraci None.
|
|
"""
|
|
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"
|
|
# Zkontroluj SOAP Fault
|
|
fault = body.find(f"{{{NS_SOAP}}}Fault")
|
|
if fault is None:
|
|
fault = body.find("Fault") # nektery server posila bez namespace
|
|
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
|
|
# Zkontroluj, ze odpoved je spravneho typu (NacistLekovyZaznamOdpoved)
|
|
odpoved = body.find(f"{{{NS_SUKL}}}NacistLekovyZaznamOdpoved")
|
|
if odpoved is None:
|
|
# Neznamy format — vrat prvni tag jako info
|
|
first = list(body)
|
|
tag = first[0].tag if first else "prazdne Body"
|
|
return f"Neocekavana odpoved: {tag}"
|
|
return None # vse OK
|
|
except Exception as e:
|
|
return f"Chyba pri parsovani odpovedi: {e}"
|
|
|
|
|
|
def uloz_poznamku(conn, pacient_id, poznamka):
|
|
"""Ulozi nebo vymaze poznamku (chybu) u pacienta."""
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"UPDATE pacient SET poznamka = %s WHERE id = %s",
|
|
(poznamka, pacient_id)
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
def nacti_lekovy_zaznam(sess, prijmeni, jmena, datum_narozeni, pocet_mesicu):
|
|
"""
|
|
Zavola NacistLekovyZaznam pro jednoho pacienta.
|
|
Vraci xml_text (str). Vyhazuje RuntimeError pri HTTP chybe.
|
|
"""
|
|
id_zpravy = str(uuid.uuid4())
|
|
odeslano = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S+00:00")
|
|
|
|
soap = (
|
|
'<?xml version="1.0" encoding="UTF-8"?>'
|
|
'<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">'
|
|
'<soapenv:Body>'
|
|
f'<NacistLekovyZaznamLekarDotaz xmlns="http://www.sukl.cz/erp/201912">'
|
|
f'<Doklad>'
|
|
f'<Pristupujici>'
|
|
f'<Uzivatel>{UZIVATEL}</Uzivatel>'
|
|
f'<Pracoviste>{PRACOVISTE}</Pracoviste>'
|
|
f'</Pristupujici>'
|
|
f'<PocetZnakuATC>{POCET_ZNAKU_ATC}</PocetZnakuATC>'
|
|
f'<PocetMesicu>{pocet_mesicu}</PocetMesicu>'
|
|
f'<Pacient><Totoznost><Jmeno>'
|
|
f'<Prijmeni>{xml_escape(prijmeni)}</Prijmeni>'
|
|
f'<Jmena>{xml_escape(jmena)}</Jmena>'
|
|
f'</Jmeno><DatumNarozeni>{datum_narozeni}</DatumNarozeni>'
|
|
f'</Totoznost></Pacient>'
|
|
f'</Doklad>'
|
|
f'<Zprava>'
|
|
f'<ID_Zpravy>{id_zpravy}</ID_Zpravy>'
|
|
f'<Verze>202501A</Verze>'
|
|
f'<Odeslano>{odeslano}</Odeslano>'
|
|
f'<SW_Klienta>MEDICUS_____</SW_Klienta>'
|
|
f'</Zprava>'
|
|
f'</NacistLekovyZaznamLekarDotaz>'
|
|
'</soapenv:Body>'
|
|
'</soapenv:Envelope>'
|
|
)
|
|
|
|
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):
|
|
"""
|
|
Ulozi XML do xml_archive/YYYY-MM-DD/{Prijmeni}_{Jmena}_{datnar}.xml
|
|
Vraci relativni cestu (str) vuci adresari skriptu.
|
|
"""
|
|
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.relative_to(Path(__file__).parent))
|
|
|
|
|
|
# ── Hlavni smycka ─────────────────────────────────────────────────────────────
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser(
|
|
description="Hromadne stazeni lekovych zaznamu z eReceptu"
|
|
)
|
|
ap.add_argument(
|
|
"--prijmeni",
|
|
help="Filtr prijmeni oddelena carkou, napr: Buzalka,Buzalkova,Kusinova",
|
|
default=None,
|
|
)
|
|
ap.add_argument(
|
|
"--limit",
|
|
type=int,
|
|
default=None,
|
|
help="Zpracuj pouze N pacientu",
|
|
)
|
|
ap.add_argument(
|
|
"--offset",
|
|
type=int,
|
|
default=0,
|
|
help="Preskoc prvnich N pacientu (pro postupne davkovani)",
|
|
)
|
|
args = ap.parse_args()
|
|
|
|
prijmeni_filtr = None
|
|
if args.prijmeni:
|
|
prijmeni_filtr = [p.strip() for p in args.prijmeni.split(",")]
|
|
print(f"Filtr prijmeni: {prijmeni_filtr}")
|
|
|
|
# 1. Nacti pacienty z Medicusu
|
|
print("Nacitam pacienty z Medicusu (Firebird)...")
|
|
pacienti = nacti_pacienty(prijmeni_filtr)
|
|
print(f" Nalezeno: {len(pacienti)} pacientu")
|
|
if args.offset:
|
|
pacienti = pacienti[args.offset:]
|
|
print(f" Preskoceno: {args.offset} (--offset {args.offset})")
|
|
if args.limit:
|
|
pacienti = pacienti[:args.limit]
|
|
print(f" Omezeno na: {len(pacienti)} (--limit {args.limit})")
|
|
if not pacienti:
|
|
print("Zadni pacienti — konec.")
|
|
return
|
|
|
|
# 2. Pripoj se k MySQL, over schema (CREATE IF NOT EXISTS)
|
|
print("Pripojuji k MySQL...")
|
|
conn = pymysql.connect(**DB)
|
|
inicializuj_schema(conn)
|
|
|
|
# 3. Priprav SOAP session (sdilena pro vsechny pacienty)
|
|
sess = Session()
|
|
sess.mount("https://", Pkcs12Adapter(
|
|
pkcs12_filename=PFX_FILE,
|
|
pkcs12_password=PFX_PASS,
|
|
))
|
|
sess.auth = (API_USER, API_PASS)
|
|
|
|
dnes_str = date.today().isoformat()
|
|
ok = 0
|
|
chyby = 0
|
|
celkem = len(pacienti)
|
|
|
|
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)
|
|
|
|
print(f"\n[{i}/{celkem}] {prijmeni} {jmena} (*{datnar_str})")
|
|
|
|
# UPSERT pacienta, zjisti kdy byl naposledy stazen
|
|
with conn.cursor() as cur:
|
|
pacient_id = upsert_pacient(cur, pac)
|
|
posledni = posledni_stazeni(cur, pacient_id)
|
|
conn.commit()
|
|
|
|
pocet_mesicu = vypocti_pocet_mesicu(posledni)
|
|
print(f" Stahuju {pocet_mesicu} mesicu "
|
|
f"(posledni stazeni: {posledni.strftime('%Y-%m-%d') if posledni else 'nikdy'})")
|
|
|
|
# Zavolej API
|
|
try:
|
|
xml_text = nacti_lekovy_zaznam(sess, prijmeni, jmena, datnar_str, pocet_mesicu)
|
|
except Exception as e:
|
|
zprava_chyby = str(e)[:400]
|
|
print(f" CHYBA API: {zprava_chyby}")
|
|
uloz_poznamku(conn, pacient_id, zprava_chyby)
|
|
chyby += 1
|
|
continue
|
|
|
|
# Detekuj SOAP Fault v odpovedi (HTTP 200 ale chyba uvnitr)
|
|
soap_fault = extrahuj_soap_fault(xml_text)
|
|
if soap_fault:
|
|
print(f" SOAP FAULT: {soap_fault}")
|
|
uloz_poznamku(conn, pacient_id, soap_fault[:400])
|
|
chyby += 1
|
|
continue
|
|
|
|
# Uloz XML na disk
|
|
xml_soubor = uloz_xml_na_disk(xml_text, prijmeni, jmena, datnar_str, dnes_str)
|
|
xml_path = Path(__file__).parent / xml_soubor
|
|
print(f" XML: {xml_soubor} ({xml_path.stat().st_size // 1024} KB)")
|
|
|
|
# Parsuj + uloz do MySQL
|
|
try:
|
|
zprava_d, predpisy, vydeji, predepisujici, vydavajici = parsuj_xml(xml_path)
|
|
uloz(conn, zprava_d, predpisy, vydeji, predepisujici, vydavajici,
|
|
pacient_id=pacient_id, xml_soubor=xml_soubor)
|
|
uloz_poznamku(conn, pacient_id, None) # vymaz predchozi chybu
|
|
ok += 1
|
|
except Exception as e:
|
|
zprava_chyby = str(e)[:400]
|
|
print(f" CHYBA parsovani/ulozeni: {zprava_chyby}")
|
|
uloz_poznamku(conn, pacient_id, zprava_chyby)
|
|
chyby += 1
|
|
|
|
# Pauza mezi volanimi API (neplati po poslednim pacientovi)
|
|
if i < celkem:
|
|
print(f" Cekam {PAUZA_MEZI_VOLANIMI}s ...")
|
|
time.sleep(PAUZA_MEZI_VOLANIMI)
|
|
|
|
finally:
|
|
conn.close()
|
|
sess.close()
|
|
|
|
print(f"\n{'=' * 55}")
|
|
print(f"Hotovo: {ok} OK | {chyby} chyb | celkem {celkem} pacientu")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|