""" 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 os import re import sys 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 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/. """ 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) 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, }], accept_downloads=True, ) 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) 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) print("[stahování] Přihlášeno.") # 2. Stránka s dokumenty page.goto("https://point.vzp.cz/Cms/Document", wait_until="networkidle", timeout=30_000) # 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] # 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 zip_path = os.path.join(IMPORT_DIR, download.suggested_filename) download.save_as(zip_path) browser.close() # 5. Rozbal Lh7 z archivu 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") 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: 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()