notebookvb

This commit is contained in:
Vladimir Buzalka
2026-06-17 05:22:34 +02:00
parent 39d33b76f3
commit 45c32a37c4
4 changed files with 430 additions and 21 deletions
@@ -0,0 +1,96 @@
# OZP (207) — Stahování seznamu registrovaných pojištěnců
## Co skript dělá
`StahniSeznamPojistencuOZP.py` provede v jednom spuštění čtyři kroky:
1. **Přihlásí se** certifikátem na portál OZP (čistý Python, bez NMSigneru)
— uloží cookies do sdíleného `StahováníZpráv/207 OZP/ozp_cookies.json`
2. **Stáhne nové výpisy** z výpisové schránky `schranky-vypis-pojistencu-v-kapitaci`
— stahuje soubory, jejichž obsah začíná `H09305001`
— ukládá do `…\Zúčtovací zprávy\SeznamyPojištěnců\` (Dropbox)
— zastaví se při první již stažené zprávě
— po stahování se **znovu přihlásí** (Playwright invaliduje requests session)
3. **Podá žádost** o aktuální výpis (typ=soubor, třídění dle příjmení)
## Platforma
OZP běží na stejné platformě jako **ZPŠ, VoZP, RBP** (portalzp.cz / json-api).
Login je identický se ZPŠ. Liší se URL schránky, ID formuláře a názvy filtru/položek.
## Flow přihlášení (stejné jako ZPŠ)
1. GET `/app/prihlaseni` → session cookie
2. POST `/json-api/prihlaseni/prihlasovaci-zprava` → challenge (`zprava`)
3. Podpis challenge certifikátem (PKCS7/SHA-256, **s** certifikátem)
4. POST `/json-api/prihlaseni/prihlaseni-certifikatem` → autentizovaná session
## Stažení přílohy
GET `/html/prehled-zprav-ve-schrankach/zobrazit-prilohu?zprava_id={fileId}`
`fileId` se získá z `onclick="SchrPolOpenFile(<id>)"` v řádcích tabulky schránky.
Soubory ve schránce mají název `F207MMRR.xxx` (MM/RR = měsíc/rok generování).
## Podání žádosti (KLÍČOVÝ ROZDÍL oproti ZPŠ)
OZP **nemá pole „datum/měsíc"** — výpis je *aktuální snímek* platných registrací
(„připraveno do příštího dne"). Nepodává se za konkrétní měsíc, nepočítá se „další měsíc".
Při každém běhu se podá jedna žádost o aktuální výpis. Žádný stavový soubor s měsícem.
POST `https://portal.ozp.cz/json-api/formular-schranky/108-vypis-pojistencu-v-registraci/ulozit-formular`
Body: `{"schrXml": "...", "schrSign": "-----BEGIN PKCS7-----...", "schrFiles": []}`
### XML žádosti (řádky `\r\n`)
```xml
<SchrankaZadost NazevSchranky="SEZNAM_KAP" NazevFiltru="SEZNAM_KAP">
<PolozkaFiltru Nazev="nicoz">13074913</PolozkaFiltru>
<PolozkaFiltru Nazev="trideni">p</PolozkaFiltru>
<PolozkaFiltru Nazev="typ">soubor</PolozkaFiltru>
</SchrankaZadost>
```
| Položka | Hodnota | Význam |
|---------|---------|--------|
| `nicoz` | `13074913` | **interní ID** položky IČZ (zobrazené IČZ = 09305000). Ověřeno: posílá se interní ID, ne číslo IČZ. |
| `trideni` | `p` | `p`=podle příjmení, `i`=IČP+příjmení, `r`=rodná čísla |
| `typ` | `soubor` | `soubor`=datový soubor dle rozhraní, `sestava`=tiskový výstup |
### Podpis XML
PKCS7/SHA-256, **bez** certifikátu v podpisu (`NoCerts`) — stejně jako ZPŠ formulář.
Server certifikát v podpisu odmítá.
## Jak byly endpointy zjištěny
Odposlechem reálného podání v Chrome (MCP) — `data-xml-*` atributy formuláře daly názvy
schránky/filtru a položek, odchycený XHR na `ulozit-formular` potvrdil přesný payload.
První ostré podání: **ref. 179774883** (17.06.2026).
## Srovnání se ZPŠ
| | ZPŠ (209) | OZP (207) |
|--|-----------|-----------|
| Schránka URL | `schranka-vypis-…` (jedn.) | `schranky-vypis-…` (množ.) |
| Formulář | `29-vypis-registrov-pojistencu` | `108-vypis-pojistencu-v-registraci` |
| NazevSchranky / NazevFiltru | `VypisPojKap` / `ZZ_VYP_REG` | `SEZNAM_KAP` / `SEZNAM_KAP` |
| Položka IČZ | `icz` = 25520 | `nicoz` = 13074913 (interní ID) |
| Pole datum | ano (za měsíc) | **ne** (aktuální snímek) |
| Stav | `stav.json` (měsíc) | jen `log_podani.json` |
## Soubory
| Soubor | Popis |
|--------|-------|
| `StahniSeznamPojistencuOZP.py` | Hlavní skript — stažení výpisů + podání žádosti |
| `log_podani.json` | Historie podání s referenčními čísly |
## Parametry
- **IČZ**: 09305000 (IČP: 09305001, MUDr. Michaela Buzalková), interní ID `13074913`
- **Certifikát**: `Insurance/Certificates/MBQualifiedCert.pfx`
## Stav
Hotovo a otestováno (17.06.2026): login ✓, stažení ✓ (3 výpisy), podání ✓ (ref. 179774883).
Výpis z prvního podání dorazí do schránky do příštího dne.
@@ -1,35 +1,80 @@
"""
Stahování seznamu registrovaných pojištěnců OZP (207) — čistý Python.
Stahování seznamu registrovaných pojištěnců OZP (207) — čistý Python, bez NMSigneru.
OZP běží na stejné platformě jako ZPŠ (portalzp.cz / json-api).
Tento skript zatím umí jen KROK 1: přihlášení certifikátem.
OZP běží na stejné platformě jako ZPŠ (portalzp.cz / json-api), ale s rozdíly:
- schránka: /app/schranky-vypis-pojistencu-v-kapitaci (množné "schranky")
- formulář: 108-vypis-pojistencu-v-registraci
- filtr XML: NazevSchranky = NazevFiltru = "SEZNAM_KAP"
- položky: nicoz (IČZ interní ID), trideni (p/i/r), typ (soubor/sestava)
- BEZ pole "datum" — výpis je aktuální snímek platných registrací
("připraveno do příštího dne"), nepodává se za konkrétní měsíc.
Stavba poběží postupně:
[krok 1] přihlášení certifikátem ← hotovo
[krok 2] stažení nových výpisů ze schránky
[krok 3] podání žádosti o další měsíc
Co skript dělá v jednom spuštění:
1. Přihlásí se certifikátem (uloží cookies pro Playwright)
2. Stáhne nové výpisy z výpisové schránky (soubory s hlavičkou H09305001)
3. Znovu se přihlásí (Playwright invaliduje requests session)
4. Podá jednu žádost o aktuální výpis (typ=soubor, třídění dle příjmení)
Log podání: log_podani.json — seznam { ref_cislo, podano_kdy }
"""
import io
import json
import os
import re
import sys
import time
from datetime import datetime
from pathlib import Path
import requests
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.serialization import pkcs7, pkcs12
# UTF-8 výstup i na Windows konzoli (cp1252 by padal na českých znacích)
try:
sys.stdout.reconfigure(encoding="utf-8")
sys.stderr.reconfigure(encoding="utf-8")
except Exception:
pass
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "..")))
from Knihovny.najdi_dropbox import get_dropbox_root
PFX_PATH = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "Certificates", "MBQualifiedCert.pfx"))
PFX_PASSWORD = b"Vlado7309208104++"
BASE_URL = "https://portal.ozp.cz"
CHALLENGE_URL = f"{BASE_URL}/json-api/prihlaseni/prihlasovaci-zprava"
CERTLOGIN_URL = f"{BASE_URL}/json-api/prihlaseni/prihlaseni-certifikatem"
SUBMIT_URL = f"{BASE_URL}/json-api/formular-schranky/108-vypis-pojistencu-v-registraci/ulozit-formular"
# Sdílený cookies soubor s OZP skriptem pro stahování zpráv
COOKIES_FILE = os.path.abspath(os.path.join(
os.path.dirname(__file__), "..", "..", "StahováníZpráv", "207 OZP", "ozp_cookies.json"
))
VYPIS_URL = f"{BASE_URL}/app/schranky-vypis-pojistencu-v-kapitaci"
DOWNLOAD_URL = f"{BASE_URL}/html/prehled-zprav-ve-schrankach/zobrazit-prilohu"
# Hodnoty filtru (ověřeno odchytem reálného podání na portálu)
ICZ_INTERNAL = "13074913" # IČZ 09305000 — interní ID položky "nicoz"
TRIDENI = "p" # p = podle příjmení, i = IČP+příjmení, r = rodná čísla
TYP = "soubor" # soubor = datový soubor, sestava = tiskový výstup
# Hlavička platného výpisu pojištěnců (IČP 09305001 = MUDr. Buzalková)
HLAVICKA = "H09305001"
LOG_FILE = os.path.join(os.path.dirname(__file__), "log_podani.json")
# Sdílené soubory s OZP skriptem pro stahování zpráv
STAHUJ_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "StahováníZpráv", "207 OZP"))
COOKIES_FILE = os.path.join(STAHUJ_DIR, "ozp_cookies.json")
CHROME_PROFILE = os.path.join(STAHUJ_DIR, "chrome_profile")
DOWNLOAD_DIR = os.path.join(
get_dropbox_root(),
"Ordinace", "Dokumentace_ke_zpracování", "Zúčtovací zprávy", "SeznamyPojištěnců",
)
# ---------------------------------------------------------------------------
# Přihlášení
# ---------------------------------------------------------------------------
def prihlaseni() -> requests.Session:
"""Přihlásí se certifikátem, vrátí autentizovanou session. Uloží cookies pro Playwright."""
@@ -41,19 +86,15 @@ def prihlaseni() -> requests.Session:
"Referer": BASE_URL + "/",
})
# 1. Načti login stránku → session cookie
r = session.get(f"{BASE_URL}/app/prihlaseni")
r.raise_for_status()
session.cookies.set("pzp_sign", "CERT", domain="portal.ozp.cz", path="/")
# 2. Získej challenge
r = session.post(CHALLENGE_URL, json={"login_sign": "CERT"},
headers={"Content-Type": "application/json; charset=UTF-8"})
r.raise_for_status()
zprava = r.json()["data"]["zprava"]
print(f"Challenge: {zprava[:60]}...")
# 3. Podepiš challenge certifikátem (PKCS7 detached, RSA + SHA-256, s certifikátem)
with open(PFX_PATH, "rb") as f:
private_key, cert, _ = pkcs12.load_key_and_certificates(f.read(), PFX_PASSWORD)
@@ -65,7 +106,6 @@ def prihlaseni() -> requests.Session:
.decode("ascii").strip()
)
# 4. Přihlas se
r = session.post(CERTLOGIN_URL, json={"zprava": zprava, "podpis": podpis},
headers={"Content-Type": "application/json; charset=UTF-8"})
r.raise_for_status()
@@ -74,9 +114,8 @@ def prihlaseni() -> requests.Session:
if not data.get("prihlasen"):
raise RuntimeError(f"Přihlášení selhalo: {r.json().get('errMsg', '')}")
print(f"Přihlášení úspěšné! {data.get('url', '')}")
print("Přihlášení úspěšné!")
# 5. Ulož cookies pro Playwright
cookies = [
{
"name": c.name,
@@ -92,14 +131,282 @@ def prihlaseni() -> requests.Session:
]
with open(COOKIES_FILE, "w", encoding="utf-8") as f:
json.dump(cookies, f, indent=2, ensure_ascii=False)
print(f"Cookies uloženy: {len(cookies)}{COOKIES_FILE}")
return session
# ---------------------------------------------------------------------------
# Stahování z výpisové schránky
# ---------------------------------------------------------------------------
def safe_filename(name: str) -> str:
return re.sub(r'[\\/:*?"<>|]', "_", name).strip()
def parse_date(date_str: str) -> str:
try:
return datetime.strptime(date_str.strip()[:19], "%d.%m.%Y %H:%M:%S").strftime("%Y-%m-%d")
except Exception:
try:
return datetime.strptime(date_str.strip()[:10], "%d.%m.%Y").strftime("%Y-%m-%d")
except Exception:
return "0000-00-00"
def parse_row(cells: list) -> dict:
"""Z buněk řádku schránky vytvoří popis a cílový název souboru."""
date_raw = cells[1].strip() if len(cells) > 1 else ""
desc_raw = cells[2].strip() if len(cells) > 2 else ""
fname_raw = cells[3].strip() if len(cells) > 3 else ""
desc_lines = [l.strip() for l in desc_raw.split("\n") if l.strip()]
if len(desc_lines) >= 3:
description = desc_lines[2]
elif len(desc_lines) >= 2:
description = desc_lines[1]
else:
description = desc_lines[0] if desc_lines else ""
description = description[:80]
fname_match = re.match(r'^(.+?)\s*\(\d{2}\.\d{2}\.\d{4}\)\s*$', fname_raw)
original = fname_match.group(1).strip() if fname_match else fname_raw.split("(")[0].strip()
orig_path = Path(original)
stem = orig_path.stem or "zprava"
ext = orig_path.suffix or ""
date_iso = parse_date(date_raw)
name = f"{date_iso} {safe_filename(description)} ({safe_filename(stem)}){ext}"
if len(name) > 240:
name = f"{date_iso} ({safe_filename(stem)}){ext}"
return {"date": date_iso, "desc": description, "original": original, "filename": name}
def stahni_nove_vypisy() -> int:
"""Stáhne nové výpisy z výpisové schránky. Vrátí počet stažených souborů."""
try:
from playwright.sync_api import sync_playwright
except ImportError:
print("Chybí playwright: pip install playwright && playwright install chrome")
return 0
os.makedirs(DOWNLOAD_DIR, exist_ok=True)
with open(COOKIES_FILE, encoding="utf-8") as f:
cookies = json.load(f)
downloaded = 0
with sync_playwright() as p:
context = p.chromium.launch_persistent_context(
user_data_dir=CHROME_PROFILE,
channel="chrome",
headless=False,
slow_mo=100,
ignore_https_errors=True,
)
try:
context.add_cookies(cookies)
page = context.new_page()
page.goto(f"{VYPIS_URL}/", wait_until="domcontentloaded", timeout=30_000)
if "prihlaseni" in page.url or "login" in page.url.lower():
print("Session v prohlížeči expirovala — stahování přeskočeno")
return 0
print("Prohlížeč přihlášen OK\n")
already = set(os.listdir(DOWNLOAD_DIR))
print(f"V archivu: {len(already)} souborů.\n")
page_num = 1
seen_ids: set = set()
while True:
url = f"{VYPIS_URL}/stranka-{page_num}"
print(f" Stránka {page_num}: {url}")
try:
page.goto(url, wait_until="domcontentloaded", timeout=30_000)
except Exception as e:
print(f" Navigace selhala: {e}")
break
page.wait_for_load_state("networkidle", timeout=15_000)
data = page.evaluate("""() => {
const rows = [];
for (const tr of document.querySelectorAll('table tr')) {
const cells = Array.from(tr.querySelectorAll('td')).map(td => td.innerText.trim());
if (cells.length < 4) continue;
const dlLink = tr.querySelector('a[onclick*="SchrPolOpenFile"]');
if (!dlLink) continue;
const mFile = dlLink.getAttribute('onclick').match(/\\d+/);
rows.push({ cells, fileId: mFile ? mFile[0] : null });
}
return rows;
}""")
rows = [r for r in data if r["fileId"]]
if not rows:
print(f" Stránka {page_num} — žádné řádky, konec schránky.")
break
current_ids = {r["fileId"] for r in rows}
if current_ids & seen_ids:
print(f" Stránka {page_num} — opakující se obsah, konec schránky.")
break
seen_ids.update(current_ids)
print(f" Nalezeno {len(rows)} zpráv.")
stop = False
for row in rows:
info = parse_row(row["cells"])
target = os.path.join(DOWNLOAD_DIR, info["filename"])
if info["filename"] in already or os.path.exists(target):
print(f" [stop] Nalezena již stažená zpráva: {info['filename']}")
stop = True
break
dl_url = f"{DOWNLOAD_URL}?zprava_id={row['fileId']}"
try:
r = context.request.get(dl_url, headers={"Referer": VYPIS_URL}, timeout=30_000)
if not r.ok:
print(f" HTTP {r.status} příloha (id={row['fileId']})")
else:
body = r.body()
if not body[:len(HLAVICKA)].decode("ascii", errors="ignore").startswith(HLAVICKA):
print(f" přeskočeno (není výpis pojištěnců): {info['filename']}")
else:
with open(target, "wb") as fh:
fh.write(body)
print(f" OK: {info['filename']}")
already.add(info["filename"])
downloaded += 1
except Exception as e:
print(f" Chyba příloha (id={row['fileId']}): {e}")
time.sleep(1.0)
if stop:
break
page_num += 1
finally:
context.close()
return downloaded
# ---------------------------------------------------------------------------
# Sestavení XML a podpis žádosti
# ---------------------------------------------------------------------------
def build_xml() -> str:
"""Sestaví XML žádosti o aktuální výpis pojištěnců (bez data — aktuální snímek)."""
return (
f'<SchrankaZadost NazevSchranky="SEZNAM_KAP" NazevFiltru="SEZNAM_KAP">\r\n'
f'<PolozkaFiltru Nazev="nicoz">{ICZ_INTERNAL}</PolozkaFiltru>\r\n'
f'<PolozkaFiltru Nazev="trideni">{TRIDENI}</PolozkaFiltru>\r\n'
f'<PolozkaFiltru Nazev="typ">{TYP}</PolozkaFiltru>\r\n'
f'</SchrankaZadost>'
)
def sign_xml(xml: str) -> str:
"""Podepíše XML certifikátem (PKCS7 detached, bez certifikátu — server cert v podpisu odmítá)."""
with open(PFX_PATH, "rb") as f:
private_key, cert, _ = pkcs12.load_key_and_certificates(f.read(), PFX_PASSWORD)
pem = (
pkcs7.PKCS7SignatureBuilder()
.set_data(xml.encode("utf-8"))
.add_signer(cert, private_key, hashes.SHA256())
.sign(serialization.Encoding.PEM, [pkcs7.PKCS7Options.DetachedSignature, pkcs7.PKCS7Options.NoCerts])
.decode("ascii")
)
return pem.replace("\r\n", "\n").replace("\n", "\r\n")
def odeslat_zadost(session: requests.Session) -> str | None:
"""Odešle podepsanou žádost o aktuální výpis. Vrátí referenční číslo nebo None."""
xml = build_xml()
podpis = sign_xml(xml)
payload = {"schrXml": xml, "schrSign": podpis, "schrFiles": []}
r = session.post(SUBMIT_URL, json=payload, headers={
"Content-Type": "application/json; charset=UTF-8",
"X-Requested-With": "XMLHttpRequest",
"Referer": BASE_URL + "/",
})
r.raise_for_status()
try:
resp = r.json()
except Exception:
print(f" Odpověď není JSON: {r.text[:300]}")
return None
resp_str = json.dumps(resp, ensure_ascii=False)
if resp.get("errMsg") or resp.get("error"):
print(f" Chyba od serveru: {resp.get('errMsg') or resp.get('error')}")
return None
m = re.search(r'\b(1[5-9]\d{7})\b', resp_str)
ref = m.group(1) if m else None
if ref:
print(f" OK — ref. číslo: {ref}")
else:
print(f" Odpověď (bez ref. čísla): {resp_str[:300]}")
return ref or ("OK" if r.ok else None)
# ---------------------------------------------------------------------------
# Log
# ---------------------------------------------------------------------------
def uloz_log(ref_cislo: str) -> None:
log = []
if os.path.exists(LOG_FILE):
with open(LOG_FILE, encoding="utf-8") as f:
log = json.load(f)
log.append({
"ref_cislo": ref_cislo,
"podano_kdy": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
})
with open(LOG_FILE, "w", encoding="utf-8") as f:
json.dump(log, f, indent=2, ensure_ascii=False)
# ---------------------------------------------------------------------------
# Hlavní funkce
# ---------------------------------------------------------------------------
def hlavni() -> None:
# 1. Přihlášení — uloží cookies pro Playwright
prihlaseni()
# 2. Stažení nových výpisů z výpisové schránky
print("\n=== Stahování nových výpisů ===")
stazeno = stahni_nove_vypisy()
print(f"Staženo: {stazeno} souborů.\n")
# 3. Znovu přihlásit — Playwright mohl invalidovat předchozí session
print("=== Znovu přihlašuji před podáním ===")
session = prihlaseni()
# 4. Podání žádosti o aktuální výpis
print("=== Podávám žádost o aktuální výpis ===")
ref = odeslat_zadost(session)
if ref:
uloz_log(ref)
print(f"\nHotovo — žádost podána, ref: {ref}")
else:
print("\nPodání selhalo — žádost nebyla zaevidována.")
if __name__ == "__main__":
hlavni()
@@ -0,0 +1,6 @@
[
{
"ref_cislo": "179774959",
"podano_kdy": "2026-06-17 05:21:08"
}
]
@@ -1,7 +1,7 @@
[
{
"name": "SID",
"value": "d5b299e911197a26c190b5a84bebeac8",
"value": "786ed43afb46b3c7432371f7f2ee282e",
"domain": ".portal.ozp.cz",
"path": "/",
"expires": -1,
@@ -14,7 +14,7 @@
"value": "CERT",
"domain": ".portal.ozp.cz",
"path": "/",
"expires": 1813134421,
"expires": 1813202467,
"secure": true,
"httpOnly": false,
"sameSite": "Lax"