commit c2dec96b1fa449a43140204496dab0ca2c57d796 Author: lubna-fatima Date: Tue Jun 16 14:55:29 2026 +0530 Initial commit: MISF VOB download and upload automation Co-Authored-By: Claude Sonnet 4.6 diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..17c91d7 --- /dev/null +++ b/.env.example @@ -0,0 +1,23 @@ +# Copy to .env and fill in (do not commit .env) +ONEDRIVE_USERNAME=your.email@example.com +ONEDRIVE_PASSWORD=your_password +MICROSOFT_TOTP_SECRET=your_base32_authenticator_secret + +# Browser: uses chrome-win\chrome-win\chrome.exe by default (bundled Chromium) +# MISF_USE_INSTALLED_CHROME=1 # skip chrome-win, use system Chromium +# CHROMIUM_PATH=C:\full\path\to\chrome.exe +# MISF_BROWSER_FOREGROUND=1 # show browser (default: minimized background) +# MISF_ALLOW_CHROME=1 # allow Google Chrome fallback + +# Optional — Planner grid URL (tenant tid= in query string) +# PLANNER_URL=https://planner.cloud.microsoft/webui/mytasks/assignedtome/view/grid?tid=... + +# Optional +# ONEDRIVE_URL=https://... +# PDF_FOLDER_PATH=C:\path\to\Input\PDFs +# MISF_DEBUG=1 +# MISF_LOG_LEVEL=DEBUG +# PROXY_IP= +# PROXY_PORT= +# PROXY_USERNAME= +# PROXY_PASSWORD= diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..3111959 --- /dev/null +++ b/.gitignore @@ -0,0 +1,41 @@ +# Build artifacts +build/ +dist/ +_internal/ +*.exe +*.spec + +# Chrome binary +chrome-win/ + +# Input/Output data folders +Input/ +Output/ + +# Python cache +__pycache__/ +*.pyc +*.pyo +*.pyd +.Python +*.egg-info/ +.eggs/ + +# Virtual environments +venv/ +env/ +.env +.venv/ + +# IDE +.vscode/ +.idea/ +*.swp + +# Logs +*.log +logs/ + +# OS +.DS_Store +Thumbs.db diff --git a/README.md b/README.md new file mode 100644 index 0000000..98a708f --- /dev/null +++ b/README.md @@ -0,0 +1,228 @@ +# OneDrive / Planner download-only automation + +Automates the MISF VOB **face-sheet download** workflow: sign in to OneDrive, open Planner, search each facility plan, download attachment PDFs, and mark tasks **In progress**. + +Reference: [`docs/MISF-VOB-Planner-Workflow.docx`](docs/MISF-VOB-Planner-Workflow.docx) + +This repository contains **only** OneDrive / Planner download automation. There is no Noridian Medicare, Medi-Cal, PDF extraction, or eligibility verification code. + +--- + +## Source requirements + +| Doc step | Action | +|----------|--------| +| 1 | Open OneDrive URL, enter username / password, complete MFA (Google Authenticator 6-digit code) | +| 2–4 | Open **Planner** (direct URL in config; default: Assigned to me grid) | +| 5 | For each facility: search/select plan → open tasks → **download attachment PDFs** → set task status **In progress** (board card, or Task details → Status) | + +**Clarifications:** + +- Process **54 facilities** by typing each name into Planner’s **search box**, one at a time. +- MFA via **TOTP secret** in `Input/input.xlsx` (automated with `pyotp`). + +**Not in this repo (upload path from the MISF guide):** + +- Uploading completed work to OneDrive +- Marking tasks **Completed** after eligibility verification + +```mermaid +flowchart TD + start[run_download_workflow] --> config[Read input.xlsx] + config --> chrome[start_chrome_debug + attach_browser] + chrome --> login[Microsoft login + TOTP] + login --> planner[Open Planner / In my plans] + planner --> loop{For each facility} + loop --> search[Search facility in Planner] + search --> dl[Download task attachments] + dl --> status[Mark tasks In progress] + status --> loop + loop --> summary[Log summary + Excel report] +``` + +--- + +## Repository layout + +| File | Role | +|------|------| +| [`run_download.py`](run_download.py) | Main entry point | +| [`src/planner_download.py`](src/planner_download.py) | Orchestrator: config → Chrome → login → facility loop → summary | +| [`src/onedrive_planner.py`](src/onedrive_planner.py) | Selenium: Microsoft login, Planner, downloads, In-Process | +| [`src/config.py`](src/config.py) | `read_app_config()`, `read_facility_list()` | +| [`src/totp_utils.py`](src/totp_utils.py) | TOTP secret sanitization / validation | +| [`scripts/validate_config.py`](scripts/validate_config.py) | Pre-flight config check (no browser) | +| [`src/chrome.py`](src/chrome.py) | Chrome debug startup, Selenium attach, proxy auth | +| [`scripts/bootstrap_input_xlsx.py`](scripts/bootstrap_input_xlsx.py) | Create config template | +| [`tests/test_download_config.py`](tests/test_download_config.py) | Config smoke tests | + +``` +. +├── README.md +├── requirements.txt +├── run_download.py +├── docs/ +├── Input/ +│ ├── input.xlsx.example +│ ├── input.xlsx # gitignored +│ └── PDFs/ +├── Output/ +├── scripts/ +├── tests/ +└── src/ + ├── planner_download.py + ├── onedrive_planner.py + ├── config.py + └── chrome.py +``` + +--- + +## Configuration (`Input/input.xlsx`) + +```powershell +copy Input\input.xlsx.example Input\input.xlsx +copy .env.example .env +``` + +Fill credentials in **either** place: + +- **`Input/input.xlsx`** → Sheet1, column **B** (`OneDrive Username`, `OneDrive Password`, `Microsoft TOTP Secret`), **or** +- **`.env`** in the project root (`ONEDRIVE_USERNAME`, `ONEDRIVE_PASSWORD`, `MICROSOFT_TOTP_SECRET`) + +Check config before running: + +```powershell +python scripts\validate_config.py +``` + +### Sheet `Sheet1` + +| Key | Purpose | +|-----|---------| +| `OneDrive Username` | Microsoft login email | +| `OneDrive Password` | Password | +| `Microsoft TOTP Secret` | Base32 authenticator secret (`pyotp`) | +| `OneDrive URL` | OneDrive landing URL | +| `pdf_folder_path` | Download root (default: `Input/PDFs`) | +| `Proxy IP` / `Proxy Port` / `Proxy Username` / `Proxy Password` | Optional | + +### Sheet `Facilities` — **put all 54 facility names here** + +Open **`Input/input.xlsx`** (not the `.example` file) and edit the **`Facilities`** sheet: + +| Column | Purpose | +|--------|---------| +| `Facility Name` | **One row per facility** (54 rows). Use the exact text you type in Planner search. | + +Example: + +| Facility Name | +|---------------| +| Amaya | +| Brighton Place San Diego | +| … (52 more rows) | + +The bot reads this sheet via `read_facility_list()` in `src/config.py`. For a first test, keep only **2 rows** and run `python run_download.py --limit 2`. + +--- + +## Production `.exe` (double-click) + +Build a Windows folder you can copy to any PC: + +```powershell +powershell -ExecutionPolicy Bypass -File scripts\build_production_exe.ps1 +``` + +Output at **project root**: `MISF_VOB_Download.exe` and `_internal\` (same folder as `Input/`, `Output/`, `run_download.py`). + +**Deploy:** zip the project folder (omit `.venv`, `build`, `src` if you only ship the exe build). On the target machine: + +1. Unzip anywhere +2. Copy `Input\input.xlsx.example` → `Input\input.xlsx` and configure credentials + facilities +3. Double-click `MISF_VOB_Download.exe` + +Config, `Input/`, and `Output/` are read from the **project root** (folder containing the `.exe`). See [`README_PRODUCTION.txt`](README_PRODUCTION.txt) after build. + +--- + +## Quick start + +```powershell +python -m venv .venv +.\.venv\Scripts\Activate.ps1 +pip install -r requirements.txt +copy Input\input.xlsx.example Input\input.xlsx +# Edit Input\input.xlsx + +python run_download.py --limit 2 # dry-run (first N facilities) +python run_download.py --facility "Amaya" # single facility +python run_download.py --resume # continue after last checkpoint +python run_download.py --keep-browser # leave browser open when done +python run_download.py # full run +``` + +**Outputs:** + +- `Input/PDFs//*.pdf` +- `Output//Summary/Download_Summary_.xlsx` (includes run timestamp and Planner URL) +- `Output//logs/download.log` +- `Output//checkpoint.json` (while a run is in progress; cleared on success) + +Set `MISF_DEBUG=1` in `.env` to enable NDJSON debug traces (`debug-fc545e.log`). + +--- + +## Tests + +```powershell +python -m pytest +python scripts\validate_config.py +``` + +--- + +## Troubleshooting + +| Issue | What to check | +|-------|----------------| +| Login / MFA fails | TOTP secret; system clock (NTP) | +| Facility not found | `Facility Name` matches Planner search | +| No PDFs downloaded | Attachments exist; tune `src/onedrive_planner.py` selectors | +| Proxy errors | Proxy IP/Port in `input.xlsx`; profile `%USERPROFILE%\Automation\ChromiumProfile_MISF_Download` | +| Browser not found | Install [Chromium](https://www.chromium.org/getting-involved/download-chromium/) or set `CHROMIUM_PATH` in `.env` (Google Chrome only with `MISF_ALLOW_CHROME=1`) | +| `ChromeDriver only supports Chrome version 148` / attach fails | Install Chromium matching driver version; driver auto-downloads to `.drivers/` | + +--- + +## Browser + +The automation uses **Chromium only** (minimized in the background by default). It looks for, in order: + +1. `CHROMIUM_PATH` or `BROWSER_PATH` environment variable +2. **`chrome-win/chrome-win/chrome.exe`** (portable bundle at project root — default for production) +3. Bundled **`chromedriver.exe`** in the same folder (no download needed) +4. System Chromium under `Program Files\Chromium`, `%LOCALAPPDATA%\Chromium`, Scoop (`MISF_USE_INSTALLED_CHROME=1` to prefer this over `chrome-win`) +5. Google Chrome only when `MISF_ALLOW_CHROME=1` + +Set `MISF_BROWSER_FOREGROUND=1` to show/maximize the browser window. The production `.exe` runs without a console window; see `Output//logs/download.log`. + +**Chromedriver:** `webdriver-manager` downloads a matching driver. If attach fails, install Chromium that matches your driver or set `CHROMIUM_PATH` explicitly. + +User profile directory: `%USERPROFILE%\Automation\ChromiumProfile_MISF_Download` + +--- + +## UiPath migration + +An equivalent **UiPath Studio (Modern)** project lives in [`uipath/MISF_Planner_Download/`](uipath/MISF_Planner_Download/). Open `project.json` in UiPath Studio. Pre-flight check: + +```powershell +powershell -File scripts\validate_uipath_pilot.ps1 +python -m pytest tests\test_uipath_pilot.py -v +``` + +--- + +Internal MISF / Access Psychology automation. diff --git a/README_PRODUCTION.txt b/README_PRODUCTION.txt new file mode 100644 index 0000000..6321945 --- /dev/null +++ b/README_PRODUCTION.txt @@ -0,0 +1,42 @@ +MISF VOB Download — Production (Windows) +========================================= + +Double-click MISF_VOB_Download.exe to run the full facility download workflow. + +FIRST-TIME SETUP +---------------- +1. Copy Input\input.xlsx.example to Input\input.xlsx +2. Open Input\input.xlsx and fill: + - Sheet1: OneDrive Username, Password, Microsoft TOTP Secret + - Facilities: one facility name per row (Planner search text) +3. Keep chrome-win\chrome-win\ (chrome.exe + chromedriver.exe) in this folder. + Browser starts minimized in background. No separate Chromium install required. +4. Optional: copy .env.example to .env for credentials instead of Excel + +RUN +--- +- Double-click MISF_VOB_Download.exe +- A console window shows progress; press Enter when finished to close +- For scheduled tasks (no pause): set environment variable MISF_NO_PAUSE=1 + +OUTPUT +------ +- Output\\Original\\ — downloaded PDFs +- Output\\Summary\Download_Summary_.xlsx +- Output\\logs\download.log + +OPTIONAL COMMAND-LINE (run from cmd in this folder) +--------------------------------------------------- + MISF_VOB_Download.exe --limit 2 + MISF_VOB_Download.exe --facility "Amaya" + MISF_VOB_Download.exe --resume + +REQUIREMENTS +------------ +- Windows 10/11 +- Network access to Microsoft 365 / Planner +- chrome-win\ folder with portable Chromium (or system Chromium / CHROMIUM_PATH) +- Progress log: Output\\logs\download.log (no console window when using .exe) + +Keep MISF_VOB_Download.exe, _internal, and chrome-win together in the project root +(same directory as Input and Output). diff --git a/docs/MISF-VOB-Planner-Workflow.docx b/docs/MISF-VOB-Planner-Workflow.docx new file mode 100644 index 0000000..0757bb6 Binary files /dev/null and b/docs/MISF-VOB-Planner-Workflow.docx differ diff --git a/docs/PRODUCTION_README.txt b/docs/PRODUCTION_README.txt new file mode 100644 index 0000000..6321945 --- /dev/null +++ b/docs/PRODUCTION_README.txt @@ -0,0 +1,42 @@ +MISF VOB Download — Production (Windows) +========================================= + +Double-click MISF_VOB_Download.exe to run the full facility download workflow. + +FIRST-TIME SETUP +---------------- +1. Copy Input\input.xlsx.example to Input\input.xlsx +2. Open Input\input.xlsx and fill: + - Sheet1: OneDrive Username, Password, Microsoft TOTP Secret + - Facilities: one facility name per row (Planner search text) +3. Keep chrome-win\chrome-win\ (chrome.exe + chromedriver.exe) in this folder. + Browser starts minimized in background. No separate Chromium install required. +4. Optional: copy .env.example to .env for credentials instead of Excel + +RUN +--- +- Double-click MISF_VOB_Download.exe +- A console window shows progress; press Enter when finished to close +- For scheduled tasks (no pause): set environment variable MISF_NO_PAUSE=1 + +OUTPUT +------ +- Output\\Original\\ — downloaded PDFs +- Output\\Summary\Download_Summary_.xlsx +- Output\\logs\download.log + +OPTIONAL COMMAND-LINE (run from cmd in this folder) +--------------------------------------------------- + MISF_VOB_Download.exe --limit 2 + MISF_VOB_Download.exe --facility "Amaya" + MISF_VOB_Download.exe --resume + +REQUIREMENTS +------------ +- Windows 10/11 +- Network access to Microsoft 365 / Planner +- chrome-win\ folder with portable Chromium (or system Chromium / CHROMIUM_PATH) +- Progress log: Output\\logs\download.log (no console window when using .exe) + +Keep MISF_VOB_Download.exe, _internal, and chrome-win together in the project root +(same directory as Input and Output). diff --git a/launcher.py b/launcher.py new file mode 100644 index 0000000..8c141c1 --- /dev/null +++ b/launcher.py @@ -0,0 +1,53 @@ +#!/usr/bin/env python3 +""" +MISF VOB Download — production entry (PyInstaller / double-click). + +Config, Input/, and Output/ live in the same folder as the .exe. +""" +from __future__ import annotations + +import os +import sys +import traceback +from pathlib import Path + +ROOT = Path(__file__).resolve().parent +SRC = ROOT / "src" +if not getattr(sys, "frozen", False) and str(SRC) not in sys.path: + sys.path.insert(0, str(SRC)) + +from planner_download import build_arg_parser, run_download_workflow # noqa: E402 + + +def _run_workflow() -> int: + exit_code = 0 + try: + args = build_arg_parser().parse_args() + run_download_workflow( + facility_limit=max(0, args.limit), + facility_name=args.facility.strip() or None, + resume=args.resume, + keep_browser=args.keep_browser, + screenshot_on_error=args.screenshot and not args.no_screenshot, + ) + except KeyboardInterrupt: + print("\nCancelled by user.") + exit_code = 130 + except SystemExit as exc: + exit_code = int(exc.code) if exc.code is not None else 0 + except Exception: + traceback.print_exc() + exit_code = 1 + return exit_code + + +def main() -> int: + if getattr(sys, "frozen", False): + from log_window import run_with_live_logs + + return run_with_live_logs(_run_workflow) + return _run_workflow() + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..8a9708a --- /dev/null +++ b/pytest.ini @@ -0,0 +1,4 @@ +[pytest] +testpaths = tests +pythonpath = src +addopts = -q diff --git a/requirements-build.txt b/requirements-build.txt new file mode 100644 index 0000000..ae6536b --- /dev/null +++ b/requirements-build.txt @@ -0,0 +1,3 @@ +# Build-time only (production .exe) +-r requirements.txt +pyinstaller>=6.10 diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..e25dde6 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,6 @@ +pandas==2.2.3 +openpyxl==3.1.5 +selenium==4.27.1 +webdriver-manager==4.0.2 +pyotp==2.9.0 +pytest==8.3.4 diff --git a/run_download.py b/run_download.py new file mode 100644 index 0000000..8207596 --- /dev/null +++ b/run_download.py @@ -0,0 +1,49 @@ +#!/usr/bin/env python3 +""" +MISF VOB — OneDrive / Planner face-sheet download. + +Usage: + python run_download.py + python run_download.py --limit 2 + python run_download.py --facility "Amaya Healthcare" + python run_download.py --resume + python run_download.py --keep-browser +""" +from __future__ import annotations + +import sys +import traceback +from pathlib import Path + +ROOT = Path(__file__).resolve().parent +SRC = ROOT / "src" +if str(SRC) not in sys.path: + sys.path.insert(0, str(SRC)) + +from planner_download import build_arg_parser, run_download_workflow # noqa: E402 + + +def main() -> int: + try: + args = build_arg_parser().parse_args() + run_download_workflow( + facility_limit=max(0, args.limit), + facility_name=args.facility.strip() or None, + resume=args.resume, + keep_browser=args.keep_browser, + screenshot_on_error=args.screenshot and not args.no_screenshot, + ) + return 0 + except KeyboardInterrupt: + print("\nCancelled by user.") + return 130 + except SystemExit as exc: + return int(exc.code) if exc.code is not None else 0 + except Exception: + traceback.print_exc() + print("\nRun failed. See traceback above.", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/bootstrap_input_xlsx.py b/scripts/bootstrap_input_xlsx.py new file mode 100644 index 0000000..1b35b45 --- /dev/null +++ b/scripts/bootstrap_input_xlsx.py @@ -0,0 +1,86 @@ +""" +Create Input/input.xlsx (if missing) and refresh Input/input.xlsx.example. + +Run from repository root: + python scripts/bootstrap_input_xlsx.py +""" +from __future__ import annotations + +from pathlib import Path + +import pandas as pd + +PROJECT_ROOT = Path(__file__).resolve().parent.parent +INPUT_DIR = PROJECT_ROOT / "Input" +CONFIG_PATH = INPUT_DIR / "input.xlsx" +EXAMPLE_PATH = INPUT_DIR / "input.xlsx.example" + +FACILITY_KEYWORDS = [ + "amaya", "brighton place san diego", "brighton place sd", "brighton place spring valley", + "brighton place-spring valley", "granite hill", "anberry", "arbor hills", "bayberry", + "bradley court", "castle manor", "friendship", "gramercy", "kearny", "lakeside", + "newport", "siena", "stanford court", "temecula", "walnut", "balboa", "cottonwood", + "country hills", "la mesa", "university care", "blue oak", "bridgewood", "capistrano", + "eden", "hillcrest heights", "la casa via", "torrey pines", "bradley gardens", "horizon", + "community care", "jacob", "the shores", "valley vista", "valle vista", "golden hill", + "grossmont post acute", "lemon grove", "mission hills", "parkside", "somerset", + "the springs", "victoria", "reo vista", "san diego post acute", "villa las palmas", + "ridgeview", "windsor gardens", "windsor rosewood", +] +FACILITIES = [k.title() for k in FACILITY_KEYWORDS] +FACILITIES.append("REPLACE_WITH_54TH_FACILITY_NAME") + +DEFAULT_PLANNER_URL = ( + "https://planner.cloud.microsoft/webui/mytasks/assignedtome/view/grid" + "?tid=0c15ad2c-19dd-4ded-9cd1-20d344386414" +) + +SHEET1_ROWS = [ + ("Proxy IP", ""), + ("Proxy Port", ""), + ("Proxy Username", ""), + ("Proxy Password", ""), + ("OneDrive Username", ""), + ("OneDrive Password", ""), + ("Microsoft TOTP Secret", ""), + ("Planner URL", DEFAULT_PLANNER_URL), + ( + "OneDrive URL", + "https://netorg393890-my.sharepoint.com/personal/joy_accesspsychology_org" + "/_layouts/15/onedrive.aspx?id=%2Fpersonal%2Fjoy%5Faccesspsychology%5Forg%2FDocuments" + "%2FAccess%20Facility%20Spreadsheets3", + ), + ("pdf_folder_path", str(PROJECT_ROOT / "Input" / "PDFs")), +] + + +def _build_workbook() -> tuple[pd.DataFrame, pd.DataFrame]: + sheet1 = pd.DataFrame(SHEET1_ROWS, columns=["Key", "Value"]) + facilities = pd.DataFrame({"Facility Name": FACILITIES}) + return sheet1, facilities + + +def _write(path: Path, sheet1: pd.DataFrame, facilities: pd.DataFrame) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + with pd.ExcelWriter(path, engine="openpyxl") as writer: + sheet1.to_excel(writer, sheet_name="Sheet1", index=False, header=False) + facilities.to_excel(writer, sheet_name="Facilities", index=False) + + +def main() -> None: + sheet1, facilities = _build_workbook() + + _write(EXAMPLE_PATH, sheet1, facilities) + print(f"Wrote template -> {EXAMPLE_PATH} ({len(FACILITIES)} facilities)") + + if CONFIG_PATH.exists(): + print(f"Skipped (already exists): {CONFIG_PATH}") + return + + _write(CONFIG_PATH, sheet1, facilities) + print(f"Created local config -> {CONFIG_PATH}") + print("Edit input.xlsx with your credentials before running run_download.py") + + +if __name__ == "__main__": + main() diff --git a/scripts/build_production_exe.ps1 b/scripts/build_production_exe.ps1 new file mode 100644 index 0000000..8b9fc19 --- /dev/null +++ b/scripts/build_production_exe.ps1 @@ -0,0 +1,88 @@ +# Build MISF_VOB_Download.exe and install at project root (double-click to run). +# Output: \MISF_VOB_Download.exe + _internal\ + +$ErrorActionPreference = "Stop" +$Root = Split-Path -Parent $PSScriptRoot +Set-Location $Root + +Write-Host "=== MISF VOB Download - production build (project root) ===" -ForegroundColor Cyan + +$PortableChrome = Join-Path $Root "chrome-win\chrome-win\chrome.exe" +$PortableDriver = Join-Path $Root "chrome-win\chrome-win\chromedriver.exe" +if (-not (Test-Path $PortableChrome)) { + throw "Portable Chromium not found: $PortableChrome (required for production build)" +} +if (-not (Test-Path $PortableDriver)) { + Write-Warning "Bundled chromedriver missing: $PortableDriver (will download at runtime)" +} else { + Write-Host "Using portable Chromium: $PortableChrome" +} + +if (-not (Test-Path ".venv\Scripts\python.exe")) { + Write-Host "Creating virtual environment..." + python -m venv .venv +} + +$Python = Join-Path $Root ".venv\Scripts\python.exe" +& $Python -m pip install --upgrade pip -q +& $Python -m pip install -r requirements.txt -q +& $Python -m pip install "pyinstaller>=6.10" -q + +Write-Host "Running PyInstaller..." +& $Python -m PyInstaller --noconfirm MISF_VOB_Download.spec +if ($LASTEXITCODE -ne 0) { + throw "PyInstaller failed with exit code $LASTEXITCODE" +} + +$BuiltExe = Join-Path $Root "dist\MISF_VOB_Download.exe" +$BuiltDir = Join-Path $Root "dist\MISF_VOB_Download" +$BuiltDirExe = Join-Path $BuiltDir "MISF_VOB_Download.exe" + +if (Test-Path $BuiltExe) { + Write-Host "Installing one-file build to project root..." + $ExeDest = Join-Path $Root "MISF_VOB_Download.exe" + Get-Process -Name "MISF_VOB_Download" -ErrorAction SilentlyContinue | Stop-Process -Force + Start-Sleep -Seconds 1 + Copy-Item $BuiltExe $ExeDest -Force + Remove-Item $BuiltExe -Force +} elseif (Test-Path $BuiltDirExe) { + Write-Host "Installing folder build to project root..." + $ExeDest = Join-Path $Root "MISF_VOB_Download.exe" + Get-Process -Name "MISF_VOB_Download" -ErrorAction SilentlyContinue | Stop-Process -Force + Start-Sleep -Seconds 1 + $InternalDest = Join-Path $Root "_internal" + + Copy-Item $BuiltDirExe $ExeDest -Force + if (Test-Path $InternalDest) { + Remove-Item $InternalDest -Recurse -Force + } + Move-Item (Join-Path $BuiltDir "_internal") $InternalDest -Force + Remove-Item $BuiltDir -Recurse -Force +} else { + throw "Expected build output not found: $BuiltExe or $BuiltDirExe" +} +$DistFolder = Join-Path $Root "dist" +if (Test-Path $DistFolder) { + $remaining = @(Get-ChildItem $DistFolder -Force -ErrorAction SilentlyContinue) + if ($remaining.Count -eq 0) { + Remove-Item $DistFolder -Force + } +} + +$InputDir = Join-Path $Root "Input" +$OutputDir = Join-Path $Root "Output" +New-Item -ItemType Directory -Force -Path (Join-Path $InputDir "PDFs") | Out-Null +New-Item -ItemType Directory -Force -Path $OutputDir | Out-Null +Copy-Item (Join-Path $Root "docs\PRODUCTION_README.txt") (Join-Path $Root "README_PRODUCTION.txt") -Force + +Write-Host "" +Write-Host "Build complete." -ForegroundColor Green +Write-Host " Run: $ExeDest" +Write-Host "" +Write-Host "First-time setup:" -ForegroundColor Yellow +Write-Host " 1. Copy Input\input.xlsx.example to Input\input.xlsx and fill credentials" +Write-Host " 2. Add facility names on the Facilities sheet" +Write-Host " 3. Keep chrome-win\ next to the .exe" +Write-Host " 4. Double-click MISF_VOB_Download.exe in this folder" +Write-Host "" +Write-Host 'See README_PRODUCTION.txt. Exe is at project root (no dist folder).' diff --git a/scripts/validate_config.py b/scripts/validate_config.py new file mode 100644 index 0000000..2d6baf6 --- /dev/null +++ b/scripts/validate_config.py @@ -0,0 +1,68 @@ +"""Print config status (no browser). Run: python scripts/validate_config.py""" +from __future__ import annotations + +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(ROOT / "src")) + +from config import ( # noqa: E402 + CONFIG_FILE, + DEFAULT_PLANNER_URL, + EXPECTED_FACILITY_COUNT, + credentials_setup_message, + missing_onedrive_credentials, + read_app_config, + read_facility_list, +) +from totp_utils import validate_totp_secret # noqa: E402 + + +def main() -> int: + path = CONFIG_FILE if CONFIG_FILE.exists() else ROOT / "Input" / "input.xlsx.example" + if not path.exists(): + print(f"[FAIL] Missing {path}") + print(" copy Input\\input.xlsx.example Input\\input.xlsx") + return 1 + + cfg = read_app_config(path) + missing = missing_onedrive_credentials( + cfg.onedrive_username, + cfg.onedrive_password, + cfg.microsoft_totp_secret, + ) + if missing and path == CONFIG_FILE: + print("[FAIL] Credentials not configured:") + print(credentials_setup_message(missing)) + return 1 + + try: + validate_totp_secret(cfg.microsoft_totp_secret or "JBSWY3DPEHPK3PXP") + except ValueError as exc: + if cfg.microsoft_totp_secret: + print(f"[FAIL] TOTP secret: {exc}") + return 1 + print(f"[WARN] TOTP not set — format check skipped ({exc})") + + facilities = read_facility_list(path, warn_if_not_expected_count=True) + planner = cfg.planner_url or DEFAULT_PLANNER_URL + if not planner.startswith("https://"): + print(f"[FAIL] Planner URL must be https — got: {planner[:80]}") + return 1 + + print(f"[OK] Config file: {path.name}") + if cfg.onedrive_username: + print(f"[OK] Credentials set for {cfg.onedrive_username}") + else: + print("[OK] Template mode (no credentials in example file)") + print(f"[OK] Planner URL: {planner[:70]}...") + print(f"[OK] PDF folder: {cfg.pdf_folder_path or '(default Input/PDFs)'}") + print(f"[OK] {len(facilities)} facilities (expected {EXPECTED_FACILITY_COUNT})") + if len(facilities) != EXPECTED_FACILITY_COUNT: + print("[WARN] Facility count differs from expected 54 — verify Facilities sheet") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/validate_uipath_pilot.ps1 b/scripts/validate_uipath_pilot.ps1 new file mode 100644 index 0000000..d1bef4d --- /dev/null +++ b/scripts/validate_uipath_pilot.ps1 @@ -0,0 +1,71 @@ +#Requires -Version 5.1 +<# +.SYNOPSIS + Pre-flight validation for UiPath MISF Planner Download pilot (mirrors Python validate_config + output checks). +#> +param( + [string]$ProjectRoot = (Resolve-Path (Join-Path $PSScriptRoot "..")).Path, + [int]$ExpectedFacilities = 54, + [switch]$CheckOutput +) + +$ErrorActionPreference = "Stop" +$configPath = Join-Path $ProjectRoot "Input\input.xlsx" +$uipathProject = Join-Path $ProjectRoot "uipath\MISF_Planner_Download\project.json" +$today = Get-Date -Format "yyyy-MM-dd" + +Write-Host "MISF UiPath pilot validation" +Write-Host "Project root: $ProjectRoot" + +if (-not (Test-Path $configPath)) { + Write-Error "Missing config: $configPath (copy Input\input.xlsx.example)" +} +if (-not (Test-Path $uipathProject)) { + Write-Error "Missing UiPath project: $uipathProject" +} + +# Run Python config validator if available +$pyValidate = Join-Path $ProjectRoot "scripts\validate_config.py" +if (Test-Path $pyValidate) { + Write-Host "Running Python validate_config.py..." + python $pyValidate + if ($LASTEXITCODE -ne 0) { exit $LASTEXITCODE } +} + +# Verify UiPath workflow files exist +$requiredWorkflows = @( + "Main.xaml", + "01_ReadConfig.xaml", + "02_LoginPlanner.xaml", + "03_ProcessFacility.xaml", + "04_DownloadTaskAttachments.xaml", + "05_SetTaskInProgress.xaml", + "06_WriteSummary.xaml", + "Helpers\GenerateTotp.xaml", + "Helpers\SanitizeFolderName.xaml", + "Selectors\PlannerSelectors.json" +) +$uipathDir = Join-Path $ProjectRoot "uipath\MISF_Planner_Download" +foreach ($wf in $requiredWorkflows) { + $p = Join-Path $uipathDir $wf + if (-not (Test-Path $p)) { + Write-Error "Missing workflow file: $p" + } +} +Write-Host "All UiPath workflow files present." + +if ($CheckOutput) { + $summary = Join-Path $ProjectRoot "Output\$today\Summary\Download_Summary_$today.xlsx" + $original = Join-Path $ProjectRoot "Output\$today\Original" + if (-not (Test-Path $summary)) { + Write-Warning "No summary yet: $summary (run UiPath Main with in_FacilityLimit=2)" + } else { + Write-Host "Summary found: $summary" + } + if (Test-Path $original) { + $pdfCount = (Get-ChildItem -Path $original -Recurse -Filter "*.pdf" -ErrorAction SilentlyContinue).Count + Write-Host "PDFs under Original: $pdfCount" + } +} + +Write-Host "Pilot pre-flight OK." diff --git a/src/__init__.py b/src/__init__.py new file mode 100644 index 0000000..2941540 --- /dev/null +++ b/src/__init__.py @@ -0,0 +1 @@ +"""MISF OneDrive / Planner download automation.""" diff --git a/src/checkpoint.py b/src/checkpoint.py new file mode 100644 index 0000000..1b5d87f --- /dev/null +++ b/src/checkpoint.py @@ -0,0 +1,51 @@ +"""Resume checkpoint for multi-facility download runs.""" +from __future__ import annotations + +import json +from pathlib import Path + + +def load_checkpoint(path: Path) -> dict: + """ + Return checkpoint dict: + - last_completed_index: int + - completed_facilities: list[str] + + Backward compatible with older files. + """ + if not path.is_file(): + return {"last_completed_index": 0, "completed_facilities": []} + try: + data = json.loads(path.read_text(encoding="utf-8")) + return { + "last_completed_index": int(data.get("last_completed_index", 0) or 0), + "completed_facilities": list(data.get("completed_facilities", []) or []), + } + except (json.JSONDecodeError, TypeError, ValueError): + return {"last_completed_index": 0, "completed_facilities": []} + + +def save_checkpoint( + path: Path, + *, + last_completed_index: int, + facility_name: str, + completed_facilities: list[str] | None = None, +) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + json.dumps( + { + "last_completed_index": last_completed_index, + "last_facility": facility_name, + "completed_facilities": completed_facilities or [], + }, + indent=2, + ), + encoding="utf-8", + ) + + +def clear_checkpoint(path: Path) -> None: + if path.is_file(): + path.unlink() diff --git a/src/chrome.py b/src/chrome.py new file mode 100644 index 0000000..7ef52d4 --- /dev/null +++ b/src/chrome.py @@ -0,0 +1,891 @@ +"""Chromium (Chrome-compatible) remote-debug startup and Selenium attach.""" +from __future__ import annotations + +import atexit +import json +import os +import re +import shutil +import socket +import subprocess +import sys +import tempfile +import time +import zipfile +from datetime import datetime +from pathlib import Path +from urllib.error import URLError +from urllib.request import urlopen + +from selenium import webdriver +from selenium.common.exceptions import WebDriverException +from selenium.webdriver.chrome.options import Options +from selenium.webdriver.chrome.service import Service +from selenium.webdriver.common.by import By +from selenium.webdriver.common.keys import Keys + +from config import PROJECT_ROOT +from debug_log import write_debug as _agent_log +from run_log import log_step + +_DEBUG_PORT = 9222 +DEBUG_PORT = _DEBUG_PORT +_PROXY_CREDENTIALS: dict[str, str] = {} +_LAST_BROWSER_PATH: str | None = None +_BROWSER_PROCESS: subprocess.Popen | None = None + +USER_DATA_DIR = os.path.join( + os.environ.get("USERPROFILE", str(Path.home())), + "Automation", + "ChromiumProfile_MISF_Download", +) + +_DRIVERS_DIR = PROJECT_ROOT / ".drivers" +_CFT_DOWNLOAD_BASE = "https://storage.googleapis.com/chrome-for-testing-public" +DOWNLOADS_DIR = PROJECT_ROOT / "Input" / "PDFs" +_VERSION_RE = re.compile(r"(\d+\.\d+\.\d+\.\d+)") + +# Microsoft 365 hosts bypass HTTP proxy when split-tunnel is configured. +CHROME_PROXY_BYPASS_OFFICE = ( + "*.office.com;*.office365.com;*.microsoft.com;*.live.com;*.outlook.com;" + "outlook.office.com;outlook.office365.com;login.microsoftonline.com;login.live.com;" + "aadcdn.msauth.net;aadcdn.msftauth.net" +) + + + +def set_default_download_directory(path: Path | str, *, create: bool = False) -> None: + """Set browser download prefs root (used at attach time). Per-facility CDP still applies.""" + global DOWNLOADS_DIR + DOWNLOADS_DIR = Path(path).resolve() + if create: + DOWNLOADS_DIR.mkdir(parents=True, exist_ok=True) + + +def _allow_google_chrome_fallback() -> bool: + return os.environ.get("MISF_ALLOW_CHROME", "").strip().lower() in ("1", "true", "yes") + + +def _portable_chrome_win_roots() -> list[Path]: + return [ + PROJECT_ROOT / "chrome-win" / "chrome-win", + PROJECT_ROOT / "chrome-win", + ] + + +def _portable_chromium_executable() -> str | None: + """Bundled Chromium under project ``chrome-win/`` (preferred for production).""" + for root in _portable_chrome_win_roots(): + for name in ("chrome.exe", "chromium.exe"): + path = root / name + if path.is_file(): + return str(path) + return None + + +def _bundled_chromedriver_for(browser_path: str) -> str | None: + """``chromedriver.exe`` shipped next to portable ``chrome.exe``.""" + if not browser_path: + return None + driver = Path(browser_path).parent / "chromedriver.exe" + if driver.is_file(): + return str(driver) + return None + + +def _installed_chromium_paths() -> list[str]: + """System-installed Chromium (not Google Chrome).""" + local = os.environ.get("LOCALAPPDATA", "") + profile = os.environ.get("USERPROFILE", str(Path.home())) + return [ + r"C:\Program Files\Chromium\Application\chrome.exe", + r"C:\Program Files\Chromium\Application\chromium.exe", + r"C:\Program Files (x86)\Chromium\Application\chrome.exe", + os.path.join(local, r"Chromium\Application\chrome.exe"), + os.path.join(profile, r"scoop\apps\chromium\current\chrome.exe"), + ] + + +def _prefer_installed_chromium() -> bool: + """Skip ``chrome-win/`` and use a system Chromium install instead.""" + return os.environ.get("MISF_USE_INSTALLED_CHROME", "").strip().lower() in ( + "1", + "true", + "yes", + ) + + +def _google_chrome_fallback_paths() -> list[str]: + local = os.environ.get("LOCALAPPDATA", "") + return [ + r"C:\Program Files\Google\Chrome\Application\chrome.exe", + r"C:\Program Files (x86)\Google\Chrome\Application\chrome.exe", + os.path.join(local, r"Google\Chrome\Application\chrome.exe"), + ] + + +def find_chromium_executable() -> str | None: + """ + Resolve Chromium only (Google Chrome is opt-in via ``MISF_ALLOW_CHROME=1``). + + Default order: ``chrome-win/`` portable bundle, then system Chromium. + Override: ``CHROMIUM_PATH`` / ``BROWSER_PATH``, or ``MISF_USE_INSTALLED_CHROME=1``. + """ + for env_key in ("CHROMIUM_PATH", "BROWSER_PATH"): + raw = os.environ.get(env_key, "").strip().strip('"') + if raw and os.path.isfile(raw): + return raw + + portable = _portable_chromium_executable() + installed = [p for p in _installed_chromium_paths() if os.path.isfile(p)] + + if _prefer_installed_chromium(): + candidates = installed + ([portable] if portable else []) + else: + candidates = ([portable] if portable else []) + installed + + if _allow_google_chrome_fallback(): + candidates = candidates + [ + p for p in _google_chrome_fallback_paths() if os.path.isfile(p) + ] + + for path in candidates: + if path: + _agent_log( + "A", + "src/chrome.py:find_chromium_executable", + "Selected browser executable", + {"basename": os.path.basename(path), "from_portable": "chrome-win" in path.lower()}, + ) + return path + _agent_log( + "A", + "src/chrome.py:find_chromium_executable", + "No Chromium executable found", + {"chrome_win_checked": [str(r) for r in _portable_chrome_win_roots()]}, + ) + return None + + +def _browser_runs_in_background() -> bool: + """Default: foreground (visible). Set MISF_BROWSER_BACKGROUND=1 to start minimized.""" + return os.environ.get("MISF_BROWSER_BACKGROUND", "").strip().lower() in ( + "1", + "true", + "yes", + ) + + +def _launch_browser_process(cmd: list[str]) -> subprocess.Popen: + """Start Chromium detached; minimized on Windows when running in background.""" + global _BROWSER_PROCESS + kwargs: dict = { + "stdout": subprocess.DEVNULL, + "stderr": subprocess.DEVNULL, + } + if sys.platform == "win32" and _browser_runs_in_background(): + startupinfo = subprocess.STARTUPINFO() + startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW + startupinfo.wShowWindow = 7 # SW_SHOWMINNOACTIVE — minimized, no focus steal + kwargs["startupinfo"] = startupinfo + _BROWSER_PROCESS = subprocess.Popen(cmd, **kwargs) + return _BROWSER_PROCESS + + +def stop_chrome_browser(driver: webdriver.Chrome | None = None) -> None: + """Disconnect Selenium and terminate the Chromium process we launched.""" + if driver is not None: + try: + driver.quit() + except Exception: + pass + + global _BROWSER_PROCESS + proc = _BROWSER_PROCESS + _BROWSER_PROCESS = None + if proc is not None and proc.poll() is None: + try: + proc.terminate() + proc.wait(timeout=8) + except Exception: + try: + proc.kill() + except Exception: + pass + + if wait_for_debug_port(DEBUG_PORT, timeout=1): + _kill_processes_listening_on_port(DEBUG_PORT) + + +def _kill_processes_listening_on_port(port: int) -> None: + """Best-effort cleanup when the stored Popen handle is unavailable.""" + if sys.platform != "win32": + return + try: + output = subprocess.check_output( + ["netstat", "-ano", "-p", "tcp"], + text=True, + errors="replace", + **_hidden_subprocess_kwargs(), + ) + except Exception: + return + target_pids: set[int] = set() + needle = f":{port}" + for line in output.splitlines(): + if needle not in line or "LISTENING" not in line.upper(): + continue + parts = line.split() + if not parts: + continue + try: + target_pids.add(int(parts[-1])) + except ValueError: + continue + for pid in target_pids: + if pid <= 0: + continue + try: + subprocess.run( + ["taskkill", "/PID", str(pid), "/T", "/F"], + check=False, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + **_hidden_subprocess_kwargs(), + ) + except Exception: + pass + + +def _apply_background_window(driver: webdriver.Chrome) -> None: + try: + driver.maximize_window() + except WebDriverException: + pass + + +def _bring_browser_to_front(driver: webdriver.Chrome) -> None: + """Best-effort focus for scheduled/server runs so captures are not blank.""" + if sys.platform != "win32": + return + try: + import ctypes + + user32 = ctypes.windll.user32 + hwnd = user32.GetForegroundWindow() + try: + rect = driver.get_window_rect() + target_x = int(rect.get("x", 0) + max(rect.get("width", 800) // 2, 1)) + target_y = int(rect.get("y", 0) + max(rect.get("height", 600) // 2, 1)) + user32.SetCursorPos(target_x, target_y) + except Exception: + pass + if hwnd: + user32.SetForegroundWindow(hwnd) + driver.switch_to.window(driver.current_window_handle) + except Exception: + pass + + +def prepare_browser_for_screenshot(driver: webdriver.Chrome) -> None: + """Maximize and focus the automation browser before board screenshots.""" + try: + driver.switch_to.default_content() + except Exception: + pass + _apply_background_window(driver) + _bring_browser_to_front(driver) + time.sleep(0.6) + + +def _hidden_subprocess_kwargs() -> dict: + """Run helper processes without flashing a console window on Windows.""" + if sys.platform != "win32": + return {} + kwargs: dict = {"creationflags": subprocess.CREATE_NO_WINDOW} + startupinfo = subprocess.STARTUPINFO() + startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW + startupinfo.wShowWindow = 0 + kwargs["startupinfo"] = startupinfo + return kwargs + + +def _capture_desktop_powershell(path: Path) -> bool: + """Primary monitor capture on Windows (includes taskbar date/time).""" + if sys.platform != "win32": + return False + out = str(path.resolve()) + ps_script = f""" +Add-Type -AssemblyName System.Windows.Forms,System.Drawing +$bounds = [System.Windows.Forms.Screen]::PrimaryScreen.Bounds +$bmp = New-Object System.Drawing.Bitmap $bounds.Width, $bounds.Height +$graphics = [System.Drawing.Graphics]::FromImage($bmp) +$graphics.CopyFromScreen($bounds.Location, [System.Drawing.Point]::Empty, $bounds.Size) +$bmp.Save('{out.replace("'", "''")}', [System.Drawing.Imaging.ImageFormat]::Png) +$graphics.Dispose() +$bmp.Dispose() +""" + try: + subprocess.run( + [ + "powershell", + "-NoProfile", + "-WindowStyle", + "Hidden", + "-Command", + ps_script, + ], + check=True, + timeout=20, + capture_output=True, + **_hidden_subprocess_kwargs(), + ) + return path.is_file() and path.stat().st_size > 0 + except Exception: + return False + + +def capture_desktop_screenshot(path: Path) -> bool: + """Save a full-desktop PNG (taskbar clock visible). Falls back to False if unavailable.""" + path = Path(path) + path.parent.mkdir(parents=True, exist_ok=True) + try: + from PIL import ImageGrab + + ImageGrab.grab(all_screens=False).save(str(path), "PNG") + if path.is_file(): + return True + except Exception: + pass + try: + import mss + import mss.tools + + with mss.mss() as sct: + monitor = sct.monitors[1] + shot = sct.grab(monitor) + mss.tools.to_png(shot.rgb, shot.size, output=str(path)) + return path.is_file() + except Exception: + pass + if _capture_desktop_powershell(path): + return True + try: + import pyautogui + + pyautogui.screenshot(str(path)) + return path.is_file() + except Exception: + pass + return False + + +def _is_image_mostly_black(path: Path, *, threshold: int = 18) -> bool: + """True when a PNG is nearly all black (common on headless/disconnected server sessions).""" + try: + from PIL import Image + + with Image.open(path) as img: + rgb = img.convert("RGB") + sample = rgb.resize((64, 64)) + pixels = list(sample.getdata()) + if not pixels: + return True + avg = sum(sum(p) for p in pixels) / (len(pixels) * 3) + return avg < threshold + except Exception: + return False + + +def _capture_browser_with_timestamp_bar(driver: webdriver.Chrome, path: Path) -> bool: + """ + Fallback when desktop capture is black: maximized browser PNG plus a taskbar-style clock strip. + """ + try: + import io + + from PIL import Image, ImageDraw + + prepare_browser_for_screenshot(driver) + png = driver.get_screenshot_as_png() + browser = Image.open(io.BytesIO(png)).convert("RGB") + bar_h = 48 + canvas = Image.new("RGB", (browser.width, browser.height + bar_h), (30, 30, 30)) + canvas.paste(browser, (0, 0)) + draw = ImageDraw.Draw(canvas) + stamp = datetime.now().strftime("%I:%M %p %m/%d/%Y").lstrip("0") + draw.text((max(browser.width - 260, 8), browser.height + 14), stamp, fill=(255, 255, 255)) + canvas.save(str(path), "PNG") + return path.is_file() and path.stat().st_size > 0 + except Exception: + return False + + +def capture_fullscreen_screenshot(driver: webdriver.Chrome, path: Path) -> bool: + """ + Full-screen capture with system taskbar visible when an interactive desktop exists. + + On scheduled/server sessions without a visible desktop, falls back to a maximized + browser screenshot plus a timestamp bar (instead of an all-black PNG). + """ + path = Path(path) + path.parent.mkdir(parents=True, exist_ok=True) + log_step("Fullscreen screenshot", path=str(path)) + prepare_browser_for_screenshot(driver) + if capture_desktop_screenshot(path) and not _is_image_mostly_black(path): + log_step("Fullscreen screenshot saved", path=str(path)) + return True + if _is_image_mostly_black(path): + try: + path.unlink(missing_ok=True) + except OSError: + pass + if _capture_browser_with_timestamp_bar(driver, path): + log_step("Browser screenshot with timestamp bar saved", path=str(path), mode="server_fallback") + return True + driver.save_screenshot(str(path)) + log_step("Viewport screenshot saved (desktop capture unavailable)", path=str(path)) + return path.is_file() + + +def wait_for_debug_port(port: int = DEBUG_PORT, timeout: int = 40) -> bool: + start = time.time() + while time.time() - start < timeout: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + if s.connect_ex(("127.0.0.1", port)) == 0: + return True + time.sleep(1) + return False + + +def is_debug_port_open(port: int = DEBUG_PORT) -> bool: + return wait_for_debug_port(port, timeout=1) + + +def _version_from_portable_dir(browser_path: str) -> str: + """Read e.g. 150.0.7852.0 from 150.0.7852.0.manifest next to chrome.exe.""" + folder = Path(browser_path).parent + for manifest in folder.glob("*.manifest"): + match = _VERSION_RE.search(manifest.name) + if match: + return match.group(1) + return "" + + +def _resolve_browser_version(browser_path: str) -> str: + """Best-effort version for the browser we launched (no hanging --version).""" + version = _get_browser_version_from_debug_port(timeout_secs=5.0) + if version: + return version + if browser_path: + version = _version_from_portable_dir(browser_path) + if version: + return version + return "" + + +def _ensure_chromedriver_cft(browser_version: str) -> str: + """ + Download chromedriver matching ``browser_version`` from Chrome for Testing. + Cached under ``.drivers//chromedriver.exe``. + """ + dest = _DRIVERS_DIR / browser_version / "chromedriver.exe" + if dest.is_file(): + return str(dest) + + dest.parent.mkdir(parents=True, exist_ok=True) + zip_url = f"{_CFT_DOWNLOAD_BASE}/{browser_version}/win64/chromedriver-win64.zip" + zip_path = dest.parent / "chromedriver-win64.zip" + print(f"[INFO] Downloading ChromeDriver {browser_version} (Chrome for Testing)...") + + _download_timeout = int( + os.environ.get("MISF_CHROMEDRIVER_DOWNLOAD_TIMEOUT", "60") + ) + last_exc: Exception | None = None + for attempt in range(1, 4): + try: + req = urlopen(zip_url, timeout=_download_timeout) + with open(zip_path, "wb") as fh: + shutil.copyfileobj(req, fh) + break + except Exception as exc: + last_exc = exc + print(f"[WARN] ChromeDriver download attempt {attempt}/3 failed: {exc}") + if zip_path.exists(): + zip_path.unlink(missing_ok=True) + time.sleep(2 * attempt) + else: + raise RuntimeError( + f"ChromeDriver {browser_version} download failed after 3 attempts: {last_exc}" + ) + + if not zipfile.is_zipfile(zip_path): + zip_path.unlink(missing_ok=True) + raise RuntimeError( + f"Downloaded ChromeDriver archive is not a valid ZIP: {zip_url}" + ) + + with zipfile.ZipFile(zip_path) as archive: + member = next( + (name for name in archive.namelist() if name.endswith("chromedriver.exe")), + None, + ) + if not member: + raise RuntimeError(f"chromedriver.exe not found in {zip_url}") + archive.extract(member, dest.parent) + extracted = dest.parent / member + extracted.rename(dest) + + try: + zip_path.unlink(missing_ok=True) + nested = dest.parent / "chromedriver-win64" + if nested.is_dir(): + shutil.rmtree(nested, ignore_errors=True) + except OSError: + pass + + print(f"[INFO] ChromeDriver installed -> {dest}") + return str(dest) + + +def _get_browser_version_from_debug_port(timeout_secs: float = 2.0) -> str: + """ + Read Chromium version from the running remote-debug instance. + + Example payload: {"Browser":"Chrome/150.0.7852.0", ...} + """ + try: + with urlopen(f"http://127.0.0.1:{DEBUG_PORT}/json/version", timeout=timeout_secs) as resp: + raw = resp.read().decode("utf-8", errors="replace") + data = json.loads(raw) + browser = str(data.get("Browser", "")).strip() + # "Chrome/150.0.7852.0" + if "/" in browser: + return browser.split("/", 1)[1].strip() + return browser + except (URLError, TimeoutError, ValueError, OSError): + return "" + + +def create_proxy_auth_extension( + proxy_ip: str, proxy_port: str, proxy_username: str, proxy_password: str +) -> str: + manifest_json = """{ + "version": "1.0.0", + "manifest_version": 2, + "name": "Chrome Proxy Auth Extension", + "permissions": [ + "proxy", "tabs", "unlimitedStorage", "storage", + "", "webRequest", "webRequestBlocking" + ], + "background": { "scripts": ["background.js"] }, + "minimum_chrome_version": "22.0.0" +}""" + _u = json.dumps(proxy_username) + _p = json.dumps(proxy_password) + background_js = f"""function callbackFn(details) {{ + return {{ authCredentials: {{ username: {_u}, password: {_p} }} }}; +}} +chrome.webRequest.onAuthRequired.addListener( + callbackFn, {{urls: [""]}}, ["blocking"] +);""" + ext_dir = Path(tempfile.mkdtemp(prefix="proxy_auth_ext_")) + atexit.register(shutil.rmtree, str(ext_dir), True) + (ext_dir / "manifest.json").write_text(manifest_json, encoding="utf-8") + (ext_dir / "background.js").write_text(background_js, encoding="utf-8") + print(f"[INFO] Proxy auth extension folder -> {ext_dir}") + return str(ext_dir) + + +def start_chrome_debug( + proxy_ip=None, + proxy_port=None, + proxy_username=None, + proxy_password=None, +) -> bool: + global _PROXY_CREDENTIALS + + browser_path = find_chromium_executable() + if not browser_path: + raise RuntimeError( + "Chromium not found. Place the portable bundle in chrome-win\\chrome-win\\ " + "(chrome.exe + chromedriver.exe), install Chromium, or set CHROMIUM_PATH. " + "(Google Chrome: MISF_ALLOW_CHROME=1.)" + ) + + log_step( + "Launch Chromium", + executable=browser_path, + background=str(_browser_runs_in_background()).lower(), + ) + global _LAST_BROWSER_PATH + _LAST_BROWSER_PATH = browser_path + os.makedirs(USER_DATA_DIR, exist_ok=True) + + cmd = [ + browser_path, + f"--remote-debugging-port={DEBUG_PORT}", + "--remote-allow-origins=*", + f"--user-data-dir={USER_DATA_DIR}", + "--disable-blink-features=AutomationControlled", + "--disable-popup-blocking", + "--no-first-run", + "--no-default-browser-check", + "--password-store=basic", + "--disable-restore-session-state", + "--disable-session-crashed-bubble", + "--hide-crash-restore-bubble", + "about:blank", + ] + if _browser_runs_in_background(): + cmd.append("--start-minimized") + + if proxy_ip and proxy_port: + port_str = str(proxy_port).strip() + if not port_str.isdigit(): + print(f"[WARN] Invalid proxy port '{port_str}' — skipping proxy") + _PROXY_CREDENTIALS = {} + else: + cmd.append(f"--proxy-server=http://{proxy_ip.strip()}:{port_str}") + cmd.append( + "--proxy-bypass-list=127.0.0.1;localhost;<-loopback>;" + + CHROME_PROXY_BYPASS_OFFICE + ) + log_step("Proxy configured", host=proxy_ip.strip(), port=port_str) + if proxy_username and proxy_password: + ext_folder = create_proxy_auth_extension( + proxy_ip.strip(), + port_str, + proxy_username.strip(), + proxy_password, + ) + cmd.append(f"--load-extension={ext_folder}") + _PROXY_CREDENTIALS = { + "username": proxy_username.strip(), + "password": str(proxy_password), + } + else: + _PROXY_CREDENTIALS = {} + else: + print("[INFO] No proxy configured — direct connection") + _PROXY_CREDENTIALS = {} + + if is_debug_port_open(DEBUG_PORT): + log_step("Debug port already in use — closing previous browser session") + stop_chrome_browser() + time.sleep(2) + + _launch_browser_process(cmd) + + if not wait_for_debug_port(DEBUG_PORT, timeout=40): + print("[ERROR] Browser debug port did not open in time") + return False + + print("[INFO] Browser debug port ready") + return True + + +def _find_chromedriver_path() -> str | None: + """ + Return a chromedriver path matching the browser we started. + + Prefers ``chromedriver.exe`` bundled in ``chrome-win/``. Otherwise downloads + a matching Chrome-for-Testing driver, then Selenium Manager. + """ + for env_key in ("CHROMEDRIVER_PATH", "WEBDRIVER_CHROME_DRIVER"): + raw = os.environ.get(env_key, "").strip().strip('"') + if raw and os.path.isfile(raw): + _agent_log( + "A", + "src/chrome.py:_find_chromedriver_path", + "Using explicit chromedriver path", + {"basename": os.path.basename(raw)}, + ) + return raw + + browser_path = _LAST_BROWSER_PATH or "" + bundled = _bundled_chromedriver_for(browser_path) + if bundled: + log_step("Using bundled chromedriver", path=bundled) + _agent_log( + "A", + "src/chrome.py:_find_chromedriver_path", + "Using chrome-win chromedriver", + {"basename": os.path.basename(bundled)}, + ) + return bundled + + browser_version = _resolve_browser_version(browser_path) + _agent_log( + "A", + "src/chrome.py:_find_chromedriver_path", + "Detected browser version", + { + "browser_version": browser_version, + "from_portable": "chrome-win" in (browser_path or "").lower(), + }, + ) + + if browser_version: + try: + driver_path = _ensure_chromedriver_cft(browser_version) + _agent_log( + "A", + "src/chrome.py:_find_chromedriver_path", + "Chrome for Testing chromedriver ready", + {"browser_version": browser_version}, + ) + return driver_path + except Exception as exc: + print( + f"[WARN] Could not download ChromeDriver {browser_version}: {exc}" + ) + _agent_log( + "A", + "src/chrome.py:_find_chromedriver_path", + "CfT chromedriver download failed", + {"error": str(exc)[:200]}, + ) + + _agent_log( + "A", + "src/chrome.py:_find_chromedriver_path", + "Using Selenium Manager for chromedriver", + {}, + ) + print( + "[WARN] Browser version unknown — Selenium Manager may pick the wrong driver. " + "Set CHROMEDRIVER_PATH or use installed Chrome (default)." + ) + return None + + +def _fill_proxy_auth_js_alert_if_present( + driver, username: str, password: str, wait_secs: float +) -> bool: + deadline = time.monotonic() + wait_secs + while time.monotonic() < deadline: + try: + alert = driver.switch_to.alert + alert.send_keys(Keys.CONTROL + "a") + alert.send_keys(username) + alert.send_keys(Keys.TAB) + alert.send_keys(Keys.CONTROL + "a") + alert.send_keys(password) + alert.accept() + return True + except Exception: + time.sleep(0.25) + return False + + +def _try_fill_proxy_auth_native_dialog(username: str, password: str) -> bool: + if sys.platform != "win32": + return False + try: + import pyautogui + except ImportError: + print("[INFO] pyautogui not installed — skip native proxy dialog fill") + return False + try: + time.sleep(0.6) + pyautogui.FAILSAFE = True + pyautogui.hotkey("ctrl", "a") + pyautogui.press("backspace") + pyautogui.write(username, interval=0.02) + pyautogui.press("tab") + pyautogui.hotkey("ctrl", "a") + pyautogui.press("backspace") + pyautogui.write(password, interval=0.02) + pyautogui.press("enter") + return True + except Exception as exc: + print(f"[WARN] Native proxy keyboard fill failed: {exc}") + return False + + +def _microsoft_login_visible(driver) -> bool: + try: + url = (driver.current_url or "").lower() + if "login.microsoftonline.com" in url or "login.live.com" in url: + return True + return bool( + driver.find_elements(By.CSS_SELECTOR, "input[type='email']") + or driver.find_elements(By.NAME, "loginfmt") + ) + except Exception: + return False + + +def dismiss_proxy_auth_popup( + driver, + username: str, + password: str, + wait_secs: float = 5.0, + max_attempts: int = 3, +) -> bool: + for attempt in range(1, max_attempts + 1): + if _fill_proxy_auth_js_alert_if_present(driver, username, password, wait_secs): + print(f"[INFO] Proxy auth handled (JS dialog) attempt {attempt}/{max_attempts}") + return True + if attempt < max_attempts: + time.sleep(2) + + if _try_fill_proxy_auth_native_dialog(username, password): + time.sleep(2) + if _microsoft_login_visible(driver): + print("[INFO] Microsoft login visible after proxy auth — proceeding") + return True + + if _microsoft_login_visible(driver): + print("[INFO] Microsoft login page present — proxy session usable") + return True + + print("[WARN] Proxy auth popup could not be dismissed") + return False + + +def attach_browser( + retries: int = 5, + download_dir: Path | str | None = None, + *, + create_download_dir: bool = False, +) -> webdriver.Chrome | None: + driver_path = _find_chromedriver_path() + service = Service(driver_path) if driver_path else Service() + dl_dir = Path(download_dir) if download_dir else DOWNLOADS_DIR + if create_download_dir: + dl_dir.mkdir(parents=True, exist_ok=True) + + for attempt in range(1, retries + 1): + try: + log_step("Attach browser", attempt=f"{attempt}/{retries}", download_dir=str(dl_dir)) + chrome_options = Options() + chrome_options.add_experimental_option( + "debuggerAddress", f"127.0.0.1:{DEBUG_PORT}" + ) + chrome_options.add_experimental_option( + "prefs", + { + "download.default_directory": str(dl_dir), + "download.prompt_for_download": False, + "plugins.always_open_pdf_externally": True, + "download.directory_upgrade": True, + "credentials_enable_service": False, + "profile.password_manager_enabled": False, + }, + ) + driver = webdriver.Chrome(service=service, options=chrome_options) + _ = driver.current_url + _apply_background_window(driver) + driver.set_page_load_timeout(90) + driver.set_script_timeout(30) + log_step("Browser attached successfully") + return driver + except WebDriverException as exc: + log_step("Attach failed", attempt=f"{attempt}/{retries}", error=str(exc)[:200]) + if attempt < retries: + time.sleep(3) + + print("[ERROR] Failed to attach browser after all retries") + return None diff --git a/src/config.py b/src/config.py new file mode 100644 index 0000000..6078c8a --- /dev/null +++ b/src/config.py @@ -0,0 +1,273 @@ +"""OneDrive / Planner download configuration from Input/input.xlsx.""" +from __future__ import annotations + +import logging +import os +import re +import sys +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path + +import pandas as pd + +from totp_utils import validate_totp_secret + +if getattr(sys, "frozen", False): + PROJECT_ROOT = Path(sys.executable).parent +else: + PROJECT_ROOT = Path(__file__).resolve().parent.parent + +CONFIG_FILE = PROJECT_ROOT / "Input" / "input.xlsx" +TODAY_ISO_DATE = datetime.now().strftime("%Y-%m-%d") +EXPECTED_FACILITY_COUNT = 54 + +DEFAULT_ONEDRIVE_URL = ( + "https://netorg393890-my.sharepoint.com/personal/joy_accesspsychology_org" + "/_layouts/15/onedrive.aspx" + "?id=%2Fpersonal%2Fjoy%5Faccesspsychology%5Forg%2FDocuments" + "%2FAccess%20Facility%20Spreadsheets3" +) + +DEFAULT_PLANNER_URL = ( + "https://planner.cloud.microsoft/webui/mytasks/assignedtome/view/grid" + "?tid=0c15ad2c-19dd-4ded-9cd1-20d344386414" +) + +# Rows like REPLACE_WITH_54TH_FACILITY_NAME are skipped when loading facilities. +_FACILITY_PLACEHOLDER_RE = re.compile( + r"^REPLACE_WITH_.*", + re.IGNORECASE, +) + +_log = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class AppConfig: + proxy_ip: str + proxy_port: str + proxy_username: str + proxy_password: str + onedrive_username: str + onedrive_password: str + microsoft_totp_secret: str + onedrive_url: str + planner_url: str + pdf_folder_path: str + + +def _load_dotenv_file() -> None: + """Load .env from project root into os.environ (no-override: existing vars take priority).""" + env_path = PROJECT_ROOT / ".env" + if not env_path.is_file(): + return + try: + from dotenv import load_dotenv + load_dotenv(dotenv_path=env_path, override=False) + except ImportError: + # Fallback: basic KEY=VALUE parser (no multi-line or comment support) + for raw_line in env_path.read_text(encoding="utf-8").splitlines(): + line = raw_line.strip() + if not line or line.startswith("#") or "=" not in line: + continue + key, _, value = line.partition("=") + key = key.strip() + value = value.strip().strip('"').strip("'") + if key and key not in os.environ: + os.environ[key] = value + + +def _env(name: str, *aliases: str) -> str: + for key in (name, *aliases): + val = os.environ.get(key, "").strip() + if val: + return val + return "" + + +def _read_config_key_value_sheet(config_path: Path | None = None) -> dict[str, str]: + path = config_path or CONFIG_FILE + if not path.exists(): + raise FileNotFoundError(f"input.xlsx not found at {path}") + + try: + df = pd.read_excel(str(path), sheet_name="Sheet1", header=None, usecols=[0, 1]) + except Exception as exc: + raise RuntimeError( + f"Could not read config from {path} (Sheet1): {exc}" + ) from exc + + def _clean(val) -> str: + s = str(val).strip() + return "" if s.lower() in ("nan", "none", "") else s + + if len(df.columns) < 2: + raise RuntimeError( + f"Config sheet Sheet1 in {path} must have at least 2 columns (Key | Value)." + ) + + config: dict[str, str] = {} + for _, row in df.iterrows(): + key = _clean(row[0]) + value = _clean(row[1]) + if key: + config[key] = value + return config + + +def missing_onedrive_credentials( + onedrive_username: str, + onedrive_password: str, + microsoft_totp_secret: str, +) -> list[str]: + missing = [] + if not onedrive_username: + missing.append("OneDrive Username") + if not onedrive_password: + missing.append("OneDrive Password") + if not microsoft_totp_secret: + missing.append("Microsoft TOTP Secret") + return missing + + +def credentials_setup_message(missing: list[str]) -> str: + keys = ", ".join(missing) + return ( + f"Missing: {keys}\n\n" + f"Option A — edit {CONFIG_FILE} Sheet1, column B next to each key.\n" + f"Option B — copy .env.example to .env in the project root and set:\n" + " ONEDRIVE_USERNAME, ONEDRIVE_PASSWORD, MICROSOFT_TOTP_SECRET\n" + " PLANNER_URL (optional)\n" + ) + + +def read_app_config(config_path: Path | None = None) -> AppConfig: + """Load full application config from input.xlsx with .env fallbacks.""" + _load_dotenv_file() + config = _read_config_key_value_sheet(config_path) + + proxy_ip = config.get("Proxy IP", "") or _env("PROXY_IP") + proxy_port = config.get("Proxy Port", "") or _env("PROXY_PORT") + proxy_username = config.get("Proxy Username", "") or _env("PROXY_USERNAME") + proxy_password = config.get("Proxy Password", "") or _env("PROXY_PASSWORD") + + onedrive_username = config.get("OneDrive Username", "") or _env( + "ONEDRIVE_USERNAME", "OneDrive Username" + ) + onedrive_password = config.get("OneDrive Password", "") or _env( + "ONEDRIVE_PASSWORD", "OneDrive Password" + ) + microsoft_totp_secret = ( + config.get("Microsoft TOTP Secret", "") + or config.get("OTP Secret", "") + or _env("MICROSOFT_TOTP_SECRET", "OTP_SECRET", "TOTP_SECRET") + ) + onedrive_url = config.get("OneDrive URL", "") or _env("ONEDRIVE_URL") or DEFAULT_ONEDRIVE_URL + planner_url = ( + config.get("Planner URL", "") + or config.get("Planner URL ", "") + or _env("PLANNER_URL") + or DEFAULT_PLANNER_URL + ) + pdf_folder_path = config.get("pdf_folder_path", "") or _env("PDF_FOLDER_PATH") + + _log.info("Proxy -> %s:%s", proxy_ip, proxy_port) + _log.info("OneDrive User -> %s", onedrive_username) + _log.info("Planner URL -> %s...", planner_url[:80]) + _log.info("PDF folder path -> %s", pdf_folder_path or "(default Input/PDFs)") + + return AppConfig( + proxy_ip=proxy_ip, + proxy_port=proxy_port, + proxy_username=proxy_username, + proxy_password=proxy_password, + onedrive_username=onedrive_username, + onedrive_password=onedrive_password, + microsoft_totp_secret=microsoft_totp_secret, + onedrive_url=onedrive_url, + planner_url=planner_url, + pdf_folder_path=pdf_folder_path, + ) + + +def read_onedrive_config(config_path: Path | None = None) -> tuple: + """ + Legacy tuple API. Prefer :func:`read_app_config`. + + Returns + ------- + (proxy_ip, proxy_port, proxy_username, proxy_password, + onedrive_username, onedrive_password, microsoft_totp_secret, + onedrive_url, pdf_folder_path, planner_url) + """ + cfg = read_app_config(config_path) + return ( + cfg.proxy_ip, + cfg.proxy_port, + cfg.proxy_username, + cfg.proxy_password, + cfg.onedrive_username, + cfg.onedrive_password, + cfg.microsoft_totp_secret, + cfg.onedrive_url, + cfg.pdf_folder_path, + cfg.planner_url, + ) + + +def _is_placeholder_facility(name: str) -> bool: + return bool(_FACILITY_PLACEHOLDER_RE.match(name.strip())) + + +def read_facility_list( + config_path: Path | None = None, + *, + skip_placeholders: bool = True, + warn_if_not_expected_count: bool = True, +) -> list[str]: + path = config_path or CONFIG_FILE + if not path.exists(): + raise FileNotFoundError(f"input.xlsx not found at {path}") + + try: + df = pd.read_excel(str(path), sheet_name="Facilities", usecols=[0]) + except ValueError as exc: + raise RuntimeError( + f"Sheet 'Facilities' not found in {path}. " + "Add a sheet with column 'Facility Name' (one facility per row)." + ) from exc + + # Always use the first column regardless of its header name. + col = df.columns[0] + + names: list[str] = [] + skipped: list[str] = [] + for val in df[col]: + s = str(val).strip() + if not s or s.lower() in ("nan", "none", "facility name"): + continue + if skip_placeholders and _is_placeholder_facility(s): + skipped.append(s) + continue + names.append(s) + + if skipped: + _log.warning( + "Skipped %d placeholder facility row(s): %s", + len(skipped), + ", ".join(skipped[:3]), + ) + + if not names: + raise RuntimeError(f"No facility names found in 'Facilities' sheet of {path}.") + + if warn_if_not_expected_count and len(names) != EXPECTED_FACILITY_COUNT: + _log.warning( + "Expected %d facilities, found %d (check Facilities sheet names).", + EXPECTED_FACILITY_COUNT, + len(names), + ) + + _log.info("Loaded %d facility name(s) from Facilities sheet", len(names)) + return names diff --git a/src/debug_log.py b/src/debug_log.py new file mode 100644 index 0000000..a06c865 --- /dev/null +++ b/src/debug_log.py @@ -0,0 +1,60 @@ +"""Shared NDJSON debug logging (MISF_DEBUG=1 only). No secrets/PII.""" +from __future__ import annotations + +import json +import os +import time +import uuid + +from config import PROJECT_ROOT + +_DEBUG_LOG = PROJECT_ROOT / "debug.log" +_AGENT_SESSION_LOG = PROJECT_ROOT / "debug-6d994d.log" +_AGENT_SESSION_ID = "6d994d" + + +def write_agent_session_debug( + hypothesis_id: str, location: str, message: str, data: dict +) -> None: + """NDJSON debug log for Cursor agent debug sessions (MISF_DEBUG=1 only).""" + # region agent log + if os.environ.get("MISF_DEBUG", "").strip() not in ("1", "true", "yes"): + return + try: + ts_ms = int(time.time() * 1000) + payload = { + "sessionId": _AGENT_SESSION_ID, + "runId": os.environ.get("MISF_RUN_ID", "run1"), + "hypothesisId": hypothesis_id, + "id": f"log_{ts_ms}_{uuid.uuid4().hex[:8]}", + "timestamp": ts_ms, + "location": location, + "message": message, + "data": data, + } + with _AGENT_SESSION_LOG.open("a", encoding="utf-8") as f: + f.write(json.dumps(payload, ensure_ascii=False) + "\n") + except Exception: + pass + # endregion + + +def write_debug(hypothesis_id: str, location: str, message: str, data: dict) -> None: + if os.environ.get("MISF_DEBUG", "").strip() not in ("1", "true", "yes"): + return + try: + ts_ms = int(time.time() * 1000) + payload = { + "sessionId": os.environ.get("MISF_SESSION_ID", "local"), + "runId": os.environ.get("MISF_RUN_ID", "run1"), + "hypothesisId": hypothesis_id, + "id": f"log_{ts_ms}_{uuid.uuid4().hex[:8]}", + "timestamp": ts_ms, + "location": location, + "message": message, + "data": data, + } + with _DEBUG_LOG.open("a", encoding="utf-8") as f: + f.write(json.dumps(payload, ensure_ascii=False) + "\n") + except Exception: + pass diff --git a/src/log_window.py b/src/log_window.py new file mode 100644 index 0000000..6a72517 --- /dev/null +++ b/src/log_window.py @@ -0,0 +1,118 @@ +"""Live log viewer for windowed MISF_VOB_Download.exe (no console).""" +from __future__ import annotations + +import logging +import queue +import threading +import tkinter as tk +from collections.abc import Callable +from tkinter import scrolledtext + + +class _TkLogHandler(logging.Handler): + """Thread-safe logging handler that forwards records to a Tk text widget.""" + + def __init__(self, text_widget: scrolledtext.ScrolledText, root: tk.Tk) -> None: + super().__init__() + self._text = text_widget + self._root = root + self._queue: queue.Queue[str] = queue.Queue() + self._root.after(100, self._drain_queue) + + def emit(self, record: logging.LogRecord) -> None: + try: + msg = self.format(record) + except Exception: + self.handleError(record) + return + self._queue.put(msg + "\n") + + def _drain_queue(self) -> None: + while True: + try: + line = self._queue.get_nowait() + except queue.Empty: + break + self._text.configure(state="normal") + self._text.insert(tk.END, line) + self._text.see(tk.END) + self._text.configure(state="disabled") + self._root.after(100, self._drain_queue) + + +def run_with_live_logs(workflow: Callable[[], int]) -> int: + """ + Show a scrollable log window and run *workflow* on a background thread. + + Returns the workflow exit code after the user closes the window. + """ + root = tk.Tk() + root.title("MISF VOB Download — Live Log") + root.geometry("920x560") + root.minsize(640, 360) + + header = tk.Label( + root, + text="Download in progress — logs update live below", + anchor="w", + font=("Segoe UI", 10, "bold"), + ) + header.pack(fill="x", padx=10, pady=(10, 4)) + + text = scrolledtext.ScrolledText( + root, + wrap="word", + state="disabled", + font=("Consolas", 10), + ) + text.pack(fill="both", expand=True, padx=10, pady=4) + + status_var = tk.StringVar(value="Running...") + status_label = tk.Label(root, textvariable=status_var, anchor="w") + status_label.pack(fill="x", padx=10, pady=(0, 6)) + + close_btn = tk.Button(root, text="Close", state="disabled", width=12) + close_btn.pack(pady=(0, 10)) + + fmt = logging.Formatter( + "%(asctime)s [%(levelname)s] %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", + ) + handler = _TkLogHandler(text, root) + handler.setFormatter(fmt) + handler.setLevel(logging.INFO) + + root_logger = logging.getLogger() + root_logger.addHandler(handler) + + result: dict[str, int] = {"code": 0} + + def _run_workflow() -> None: + try: + result["code"] = workflow() + except SystemExit as exc: + result["code"] = int(exc.code) if exc.code is not None else 0 + except Exception: + result["code"] = 1 + logging.getLogger(__name__).exception("Workflow failed") + finally: + def _on_done() -> None: + if result["code"] == 0: + status_var.set("Complete — you may close this window.") + else: + status_var.set(f"Finished with errors (exit {result['code']}). See log above.") + close_btn.configure(state="normal") + + root.after(0, _on_done) + + worker = threading.Thread(target=_run_workflow, daemon=True) + worker.start() + + def _close() -> None: + root_logger.removeHandler(handler) + root.destroy() + + close_btn.configure(command=_close) + root.protocol("WM_DELETE_WINDOW", _close) + root.mainloop() + return result["code"] diff --git a/src/logging_setup.py b/src/logging_setup.py new file mode 100644 index 0000000..2470a27 --- /dev/null +++ b/src/logging_setup.py @@ -0,0 +1,68 @@ +"""Structured logging for MISF download runs.""" +from __future__ import annotations + +import logging +import os +import sys +from datetime import datetime +from pathlib import Path + + +def setup_logging(project_root: Path, *, run_date: str | None = None) -> Path: + """ + Configure root logger and misf.run logger: console + download.log + moves.log. + Returns the main download log file path. + """ + date_str = run_date or datetime.now().strftime("%Y-%m-%d") + log_dir = project_root / "Output" / date_str / "logs" + log_dir.mkdir(parents=True, exist_ok=True) + log_path = log_dir / "download.log" + moves_path = log_dir / "moves.log" + + level_name = os.environ.get("MISF_LOG_LEVEL", "INFO").upper() + level = getattr(logging, level_name, logging.INFO) + + fmt = logging.Formatter( + "%(asctime)s [%(levelname)s] %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", + ) + + root = logging.getLogger() + root.setLevel(level) + + # Avoid duplicate handlers on repeated setup in same process. + for h in list(root.handlers): + root.removeHandler(h) + + # Windowed .exe: log files only (no console / live log window). + if not getattr(sys, "frozen", False): + console = logging.StreamHandler() + console.setLevel(level) + console.setFormatter(fmt) + root.addHandler(console) + + file_handler = logging.FileHandler(log_path, encoding="utf-8") + file_handler.setLevel(level) + file_handler.setFormatter(fmt) + root.addHandler(file_handler) + + moves_handler = logging.FileHandler(moves_path, encoding="utf-8") + moves_handler.setLevel(level) + moves_handler.setFormatter(fmt) + + run_logger = logging.getLogger("misf.run") + run_logger.setLevel(level) + run_logger.propagate = True + run_logger.addHandler(moves_handler) + + # Module loggers used across the app + for name in ("misf", "src.planner_download", "src.onedrive_planner", "src.chrome", "src.config"): + logging.getLogger(name).setLevel(level) + + logging.getLogger("misf.run").info( + "Logging enabled -> %s and %s (filter [STEP] in moves.log)", + log_path, + moves_path, + ) + + return log_path diff --git a/src/onedrive_planner.py b/src/onedrive_planner.py new file mode 100644 index 0000000..5b5ffd0 --- /dev/null +++ b/src/onedrive_planner.py @@ -0,0 +1,6305 @@ +# ================================================================ +# 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_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 'Couldn't open your plan' (URL may still be /plan/).""" + phrases = ( + "couldn't open your plan", + "could not open your plan", + "try again in a few minutes", + ) + 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 _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. + """ + 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) + 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 _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 = [ + (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']"), + ] + 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 (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[str]: + """Scroll the virtualized grid and collect Not started task titles.""" + _scroll_grid_to_top(driver) + seen: set[str] = set() + titles: list[str] = [] + 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 + title = _grid_row_task_name(row).strip() + if not title or _is_weak_task_title_key(title.lower()): + continue + key = title.lower() + if key in seen: + continue + seen.add(key) + titles.append(title) + 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(titles)) + return titles + + +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