""" Import VZP číselníku pracovišť (soubory *.Lh7) do MySQL tabulky vzp_pracoviste. Před importem automaticky stáhne nejnovější soubor z VZP Point (vyžaduje certifikát). 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 # 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") import mysql.connector DB_CONFIG = { "host": "192.168.1.76", "user": "root", "password": "Vlado9674+", "database": "medicus", "charset": "utf8mb4", } IMPORT_DIR = os.path.join(os.path.dirname(__file__), "Import") VZP_CERT_FILE = os.path.join(os.path.dirname(__file__), "MichalkaPublicCertProPython.pfx") VZP_CERT_PASSWORD = "Vlado7309208104++" VZP_DOCUMENT_ID = 5283 # "Soubor platných IČP" na point.vzp.cz/Cms/Document CREATE_TABLE_SQL = """ CREATE TABLE IF NOT EXISTS vzp_pracoviste ( id INT NOT NULL AUTO_INCREMENT, ico CHAR(8) NOT NULL, icz CHAR(8) NOT NULL, icp CHAR(8) NOT NULL, odbornost VARCHAR(4) NOT NULL, platnost_od DATE NOT NULL, platnost_do DATE NOT NULL, nazev_zarizeni VARCHAR(200), nazev_pracoviste VARCHAR(200), ulice VARCHAR(150), mesto VARCHAR(100), psc CHAR(5), PRIMARY KEY (id), INDEX idx_icp (icp), INDEX idx_icz (icz), INDEX idx_odbornost (odbornost), INDEX idx_platnost (platnost_od, platnost_do) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """ BATCH_SIZE = 1000 def parse_date(s: str) -> date | None: """Převede DDMMYYYY na date. Rok 3000 → 9999-12-31.""" s = s.strip() if len(s) != 8: return None try: d, m, y = int(s[0:2]), int(s[2:4]), int(s[4:8]) if y >= 3000: return date(9999, 12, 31) return date(y, m, d) except ValueError: 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 (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 except ImportError: print("[stahování] pip install playwright && playwright install chromium") return 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: 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 ) try: page = context.new_page() # 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}") # 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) cert_btn.click(no_wait_after=True) # 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 # 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}") 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}") if not page.url.startswith("https://point.vzp.cz"): print(f"[stahování] Přihlášení selhalo, URL: {page.url}") return None 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í] 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()) print(f"[stahování] Rozbaleno: {os.path.basename(dest)} ({os.path.getsize(dest):,} B)") return dest except Exception as e: print(f"[stahování] Chyba při rozbalování: {e}") return None def find_latest_file() -> str: files = glob.glob(os.path.join(IMPORT_DIR, "*.Lh7")) if not files: raise FileNotFoundError(f"Žádný *.Lh7 soubor nenalezen v {IMPORT_DIR}") return max(files, key=os.path.getmtime) def import_file(filepath: str, conn: mysql.connector.MySQLConnection) -> int: cursor = conn.cursor() cursor.execute("DROP TABLE IF EXISTS vzp_pracoviste") cursor.execute(CREATE_TABLE_SQL) conn.commit() insert_sql = """ INSERT INTO vzp_pracoviste (ico, icz, icp, odbornost, platnost_od, platnost_do, nazev_zarizeni, nazev_pracoviste, ulice, mesto, psc) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """ batch = [] total = 0 skipped = 0 with open(filepath, encoding="cp1250", errors="replace", newline="") as f: reader = csv.reader(f, quotechar='"', skipinitialspace=True) for row in reader: if len(row) < 10: skipped += 1 continue platnost_od = parse_date(row[4]) platnost_do = parse_date(row[5]) if platnost_od is None or platnost_do is None: skipped += 1 continue # ulice = název ulice + číslo popisné + číslo orientační (sloupce 11,12,13) ulice_parts = [row[11].strip(), row[12].strip(), row[13].strip()] if len(row) > 13 else [] ulice = " ".join(p for p in ulice_parts if p) or row[8].strip() psc = row[14].strip() if len(row) > 14 else "" if len(psc) > 5: psc = psc[:5] batch.append(( row[0].strip(), # ico row[1].strip(), # icz row[2].strip(), # icp row[3].strip(), # odbornost platnost_od, platnost_do, row[6].strip()[:200] if len(row) > 6 else "", # nazev_zarizeni row[7].strip()[:200] if len(row) > 7 else "", # nazev_pracoviste ulice[:150], row[9].strip()[:100] if len(row) > 9 else "", # mesto psc, )) if len(batch) >= BATCH_SIZE: cursor.executemany(insert_sql, batch) conn.commit() total += len(batch) batch.clear() if batch: cursor.executemany(insert_sql, batch) conn.commit() total += len(batch) cursor.close() return total, skipped def main(): args = sys.argv[1:] no_download = "--no-download" in args args = [a for a in args if a != "--no-download"] if args: filepath = args[0] else: if not no_download: downloaded = download_latest_file() if downloaded is None: print("[stahování] Pokračuji s lokálním souborem...") filepath = find_latest_file() filename = os.path.basename(filepath) print(f"Soubor: {filename}") print(f"Databáze: {DB_CONFIG['host']}/{DB_CONFIG['database']}") print(f"Začátek: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") conn = mysql.connector.connect(**DB_CONFIG) try: total, skipped = import_file(filepath, conn) finally: conn.close() print(f"Importováno: {total} záznamů") if skipped: print(f"Přeskočeno: {skipped} řádků (neúplná data)") print(f"Hotovo: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") if __name__ == "__main__": main()