132 lines
4.1 KiB
Python
132 lines
4.1 KiB
Python
# 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'''<?xml version="1.0" encoding="utf-8"?>
|
|
<soap:Envelope xmlns:soap="{SOAP_NS}">
|
|
<soap:Body>{body_xml}</soap:Body>
|
|
</soap:Envelope>'''
|
|
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"""
|
|
<ns1:stavPojisteniB2B xmlns:ns1="{NS_STAV}">
|
|
<ns1:cisloPojistence>{rc_norm}</ns1:cisloPojistence>
|
|
<ns1:kDatu>{K_DATU}</ns1:kDatu>
|
|
</ns1:stavPojisteniB2B>""".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)
|