# ================================================================ # OneDrive / Microsoft Planner — face-sheet download automation # ================================================================ from __future__ import annotations import base64 from dataclasses import dataclass, field import json import logging import os import re import shutil import time from datetime import datetime import urllib.error import urllib.parse import urllib.request from pathlib import Path import pyotp from chrome import DOWNLOADS_DIR as _CHROME_DOWNLOADS_DIR from chrome import capture_browser_screenshot from chrome import capture_fullscreen_screenshot from config import DEFAULT_ONEDRIVE_URL, DEFAULT_PLANNER_URL, PROJECT_ROOT, TODAY_ISO_DATE from debug_log import write_agent_session_debug as _agent_session_log from debug_log import write_debug as _write_debug from run_log import log_step, log_url from totp_utils import validate_totp_secret from selenium.common.exceptions import ( InvalidSelectorException, NoSuchElementException, StaleElementReferenceException, TimeoutException, ) from selenium.webdriver import ActionChains from selenium.webdriver.common.by import By from selenium.webdriver.common.keys import Keys from selenium.webdriver.remote.webdriver import WebDriver from selenium.webdriver.support import expected_conditions as EC from selenium.webdriver.support.ui import WebDriverWait _log = logging.getLogger(__name__) _COOKIE_JAR = PROJECT_ROOT / "Output" / "session" / "cookies_planner.json" WAIT_SHORT = 15 WAIT_MED = 30 WAIT_LONG = 60 POST_FACILITY_CLICK_WAIT = 30 POST_FACILITY_LOAD_DELAY = 3 PANEL_TAB_LOAD_DELAY = 2 # let task panel / tab content render before reads FACILITY_LOAD_AFTER_REFRESH = 45 DOWNLOAD_POLL_INTERVAL = 0.5 DOWNLOAD_TIMEOUT = 120 MIN_PDF_BYTES = 5000 _ACTIVE_PAPERCLIP_XPATH = ( "//button[@data-testid='attachment-indicator']" "[not(ancestor::*[contains(@aria-label,'Completed tasks')])]" ) # Planner 2024+ board: complete control is often button[role=checkbox], not label. _COMPLETION_CHECKBOX_UNCHECKED_XPATH = ( ".//label[@role='checkbox' and @aria-checked='false']" " | .//button[@role='checkbox' and @aria-checked='false']" " | .//*[@role='checkbox' and @aria-checked='false']" ) _COMPLETION_CHECKBOX_CHECKED_XPATH = ( ".//label[@role='checkbox' and @aria-checked='true']" " | .//button[@role='checkbox' and @aria-checked='true']" ) _PLANNER_TASK_CARD_XPATHS = ( "./ancestor::*[@data-dnd-role='card'][1]", "./ancestor::div[contains(@class,'planner-draggable-task-card')][1]", "./ancestor::*[@role='listitem'][@data-dnd-role='card'][1]", ) # Avoid duplicate processing within a single run (Shared list can contain duplicates). _PROCESSED_SHARED_FACILITIES: set[str] = set() FACILITY_NOT_APPEAR_MSG = "Facility did not appear in Shared plans search" def reset_run_state() -> None: """Clear per-run module-level state. Call at the start of each workflow run.""" global _PROCESSED_SHARED_FACILITIES _PROCESSED_SHARED_FACILITIES = set() class FacilityNotFoundError(Exception): """Facility could not be opened from My plans -> Shared (list or filter).""" _FILTER_BUTTON_LOCATORS = [ (By.CSS_SELECTOR, "button[aria-label='Filter by keyword']"), (By.XPATH, "//button[@aria-label='Filter by keyword']"), ] _FILTER_INPUT_LOCATORS = [ (By.CSS_SELECTOR, "input[aria-label='Filter by keyword']"), (By.CSS_SELECTOR, "input[placeholder*='Filter'][placeholder*='keyword']"), (By.XPATH, "//input[contains(@placeholder,'Filter') and contains(@placeholder,'keyword')]"), (By.CSS_SELECTOR, "input[type='search']"), (By.CSS_SELECTOR, "[role='searchbox']"), ] _FILTER_EMPTY_STATE_XPATHS = [ "//*[contains(.,'Clear your filters')]", "//*[contains(.,'No results')]", "//*[contains(.,'Search more plans')]", "//*[contains(.,'no plans')]", ] # Optional corporate proxy for HTTP downloads (same as browser proxy). _HTTP_PROXY: tuple[str, str, str, str] | None = None def configure_http_download_proxy( proxy_ip: str | None, proxy_port: str | None, proxy_username: str | None = None, proxy_password: str | None = None, ) -> None: """Use the same proxy as Chromium for SharePoint HTTP downloads.""" global _HTTP_PROXY if proxy_ip and proxy_port: _HTTP_PROXY = ( proxy_ip.strip(), str(proxy_port).strip(), (proxy_username or "").strip(), (proxy_password or "").strip(), ) else: _HTTP_PROXY = None _agent_log = _write_debug _agent_dbg_eb = _write_debug _dbg = _write_debug def _sanitize_folder_name(name: str) -> str: cleaned = re.sub(r'[<>:"/\\|?*]', "_", name.strip()) return cleaned or "unknown_facility" def _xpath_literal(text: str) -> str: """Build an XPath string literal safe for quotes in facility names.""" if "'" not in text: return f"'{text}'" if '"' not in text: return f'"{text}"' parts = text.split("'") return "concat(" + ", \"'\", ".join(f"'{p}'" for p in parts) + ")" def _wait(driver: WebDriver, timeout: int = WAIT_MED) -> WebDriverWait: return WebDriverWait(driver, timeout) def _is_facility_plan_url(url: str) -> bool: u = (url or "").lower() return "/plan/" in u or "planid=" in u def _is_planner_hub_url(url: str) -> bool: """My Tasks / My plans list — not an open facility board.""" u = (url or "").lower() if u in ("about:blank", "", "data:,"): return False if "planner.cloud.microsoft" not in u and "tasks.office.com" not in u: return False return not _is_facility_plan_url(u) def consolidate_browser_to_single_tab(driver: WebDriver) -> None: """ Close extra tabs from Chromium session restore or about:blank at startup. The automation expects one working tab before navigating to Planner. """ try: handles = list(driver.window_handles) if len(handles) <= 1: return def _tab_url(handle: str) -> str: try: driver.switch_to.window(handle) return (driver.current_url or "").strip().lower() except Exception: return "" keep = handles[-1] for handle in reversed(handles): url = _tab_url(handle) if url not in ("about:blank", "", "data:,"): keep = handle break closed = 0 for handle in handles: if handle == keep: continue try: driver.switch_to.window(handle) driver.close() closed += 1 except Exception: pass driver.switch_to.window(keep) log_step("Consolidated browser to single tab", closed=closed) except Exception as exc: _log.warning("consolidate_browser_to_single_tab: %s", exc) def focus_facility_plan_tab(driver: WebDriver) -> bool: """ Switch to the facility plan tab and close My Tasks / Shared list tabs. Planner opens a plan in a new tab when clicked from My plans -> Shared, leaving the original My Tasks tab open. """ try: plan_handle = None for handle in driver.window_handles: driver.switch_to.window(handle) if _is_facility_plan_url(driver.current_url or ""): plan_handle = handle break if not plan_handle: return False closed = 0 for handle in list(driver.window_handles): if handle == plan_handle: continue try: driver.switch_to.window(handle) url = (driver.current_url or "").strip().lower() if url in ("about:blank", "", "data:,") or _is_planner_hub_url(url): driver.close() closed += 1 except Exception: pass driver.switch_to.window(plan_handle) if closed: log_step( "Closed planner hub tab(s)", count=closed, plan_url=(driver.current_url or "")[:120], ) return True except Exception as exc: _log.warning("focus_facility_plan_tab: %s", exc) return False def _save_cookies(driver: WebDriver) -> None: try: _COOKIE_JAR.parent.mkdir(parents=True, exist_ok=True) cookies = driver.get_cookies() # Do not log cookie values; just store to disk. _COOKIE_JAR.write_text(json.dumps(cookies), encoding="utf-8") except Exception as exc: _log.warning("Could not save session cookies: %s", exc) def _load_cookies(driver: WebDriver, *, domain_url: str) -> bool: """ Best-effort cookie restore. We must visit the domain first before adding cookies. """ if not _COOKIE_JAR.is_file(): return False try: driver.get(domain_url) time.sleep(2) cookies = json.loads(_COOKIE_JAR.read_text(encoding="utf-8")) if not isinstance(cookies, list): return False added = 0 for c in cookies: if not isinstance(c, dict): continue # Selenium rejects some keys; keep the common ones. slim = {k: c.get(k) for k in ("name", "value", "domain", "path", "expiry", "secure", "httpOnly", "sameSite") if k in c} if "name" in slim and "value" in slim: try: driver.add_cookie(slim) added += 1 except Exception: continue if added: driver.get(domain_url) time.sleep(2) return added > 0 except Exception as exc: _log.warning("Could not restore session cookies: %s", exc) return False def _js_click(driver: WebDriver, element) -> None: try: tag = element.tag_name label = (element.text or element.get_attribute("aria-label") or "")[:80] log_step("Click element", tag=tag, label=label) except Exception: log_step("Click element") driver.execute_script("arguments[0].scrollIntoView({block:'center'});", element) time.sleep(0.3) driver.execute_script("arguments[0].click();", element) def _first_clickable(driver: WebDriver, locators: list[tuple[str, str]], timeout: int = WAIT_MED): end = time.time() + timeout last_err = None while time.time() < end: for by, value in locators: try: el = driver.find_element(by, value) if el.is_displayed() and el.is_enabled(): return el except (NoSuchElementException, StaleElementReferenceException) as exc: last_err = exc time.sleep(0.4) raise TimeoutException(f"No clickable element for locators (last: {last_err})") def _click_any(driver: WebDriver, locators: list[tuple[str, str]], timeout: int = WAIT_MED) -> None: el = _first_clickable(driver, locators, timeout=timeout) _js_click(driver, el) def _type_into(driver: WebDriver, locators: list[tuple[str, str]], text: str, clear: bool = True) -> None: el = _first_clickable(driver, locators, timeout=WAIT_LONG) _js_click(driver, el) if clear: el.send_keys(Keys.CONTROL, "a") el.send_keys(Keys.BACKSPACE) el.send_keys(text) def _on_login_host(driver: WebDriver) -> bool: url = (driver.current_url or "").lower() return any( h in url for h in ( "login.microsoftonline.com", "login.live.com", "login.microsoft.com", ) ) def _on_planner_app(driver: WebDriver) -> bool: url = (driver.current_url or "").lower() return any( h in url for h in ( "planner.cloud.microsoft", "tasks.office.com", "tasks.office365.us", ) ) def _on_planner_plan_board(driver: WebDriver) -> bool: url = (driver.current_url or "").lower() return ( "planner.cloud.microsoft" in url and "/plan/" in url and ("board" in url or "grid" in url) ) def _plan_load_error_present(driver: WebDriver) -> bool: """True when Planner shows any error overlay that blocks normal operation.""" phrases = ( "couldn't open your plan", "could not open your plan", "try again in a few minutes", "something went wrong", "something's not right", "we couldn't load", "please try again", ) try: blob = (driver.find_element(By.TAG_NAME, "body").text or "").lower() return any(p in blob for p in phrases) except Exception: return False def _recover_planner_error(driver: WebDriver) -> bool: """Detect and dismiss a Planner error overlay ('Something went wrong', etc). Tries clicking the 'Try again' button first; falls back to a full page refresh. Returns True if an error was detected (regardless of whether recovery succeeded). """ if not _plan_load_error_present(driver): return False log_step("Planner error overlay detected — attempting recovery") if _click_plan_try_again(driver): time.sleep(4) if not _plan_load_error_present(driver): log_step("Planner error overlay cleared via Try again") return True # Fallback: full page refresh try: log_step("Refreshing page to clear Planner error overlay") driver.refresh() time.sleep(6) driver.switch_to.default_content() _switch_to_planner_context(driver) log_step("Page refreshed — error overlay cleared") except Exception as exc: log_step("Refresh failed during error recovery", error=str(exc)[:120]) return True def _plan_page_ready(driver: WebDriver) -> bool: """Plan UI is usable — not the error page.""" if _plan_load_error_present(driver): return False url = (driver.current_url or "").lower() if not ("/plan/" in url or "planid=" in url): return False return bool( driver.find_elements(By.XPATH, "//*[@role='tab' and normalize-space()='Board']") or driver.find_elements(By.XPATH, "//*[@role='tab' and normalize-space()='Grid']") or driver.find_elements( By.XPATH, "//*[contains(@aria-label,'Add task') or normalize-space()='Add task']", ) or driver.find_elements(By.CSS_SELECTOR, "[role='grid'].grid-container, .grid-container") or driver.find_elements( By.XPATH, "//*[contains(@data-testid,'TaskCard') or contains(@class,'TaskCard')]", ) ) def _click_plan_try_again(driver: WebDriver) -> bool: """Click Planner's 'Try again' button on the plan load error page.""" locators = [ (By.XPATH, "//button[normalize-space()='Try again']"), (By.XPATH, "//*[self::button or self::a][normalize-space()='Try again']"), ] try: log_step("Click Try again on plan error page") _click_any(driver, locators, timeout=8) time.sleep(4) return True except TimeoutException: return False def _board_probes_present(driver: WebDriver) -> bool: """True when the plan board UI (task cards) is visible in the current frame.""" probes = [ "//button[@data-testid='attachment-indicator']", "//label[@role='checkbox']", "//*[@role='checkbox']", "//*[normalize-space()='Add task']", "//*[@role='tab' and normalize-space()='Board']", ] for xpath in probes: try: for el in driver.find_elements(By.XPATH, xpath)[:12]: if el.is_displayed(): return True except Exception: continue return False def _on_sharepoint_or_onedrive(driver: WebDriver) -> bool: url = (driver.current_url or "").lower() return any( h in url for h in ("sharepoint.com", "onedrive.live.com", "office.com", "office365.com") ) def _on_m365_app(driver: WebDriver) -> bool: return _on_planner_app(driver) or _on_sharepoint_or_onedrive(driver) def _is_stay_signed_in_prompt(driver: WebDriver) -> bool: try: if driver.find_elements(By.ID, "idBtn_Back"): return True if driver.find_elements( By.XPATH, "//*[contains(translate(normalize-space(.)," "'ABCDEFGHIJKLMNOPQRSTUVWXYZ','abcdefghijklmnopqrstuvwxyz')," "'stay signed in')]", ): return True return "stay signed in" in (driver.page_source or "").lower() except Exception: return False def _otp_input_visible(driver: WebDriver) -> bool: try: for by, sel in ( (By.CSS_SELECTOR, "input[name='otc']"), (By.ID, "idTxtBx_SAOTCC_OTC"), (By.CSS_SELECTOR, "input[type='tel']"), ): for el in driver.find_elements(by, sel): if el.is_displayed(): return True except Exception: pass return False def _email_form_visible(driver: WebDriver) -> bool: """True when the Microsoft login page shows an email / username field.""" try: for by, sel in ( (By.CSS_SELECTOR, "input[type='email']"), (By.NAME, "loginfmt"), (By.ID, "i0116"), ): for el in driver.find_elements(by, sel): if el.is_displayed(): return True except Exception: pass return False def _password_form_visible(driver: WebDriver) -> bool: """True when the Microsoft login page shows a password field (with or without email step).""" try: for by, sel in ( (By.CSS_SELECTOR, "input[type='password']"), (By.NAME, "passwd"), (By.ID, "i0118"), ): for el in driver.find_elements(by, sel): if el.is_displayed(): return True except Exception: pass return False def _login_form_visible(driver: WebDriver) -> bool: return _email_form_visible(driver) or _password_form_visible(driver) def _needs_microsoft_login(driver: WebDriver) -> bool: """Planner launch may redirect to login or show password directly (remembered account).""" return _on_login_host(driver) or _login_form_visible(driver) def _planner_session_ready(driver: WebDriver) -> bool: """Signed in on Planner — no Microsoft login/password prompt visible.""" return ( _on_planner_app(driver) and not _on_login_host(driver) and not _login_form_visible(driver) ) def _remove_empty_directory(path: Path) -> None: """Remove facility folder if it was created for downloads but stayed empty.""" try: if path.is_dir() and not any(path.iterdir()): path.rmdir() except OSError: pass try: parent = path.parent if parent.is_dir() and parent != path and not any(parent.iterdir()): parent.rmdir() except OSError: pass def _ensure_download_folder(folder: Path) -> None: """Create Original// only when a download is about to start.""" folder.mkdir(parents=True, exist_ok=True) def set_chrome_download_directory( driver: WebDriver, folder: Path, *, create_dir: bool = False, ) -> None: """Set download path for the current tab via CDP.""" if create_dir: _ensure_download_folder(folder) try: driver.execute_cdp_cmd( # pyright: ignore[reportAttributeAccessIssue] "Page.setDownloadBehavior", {"behavior": "allow", "downloadPath": str(folder.resolve())}, ) except Exception as exc: print(f"[WARN] Could not set CDP download path ({exc}); using default Downloads") def _relocate_new_pdfs_from_dirs( extra_dirs: list[Path], dest_folder: Path, before: set[str], ) -> list[Path]: """Move new PDFs from Chrome default download dirs into the facility folder.""" _ensure_download_folder(dest_folder) moved: list[Path] = [] for folder in extra_dirs: if not folder.exists(): continue for src in sorted(folder.glob("*.pdf"), key=lambda p: p.stat().st_mtime): if src.name in before: continue dest = _unique_file_path(dest_folder, src.name) try: shutil.move(str(src), str(dest)) before.add(dest.name) moved.append(dest) log_step("Relocated PDF from Chrome download dir", file=dest.name, from_dir=str(folder)) except Exception as exc: _log.warning("Could not move %s -> %s: %s", src, dest, exc) return moved def _wait_for_new_download( folder: Path, before: set[str], timeout: float = DOWNLOAD_TIMEOUT, *, extra_watch_dirs: list[Path] | None = None, ) -> list[Path]: """Poll folder until new PDF(s) appear and .crdownload files finish.""" deadline = time.time() + timeout if not folder.exists(): folder.mkdir(parents=True, exist_ok=True) watch = [folder] if extra_watch_dirs: watch.extend(extra_watch_dirs) while time.time() < deadline: partial: list[Path] = [] for d in watch: if d.exists(): partial.extend(d.glob("*.crdownload")) partial.extend(d.glob("*.tmp")) if partial: time.sleep(DOWNLOAD_POLL_INTERVAL) continue after = {p.name for p in folder.iterdir() if p.is_file()} new_names = after - before new_pdfs = [ folder / n for n in new_names if n.lower().endswith(".pdf") ] if new_pdfs: return sorted(new_pdfs, key=lambda p: p.stat().st_mtime) relocated = _relocate_new_pdfs_from_dirs( [d for d in watch if d != folder], folder, before, ) if relocated: return relocated time.sleep(DOWNLOAD_POLL_INTERVAL) return _relocate_new_pdfs_from_dirs( [d for d in watch if d != folder], folder, before, ) def _submit_microsoft_totp(driver: WebDriver, totp_secret: str) -> None: raw = (totp_secret or "").strip() secret_mapped = validate_totp_secret(raw) _dbg( "H1", "src/onedrive_planner.py:_submit_microsoft_totp", "TOTP secret sanitized", { "raw_len": len(raw), "sanitized_len": len(secret_mapped), "had_separators": bool(re.search(r"[\s\-]", raw)), }, ) totp_input_locators = [ (By.CSS_SELECTOR, "input[name='otc']"), (By.CSS_SELECTOR, "input[type='tel']"), (By.CSS_SELECTOR, "input[inputmode='numeric']"), (By.XPATH, "//input[contains(@aria-label,'code') or contains(@placeholder,'code')]"), (By.ID, "idTxtBx_SAOTCC_OTC"), ] for attempt in range(2): if _is_stay_signed_in_prompt(driver): _dbg( "H4", "src/onedrive_planner.py:_submit_microsoft_totp", "Stay-signed-in before OTP entry", {"attempt": attempt + 1}, ) return if not _otp_input_visible(driver): _dbg( "H4", "src/onedrive_planner.py:_submit_microsoft_totp", "No OTP field visible", {"attempt": attempt + 1, "url": (driver.current_url or "")[:120]}, ) return code = pyotp.TOTP(secret_mapped).now() print(f"[INFO] Entering Microsoft Authenticator code (attempt {attempt + 1})") _type_into(driver, totp_input_locators, code, clear=True) time.sleep(0.5) verify_locators = [ (By.CSS_SELECTOR, "input[type='submit']"), (By.XPATH, "//input[@type='submit' and (@value='Verify' or @value='Sign in')]"), (By.XPATH, "//button[@type='submit']"), (By.ID, "idSubmit_SAOTCC_Continue"), (By.ID, "idSIButton9"), ] try: _click_any(driver, verify_locators, timeout=WAIT_SHORT) except TimeoutException: if _otp_input_visible(driver): _first_clickable(driver, totp_input_locators).send_keys(Keys.ENTER) time.sleep(4) _dismiss_stay_signed_in(driver) stay_prompt = _is_stay_signed_in_prompt(driver) otp_visible = _otp_input_visible(driver) on_login = _on_login_host(driver) _dbg( "H4", "src/onedrive_planner.py:_submit_microsoft_totp", "After OTP submit", { "attempt": attempt + 1, "url": (driver.current_url or "")[:120], "on_login_host": on_login, "otp_field_present": otp_visible, "stay_prompt_present": stay_prompt, }, ) if stay_prompt or not on_login or not otp_visible: return print("[WARN] OTP still visible on login host — retrying with fresh code") time.sleep(5) def _dismiss_stay_signed_in(driver: WebDriver) -> None: _dbg( "H4", "src/onedrive_planner.py:_dismiss_stay_signed_in", "Checking stay signed in prompt", { "url": (driver.current_url or "")[:120], }, ) locators = [ (By.ID, "idSIButton9"), (By.ID, "idBtn_Back"), (By.XPATH, "//input[@type='submit' and @value='Yes']"), (By.XPATH, "//button[contains(.,'Yes')]"), (By.XPATH, "//input[@type='submit' and @value='No']"), (By.XPATH, "//button[contains(.,'No')]"), ] try: # Prefer "No" to reduce extra consent-style steps. no_first = [ (By.ID, "idBtn_Back"), (By.XPATH, "//input[@type='submit' and @value='No']"), (By.XPATH, "//button[contains(.,'No')]"), ] try: _click_any(driver, no_first, timeout=3) _dbg("H4", "src/onedrive_planner.py:_dismiss_stay_signed_in", "Clicked No", {}) except TimeoutException: _click_any(driver, locators, timeout=8) _dbg("H4", "src/onedrive_planner.py:_dismiss_stay_signed_in", "Clicked a stay-signed-in button", {}) time.sleep(2) except TimeoutException: pass def _perform_microsoft_login( driver: WebDriver, username: str, password: str, totp_secret: str, ) -> None: """Username → password → TOTP → dismiss stay-signed-in. Handles password-only prompts when Planner opens with a remembered account. """ email_locators = [ (By.CSS_SELECTOR, "input[type='email']"), (By.NAME, "loginfmt"), (By.ID, "i0116"), ] next_locators = [ (By.ID, "idSIButton9"), (By.CSS_SELECTOR, "input[type='submit']"), (By.XPATH, "//input[@type='submit' and @value='Next']"), ] password_locators = [ (By.CSS_SELECTOR, "input[type='password']"), (By.NAME, "passwd"), (By.ID, "i0118"), ] signin_locators = [ (By.ID, "idSIButton9"), (By.CSS_SELECTOR, "input[type='submit']"), (By.XPATH, "//input[@type='submit' and @value='Sign in']"), (By.XPATH, "//button[@type='submit']"), ] if _email_form_visible(driver): _log.info("Entering Microsoft username...") log_step("Microsoft login — enter username") _wait(driver, WAIT_LONG).until( EC.element_to_be_clickable(email_locators[0]) ) _type_into(driver, email_locators, username.strip()) _click_any(driver, next_locators) try: _wait(driver, WAIT_MED).until( lambda d: _password_form_visible(d) or not _on_login_host(d) ) except TimeoutException: pass elif _password_form_visible(driver): log_step("Microsoft login — password only (account already selected)") _log.info("Password-only login prompt — skipping username entry") if not _password_form_visible(driver): if not _on_login_host(driver): return _log.warning("Expected password field on login page — retrying after brief wait") try: _wait(driver, WAIT_SHORT).until(_password_form_visible) except TimeoutException: _log.warning("Password field not found on Microsoft login page") return _log.info("Entering Microsoft password...") log_step("Microsoft login — enter password") _type_into(driver, password_locators, password, clear=True) _click_any(driver, signin_locators) if _on_login_host(driver): try: _wait(driver, WAIT_MED).until( lambda d: _otp_input_visible(d) or _is_stay_signed_in_prompt(d) or not _on_login_host(d) ) except TimeoutException: pass if _is_stay_signed_in_prompt(driver): _dismiss_stay_signed_in(driver) elif _on_login_host(driver) and _otp_input_visible(driver): _log.info("MFA / additional verification detected") _submit_microsoft_totp(driver, totp_secret) _dismiss_stay_signed_in(driver) _dismiss_stay_signed_in(driver) def login_and_open_planner( driver: WebDriver, username: str, password: str, totp_secret: str, planner_url: str = DEFAULT_PLANNER_URL, ) -> None: """ Open Planner directly (planner.cloud.microsoft), sign in if needed, land on Planner. """ if not username.strip() or not password: raise ValueError("OneDrive Username and OneDrive Password are required in input.xlsx") target = (planner_url or DEFAULT_PLANNER_URL).strip() log_step("Open Planner URL", url=target[:200]) _dbg("H5", "src/onedrive_planner.py:login_and_open_planner", "Navigate", {"url": target[:160]}) if _load_cookies(driver, domain_url="https://planner.cloud.microsoft/"): log_step("Loaded saved session cookies") else: log_step("No cookie jar — full login may be required") driver.get(target) try: _wait(driver, WAIT_MED).until( lambda d: _planner_session_ready(d) or _needs_microsoft_login(d) ) except TimeoutException: time.sleep(3) log_url(driver, "after_navigate") if _planner_session_ready(driver): log_step("Already signed in (cookies/session)") _switch_to_planner_context(driver) return if _needs_microsoft_login(driver): log_step("Microsoft login required") _perform_microsoft_login(driver, username, password, totp_secret) _dismiss_stay_signed_in(driver) if _needs_microsoft_login(driver): log_step("Microsoft login still required — retrying") _perform_microsoft_login(driver, username, password, totp_secret) _dismiss_stay_signed_in(driver) try: _wait(driver, WAIT_LONG).until(_planner_session_ready) except TimeoutException as exc: _dbg( "H5", "src/onedrive_planner.py:login_and_open_planner", "Planner not reached in time", {"url": (driver.current_url or "")[:160]}, ) raise RuntimeError( "Login did not reach Microsoft Planner. " f"Current URL: {driver.current_url}" ) from exc _dbg( "H5", "src/onedrive_planner.py:login_and_open_planner", "Planner reached", {"url": (driver.current_url or "")[:160]}, ) _switch_to_planner_context(driver) _dbg( "H5", "src/onedrive_planner.py:login_and_open_planner", "Planner ready", {"url": (driver.current_url or "")[:160]}, ) log_step("Microsoft Planner login complete") _save_cookies(driver) log_step("Saved session cookies for next run") def login_microsoft_onedrive( driver: WebDriver, username: str, password: str, totp_secret: str, onedrive_url: str = DEFAULT_ONEDRIVE_URL, ) -> None: """Deprecated: sign in via OneDrive URL. Use :func:`login_and_open_planner` instead.""" if not username.strip() or not password: raise ValueError("OneDrive Username and OneDrive Password are required in input.xlsx") print(f"[INFO] Navigating to OneDrive → {onedrive_url[:90]}...") driver.get(onedrive_url) time.sleep(3) if _on_m365_app(driver) and not _needs_microsoft_login(driver): print("[INFO] Already signed in to Microsoft 365") return if _needs_microsoft_login(driver): _perform_microsoft_login(driver, username, password, totp_secret) try: _wait(driver, WAIT_LONG).until( lambda d: _on_m365_app(d) and not _on_login_host(d) ) except TimeoutException as exc: raise RuntimeError( "Microsoft login did not complete. " f"Current URL: {driver.current_url}" ) from exc print("[SUCCESS] Microsoft login complete") def _switch_to_planner_context(driver: WebDriver, *, retries: int = 3) -> bool: """ Switch into the iframe that hosts the Planner React app. Planner uses different surfaces (My tasks grid, My plans list, plan board) and the previous "search box" probe is not always present. This version searches for a frame where *any* Planner UI probes exist and supports one nested iframe. """ probes: list[tuple[str, str]] = [ # Board task cards (planner.cloud.microsoft plan board) (By.CSS_SELECTOR, "button[data-testid='attachment-indicator']"), (By.XPATH, "//label[@role='checkbox']"), (By.XPATH, "//*[@role='checkbox']"), # My plans -> Shared filter box (By.CSS_SELECTOR, "input[placeholder*='Filter'][placeholder*='keyword']"), (By.XPATH, "//input[contains(@placeholder,'Filter') and contains(@placeholder,'keyword')]"), # Left-nav / section labels (By.XPATH, "//*[self::span or self::div][normalize-space()='My plans']"), (By.XPATH, "//*[self::span or self::div][normalize-space()='My tasks']"), # Plan view tabs (By.XPATH, "//*[@role='tab' and (normalize-space()='Board' or normalize-space()='Grid')]"), (By.XPATH, "//*[contains(@aria-label,'Add task') or normalize-space()='Add task']"), # Generic search (some views) (By.CSS_SELECTOR, "input[type='search']"), (By.CSS_SELECTOR, "[role='searchbox']"), ] def _has_probe() -> bool: for by, sel in probes: try: if driver.find_elements(by, sel): return True except Exception: continue return False def _try_frames(frames) -> bool: for fr in frames: try: # Wait for frame availability (helps when iframe is present but still loading). try: _wait(driver, WAIT_SHORT).until(EC.frame_to_be_available_and_switch_to_it(fr)) except Exception: driver.switch_to.frame(fr) # Give the React app a moment to paint inside this frame. end = time.time() + 3.0 while time.time() < end: if _has_probe(): return True time.sleep(0.2) # One level of nesting (Planner sometimes nests) nested = driver.find_elements(By.TAG_NAME, "iframe") if nested: for nfr in nested: try: driver.switch_to.frame(nfr) end2 = time.time() + 2.0 while time.time() < end2: if _has_probe(): return True time.sleep(0.2) driver.switch_to.parent_frame() except Exception: try: driver.switch_to.parent_frame() except Exception: pass driver.switch_to.default_content() except Exception: try: driver.switch_to.default_content() except Exception: pass return False for attempt in range(1, retries + 1): driver.switch_to.default_content() if _has_probe() or (_on_planner_plan_board(driver) and _board_probes_present(driver)): return True iframes = driver.find_elements(By.TAG_NAME, "iframe") if not iframes: time.sleep(1) continue # Prefer likely Planner frames first, but verify using probes. scored = [] for fr in iframes: src = (fr.get_attribute("src") or "").lower() title = (fr.get_attribute("title") or "").lower() name = (fr.get_attribute("name") or "").lower() portal = (fr.get_attribute("data-portal-node") or "").lower() score = 0 if "planner" in src or "planner" in title or "planner" in name: score += 3 if "myplans" in src or "/webui/" in src: score += 2 if portal == "true": score += 2 if "tasks" in src or "office" in src: score += 1 scored.append((score, fr)) scored.sort(key=lambda x: x[0], reverse=True) if _try_frames([fr for _, fr in scored]): _log.info("Switched into Planner iframe (attempt %d)", attempt) return True time.sleep(1) _log.warning("Planner iframe not detected after %d context attempts", retries) driver.switch_to.default_content() return False _LEFT_NAV_TOGGLE_LOCATORS = [ (By.XPATH, "//button[@aria-label='Left navigation pane']"), (By.XPATH, "//button[contains(@aria-label,'Left navigation pane')]"), (By.XPATH, "//button[contains(@aria-label,'navigation pane')]"), ] _MY_PLANS_NAV_LOCATORS = [ (By.XPATH, "//*[self::span or self::div][normalize-space()='My plans']"), (By.XPATH, "//*[contains(@aria-label,'My plans') or contains(@title,'My plans')]"), (By.XPATH, "//button[contains(.,'My plans')]"), (By.XPATH, "//a[contains(.,'My plans')]"), ] def _my_plans_nav_visible(driver: WebDriver) -> bool: """True when the left-nav My plans entry is visible and clickable.""" for by, sel in _MY_PLANS_NAV_LOCATORS: try: for el in driver.find_elements(by, sel): if el.is_displayed() and el.is_enabled(): return True except Exception: continue return False def _left_nav_pane_collapsed(driver: WebDriver) -> bool: """True when the Planner hamburger menu shows aria-expanded=false.""" for by, sel in _LEFT_NAV_TOGGLE_LOCATORS: try: for btn in driver.find_elements(by, sel): if not btn.is_displayed(): continue expanded = (btn.get_attribute("aria-expanded") or "").strip().lower() return expanded != "true" except Exception: continue return False def _ensure_left_nav_open(driver: WebDriver) -> None: """ Expand the left navigation pane when collapsed. On My tasks grid the hamburger (aria-label='Left navigation pane') is often closed — My plans is hidden until it is clicked. """ if _my_plans_nav_visible(driver) and not _left_nav_pane_collapsed(driver): return if _my_plans_nav_visible(driver): return log_step("Open left navigation pane") _log.info("Left nav collapsed — clicking navigation pane toggle before My plans") try: _click_any(driver, _LEFT_NAV_TOGGLE_LOCATORS, timeout=WAIT_SHORT) time.sleep(1) except TimeoutException: _log.warning("Left navigation pane toggle not found") return try: _wait(driver, WAIT_MED).until(_my_plans_nav_visible) log_step("Left navigation pane opened") except TimeoutException: _log.warning("My plans not visible after opening left navigation pane") def open_my_plans_shared(driver: WebDriver) -> None: """ Left nav: My plans -> Shared tab. This matches your current UI flow and avoids search-based plan opening. """ _recover_planner_error(driver) log_step("Navigate: My plans -> Shared") log_url(driver, "before") driver.switch_to.default_content() _switch_to_planner_context(driver) _ensure_left_nav_open(driver) log_step("Click: My plans") _click_any(driver, _MY_PLANS_NAV_LOCATORS, timeout=WAIT_LONG) time.sleep(2) shared_locators = [ (By.XPATH, "//*[self::button or self::div or self::span][normalize-space()='Shared']"), (By.XPATH, "//*[@role='tab' and normalize-space()='Shared']"), ] log_step("Click: Shared tab") _click_any(driver, shared_locators, timeout=WAIT_LONG) time.sleep(2) log_url(driver, "shared_list") log_step("Opened My plans -> Shared") def _facility_search_terms(facility_name: str) -> list[str]: """Filter queries to try: full name, then first 3 words, then first 2.""" name = (facility_name or "").strip() if not name: return [] words = name.split() terms: list[str] = [] seen: set[str] = set() def _add(term: str) -> None: t = term.strip() if not t: return key = t.lower() if key not in seen: seen.add(key) terms.append(t) _add(name) if len(words) >= 3: _add(" ".join(words[:3])) if len(words) >= 2: _add(" ".join(words[:2])) return terms def _shared_list_data_rows(driver: WebDriver) -> list: """Visible data rows on Shared plans grid (exclude header).""" rows: list = [] for row in driver.find_elements(By.CSS_SELECTOR, "[role='row']"): try: if not row.is_displayed(): continue txt = (row.text or "").strip() if not txt: continue low = txt.lower() if low.startswith("name") and "shared" in low: continue rows.append(row) except StaleElementReferenceException: continue return rows def _row_display_name(row) -> str: return ((row.text or "").strip().splitlines() or [""])[0].strip() def _row_matches_facility_search(row_name: str, facility_name: str, search_term: str) -> bool: """True when a filter/list row matches the facility we want.""" name = (row_name or "").strip().lower() if not name: return False full = (facility_name or "").strip().lower() term = (search_term or "").strip().lower() if name == full or full in name or name in full: return True if term and (term in name or name.startswith(term)): return True words = (facility_name or "").split() if len(words) >= 2: short2 = " ".join(words[:2]).lower() if short2 in name: return True if len(words) >= 3: short3 = " ".join(words[:3]).lower() if short3 in name: return True return False def _filter_shows_no_results(driver: WebDriver) -> bool: for xpath in _FILTER_EMPTY_STATE_XPATHS: try: for el in driver.find_elements(By.XPATH, xpath): if el.is_displayed(): return True except Exception: continue return len(_shared_list_data_rows(driver)) == 0 def _type_facility_filter(driver: WebDriver, query: str) -> None: """Type a keyword into the Shared plans filter box.""" try: _click_any(driver, _FILTER_BUTTON_LOCATORS, timeout=3) time.sleep(0.4) except TimeoutException: pass typed = False for by, sel in _FILTER_INPUT_LOCATORS: try: el = driver.find_element(by, sel) if el.is_displayed() and el.is_enabled(): _js_click(driver, el) el.send_keys(Keys.CONTROL, "a") el.send_keys(Keys.BACKSPACE) el.send_keys(query) try: el.send_keys(Keys.ENTER) except Exception: pass typed = True break except Exception: continue if not typed: _type_into(driver, _FILTER_INPUT_LOCATORS, query, clear=True) try: _first_clickable(driver, _FILTER_INPUT_LOCATORS, timeout=3).send_keys(Keys.ENTER) except Exception: pass time.sleep(1.2) try: _wait(driver, WAIT_MED).until( lambda d: bool(_shared_list_data_rows(d)) or _filter_shows_no_results(d) ) except TimeoutException: pass def _click_row_open_plan(driver: WebDriver, row) -> None: for xpath in ( ".//a[1]", ".//button[1]", ".//*[@role='gridcell']//a[1]", ".//*[@role='gridcell']//button[1]", ".//*[@role='gridcell']", ): try: el = row.find_element(By.XPATH, xpath) if el.is_displayed() and el.is_enabled(): _js_click(driver, el) return except Exception: continue ActionChains(driver).double_click(row).perform() def _open_facility_from_shared_list(driver: WebDriver, facility_name: str) -> None: """ From My plans -> Shared list, try clicking the facility directly from the list (scrolling if needed). If it cannot be found, fall back to Filter by keyword (type + Enter) and click the first row. After clicking a facility, wait 30s for the plan page. If it is still not ready, refresh and reopen the facility once. """ log_step("Open facility from Shared list", facility=facility_name) log_url(driver, "shared") _agent_log( "F1", "src/onedrive_planner.py:_open_facility_from_shared_list", "Enter", {"facility": facility_name[:80], "url": (driver.current_url or "")[:160]}, ) def _plan_loaded(d: WebDriver) -> bool: return _plan_page_ready(d) def _click_filter_result_row(search_term: str) -> bool: rows = _shared_list_data_rows(driver) if not rows: return False chosen = None for row in rows: name = _row_display_name(row) if _row_matches_facility_search(name, facility_name, search_term): chosen = row break if chosen is None: chosen = rows[0] row_name = _row_display_name(chosen) log_step( "Click filter result row", facility=facility_name, row=row_name[:60], query=search_term[:40], ) _agent_dbg_eb( "F3", "onedrive_planner:_open_facility_from_shared_list", "filter_row_click", {"row": row_name[:80], "query": search_term[:40]}, ) _click_row_open_plan(driver, chosen) time.sleep(0.5) return True def _try_filter_search() -> bool: for term in _facility_search_terms(facility_name): log_step("Filter by keyword", facility=facility_name, query=term[:60]) _type_facility_filter(driver, term) if _filter_shows_no_results(driver): log_step("Filter returned no results", query=term[:60], facility=facility_name) # A Planner error overlay may be blocking the search UI — try to recover if _recover_planner_error(driver): log_step("Recovered from error overlay during filter search", facility=facility_name) # Re-open My plans -> Shared after the page refreshed _click_any(driver, _MY_PLANS_NAV_LOCATORS, timeout=WAIT_LONG) time.sleep(2) shared_locators = [ (By.XPATH, "//*[self::button or self::div or self::span][normalize-space()='Shared']"), (By.XPATH, "//*[@role='tab' and normalize-space()='Shared']"), ] try: _click_any(driver, shared_locators, timeout=WAIT_LONG) time.sleep(2) except TimeoutException: pass _type_facility_filter(driver, term) if _filter_shows_no_results(driver): continue else: continue rows = _shared_list_data_rows(driver) log_step( "Filter results", rows=len(rows), query=term[:60], facility=facility_name, ) if rows and _click_filter_result_row(term): return True return False def _click_facility_from_list() -> bool: """ Try to click the facility in the visible Shared list without typing. Scrolls the grid to load more rows. """ target = facility_name.strip().lower() if not target: return False grid_body_locators = [ (By.XPATH, "//*[contains(@class,'DataGridBody')]"), (By.XPATH, "//*[@role='rowgroup']"), ] try: grid = _first_clickable(driver, grid_body_locators, timeout=5) except Exception: grid = None def _rows(): # Some builds don't expose aria-rowindex. Use role=row and filter out header-like rows. rows = driver.find_elements(By.CSS_SELECTOR, "[role='row']") filtered = [] for r in rows: try: if not r.is_displayed(): continue txt = (r.text or "").strip().lower() if not txt: continue # Skip header rows that contain column names if txt.startswith("name") and "shared" in txt: continue filtered.append(r) except Exception: continue return filtered for _ in range(30): for row in _rows(): try: if not row.is_displayed(): continue txt = (row.text or "").strip() if not txt: continue name = txt.splitlines()[0].strip() key = name.lower() if key in _PROCESSED_SHARED_FACILITIES: continue if not _row_matches_facility_search(name, facility_name, facility_name): continue log_step("Matched facility in Shared list", facility=facility_name, row=name) _agent_log( "F2", "src/onedrive_planner.py:_open_facility_from_shared_list", "Matched facility in list", {"facility": facility_name[:80], "row_name": name[:80]}, ) _click_row_open_plan(driver, row) _PROCESSED_SHARED_FACILITIES.add(key) return True except StaleElementReferenceException: continue try: if grid is not None: driver.execute_script("arguments[0].scrollTop = arguments[0].scrollTop + 700;", grid) else: driver.find_element(By.TAG_NAME, "body").send_keys(Keys.PAGE_DOWN) except Exception: pass time.sleep(0.6) return False driver.switch_to.default_content() ctx_ok = _switch_to_planner_context(driver) _agent_log( "F1", "src/onedrive_planner.py:_open_facility_from_shared_list", "After switch_to_planner_context", {"ok": bool(ctx_ok), "url": (driver.current_url or "")[:160]}, ) def _click_facility_open() -> None: opened = False if _click_facility_from_list(): log_step("Opened facility from visible list", facility=facility_name) opened = True elif _try_filter_search(): log_step("Opened facility from filter search", facility=facility_name) opened = True if not opened: tried = _facility_search_terms(facility_name) log_step( "Facility not appear in Shared search", facility=facility_name, queries="; ".join(tried[:3]), ) _agent_dbg_eb( "F4", "onedrive_planner:_open_facility_from_shared_list", "not_appear", {"facility": facility_name[:80], "queries": tried}, ) raise FacilityNotFoundError(f"{FACILITY_NOT_APPEAR_MSG}: {facility_name}") print(f"[MISF] Facility clicked: {facility_name}", flush=True) log_step("Facility clicked", facility=facility_name) def _wait_after_facility_click() -> None: print( f"[MISF] Waiting up to {POST_FACILITY_CLICK_WAIT}s for plan page to load " f"({facility_name})...", flush=True, ) log_step( "Wait after facility click", facility=facility_name, seconds=POST_FACILITY_CLICK_WAIT, ) remaining = POST_FACILITY_CLICK_WAIT poll_secs = 2 while remaining > 0: if _plan_load_error_present(driver): log_step("Plan load error during wait", facility=facility_name) _click_plan_try_again(driver) if _plan_loaded(driver): waited = POST_FACILITY_CLICK_WAIT - remaining print( f"[MISF] Plan page loaded after {waited}s (stopped early).", flush=True, ) log_step( "Plan loaded during wait", facility=facility_name, waited_secs=waited, ) return chunk = min(poll_secs, remaining) time.sleep(chunk) remaining -= chunk if remaining > 0: print(f"[MISF] ... {remaining}s remaining", flush=True) print(f"[MISF] Done waiting {POST_FACILITY_CLICK_WAIT} seconds.", flush=True) log_step("Finished wait after facility click", facility=facility_name, seconds=POST_FACILITY_CLICK_WAIT) def _finish_if_plan_ready() -> bool: if _plan_load_error_present(driver): log_step("Plan load error before finish", facility=facility_name) _click_plan_try_again(driver) time.sleep(4) if not _plan_loaded(driver): focus_facility_plan_tab(driver) if _plan_load_error_present(driver): _click_plan_try_again(driver) time.sleep(4) if not _plan_loaded(driver): return False focus_facility_plan_tab(driver) log_url(driver, "plan_loaded") log_step("Plan loaded", facility=facility_name) time.sleep(POST_FACILITY_LOAD_DELAY) log_step("Post-load stabilize complete", seconds=POST_FACILITY_LOAD_DELAY) return True def _refresh_and_reopen_facility() -> None: log_step("Plan still not ready — refreshing page", facility=facility_name) _log.warning("Plan not ready after %ss wait — refreshing: %s", POST_FACILITY_CLICK_WAIT, facility_name) driver.refresh() time.sleep(5) driver.switch_to.default_content() _switch_to_planner_context(driver) open_my_plans_shared(driver) _click_facility_open() _wait_after_facility_click() # Primary: click from list (scroll). Fallback: filter + first row. try: _click_facility_open() except Exception: log_step("Facility open failed — will refresh and retry", facility=facility_name) _log.warning("Facility open failed; refreshing and retrying: %s", facility_name) driver.refresh() time.sleep(5) driver.switch_to.default_content() _switch_to_planner_context(driver) open_my_plans_shared(driver) _click_facility_open() _wait_after_facility_click() if _finish_if_plan_ready(): return _refresh_and_reopen_facility() if _finish_if_plan_ready(): log_step("Plan loaded after refresh", facility=facility_name) return log_step("Wait for plan after refresh", timeout_secs=FACILITY_LOAD_AFTER_REFRESH) try: _wait(driver, FACILITY_LOAD_AFTER_REFRESH).until(_plan_loaded) time.sleep(POST_FACILITY_LOAD_DELAY) log_step("Plan loaded after refresh (waited)", facility=facility_name) except TimeoutException as exc: raise TimeoutException( f"Plan did not load for facility '{facility_name}' after " f"{POST_FACILITY_CLICK_WAIT}s wait + refresh" ) from exc def _ensure_board_view(driver: WebDriver) -> None: board_locators = [ (By.XPATH, "//*[@role='tab' and normalize-space()='Board']"), (By.XPATH, "//*[self::button or self::div][normalize-space()='Board']"), ] try: log_step("Switch to Board view") _click_any(driver, board_locators, timeout=8) time.sleep(2) log_url(driver, "board") except TimeoutException: log_step("Board tab not found (continuing on current view)") def _wait_for_board_loaded(driver: WebDriver, timeout: int = 20) -> None: """Switch to Board tab and wait until board content is visible before screenshotting.""" board_locators = [ (By.XPATH, "//*[@role='tab' and normalize-space()='Board']"), (By.XPATH, "//*[self::button or self::div][normalize-space()='Board']"), ] # Click the Board tab (ignore if already on board or tab not found) try: tab = _first_clickable(driver, board_locators, timeout=8) # Only click if the tab isn't already selected selected = (tab.get_attribute("aria-selected") or "").lower() if selected != "true": log_step("Switch to Board view") _js_click(driver, tab) else: log_step("Board tab already active") except TimeoutException: log_step("Board tab not found (continuing on current view)") # Poll until board content is visible (task cards, bucket columns, or Add task) content_xpaths = [ "//button[@data-testid='attachment-indicator']", "//*[@role='checkbox']", "//label[@role='checkbox']", "//*[contains(@data-testid,'TaskCard') or contains(@class,'TaskCard')]", "//*[@role='list' and ancestor::*[@role='group']]", "//*[normalize-space()='Add task' and ancestor::*[@role='group']]", "//*[normalize-space()='Add task']", ] log_step("Waiting for board content to load") deadline = time.time() + timeout while time.time() < deadline: for xpath in content_xpaths: try: for el in driver.find_elements(By.XPATH, xpath)[:6]: if el.is_displayed(): log_step("Board content visible — ready for screenshot") return except Exception: continue time.sleep(0.5) log_step("Board content wait timed out — proceeding with screenshot anyway") def _ensure_grid_view(driver: WebDriver) -> None: grid_locators = [ (By.CSS_SELECTOR, "button#gridViewPivot"), (By.XPATH, "//*[@role='tab' and normalize-space()='Grid']"), (By.XPATH, "//button[@value='grid' and @id='gridViewPivot']"), (By.XPATH, "//*[self::button or self::div][normalize-space()='Grid']"), ] try: log_step("Switch to Grid view") _click_any(driver, grid_locators, timeout=8) time.sleep(PANEL_TAB_LOAD_DELAY) log_url(driver, "grid") except TimeoutException: log_step("Grid tab not found (continuing on current view)") def _status_column_header_locators() -> list[tuple[str, str]]: return [ (By.CSS_SELECTOR, "#ColumnHeaderCell_status"), (By.CSS_SELECTOR, "#ColumnHeaderCell_progress"), (By.XPATH, "//div[@role='columnheader' and @aria-label='Status']"), (By.XPATH, "//div[@role='columnheader' and @aria-label='Progress']"), ] def _is_status_sorted_descending(header) -> bool: sort = (header.get_attribute("aria-sort") or "").strip().lower() return sort == "descending" def _sort_descending_menu_selected(driver: WebDriver) -> bool: """True when the column header menu shows Sort descending as active.""" probes = [ "//button[@name='Sort descending' and @aria-checked='true']", "//button[@name='Sort descending' and contains(@class,'is-checked')]", "//button[@name='Sort descending']//*[@data-icon-name='CheckMark']", ( "//button[@name='Sort descending']" "/ancestor::li[contains(@class,'is-checked') or contains(@class,'is-selected')]" ), ] for xpath in probes: try: for el in driver.find_elements(By.XPATH, xpath): if el.is_displayed(): return True except StaleElementReferenceException: continue return False def _dismiss_open_menu(driver: WebDriver) -> None: try: driver.find_element(By.TAG_NAME, "body").send_keys(Keys.ESCAPE) time.sleep(0.3) except Exception: pass def _ensure_status_sort_descending(driver: WebDriver) -> None: """Open Status column menu and apply Sort Descending when not already set.""" log_step("Ensure Status column sorted descending") try: header = _first_clickable(driver, _status_column_header_locators(), timeout=8) except TimeoutException: log_step("Status column header not found (continuing on current sort)") return if _is_status_sorted_descending(header): log_step("Status column already sorted descending") return _js_click(driver, header) time.sleep(0.5) if _sort_descending_menu_selected(driver): log_step("Status column already sorted descending (menu check)") _dismiss_open_menu(driver) return sort_locators = [ # Exact match — actual Planner DOM uses lowercase 'd' (By.XPATH, "//button[@name='Sort descending' and @role='menuitem']"), (By.XPATH, "//ul[contains(@class,'ContextualMenu')]//button[normalize-space()='Sort descending']"), (By.XPATH, "//button[@role='menuitem' and normalize-space()='Sort descending']"), (By.XPATH, "//span[normalize-space()='Sort descending']/ancestor::button[@role='menuitem']"), # Uppercase fallback (in case Planner localises differently) (By.XPATH, "//button[@name='Sort Descending' and @role='menuitem']"), (By.XPATH, "//button[@role='menuitem' and normalize-space()='Sort Descending']"), # Broad contains — catches any capitalisation (By.XPATH, "//*[@role='menuitem' and contains(normalize-space(),'Sort descending')]"), (By.XPATH, "//*[@role='menuitem' and contains(normalize-space(),'Sort Descending')]"), (By.XPATH, "//li[contains(@class,'ms-ContextualMenu-item')]//button[contains(normalize-space(),'escending')]"), ] try: _click_any(driver, sort_locators, timeout=5) time.sleep(1) try: header = _first_clickable(driver, _status_column_header_locators(), timeout=3) if _is_status_sorted_descending(header): log_step("Status column sorted descending") else: log_step("Status column sort descending applied (aria-sort not confirmed)") except TimeoutException: log_step("Status column sort descending clicked") except TimeoutException: log_step("Sort descending menu item not found via XPath — trying JS click") clicked = False try: clicked = driver.execute_script( """ var items = Array.from(document.querySelectorAll( '[role="menuitem"], [role="menuitem"] button, button[role="menuitem"]' )); var target = items.find(function(el) { var t = (el.innerText || el.textContent || '').trim().toLowerCase(); return t === 'sort descending'; }); if (target) { target.click(); return true; } return false; """ ) except Exception: pass if clicked: time.sleep(1) log_step("Sort descending clicked via JS") else: log_step("Sort descending not found (continuing on current sort)") _dismiss_open_menu(driver) def _grid_probes_present(driver: WebDriver) -> bool: probes = ( "[role='grid'].grid-container", ".grid-container[role='grid']", "div.grid-container", "button#gridViewPivot", ) for sel in probes: for el in driver.find_elements(By.CSS_SELECTOR, sel): if el.is_displayed(): return True return False def _grid_scrollable_element(driver: WebDriver): for sel in ( ".grid-virtualized-container", "[role='grid'].grid-container", ".grid-container[role='grid']", ): for el in driver.find_elements(By.CSS_SELECTOR, sel): if el.is_displayed(): return el return None def _grid_visible_data_rows(driver: WebDriver) -> list: rows: list = [] for row in driver.find_elements(By.CSS_SELECTOR, "div[role='row'][aria-rowindex]"): try: if not row.is_displayed(): continue idx_raw = row.get_attribute("aria-rowindex") or "" if idx_raw.isdigit() and int(idx_raw) < 1: continue rows.append(row) except StaleElementReferenceException: continue return rows def _parse_status_blob(blob: str) -> str | None: low = (blob or "").strip().lower() if not low: return None if "not started" in low or low in ("notstarted", "not_started"): return "not started" if "in progress" in low or low in ("inprogress", "in_progress"): return "in progress" if "completed" in low or low == "completed": return "completed" return None def _grid_row_status(row) -> str | None: """Read Status column: not started / in progress / completed.""" try: for cell in row.find_elements(By.XPATH, ".//*[@role='gridcell']"): aria = (cell.get_attribute("aria-label") or "").strip() if not aria.lower().startswith("status"): continue for btn in cell.find_elements(By.XPATH, ".//button[@role='combobox']"): if not btn.is_displayed(): continue parsed = _parse_status_blob( (btn.get_attribute("value") or "") + " " + (btn.text or "") ) if parsed: return parsed parsed = _parse_status_blob((cell.text or "") + " " + aria) if parsed: return parsed for btn in row.find_elements(By.XPATH, ".//button[@role='combobox']"): if not btn.is_displayed(): continue parsed = _parse_status_blob( (btn.get_attribute("value") or "") + " " + (btn.text or "") ) if parsed: return parsed for cell in row.find_elements(By.XPATH, ".//*[@role='gridcell']"): parsed = _parse_status_blob(cell.text or "") if parsed: return parsed except StaleElementReferenceException: return None return None def _grid_row_task_name(row) -> str: try: for cell in row.find_elements(By.XPATH, ".//*[@role='gridcell']"): aria = (cell.get_attribute("aria-label") or "").strip() if aria.lower().startswith("task name"): # e.g. "Task Name Missing Insurance Info" prefix = "task name" rest = _clean_grid_task_title(aria[len(prefix) :].strip()) if rest: return rest for span in cell.find_elements(By.XPATH, ".//span[normalize-space()]"): text = _clean_grid_task_title((span.text or "").strip()) if not text: continue low = text.lower() if low in ("not started", "in progress", "completed", "to do", "important", "medium"): continue if len(text) > 2: return text for line in (row.text or "").splitlines(): line = _clean_grid_task_title(line.strip()) if line and line.lower() not in ("not started", "in progress", "completed", "to do"): return line except StaleElementReferenceException: return "" return "" def _grid_task_name_cell(row): try: for cell in row.find_elements(By.XPATH, ".//*[@role='gridcell']"): aria = (cell.get_attribute("aria-label") or "").lower() if aria.startswith("task name"): return cell cells = row.find_elements(By.XPATH, ".//*[@role='gridcell']") return cells[0] if cells else None except StaleElementReferenceException: return None def _open_grid_row_details(driver: WebDriver, row) -> bool: """Hover task name, click Open details icon.""" try: name_cell = _grid_task_name_cell(row) if name_cell is None: return False _hover_element(driver, name_cell) time.sleep(0.6) open_xpaths = ( ".//button[contains(@aria-label,'Open details')]", ".//*[@data-testid='quick-actions-wrapper']//button[1]", ".//button[contains(@aria-label,'details')]", ) for xpath in open_xpaths: for btn in row.find_elements(By.XPATH, xpath): if not btn.is_displayed(): continue _js_click(driver, btn) time.sleep(PANEL_TAB_LOAD_DELAY) if _task_details_panel_open(driver): time.sleep(PANEL_TAB_LOAD_DELAY) log_step("Opened task via grid Open details") return True for btn in name_cell.find_elements(By.XPATH, ".//button"): if not btn.is_displayed(): continue aria = (btn.get_attribute("aria-label") or "").lower() if "open details" in aria or aria == "open details (alt+i)": _js_click(driver, btn) time.sleep(PANEL_TAB_LOAD_DELAY) if _task_details_panel_open(driver): time.sleep(PANEL_TAB_LOAD_DELAY) log_step("Opened task via grid Open details (name cell)") return True except StaleElementReferenceException: return False return False def _scroll_grid_to_top(driver: WebDriver) -> None: scroll_el = _grid_scrollable_element(driver) if scroll_el is not None: driver.execute_script("arguments[0].scrollTop = 0;", scroll_el) time.sleep(0.4) def _scroll_grid_down(driver: WebDriver, pixels: int = 450) -> None: scroll_el = _grid_scrollable_element(driver) if scroll_el is not None: driver.execute_script("arguments[0].scrollTop += arguments[1];", scroll_el, pixels) else: driver.execute_script("window.scrollBy(0, arguments[0]);", pixels) time.sleep(0.4) def _scan_grid_not_started_titles(driver: WebDriver, *, max_scrolls: int = 100) -> list[tuple[str, int]]: """Scroll the virtualized grid and collect (title, aria-rowindex) for Not started tasks. Uses aria-rowindex as the unique key so duplicate task names are all captured. """ _scroll_grid_to_top(driver) seen_indices: set[int] = set() tasks: list[tuple[str, int]] = [] stagnant = 0 for _ in range(max_scrolls): new_this_pass = 0 for row in _grid_visible_data_rows(driver): status = _grid_row_status(row) if status != "not started": continue idx_raw = row.get_attribute("aria-rowindex") or "" row_idx = int(idx_raw) if idx_raw.isdigit() else -1 if row_idx < 1 or row_idx in seen_indices: continue title = _grid_row_task_name(row).strip() if not title or _is_weak_task_title_key(title.lower()): continue seen_indices.add(row_idx) tasks.append((title, row_idx)) new_this_pass += 1 if new_this_pass == 0: stagnant += 1 if stagnant >= 4: break else: stagnant = 0 _scroll_grid_down(driver) _scroll_grid_to_top(driver) log_step("Grid not-started tasks scanned", count=len(tasks)) return tasks def _find_grid_row_by_index(driver: WebDriver, row_index: int, *, max_scrolls: int = 100): """Find a grid row by its aria-rowindex — stable even when multiple rows share the same title.""" if row_index < 1: return None _scroll_grid_to_top(driver) for _ in range(max_scrolls): for row in _grid_visible_data_rows(driver): idx_raw = row.get_attribute("aria-rowindex") or "" if idx_raw.isdigit() and int(idx_raw) == row_index: return row _scroll_grid_down(driver) return None def _find_grid_row_by_title(driver: WebDriver, title: str, *, max_scrolls: int = 100): want = _clean_grid_task_title(title).strip().lower() if not want: return None _scroll_grid_to_top(driver) for _ in range(max_scrolls): for row in _grid_visible_data_rows(driver): name = _grid_row_task_name(row).strip().lower() if name == want or (name and (want in name or name in want)): return row _scroll_grid_down(driver) return None def _panel_task_status(driver: WebDriver) -> str | None: """Read Not started / In progress / Completed from the open task panel.""" _task_details_tab_locators = [ (By.XPATH, "//*[@role='tab' and normalize-space()='Task details']"), (By.XPATH, "//*[@role='tab' and contains(normalize-space(),'Task details')]"), (By.XPATH, "//button[normalize-space()='Task details']"), (By.XPATH, "//*[self::button or self::div][normalize-space()='Task details']"), ] panel_status = None for _ps_round in range(2): try: _click_any(driver, _task_details_tab_locators, timeout=WAIT_SHORT) time.sleep(PANEL_TAB_LOAD_DELAY) except Exception: log_step("Task details tab not clickable for status check") for _ps_attempt in range(4): panel_status = _task_details_progress_status(driver) if panel_status: break time.sleep(1.0) if panel_status: break return panel_status def _ensure_board_scan_context(driver: WebDriver) -> str: """ Leave WebDriver focused on the document/frame that contains the plan board. planner.cloud.microsoft plan/board often renders in default_content, not an iframe. """ driver.switch_to.default_content() if _board_probes_present(driver): log_step("Board scan context", frame="default_content") return "default_content" if _switch_to_planner_context(driver) and _board_probes_present(driver): log_step("Board scan context", frame="planner_iframe") return "planner_iframe" driver.switch_to.default_content() for idx, fr in enumerate(driver.find_elements(By.TAG_NAME, "iframe")): try: driver.switch_to.default_content() driver.switch_to.frame(fr) if _board_probes_present(driver): log_step("Board scan context", frame=f"iframe_{idx}") return f"iframe_{idx}" except Exception: continue driver.switch_to.default_content() log_step("Board scan context", frame="default_content_fallback") return "default_content_fallback" def _find_in_best_board_context(driver: WebDriver, xpath: str) -> list: """Search default content and all iframes; stay in the frame with the most matches.""" best: list = [] best_switch = None def _count_in_current() -> list: visible = [] for el in driver.find_elements(By.XPATH, xpath): try: if el.is_displayed(): visible.append(el) except StaleElementReferenceException: continue return visible driver.switch_to.default_content() default_hits = _count_in_current() if len(default_hits) >= len(best): best = default_hits best_switch = lambda: driver.switch_to.default_content() driver.switch_to.default_content() for idx, fr in enumerate(driver.find_elements(By.TAG_NAME, "iframe")): try: driver.switch_to.default_content() driver.switch_to.frame(fr) hits = _count_in_current() if len(hits) > len(best): def _make_switch(f=fr): def _sw(): driver.switch_to.default_content() driver.switch_to.frame(f) return _sw best = hits best_switch = _make_switch() except Exception: continue if best_switch is not None: best_switch() return best def _wait_for_board_ready(driver: WebDriver, timeout: float = 45.0) -> None: """Wait until task cards or attachment indicators appear on the board.""" _ensure_board_scan_context(driver) deadline = time.time() + timeout while time.time() < deadline: if _find_in_best_board_context( driver, "//button[@data-testid='attachment-indicator']", ): return if _find_in_best_board_context(driver, "//label[@role='checkbox' and @aria-checked='false']"): return if _find_in_best_board_context(driver, "//*[@role='checkbox' and @aria-checked='false']"): return if driver.find_elements(By.CSS_SELECTOR, "[data-testid='task-board-column-body']"): return time.sleep(0.5) log_step("Board task elements not visible yet (continuing scan)") _TASK_CARD_ANCESTOR_XPATHS = [ *_PLANNER_TASK_CARD_XPATHS, # UiPath: single card = checkbox + title textbox (tightest match). "./ancestor::div[(.//label[@role='checkbox'] or .//button[@role='checkbox']) and .//*[@role='textbox']][1]", "./ancestor::button[(.//label[@role='checkbox'] or .//button[@role='checkbox']) and .//*[@role='textbox']][1]", "./ancestor::div[.//button[@data-testid='attachment-indicator']][(.//label[@role='checkbox'] or .//button[@role='checkbox'])][1]", "./ancestor::div[(.//label[@role='checkbox'] or .//button[@role='checkbox']) and (.//*[@role='textbox'] or .//*[contains(@class,'textbox')])][1]", "./ancestor::div[.//button[@data-testid='attachment-indicator']][1]", "./ancestor::*[contains(@data-testid,'task')][1]", "./ancestor::button[contains(@data-testid,'task') or contains(@class,'Task')][1]", "./ancestor::div[contains(@class,'TaskCard') or contains(@class,'taskCard')][1]", "./ancestor::div[contains(@class,'baseCard')][1]", # Sibling-layout: checkbox and indicator share a parent/grandparent without nesting # (occurs when task-card is a