notebookvb

2026-05-01 09:43:21 +02:00
parent 1b904e3da0
commit 88602cb406
3 changed files with 468 additions and 6 deletions
@@ -0,0 +1,167 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+import sys as _sys
+_sys.stdout.reconfigure(encoding="utf-8", errors="replace")
+_sys.stderr.reconfigure(encoding="utf-8", errors="replace")
+"""
+zadej_davku.py
+==============
+Odešle asynchronní požadavek na VZP B2B službu SeznamRegPojistencuB2B
+(seznam zakapitovaných/registrovaných pojištěnců za dané období).
+
+Použití:
+    python zadej_davku.py [mesic] [rok]
+    python zadej_davku.py 2 2025       # únor 2025
+    python zadej_davku.py              # předchozí měsíc (výchozí)
+
+Výstup:
+    - korelační ID zprávy (pro pozdější spárování asynchronní odpovědi)
+    - stavVyrizeniPozadavku=2 znamená "přijato, VZP zpracovává"
+    - finální odpověď přijde asynchronně na AS2 endpoint partnera
+"""
+
+import sys
+import uuid
+import argparse
+from pathlib import Path
+from datetime import date, timedelta
+
+from requests_pkcs12 import Pkcs12Adapter
+import requests
+import xml.etree.ElementTree as ET
+
+# ── KONFIGURACE ───────────────────────────────────────────────────────────────
+
+PFX_PATH     = Path(__file__).resolve().parent.parent / "Certificates" / "picka.pfx"
+PFX_PASSWORD = "Vlado7309208104+"
+
+ICZ = "09305000"
+ENV = "prod"
+
+NS_VZP    = "http://xmlns.gemsystem.cz/SeznamRegPojistencuB2B"
+NS_COMMON = "http://xmlns.gemsystem.cz/CommonB2B"
+NS_SOAP   = "http://schemas.xmlsoap.org/soap/envelope/"
+
+ENDPOINT_SIMU = "https://simu.b2b.vzp.cz/B2BProxy/HttpProxy/SIMUSeznamRegPojistencuB2B?sluzba=SIMUSeznamRegPojistencuB2B"
+ENDPOINT_PROD = "https://prod.b2b.vzp.cz/B2BProxy/HttpProxy/SeznamRegPojistencuB2B"
+
+# ── ARGUMENTY ────────────────────────────────────────────────────────────────
+
+parser = argparse.ArgumentParser(description="Požadavek na seznam registrovaných pojištěnců VZP")
+parser.add_argument("mesic", nargs="?", type=int, help="Měsíc (1-12)")
+parser.add_argument("rok",   nargs="?", type=int, help="Rok (např. 2025)")
+parser.add_argument("--simu", action="store_true", help="Použít simulační prostředí")
+parser.add_argument("--pdf",  action="store_true", help="Výstup jako PDF (výchozí: text/plain)")
+args = parser.parse_args()
+
+# Výchozí: předchozí měsíc
+if args.mesic and args.rok:
+    mesic, rok = args.mesic, args.rok
+else:
+    prvni_tohoto = date.today().replace(day=1)
+    predchozi    = prvni_tohoto - timedelta(days=1)
+    mesic, rok   = predchozi.month, predchozi.year
+
+format_vystupu = "application/pdf" if args.pdf else "text/plain"
+endpoint       = ENDPOINT_SIMU if args.simu else ENDPOINT_PROD
+
+# ── KONTROLY ─────────────────────────────────────────────────────────────────
+
+if not PFX_PATH.exists():
+    print(f"CHYBA: certifikát nenalezen: {PFX_PATH}")
+    sys.exit(1)
+
+if not (1 <= mesic <= 12):
+    print(f"CHYBA: neplatný měsíc: {mesic}")
+    sys.exit(1)
+
+# ── SESTAVENÍ POŽADAVKU ───────────────────────────────────────────────────────
+
+id_zpravy = uuid.uuid4().hex[:12].upper()
+
+soap = f"""<?xml version="1.0" encoding="utf-8"?>
+<soap:Envelope
+    xmlns:soap="{NS_SOAP}"
+    xmlns:vzp="{NS_VZP}"
+    xmlns:com="{NS_COMMON}">
+
+  <soap:Header>
+    <com:idZpravy>{id_zpravy}</com:idZpravy>
+    <com:idSubjektu>
+      <com:icz>{ICZ}</com:icz>
+    </com:idSubjektu>
+  </soap:Header>
+
+  <soap:Body>
+    <vzp:seznamRegistrovanychPojistencuB2BPozadavek>
+      <vzp:idZpravy>{id_zpravy}</vzp:idZpravy>
+      <vzp:idSubjektu>{ICZ}</vzp:idSubjektu>
+      <vzp:typSubjektu>zp</vzp:typSubjektu>
+      <vzp:seznam>
+        <vzp:obdobiDavky>
+          <vzp:mesic>{mesic}</vzp:mesic>
+          <vzp:rok>{rok}</vzp:rok>
+        </vzp:obdobiDavky>
+        <vzp:formatVystupu>{format_vystupu}</vzp:formatVystupu>
+      </vzp:seznam>
+    </vzp:seznamRegistrovanychPojistencuB2BPozadavek>
+  </soap:Body>
+
+</soap:Envelope>"""
+
+# ── ODESLÁNÍ ─────────────────────────────────────────────────────────────────
+
+print(f"Období:      {mesic:02d}/{rok}")
+print(f"Formát:      {format_vystupu}")
+print(f"Prostředí:   {'SIMU' if args.simu else 'PROD'}")
+print(f"IČZ:         {ICZ}")
+print(f"ID zprávy:   {id_zpravy}")
+print(f"Endpoint:    {endpoint}")
+print()
+
+session = requests.Session()
+session.mount("https://", Pkcs12Adapter(
+    pkcs12_filename=str(PFX_PATH),
+    pkcs12_password=PFX_PASSWORD
+))
+
+try:
+    resp = session.post(
+        endpoint,
+        data=soap.encode("utf-8"),
+        headers={"Content-Type": "text/xml; charset=utf-8", "SOAPAction": "process"},
+        timeout=30
+    )
+except requests.RequestException as e:
+    print(f"CHYBA při spojení: {e}")
+    sys.exit(1)
+
+print(f"HTTP status: {resp.status_code}")
+print()
+
+# ── PARSOVÁNÍ ODPOVĚDI ────────────────────────────────────────────────────────
+
+if not resp.content:
+    print("→ Požadavek přijat (prázdná odpověď = asynchronní služba).")
+    print(f"→ VZP zpracuje a odešle výsledek na AS2 endpoint.")
+    print(f"→ ID zprávy pro spárování: {id_zpravy}")
+else:
+    try:
+        root = ET.fromstring(resp.text)
+        NS = {"soap": NS_SOAP, "vzp": NS_VZP}
+        korelace = root.find(".//vzp:korelaceZpravy", NS)
+        text_odp = root.find(".//vzp:textOdpovedi",   NS)
+        stav     = root.find(".//vzp:stavVyrizeniPozadavku", NS)
+
+        print(f"Korelace zprávy: {korelace.text if korelace is not None else '–'}")
+        print(f"Text odpovědi:   {text_odp.text  if text_odp  is not None else '–'}")
+        print(f"Stav vyřízení:   {stav.text      if stav      is not None else '–'}")
+
+        if stav is not None and stav.text == "2":
+            print()
+            print("→ VZP zpracovává — finální odpověď přijde asynchronně na AS2 endpoint.")
+            print(f"→ ID zprávy pro spárování: {id_zpravy}")
+
+    except ET.ParseError:
+        print("Nepodařilo se parsovat XML odpověď:")
+        print(resp.text)
@@ -1,7 +1,7 @@
 """
-Přihlásí se na VZP Point a stáhne nové zprávy.
+Přihlásí se na VZP Point, stáhne nové zprávy a aktualizuje číselníky.

-Kombinuje 01_prihlaseni.py + 03_stahuj_nove.py do jednoho spuštění.
+Kombinuje 01_prihlaseni.py + 03_stahuj_nove.py + 01_stahni_ciselniky.py.
 Přihlášení probíhá plně automaticky (Chrome auto-vybere certifikát).

 POUŽITÍ:
@@ -13,23 +13,29 @@ import sys
 import os

 DIR = os.path.dirname(os.path.abspath(__file__))
+CISELNIKY_SCRIPT = os.path.abspath(
+    os.path.join(DIR, "..", "..", "..", "Recepty", "StahovánízVZPWithClaude", "01_stahni_ciselniky.py")
+)


 def run(script: str) -> None:
    result = subprocess.run(
-        [sys.executable, os.path.join(DIR, script)],
+        [sys.executable, script],
        check=False,
    )
    if result.returncode != 0:
-        raise SystemExit(f"Skript {script} skončil s chybou (kód {result.returncode})")
+        raise SystemExit(f"Skript {os.path.basename(script)} skončil s chybou (kód {result.returncode})")


 def main() -> None:
    print("=== Přihlášení ===")
-    run("01_prihlaseni.py")
+    run(os.path.join(DIR, "01_prihlaseni.py"))

    print("\n=== Stahování nových zpráv ===")
-    run("03_stahuj_nove.py")
+    run(os.path.join(DIR, "03_stahuj_nove.py"))
+
+    print("\n=== Stahování číselníků VZP ===")
+    run(CISELNIKY_SCRIPT)


 if __name__ == "__main__":
@@ -0,0 +1,289 @@
+"""
+01 — Stáhni číselníky VZP z point.vzp.cz/Cms/Document (sekce „Číselníky").
+Soubory přejmenuje do formátu: YYYY-MM-DD NázevLinku (popis).zip
+Ověří co již existuje v cílovém adresáři a stáhne pouze chybějící soubory.
+Použití: python 01_stahni_ciselniky.py [--dry-run]
+"""
+
+import json
+import os
+import re
+import sys
+
+# Windows konzole - povol UTF-8 výstup
+if sys.stdout.encoding != "utf-8":
+    import io
+    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "Knihovny"))
+from najdi_dropbox import get_dropbox_root
+
+TARGET_DIR = os.path.join(
+    get_dropbox_root(),
+    "Ordinace", "Dokumentace_ke_zpracování", "Zúčtovací zprávy", "Číselníky"
+)
+
+DRY_RUN = False  # True = pouze zobraz co by se stáhlo, nic nestahuj
+
+CHROME_PROFILE = os.path.join(os.path.dirname(__file__), "chrome_profile")
+COOKIES_FILE   = os.path.join(os.path.dirname(__file__), "vzp_cookies.json")
+
+# Issuer CN certifikátu v Windows store (CurrentUser\My)
+CERT_ISSUER_CN = "I.CA Public CA/RSA 06/2022"
+
+
+def _set_chrome_cert_policy() -> None:
+    """Nastaví Chrome politiku AutoSelectCertificateForUrls — Chrome vybere certifikát automaticky."""
+    import winreg
+    policy = json.dumps({
+        "pattern": "https://[*.]vzp.cz",
+        "filter": {"ISSUER": {"CN": CERT_ISSUER_CN}},
+    })
+    key_path = r"SOFTWARE\Policies\Google\Chrome\AutoSelectCertificateForUrls"
+    try:
+        key = winreg.CreateKey(winreg.HKEY_CURRENT_USER, key_path)
+        winreg.SetValueEx(key, "1", 0, winreg.REG_SZ, policy)
+        winreg.CloseKey(key)
+        print(f"  Chrome politika nastavena (issuer: {CERT_ISSUER_CN})")
+    except Exception as e:
+        print(f"  Varování: nelze nastavit Chrome politiku: {e}")
+
+
+def load_cookies(context) -> int:
+    if not os.path.exists(COOKIES_FILE):
+        return 0
+    try:
+        with open(COOKIES_FILE, "r", encoding="utf-8") as f:
+            cookies = json.load(f)
+        context.add_cookies(cookies)
+        return len(cookies)
+    except Exception as e:
+        print(f"  Chyba při načítání cookies: {e}")
+        return 0
+
+
+def save_cookies(context) -> int:
+    try:
+        vzp = [c for c in context.cookies() if "vzp.cz" in c.get("domain", "")]
+        with open(COOKIES_FILE, "w", encoding="utf-8") as f:
+            json.dump(vzp, f, indent=2, ensure_ascii=False)
+        return len(vzp)
+    except Exception as e:
+        print(f"  Chyba při ukládání cookies: {e}")
+        return 0
+
+
+def sanitize_filename(name: str) -> str:
+    """Odstraní znaky nepovolené ve Windows názvech souborů."""
+    return re.sub(r'[\\/:*?"<>|]', '_', name).strip()
+
+
+# True = přidat popis v závorce do názvu souboru
+SECTIONS = {
+    "Číselníky":          True,
+    "Soubor platných IČP": False,
+}
+
+
+def extract_entries(page) -> list[dict]:
+    """
+    Projde tabulky sekcí definovaných v SECTIONS a vrátí seznam slovníků:
+      orig_fname  — původní název souboru z atributu download
+      new_fname   — nový název: YYYY-MM-DD Název [(popis)].zip
+      link        — Playwright Locator pro kliknutí
+    """
+    entries = []
+    for section, include_description in SECTIONS.items():
+        rows = page.locator(f"h3:has-text('{section}') + table tbody tr").all()
+        entries.extend(_extract_rows(rows, include_description=include_description))
+    return entries
+
+
+def _extract_rows(rows: list, include_description: bool = True) -> list[dict]:
+    entries = []
+    i = 0
+    while i < len(rows):
+        row = rows[i]
+        dl_links = row.locator("a[download]").all()
+
+        if not dl_links:
+            i += 1
+            continue
+
+        # Datum platnosti z <time datetime="YYYY-MM-DD">
+        time_el = row.locator("time[datetime]")
+        date_iso = time_el.get_attribute("datetime") if time_el.count() > 0 else ""
+
+        # Zobrazený název — přímý link nebo dropdown toggle
+        span = row.locator("a[download] span[title='Stáhnout']")
+        if span.count() > 0:
+            display_name = span.first.inner_text().strip()
+        else:
+            toggle = row.locator(".dropdown-toggle span[title='Stáhnout']")
+            display_name = toggle.first.inner_text().strip() if toggle.count() > 0 else ""
+
+        # Popis z následující řádky (má colspan="5" a div.text-gray)
+        description = ""
+        if i + 1 < len(rows):
+            desc_div = rows[i + 1].locator(".text-gray")
+            if desc_div.count() > 0:
+                description = desc_div.first.inner_text().strip()
+
+        is_multi = len(dl_links) > 1
+        for link in dl_links:
+            orig_fname = link.get_attribute("download") or ""
+            ext = os.path.splitext(orig_fname)[1] or ".zip"
+
+            if is_multi:
+                # Pro více variant (např. LEKY má "leky" a "lekyp") přidej základ
+                # původního názvu jako rozlišení: odstraní timestamp prefix (12 číslic-)
+                base = re.sub(r'^\d{12}-', '', orig_fname)
+                base = os.path.splitext(base)[0]
+                name = f"{date_iso} {display_name} {base} ({description}){ext}" if include_description else f"{date_iso} {display_name} {base}{ext}"
+            else:
+                name = f"{date_iso} {display_name} ({description}){ext}" if include_description else f"{date_iso} {display_name}{ext}"
+
+            entries.append({
+                "orig_fname": orig_fname,
+                "new_fname": sanitize_filename(name),
+                "link": link,
+            })
+
+        i += 1
+
+    return entries
+
+
+def get_existing_files() -> set[str]:
+    if not os.path.isdir(TARGET_DIR):
+        return set()
+    return {f for f in os.listdir(TARGET_DIR) if f.lower().endswith(".zip")}
+
+
+def ensure_logged_in(page, context):
+    """
+    Pokud jsme přesměrováni na login, provede přihlášení certifikátem
+    a vrátí stránku po přesměrování zpět na point.vzp.cz. Jinak vrátí původní page.
+    Při selhání vrátí None.
+    """
+    if not page.url.startswith("https://auth.vzp.cz/signin"):
+        return page
+
+    print("[stahování] Přihlašovací stránka — klikám na certifikát...")
+    cert_btn = page.locator("a, button").filter(has_text=re.compile(r"certifikát", re.I)).first
+    cert_btn.wait_for(state="visible", timeout=10_000)
+    cert_btn.click(no_wait_after=True)
+
+    print("[stahování] Čekám na přesměrování (max 60 s)...")
+    try:
+        page.wait_for_url("https://point.vzp.cz/**", timeout=60_000)
+    except Exception:
+        print(f"[stahování] Timeout čekání na přesměrování. URL: {page.url}")
+
+    if not page.url.startswith("https://point.vzp.cz"):
+        print(f"[stahování] Auth selhala, URL: {page.url}")
+        return None
+
+    print("[stahování] Přihlášení úspěšné.")
+    return page
+
+
+def download_ciselniky(dry_run: bool = False) -> None:
+    try:
+        from playwright.sync_api import sync_playwright
+    except ImportError:
+        print("[chyba] pip install playwright && playwright install chromium")
+        return
+
+    os.makedirs(TARGET_DIR, exist_ok=True)
+    existing = get_existing_files()
+    print(f"Cílový adresář:  {TARGET_DIR}")
+    print(f"Již staženo:     {len(existing)} souborů")
+
+    _set_chrome_cert_policy()
+
+    with sync_playwright() as p:
+        context = p.chromium.launch_persistent_context(
+            user_data_dir=CHROME_PROFILE,
+            channel="chrome",
+            headless=False,
+            slow_mo=200,
+            ignore_https_errors=True,
+            accept_downloads=True,
+        )
+        try:
+            loaded = load_cookies(context)
+            print(f"Cookies načtené z JSON: {loaded}")
+
+            page = context.new_page()
+
+            print("[stahování] Naviguji na VZP Point...")
+            try:
+                page.goto("https://point.vzp.cz/Cms/Document", wait_until="domcontentloaded", timeout=30_000)
+            except Exception as e:
+                print(f"[stahování] Navigace: {e}")
+
+            page = ensure_logged_in(page, context)
+            if page is None:
+                return
+
+            page.wait_for_load_state("networkidle", timeout=30_000)
+
+            entries = extract_entries(page)
+            print(f"[stahování] Nalezeno {len(entries)} souborů v sekci Číselníky.")
+            print()
+
+            downloaded = 0
+            skipped = 0
+
+            for entry in entries:
+                new_fname = entry["new_fname"]
+                orig_fname = entry["orig_fname"]
+
+                if new_fname in existing:
+                    print(f"  ✓ {new_fname}")
+                    skipped += 1
+                    continue
+
+                if dry_run:
+                    print(f"  → {new_fname}")
+                    print(f"    orig: {orig_fname}")
+                    downloaded += 1
+                    continue
+
+                dest = os.path.join(TARGET_DIR, new_fname)
+                try:
+                    with page.expect_download(timeout=60_000) as dl_info:
+                        entry["link"].dispatch_event("click")
+                    dl = dl_info.value
+                    dl.save_as(dest)
+                    size = os.path.getsize(dest)
+                    print(f"  ↓ {new_fname}  ({size:,} B)")
+                    downloaded += 1
+                except Exception as e:
+                    print(f"  ! {orig_fname}  CHYBA: {e}")
+
+        finally:
+            saved = save_cookies(context)
+            print(f"Uloženo {saved} VZP cookies.")
+            try:
+                context.close()
+            except Exception:
+                pass
+
+    print()
+    if dry_run:
+        print(f"[dry-run] Ke stažení: {downloaded}, přeskočeno: {skipped}")
+    else:
+        print(f"Staženo: {downloaded}, přeskočeno (již existovalo): {skipped}")
+
+
+def main():
+    dry_run = DRY_RUN or "--dry-run" in sys.argv
+    if dry_run:
+        print("[dry-run] Pouze zobrazuji co by se stáhlo, nic nestahuju.\n")
+    download_ciselniky(dry_run=dry_run)
+
+
+if __name__ == "__main__":
+    main()