#!/usr/bin/env python3 """ Pull_On_Hand_Footwear.py (v14) Attaches to an already-open Firefox session created by Open_PowerBi_patched.py (via powerbi_session.json), loads the Inventory Detail paginated report, switches into the paginated-reports iframe, sets: Store Type = Dealership As Of = Current Date Category = Footwear (multi-select: turns off Any, turns on Footwear) Store No = 614-WALDORF, MD Bucket = On hand Then: - Clicks View report (if needed) - Clicks Export -> Comma Separated Values (.csv) - Waits for the download to complete (monitors download folder) - Moves/renames it into the CURRENT WORKING DIRECTORY as: yyyymmdd_On_Hand_Footwear.csv Env (optional) -------------- POWERBI_SESSION_FILE=/full/path/to/powerbi_session.json POWERBI_DOWNLOAD_DIR=/full/path/to/downloads Debug (optional) -------------- If the script cannot find/click the Export (or Download) button, set: POWERBI_DEBUG_EXPORT=1 It will save: export_debug/export_debug_default.png export_debug/export_debug_default.json export_debug/export_debug_frame.png export_debug/export_debug_frame.json Send me those JSON contents (or paste the console output) and I can lock the selector. Requires -------- pip install selenium """ from __future__ import annotations import json import os import time import traceback from datetime import datetime from pathlib import Path from typing import Iterable, Optional, Tuple from urllib.parse import urlparse from selenium import webdriver from selenium.webdriver.common.by import By from selenium.webdriver.common.keys import Keys from selenium.webdriver.common.action_chains import ActionChains from selenium.webdriver.remote.webdriver import WebDriver as RemoteWebDriver from selenium.webdriver.firefox.options import Options as FirefoxOptions from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC # ---- Targets ---- TARGET_REPORT_URL = ( "https://app.powerbi.com/groups/me/apps/e65cf9b1-6444-4b63-8a86-b94e47839c83/rdlreports/c98f4a29-f532-4bf7-929d-4328ef57a6c2?experience=power-bi" ) STORE_TYPE_VALUE = "Dealership" AS_OF_VALUE = os.environ.get("POWERBI_AS_OF_VALUE", "Current Date").strip() or "Current Date" STORE_NO_VALUE = "614-WALDORF, MD" BUCKET_VALUE = "On hand" # case-insensitive match CATEGORY_ON = "Footwear" CATEGORY_OFF = ["Select All", "Any", "Footwear Accessories"] SESSION_FILENAME = "config/powerbi_session.json" IFRAME_SRC_HINT = "paginated-reports.powerbi.com" IFRAME_SRC_HINTS = ( "paginated-reports.powerbi.com", "/rdlreports/", "rdlembed", "rdl", ) PAGE_TIMEOUT = 120 IFRAME_TIMEOUT = 30 FIELD_TIMEOUT = 25 EXPORT_TIMEOUT = 60 DOWNLOAD_TIMEOUT = 240 def _v(msg: str) -> None: """Verbose logging controlled by POWERBI_VERBOSE=1.""" if os.environ.get("POWERBI_VERBOSE", "").strip().lower() in ("1", "true", "yes", "y", "on"): print(msg) # --------------------------- # Session file discovery # --------------------------- def _walk_parents(start: Path) -> Iterable[Path]: p = start.resolve() yield p for parent in p.parents: yield parent def _candidate_session_paths() -> list[Path]: candidates: list[Path] = [] override = os.environ.get("POWERBI_SESSION_FILE", "").strip() if override: candidates.append(Path(override).expanduser().resolve()) candidates.append(Path.cwd().resolve() / SESSION_FILENAME) candidates.append(Path(__file__).resolve().parent / SESSION_FILENAME) candidates.append(Path.cwd().resolve() / "Powerbi_Background" / SESSION_FILENAME) candidates.append(Path(__file__).resolve().parent / "Powerbi_Background" / SESSION_FILENAME) for base in _walk_parents(Path.cwd()): candidates.append(base / SESSION_FILENAME) candidates.append(base / "Powerbi_Background" / SESSION_FILENAME) for base in _walk_parents(Path(__file__).resolve().parent): candidates.append(base / SESSION_FILENAME) candidates.append(base / "Powerbi_Background" / SESSION_FILENAME) seen = set() out: list[Path] = [] for c in candidates: s = str(c) if s not in seen: seen.add(s) out.append(c) return out def _pick_session_file() -> Path: for p in _candidate_session_paths(): if p.exists(): return p return Path.cwd().resolve() / SESSION_FILENAME def _load_session_file(path: Path) -> Tuple[str, str, Optional[str]]: if not path.exists(): searched = _candidate_session_paths() msg = ["Session file not found.", "Searched:"] msg += [f" - {p}" for p in searched[:20]] if len(searched) > 20: msg.append(f" ... and {len(searched) - 20} more") msg += [ "", "Fix:", " - Run from folder containing powerbi_session.json, OR", " - export POWERBI_SESSION_FILE=/full/path/to/powerbi_session.json", ] raise FileNotFoundError("\n".join(msg)) data = json.loads(path.read_text(encoding="utf-8")) executor_url = data.get("executor_url") session_id = data.get("session_id") download_dir = data.get("download_dir") if not executor_url or not session_id: raise ValueError(f"Bad session file {path} keys={list(data.keys())}") return executor_url, session_id, download_dir # --------------------------- # Attach to existing session # --------------------------- def attach_to_session(executor_url: str, session_id: str) -> RemoteWebDriver: original_execute = RemoteWebDriver.execute def _patched_execute(self, command, params=None): if command == "newSession": return {"value": {"capabilities": {}}, "sessionId": session_id} return original_execute(self, command, params) RemoteWebDriver.execute = _patched_execute try: options = FirefoxOptions() try: driver = webdriver.Remote(command_executor=executor_url, options=options) except TypeError: try: driver = webdriver.Remote(command_executor=executor_url) except TypeError: driver = webdriver.Remote(command_executor=executor_url, desired_capabilities={}) driver.session_id = session_id finally: RemoteWebDriver.execute = original_execute _ = driver.current_url return driver # --------------------------- # Navigation / iframe # --------------------------- def _url_same_page(a: str, b: str) -> bool: try: pa = urlparse(a) pb = urlparse(b) return (pa.scheme, pa.netloc, pa.path) == (pb.scheme, pb.netloc, pb.path) except Exception: return False def _find_report_frame_in_current_context(driver: RemoteWebDriver): selectors = [f"iframe[src*='{h}']" for h in IFRAME_SRC_HINTS] selectors.append("iframe[name*='rdl' i]") selectors.append("iframe[id*='rdl' i]") for sel in selectors: try: frames = driver.find_elements(By.CSS_SELECTOR, sel) if frames: return frames[0] except Exception: continue return None def _switch_to_report_iframe_anywhere(driver: RemoteWebDriver): """Find report iframe from default content or one nested level down.""" try: driver.switch_to.default_content() except Exception: pass frame = _find_report_frame_in_current_context(driver) if frame is not None: driver.switch_to.frame(frame) return True # One nested level for shells that host an outer iframe first. try: outer_frames = driver.find_elements(By.CSS_SELECTOR, "iframe") except Exception: outer_frames = [] for outer in outer_frames[:20]: try: driver.switch_to.default_content() driver.switch_to.frame(outer) inner = _find_report_frame_in_current_context(driver) if inner is not None: driver.switch_to.frame(inner) return True except Exception: continue try: driver.switch_to.default_content() except Exception: pass return False def ensure_report_loaded(driver: RemoteWebDriver, timeout: int = PAGE_TIMEOUT) -> None: desired = TARGET_REPORT_URL found = False try: handles = driver.window_handles except Exception: handles = [] for h in handles: try: driver.switch_to.window(h) cur = driver.current_url or "" if _url_same_page(cur, desired): found = True break except Exception: continue if not found: driver.get(desired) deadline = time.time() + timeout last_url = "" while time.time() < deadline: try: last_url = driver.current_url or "" except Exception: last_url = "" # If auth bounced us to M365/AAD, fail fast with a clearer message. if any( h in last_url for h in ( "login.microsoftonline.com", "aadcdn", "microsoftonline", ) ): raise RuntimeError( f"Power BI session is not authenticated. Landed on: {last_url}. " "Re-run Open_PowerBi and complete login/MFA." ) try: ready = driver.execute_script("return document.readyState") except Exception: ready = "" if ready in ("interactive", "complete"): if _switch_to_report_iframe_anywhere(driver): try: driver.switch_to.default_content() except Exception: pass return time.sleep(0.35) title = "" try: title = driver.title except Exception: pass raise TimeoutError( "Timed out waiting for paginated report iframe to load. " f"Last URL: {last_url!r}; title: {title!r}" ) def switch_to_paginated_report_iframe(driver: RemoteWebDriver, timeout: int = IFRAME_TIMEOUT) -> None: wait = WebDriverWait(driver, timeout) ok = wait.until(lambda d: _switch_to_report_iframe_anywhere(d)) if not ok: raise TimeoutError("Could not switch to paginated report iframe.") wait.until(lambda d: len(d.find_elements(By.XPATH, "//input[@role='combobox']")) > 0) # --------------------------- # UI helpers # --------------------------- def _scroll_into_view(driver: RemoteWebDriver, element) -> None: try: driver.execute_script("arguments[0].scrollIntoView({block:'center', inline:'nearest'});", element) except Exception: pass def _safe_click(driver: RemoteWebDriver, element) -> None: _scroll_into_view(driver, element) try: element.click() return except Exception: pass try: ActionChains(driver).move_to_element(element).pause(0.05).click().perform() return except Exception: pass try: driver.execute_script("arguments[0].click();", element) except Exception: pass def _wait_enabled(element, timeout: int = FIELD_TIMEOUT) -> None: end = time.time() + timeout while time.time() < end: try: dis = element.get_attribute("disabled") aria_dis = element.get_attribute("aria-disabled") if (dis is None or dis == "") and (aria_dis is None or aria_dis.lower() == "false"): return except Exception: pass time.sleep(0.15) raise TimeoutError("Field stayed disabled too long.") def _visible_value(inp) -> str: try: return (inp.get_attribute("value") or "").strip() except Exception: return "" def _find_combobox_input(driver: RemoteWebDriver, label: str): id_map = { "Store Type": "StoreType-input", "As Of": "AsOf-input", "Category": "Category-input", "Store No": "StoreNo-input", "Bucket": "Bucket-input", } if label in id_map: els = driver.find_elements(By.ID, id_map[label]) if els: return els[0] xp = f"//input[@role='combobox' and @aria-label={json.dumps(label)}]" els = driver.find_elements(By.XPATH, xp) if els: return els[0] xp2 = f"//input[@role='combobox' and contains(@aria-label, {json.dumps(label)})]" els = driver.find_elements(By.XPATH, xp2) if els: return els[0] raise RuntimeError(f"Could not find combobox input for label: {label}") def _find_category_listbox(driver: RemoteWebDriver): boxes = driver.find_elements(By.XPATH, "//div[@role='listbox' and @aria-labelledby='Category-label']") if boxes: return boxes[0] boxes = driver.find_elements(By.XPATH, "//div[@role='listbox' and .//*[@id[starts-with(.,'Category-list')]]]") if boxes: return boxes[0] return None def _open_dropdown(driver: RemoteWebDriver, label: str, inp, tries: int = 8) -> None: for _ in range(tries): try: exp = driver.find_element(By.CSS_SELECTOR, f"[aria-label='Open {label}']") _safe_click(driver, exp) except Exception: _safe_click(driver, inp) time.sleep(0.20) try: inp.send_keys(Keys.ARROW_DOWN) except Exception: pass time.sleep(0.15) if driver.find_elements(By.XPATH, "//*[@role='option']"): return if label == "Category" and _find_category_listbox(driver) is not None: return def _find_option_clickable(driver: RemoteWebDriver, value: str): low = value.lower() opts = driver.find_elements(By.XPATH, "//*[@role='option']") for opt in opts: try: if not opt.is_displayed(): continue except Exception: continue t = (opt.text or "").strip() or (opt.get_attribute("title") or "").strip() if t.lower() == low: return opt for opt in opts: t = (opt.text or "").strip() or (opt.get_attribute("title") or "").strip() if low in t.lower(): return opt return None def set_single_combobox(driver: RemoteWebDriver, label: str, value: str) -> None: inp = _find_combobox_input(driver, label) _wait_enabled(inp) _scroll_into_view(driver, inp) if _visible_value(inp).lower() == value.lower(): print(f"{label}: set to '{value}'.") return _open_dropdown(driver, label, inp) _safe_click(driver, inp) try: inp.send_keys(Keys.CONTROL, "a") except Exception: pass inp.send_keys(value) time.sleep(0.40) opt = _find_option_clickable(driver, value) if opt is not None: _safe_click(driver, opt) else: try: inp.send_keys(Keys.ARROW_DOWN) inp.send_keys(Keys.ENTER) except Exception: pass try: inp.send_keys(Keys.TAB) except Exception: pass end = time.time() + FIELD_TIMEOUT last = "" while time.time() < end: last = _visible_value(inp) if last.lower() == value.lower(): print(f"{label}: set to '{value}'.") return if value.lower() in last.lower(): print(f"{label}: set to '{last}'.") return time.sleep(0.2) raise TimeoutError(f"{label}: did not become '{value}'. Last value: {last!r}") def wait_for_filter_panel_ready( driver: RemoteWebDriver, labels: tuple[str, ...] = ("Category", "Store No", "Bucket"), timeout: int = FIELD_TIMEOUT, ) -> None: """Wait for filter controls to be rebuilt after a report parameter changes.""" end = time.time() + timeout last_error: Optional[str] = None while time.time() < end: try: switch_to_paginated_report_iframe(driver, timeout=5) for label in labels: inp = _find_combobox_input(driver, label) _wait_enabled(inp, timeout=5) return except Exception as e: last_error = f"{type(e).__name__}: {e}" time.sleep(0.4) raise TimeoutError(f"Filter panel did not become ready. Last error: {last_error}") # --------------------------- # Category checkbox multi-select # --------------------------- def _listbox_items(listbox) -> list: try: return listbox.find_elements(By.XPATH, ".//*[@role='menuitemcheckbox']") except Exception: return [] def _find_menuitemcheckbox(listbox, label_text: str): target = label_text.strip().lower() for it in _listbox_items(listbox): try: txt = (it.text or "").strip().lower() except Exception: txt = "" if txt == target: return it return None def _aria_checked(el) -> Optional[bool]: try: v = (el.get_attribute("aria-checked") or "").lower() if v in ("true", "false"): return v == "true" except Exception: pass return None def _set_menuitemcheckbox(driver: RemoteWebDriver, listbox, label_text: str, desired: bool, allow_missing: bool = False) -> None: end = time.time() + FIELD_TIMEOUT last_items: list[str] = [] while time.time() < end: item = _find_menuitemcheckbox(listbox, label_text) if item is None: lb2 = _find_category_listbox(driver) if lb2 is not None: listbox = lb2 last_items = [(it.text or "").strip() for it in _listbox_items(listbox)] if allow_missing: return time.sleep(0.2) continue cur = _aria_checked(item) if cur == desired: return _safe_click(driver, item) time.sleep(0.25) lb2 = _find_category_listbox(driver) if lb2 is not None: listbox = lb2 item2 = _find_menuitemcheckbox(listbox, label_text) if item2 is None: continue cur2 = _aria_checked(item2) if cur2 == desired: return if allow_missing: return preview = [s for s in last_items if s][:30] raise TimeoutError(f"Could not set '{label_text}' to {desired}. Items seen: {preview}") def set_category_to_footwear_only(driver: RemoteWebDriver) -> None: label = "Category" inp = _find_combobox_input(driver, label) _wait_enabled(inp) _scroll_into_view(driver, inp) _open_dropdown(driver, label, inp) wait = WebDriverWait(driver, FIELD_TIMEOUT) listbox = wait.until(lambda d: _find_category_listbox(d)) if listbox is None: raise TimeoutError("Category listbox did not appear.") for off in CATEGORY_OFF: _set_menuitemcheckbox(driver, listbox, off, False, allow_missing=True) _set_menuitemcheckbox(driver, listbox, CATEGORY_ON, True, allow_missing=False) try: inp.send_keys(Keys.ESCAPE) except Exception: pass try: inp.send_keys(Keys.TAB) except Exception: pass end = time.time() + FIELD_TIMEOUT last = "" while time.time() < end: last = _visible_value(inp) low = last.lower() if ("footwear" in low) and ("any" not in low) and ("access" not in low): print(f"Category: set to '{last}'.") return time.sleep(0.2) raise TimeoutError(f"Category: did not settle to Footwear-only. Final field text: {last!r}") # --------------------------- # View report & Export # --------------------------- def click_view_report_if_present(driver: RemoteWebDriver) -> None: # Sometimes it's already in "view report" mode; this is safe to try. wait = WebDriverWait(driver, 5) for loc in ( (By.XPATH, "//button[normalize-space()='View report']"), (By.XPATH, "//button[.//span[normalize-space()='View report']]"), (By.CSS_SELECTOR, "button[title='View report']"), (By.CSS_SELECTOR, "button[aria-label='View report']"), ): try: btn = wait.until(EC.element_to_be_clickable(loc)) _safe_click(driver, btn) time.sleep(0.5) return except Exception: continue def _element_interactable(el) -> bool: """Best-effort check that an element is *the* visible, usable instance. Power BI often renders duplicate command bar buttons (hidden/overflow variants). If we click a hidden clone, nothing happens (even with JS click). """ try: if not el.is_displayed(): return False except Exception: return False try: dis = el.get_attribute("disabled") if dis is not None and dis != "": return False except Exception: pass try: aria_dis = (el.get_attribute("aria-disabled") or "").strip().lower() if aria_dis == "true": return False except Exception: pass try: r = el.rect or {} if (r.get("width") or 0) <= 0 or (r.get("height") or 0) <= 0: return False except Exception: # rect can fail for stale elements; treat as not interactable return False return True def _first_interactable(driver: RemoteWebDriver, loc): by, sel = loc try: els = driver.find_elements(by, sel) except Exception: return None for el in els: if _element_interactable(el): return el return None def wait_for_export_button(driver: RemoteWebDriver, timeout: int = EXPORT_TIMEOUT): wait = WebDriverWait(driver, timeout) # Prefer interactable elements over Selenium's "clickable" heuristic. # Hidden/duplicate command bar buttons are a common failure mode. locators = ( (By.CSS_SELECTOR, "button[data-testid='toolbar-export-dropdown'], [data-testid='toolbar-export-dropdown']"), (By.CSS_SELECTOR, "button[data-testid='trident-export-menu-button'], [data-testid='trident-export-menu-button']"), (By.CSS_SELECTOR, "button[data-testid='toolbar-download-dropdown'], [data-testid='toolbar-download-dropdown']"), (By.CSS_SELECTOR, "button[data-testid='trident-download-menu-button'], [data-testid='trident-download-menu-button']"), (By.XPATH, "//*[@data-testid='toolbar-export-dropdown' or @data-testid='trident-export-menu-button' or @data-testid='toolbar-download-dropdown' or @data-testid='trident-download-menu-button']"), (By.XPATH, "//*[self::button or @role='button'][@title='Export' or @title='Download']"), (By.XPATH, "//*[self::button or @role='button'][contains(translate(@aria-label,'ABCDEFGHIJKLMNOPQRSTUVWXYZ','abcdefghijklmnopqrstuvwxyz'),'export') or contains(translate(@aria-label,'ABCDEFGHIJKLMNOPQRSTUVWXYZ','abcdefghijklmnopqrstuvwxyz'),'download')]"), (By.XPATH, "//*[self::button or @role='button'][normalize-space()='Export' or normalize-space()='Download' or .//span[normalize-space()='Export' or normalize-space()='Download']]"), ) for loc in locators: try: el = wait.until(lambda d: _first_interactable(d, loc)) if el is not None: _v(f"[export] picked interactable export/download via {loc}") return el except Exception: continue # Last resort: maybe it's present but Selenium can't decide visibility yet. for loc in locators: try: el = wait.until(EC.presence_of_element_located(loc)) if el is not None: _v(f"[export] fell back to presence_of_element_located via {loc}") return el except Exception: continue raise TimeoutError("Export/Download button did not appear.") def _dump_export_menu_text(driver: RemoteWebDriver) -> list[str]: # capture visible menu items for debugging if selection fails items = driver.find_elements(By.XPATH, "//*[@role='menuitem' or @role='option' or @role='menuitemcheckbox']//*[self::span or self::div or self::button]|//*[@role='menuitem' or @role='option' or @role='menuitemcheckbox']") out: list[str] = [] for it in items: try: txt = (it.text or "").strip() if txt: out.append(txt) except Exception: continue # de-dup while preserving order seen = set() uniq = [] for t in out: if t not in seen: seen.add(t) uniq.append(t) return uniq[:30] def _debug_export_enabled() -> bool: v = os.environ.get("POWERBI_DEBUG_EXPORT", "").strip().lower() return v in ("1", "true", "yes", "y", "on") def _js_collect_exportish(driver: RemoteWebDriver): """Collect potentially relevant clickable elements in the *current browsing context*. This runs inside whatever frame you're currently in. """ js = r""" const kw = /export|download|csv/i; const nodes = Array.from(document.querySelectorAll( 'button, [role="button"], [role="menuitem"], [role="option"], [data-testid], a' )); function t(v){ return (v || '').toString().trim(); } function trunc(s, n){ s = t(s); return s.length > n ? (s.slice(0, n) + '…') : s; } const out = []; for (const e of nodes) { try { const text = trunc(e.innerText || e.textContent || '', 140); const aria = trunc(e.getAttribute('aria-label') || '', 140); const title = trunc(e.getAttribute('title') || '', 140); const dt = trunc(e.getAttribute('data-testid') || '', 140); const role = t(e.getAttribute('role') || ''); const hay = (text + ' ' + aria + ' ' + title + ' ' + dt).trim(); if (!kw.test(hay)) continue; const cs = window.getComputedStyle(e); const rect = e.getBoundingClientRect(); const vis = cs && cs.display !== 'none' && cs.visibility !== 'hidden' && rect.width > 0 && rect.height > 0; out.push({ tag: e.tagName, id: t(e.id), className: trunc(e.className || '', 180), role, text, ariaLabel: aria, title, dataTestid: dt, disabled: !!e.disabled, ariaDisabled: t(e.getAttribute('aria-disabled')), href: trunc(e.getAttribute('href') || '', 200), visible: vis, rect: { x: rect.x, y: rect.y, w: rect.width, h: rect.height }, }); } catch (err) { // ignore } } const iframes = Array.from(document.querySelectorAll('iframe')).map(f => ({ id: t(f.id), name: t(f.name), className: trunc(f.className || '', 180), src: trunc(f.getAttribute('src') || '', 240) })); return { candidates: out.slice(0, 200), iframes }; """ try: return driver.execute_script(js) except Exception as e: return {"candidates": [], "iframes": [], "error": repr(e)} def dump_export_debug(driver: RemoteWebDriver, label: str) -> None: """Save screenshot + JSON of export/download/csv-related UI elements for this context.""" out_dir = Path.cwd().resolve() / "export_debug" out_dir.mkdir(parents=True, exist_ok=True) try: png = out_dir / f"export_debug_{label}.png" driver.save_screenshot(str(png)) print(f"Export debug screenshot: {png}") except Exception as e: print(f"(debug) screenshot failed in {label}: {e}") payload = { "label": label, "url": getattr(driver, "current_url", ""), "title": "", "collected": {}, } try: payload["title"] = driver.title except Exception: pass payload["collected"] = _js_collect_exportish(driver) try: jsn = out_dir / f"export_debug_{label}.json" jsn.write_text(json.dumps(payload, indent=2), encoding="utf-8") print(f"Export debug JSON: {jsn}") except Exception as e: print(f"(debug) json write failed in {label}: {e}") def _open_export_menu_via_file_right_arrow(driver: RemoteWebDriver, timeout: int = 6) -> bool: """Fallback: focus the *File* button, press Right Arrow to move to Export, then press Enter. Why this exists: - Power BI/Fluent UI command bars often render *duplicate* Export buttons (hidden/overflow variants). - Selenium may click a hidden clone successfully (no exception) but nothing opens. - Keyboard navigation on the command bar is frequently more reliable. Returns True if a menu overlay appears after attempting the key sequence. """ wait = WebDriverWait(driver, timeout) file_locs = ( (By.CSS_SELECTOR, "button[title='File'], button[aria-label='File'], [role='menuitem'][title='File']"), (By.XPATH, "//*[self::button or @role='menuitem' or @role='button'][@title='File' or normalize-space()='File' or .//span[normalize-space()='File']]"), (By.XPATH, "//*[self::button or @role='menuitem' or @role='button'][contains(translate(@aria-label,'ABCDEFGHIJKLMNOPQRSTUVWXYZ','abcdefghijklmnopqrstuvwxyz'),'file')]"), ) for loc in file_locs: try: file_btn = wait.until(lambda d: _first_interactable(d, loc) or None) except Exception: file_btn = None if file_btn is None: continue try: _scroll_into_view(driver, file_btn) except Exception: pass # Ensure focus lands on the command bar item try: driver.execute_script("arguments[0].focus();", file_btn) except Exception: pass try: ActionChains(driver).move_to_element(file_btn).pause(0.05).click().perform() except Exception: try: file_btn.click() except Exception: pass time.sleep(0.12) # Prefer sending keys to the element we just focused. try: file_btn.send_keys(Keys.ARROW_RIGHT) time.sleep(0.15) file_btn.send_keys(Keys.ENTER) except Exception: try: ActionChains(driver).send_keys(Keys.ARROW_RIGHT).pause(0.15).send_keys(Keys.ENTER).perform() except Exception: try: driver.find_element(By.TAG_NAME, 'body').send_keys(Keys.ARROW_RIGHT) time.sleep(0.15) driver.find_element(By.TAG_NAME, 'body').send_keys(Keys.ENTER) except Exception: pass # Some builds need a second ENTER (your observation) try: _wait_for_export_menu(driver, timeout=max(2, int(timeout / 2))) return True except Exception: try: ActionChains(driver).send_keys(Keys.ENTER).perform() _wait_for_export_menu(driver, timeout=max(2, int(timeout / 2))) return True except Exception: pass return False def _wait_for_export_menu(driver: RemoteWebDriver, timeout: int) -> None: """Wait for the Export/Download menu overlay to appear.""" wait = WebDriverWait(driver, timeout) for loc in ( (By.CSS_SELECTOR, "ul.ms-ContextualMenu-list"), (By.CSS_SELECTOR, "div.ms-ContextualMenu"), (By.XPATH, "//*[@role='menu']"), (By.XPATH, "//ul[contains(@class,'ContextualMenu')]") ): try: wait.until(EC.presence_of_element_located(loc)) return except Exception: continue def _try_click_csv_menu_item(driver: RemoteWebDriver, timeout: int = 10) -> bool: """Assumes the export menu is already open in the *current browsing context*.""" wait = WebDriverWait(driver, timeout) # First: the stable attribute from your screenshot. css_locs = [ (By.CSS_SELECTOR, "button[data-testid='export-csv-btn'], [data-testid='export-csv-btn']"), ] # Then: robust text/title-based fallbacks. xp_locs = [ (By.XPATH, "//*[@data-testid='export-csv-btn']"), (By.XPATH, "//span[contains(translate(normalize-space(.), 'ABCDEFGHIJKLMNOPQRSTUVWXYZ','abcdefghijklmnopqrstuvwxyz'), 'comma separated values')]/ancestor::button[1]"), (By.XPATH, "//li[contains(translate(@title,'ABCDEFGHIJKLMNOPQRSTUVWXYZ','abcdefghijklmnopqrstuvwxyz'), 'comma separated values')]/descendant::button[1]"), (By.XPATH, "//button[contains(translate(@title,'ABCDEFGHIJKLMNOPQRSTUVWXYZ','abcdefghijklmnopqrstuvwxyz'), 'comma separated values')]"), (By.XPATH, "//*[self::button or @role='menuitem' or @role='option']" "[contains(translate(normalize-space(.), 'ABCDEFGHIJKLMNOPQRSTUVWXYZ','abcdefghijklmnopqrstuvwxyz'), 'comma separated values')]"), (By.XPATH, "//*[self::button or @role='menuitem' or @role='option']" "[contains(translate(normalize-space(.), 'ABCDEFGHIJKLMNOPQRSTUVWXYZ','abcdefghijklmnopqrstuvwxyz'), '.csv')]"), ] for loc in css_locs: try: el = wait.until(EC.presence_of_element_located(loc)) if el is not None: _v(f"[export] clicking CSV via {loc}") _safe_click(driver, el) print("Export: clicked CSV option.") return True except Exception: continue for loc in xp_locs: try: el = wait.until(EC.presence_of_element_located(loc)) if el is not None: _v(f"[export] clicking CSV via {loc}") _safe_click(driver, el) print("Export: clicked CSV option.") return True except Exception: continue return False def _attempt_export_csv_in_current_context( driver: RemoteWebDriver, button_timeout: int = 12, menu_timeout: int = 18, ) -> bool: """Try clicking Export/Download and then the CSV item, without switching frames. Returns True on success, False if the CSV option wasn't found/clicked. """ try: _v(f"[export] attempt in context url={getattr(driver,'current_url','')}") except Exception: pass print("Export: searching for Export/Download button...") export_clicked = False try: export_btn = wait_for_export_button(driver, timeout=button_timeout) _v("[export] clicking export/download button") print("Export: clicking Export/Download...") _safe_click(driver, export_btn) export_clicked = True except Exception as e: _v(f"[export] export button not found/clickable in this context: {e}") print("Export: export button not found/clickable; trying JS fallback...") # Last-chance: JS click by data-testid / title / aria-label. try: export_clicked = bool(driver.execute_script( """ const sel = [ "[data-testid='toolbar-export-dropdown']", "[data-testid='trident-export-menu-button']", "[data-testid='toolbar-download-dropdown']", "[data-testid='trident-download-menu-button']", "button[title='Export']", "button[title='Download']", "button[aria-label*='Export']", "button[aria-label*='Download']", "[role='button'][title='Export']", "[role='button'][title='Download']", ].join(','); const b = document.querySelector(sel); if (b) { b.click(); return true; } return false; """ )) if export_clicked: _v("[export] JS-clicked export/download button") print("Export: JS-clicked Export/Download.") except Exception as e2: _v(f"[export] JS fallback click failed: {e2}") export_clicked = False if not export_clicked: print("Export: could not find Export/Download in this context.") return False time.sleep(0.35) print("Export: export menu should be open; selecting CSV...") # Menu overlay can render in the current frame OR bubble up to the top document. per_ctx_timeout = max(4, int(menu_timeout / 2)) # 1) Try current browsing context try: _wait_for_export_menu(driver, timeout=per_ctx_timeout) except Exception: # If click succeeded but no menu appeared, use keyboard navigation on the command bar. try: if _open_export_menu_via_file_right_arrow(driver, timeout=per_ctx_timeout): pass except Exception: pass if _try_click_csv_menu_item(driver, timeout=per_ctx_timeout): print("Export: clicked CSV option.") return True # 2) Try default content (very common for Office/Fluent contextual menus) try: driver.switch_to.default_content() try: _wait_for_export_menu(driver, timeout=per_ctx_timeout) except Exception: # Same keyboard fallback at top-level try: _open_export_menu_via_file_right_arrow(driver, timeout=per_ctx_timeout) except Exception: pass if _try_click_csv_menu_item(driver, timeout=per_ctx_timeout): print("Export: clicked CSV option.") return True except Exception: pass # 3) Try any iframes from default content (in case the menu lives in another layer) try: driver.switch_to.default_content() frames = driver.find_elements(By.CSS_SELECTOR, "iframe") for idx, fr in enumerate(frames[:15]): try: driver.switch_to.frame(fr) if _try_click_csv_menu_item(driver, timeout=3): _v(f"[export] clicked CSV inside iframe #{idx}") print("Export: clicked CSV option.") return True except Exception: pass finally: try: driver.switch_to.default_content() except Exception: pass except Exception: pass return False def export_csv(driver: RemoteWebDriver) -> None: """Click Export/Download and choose CSV. Power BI can surface the control either inside the paginated report frame *or* on the outer (app.powerbi.com) top bar. We try both. """ # 1) Try inside the current context (typically the paginated report iframe) if _attempt_export_csv_in_current_context(driver): return # 1b) Some builds nest the report toolbar in an inner iframe (e.g., rdl-host-container). # Try one level down before giving up on the report frame. try: inner_frames = driver.find_elements(By.CSS_SELECTOR, "iframe") for fr in inner_frames: try: driver.switch_to.frame(fr) if _attempt_export_csv_in_current_context(driver, button_timeout=6, menu_timeout=10): return except Exception: pass finally: try: driver.switch_to.parent_frame() except Exception: # If parent_frame fails, bail to default and continue with other strategies. try: driver.switch_to.default_content() except Exception: pass except Exception: pass # 2) Try at the top-level (default content) try: driver.switch_to.default_content() except Exception: pass if _attempt_export_csv_in_current_context(driver): return # 3) Could not find/click CSV in either place. Dump debug artifacts if requested. if _debug_export_enabled(): try: dump_export_debug(driver, "default") except Exception: pass try: switch_to_paginated_report_iframe(driver) dump_export_debug(driver, "frame") except Exception: pass seen = _dump_export_menu_text(driver) raise RuntimeError( "Could not click Export/Download -> CSV. " "Set POWERBI_DEBUG_EXPORT=1 and rerun to generate export_debug/*.json + *.png. " f"Seen menu items: {seen}" ) # --------------------------- # Download handling # --------------------------- def resolve_download_dir(session_download_dir: Optional[str]) -> Path: """ Where Firefox is actually downloading *right now*. We DO NOT force it to Inventory (or anywhere). We just watch the real folder. Priority: 1) POWERBI_DOWNLOAD_DIR env override 2) download_dir stored in powerbi_session.json 3) ./downloads fallback """ env = os.environ.get("POWERBI_DOWNLOAD_DIR", "").strip() if env: p = Path(env).expanduser().resolve() p.mkdir(parents=True, exist_ok=True) return p if session_download_dir: p = Path(session_download_dir).expanduser().resolve() p.mkdir(parents=True, exist_ok=True) return p p = (Path.cwd() / "downloads").resolve() p.mkdir(parents=True, exist_ok=True) return p def _snapshot_files(d: Path) -> set[Path]: if not d.exists(): return set() return {p for p in d.iterdir() if p.is_file()} def wait_for_new_download(download_dir: Path, before: set[Path], timeout: int = DOWNLOAD_TIMEOUT) -> Path: """ Waits for a new fully-downloaded file to appear in download_dir. Handles Firefox .part files. """ end = time.time() + timeout last_seen = None while time.time() < end: files = [p for p in download_dir.iterdir() if p.is_file()] completed = [p for p in files if not p.name.endswith(".part")] new = [p for p in completed if p not in before] if not new: time.sleep(0.35) continue new.sort(key=lambda p: p.stat().st_mtime, reverse=True) cand = new[0] last_seen = cand # If cand still has a .part sibling, download not finished if (download_dir / (cand.name + ".part")).exists(): time.sleep(0.35) continue # Or any .part still present means something is still downloading if any(p.name.endswith(".part") for p in files): time.sleep(0.35) continue return cand raise TimeoutError(f"Download did not complete in time. Last seen: {last_seen}") def move_to_dated_name( downloaded: Path, out_dir: Optional[Path] = None, stamp_override: Optional[str] = None, ) -> Path: """ Moves the downloaded file into this script's folder (or out_dir if provided) and renames it to: YYYYMMDD_On_Hand_Footwear.csv """ if out_dir is None: out_dir = Path(__file__).resolve().parent / "Footwear" out_dir.mkdir(parents=True, exist_ok=True) stamp = (stamp_override or "").strip() or datetime.now().strftime("%Y%m%d") out = out_dir / f"{stamp}_On_Hand_Footwear.csv" # If file exists, overwrite (or you can add _1 logic if you prefer) if out.exists(): out.unlink() downloaded.replace(out) return out # --------------------------- # Main # --------------------------- def main() -> int: session_path = _pick_session_file() output_stamp = os.environ.get("POWERBI_OUTPUT_STAMP", "").strip() try: executor_url, session_id, session_download_dir = _load_session_file(session_path) print(f"Using session file: {session_path}") driver = attach_to_session(executor_url, session_id) except Exception as e: print(f"ERROR: Could not attach to existing Firefox session.\n{e}") return 2 download_dir = resolve_download_dir(session_download_dir) try: ensure_report_loaded(driver) switch_to_paginated_report_iframe(driver) set_single_combobox(driver, "Store Type", STORE_TYPE_VALUE) set_single_combobox(driver, "As Of", AS_OF_VALUE) wait_for_filter_panel_ready(driver) set_category_to_footwear_only(driver) set_single_combobox(driver, "Store No", STORE_NO_VALUE) set_single_combobox(driver, "Bucket", BUCKET_VALUE) before = _snapshot_files(download_dir) click_view_report_if_present(driver) # Even if View report button is disabled/hidden, Export should still exist in the report toolbar. export_csv(driver) downloaded = wait_for_new_download(download_dir, before, timeout=DOWNLOAD_TIMEOUT) final_path = move_to_dated_name(downloaded, stamp_override=output_stamp) print(f"Saved CSV as: {final_path}") except Exception as e: try: driver.switch_to.default_content() out = Path.cwd() / "pull_on_hand_footwear_error.png" driver.save_screenshot(str(out)) print(f"Saved error screenshot: {out}") except Exception: pass tb = traceback.format_exc() print( "ERROR: Failed while running automation.\n" f"Type: {type(e).__name__}\n" f"Message: {e!r}\n" f"Traceback:\n{tb}" ) return 3 return 0 if __name__ == "__main__": raise SystemExit(main())