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

notebookvb

Browse Source
This commit is contained in:
Vlado
2026-04-13 07:13:15 +02:00
parent 068c8edbe1
commit b2b46c1571
290 changed files with 336 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
StahovánízVZPWithClaude/import_vzp_pracoviste.py
+123 -42
View File
@@ -7,9 +7,12 @@ Použití: python import_vzp_pracoviste.py [--no-download] [soubor.Lh7]
import csv
import glob
import io
import json
import os
import re
import sys
import time
import winreg
import zipfile
from datetime import date, datetime
@@ -25,6 +28,7 @@ DB_CONFIG = {
"user": "root",
"password": "Vlado9674+",
"database": "medicus",
"charset": "utf8mb4",
}
@@ -73,10 +77,22 @@ def parse_date(s: str) -> date | None:
return None
def _delete_chrome_cert_policy() -> None:
"""Smaže AutoSelectCertificateForUrls politiku — Chrome pak zobrazí dialog přirozeně."""
key_path = r"SOFTWARE\Policies\Google\Chrome\AutoSelectCertificateForUrls"
try:
key = winreg.OpenKey(winreg.HKEY_CURRENT_USER, key_path, access=winreg.KEY_SET_VALUE)
winreg.DeleteValue(key, "1")
winreg.CloseKey(key)
except Exception:
pass # Klíč neexistuje — OK
def download_latest_file() -> str | None:
"""
Použije Playwright (Chromium) s PFX certifikátem pro přihlášení na VZP Point.
Klikne na 'Soubor platných IČP', zachytí stažený ZIP, rozbalí Lh7 do Import/.
Použije Playwright (headful Chrome) s persistent profilem pro přihlášení na VZP Point.
Chrome vidí Windows Certificate Store — dialog výběru certifikátu se zobrazí automaticky.
Po přihlášení stáhne ZIP přes API a rozbalí Lh7 do Import/.
"""
try:
from playwright.sync_api import sync_playwright
@@ -86,65 +102,130 @@ def download_latest_file() -> str | None:
os.makedirs(IMPORT_DIR, exist_ok=True)
# Smaž politiku AutoSelect — VZP JavaScript potřebuje nativní Chrome cert dialog.
_delete_chrome_cert_policy()
# Vlastní profil: Chrome si pamatuje session cookies mezi spuštěními.
chrome_profile = os.path.join(os.path.dirname(__file__), "chrome_profile")
zip_path = None
zip_bytes = None
with sync_playwright() as p:
browser = p.chromium.launch(headless=True)
context = browser.new_context(
client_certificates=[{
"origin": "https://auth.vzp.cz",
"pfxPath": VZP_CERT_FILE,
"passphrase": VZP_CERT_PASSWORD,
}],
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,
args=["--force-renderer-accessibility"], # pomáhá UIA najít Chrome dialogy
)
page = context.new_page()
try:
page = context.new_page()
# 1. Přihlášení — klik na Certifikát, Playwright certifikát předloží automaticky
print("[stahování] Přihlašování na VZP Point...")
page.goto("https://point.vzp.cz/", wait_until="networkidle", timeout=30_000)
# Naviguj na cílovou stránku — server přesměruje na login pokud není session
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}")
cert_btn = page.locator("text=Certifikát").first
if cert_btn.is_visible(timeout=5_000):
cert_btn.click()
page.wait_for_url("**/point.vzp.cz/**", timeout=30_000)
page.wait_for_load_state("networkidle", timeout=30_000)
# Pokud jsme přesměrováni na přihlašovací stránku, klikni na "Certifikát"
if page.url.startswith("https://auth.vzp.cz/signin"):
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)
print("[stahování] Přihlášeno.")
cert_btn.click(no_wait_after=True)
# 2. Stránka s dokumenty
page.goto("https://point.vzp.cz/Cms/Document", wait_until="networkidle", timeout=30_000)
# Cert dialog: uživatel vybere ručně (při prvním spuštění).
# Session se uloží do chrome_profile/ — příště dialog nebude.
print("[stahování] Pokud se zobrazí cert dialog, vyberte certifikát ručně (max 60 s)...")
time.sleep(30) # čas na ruční výběr + auth redirect
# 3. Zkontroluj aktuální název souboru
icp_link = page.locator("a[download*='-icp.zip']").first
zip_name = icp_link.get_attribute("download", timeout=5_000)
if zip_name:
year = zip_name[:2]
existing = glob.glob(os.path.join(IMPORT_DIR, f"PLP111{year}.Lh7"))
if existing:
print(f"[stahování] {os.path.basename(existing[0])} již existuje — přeskočeno.")
browser.close()
return existing[0]
# Otevři novou stránku a naviguj přímo — session cookie je nastavena
page = context.new_page()
try:
page.goto(
"https://point.vzp.cz/Cms/Document",
wait_until="domcontentloaded",
timeout=30_000,
)
except Exception as e:
print(f"[stahování] Navigace po auth: {e}")
# 4. Stáhni ZIP
print(f"[stahování] Stahuji {zip_name or 'ICP soubor'}...")
with page.expect_download(timeout=60_000) as dl_info:
icp_link.click()
download = dl_info.value
if not page.url.startswith("https://point.vzp.cz"):
print(f"[stahování] Auth selhala, URL: {page.url}")
return None
print(f"[stahování] Přihlášení úspěšné. URL: {page.url}")
zip_path = os.path.join(IMPORT_DIR, download.suggested_filename)
download.save_as(zip_path)
browser.close()
if not page.url.startswith("https://point.vzp.cz"):
print(f"[stahování] Přihlášení selhalo, URL: {page.url}")
return None
# 5. Rozbal Lh7 z archivu
print(f"[stahování] Přihlášení ověřeno. URL: {page.url}")
# Naviguj na stránku s dokumenty a počkej na plné načtení
if "Cms/Document" not in page.url:
page.goto("https://point.vzp.cz/Cms/Document", wait_until="networkidle", timeout=30_000)
else:
page.wait_for_load_state("networkidle", timeout=30_000)
# === Stáhni VŠECHNY číselníky ze stránky ===
all_links = page.locator("a[download]").all()
print(f"[stahování] Nalezeno {len(all_links)} souborů ke stažení.")
for link in all_links:
fname = link.get_attribute("download")
if not fname:
continue
dest_path = os.path.join(IMPORT_DIR, fname)
if os.path.exists(dest_path):
print(f"[stahování] {fname} — již existuje, přeskočeno.")
continue
try:
with page.expect_download(timeout=30_000) as dl_info:
link.dispatch_event("click") # obejde i display:none
dl = dl_info.value
dl.save_as(dest_path)
size = os.path.getsize(dest_path)
print(f"[stahování] {fname} — staženo ({size:,} B)")
except Exception as e:
print(f"[stahování] {fname} — CHYBA: {e}")
# === IČP: najdi stažený zip a rozbal Lh7 pro import ===
icp_zips = glob.glob(os.path.join(IMPORT_DIR, "*-icp.zip"))
if icp_zips:
zip_path = max(icp_zips, key=os.path.getmtime)
zip_name = os.path.basename(zip_path)
else:
print("[stahování] IČP zip nenalezen.")
zip_path = None
finally:
# Vždy zavřít kontext — Chrome zapíše session cookies na disk
try:
context.close()
except Exception:
pass # Browser už byl zavřen uživatelem — neškodné
# Rozbal Lh7 z IČP archivu
if not zip_path:
return None
try:
with zipfile.ZipFile(zip_path) as zf:
lh7_names = [n for n in zf.namelist() if n.lower().endswith(".lh7")]
if not lh7_names:
print("[stahování] ZIP neobsahuje .Lh7 soubor")
print("[stahování] IČP ZIP neobsahuje .Lh7 soubor")
return None
dest = os.path.join(IMPORT_DIR, os.path.basename(lh7_names[0]))
with zf.open(lh7_names[0]) as src, open(dest, "wb") as out:
out.write(src.read())
os.remove(zip_path)
print(f"[stahování] Rozbaleno: {os.path.basename(dest)} ({os.path.getsize(dest):,} B)")
return dest
except Exception as e: