administrator/ordinaceprojekt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

notebookvb

Browse Source
This commit is contained in:
Vladimir Buzalka
2026-04-30 07:09:02 +02:00
parent 2e929f1d77
commit 1b904e3da0
4 changed files with 804 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Insurance/StavPojisteni/batch_stav0_20250101.py
+265
View File
@@ -0,0 +1,265 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
batch_stav0_20250101.py
========================
Zpracuje 50 pacientů, kteří k 1. 1. 2025 vrátili stavVyrizeniPozadavku=0
na dotaz registrace lékaře (= VZP pojištěnce nenašla).
Pro každého:
1. Dotáže VZP stavPojisteni k 2025-01-01 → uloží do vzp_stav_pojisteni.
2. Pokud stav != '1' a != '4', binárně hledá zlom pojištění
a uloží do vzp_sledovani_pojisteni.
Resumovatelný — pacienty, kteří už mají záznam v vzp_stav_pojisteni
k 2025-01-01, přeskočí.
"""
import sys
import time
from pathlib import Path
from datetime import date, timedelta
PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
sys.path.insert(0, str(PROJECT_ROOT))
from Knihovny.mysql_db import connect_mysql
from Knihovny.vzpb2b_client import VZPB2BClient
# ── KONFIGURACE ───────────────────────────────────────────────────────────────
K_DATU = date(2025, 1, 1)
API_DELAY = 1.5
PFX_PATH = Path(__file__).resolve().parent.parent / "Certificates" / "picka.pfx"
PFX_PASSWORD = "Vlado7309208104+"
ICZ = "00000000"
DIC = "00000000"
ENV = "prod"
# ── PACIENTI ─────────────────────────────────────────────────────────────────
PACIENTI = [
("415513130", "Rohlíková", "Marie"),
("420622031", "Hamerník", "Josef"),
("430127082", "Anderle", "Václav"),
("435625017", "Kafková", "Marie"),
("436005111", "Plzáková", "Ivana"),
("445624103", "Kloučková", "Vlasta"),
("446116017", "Strnadová", "Dagmar"),
("456016085", "Kubcová", "Anna"),
("485627038", "Poustková", "Jiřina"),
("506109148", "Holubcová", "Svatava"),
("6008040247", "Šulc", "Jiří"),
("6054574130", "Přibová", "Darina"),
("6102694070", "Elouchefoun", "Aziz"),
("6654251978", "Svozilová", "Ivana"),
("6808292018", "Moudrý", "Jiří"),
("6853222079", "Milatová", "Martina"),
("6909154934", "Novák", "Petr"),
("7056764319", "Michlíková", "Anna"),
("7157734210", "Moudry Molloy", "Joanne"),
("7309300449", "Vojáček", "Aleš"),
("7410540709", "Torres blanco", "Jose maria"),
("7459303599", "Noháčová", "Kateřina"),
("7651090106", "Matějková", "Jana"),
("7803744597", "Barrell", "Peter"),
("7855080420", "Vondřičková", "Zuzana"),
("7908030427", "Smetana", "Libor"),
("7957312099", "Nimeřická", "Michaela"),
("7961794126", "Tomyshynets", "Halyna"),
("8005291404", "Hanzl", "František"),
("8156013041", "Maršíková", "Simona"),
("8157544241", "Jarošová", "Zuzana"),
("8203120299", "Otčenášek", "Vojtěch"),
("8253215223", "Slavíková", "Kateřina"),
("8301070558", "Kříž", "Michal"),
("8352210438", "Bartáková", "Jana"),
("8412175123", "Mičulka", "Jan"),
("8454664262", "Feoktistová", "Irina"),
("8462150147", "Říhová", "Markéta"),
("8503120417", "Šindelář", "Tomáš"),
("8552170517", "Slabá", "Gabriela"),
("8558150227", "Horáková", "Lucie"),
("8652034380", "Kopová", "Jana"),
("8754280403", "Jindrová", "Helena"),
("8755075153", "Babjáková", "Jana"),
("8910584023", "Pham Van", "Duy"),
("8953010330", "Špatná", "Markéta"),
("8956039037", "Slavíková", "Zuzana"),
("9002025956", "Banáš", "Martin"),
("9010262448", "Bečica", "Marek"),
("9811040305", "Sidej", "Natan"),
]
# ── INIT ──────────────────────────────────────────────────────────────────────
vzp = VZPB2BClient(ENV, str(PFX_PATH), PFX_PASSWORD, icz=ICZ, dic=DIC)
mysql = connect_mysql()
call_count = 0
today = date.today()
# ── HELPERS ───────────────────────────────────────────────────────────────────
def check_stav(rc: str, check_date: date) -> str:
global call_count
if call_count > 0:
time.sleep(API_DELAY)
call_count += 1
print(f" [{call_count}] {check_date.isoformat()} ...", end=" ", flush=True)
xml = vzp.stav_pojisteni(rc=rc, k_datu=check_date.isoformat())
stav = vzp.parse_stav_pojisteni(xml)["stav"]
print(f"stav = {stav!r}")
return stav
def uloz_stav(rc, prijmeni, jmeno, k_datu, stav):
xml = vzp.stav_pojisteni(rc=rc, k_datu=k_datu.isoformat())
with mysql.cursor() as cur:
cur.execute("""
INSERT INTO vzp_stav_pojisteni (rc, prijmeni, jmeno, k_datu, stav, response_xml)
VALUES (%s, %s, %s, %s, %s, %s)
ON DUPLICATE KEY UPDATE stav=VALUES(stav), response_xml=VALUES(response_xml)
""", (rc, prijmeni, jmeno, k_datu, stav, xml))
def najdi_zlom(rc: str, last_ok_mysql) -> tuple:
if last_ok_mysql:
low = last_ok_mysql
high = today
print(f" MySQL last stav='1': {low} → [{low} … {high}]")
else:
print(" MySQL: žádný stav='1' — hledám zpětně po rocích ...")
prev_probe = today
low = high = None
for n in range(1, 21):
y = today.year - n
try:
probe = date(y, today.month, today.day)
except ValueError:
probe = date(y, today.month, today.day - 1)
stav = check_stav(rc, probe)
if stav == "1":
low = probe
high = prev_probe
print(f" Nalezeno stav='1' k {low} → [{low} … {high}]")
break
prev_probe = probe
if low is None:
print(" NELZE: stav='1' nenalezen ani 20 let zpět.")
return None, None
stav_low = check_stav(rc, low)
stav_high = check_stav(rc, high)
if stav_low != "1":
print(f" NELZE: dolní mez {low} má stav='{stav_low}'.")
return None, None
if stav_high == "1":
print(f" INFO: horní mez {high} má stav='1' — aktuálně pojištěn.")
return None, None
print(f" Binární hledání ({(high - low).days} dní) ...")
while (high - low).days > 1:
mid = low + timedelta(days=(high - low).days // 2)
stav = check_stav(rc, mid)
if stav == "1":
low = mid
else:
high = mid
return low, high # insured_to, uninsured_from
# ── RESUME: načti již hotové ──────────────────────────────────────────────────
with mysql.cursor() as cur:
cur.execute("SELECT rc FROM vzp_stav_pojisteni WHERE k_datu = %s", (K_DATU,))
hotove = {row[0] for row in cur.fetchall()}
zbyvaji = [(rc, p, j) for rc, p, j in PACIENTI if rc not in hotove]
print(f"Celkem pacientů: {len(PACIENTI)}, již hotovo: {len(hotove)}, zbývá: {len(zbyvaji)}")
print(f"API prodleva: {API_DELAY}s | K_DATU: {K_DATU}\n")
print("=" * 60)
# ── HLAVNÍ SMYČKA ─────────────────────────────────────────────────────────────
vysledky = []
for i, (rc, prijmeni, jmeno) in enumerate(zbyvaji, 1):
print(f"\n[{i}/{len(zbyvaji)}] {prijmeni} {jmeno} (RC: {rc})")
# Krok 1: stav k K_DATU
if call_count > 0:
time.sleep(API_DELAY)
call_count += 1
print(f" [{call_count}] {K_DATU.isoformat()} ...", end=" ", flush=True)
xml_hist = vzp.stav_pojisteni(rc=rc, k_datu=K_DATU.isoformat())
stav_hist = vzp.parse_stav_pojisteni(xml_hist)["stav"]
print(f"stav = {stav_hist!r}")
with mysql.cursor() as cur:
cur.execute("""
INSERT INTO vzp_stav_pojisteni (rc, prijmeni, jmeno, k_datu, stav, response_xml)
VALUES (%s, %s, %s, %s, %s, %s)
ON DUPLICATE KEY UPDATE stav=VALUES(stav), response_xml=VALUES(response_xml)
""", (rc, prijmeni, jmeno, K_DATU, stav_hist, xml_hist))
# Krok 2: zlom (jen pokud stav != '1' a != '4')
insured_to = uninsured_from = None
if stav_hist in ("1", "4"):
print(f" → pojištěn (stav={stav_hist!r}), zlom se nehledá")
else:
# Zkontroluj watchlist
with mysql.cursor() as cur:
cur.execute("SELECT insured_to, uninsured_from FROM vzp_sledovani_pojisteni WHERE rc = %s", (rc,))
existuje = cur.fetchone()
if existuje:
insured_to, uninsured_from = existuje
print(f" → již ve watchlistu: insured_to={insured_to}, uninsured_from={uninsured_from}")
else:
with mysql.cursor() as cur:
cur.execute(
"SELECT MAX(k_datu) FROM vzp_stav_pojisteni WHERE rc = %s AND stav = '1'",
(rc,)
)
r = cur.fetchone()
last_ok = r[0] if r and r[0] else None
insured_to, uninsured_from = najdi_zlom(rc, last_ok)
with mysql.cursor() as cur:
cur.execute("""
INSERT INTO vzp_sledovani_pojisteni
(rc, prijmeni, jmeno, insured_to, uninsured_from,
aktualni_stav, prvni_detekce, posledni_kontrola)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
ON DUPLICATE KEY UPDATE
insured_to=VALUES(insured_to),
uninsured_from=VALUES(uninsured_from),
aktualni_stav=VALUES(aktualni_stav),
posledni_kontrola=VALUES(posledni_kontrola)
""", (rc, prijmeni, jmeno, insured_to, uninsured_from,
stav_hist, today, today))
vysledky.append({
"rc": rc, "prijmeni": prijmeni, "jmeno": jmeno,
"stav_20250101": stav_hist,
"insured_to": insured_to,
"uninsured_from": uninsured_from,
})
# ── SOUHRN ────────────────────────────────────────────────────────────────────
mysql.close()
print(f"\n{'=' * 60}")
print(f" SOUHRN — {len(vysledky)} zpracovaných pacientů")
print(f"{'=' * 60}")
print(f" {'Příjmení':<20} {'Jméno':<14} {'Stav':>5} {'Pojištěn do':<13} {'Nepoj. od'}")
print(f" {'-'*58}")
for v in vysledky:
ins = str(v["insured_to"]) if v["insured_to"] else "-"
uni = str(v["uninsured_from"]) if v["uninsured_from"] else "-"
print(f" {v['prijmeni']:<20} {v['jmeno']:<14} {v['stav_20250101']:>5} {ins:<13} {uni}")
print(f"\nCelkem VZP dotazů: {call_count}")
Insurance/StavPojisteni/zkontroluj_rc_jednorazove.py
+177
View File
@@ -0,0 +1,177 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
zkontroluj_rc_jednorazove.py
=============================
Jednorázový skript pro libovolné RC:
1. Dotáže VZP na stav pojištění k zadanému K_DATU.
2. Uloží do vzp_stav_pojisteni.
3. Pokud stav != '1' a != '4', spustí binární hledání zlomu
a uloží do vzp_sledovani_pojisteni.
"""
import sys
import time
from pathlib import Path
from datetime import date, timedelta
PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
sys.path.insert(0, str(PROJECT_ROOT))
from Knihovny.mysql_db import connect_mysql
from Knihovny.vzpb2b_client import VZPB2BClient
# ── KONFIGURACE ───────────────────────────────────────────────────────────────
RC = "430127082"
PRIJMENI = "Anderle"
JMENO = "Václav"
K_DATU = date(2025, 1, 1) # datum pro první dotaz
PFX_PATH = Path(__file__).resolve().parent.parent / "Certificates" / "picka.pfx"
PFX_PASSWORD = "Vlado7309208104+"
ICZ = "00000000"
DIC = "00000000"
ENV = "prod"
API_DELAY = 2
# ── INIT ──────────────────────────────────────────────────────────────────────
vzp = VZPB2BClient(ENV, str(PFX_PATH), PFX_PASSWORD, icz=ICZ, dic=DIC)
mysql = connect_mysql()
call_count = 0
today = date.today()
def check_stav(rc: str, check_date: date) -> str:
global call_count
if call_count > 0:
time.sleep(API_DELAY)
call_count += 1
print(f" [{call_count}] VZP dotaz k {check_date.isoformat()} ...", end=" ", flush=True)
xml = vzp.stav_pojisteni(rc=rc, k_datu=check_date.isoformat())
stav = vzp.parse_stav_pojisteni(xml)["stav"]
print(f"stav = {stav!r}")
return stav
def uloz_stav(rc, prijmeni, jmeno, k_datu, stav, xml):
with mysql.cursor() as cur:
cur.execute("""
INSERT INTO vzp_stav_pojisteni
(rc, prijmeni, jmeno, k_datu, stav, response_xml)
VALUES (%s, %s, %s, %s, %s, %s)
ON DUPLICATE KEY UPDATE stav=VALUES(stav), response_xml=VALUES(response_xml)
""", (rc, prijmeni, jmeno, k_datu, stav, xml))
def najdi_zlom(rc, last_ok_mysql):
if last_ok_mysql:
low = last_ok_mysql
high = today
print(f" MySQL last stav='1': {low} -> [{low} ... {high}]")
else:
print(" MySQL: žádný stav='1' — hledám zpětně po rocích ...")
prev_probe = today
low = high = None
for n in range(1, 21):
y = today.year - n
try:
probe = date(y, today.month, today.day)
except ValueError:
probe = date(y, today.month, today.day - 1)
stav = check_stav(rc, probe)
if stav == "1":
low = probe
high = prev_probe
print(f" Nalezeno stav='1' k {low} -> [{low} ... {high}]")
break
prev_probe = probe
if low is None:
print(" NELZE: stav='1' nenalezen ani 20 let zpět.")
return None, None
stav_low = check_stav(rc, low)
stav_high = check_stav(rc, high)
if stav_low != "1":
print(f" NELZE: dolní mez {low} má stav='{stav_low}'.")
return None, None
if stav_high == "1":
print(f" INFO: horní mez {high} má stav='1' — pacient je aktuálně pojištěn, žádný zlom.")
return None, None
print(f" Binární hledání v rozsahu {(high - low).days} dní ...")
while (high - low).days > 1:
mid = low + timedelta(days=(high - low).days // 2)
stav = check_stav(rc, mid)
if stav == "1":
low = mid
else:
high = mid
return low, high # insured_to, uninsured_from
# ── KROK 1: Dotaz k zadanému datu ─────────────────────────────────────────────
print(f"\n{PRIJMENI} {JMENO} (RC: {RC})")
print(f"── Krok 1: stav pojištění k {K_DATU} ─────────────────────────────────")
xml_hist = vzp.stav_pojisteni(rc=RC, k_datu=K_DATU.isoformat())
stav_hist = vzp.parse_stav_pojisteni(xml_hist)["stav"]
call_count += 1
print(f" stav k {K_DATU}: {stav_hist!r}")
uloz_stav(RC, PRIJMENI, JMENO, K_DATU, stav_hist, xml_hist)
print(f" Uloženo do vzp_stav_pojisteni (k_datu={K_DATU})")
# ── KROK 2: Pokud není pojištěn, najdi zlom ───────────────────────────────────
if stav_hist in ("1", "4"):
print(f"\nStav '{stav_hist}' = pojištěn (nebo cizinec), zlom se nehledá.")
else:
print(f"\nStav '{stav_hist}' = nepojištěn/nenalezen")
print(f"── Krok 2: hledám zlom pojištění ─────────────────────────────────────")
# Zkontroluj, zda je pacient již ve watchlistu
with mysql.cursor() as cur:
cur.execute("SELECT insured_to, uninsured_from FROM vzp_sledovani_pojisteni WHERE rc = %s", (RC,))
existuje = cur.fetchone()
if existuje:
print(f" Pacient již ve watchlistu: insured_to={existuje[0]}, uninsured_from={existuje[1]}")
print(f" Přeskakuji hledání zlomu.")
else:
with mysql.cursor() as cur:
cur.execute(
"SELECT MAX(k_datu) FROM vzp_stav_pojisteni WHERE rc = %s AND stav = '1'",
(RC,)
)
r = cur.fetchone()
last_ok = r[0] if r and r[0] else None
insured_to, uninsured_from = najdi_zlom(RC, last_ok)
with mysql.cursor() as cur:
cur.execute("""
INSERT INTO vzp_sledovani_pojisteni
(rc, prijmeni, jmeno, insured_to, uninsured_from,
aktualni_stav, prvni_detekce, posledni_kontrola)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
ON DUPLICATE KEY UPDATE
insured_to=VALUES(insured_to),
uninsured_from=VALUES(uninsured_from),
aktualni_stav=VALUES(aktualni_stav),
posledni_kontrola=VALUES(posledni_kontrola)
""", (RC, PRIJMENI, JMENO, insured_to, uninsured_from,
stav_hist, today, today))
print(f"\n{'=' * 50}")
print(f" VÝSLEDEK — {PRIJMENI} {JMENO} (RC: {RC})")
print(f"{'=' * 50}")
print(f" Stav k {K_DATU} : {stav_hist!r}")
print(f" Poslední den pojištěn : {insured_to or 'nezjištěno'}")
print(f" První den bez pojištění : {uninsured_from or 'nezjištěno'}")
print(f" Celkem VZP dotazů : {call_count}")
print(f"{'=' * 50}\n")
mysql.close()
print(f"Hotovo. Celkem VZP dotazů: {call_count}")