notebookVB

functions.py

@@ -0,0 +1,138 @@
+# functions.py
+# pip install requests_pkcs12 pymysql
+from __future__ import annotations
+
+import os
+from datetime import date
+from pprint import pprint
+import xml.etree.ElementTree as ET
+
+import requests
+from requests_pkcs12 import Pkcs12Adapter
+import pymysql
+from pymysql.cursors import DictCursor
+
+# ======== CONFIG (env overrides allowed) ========
+PFX_PATH = os.getenv("VZP_PFX_PATH", r"mbcert.pfx")
+PFX_PASS = os.getenv("VZP_PFX_PASS", "Vlado7309208104++")
+VERIFY   = os.getenv("VZP_CA_VERIFY", "true").lower() != "false"  # set to path or "false" to disable
+
+EP_STAV  = "https://prod.b2b.vzp.cz/B2BProxy/HttpProxy/stavPojisteniB2B"
+SOAP_NS  = "http://schemas.xmlsoap.org/soap/envelope/"
+NS_STAV  = "http://xmlns.gemsystem.cz/stavPojisteniB2B"
+
+MYSQL_CFG = {
+    "host": os.getenv("MYSQL_HOST", "192.168.1.76"),
+    "port": int(os.getenv("MYSQL_PORT", "3307")),
+    "user": os.getenv("MYSQL_USER", "root"),
+    "password": os.getenv("MYSQL_PASSWORD", "Vlado9674+"),
+    "database": os.getenv("MYSQL_DB", "medevio"),
+    "cursorclass": DictCursor,
+    "autocommit": False,
+}
+
+# ======== HELPERS ========
+def normalize_rc(rc: str) -> str:
+    """Return rodné číslo without slash/spaces."""
+    return rc.replace("/", "").replace(" ", "").strip()
+
+def to_int_or_none(val):
+    if val is None:
+        return None
+    s = str(val).strip()
+    return int(s) if s.isdigit() else None
+
+def _session():
+    s = requests.Session()
+    s.mount("https://", Pkcs12Adapter(pkcs12_filename=PFX_PATH, pkcs12_password=PFX_PASS))
+    return s
+
+def post_soap(endpoint: str, inner_xml: str) -> str:
+    """Wrap body in SOAP 1.1 envelope and POST with mTLS."""
+    envelope = f'''<?xml version="1.0" encoding="utf-8"?>
+<soap:Envelope xmlns:soap="{SOAP_NS}">
+  <soap:Body>{inner_xml}</soap:Body>
+</soap:Envelope>'''
+    r = _session().post(
+        endpoint,
+        data=envelope.encode("utf-8"),
+        headers={"Content-Type": "text/xml; charset=utf-8", "SOAPAction": "process"},
+        timeout=30,
+        verify=VERIFY,
+    )
+    # You may log r.status_code if needed.
+    return r.text
+
+def parse_stav(xml_text: str) -> dict:
+    """Parse StavPojisteniB2B response → dict with keys:
+       'stav','kodPojistovny','nazevPojistovny','pojisteniKod','stavVyrizeniPozadavku'."""
+    NS = {"soap": SOAP_NS, "s": NS_STAV}
+    root = ET.fromstring(xml_text)
+    out = {"stav": None, "kodPojistovny": None, "nazevPojistovny": None,
+           "pojisteniKod": None, "stavVyrizeniPozadavku": None}
+    node = root.find(".//s:stavPojisteni", NS)
+    if node is not None:
+        for tag in ("stav", "kodPojistovny", "nazevPojistovny", "pojisteniKod"):
+            el = node.find(f"s:{tag}", NS)
+            out[tag] = el.text.strip() if (el is not None and el.text) else None
+    st = root.find(".//s:stavVyrizeniPozadavku", NS)
+    out["stavVyrizeniPozadavku"] = st.text.strip() if (st is not None and st.text) else None
+    return out
+
+def save_stav_pojisteni(rc: str, k_datu: str, parsed: dict, response_xml: str) -> None:
+    """Insert one log row into medevio.vzp_stav_pojisteni and print what is being saved."""
+    payload = {
+        "rc": normalize_rc(rc),
+        "k_datu": k_datu,
+        "stav": to_int_or_none(parsed.get("stav")),  # 'X' -> None
+        "kod_pojistovny": parsed.get("kodPojistovny"),
+        "nazev_pojistovny": parsed.get("nazevPojistovny"),
+        "pojisteni_kod": parsed.get("pojisteniKod"),
+        "stav_vyrizeni": to_int_or_none(parsed.get("stavVyrizeniPozadavku")),
+        "response_xml": response_xml,
+    }
+
+    print("\n=== vzp_stav_pojisteni: will insert ===")
+    pprint(payload)
+
+    sql = """
+    INSERT INTO vzp_stav_pojisteni
+      (rc, k_datu, stav, kod_pojistovny, nazev_pojistovny, pojisteni_kod, stav_vyrizeni, response_xml)
+    VALUES
+      (%(rc)s, %(k_datu)s, %(stav)s, %(kod_pojistovny)s, %(nazev_pojistovny)s, %(pojisteni_kod)s, %(stav_vyrizeni)s, %(response_xml)s)
+    """
+    conn = pymysql.connect(**MYSQL_CFG)
+    try:
+        with conn.cursor() as cur:
+            cur.execute(sql, payload)
+        conn.commit()
+    finally:
+        conn.close()
+    print("Inserted 1 row into vzp_stav_pojisteni.")
+
+# ======== PUBLIC API ========
+def check_insurance(rc: str, mode: str | None = None, k_datu: str | None = None):
+    """
+    Call VZP 'StavPojisteniB2B' for given RC and date.
+    Returns (is_active, info_dict).
+      - is_active: True iff info_dict['stav'] == '1'
+      - info_dict: {'stav','kodPojistovny','nazevPojistovny','pojisteniKod','stavVyrizeniPozadavku'}
+    If mode == 'nodb' (case-insensitive), DOES NOT write to DB. Otherwise logs into medevio.vzp_stav_pojisteni.
+    """
+    rc_norm = normalize_rc(rc)
+    kd = k_datu or date.today().isoformat()
+
+    body = f"""
+<ns1:stavPojisteniB2B xmlns:ns1="{NS_STAV}">
+  <ns1:cisloPojistence>{rc_norm}</ns1:cisloPojistence>
+  <ns1:kDatu>{kd}</ns1:kDatu>
+</ns1:stavPojisteniB2B>""".strip()
+
+    xml = post_soap(EP_STAV, body)
+    info = parse_stav(xml)
+    is_active = (info.get("stav") == "1")
+
+    if not (mode and mode.lower() == "nodb"):
+        save_stav_pojisteni(rc_norm, kd, info, xml)
+
+    return is_active, info