notebookVB
This commit is contained in:
213
10 Tests/vzpb2b_client.py
Normal file
213
10 Tests/vzpb2b_client.py
Normal file
@@ -0,0 +1,213 @@
|
||||
#!/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 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"),
|
||||
}
|
||||
Reference in New Issue
Block a user