405 lines
16 KiB
Python
405 lines
16 KiB
Python
# ============================================================
|
||
# download_vault_v1.5.py
|
||
# Verze: 1.5
|
||
# Datum: 2026-06-12
|
||
# Popis: Stažení source-file dokumentu z Veeva Vault (J&J V-TMF).
|
||
# Perzistentní Chromium profil (session přežije běhy),
|
||
# login přes J&J SSO formulář, ruční potvrzení 2FA na
|
||
# telefonu, stažení Source File a přejmenování podle
|
||
# VTMF čísla do U:\Dropbox\!!!Days\Downloads Z230\.
|
||
#
|
||
# v1.1: VAULT_USER / VAULT_PASS se čtou ze souboru .env v rootu
|
||
# projektu Janssen; env proměnné mají přednost.
|
||
# v1.2: když údaje chybí (nový počítač — .env se přes Giteu
|
||
# nepřenáší), skript .env založí / doplní v něm chybějící
|
||
# řádky VAULT_USER=/VAULT_PASS=, vyzve k doplnění a skončí.
|
||
# v1.3: robustnější odeslání login formuláře (Sign On / Login /
|
||
# submit / Enter fallbacky).
|
||
# v1.4: maintenance okno je jQuery UI dialog (.vv_login_msg_dialog);
|
||
# jeho "OK" v patičce je nefunkční — jediné funkční zavření je
|
||
# křížek .ui-dialog-titlebar-close. Zavírá se ve smyčce (dialogy
|
||
# umí být ve frontě), s opakovanou kontrolou kvůli zpožděnému
|
||
# zobrazení; záloha klávesa Escape.
|
||
# v1.5: dialog se hledá ve VŠECH frames (Veeva renderuje části UI
|
||
# v iframech — selektor na hlavní stránce ho pak nevidí).
|
||
# Když dialog zůstane viditelný i po pokusech o zavření,
|
||
# skript sám uloží diagnostiku do debug/<čas>/ (screenshot
|
||
# + HTML všech frames + výpis kandidátů na tlačítka).
|
||
# Heslo se NIKDY nedává natvrdo do skriptu.
|
||
# ============================================================
|
||
|
||
import os
|
||
import re
|
||
import sys
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
from playwright.sync_api import sync_playwright, TimeoutError as PWTimeout
|
||
|
||
# --- Konfigurace -------------------------------------------------------
|
||
|
||
LOGIN_URL = ("https://fedlogin.jnj.com/idp/eyJ2c2lkIjoiam5qX3ZlZXZhIn0/"
|
||
"startSSO.ping?PartnerSpId=janssenetmf.veevavault.com"
|
||
"&IdpAdapterId=CompIWALDAPEXTFORM"
|
||
"&TargetResource=https%3A%2F%2Fvtmf.veevavault.com%2F")
|
||
|
||
VAULT_UI_PATTERN = "**vtmf.veevavault.com/ui**" # úspěšný vstup do Vaultu
|
||
|
||
SCRIPT_DIR = Path(__file__).resolve().parent
|
||
PROFILE_DIR = SCRIPT_DIR / "vault_profile" # perzistentní session
|
||
ENV_FILE = SCRIPT_DIR.parent / ".env" # root projektu Janssen
|
||
DEBUG_DIR = SCRIPT_DIR / "debug" # diagnostické výstupy
|
||
DOWNLOAD_DIR = Path(r"U:\Dropbox\!!!Days\Downloads Z230")
|
||
|
||
# Zatím jeden dokument; struktura připravená na pozdější seznam
|
||
# (ve fázi 2 se naplní z Excelu).
|
||
DOCUMENTS = [
|
||
("https://vtmf.veevavault.com/ui/#doc_info/31824736/1/0", "VTMF-25690359"),
|
||
]
|
||
|
||
|
||
def log(msg):
|
||
print(msg, flush=True)
|
||
|
||
|
||
def load_env_file(path):
|
||
"""Načte KEY=VALUE řádky z .env do os.environ.
|
||
Už nastavené env proměnné mají přednost, .env je nepřepisuje."""
|
||
if not path.exists():
|
||
log(f"[!] .env nenalezen: {path}")
|
||
return
|
||
for line in path.read_text(encoding="utf-8").splitlines():
|
||
line = line.strip()
|
||
if not line or line.startswith("#") or "=" not in line:
|
||
continue
|
||
key, _, value = line.partition("=")
|
||
key, value = key.strip(), value.strip().strip('"').strip("'")
|
||
if value and key not in os.environ:
|
||
os.environ[key] = value
|
||
|
||
|
||
ENV_SECTION_HEADER = "# --- Veeva Vault (J&J V-TMF) — VTMFDownloadFiles/download_vault ---"
|
||
ENV_KEYS = ("VAULT_USER", "VAULT_PASS")
|
||
|
||
|
||
def ensure_credentials():
|
||
"""Načte .env; pokud VAULT_USER/VAULT_PASS chybí, založí/doplní
|
||
v .env šablonu, vyzve uživatele k doplnění a ukončí skript."""
|
||
load_env_file(ENV_FILE)
|
||
if all(os.environ.get(k) for k in ENV_KEYS):
|
||
return
|
||
|
||
existing = ENV_FILE.read_text(encoding="utf-8") if ENV_FILE.exists() else ""
|
||
# doplnit jen řádky, které v souboru úplně chybí (prázdné hodnoty nechat)
|
||
missing_lines = [f"{k}=" for k in ENV_KEYS
|
||
if not re.search(rf"^\s*{k}\s*=", existing, re.M)]
|
||
|
||
if not ENV_FILE.exists():
|
||
ENV_FILE.write_text(
|
||
"# .env — lokální přihlašovací údaje (NEVERZOVAT, je v .gitignore)\n\n"
|
||
+ ENV_SECTION_HEADER + "\n"
|
||
+ "\n".join(missing_lines) + "\n",
|
||
encoding="utf-8")
|
||
log(f"[i] Založil jsem nový .env: {ENV_FILE}")
|
||
elif missing_lines:
|
||
with open(ENV_FILE, "a", encoding="utf-8") as f:
|
||
f.write("\n" + ENV_SECTION_HEADER + "\n"
|
||
+ "\n".join(missing_lines) + "\n")
|
||
log(f"[i] Doplnil jsem chybějící řádky do .env: {ENV_FILE}")
|
||
|
||
print("\n" + "=" * 60)
|
||
print(" CHYBÍ PŘIHLAŠOVACÍ ÚDAJE.")
|
||
print(f" Doplň VAULT_USER a VAULT_PASS do souboru:")
|
||
print(f" {ENV_FILE}")
|
||
print(" a spusť skript znovu.")
|
||
print("=" * 60)
|
||
sys.exit(1)
|
||
|
||
|
||
# --- Přihlášení --------------------------------------------------------
|
||
|
||
def submit_login_form(page, password_box):
|
||
"""Odešle login formulář. Zkouší postupně tlačítka Sign On / Login /
|
||
OK / submit input; když žádné nenajde, stiskne Enter v poli hesla."""
|
||
candidates = [
|
||
page.get_by_role("button", name=re.compile("sign\\s*on", re.I)),
|
||
page.get_by_role("button", name=re.compile("log\\s*in|sign\\s*in", re.I)),
|
||
page.locator("input[type='submit']"),
|
||
page.locator("button[type='submit']"),
|
||
page.get_by_role("button", name=re.compile("^ok$", re.I)),
|
||
]
|
||
for loc in candidates:
|
||
try:
|
||
if loc.count() and loc.first.is_visible():
|
||
label = (loc.first.inner_text() or
|
||
loc.first.get_attribute("value") or "submit").strip()
|
||
log(f"[i] Odesílám formulář tlačítkem '{label}'...")
|
||
loc.first.click()
|
||
return
|
||
except Exception:
|
||
continue
|
||
log("[i] Tlačítko nenalezeno, odesílám Enterem v poli hesla...")
|
||
password_box.press("Enter")
|
||
|
||
|
||
def login_if_needed(page):
|
||
"""Otevře login URL, vyplní jméno+heslo, detekuje 2FA a počká na
|
||
ruční potvrzení. Pokud perzistentní session žije, login přeskočí."""
|
||
log(f"[i] Otevírám přihlašovací URL...")
|
||
page.goto(LOGIN_URL, wait_until="domcontentloaded")
|
||
|
||
# Perzistentní session může rovnou přesměrovat do Vaultu
|
||
if "vtmf.veevavault.com/ui" in page.url:
|
||
log("[i] Už přihlášen (perzistentní session).")
|
||
return
|
||
|
||
# SSO redirecty chvíli trvají — čekáme buď na login formulář,
|
||
# nebo na samovolný vstup do Vaultu.
|
||
user_box = page.locator("input[type='text']").first
|
||
try:
|
||
user_box.wait_for(timeout=8000)
|
||
except PWTimeout:
|
||
if "vtmf.veevavault.com/ui" in page.url:
|
||
log("[i] Přihlášen bez formuláře (session redirect).")
|
||
return
|
||
raise RuntimeError(
|
||
f"Nenašel jsem login formulář ani Vault. Aktuální URL: {page.url}")
|
||
|
||
# údaje z .env v rootu projektu (případně env proměnné);
|
||
# jejich existenci už ověřil ensure_credentials() na startu
|
||
username = os.environ["VAULT_USER"]
|
||
password = os.environ["VAULT_PASS"]
|
||
|
||
log("[i] Vyplňuji přihlašovací údaje...")
|
||
user_box.fill(username)
|
||
password_box = page.locator("input[type='password']").first
|
||
password_box.fill(password)
|
||
submit_login_form(page, password_box)
|
||
|
||
# --- Detekce 2FA / čekání na vstup do Vaultu ---
|
||
log("[i] Odeslán login, čekám na výsledek...")
|
||
try:
|
||
page.wait_for_url(VAULT_UI_PATTERN, timeout=15000)
|
||
log("[ok] Přihlášen rovnou (bez 2FA).")
|
||
return
|
||
except PWTimeout:
|
||
pass # nejsme ve Vaultu -> pravděpodobně 2FA výzva
|
||
|
||
# Pojistka: pokud formulář hlásí chybu (špatné heslo), nečekat na 2FA
|
||
err = page.locator("text=/invalid|incorrect|failed/i")
|
||
try:
|
||
if err.count() and err.first.is_visible():
|
||
raise RuntimeError(f"Login selhal: {err.first.inner_text().strip()}")
|
||
except PWTimeout:
|
||
pass
|
||
|
||
print("\n" + "=" * 60)
|
||
print(" VYŽADOVÁNO OVĚŘENÍ NA TELEFONU (2FA).")
|
||
print(" Potvrď přihlášení v mobilní aplikaci.")
|
||
print("=" * 60)
|
||
input(" Až to potvrdíš, stiskni ENTER pro pokračování... ")
|
||
|
||
page.wait_for_url(VAULT_UI_PATTERN, timeout=120000)
|
||
log("[ok] Přihlášení dokončeno.")
|
||
|
||
|
||
def verify_inside(page):
|
||
"""Ověří, že jsme uvnitř Vaultu (URL na /ui)."""
|
||
page.wait_for_url(VAULT_UI_PATTERN, timeout=30000)
|
||
log(f"[ok] Uvnitř Vaultu: {page.url}")
|
||
|
||
|
||
def dialog_visible_anywhere(page):
|
||
"""Vrátí frame, ve kterém je viditelný jQuery UI dialog, jinak None.
|
||
Prochází hlavní stránku i všechny iframy."""
|
||
for frame in page.frames:
|
||
try:
|
||
dlg = frame.locator(".ui-dialog")
|
||
if dlg.count() and dlg.first.is_visible():
|
||
return frame
|
||
except Exception:
|
||
continue # frame se mezitím odpojil
|
||
return None
|
||
|
||
|
||
def save_dialog_debug(page, tag):
|
||
"""Uloží diagnostiku neumlčitelného dialogu: screenshot, HTML všech
|
||
frames a výpis kandidátů na zavírací tlačítka. Vrátí cestu složky."""
|
||
out = DEBUG_DIR / datetime.now().strftime(f"%Y-%m-%d_%H-%M-%S_{tag}")
|
||
out.mkdir(parents=True, exist_ok=True)
|
||
try:
|
||
page.screenshot(path=str(out / "screenshot.png"), full_page=False)
|
||
except Exception as e:
|
||
(out / "screenshot_error.txt").write_text(str(e), encoding="utf-8")
|
||
report = []
|
||
for i, frame in enumerate(page.frames):
|
||
report.append(f"=== frame[{i}] url={frame.url}")
|
||
try:
|
||
(out / f"frame_{i}.html").write_text(frame.content(),
|
||
encoding="utf-8")
|
||
for sel in (".ui-dialog", ".ui-dialog-titlebar-close",
|
||
"button", "input[type='button']"):
|
||
n = frame.locator(sel).count()
|
||
if n:
|
||
report.append(f" {sel}: {n}x")
|
||
except Exception as e:
|
||
report.append(f" [chyba čtení framu: {e}]")
|
||
(out / "frames_report.txt").write_text("\n".join(report),
|
||
encoding="utf-8")
|
||
log(f"[!] Diagnostika dialogu uložena do: {out}")
|
||
return out
|
||
|
||
|
||
def dismiss_maintenance_popup(page, attempts=3):
|
||
"""Zavře Veeva login/maintenance jQuery-UI dialog(y), pokud jsou
|
||
zobrazené. Modré 'OK' v patičce dialogu je nefunkční — jediné
|
||
funkční zavření je křížek .ui-dialog-titlebar-close v titulku.
|
||
Hledá se ve VŠECH frames (dialog může žít v iframe). Dialogy umí
|
||
být ve frontě (zavírá se ve smyčce) a objevují se i se zpožděním
|
||
po načtení stránky (attempts × 1 s). Bezpečné volat vždy."""
|
||
closed = 0
|
||
for attempt in range(attempts):
|
||
# zavírej křížkem ve všech frames, dokud nějaký dialog existuje
|
||
for _ in range(5):
|
||
clicked = False
|
||
for frame in page.frames:
|
||
try:
|
||
btn = frame.locator(".ui-dialog-titlebar-close")
|
||
if btn.count() and btn.first.is_visible():
|
||
btn.first.click()
|
||
page.wait_for_timeout(300)
|
||
closed += 1
|
||
where = ("hlavní stránka" if frame is page.main_frame
|
||
else f"iframe {frame.url[:60]}")
|
||
log(f"[i] Zavřel jsem dialog křížkem ({where}).")
|
||
clicked = True
|
||
break
|
||
except Exception:
|
||
continue # frame/dialog mezitím zmizel
|
||
if not clicked:
|
||
break
|
||
if closed and not dialog_visible_anywhere(page):
|
||
return True
|
||
if attempt < attempts - 1:
|
||
page.wait_for_timeout(1000) # okno se může teprve objevit
|
||
|
||
# záloha: jQuery UI dialog s closeOnEscape reaguje na Escape
|
||
if dialog_visible_anywhere(page):
|
||
page.keyboard.press("Escape")
|
||
page.wait_for_timeout(500)
|
||
log("[i] Zkusil jsem dialog zavřít klávesou Escape.")
|
||
|
||
# dialog pořád viditelný -> zachytit diagnostiku pro analýzu
|
||
stuck = dialog_visible_anywhere(page)
|
||
if stuck:
|
||
save_dialog_debug(page, "dialog")
|
||
print("\n" + "=" * 60)
|
||
print(" DIALOG SE NEPODAŘILO ZAVŘÍT AUTOMATICKY.")
|
||
print(" Zavři ho prosím ručně v prohlížeči.")
|
||
print(" (Diagnostika uložena, viz cesta výše.)")
|
||
print("=" * 60)
|
||
input(" Po ručním zavření stiskni ENTER... ")
|
||
return bool(closed)
|
||
|
||
|
||
# --- Stažení -----------------------------------------------------------
|
||
|
||
def build_target_name(suggested, vtmf):
|
||
"""Původní název + ' [VTMF-XXXXXX]' před příponou."""
|
||
if "." in suggested:
|
||
base, ext = suggested.rsplit(".", 1)
|
||
return f"{base} [{vtmf}].{ext}"
|
||
return f"{suggested} [{vtmf}]"
|
||
|
||
|
||
def find_source_file_button(page):
|
||
"""Najde ikonu Source File (list papíru se šipkou dolů, vpravo nahoře).
|
||
Více fallback selektorů — DOM se může lišit podle typu dokumentu."""
|
||
candidates = [
|
||
"[title='Source File']",
|
||
"[aria-label='Source File']",
|
||
]
|
||
for sel in candidates:
|
||
loc = page.locator(sel)
|
||
if loc.count():
|
||
return loc.first
|
||
# fallback: tlačítko/role s názvem Source File
|
||
loc = page.get_by_role("button", name=re.compile("Source File", re.I))
|
||
if loc.count():
|
||
return loc.first
|
||
return None
|
||
|
||
|
||
def download_source_file(page, doc_url, vtmf):
|
||
log(f"[i] Otevírám dokument {vtmf} ...")
|
||
page.goto(doc_url, wait_until="domcontentloaded")
|
||
try:
|
||
page.wait_for_load_state("networkidle", timeout=30000)
|
||
except PWTimeout:
|
||
log("[!] networkidle nenastal do 30 s, zkouším pokračovat...")
|
||
dismiss_maintenance_popup(page)
|
||
|
||
target = find_source_file_button(page)
|
||
if target is None:
|
||
raise RuntimeError(
|
||
f"Nenašel jsem ikonu 'Source File' na stránce dokumentu {vtmf}.")
|
||
|
||
log("[i] Klikám na Source File a čekám na download...")
|
||
with page.expect_download(timeout=60000) as dl_info:
|
||
target.click()
|
||
# Varianta s dropdownem (Source File + Viewable Rendition):
|
||
# pokud klik otevřel menu místo downloadu, klikni na položku menu.
|
||
try:
|
||
item = page.get_by_role("menuitem",
|
||
name=re.compile("Source File", re.I))
|
||
if item.count() and item.first.is_visible():
|
||
log("[i] Otevřel se dropdown, vybírám 'Source File'...")
|
||
item.first.click()
|
||
except Exception:
|
||
pass
|
||
download = dl_info.value
|
||
|
||
new_name = build_target_name(download.suggested_filename, vtmf)
|
||
DOWNLOAD_DIR.mkdir(parents=True, exist_ok=True)
|
||
dest = DOWNLOAD_DIR / new_name
|
||
download.save_as(str(dest))
|
||
log(f"[ok] Uloženo: {dest}")
|
||
return dest
|
||
|
||
|
||
# --- Main --------------------------------------------------------------
|
||
|
||
def main():
|
||
ensure_credentials()
|
||
DOWNLOAD_DIR.mkdir(parents=True, exist_ok=True)
|
||
with sync_playwright() as p:
|
||
ctx = p.chromium.launch_persistent_context(
|
||
user_data_dir=str(PROFILE_DIR),
|
||
headless=False,
|
||
accept_downloads=True,
|
||
no_viewport=True, # okno se chová nativně
|
||
args=["--start-maximized"],
|
||
)
|
||
page = ctx.pages[0] if ctx.pages else ctx.new_page()
|
||
ok, failed = [], []
|
||
try:
|
||
login_if_needed(page)
|
||
verify_inside(page)
|
||
dismiss_maintenance_popup(page)
|
||
for doc_url, vtmf in DOCUMENTS:
|
||
try:
|
||
ok.append(download_source_file(page, doc_url, vtmf))
|
||
except Exception as e:
|
||
log(f"[CHYBA] {vtmf}: {e}")
|
||
failed.append(vtmf)
|
||
finally:
|
||
log(f"\n[i] Hotovo: {len(ok)} staženo, {len(failed)} chyb"
|
||
+ (f" ({', '.join(failed)})" if failed else ""))
|
||
input("ENTER pro zavření prohlížeče...")
|
||
ctx.close()
|
||
sys.exit(1 if failed else 0)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|