Přejmenování Insurance/KdoJeLékař → KdoJeLekar (bez diakritiky)

Adresář přejmenován bez diakritiky kvůli problémům s kódováním cesty.
Historie zachována přes git mv. Nadpis v NOTES.md sjednocen s názvem.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Vladimir Buzalka
2026-06-15 23:10:59 +02:00
parent 79216dfbdb
commit e23d61de84
7 changed files with 1 additions and 1 deletions
+108
View File
@@ -0,0 +1,108 @@
# KdoJeLekar — poznámky k vývoji
## Cíl
Zjistit pro pacienty z Medicus DB, kdo je jejich registrující **praktický lékař (001)**, **gynekolog (002)** a **stomatolog (014)** — dotazem na VZP B2B portál.
---
## Stav k 29. 4. 2026 — hotovo
- Certifikát ✅, tabulky ✅, produkční skript ✅
- Připraveno ke spuštění — přepnout `TEST_MODE = False`
---
## Soubory v tomto adresáři
| Soubor | Popis |
|--------|-------|
| `kdojelekar_tydenni.py` | Produkční skript — batch všech pacientů, ukládá do MySQL |
| `_test_temp.py` | Testovací skript — dotaz na jedno RC, výpis XML + parsovaný výsledek |
| `_test_no_odb.py` | Test bez filtru odborností — sloužil k ověření struktury odpovědi |
---
## Certifikát
**`u:\ordinaceprojekt\Insurance\Certificates\picka.pfx`** / heslo **`Vlado7309208104+`**
Ověřeno 29. 4. 2026 (HTTP 200). Stejný certifikát používá i `StavPojisteni\zkontroluj_a_odesli_zlomy.py`.
---
## VZP B2B služba: `RegistracePojistencePZSB2B`
### Endpoint (produkce)
```
https://prod.b2b.vzp.cz/B2BProxy/HttpProxy/RegistracePojistencePZSB2B
```
### Autentizace
mTLS — klientský certifikát `.pfx`, stejný mechanismus jako u `stavPojisteniB2B`.
### Struktura odpovědi
Pro každou odbornost kde má pacient lékaře vrátí jeden `<odbornost>` element.
Pokud lékař není, VZP element vynechá — skript ukládá placeholder řádek s `ma_lekare=0`.
| XML tag | Uloženo jako | Popis |
|---------|-------------|-------|
| `ICZ` | `ICZ` | IČZ zdravotnického zařízení |
| `ICP` | `ICP` | IČP lékaře |
| `nazevICP` | `nazev_lekare` | Název pracoviště |
| `nazevSZZ` | `nazev_zzz` | Jméno lékaře |
| `zdravotniPojistovna/kod` | `poj_kod` | Kód pojišťovny pacienta |
| `zdravotniPojistovna/zkratka` | `poj_zkratka` | Zkratka pojišťovny |
| `odbornost/kod` | `kod_odbornosti` | Kód odbornosti (001/002/014) |
| `datumRegistrace` | `datum_registrace` | Kdy pacient podepsal registraci |
| `datumZahajeni` | `datum_zahajeni` | Od kdy registrace platí u VZP |
| `datumUkonceni` | `datum_ukonceni` | Do kdy (3000-01-01 = bez konce) |
| `stavVyrizeniPozadavku` | `stav_vyrizeni` | Stavový kód odpovědi VZP |
**Poznámka k parsování:** VZP vrací pro každý nalezený záznam dva `<odbornost>` elementy —
vnější (s ICZ/ICP/jménem) a vnořený subelement (jen kód+název). Parser používá
`findall(".//seznamOdbornosti/odbornost")` který zachytí jen vnější.
---
## MySQL tabulky
### `vzp_registrace_lekari`
Jeden řádek na `(rc, k_datu, kod_odbornosti)`. UNIQUE klíč = `(rc, k_datu, kod_odbornosti)`.
Historie se hromadí — každý týdenní běh přidá nové řádky.
### `vzp_registrace_raw`
Jeden řádek na `(rc, k_datu)` — celé raw XML odpovědi.
Slouží k případnému přepočtu bez opakování API dotazů. UNIQUE klíč = `(rc, k_datu)`.
---
## Produkční skript `kdojelekar_tydenni.py`
### Konfigurace (začátek souboru)
| Proměnná | Výchozí | Popis |
|----------|---------|-------|
| `API_PAUSE` | `2` | Sekundy mezi VZP dotazy |
| `TEST_MODE` | `True` | False = produkční běh |
| `ODBORNOSTI` | `["001","002","014"]` | Dotazované odbornosti |
### Logika
1. Načte aktivně registrované pacienty z Medicus (přesný select dle SELECTS.md, IČP 09305001)
2. V produkčním běhu přeskočí pacienty, kteří už mají záznam v `vzp_registrace_raw` pro dnešní datum — **resumovatelný běh**
3. Pro každého pacienta zavolá VZP B2B, uloží raw XML + parsované záznamy
4. Placeholdery pro odbornosti bez lékaře ukládá s `ma_lekare=0`
### Knihovny
- `Knihovny/vzpb2b_client.py` → metody `registrace_lekare()` a `parse_registrace_lekare()`
- `Knihovny/medicus_db.py``get_active_registered_patients()` (opraveno 29. 4. 2026)
- `Knihovny/mysql_db.py``connect_mysql()`
---
## Plán dalšího postupu
1. ~~Certifikát~~ — vyřešeno, `picka.pfx` / `Vlado7309208104+`
2. ~~Ověřit funkčnost~~ — hotovo, HTTP 200 s daty
3. ~~Produkční skript~~ — hotovo, `kdojelekar_tydenni.py`
4. ~~MySQL tabulky~~ — hotovo, `vzp_registrace_lekari` + `vzp_registrace_raw`
5. Naplánovat týdenní spouštění (Windows Task Scheduler nebo Claude schedule)
6. Zvážit detekci změn lékaře (analogie zlomů u StavPojisteni) — zatím není v plánu
@@ -0,0 +1,42 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import sys, requests
sys.stdout.reconfigure(encoding='utf-8')
from requests_pkcs12 import Pkcs12Adapter
from datetime import date
import xml.etree.ElementTree as ET
ENDPOINT = "https://prod.b2b.vzp.cz/B2BProxy/HttpProxy/RegistracePojistencePZSB2B"
PFX_PATH = r"/Insurance/Certificates/picka.pfx"
PFX_PASS = "Vlado7309208104+"
NS = {
"soap": "http://schemas.xmlsoap.org/soap/envelope/",
"rp": "http://xmlns.gemsystem.cz/B2B/RegistracePojistencePZSB2B/1",
}
envelope = """<?xml version="1.0" encoding="utf-8"?>
<soap:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/">
<soap:Body>
<ns1:registracePojistencePZSB2B xmlns:ns1="http://xmlns.gemsystem.cz/B2B/RegistracePojistencePZSB2B/1">
<ns1:cisloPojistence>7309208104</ns1:cisloPojistence>
<ns1:kDatu>2026-04-29</ns1:kDatu>
</ns1:registracePojistencePZSB2B>
</soap:Body>
</soap:Envelope>"""
session = requests.Session()
session.mount("https://", Pkcs12Adapter(pkcs12_filename=PFX_PATH, pkcs12_password=PFX_PASS))
resp = session.post(ENDPOINT, data=envelope.encode("utf-8"),
headers={"Content-Type": "text/xml; charset=utf-8", "SOAPAction": "process"},
timeout=30, verify=True)
print(f"HTTP: {resp.status_code}")
root = ET.fromstring(resp.text)
items = root.findall(".//rp:odbornost", NS)
print(f"Pocet odbornosti: {len(items)}")
for i, it in enumerate(items):
print(f"\n--- odbornost #{i+1} ---")
print(ET.tostring(it, encoding="unicode"))
st = root.find(".//rp:stavVyrizeniPozadavku", NS)
print(f"stav: {st.text if st is not None else '?'}")
+70
View File
@@ -0,0 +1,70 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import sys
sys.path.insert(0, r"u:\insurance")
from requests_pkcs12 import Pkcs12Adapter
import requests
import xml.etree.ElementTree as ET
from datetime import date
ENDPOINT = "https://prod.b2b.vzp.cz/B2BProxy/HttpProxy/RegistracePojistencePZSB2B"
PFX_PATH = r"/Insurance/Certificates/picka.pfx"
PFX_PASS = "Vlado7309208104+"
RC = "7309208104"
K_DATU = date.today().isoformat()
NS = {
"soap": "http://schemas.xmlsoap.org/soap/envelope/",
"rp": "http://xmlns.gemsystem.cz/B2B/RegistracePojistencePZSB2B/1",
}
envelope = f"""<?xml version="1.0" encoding="utf-8"?>
<soap:Envelope xmlns:soap="{NS['soap']}">
<soap:Body>
<ns1:registracePojistencePZSB2B xmlns:ns1="{NS['rp']}">
<ns1:cisloPojistence>{RC}</ns1:cisloPojistence>
<ns1:kDatu>{K_DATU}</ns1:kDatu>
</ns1:registracePojistencePZSB2B>
</soap:Body>
</soap:Envelope>"""
session = requests.Session()
session.mount("https://", Pkcs12Adapter(pkcs12_filename=PFX_PATH, pkcs12_password=PFX_PASS))
resp = session.post(ENDPOINT, data=envelope.encode("utf-8"),
headers={"Content-Type": "text/xml; charset=utf-8", "SOAPAction": "process"},
timeout=30, verify=True)
print(f"HTTP: {resp.status_code}\n")
print("=== RAW XML ===")
print(resp.text)
print("\n=== PARSED ===")
root = ET.fromstring(resp.text)
items = root.findall(".//rp:seznamOdbornosti/rp:odbornost", NS)
if not items:
st = root.find(".//rp:stavVyrizeniPozadavku", NS)
print(f"Žádné záznamy. stavVyrizeniPozadavku={st.text if st is not None else '?'}")
else:
for it in items:
def g(tag):
el = it.find(f"rp:{tag}", NS)
return el.text.strip() if el is not None and el.text else None
odb = it.find("rp:odbornost", NS)
odb_kod = odb.find("rp:kod", NS).text.strip() if odb is not None and odb.find("rp:kod", NS) is not None else None
odb_naz = odb.find("rp:nazev", NS).text.strip() if odb is not None and odb.find("rp:nazev", NS) is not None else None
print(f" odbornost: {odb_kod} {odb_naz}")
print(f" ICZ: {g('ICZ')}")
print(f" ICP: {g('ICP')}")
print(f" nazevICP: {g('nazevICP')}")
print(f" nazevSZZ: {g('nazevSZZ')}")
print(f" datumRegistrace: {g('datumRegistrace')}")
print(f" datumZahajeni: {g('datumZahajeni')}")
print(f" datumUkonceni: {g('datumUkonceni')}")
print()
st = root.find(".//rp:stavVyrizeniPozadavku", NS)
print(f"stavVyrizeniPozadavku: {st.text.strip() if st is not None and st.text else '?'}")
@@ -0,0 +1,362 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Exportuje 151 pacientů registrovaných v Medicusu k 1.1.2025,
u nichž VZP k tomuto datu nevykazuje registraci v odbornosti 001 u IČP 09305001.
Výstup: Excel s komentářem a aktuálním stavem v Medicusu.
"""
import sys
from pathlib import Path
from datetime import date
from collections import defaultdict
sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent))
from Knihovny.mysql_db import connect_mysql
import fdb, socket
from openpyxl import Workbook
from openpyxl.styles import (Font, PatternFill, Alignment, Border, Side,
GradientFill)
from openpyxl.utils import get_column_letter
# ── Konfigurace ────────────────────────────────────────────────────────────────
K_DATU_HIST = "2025-01-01"
TODAY = date.today()
OUT_FILE = Path(__file__).resolve().parent / f"neregistrovani_vzp_20250101.xlsx"
POJ_NAZVY = {
"111": "VZP",
"201": "ČPZP",
"205": "ČPZP (ex-OZP)",
"207": "OZP",
"209": "ZPŠ",
"211": "ZPMV",
"213": "RBP",
}
# ── Barvy ──────────────────────────────────────────────────────────────────────
BLUE_HEADER = "1F497D"
WHITE = "FFFFFF"
LIGHT_BLUE = "DCE6F1"
LIGHT_GREEN = "EBF1DE"
LIGHT_YELLOW = "FFFFC0"
LIGHT_RED = "FCE4D6"
LIGHT_GREY = "F2F2F2"
ORANGE = "F4B942"
# ── Data z MySQL ───────────────────────────────────────────────────────────────
mysql = connect_mysql()
cur = mysql.cursor()
cur.execute("""
SELECT rc FROM vzp_registrace_raw WHERE k_datu = %s
AND rc NOT IN (
SELECT rc FROM vzp_registrace_lekari
WHERE k_datu = %s AND kod_odbornosti = '001'
AND ICP = '09305001' AND ma_lekare = 1
)
""", (K_DATU_HIST, K_DATU_HIST))
problematicke_rcs = [row[0] for row in cur.fetchall()]
ph = ",".join(["%s"] * len(problematicke_rcs))
cur.execute(f"""
SELECT rc, prijmeni, jmeno, kod_odbornosti, ma_lekare, ICP,
nazev_lekare, nazev_zzz, poj_kod, poj_zkratka
FROM vzp_registrace_lekari
WHERE k_datu = %s AND rc IN ({ph}) AND kod_odbornosti = '001'
""", (K_DATU_HIST, *problematicke_rcs))
vzp = {}
for rc, prijmeni, jmeno, odb, ma, icp, nazev_lek, nazev_zzz, poj_kod, poj_zkr in cur.fetchall():
vzp[rc] = {"prijmeni": prijmeni or "", "jmeno": jmeno or "",
"ma_lekare": bool(ma), "ICP": icp or "",
"nazev_lekare": nazev_lek or "", "nazev_zzz": nazev_zzz or "",
"poj_kod": poj_kod or "", "poj_zkratka": poj_zkr or ""}
mysql.close()
# ── Data z Medicusu ────────────────────────────────────────────────────────────
computer_name = socket.gethostname().upper()
dsn_map = {
"LEKAR": r"localhost:M:\medicus\data\medicus.fdb",
"SESTRA": r"192.168.1.10:m:\medicus\data\medicus.fdb",
"LENOVO": r"192.168.1.10:m:\medicus\data\medicus.fdb",
}
dsn = dsn_map.get(computer_name, r"localhost:c:\medicus 3\data\medicus.fdb")
fb_conn = fdb.connect(dsn=dsn, user="SYSDBA", password="masterkey", charset="win1250")
fb_cur = fb_conn.cursor()
# Aktuálně aktivní pacienti
fb_cur.execute("""
SELECT kar.rodcis FROM kar
WHERE kar.vyrazen = 'N' AND kar.rodcis IS NOT NULL AND kar.rodcis <> ''
AND EXISTS (
SELECT 1 FROM registr r JOIN icp i ON r.idicp = i.idicp
WHERE r.idpac = kar.idpac
AND r.datum <= CURRENT_DATE
AND (r.datum_zruseni IS NULL OR r.datum_zruseni >= CURRENT_DATE)
AND r.priznak IN ('V','D','A')
AND i.icp = '09305001' AND i.odb = '001'
)
""")
aktualne_aktivni = {(row[0] or "").strip() for row in fb_cur.fetchall()}
# Detail všech 151 pacientů
ph_fb = ",".join(["?" for _ in problematicke_rcs])
fb_cur.execute(f"""
SELECT kar.rodcis, kar.prijmeni, kar.jmeno, kar.poj, kar.vyrazen,
r.datum, r.datum_zruseni, r.priznak
FROM kar
LEFT JOIN registr r ON r.idpac = kar.idpac
LEFT JOIN icp i ON r.idicp = i.idicp AND i.icp = '09305001' AND i.odb = '001'
WHERE kar.rodcis IN ({ph_fb})
ORDER BY kar.rodcis, r.datum DESC
""", problematicke_rcs)
medicus = {}
for row in fb_cur.fetchall():
rc = (row[0] or "").strip()
if rc not in medicus:
medicus[rc] = {
"prijmeni": (row[1] or "").strip(),
"jmeno": (row[2] or "").strip(),
"poj": str(row[3] or "").strip(),
"vyrazen": (row[4] or "").strip(),
"reg_datum": row[5],
"reg_datum_zruseni":row[6],
"reg_priznak": (row[7] or "").strip(),
}
fb_conn.close()
# ── Kategorizace ───────────────────────────────────────────────────────────────
def kategorie(rc, vzp_row, med_row, aktivni_set):
poj = med_row.get("poj", "") if med_row else ""
if not vzp_row:
if poj != "111":
return "JINÁ POJIŠŤOVNA", "VZP neregistruje — pojištěnec jiné pojišťovny.", LIGHT_BLUE
return "BEZ VZP ZÁZNAMU", "VZP nevrátila žádný záznam přesto, že jde o pojištěnce VZP. Pravděpodobně chybí registrace u VZP.", LIGHT_RED
if vzp_row["ma_lekare"]:
return "REGISTROVÁN JINDE", f"VZP eviduje registraci u jiného lékaře: {vzp_row['nazev_zzz']} (ICP {vzp_row['ICP']}).", LIGHT_RED
return "BEZ LÉKAŘE U VZP", "VZP eviduje pojištěnce, ale bez registrujícího lékaře v odbornosti 001.", LIGHT_YELLOW
# ── Sestavení řádků ────────────────────────────────────────────────────────────
rows = []
for rc in problematicke_rcs:
vzp_row = vzp.get(rc)
med_row = medicus.get(rc)
aktivni = rc in aktualne_aktivni
prijmeni = (med_row or vzp_row or {}).get("prijmeni", "")
jmeno = (med_row or vzp_row or {}).get("jmeno", "")
poj_kod = med_row.get("poj", "") if med_row else (vzp_row or {}).get("poj_kod", "")
poj_nazev = POJ_NAZVY.get(poj_kod, poj_kod)
med_stav = "Aktivní" if aktivni else ("Odregistrován" if med_row else "Nenalezen v Medicusu")
reg_datum = med_row.get("reg_datum") if med_row else None
reg_datum_zrus = med_row.get("reg_datum_zruseni") if med_row else None
kat, komentar, barva = kategorie(rc, vzp_row, med_row, aktualne_aktivni)
vzp_icp = vzp_row["ICP"] if vzp_row and vzp_row["ma_lekare"] else ""
vzp_lek = vzp_row["nazev_zzz"] if vzp_row and vzp_row["ma_lekare"] else ""
vzp_zzz = vzp_row["nazev_lekare"] if vzp_row and vzp_row["ma_lekare"] else ""
rows.append({
"prijmeni": prijmeni,
"jmeno": jmeno,
"rc": rc,
"poj_kod": poj_kod,
"poj_nazev": poj_nazev,
"med_stav": med_stav,
"reg_datum": reg_datum.strftime("%d.%m.%Y") if reg_datum else "",
"reg_zruseni": reg_datum_zrus.strftime("%d.%m.%Y") if reg_datum_zrus else "",
"kategorie": kat,
"komentar": komentar,
"vzp_icp": vzp_icp,
"vzp_lek": vzp_lek,
"vzp_zzz": vzp_zzz,
"barva": barva,
})
rows.sort(key=lambda r: (r["kategorie"], r["prijmeni"], r["jmeno"]))
# ── Excel ──────────────────────────────────────────────────────────────────────
wb = Workbook()
# ────────────────────────────────────────────────────────────────────────────────
# SHEET 1: Přehled
# ────────────────────────────────────────────────────────────────────────────────
ws_info = wb.active
ws_info.title = "Přehled"
def hdr_cell(ws, row, col, value):
c = ws.cell(row=row, column=col, value=value)
c.font = Font(name="Arial", bold=True, color=WHITE, size=11)
c.fill = PatternFill("solid", fgColor=BLUE_HEADER)
c.alignment = Alignment(horizontal="center", vertical="center")
return c
def val_cell(ws, row, col, value, bold=False, bg=None):
c = ws.cell(row=row, column=col, value=value)
c.font = Font(name="Arial", bold=bold, size=10)
c.alignment = Alignment(wrap_text=True, vertical="top")
if bg:
c.fill = PatternFill("solid", fgColor=bg)
return c
ws_info.column_dimensions["A"].width = 36
ws_info.column_dimensions["B"].width = 18
ws_info.column_dimensions["C"].width = 60
# Titulek
ws_info.merge_cells("A1:C1")
t = ws_info["A1"]
t.value = f"Pacienti registrovaní v Medicusu k 1. 1. 2025, ale dle VZP bez registrace u IČP 09305001"
t.font = Font(name="Arial", bold=True, size=13, color=WHITE)
t.fill = PatternFill("solid", fgColor=BLUE_HEADER)
t.alignment = Alignment(horizontal="center", vertical="center", wrap_text=True)
ws_info.row_dimensions[1].height = 36
ws_info.merge_cells("A2:C2")
ws_info["A2"].value = f"Vygenerováno: {TODAY.strftime('%d. %m. %Y')} | Stav v Medicusu k dnešnímu dni"
ws_info["A2"].font = Font(name="Arial", italic=True, size=9, color="595959")
ws_info["A2"].alignment = Alignment(horizontal="center")
# Souhrn počtů
counts = defaultdict(int)
counts_aktivni = defaultdict(int)
for r in rows:
counts[r["kategorie"]] += 1
if r["med_stav"] == "Aktivní":
counts_aktivni[r["kategorie"]] += 1
hdr_cell(ws_info, 4, 1, "Kategorie")
hdr_cell(ws_info, 4, 2, "Počet pacientů")
hdr_cell(ws_info, 4, 3, "Komentář")
KAT_BARVY = {
"REGISTROVÁN JINDE": LIGHT_RED,
"BEZ LÉKAŘE U VZP": LIGHT_YELLOW,
"JINÁ POJIŠŤOVNA": LIGHT_BLUE,
"BEZ VZP ZÁZNAMU": LIGHT_RED,
}
KAT_POPIS = {
"REGISTROVÁN JINDE": "VZP k 1.1.2025 eviduje registraci u jiného praktického lékaře. Pacient se pravděpodobně přeregistroval jinam, aniž by byl v Medicusu odregistrován.",
"BEZ LÉKAŘE U VZP": "VZP eviduje pojištěnce, ale v odbornosti 001 mu neeviduje žádného lékaře. Může jít o opožděné zpracování přihlášky nebo technickou chybu.",
"JINÁ POJIŠŤOVNA": "Pacient není pojištěncem VZP — VZP o něm data nemá, proto nebylo vráceno nic. To je očekávané chování.",
"BEZ VZP ZÁZNAMU": "VZP pojištěnec (111), ale VZP nevrátila žádný záznam. Může jít o nesprávné RC, neaktivní pojistný vztah nebo chybu v komunikaci.",
}
for i, kat in enumerate(["REGISTROVÁN JINDE", "BEZ LÉKAŘE U VZP", "JINÁ POJIŠŤOVNA", "BEZ VZP ZÁZNAMU"]):
r = 5 + i
bg = KAT_BARVY[kat]
val_cell(ws_info, r, 1, kat, bold=True, bg=bg)
val_cell(ws_info, r, 2, f"{counts[kat]} ({counts_aktivni[kat]} stále aktivní)", bg=bg)
val_cell(ws_info, r, 3, KAT_POPIS[kat], bg=bg)
ws_info.row_dimensions[r].height = 42
ws_info.row_dimensions[4].height = 20
# Celkem
val_cell(ws_info, 10, 1, "CELKEM", bold=True)
val_cell(ws_info, 10, 2, f"{len(rows)} ({sum(counts_aktivni.values())} aktivní)", bold=True)
# Metodika
ws_info.merge_cells("A12:C12")
ws_info["A12"].value = "Metodika"
ws_info["A12"].font = Font(name="Arial", bold=True, size=11, color=BLUE_HEADER)
metodika_text = (
"Skript kdojelekar_20250101.py dotázal VZP B2B na registrujícího lékaře (odbornost 001) "
"pro každého pacienta registrovaného k 1. 1. 2025 v Medicusu u IČP 09305001. "
"Pacienti v tomto souboru jsou ti, u nichž VZP k danému datu nevrátila záznam s ICP=09305001 a ma_lekare=1. "
"Stav v Medicusu je aktuální k dnešnímu dni (" + TODAY.strftime("%d. %m. %Y") + ")."
)
ws_info.merge_cells("A13:C13")
c = ws_info["A13"]
c.value = metodika_text
c.font = Font(name="Arial", size=9, color="595959")
c.alignment = Alignment(wrap_text=True, vertical="top")
ws_info.row_dimensions[13].height = 56
# Doporučení
ws_info.merge_cells("A15:C15")
ws_info["A15"].value = "Doporučení"
ws_info["A15"].font = Font(name="Arial", bold=True, size=11, color=BLUE_HEADER)
ws_info.merge_cells("A16:C16")
c = ws_info["A16"]
c.value = (
"1. REGISTROVÁN JINDE — ověřit s pacientem při návštěvě, zda se přeregistroval; pokud ano, odregistrovat v Medicusu.\n"
"2. BEZ LÉKAŘE U VZP — zkontrolovat, zda přihláška registrace byla správně odeslána a VZP ji eviduje.\n"
"3. JINÁ POJIŠŤOVNA — tyto pacienty prověřit u příslušné pojišťovny (ČPZP, OZP…) analogickým dotazem.\n"
"4. BEZ VZP ZÁZNAMU — ověřit správnost RC a aktivitu pojistného vztahu přímo u VZP."
)
c.font = Font(name="Arial", size=10)
c.alignment = Alignment(wrap_text=True, vertical="top")
ws_info.row_dimensions[16].height = 80
# ────────────────────────────────────────────────────────────────────────────────
# SHEET 2: Data
# ────────────────────────────────────────────────────────────────────────────────
ws = wb.create_sheet("Pacienti")
COLS = [
("Příjmení", 20),
("Jméno", 14),
("Rodné číslo", 14),
("Pojišťovna", 12),
("Stav v Medicusu\ndnes", 16),
("Datum registrace\nv Medicusu", 16),
("Datum zrušení\nv Medicusu", 16),
("Kategorie VZP problému", 22),
("Komentář", 52),
("VZP ICP jiného lékaře", 18),
("VZP — jméno lékaře", 28),
("VZP — název ZZZ", 36),
]
for col_idx, (header, width) in enumerate(COLS, 1):
c = ws.cell(row=1, column=col_idx, value=header)
c.font = Font(name="Arial", bold=True, color=WHITE, size=10)
c.fill = PatternFill("solid", fgColor=BLUE_HEADER)
c.alignment = Alignment(horizontal="center", vertical="center", wrap_text=True)
ws.column_dimensions[get_column_letter(col_idx)].width = width
ws.row_dimensions[1].height = 32
ws.freeze_panes = "A2"
thin = Side(style="thin", color="BFBFBF")
border = Border(left=thin, right=thin, top=thin, bottom=thin)
for row_idx, r in enumerate(rows, 2):
bg = r["barva"]
data = [
r["prijmeni"], r["jmeno"], r["rc"], f"{r['poj_kod']} {r['poj_nazev']}",
r["med_stav"], r["reg_datum"], r["reg_zruseni"],
r["kategorie"], r["komentar"],
r["vzp_icp"], r["vzp_lek"], r["vzp_zzz"],
]
for col_idx, value in enumerate(data, 1):
c = ws.cell(row=row_idx, column=col_idx, value=value)
c.font = Font(name="Arial", size=9)
c.fill = PatternFill("solid", fgColor=bg)
c.border = border
c.alignment = Alignment(vertical="top", wrap_text=(col_idx in (9, 11, 12)))
if r["med_stav"] == "Aktivní" and col_idx == 5:
c.font = Font(name="Arial", size=9, bold=True, color="375623")
elif r["med_stav"] != "Aktivní" and col_idx == 5:
c.font = Font(name="Arial", size=9, color="843C0C")
ws.row_dimensions[row_idx].height = 32
# AutoFilter
ws.auto_filter.ref = f"A1:{get_column_letter(len(COLS))}1"
wb.save(OUT_FILE)
print(f"Uloženo: {OUT_FILE}")
print(f"Celkem řádků: {len(rows)}")
+158
View File
@@ -0,0 +1,158 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
kdojelekar_20250101.py
======================
Jednorázový skript: vybere pacienty registrované k 01.01.2025
a dotáže se VZP B2B na jejich registrujícího lékaře k tomuto datu.
Výsledky uloží do stejných MySQL tabulek jako týdenní skript
(vzp_registrace_lekari, vzp_registrace_raw) s k_datu = 2025-01-01.
Resumovatelný — přeskočí pacienty, kteří již mají raw XML pro k_datu=2025-01-01.
"""
import sys
import time
from pathlib import Path
from datetime import date
PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
sys.path.insert(0, str(PROJECT_ROOT))
from Knihovny.vzpb2b_client import VZPB2BClient
from Knihovny.mysql_db import connect_mysql
from Knihovny.medicus_db import get_medicus_connection
# ── KONFIGURACE ───────────────────────────────────────────────────────────────
K_DATU = date(2025, 1, 1)
API_PAUSE = 2
PFX_PATH = Path(__file__).resolve().parent.parent / "Certificates" / "picka.pfx"
PFX_PASS = "Vlado7309208104+"
ODBORNOSTI = None # None = bez filtru odborností
# ── INIT ──────────────────────────────────────────────────────────────────────
vzp = VZPB2BClient("prod", str(PFX_PATH), PFX_PASS)
mysql = connect_mysql()
# ── PACIENTI Z MEDICUS (registrovaní k 01.01.2025) ───────────────────────────
import fdb, socket
computer_name = socket.gethostname().upper()
dsn_map = {
"LEKAR": r"localhost:M:\medicus\data\medicus.fdb",
"SESTRA": r"192.168.1.10:m:\medicus\data\medicus.fdb",
"LENOVO": r"192.168.1.10:m:\medicus\data\medicus.fdb",
}
dsn = dsn_map.get(computer_name, r"localhost:c:\medicus 3\data\medicus.fdb")
fb_conn = fdb.connect(dsn=dsn, user="SYSDBA", password="masterkey", charset="win1250")
fb_cur = fb_conn.cursor()
fb_cur.execute("""
SELECT kar.rodcis, kar.prijmeni, kar.jmeno, kar.poj
FROM kar
WHERE kar.vyrazen = 'N'
AND kar.rodcis IS NOT NULL
AND kar.rodcis <> ''
AND EXISTS (
SELECT r.id FROM registr r
JOIN icp i ON r.idicp = i.idicp
WHERE r.idpac = kar.idpac
AND r.datum <= ?
AND (r.datum_zruseni IS NULL OR r.datum_zruseni >= ?)
AND r.priznak IN ('V', 'D', 'A')
AND i.icp = '09305001'
AND i.odb = '001'
)
ORDER BY kar.prijmeni, kar.rodcis
""", (K_DATU.isoformat(), K_DATU.isoformat()))
cols = [d[0].strip().lower() for d in fb_cur.description]
pacienti = [dict(zip(cols, row)) for row in fb_cur.fetchall()]
fb_conn.close()
print(f"Pacientu registrovanych k {K_DATU}: {len(pacienti)}")
# ── RESUME: přeskočit již hotové ─────────────────────────────────────────────
with mysql.cursor() as cur:
cur.execute("SELECT rc FROM vzp_registrace_raw WHERE k_datu = %s", (K_DATU,))
hotove = {row[0] for row in cur.fetchall()}
pacienti = [p for p in pacienti if (p.get("rodcis") or "").strip() not in hotove]
print(f"Zbyvá zpracovat: {len(pacienti)} ({len(hotove)} již hotovo)\n")
# ── BATCH ─────────────────────────────────────────────────────────────────────
call_count = 0
for i, pac in enumerate(pacienti):
rc = (pac.get("rodcis") or "").strip()
prijmeni = (pac.get("prijmeni") or "").strip()
jmeno = (pac.get("jmeno") or "").strip()
if not rc:
continue
if call_count > 0:
time.sleep(API_PAUSE)
call_count += 1
print(f"[{i+1}/{len(pacienti)}] {prijmeni} {jmeno} ({rc}) ...", end=" ", flush=True)
try:
xml = vzp.registrace_lekare(rc=rc, k_datu=K_DATU.isoformat(), odbornosti=ODBORNOSTI)
zaznamy = vzp.parse_registrace_lekare(xml)
except Exception as e:
print(f"CHYBA: {e}")
continue
print(f"{len(zaznamy)} lekar(u)")
for z in zaznamy:
print(f" {z['kod_odbornosti']}: {z['nazev_lekare']} / {z['nazev_zzz']}"
f" [{z['datum_zahajeni']} - {z['datum_ukonceni']}]")
if not zaznamy:
print(" (zadny lekar v zadne odbornosti)")
with mysql.cursor() as cur:
cur.execute("""
INSERT INTO vzp_registrace_raw (rc, k_datu, raw_xml)
VALUES (%s, %s, %s)
ON DUPLICATE KEY UPDATE raw_xml=VALUES(raw_xml)
""", (rc, K_DATU, xml))
with mysql.cursor() as cur:
for z in zaznamy:
cur.execute("""
INSERT INTO vzp_registrace_lekari
(rc, prijmeni, jmeno, k_datu, kod_odbornosti, ma_lekare,
ICZ, ICP, nazev_lekare, nazev_zzz,
poj_kod, poj_zkratka,
datum_registrace, datum_zahajeni, datum_ukonceni,
stav_vyrizeni)
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
ON DUPLICATE KEY UPDATE
prijmeni=VALUES(prijmeni), jmeno=VALUES(jmeno),
ma_lekare=VALUES(ma_lekare),
ICZ=VALUES(ICZ), ICP=VALUES(ICP),
nazev_lekare=VALUES(nazev_lekare), nazev_zzz=VALUES(nazev_zzz),
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)
""", (
rc, prijmeni, jmeno, K_DATU, z["kod_odbornosti"], 1 if z["ma_lekare"] else 0,
z["ICZ"], z["ICP"], z["nazev_lekare"], z["nazev_zzz"],
z["poj_kod"], z["poj_zkratka"],
z["datum_registrace"], z["datum_zahajeni"], z["datum_ukonceni"],
z["stav_vyrizeni"],
))
print(f"\nHotovo. VZP dotazu: {call_count}")
mysql.close()
+142
View File
@@ -0,0 +1,142 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
kdojelekar_tydenni.py
======================
Projde aktivně registrované pacienty z Medicus a pro každého zjistí
registrujícího lékaře u VZP (odbornosti 001, 002, 014).
Výsledky uloží do MySQL tabulky vzp_registrace_lekari.
Spouštět týdně. Mezi dotazy 2s prodleva (API_DELAY).
TEST_MODE = True → zpracuje jen 3 náhodné pacienty, bez zápisu do DB.
TEST_MODE = False → produkční běh, zapíše vše.
"""
import sys
import time
import random
from pathlib import Path
from datetime import date
PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
sys.path.insert(0, str(PROJECT_ROOT))
from Knihovny.vzpb2b_client import VZPB2BClient
from Knihovny.mysql_db import connect_mysql
from Knihovny.medicus_db import get_medicus_db
# ── KONFIGURACE ───────────────────────────────────────────────────────────────
API_PAUSE = 2 # sekundy mezi VZP dotazy
TEST_MODE = False # False = produkční běh
PFX_PATH = Path(__file__).resolve().parent.parent / "Certificates" / "picka.pfx"
PFX_PASS = "Vlado7309208104+"
ODBORNOSTI = None # None = bez filtru, VZP vrátí všechny odbornosti
TODAY = date.today()
# ── INIT ──────────────────────────────────────────────────────────────────────
vzp = VZPB2BClient("prod", str(PFX_PATH), PFX_PASS)
mysql = connect_mysql()
# ── PACIENTI Z MEDICUS ────────────────────────────────────────────────────────
medicus = get_medicus_db()
pacienti = medicus.get_active_registered_patients(as_dict=True)
medicus.close()
print(f"Aktivně registrovaných pacientů: {len(pacienti)}")
if TEST_MODE:
pacienti = random.sample(pacienti, min(3, len(pacienti)))
print(f"TEST MODE — zpracuji {len(pacienti)} náhodné pacienty, BEZ zápisu do DB\n")
else:
# Načti RC která už dnes mají uložený raw XML — ty přeskočíme
with mysql.cursor() as cur:
cur.execute("SELECT rc FROM vzp_registrace_raw WHERE k_datu = %s", (TODAY,))
hotove = {row[0] for row in cur.fetchall()}
pacienti = [p for p in pacienti if (p.get("rodcis") or "").strip() not in hotove]
print(f"PRODUKČNÍ běh — k_datu={TODAY}, zbývá zpracovat: {len(pacienti)} pacientů"
f" ({len(hotove)} již hotovo dnes)\n")
# ── BATCH ─────────────────────────────────────────────────────────────────────
call_count = 0
for i, pac in enumerate(pacienti):
rc = (pac.get("rodcis") or "").strip()
prijmeni = (pac.get("prijmeni") or "").strip()
jmeno = (pac.get("jmeno") or "").strip()
if not rc:
continue
if call_count > 0:
time.sleep(API_PAUSE)
call_count += 1
print(f"[{i+1}/{len(pacienti)}] {prijmeni} {jmeno} ({rc}) ...", end=" ", flush=True)
try:
xml = vzp.registrace_lekare(rc=rc, k_datu=TODAY.isoformat(), odbornosti=ODBORNOSTI)
zaznamy = vzp.parse_registrace_lekare(xml)
except Exception as e:
print(f"CHYBA: {e}")
continue
print(f"{len(zaznamy)} lékař(ů)")
for z in zaznamy:
print(f" {z['kod_odbornosti']}: {z['nazev_lekare']} / {z['nazev_zzz']}"
f" [{z['datum_zahajeni']} {z['datum_ukonceni']}]")
if not zaznamy:
print(f" (žádný lékař v žádné odbornosti)")
if TEST_MODE:
continue
with mysql.cursor() as cur:
cur.execute("""
INSERT INTO vzp_registrace_raw (rc, k_datu, raw_xml)
VALUES (%s, %s, %s)
ON DUPLICATE KEY UPDATE raw_xml=VALUES(raw_xml)
""", (rc, TODAY, xml))
with mysql.cursor() as cur:
for z in zaznamy:
cur.execute("""
INSERT INTO vzp_registrace_lekari
(rc, prijmeni, jmeno, k_datu, kod_odbornosti, ma_lekare,
ICZ, ICP, nazev_lekare, nazev_zzz,
poj_kod, poj_zkratka,
datum_registrace, datum_zahajeni, datum_ukonceni,
stav_vyrizeni)
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
ON DUPLICATE KEY UPDATE
prijmeni=VALUES(prijmeni), jmeno=VALUES(jmeno),
ma_lekare=VALUES(ma_lekare),
ICZ=VALUES(ICZ), ICP=VALUES(ICP),
nazev_lekare=VALUES(nazev_lekare), nazev_zzz=VALUES(nazev_zzz),
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)
""", (
rc, prijmeni, jmeno, TODAY, z["kod_odbornosti"], 1 if z["ma_lekare"] else 0,
z["ICZ"], z["ICP"], z["nazev_lekare"], z["nazev_zzz"],
z["poj_kod"], z["poj_zkratka"],
z["datum_registrace"], z["datum_zahajeni"], z["datum_ukonceni"],
z["stav_vyrizeni"],
))
print(f"\nHotovo. VZP dotazů: {call_count}")
if TEST_MODE:
print("(TEST MODE — nic nebylo zapsáno do DB)")
mysql.close()