z230
This commit is contained in:
@@ -0,0 +1,247 @@
|
||||
# ============================================================
|
||||
# download_vault_v1.1.py
|
||||
# Verze: 1.1
|
||||
# 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 (sekce Veeva Vault); env proměnné mají
|
||||
# přednost, při chybějících údajích se zeptá interaktivně.
|
||||
# Heslo se NIKDY nedává natvrdo do skriptu.
|
||||
# ============================================================
|
||||
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
from getpass import getpass
|
||||
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
|
||||
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
|
||||
|
||||
|
||||
# --- Přihlášení --------------------------------------------------------
|
||||
|
||||
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é); jinak interaktivně
|
||||
username = os.environ.get("VAULT_USER") or input("WWID/username: ")
|
||||
password = os.environ.get("VAULT_PASS") or getpass("Network Password: ")
|
||||
if not username or not password:
|
||||
raise RuntimeError("Chybí přihlašovací údaje.")
|
||||
|
||||
log("[i] Vyplňuji přihlašovací údaje...")
|
||||
user_box.fill(username)
|
||||
page.locator("input[type='password']").first.fill(password)
|
||||
page.get_by_role("button", name=re.compile("Sign On", re.I)).click()
|
||||
|
||||
# --- 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 dismiss_maintenance_popup(page):
|
||||
"""Zavře informační okno o údržbě, pokud je zobrazené."""
|
||||
for name in ("OK", "Close", "Continue"):
|
||||
btn = page.get_by_role("button", name=re.compile(f"^{name}$", re.I))
|
||||
try:
|
||||
if btn.count() and btn.first.is_visible():
|
||||
btn.first.click()
|
||||
log(f"[i] Zavřel jsem dialog tlačítkem '{name}'.")
|
||||
page.wait_for_timeout(500)
|
||||
except Exception:
|
||||
pass # dialog není / mezitím zmizel — pokračujeme
|
||||
|
||||
|
||||
# --- 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():
|
||||
load_env_file(ENV_FILE)
|
||||
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()
|
||||
Reference in New Issue
Block a user