#!/usr/bin/env python3 """ Open_PowerBi.py Purpose ------- Launch a *private* (incognito) Firefox window, navigate to Power BI, and perform Microsoft login if prompted. This script ALSO writes out a small session-info JSON file so other scripts can attach to the already-open browser session (without reopening/logging in again). Key fix (Jan 2026) ------------------ When this script is launched *from another process* (like Enter_PowerBi_Selector), stdin is often NOT an interactive TTY. In that case, calling input() raises EOFError immediately, which causes this script to exit and kills the geckodriver port before your attach scripts can connect. So: if stdin is not a TTY (or input() hits EOF), we keep the session alive by sleeping until we receive SIGTERM/SIGINT (the selector will terminate us when "Keep Power BI open" is unchecked). Session handoff file -------------------- By default, this script writes: config/powerbi_session.json to BOTH: 1) the folder containing this script 2) the current working directory (if different) You can override the path with: - env var POWERBI_SESSION_FILE=/full/path/to/powerbi_session.json The JSON contains: { "executor_url": "http://127.0.0.1:PORT", "session_id": "...", "download_dir": "/abs/path/to/downloads" } Dependencies ------------ - Python 3.9+ - Selenium 4.x - Firefox installed - geckodriver available on PATH (recommended) or set GECKODRIVER_PATH Usage ----- python Open_PowerBi.py python Open_PowerBi.py --headless """ from __future__ import annotations import json import os import sys import time import shutil from pathlib import Path from selenium import webdriver from selenium.webdriver.common.by import By from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC from selenium.webdriver.firefox.service import Service as FirefoxService from selenium.webdriver.firefox.options import Options as FirefoxOptions # PowerBI URL to the report. Do not modify unless the link changes. POWERBI_REPORT_URL = ( "https://app.powerbi.com/groups/me/apps/e65cf9b1-6444-4b63-8a86-b94e47839c83/rdlreports/c98f4a29-f532-4bf7-929d-4328ef57a6c2?experience=power-bi" ) # Desired filter settings (kept for your later automation; not used here yet) STORE_TYPE = "Dealership" STORE_NO = "614-WALDORF, MD" CATEGORY = "Footwear" # Hard-coded credentials (user provided). These are used when # environment variables POWERBI_USERNAME and POWERBI_PASSWORD are not set. HARDCODED_USERNAME = "jusstin@rwsext.net" HARDCODED_PASSWORD = "FluppyNupps49!" # Default download directory. # Override with env var POWERBI_DOWNLOAD_DIR=/full/path/to/folder DOWNLOAD_DIR = os.path.join(os.getcwd(), "downloads") # Session handoff file name (unless POWERBI_SESSION_FILE overrides) SESSION_FILENAME = "config/powerbi_session.json" # Wait time before declaring login/session setup failed. READY_TIMEOUT_SECONDS = int(os.environ.get("POWERBI_READY_TIMEOUT", "240")) # Throttle repeated MFA actions while we poll the login flow. _LAST_MFA_ACTION_TS = 0.0 MFA_ACTION_COOLDOWN_SECONDS = int(os.environ.get("POWERBI_MFA_ACTION_COOLDOWN_SECONDS", "30")) MFA_POST_TRIGGER_WAIT_SECONDS = int(os.environ.get("POWERBI_MFA_POST_TRIGGER_WAIT_SECONDS", "35")) def _ensure_selenium_tempdir() -> None: """Force Selenium temp files under $HOME (avoid /tmp issues on snap/locked-down Firefox). On some server installs (notably snap-packaged Firefox), the browser process cannot access host paths like /tmp created by Selenium for its temporary Firefox profile. That can cause geckodriver to hang during NEW_SESSION until Selenium times out. We avoid that by pointing TMPDIR/TMP/TEMP to a normal, user-writable folder. You can override the base folder with POWERBI_TMPDIR. """ preferred = os.environ.get("POWERBI_TMPDIR", "").strip() base = Path(preferred).expanduser() if preferred else (Path.home() / "selenium-tmp") base = base.resolve() base.mkdir(parents=True, exist_ok=True) for key in ("TMPDIR", "TMP", "TEMP"): cur = os.environ.get(key, "").strip() # Override empty or /tmp-ish temp dirs (common cause of snap Firefox profile access issues) if (not cur) or cur in ("/tmp", "/var/tmp") or cur.startswith("/tmp/") or cur.startswith("/var/tmp/"): os.environ[key] = str(base) def _ensure_xdg_runtime_dir() -> str: """Provide a private runtime dir when cron does not export one.""" current = os.environ.get("XDG_RUNTIME_DIR", "").strip() candidates = [] if current: candidates.append(Path(current).expanduser()) override = os.environ.get("POWERBI_XDG_RUNTIME_DIR", "").strip() if override: candidates.insert(0, Path(override).expanduser()) candidates.extend([ Path.home() / ".local" / "run" / "powerbi", Path.home() / ".cache" / "powerbi-xdg-runtime", ]) for candidate in candidates: try: candidate.mkdir(parents=True, exist_ok=True) os.chmod(candidate, 0o700) if candidate.is_dir() and os.access(candidate, os.W_OK | os.X_OK): os.environ["XDG_RUNTIME_DIR"] = str(candidate) return str(candidate) except Exception: continue return current def ensure_download_directory(path: str) -> None: """Ensure the download directory exists.""" os.makedirs(path, exist_ok=True) def _resolve_firefox_binary() -> str | None: """Resolve Firefox even when cron provides a reduced PATH.""" candidates: list[str] = [] env_path = os.environ.get("FIREFOX_BINARY", "").strip() if env_path: candidates.append(env_path) for fixed in ( "/snap/firefox/current/usr/lib/firefox/firefox", "/snap/bin/firefox", "/usr/bin/firefox", "/usr/bin/firefox-esr", "/usr/local/bin/firefox", "/usr/local/bin/firefox-esr", "/usr/lib/firefox/firefox", "/usr/lib64/firefox/firefox", ): candidates.append(fixed) for name in ("firefox", "firefox-esr"): which_path = shutil.which(name) if which_path: candidates.append(which_path) seen: set[str] = set() for candidate in candidates: if not candidate or candidate in seen: continue seen.add(candidate) if os.path.isfile(candidate) and os.access(candidate, os.X_OK): return os.path.abspath(candidate) return None def _resolve_geckodriver() -> str | None: """ Resolve geckodriver path cross-platform. Priority: 1) GECKODRIVER_PATH env var 2) PATH (shutil.which) Returns an absolute path string, or None to let Selenium try defaults. """ env_path = os.environ.get("GECKODRIVER_PATH", "").strip() if env_path: # Allow passing a folder or the full executable path. if os.path.isdir(env_path): cand = os.path.join( env_path, "geckodriver.exe" if os.name == "nt" else "geckodriver", ) if os.path.isfile(cand): return os.path.abspath(cand) if os.path.isfile(env_path): return os.path.abspath(env_path) candidates: list[str] = [] for name in ("geckodriver", "geckodriver.exe"): which_path = shutil.which(name) if which_path: candidates.append(which_path) candidates.extend([ str(Path.home() / "bin" / "geckodriver"), "/snap/bin/geckodriver", "/usr/bin/geckodriver", "/usr/local/bin/geckodriver", ]) seen: set[str] = set() for candidate in candidates: if not candidate or candidate in seen: continue seen.add(candidate) if os.path.isfile(candidate) and os.access(candidate, os.X_OK): return os.path.abspath(candidate) return None def _get_executor_url(driver: webdriver.Firefox) -> str: """ Determine the running WebDriver command executor URL. """ ce = getattr(driver, "command_executor", None) if ce is not None: url = getattr(ce, "_url", None) if url: return str(url) url = getattr(ce, "url", None) if url: return str(url) service = getattr(driver, "service", None) if service is not None: try: url = getattr(service, "service_url", None) if url: return str(url) except Exception: pass if ce is not None: url = getattr(ce, "address", None) if url: return str(url) url = getattr(ce, "remote_server_addr", None) if url: return str(url) raise RuntimeError( "Could not determine executor URL from WebDriver. Please ensure that " "your version of Selenium exposes either 'command_executor._url', 'command_executor.url' " "or that the driver service provides 'service_url'." ) def write_session_file(driver: webdriver.Firefox, download_dir: str) -> list[Path]: """ Write session info to a JSON file so other scripts can attach. Returns a list of paths written. """ executor_url = _get_executor_url(driver) session_id = driver.session_id payload = { "executor_url": executor_url, "session_id": session_id, "download_dir": str(Path(download_dir).expanduser().resolve()), } override = os.environ.get("POWERBI_SESSION_FILE", "").strip() written: list[Path] = [] if override: p = Path(override).expanduser().resolve() p.parent.mkdir(parents=True, exist_ok=True) p.write_text(json.dumps(payload, indent=2), encoding="utf-8") written.append(p) return written script_dir = Path(__file__).resolve().parent cwd_dir = Path.cwd().resolve() for base in {script_dir, cwd_dir}: p = base / SESSION_FILENAME try: p.parent.mkdir(parents=True, exist_ok=True) p.write_text(json.dumps(payload, indent=2), encoding="utf-8") written.append(p) except Exception: continue if not written: raise RuntimeError("Failed to write session file anywhere (permissions/path issue).") return written def login_if_required(driver: webdriver.Firefox) -> None: """ Perform a two-step login into the Microsoft account used by Power BI. """ username = os.environ.get("POWERBI_USERNAME", HARDCODED_USERNAME) password = os.environ.get("POWERBI_PASSWORD", HARDCODED_PASSWORD) if not (username and password): return wait = WebDriverWait(driver, 6) # ---- Username/email step (if present) ---- email_input = None for locator in ( (By.ID, "email"), (By.ID, "i0116"), (By.NAME, "loginfmt"), (By.XPATH, "//input[@type='email' or contains(@placeholder, 'Email') or contains(@placeholder, 'email')]"), ): try: email_input = wait.until(EC.presence_of_element_located(locator)) if email_input is not None: break except Exception: continue if email_input is not None: try: email_input.clear() except Exception: pass try: email_input.send_keys(username) print("Entered Microsoft username.") except Exception: pass for locator in ( (By.ID, "submitBtn"), (By.ID, "idSIButton9"), (By.CSS_SELECTOR, "input[type='submit']"), (By.CSS_SELECTOR, "button[type='submit']"), ): try: btn = driver.find_element(*locator) btn.click() print("Submitted username step.") break except Exception: continue # ---- Password step (if present) ---- password_input = None for locator in ( (By.ID, "i0118"), (By.NAME, "passwd"), (By.XPATH, "//input[@type='password']"), ): try: password_input = wait.until(EC.presence_of_element_located(locator)) if password_input is not None: break except Exception: continue if password_input is not None: try: password_input.clear() except Exception: pass try: password_input.send_keys(password) print("Entered Microsoft password.") except Exception: pass for locator in ( (By.ID, "idSIButton9"), (By.CSS_SELECTOR, "input[type='submit']"), (By.CSS_SELECTOR, "button[type='submit']"), ): try: btn = driver.find_element(*locator) btn.click() print("Submitted password step.") break except Exception: continue # ---- "Stay signed in?" prompt ---- try: stay_btn = driver.find_element(By.ID, "idSIButton9") stay_btn.click() print("Confirmed stay signed in prompt.") except Exception: pass # ---- 2FA: "Call me" option tile / retry ---- global _LAST_MFA_ACTION_TS now = time.time() cooldown = max(8, MFA_ACTION_COOLDOWN_SECONDS) if now - _LAST_MFA_ACTION_TS < cooldown: return mfa_error_text = "" for locator in ( (By.ID, "idSpan_SAOTCC_Error_OTC"), (By.CSS_SELECTOR, "#idDiv_SAOTCC_ErrorMsg_OTC"), (By.XPATH, "//*[contains(@class,'alert-error') or contains(@id,'ErrorMsg')]"), ): try: txt = (driver.find_element(*locator).text or "").strip() if txt: mfa_error_text = txt break except Exception: continue if mfa_error_text: print(f"MFA notice: {mfa_error_text}") retry_clicked = False if mfa_error_text: for locator in ( (By.XPATH, "//input[@type='button' and contains(translate(@value,'ABCDEFGHIJKLMNOPQRSTUVWXYZ','abcdefghijklmnopqrstuvwxyz'),'try again')]"), (By.XPATH, "//button[contains(translate(normalize-space(.),'ABCDEFGHIJKLMNOPQRSTUVWXYZ','abcdefghijklmnopqrstuvwxyz'),'try again')]"), (By.XPATH, "//a[contains(translate(normalize-space(.),'ABCDEFGHIJKLMNOPQRSTUVWXYZ','abcdefghijklmnopqrstuvwxyz'),'try again')]"), ): try: btn = WebDriverWait(driver, 2).until(EC.element_to_be_clickable(locator)) btn.click() retry_clicked = True print("Clicked MFA 'Try again'.") time.sleep(1.0) break except Exception: continue call_clicked = False for locator in ( (By.XPATH, "//div[@data-value='TwoWayVoiceOffice' and contains(., 'Call')]"), (By.XPATH, "//*[self::div or self::button][contains(translate(normalize-space(.), 'ABCDEFGHIJKLMNOPQRSTUVWXYZ', 'abcdefghijklmnopqrstuvwxyz'), 'call me')]"), (By.XPATH, "//*[self::div or self::button][contains(translate(normalize-space(.), 'ABCDEFGHIJKLMNOPQRSTUVWXYZ', 'abcdefghijklmnopqrstuvwxyz'), 'call')]"), ): try: voice_call_tile = WebDriverWait(driver, 2).until(EC.element_to_be_clickable(locator)) voice_call_tile.click() call_clicked = True print("Clicked the 'Call me' option.") break except Exception: continue verify_clicked = False if call_clicked or retry_clicked: for locator in ( (By.ID, "idSubmit_SAOTCC_Continue"), (By.XPATH, "//input[@type='submit' and contains(translate(@value,'ABCDEFGHIJKLMNOPQRSTUVWXYZ','abcdefghijklmnopqrstuvwxyz'),'verify')]"), (By.XPATH, "//button[contains(translate(normalize-space(.),'ABCDEFGHIJKLMNOPQRSTUVWXYZ','abcdefghijklmnopqrstuvwxyz'),'verify')]"), ): try: btn = WebDriverWait(driver, 2).until(EC.element_to_be_clickable(locator)) btn.click() verify_clicked = True print("Triggered MFA verification call.") break except Exception: continue if call_clicked or retry_clicked or verify_clicked: _LAST_MFA_ACTION_TS = now time.sleep(max(10, MFA_POST_TRIGGER_WAIT_SECONDS)) def login_microsoft_if_prompted(driver: webdriver.Firefox) -> None: login_if_required(driver) def _safe_current_url(driver: webdriver.Firefox) -> str: try: return (driver.current_url or "").strip() except Exception: return "" def _safe_title(driver: webdriver.Firefox) -> str: try: return (driver.title or "").strip() except Exception: return "" def _is_auth_or_sso_page(url: str, title: str) -> bool: u = url.lower() t = title.lower() return ( "login.microsoftonline.com" in u or "/singlesignon" in u or "/signin" in u or "sign in" in t ) def _extract_login_error_text(driver: webdriver.Firefox) -> str: """Best-effort capture of visible Microsoft login error/challenge text.""" candidates: list[str] = [] for locator in ( (By.ID, "usernameError"), (By.ID, "passwordError"), (By.ID, "i0116Error"), (By.ID, "i0118Error"), (By.CSS_SELECTOR, "[role='alert']"), (By.CSS_SELECTOR, ".error, .error-msg, .error-message"), (By.XPATH, "//*[contains(@class,'error') or contains(@id,'error')]"), ): try: for el in driver.find_elements(*locator): txt = (el.text or "").strip() if txt: candidates.append(txt) except Exception: continue seen = set() out: list[str] = [] for txt in candidates: if txt not in seen: seen.add(txt) out.append(txt) return " | ".join(out[:4]) def _dump_login_debug(driver: webdriver.Firefox) -> None: out_dir = Path.cwd().resolve() / "powerbi_login_debug" out_dir.mkdir(parents=True, exist_ok=True) stamp = str(int(time.time())) png = out_dir / f"login_timeout_{stamp}.png" html = out_dir / f"login_timeout_{stamp}.html" try: driver.save_screenshot(str(png)) print(f"Saved login debug screenshot: {png}") except Exception: pass try: html.write_text(driver.page_source or "", encoding="utf-8") print(f"Saved login debug HTML: {html}") except Exception: pass def _is_report_ready(driver: webdriver.Firefox) -> bool: """True when authenticated and on a report-capable Power BI page.""" url = _safe_current_url(driver) title = _safe_title(driver) if not url: return False if _is_auth_or_sso_page(url, title): return False if "app.powerbi.com" not in url.lower(): return False # Strong positive signal: reached report route. low = url.lower() if "/rdlreports/" in low and "/groups/" in low: return True # Fallback: report iframe exists in current document. for sel in ( "iframe[src*='paginated-reports.powerbi.com']", "iframe[src*='/rdlreports/']", "iframe[src*='rdlembed']", ): try: if driver.find_elements(By.CSS_SELECTOR, sel): return True except Exception: continue return False def wait_for_authenticated_report(driver: webdriver.Firefox, timeout: int = READY_TIMEOUT_SECONDS) -> None: """Wait until login is truly complete and report context is available.""" deadline = time.time() + timeout last_log = 0.0 last_nav = 0.0 while time.time() < deadline: login_microsoft_if_prompted(driver) if _is_report_ready(driver): return now = time.time() if now - last_log >= 8: login_err = "" if _is_auth_or_sso_page(_safe_current_url(driver), _safe_title(driver)): login_err = _extract_login_error_text(driver) print( "Waiting for authenticated report page...", f"url={_safe_current_url(driver)!r}", f"title={_safe_title(driver)!r}", (f"login_error={login_err!r}" if login_err else ""), ) last_log = now # Re-navigate to the report URL when we're on generic Power BI shell pages. cur = _safe_current_url(driver).lower() if ( now - last_nav >= 15 and "app.powerbi.com" in cur and not _is_auth_or_sso_page(cur, _safe_title(driver)) and "/rdlreports/" not in cur ): try: driver.get(POWERBI_REPORT_URL) except Exception: pass last_nav = now time.sleep(1.0) _dump_login_debug(driver) raise TimeoutError( "Timed out waiting for authenticated Power BI report page. " f"Last URL: {_safe_current_url(driver)!r}; title: {_safe_title(driver)!r}" ) def build_driver(headless: bool = False) -> tuple[webdriver.Firefox, str]: _ensure_selenium_tempdir() runtime_dir = _ensure_xdg_runtime_dir() os.environ.setdefault("NO_AT_BRIDGE", "1") if headless: os.environ.setdefault("MOZ_HEADLESS", "1") opts = FirefoxOptions() firefox_binary = _resolve_firefox_binary() if firefox_binary: opts.binary_location = firefox_binary # Private browsing opts.set_preference("browser.privatebrowsing.autostart", True) # ---- Downloads (quiet + auto-accept) ---- download_dir = os.environ.get("POWERBI_DOWNLOAD_DIR", DOWNLOAD_DIR) download_dir = str(Path(download_dir).expanduser().resolve()) ensure_download_directory(download_dir) opts.set_preference("browser.download.folderList", 2) opts.set_preference("browser.download.dir", download_dir) opts.set_preference("browser.download.useDownloadDir", True) opts.set_preference("browser.download.manager.showWhenStarting", False) opts.set_preference("browser.download.manager.focusWhenStarting", False) opts.set_preference("browser.download.alwaysOpenPanel", False) opts.set_preference("browser.download.manager.alertOnEXEOpen", False) never_ask = ",".join([ "application/octet-stream", "binary/octet-stream", "application/download", "application/x-download", "application/pdf", "text/csv", "application/csv", "application/vnd.ms-excel", "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", "application/zip", "application/x-zip-compressed", "text/plain", ]) opts.set_preference("browser.helperApps.neverAsk.saveToDisk", never_ask) opts.set_preference("browser.helperApps.neverAsk.openFile", never_ask) opts.set_preference("browser.helperApps.alwaysAsk.force", False) opts.set_preference("pdfjs.disabled", True) if headless: opts.add_argument("-headless") geckodriver = _resolve_geckodriver() print(f"Firefox binary: {firefox_binary or 'default lookup'}") print(f"Geckodriver: {geckodriver or 'default lookup'}") print(f"TMPDIR: {os.environ.get('TMPDIR', '')}") print(f"XDG_RUNTIME_DIR: {runtime_dir or os.environ.get('XDG_RUNTIME_DIR', '')}") service = FirefoxService(executable_path=geckodriver) if geckodriver else FirefoxService() return webdriver.Firefox(service=service, options=opts), download_dir def open_powerbi_session(headless: bool = False) -> tuple[webdriver.Firefox, str]: driver, download_dir = build_driver(headless=headless) try: driver.maximize_window() except Exception: pass driver.get(POWERBI_REPORT_URL) login_microsoft_if_prompted(driver) wait_for_authenticated_report(driver, timeout=READY_TIMEOUT_SECONDS) try: print("Current URL after login attempt:", driver.current_url) except Exception: pass return driver, download_dir def _keep_alive_loop() -> None: # Sleep forever (until SIGTERM/SIGINT). This keeps the webdriver service alive. while True: time.sleep(3600) def keep_open_forever() -> None: headless = any(a in sys.argv for a in ("-headless", "--headless")) driver, download_dir = open_powerbi_session(headless=headless) try: written_paths = write_session_file(driver, download_dir) print("\nWrote session file(s):") for p in written_paths: print(f" - {p}") print("Other scripts can now attach using this file.\n") except Exception as e: print(f"\nWARNING: Could not write session file: {e}\n") print("Power BI window is open.") print("Leave this terminal running to keep the session alive.") print("Press ENTER to close the browser (TTY only), or Ctrl+C to stop.\n") try: # If we're attached to a real terminal, allow ENTER-to-close. if sys.stdin is not None and hasattr(sys.stdin, "isatty") and sys.stdin.isatty(): input() else: _keep_alive_loop() except (KeyboardInterrupt, EOFError): # EOFError happens when stdin is not interactive (common when launched from GUI/selector). pass finally: try: driver.quit() except Exception: pass if __name__ == "__main__": keep_open_forever()