torrents/10 Tests/vzpb2b_client.py

#!/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"),
        }