This commit is contained in:
2026-04-28 16:40:04 +02:00
parent 6cff5f1b91
commit b1f246bc54
2 changed files with 510 additions and 0 deletions
@@ -0,0 +1,146 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Najde přesný den zlomu pojištění pro konkrétního pacienta.
Algoritmus:
1. Z MySQL vezme MAX(k_datu) WHERE stav='1' pro daného pacienta.
Pokud neexistuje, prochází zpět po 1 roce a hledá první stav='1'.
2. Binárním hledáním najde přesný den:
low = poslední den kdy byl stav='1'
high = první den kdy byl stav!='1'
"""
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 = "500208129"
PRIJMENI = "Zuzák"
JMENO = "Viktor"
PFX_PATH = Path(__file__).resolve().parent.parent / "Certificates" / "picka.pfx"
PFX_PASSWORD = "Vlado7309208104+"
ICZ = "00000000"
DIC = "00000000"
ENV = "prod"
API_DELAY = 2 # sekundy mezi dotazy na VZP
# ── INIT VZP ──────────────────────────────────────────────────────────────────
if not PFX_PATH.exists():
print(f"CHYBA: PFX certifikat nenalezen: {PFX_PATH}")
sys.exit(1)
vzp = VZPB2BClient(ENV, str(PFX_PATH), PFX_PASSWORD, icz=ICZ, dic=DIC)
call_count = 0
def check_stav(rc: str, check_date: date) -> str | None:
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
# ── KROK 1: DOLNÍ MEZ ─────────────────────────────────────────────────────────
print(f"\nHledám zlom pojištění pro {PRIJMENI} {JMENO} (RC: {RC})\n")
print("── Krok 1: dolní mez ──────────────────────────────────────────────────")
mysql = connect_mysql()
with mysql.cursor() as cur:
cur.execute(
"SELECT MAX(k_datu) FROM vzp_stav_pojisteni WHERE rc = %s AND stav = '1'",
(RC,)
)
row = cur.fetchone()
mysql.close()
today = date.today()
last_ok = row[0] if row and row[0] else None
if last_ok:
low = last_ok
high = today
print(f" MySQL: poslední stav='1' k datu {low}")
print(f" → binary search [{low}{high}]")
else:
print(" MySQL: žádný záznam se stavem='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"\n Nalezeno stav='1' k {low}")
print(f" → binary search [{low}{high}]")
break
prev_probe = probe
if low is None:
print("CHYBA: Stav='1' nenalezen ani 20 let zpatky. Nelze urcit zlom.")
sys.exit(1)
# ── KROK 2: OVĚŘENÍ HRANIC ────────────────────────────────────────────────────
print("\n── Krok 2: ověření hranic ─────────────────────────────────────────────")
stav_low = check_stav(RC, low)
stav_high = check_stav(RC, high)
if stav_low != "1":
print(f"CHYBA: Dolni mez {low} ma stav='{stav_low}' (ocekavam '1'). Nelze pokracovat.")
sys.exit(1)
if stav_high == "1":
print(f"INFO: Horni mez {high} ma stav='1' — pacient je aktualne pojisten, zadny zlom nenalezen.")
sys.exit(0)
print(f" OK {low} -> '{stav_low}' | {high} -> '{stav_high}' — rozsah v poradku")
# ── KROK 3: BINÁRNÍ HLEDÁNÍ ───────────────────────────────────────────────────
print(f"\n── Krok 3: binární hledání ({(high - low).days} dní v rozsahu) ─────────")
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
# ── VÝSLEDEK ──────────────────────────────────────────────────────────────────
print(f"\n{'=' * 55}")
print(f" VYSLEDEK - {PRIJMENI} {JMENO} (RC: {RC})")
print(f"{'=' * 55}")
print(f" Posledni den POJISTEN : {low}")
print(f" Prvni den BEZ pojisteni: {high}")
print(f" Celkem VZP dotazu : {call_count}")
print(f"{'=' * 55}\n")
@@ -0,0 +1,364 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
zkontroluj_a_odesli_zlomy.py
=============================
Ucel:
Sleduje registrovane pacienty, kteri maji problematicky stav pojisteni
(stav "X" = pojistovna nenalezena / nepojisten). Detekuje zmeny stavu
a odesila email s vysledky. Spousti se ad hoc nebo planovane.
Zavislosti:
Knihovny/vzpb2b_client.py -- VZP B2B SOAP API klient (mTLS pres picka.pfx)
Knihovny/mysql_db.py -- pripojeni k MySQL 192.168.1.76, DB medevio
Knihovny/EmailMessagingGraph.py -- odesilani emailu pres Microsoft Graph
MySQL tabulky:
vzp_stav_pojisteni -- denni zaznamy VZP dotazu (plni FinalSaveInsuranceScript)
vzp_sledovani_pojisteni -- watchlist pacientu se stavem X (spravuje tento skript)
vzp_sledovani_zmeny -- log vsech detektvanych zmen stavu (spravuje tento skript)
Logika:
FAZE 1 -- Re-overeni watchlistu
Pro kazdeho pacienta v tabulce vzp_sledovani_pojisteni zavola VZP API
k dnesnemu datumu. Pokud se stav zmenil (X->1, 1->X apod.), zapise
zmenu do vzp_sledovani_zmeny a aktualizuje aktualni_stav.
FAZE 2 -- Novi pacienti
Z tabulky vzp_stav_pojisteni vybere pacienty, jejichz POSLEDNI zaznam
ma stav NOT IN ('1','4') a jeste nejsou ve watchlistu.
Pro kazdeho:
a) Hleda dolni mez (posledni datum stav='1'):
- Nejprve MAX(k_datu WHERE stav='1') z MySQL.
- Pokud neni, prochazi zpetne po 1 roce az najde stav='1' (max 20 let).
b) Binarnim hledanim zpresni na konkretni den zlomu:
low = posledni den pojisten (stav='1')
high = prvni den bez pojisteni (stav!='1')
c) Vlozi pacienta do vzp_sledovani_pojisteni.
FAZE 3 -- Email
Vzdy odesle email na vladimir.buzalka@buzalka.cz.
Struktura emailu:
[NAHORE] Novi pacienti pridani toto spusteni (s datem zlomu)
[DOLE] Vsechny historicke zmeny z vzp_sledovani_zmeny, razene
podle datum_zmeny DESC
Poznamky:
- Mezi kazdym VZP API volanim je 2s prodleva (API_DELAY).
- Stav '4' (cizinec bez plneho naroku) se povazuje za OK, nesledi se.
- Tabulky se vytvori automaticky pri prvnim spusteni (CREATE TABLE IF NOT EXISTS).
- Kolace vsech tabulek: utf8mb4_unicode_ci (shoduje se s vzp_stav_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
from Knihovny.EmailMessagingGraph import send_mail
# ── KONFIGURACE ───────────────────────────────────────────────────────────────
EMAIL_PRIJEMCE = "vladimir.buzalka@buzalka.cz"
PFX_PATH = Path(__file__).resolve().parent.parent / "Certificates" / "picka.pfx"
PFX_PASSWORD = "Vlado7309208104+"
ICZ = "00000000"
DIC = "00000000"
ENV = "prod"
API_DELAY = 2
# ── INIT ──────────────────────────────────────────────────────────────────────
if not PFX_PATH.exists():
print(f"CHYBA: PFX certifikat nenalezen: {PFX_PATH}")
sys.exit(1)
vzp = VZPB2BClient(ENV, str(PFX_PATH), PFX_PASSWORD, icz=ICZ, dic=DIC)
call_count = 0
today = date.today()
# ── HELPERS ───────────────────────────────────────────────────────────────────
def check_stav(rc: str, check_date: date) -> str | None:
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 najdi_zlom(rc: str, last_ok_mysql) -> tuple:
"""Vrati (insured_to, uninsured_from) nebo (None, None) pri selhani."""
if last_ok_mysql:
low = last_ok_mysql
high = today
print(f" MySQL last stav='1': {low} -> [{low} ... {high}]")
else:
print(" MySQL: zadny stav='1' — hledam zpetne po rocich ...")
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 zpet.")
return None, None
stav_low = check_stav(rc, low)
stav_high = check_stav(rc, high)
if stav_low != "1":
print(f" NELZE: dolni mez {low} ma stav='{stav_low}'.")
return None, None
if stav_high == "1":
print(f" INFO: horni mez {high} ma stav='1' — bez zlomu.")
return None, None
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
# ── MYSQL: INIT TABULEK ───────────────────────────────────────────────────────
mysql = connect_mysql()
with mysql.cursor() as cur:
cur.execute("""
CREATE TABLE IF NOT EXISTS vzp_sledovani_pojisteni (
rc VARCHAR(20) NOT NULL PRIMARY KEY,
prijmeni VARCHAR(100),
jmeno VARCHAR(100),
insured_to DATE,
uninsured_from DATE,
aktualni_stav VARCHAR(10),
prvni_detekce DATE NOT NULL,
posledni_kontrola DATE NOT NULL
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
""")
cur.execute("""
CREATE TABLE IF NOT EXISTS vzp_sledovani_zmeny (
id INT AUTO_INCREMENT PRIMARY KEY,
rc VARCHAR(20) NOT NULL,
prijmeni VARCHAR(100),
jmeno VARCHAR(100),
datum_zmeny DATE NOT NULL,
stav_pred VARCHAR(10),
stav_po VARCHAR(10),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
INDEX idx_rc (rc),
INDEX idx_datum (datum_zmeny)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
""")
# ── FAZE 1: RE-OVERENI WATCHLISTU ─────────────────────────────────────────────
print(f"\n== Faze 1: Re-overeni watchlistu ({today}) ==")
with mysql.cursor() as cur:
cur.execute("""
SELECT rc, prijmeni, jmeno, aktualni_stav
FROM vzp_sledovani_pojisteni
ORDER BY prijmeni, jmeno
""")
watchlist = cur.fetchall()
print(f"Watchlist: {len(watchlist)} pacientu\n")
tato_zmeny = [] # zmeny detekované toto spusteni
for rc, prijmeni, jmeno, stav_pred in watchlist:
print(f" {prijmeni} {jmeno} (RC: {rc}) aktualni={stav_pred!r}")
stav_po = check_stav(rc, today)
with mysql.cursor() as cur:
cur.execute(
"UPDATE vzp_sledovani_pojisteni SET aktualni_stav=%s, posledni_kontrola=%s WHERE rc=%s",
(stav_po, today, rc)
)
if stav_po != stav_pred:
print(f" *** ZMENA: {stav_pred!r} -> {stav_po!r} ***")
with mysql.cursor() as cur:
cur.execute("""
INSERT INTO vzp_sledovani_zmeny (rc, prijmeni, jmeno, datum_zmeny, stav_pred, stav_po)
VALUES (%s, %s, %s, %s, %s, %s)
""", (rc, prijmeni, jmeno, today, stav_pred, stav_po))
tato_zmeny.append((rc, prijmeni, jmeno, stav_pred, stav_po))
# ── FAZE 2: NOVI PACIENTI S X ─────────────────────────────────────────────────
print(f"\n== Faze 2: Novi pacienti s X ==")
with mysql.cursor() as cur:
cur.execute("""
SELECT rc, prijmeni, jmeno, stav
FROM (
SELECT rc, prijmeni, jmeno, stav,
ROW_NUMBER() OVER (PARTITION BY rc ORDER BY k_datu DESC) AS rn
FROM vzp_stav_pojisteni
) t
WHERE rn = 1
AND stav NOT IN ('1', '4')
AND rc NOT IN (SELECT rc FROM vzp_sledovani_pojisteni)
ORDER BY prijmeni, jmeno
""")
novi_raw = cur.fetchall()
print(f"Novych pacientu: {len(novi_raw)}\n")
novi = []
for rc, prijmeni, jmeno, stav in novi_raw:
print(f" Novy: {prijmeni} {jmeno} (RC: {rc}) stav={stav!r}")
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)
""", (rc, prijmeni, jmeno, insured_to, uninsured_from, stav, today, today))
novi.append({
"rc": rc,
"prijmeni": prijmeni,
"jmeno": jmeno,
"stav": stav,
"insured_to": str(insured_to) if insured_to else "nezjisteno",
"uninsured_from": str(uninsured_from) if uninsured_from else "nezjisteno",
})
print()
# ── FAZE 3: NACTENI VSECH HISTORICKYCH ZMEN ──────────────────────────────────
with mysql.cursor() as cur:
cur.execute("""
SELECT rc, prijmeni, jmeno, datum_zmeny, stav_pred, stav_po
FROM vzp_sledovani_zmeny
ORDER BY datum_zmeny DESC, created_at DESC
""")
vsechny_zmeny = cur.fetchall()
mysql.close()
# ── SESTAVENI EMAILU ──────────────────────────────────────────────────────────
# --- Sekce 1: NOVI ---
if novi:
radky = "".join(f"""
<tr>
<td>{p['prijmeni']} {p['jmeno']}</td>
<td>{p['rc']}</td>
<td style="color:#c0392b;font-weight:bold">{p['stav']}</td>
<td>{p['insured_to']}</td>
<td style="color:#c0392b">{p['uninsured_from']}</td>
</tr>""" for p in novi)
sekce_novi = f"""
<h2 style="background:#c0392b;color:white;padding:8px 12px;margin-top:0">
Novi pacienti pridani do sledovani &mdash; {len(novi)}
</h2>
<table border="1" cellpadding="6" cellspacing="0"
style="border-collapse:collapse;border-color:#ccc;width:100%;margin-bottom:24px">
<thead style="background:#922b21;color:white">
<tr>
<th>Pacient</th><th>RC</th><th>Stav</th>
<th>Posledni den pojisten</th><th>Prvni den bez pojisteni</th>
</tr>
</thead>
<tbody>{radky}</tbody>
</table>"""
else:
sekce_novi = """
<h2 style="background:#27ae60;color:white;padding:8px 12px;margin-top:0">
Zadni novi pacienti
</h2>"""
# --- Sekce 2: VSECHNY HISTORICKE ZMENY ---
if vsechny_zmeny:
radky = ""
for rc, prijmeni, jmeno, datum_zmeny, stav_pred, stav_po in vsechny_zmeny:
# Zvyraznit radky z tohoto spusteni
je_dnes = (datum_zmeny == today)
bg = "#fff9e6" if je_dnes else "white"
barva = "#c0392b" if (stav_po or "") != "1" else "#27ae60"
radky += f"""
<tr style="background:{bg}">
<td>{prijmeni} {jmeno}</td>
<td>{rc}</td>
<td>{datum_zmeny}</td>
<td style="font-weight:bold">{stav_pred or '?'}</td>
<td style="color:{barva};font-weight:bold">{stav_po or '?'}</td>
</tr>"""
sekce_zmeny = f"""
<h2 style="background:#2c3e50;color:white;padding:8px 12px">
Vsechny zmeny stavu pojisteni &mdash; {len(vsechny_zmeny)} zaznam(u), razeno od nejnovejiho
</h2>
<p style="font-size:12px;color:#888">Zbarvene radky = dnesni spusteni</p>
<table border="1" cellpadding="6" cellspacing="0"
style="border-collapse:collapse;border-color:#ccc;width:100%">
<thead style="background:#2c3e50;color:white">
<tr>
<th>Pacient</th><th>RC</th><th>Datum zmeny</th>
<th>Stav pred</th><th>Stav po</th>
</tr>
</thead>
<tbody>{radky}</tbody>
</table>"""
else:
sekce_zmeny = """
<h2 style="background:#888;color:white;padding:8px 12px">
Zadne zmeny stavu v historii
</h2>"""
body = f"""<html><body style="font-family:Arial,sans-serif;font-size:14px;max-width:900px">
<h1 style="border-bottom:2px solid #2c3e50;padding-bottom:8px">
Sledovani pojisteni &mdash; {today.strftime('%d. %m. %Y')}
</h1>
{sekce_novi}
{sekce_zmeny}
<p style="color:#aaa;font-size:11px;margin-top:32px">
Celkem VZP dotazu: {call_count} &nbsp;|&nbsp; {today}
</p>
</body></html>"""
predmet = f"Pojisteni {today.strftime('%d.%m.%Y')} {len(novi)} novych, {len(tato_zmeny)} zmen dnes"
print(f"\nOdesilam email na {EMAIL_PRIJEMCE} ...")
send_mail(to=EMAIL_PRIJEMCE, subject=predmet, body=body, html=True)
print(f"Email odeslan.")
print(f"Hotovo. VZP dotazu celkem: {call_count}")