diff --git a/10 Tests/MBcert.pfx b/10 Tests/MBcert.pfx new file mode 100644 index 0000000..6fbd274 Binary files /dev/null and b/10 Tests/MBcert.pfx differ diff --git a/10 Tests/test stavpojisteni.py b/10 Tests/test stavpojisteni.py new file mode 100644 index 0000000..309b7cb --- /dev/null +++ b/10 Tests/test stavpojisteni.py @@ -0,0 +1,13 @@ +from vzpb2b_client import VZPB2BClient + +client = VZPB2BClient( + env="production", + pfx_path="mbcert.pfx", + pfx_password="Vlado7309208104++", + icz="00000000", + dic="00000000" +) + +response = client.stav_pojisteni("0308020152") +# print(response) +print(client.parse_stav_pojisteni(response)) \ No newline at end of file diff --git a/10 Tests/test.py b/10 Tests/test.py new file mode 100644 index 0000000..70f2d92 --- /dev/null +++ b/10 Tests/test.py @@ -0,0 +1,12 @@ +from vzpb2b_client import VZPB2BClient + +client = VZPB2BClient( + env="simu", # or "prod" + pfx_path="mbcert.pfx", + pfx_password="Vlado7309208104++", + icz="00000000", + dic="00000000" +) + +result_xml = client.over_prukaz_pojistence("80203111194350000001") +print(result_xml) diff --git a/10 Tests/vzpb2b_client.py b/10 Tests/vzpb2b_client.py new file mode 100644 index 0000000..a8ea67f --- /dev/null +++ b/10 Tests/vzpb2b_client.py @@ -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?sluzba=SIMU + PROD: + https://prod.b2b.vzp.cz/B2BProxy/HttpProxy/ + """ + + 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""" + {idZpravy} + + {self.icz} + {self.dic} + + """ + + # -------------------------------------------------------------- + # 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""" + + + + {self._header()} + + + + + {cislo_prukazu} + {k_datu} + + + + +""" + + 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"{prijmeni}" if prijmeni else "" + + soap = f""" + + + + {self._header()} + + + + + {rc} + {prijmeni_xml} + {k_datu} + + + + +""" + + 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"), + }