Vladimir Buzalka
304 lines
10 KiB
Python
304 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
from requests_pkcs12 import Pkcs12Adapter
|
|
import requests
|
|
import uuid
|
|
from datetime import date
|
|
|
|
|
|
|
|
class VZPB2BClient:
|
|
def __init__(self, env: str, pfx_path: str, pfx_password: str,
|
|
icz: str = "00000000", dic: str = "00000000"):
|
|
|
|
# Normalize environment name
|
|
env = env.lower().strip()
|
|
|
|
if env in ("prod", "production", "live", "real"):
|
|
self.env = "prod"
|
|
elif env in ("simu", "simulace", "test", "testing"):
|
|
self.env = "simu"
|
|
else:
|
|
raise ValueError(f"Unknown environment '{env}'. Use 'simu' or 'prod'.")
|
|
|
|
self.pfx_path = pfx_path
|
|
self.pfx_password = pfx_password
|
|
self.icz = icz
|
|
self.dic = dic
|
|
|
|
# Prepare mTLS session
|
|
session = requests.Session()
|
|
session.mount(
|
|
"https://",
|
|
Pkcs12Adapter(pkcs12_filename=pfx_path, pkcs12_password=pfx_password)
|
|
)
|
|
self.session = session
|
|
|
|
# --------------------------------------------------------------
|
|
# URL BUILDER
|
|
# --------------------------------------------------------------
|
|
def _build_endpoint(self, service_name: str) -> str:
|
|
"""
|
|
SIMU:
|
|
https://simu.b2b.vzp.cz/B2BProxy/HttpProxy/SIMU<Service>?sluzba=SIMU<Service>
|
|
PROD:
|
|
https://prod.b2b.vzp.cz/B2BProxy/HttpProxy/<Service>
|
|
"""
|
|
|
|
if self.env == "simu":
|
|
simu_service = f"SIMU{service_name}"
|
|
return (
|
|
f"https://simu.b2b.vzp.cz/B2BProxy/HttpProxy/"
|
|
f"{simu_service}?sluzba={simu_service}"
|
|
)
|
|
|
|
# Production
|
|
return (
|
|
f"https://prod.b2b.vzp.cz/B2BProxy/HttpProxy/{service_name}"
|
|
)
|
|
|
|
# --------------------------------------------------------------
|
|
# SOAP HEADER BUILDER
|
|
# --------------------------------------------------------------
|
|
def _header(self) -> str:
|
|
idZpravy = uuid.uuid4().hex[:12] # must be alphanumeric, max 12 chars
|
|
return f"""
|
|
<com:idZpravy>{idZpravy}</com:idZpravy>
|
|
<com:idSubjektu>
|
|
<com:icz>{self.icz}</com:icz>
|
|
<com:dic>{self.dic}</com:dic>
|
|
</com:idSubjektu>
|
|
"""
|
|
|
|
# --------------------------------------------------------------
|
|
# OVERPRUKAZ — EHIC CHECK
|
|
# --------------------------------------------------------------
|
|
def over_prukaz_pojistence(self, cislo_prukazu: str, k_datu: str = None) -> str:
|
|
"""
|
|
Calls OverPrukazPojistenceB2B (SIMU or PROD depending on env).
|
|
Returns raw XML string.
|
|
"""
|
|
|
|
service = "OverPrukazPojistenceB2B"
|
|
endpoint = self._build_endpoint(service)
|
|
|
|
if not k_datu:
|
|
k_datu = date.today().isoformat()
|
|
|
|
soap = f"""<?xml version="1.0" encoding="utf-8"?>
|
|
<soap:Envelope
|
|
xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/"
|
|
xmlns:vzp="http://xmlns.gemsystem.cz/OverPrukazPojistenceB2B"
|
|
xmlns:com="http://xmlns.gemsystem.cz/CommonB2B">
|
|
|
|
<soap:Header>
|
|
{self._header()}
|
|
</soap:Header>
|
|
|
|
<soap:Body>
|
|
<vzp:OverPrukazPojistenceB2BPozadavek>
|
|
<vzp:cisloPrukazu>{cislo_prukazu}</vzp:cisloPrukazu>
|
|
<vzp:kDatu>{k_datu}</vzp:kDatu>
|
|
</vzp:OverPrukazPojistenceB2BPozadavek>
|
|
</soap:Body>
|
|
|
|
</soap:Envelope>
|
|
"""
|
|
|
|
headers = {"Content-Type": "text/xml; charset=utf-8"}
|
|
|
|
print(f"Calling: {endpoint}")
|
|
response = self.session.post(
|
|
endpoint,
|
|
data=soap.encode("utf-8"),
|
|
headers=headers,
|
|
timeout=30
|
|
)
|
|
print("HTTP:", response.status_code)
|
|
return response.text
|
|
|
|
def stav_pojisteni(self, rc: str, k_datu: str = None, prijmeni: str = None):
|
|
"""
|
|
Calls stavPojisteniB2B (SIMU or PROD).
|
|
"""
|
|
service = "stavPojisteniB2B"
|
|
endpoint = self._build_endpoint(service)
|
|
|
|
if not k_datu:
|
|
k_datu = date.today().isoformat()
|
|
|
|
prijmeni_xml = f"<vzp:prijmeni>{prijmeni}</vzp:prijmeni>" if prijmeni else ""
|
|
|
|
soap = f"""<?xml version="1.0" encoding="utf-8"?>
|
|
<soap:Envelope
|
|
xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/"
|
|
xmlns:vzp="http://xmlns.gemsystem.cz/stavPojisteniB2B"
|
|
xmlns:com="http://xmlns.gemsystem.cz/CommonB2B">
|
|
|
|
<soap:Header>
|
|
{self._header()}
|
|
</soap:Header>
|
|
|
|
<soap:Body>
|
|
<vzp:stavPojisteniB2B>
|
|
<vzp:cisloPojistence>{rc}</vzp:cisloPojistence>
|
|
{prijmeni_xml}
|
|
<vzp:kDatu>{k_datu}</vzp:kDatu>
|
|
</vzp:stavPojisteniB2B>
|
|
</soap:Body>
|
|
|
|
</soap:Envelope>
|
|
"""
|
|
|
|
headers = {
|
|
"Content-Type": "text/xml; charset=utf-8",
|
|
"SOAPAction": "process"
|
|
}
|
|
|
|
print(f"Calling: {endpoint}")
|
|
resp = self.session.post(endpoint, data=soap.encode("utf-8"),
|
|
headers=headers, timeout=30)
|
|
print("HTTP:", resp.status_code)
|
|
return resp.text
|
|
|
|
def registrace_lekare(self, rc: str, k_datu: str = None,
|
|
odbornosti: list = None) -> str:
|
|
"""
|
|
Calls RegistracePojistencePZSB2B — vrátí registrující lékaře pojištěnce.
|
|
odbornosti: seznam kódů, např. ["001","002","014"]. None = bez filtru (vrátí vše).
|
|
"""
|
|
service = "RegistracePojistencePZSB2B"
|
|
endpoint = self._build_endpoint(service)
|
|
|
|
if not k_datu:
|
|
k_datu = date.today().isoformat()
|
|
|
|
ns = "http://xmlns.gemsystem.cz/B2B/RegistracePojistencePZSB2B/1"
|
|
odb_xml = ""
|
|
if odbornosti:
|
|
kody = "".join(f"<ns1:kodOdbornosti>{k}</ns1:kodOdbornosti>"
|
|
for k in odbornosti)
|
|
odb_xml = f"<ns1:seznamOdbornosti>{kody}</ns1:seznamOdbornosti>"
|
|
|
|
soap = f"""<?xml version="1.0" encoding="utf-8"?>
|
|
<soap:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/">
|
|
<soap:Body>
|
|
<ns1:registracePojistencePZSB2B xmlns:ns1="{ns}">
|
|
<ns1:cisloPojistence>{rc}</ns1:cisloPojistence>
|
|
<ns1:kDatu>{k_datu}</ns1:kDatu>
|
|
{odb_xml}
|
|
</ns1:registracePojistencePZSB2B>
|
|
</soap:Body>
|
|
</soap:Envelope>"""
|
|
|
|
resp = self.session.post(
|
|
endpoint,
|
|
data=soap.encode("utf-8"),
|
|
headers={"Content-Type": "text/xml; charset=utf-8", "SOAPAction": "process"},
|
|
timeout=30,
|
|
)
|
|
return resp.text
|
|
|
|
def parse_registrace_lekare(self, xml_text: str) -> list[dict]:
|
|
"""
|
|
Parsuje odpověď RegistracePojistencePZSB2B.
|
|
Vrátí seznam diktů — jeden na odbornost (i prázdné = ma_lekare=False).
|
|
"""
|
|
import xml.etree.ElementTree as ET
|
|
|
|
NS = {
|
|
"soap": "http://schemas.xmlsoap.org/soap/envelope/",
|
|
"rp": "http://xmlns.gemsystem.cz/B2B/RegistracePojistencePZSB2B/1",
|
|
}
|
|
|
|
root = ET.fromstring(xml_text)
|
|
stav_el = root.find(".//rp:stavVyrizeniPozadavku", NS)
|
|
stav_vyrizeni = stav_el.text.strip() if stav_el is not None and stav_el.text else None
|
|
|
|
# Jen vnější <odbornost> elementy (ty s ICZ), ne vnořené subelementy
|
|
results = []
|
|
for it in root.findall(".//rp:seznamOdbornosti/rp:odbornost", NS):
|
|
def g(tag):
|
|
el = it.find(f"rp:{tag}", NS)
|
|
return el.text.strip() if el is not None and el.text else None
|
|
|
|
odb = it.find("rp:odbornost", NS)
|
|
|
|
if odb is not None:
|
|
# Záznam s lékařem
|
|
kod = odb.find("rp:kod", NS)
|
|
naz = odb.find("rp:nazev", NS)
|
|
poj = it.find("rp:zdravotniPojistovna", NS)
|
|
results.append({
|
|
"ma_lekare": True,
|
|
"kod_odbornosti": kod.text.strip() if kod is not None and kod.text else None,
|
|
"nazev_odbornosti": naz.text.strip() if naz is not None and naz.text else None,
|
|
"ICZ": g("ICZ"),
|
|
"ICP": g("ICP"),
|
|
"nazev_lekare": g("nazevICP"),
|
|
"nazev_zzz": g("nazevSZZ"),
|
|
"poj_kod": poj.find("rp:kod", NS).text.strip() if poj is not None and poj.find("rp:kod", NS) is not None else None,
|
|
"poj_zkratka": poj.find("rp:zkratka", NS).text.strip() if poj is not None and poj.find("rp:zkratka", NS) is not None else None,
|
|
"datum_registrace": g("datumRegistrace"),
|
|
"datum_zahajeni": g("datumZahajeni"),
|
|
"datum_ukonceni": g("datumUkonceni"),
|
|
"stav_vyrizeni": stav_vyrizeni,
|
|
})
|
|
else:
|
|
# Prázdný placeholder — pacient nemá lékaře v této odbornosti
|
|
# (VZP vrací element bez ICZ/ICP — ignorujeme, zaznamená skript sám)
|
|
pass
|
|
|
|
return results
|
|
|
|
def parse_stav_pojisteni(self, xml_text: str):
|
|
"""
|
|
Parses stavPojisteniB2B SOAP response into a Python dict.
|
|
|
|
Returned structure:
|
|
{
|
|
"stavVyrizeni": int,
|
|
"stav": str | None,
|
|
"kodPojistovny": str | None,
|
|
"nazevPojistovny": str | None,
|
|
"pojisteniKod": str | None
|
|
}
|
|
"""
|
|
|
|
import xml.etree.ElementTree as ET
|
|
|
|
NS = {
|
|
"soap": "http://schemas.xmlsoap.org/soap/envelope/",
|
|
"vzp": "http://xmlns.gemsystem.cz/stavPojisteniB2B"
|
|
}
|
|
|
|
root = ET.fromstring(xml_text)
|
|
|
|
# ---- Extract status ----
|
|
stav_vyr = root.find(".//vzp:stavVyrizeniPozadavku", NS)
|
|
stav_vyr = int(stav_vyr.text.strip()) if stav_vyr is not None else None
|
|
|
|
# ---- If no stavPojisteni element present (e.g. 0 or some errors) ----
|
|
node_stav = root.find(".//vzp:stavPojisteni", NS)
|
|
if node_stav is None:
|
|
return {
|
|
"stavVyrizeni": stav_vyr,
|
|
"stav": None,
|
|
"kodPojistovny": None,
|
|
"nazevPojistovny": None,
|
|
"pojisteniKod": None,
|
|
}
|
|
|
|
def get(tag):
|
|
el = node_stav.find(f"vzp:{tag}", NS)
|
|
return el.text.strip() if el is not None and el.text else None
|
|
|
|
return {
|
|
"stavVyrizeni": stav_vyr,
|
|
"stav": get("stav"),
|
|
"kodPojistovny": get("kodPojistovny"),
|
|
"nazevPojistovny": get("nazevPojistovny"),
|
|
"pojisteniKod": get("pojisteniKod"),
|
|
}
|