commit 65ef01589cf4e008a195f780c3c751678734b9fd Author: Vladimir Buzalka Date: Sat Sep 27 15:03:31 2025 +0200 notebookVB diff --git a/.idea/Insurance.iml b/.idea/Insurance.iml new file mode 100644 index 0000000..235ece6 --- /dev/null +++ b/.idea/Insurance.iml @@ -0,0 +1,10 @@ + + + + + + + + + + \ No newline at end of file diff --git a/.idea/inspectionProfiles/profiles_settings.xml b/.idea/inspectionProfiles/profiles_settings.xml new file mode 100644 index 0000000..105ce2d --- /dev/null +++ b/.idea/inspectionProfiles/profiles_settings.xml @@ -0,0 +1,6 @@ + + + + \ No newline at end of file diff --git a/.idea/modules.xml b/.idea/modules.xml new file mode 100644 index 0000000..6239183 --- /dev/null +++ b/.idea/modules.xml @@ -0,0 +1,8 @@ + + + + + + + + \ No newline at end of file diff --git a/.idea/vcs.xml b/.idea/vcs.xml new file mode 100644 index 0000000..94a25f7 --- /dev/null +++ b/.idea/vcs.xml @@ -0,0 +1,6 @@ + + + + + + \ No newline at end of file diff --git a/.idea/workspace.xml b/.idea/workspace.xml new file mode 100644 index 0000000..6213dc3 --- /dev/null +++ b/.idea/workspace.xml @@ -0,0 +1,123 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + 1758916129960 + + + + \ No newline at end of file diff --git a/01 testik.py b/01 testik.py new file mode 100644 index 0000000..d94b674 --- /dev/null +++ b/01 testik.py @@ -0,0 +1,131 @@ +# pip install requests_pkcs12 pymysql +from requests_pkcs12 import Pkcs12Adapter +import requests +import xml.etree.ElementTree as ET +from datetime import date +import pymysql +from pymysql.cursors import DictCursor +from pprint import pprint + +# ========== CONFIG ========== +PFX_PATH = r"mbcert.pfx" +PFX_PASS = "Vlado7309208104++" +VERIFY = True + +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" + +# RC provided by you (can contain slash; we normalize before sending/saving) +RC = "280616/091" +K_DATU = date.today().isoformat() + +# MySQL (table already created as per previous DDL) +MYSQL_CFG = dict( + host="192.168.1.76", + port=3307, + user="root", + password="Vlado9674+", + database="medevio", + cursorclass=DictCursor, + autocommit=False, +) + +# ========== HELPERS ========== +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 normalize_rc(rc: str) -> str: + return rc.replace("/", "").strip() + +def post_soap(endpoint: str, body_xml: str) -> str: + env = f''' + + {body_xml} +''' + s = requests.Session() + s.mount("https://", Pkcs12Adapter(pkcs12_filename=PFX_PATH, pkcs12_password=PFX_PASS)) + r = s.post( + endpoint, + data=env.encode("utf-8"), + headers={"Content-Type": "text/xml; charset=utf-8", "SOAPAction": "process"}, + timeout=30, + verify=VERIFY, + ) + print("HTTP:", r.status_code) + return r.text + +def parse_stav(xml_text: str): + 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) -> int: + """ + Insert one log row into medevio.vzp_stav_pojisteni and print what is being saved. + Assumes the table already exists (no DDL/verification here). + """ + payload = { + "rc": normalize_rc(rc), + "k_datu": k_datu, + "stav": to_int_or_none(parsed.get("stav")), # handles '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.") + return 1 + +# ========== MAIN ========== +if __name__ == "__main__": + rc_norm = normalize_rc(RC) + body = f""" + + {rc_norm} + {K_DATU} +""".strip() + + xml = post_soap(EP_STAV, body) + parsed = parse_stav(xml) + print(parsed) # keep your quick print + + # Save to MySQL and print exactly what is saved + save_stav_pojisteni(RC, K_DATU, parsed, xml) diff --git a/02 testík.py b/02 testík.py new file mode 100644 index 0000000..d3e4d74 --- /dev/null +++ b/02 testík.py @@ -0,0 +1,203 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +Call VZP RegistracePojistencePZSB2B for one patient (001 = VPL), +parse the response, upsert rows into medevio.vzp_registrace, +and print what is being saved. +""" + +# pip install requests_pkcs12 pymysql + +from requests_pkcs12 import Pkcs12Adapter +import requests +import xml.etree.ElementTree as ET +from datetime import date +import pymysql +from pymysql.cursors import DictCursor +from pprint import pprint + +# ------------------- CONFIG ------------------- +ENDPOINT = "https://prod.b2b.vzp.cz/B2BProxy/HttpProxy/RegistracePojistencePZSB2B" # case-sensitive +PFX_PATH = r"mbcert.pfx" # <-- your .pfx path +PFX_PASS = "Vlado7309208104++" # <-- your export password +VERIFY = True # or path to CA PEM, e.g. r"C:\certs\vzp_ca.pem" + +# Patient + query +# RC = "7309208104" # rodné číslo without slash +RC = "280616/091" # rodné číslo without slash +K_DATU = date.today().isoformat() # YYYY-MM-DD +ODBORNOSTI = ["001"] # VPL (adult GP) + +# MySQL +MYSQL_CFG = dict( + host="192.168.1.76", + port=3307, + user="root", + password="Vlado9674+", + database="medevio", + cursorclass=DictCursor, + autocommit=False, +) + +# Namespaces (from your response/WSDL) +NS = { + "soap": "http://schemas.xmlsoap.org/soap/envelope/", + "rp": "http://xmlns.gemsystem.cz/B2B/RegistracePojistencePZSB2B/1", +} + + +# ------------------- HELPERS ------------------- +def normalize_rc(rc: str) -> str: + return rc.replace("/", "").strip() + +def build_envelope(rc: str, k_datu: str, odb_list: list[str]) -> str: + odb_xml = "".join([f"{kod}" for kod in odb_list]) + inner = f""" + + {rc} + {k_datu} + + {odb_xml} + +""".strip() + + return f""" + + + {inner} + +""" + +def parse_registrace(xml_text: str): + """ + Return (rows, stav_vyrizeni) where rows is a list of dicts + for each item in the response. + """ + root = ET.fromstring(xml_text) + items = root.findall(".//rp:seznamOdbornosti/rp:odbornost", NS) + out = [] + for it in items: + def g(tag, ctx=it): + el = ctx.find(f"rp:{tag}", NS) + return el.text.strip() if el is not None and el.text else None + + poj = it.find("rp:zdravotniPojistovna", NS) + 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_zkr = poj.find("rp:zkratka", NS).text.strip() if (poj is not None and poj.find("rp:zkratka", NS) is not None) else None + + spec = it.find("rp:odbornost", NS) + odb_kod = spec.find("rp:kod", NS).text.strip() if (spec is not None and spec.find("rp:kod", NS) is not None) else None + odb_naz = spec.find("rp:nazev", NS).text.strip() if (spec is not None and spec.find("rp:nazev", NS) is not None) else None + + out.append(dict( + icz=g("ICZ"), + icp=g("ICP"), + nazev_icp=g("nazevICP"), + nazev_szz=g("nazevSZZ"), + poj_kod=poj_kod, + poj_zkratka=poj_zkr, + odbornost_kod=odb_kod, + odbornost_nazev=odb_naz, + datum_registrace=g("datumRegistrace"), + datum_zahajeni=g("datumZahajeni"), + datum_ukonceni=g("datumUkonceni"), + )) + + st = root.find(".//rp:stavVyrizeniPozadavku", NS) + stav_vyrizeni = st.text.strip() if (st is not None and st.text) else None + return out, stav_vyrizeni + +def upsert_rows(rc: str, query_date: str, rows: list[dict], stav_vyrizeni: str, xml_text: str) -> int: + """ + Insert or update rows into medevio.vzp_registrace and print each payload. + Assumes the table already exists (as per your DDL). + """ + if not rows: + print("No rows found; nothing to save.") + return 0 + + sql = """ +INSERT INTO vzp_registrace + (rc, query_date, odbornost_kod, odbornost_nazev, + icz, icp, nazev_icp, nazev_szz, + poj_kod, poj_zkratka, + datum_registrace, datum_zahajeni, datum_ukonceni, + stav_vyrizeni, response_xml) +VALUES + (%(rc)s, %(query_date)s, %(odbornost_kod)s, %(odbornost_nazev)s, + %(icz)s, %(icp)s, %(nazev_icp)s, %(nazev_szz)s, + %(poj_kod)s, %(poj_zkratka)s, + %(datum_registrace)s, %(datum_zahajeni)s, %(datum_ukonceni)s, + %(stav_vyrizeni)s, %(response_xml)s) +ON DUPLICATE KEY UPDATE + odbornost_nazev = VALUES(odbornost_nazev), + nazev_icp = VALUES(nazev_icp), + nazev_szz = VALUES(nazev_szz), + poj_kod = VALUES(poj_kod), + poj_zkratka = VALUES(poj_zkratka), + datum_registrace= VALUES(datum_registrace), + datum_zahajeni = VALUES(datum_zahajeni), + datum_ukonceni = VALUES(datum_ukonceni), + stav_vyrizeni = VALUES(stav_vyrizeni), + response_xml = VALUES(response_xml), + updated_at = CURRENT_TIMESTAMP +""" + rc_norm = normalize_rc(rc) + qd = query_date or date.today().isoformat() + payloads = [] + for r in rows: + payload = { + "rc": rc_norm, + "query_date": qd, + **r, + "stav_vyrizeni": stav_vyrizeni, + "response_xml": xml_text, + } + payloads.append(payload) + + # Print what we're going to save + print("\n=== Will save the following payload(s) to medevio.vzp_registrace ===") + for p in payloads: + pprint(p) + + conn = pymysql.connect(**MYSQL_CFG) + try: + with conn.cursor() as cur: + cur.executemany(sql, payloads) + conn.commit() + finally: + conn.close() + + print(f"\nUpserted rows: {len(payloads)}") + return len(payloads) + + +# ------------------- MAIN FLOW ------------------- +def main(): + # Build SOAP envelope + envelope = build_envelope(RC, K_DATU, ODBORNOSTI) + + # mTLS session + session = requests.Session() + session.mount("https://", Pkcs12Adapter(pkcs12_filename=PFX_PATH, pkcs12_password=PFX_PASS)) + + headers = { + "Content-Type": "text/xml; charset=utf-8", + "SOAPAction": "process", # Oracle composite usually expects this + } + + # Call service + resp = session.post(ENDPOINT, data=envelope.encode("utf-8"), + headers=headers, timeout=30, verify=VERIFY) + print("HTTP:", resp.status_code) + + # (Optional) Uncomment to see raw XML + # print(resp.text) + + # Parse and save + rows, stav = parse_registrace(resp.text) + upsert_rows(RC, K_DATU, rows, stav, resp.text) + + +if __name__ == "__main__": + main() diff --git a/File3.py b/File3.py new file mode 100644 index 0000000..5a73eb7 --- /dev/null +++ b/File3.py @@ -0,0 +1,5 @@ +from functions import check_insurance +import time +import fdb + + diff --git a/functions.py b/functions.py new file mode 100644 index 0000000..5d129aa --- /dev/null +++ b/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''' + + {inner_xml} +''' + 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""" + + {rc_norm} + {kd} +""".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