Initial commit: MISF VOB download and upload automation

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
lubna-fatimaandClaude Sonnet 4.6 committed 2026-06-16 14:55:29 +05:30
commit c2dec96b1f
43 files changed
+10944

No files matched your search

+23
View File
@@ -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=
+41
View File
@@ -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
+228
View File
@@ -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/<FacilityName>/*.pdf`
- `Output/<date>/Summary/Download_Summary_<date>.xlsx` (includes run timestamp and Planner URL)
- `Output/<date>/logs/download.log`
- `Output/<date>/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/<date>/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.
+42
View File
@@ -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\<date>\Original\<facility>\ — downloaded PDFs
- Output\<date>\Summary\Download_Summary_<date>.xlsx
- Output\<date>\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\<date>\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).
Binary file not shown.
+42
View File
@@ -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\<date>\Original\<facility>\ — downloaded PDFs
- Output\<date>\Summary\Download_Summary_<date>.xlsx
- Output\<date>\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\<date>\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).
+53
View File
@@ -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())
+4
View File
@@ -0,0 +1,4 @@
[pytest]
testpaths = tests
pythonpath = src
addopts = -q
+3
View File
@@ -0,0 +1,3 @@
# Build-time only (production .exe)
-r requirements.txt
pyinstaller>=6.10
+6
View File
@@ -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
+49
View File
@@ -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())
+86
View File
@@ -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()
+88
View File
@@ -0,0 +1,88 @@
# Build MISF_VOB_Download.exe and install at project root (double-click to run).
# Output: <project root>\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).'
+68
View File
@@ -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())
+71
View File
@@ -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."
+1
View File
@@ -0,0 +1 @@
"""MISF OneDrive / Planner download automation."""
+51
View File
@@ -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()
+891
View File
@@ -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/<version>/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",
"<all_urls>", "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: ["<all_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
+273
View File
@@ -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
+60
View File
@@ -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
+118
View File
@@ -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"]
+68
View File
@@ -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
File diff suppressed because it is too large. Load diff
+747
View File
@@ -0,0 +1,747 @@
# ================================================================
# MISF — OneDrive / Planner face-sheet download (orchestrator)
#
# Entry point: python run_download.py (from repository root)
# See README.md for the full business workflow and configuration.
# ================================================================
from __future__ import annotations
import argparse
import logging
import shutil
import sys
import time
import traceback
import tkinter as tk
from datetime import datetime
from pathlib import Path
from tkinter import messagebox
import pandas as pd
from checkpoint import clear_checkpoint, load_checkpoint, save_checkpoint
from chrome import (
attach_browser,
dismiss_proxy_auth_popup,
set_default_download_directory,
start_chrome_debug,
stop_chrome_browser,
)
from config import (
CONFIG_FILE,
PROJECT_ROOT,
TODAY_ISO_DATE,
AppConfig,
credentials_setup_message,
missing_onedrive_credentials,
read_app_config,
read_facility_list,
)
from logging_setup import setup_logging
from run_log import log_step, log_url
from onedrive_planner import (
FacilityNotFoundError,
FacilityRunResult,
_remove_empty_directory,
_sanitize_folder_name,
configure_http_download_proxy,
consolidate_browser_to_single_tab,
login_and_open_planner,
process_facility,
reset_run_state,
save_failure_screenshot,
)
from totp_utils import validate_totp_secret
from debug_log import write_debug as _agent_log
_log = logging.getLogger(__name__)
def _show_error_dialog(title: str, message: str) -> None:
try:
root = tk.Tk()
root.withdraw()
root.attributes("-topmost", True)
messagebox.showerror(title, message, parent=root)
root.destroy()
except Exception as dialog_err:
_log.error("%s: %s", title, message)
_log.warning("Could not show messagebox: %s", dialog_err)
def _validate_deployment_layout() -> None:
"""Fail fast on a new machine when required folders/files are missing."""
issues: list[str] = []
example = PROJECT_ROOT / "Input" / "input.xlsx.example"
if not CONFIG_FILE.is_file():
hint = (
f"Copy and rename:\n {example}\n ->\n {CONFIG_FILE}"
if example.is_file()
else f"Create config file:\n {CONFIG_FILE}"
)
issues.append(f"Missing Input\\input.xlsx\n{hint}")
chrome_exe = PROJECT_ROOT / "chrome-win" / "chrome-win" / "chrome.exe"
chrome_driver = PROJECT_ROOT / "chrome-win" / "chrome-win" / "chromedriver.exe"
if not chrome_exe.is_file():
issues.append(
"Missing portable browser folder:\n"
f" {chrome_exe}\n"
"Copy the entire chrome-win\\ folder from the original machine "
"into the same folder as MISF_VOB_Download.exe."
)
elif not chrome_driver.is_file():
issues.append(
"Missing chromedriver next to portable chrome.exe:\n"
f" {chrome_driver}"
)
if issues:
message = (
"This PC is not set up to run MISF VOB Download yet.\n\n"
+ "\n\n".join(issues)
+ "\n\nSee README_PRODUCTION.txt in the app folder."
)
_show_error_dialog("MISF VOB Download — Setup Required", message)
raise FileNotFoundError(message.split("\n", 1)[0])
def _close_blank_tabs(driver) -> None:
"""Close any about:blank / empty tabs left by the Microsoft auth flow."""
try:
active = driver.current_window_handle
for handle in list(driver.window_handles):
if handle == active:
continue
try:
driver.switch_to.window(handle)
url = (driver.current_url or "").strip().lower()
if url in ("about:blank", "", "data:,"):
driver.close()
_log.info("Closed blank tab (%s)", handle)
except Exception:
pass
driver.switch_to.window(active)
except Exception as exc:
_log.warning("_close_blank_tabs: %s", exc)
def _quit_browser(
driver, *, keep_browser: bool, is_proxy_error: bool = False
) -> None:
"""Close the browser unless keep_browser is set or a proxy error occurred."""
if keep_browser or is_proxy_error:
return
stop_chrome_browser(driver)
log_step("Browser closed")
def _failure_status(exc: Exception) -> str:
if isinstance(exc, FacilityNotFoundError):
return "Not appear"
err_msg = str(exc).lower()
if "no planner plan found" in err_msg or "plan not found" in err_msg:
return "Plan not found"
if "no files" in err_msg or "attachment" in err_msg:
return "No attachments"
return "FAILED"
def _success_status(
file_count: int,
tasks_total: int,
*,
tasks_with_downloads: int = 0,
tasks_with_attachments: int = 0,
tasks_skipped_no_attachment: int = 0,
) -> str:
if tasks_total <= 0:
return "No tasks"
if tasks_with_attachments <= 0:
if tasks_skipped_no_attachment > 0 or tasks_total > 0:
return "No attachments"
return "No tasks"
if tasks_with_downloads <= 0 or file_count <= 0:
return "Download failed"
if tasks_with_downloads >= tasks_with_attachments:
return "Success"
return "Partial"
def _format_status_changed(
tasks_with_downloads: int,
status_changed: int,
*,
via_task_details: int = 0,
) -> str:
if tasks_with_downloads <= 0:
return "N/A"
if status_changed >= tasks_with_downloads:
prefix = "Yes"
elif status_changed > 0:
prefix = "Partial"
else:
prefix = "No"
label = f"{prefix} - {status_changed}/{tasks_with_downloads} tasks"
if via_task_details > 0:
label += f" ({via_task_details} via task details)"
return label
def _write_download_summary(
results: dict[str, dict[str, object]],
failures: list[tuple[str, str, str]],
summary_path: Path,
*,
planner_url: str,
run_timestamp: str,
) -> None:
failure_names = {facility for facility, _, _ in failures}
rows = []
for facility, payload in results.items():
if facility in failure_names:
continue
original_paths = list(payload.get("original", []) or [])
file_count = len(original_paths)
tasks_total = int(payload.get("tasks_total", 0) or 0)
tasks_with_downloads = int(payload.get("tasks_with_downloads", 0) or 0)
tasks_with_attachments = int(payload.get("tasks_with_attachments", 0) or 0)
tasks_skipped_no_attachment = int(
payload.get("tasks_skipped_no_attachment", 0) or 0
)
status_changed = int(payload.get("status_changed", 0) or 0)
via_task_details = int(payload.get("status_via_task_details", 0) or 0)
rows.append({
"Run Timestamp": run_timestamp,
"Facility Name": facility,
"Files Downloaded": file_count,
"Status": _success_status(
file_count,
tasks_total,
tasks_with_downloads=tasks_with_downloads,
tasks_with_attachments=tasks_with_attachments,
tasks_skipped_no_attachment=tasks_skipped_no_attachment,
),
"Status Changed": _format_status_changed(
tasks_with_downloads,
status_changed,
via_task_details=via_task_details,
),
})
for facility, err, status in failures:
rows.append({
"Run Timestamp": run_timestamp,
"Facility Name": facility,
"Files Downloaded": 0,
"Status": status,
"Status Changed": "N/A",
})
summary_path.parent.mkdir(parents=True, exist_ok=True)
pd.DataFrame(rows).to_excel(summary_path, index=False, sheet_name="Download")
_log.info("Download summary -> %s", summary_path)
def _unique_workable_filename(workable_root: Path, base_name: str, ext: str) -> Path:
"""
Allocate a unique file path inside Workable/ using _N suffix when needed.
"""
ext = ext if ext.startswith(".") else f".{ext}"
first = workable_root / f"{base_name}{ext}"
if not first.exists():
return first
for n in range(2, 1000):
candidate = workable_root / f"{base_name}_{n}{ext}"
if not candidate.exists():
return candidate
raise RuntimeError(f"Could not allocate workable filename for {base_name}{ext}")
def _build_workable_files(
*,
facility_name: str,
downloaded: list[Path],
output_root: Path,
today_iso: str,
) -> tuple[list[Path], Path]:
"""
Copy downloaded files into Output/<date>/Workable with standardized names.
- No facility folders are created under Workable.
- Files are written directly under Workable/ with unique names.
"""
ddmmyyyy = datetime.strptime(today_iso, "%Y-%m-%d").strftime("%d-%m-%Y")
fac_safe = _sanitize_folder_name(facility_name)
workable_root = output_root / today_iso / "Workable"
medicare_root = output_root / today_iso / "Workable_Medicare"
workable_paths: list[Path] = []
if not downloaded:
return workable_paths, workable_root
workable_root.mkdir(parents=True, exist_ok=True)
medicare_root.mkdir(parents=True, exist_ok=True)
for i, src in enumerate(downloaded, start=1):
ext = src.suffix or ".pdf"
base = f"{ddmmyyyy}_{fac_safe}" if len(downloaded) == 1 else f"{ddmmyyyy}_{fac_safe}_{i}"
dst = _unique_workable_filename(workable_root, base, ext)
try:
shutil.copy2(src, dst)
except OSError as exc:
_log.error("Failed to copy %s -> %s: %s", src.name, dst, exc)
raise
workable_paths.append(dst)
med_dst = medicare_root / dst.name
if med_dst.exists():
med_dst = _unique_workable_filename(medicare_root, dst.stem, ext)
try:
shutil.copy2(dst, med_dst)
except OSError as exc:
_log.error("Failed to mirror %s -> %s: %s", dst.name, med_dst, exc)
raise
_log.info(
"Workable files copied (%d) -> %s and %s",
len(workable_paths),
workable_root,
medicare_root,
)
return workable_paths, workable_root
def _prune_empty_date_output_dirs(output_root: Path, today_iso: str) -> None:
"""Remove empty Original/Workable day folders when no files were saved."""
for sub in ("Original", "Workable", "Workable_Medicare"):
root = output_root / today_iso / sub
if not root.is_dir():
continue
for child in list(root.iterdir()):
if child.is_dir() and not any(child.iterdir()):
try:
child.rmdir()
except OSError:
pass
try:
if not any(root.iterdir()):
root.rmdir()
except OSError:
pass
def _remove_screenshots_under_output(output_root: Path) -> int:
"""Delete legacy _screenshots folders under Output (Original/Workable/etc.)."""
removed = 0
if not output_root.is_dir():
return removed
for shots_dir in output_root.rglob("_screenshots"):
if not shots_dir.is_dir():
continue
try:
shutil.rmtree(shots_dir)
removed += 1
_log.info("Removed screenshot folder: %s", shots_dir)
except Exception as exc:
_log.warning("Could not remove %s: %s", shots_dir, exc)
return removed
def _filter_facilities(
facilities: list[str],
*,
facility_limit: int,
facility_name: str | None,
) -> list[str]:
if facility_name:
name = facility_name.strip()
matches = [f for f in facilities if f.lower() == name.lower()]
if not matches:
raise ValueError(
f"Facility '{name}' not found on Facilities sheet. "
f"Loaded {len(facilities)} names."
)
return matches
if facility_limit > 0:
return facilities[:facility_limit]
return facilities
def run_download_workflow(
*,
facility_limit: int = 0,
facility_name: str | None = None,
resume: bool = False,
keep_browser: bool = False,
screenshot_on_error: bool = False,
) -> None:
"""
Download face-sheet PDFs from Planner for configured facilities.
"""
reset_run_state()
run_timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
log_path = setup_logging(PROJECT_ROOT)
_agent_log(
"A",
"src/planner_download.py:run_download_workflow",
"Run started",
{
"limit": facility_limit,
"facility": facility_name or "",
"resume": bool(resume),
"keep_browser": bool(keep_browser),
},
)
_log.info("=" * 66)
_log.info(" MISF OneDrive / Planner Download — Start")
_log.info(" Run date: %s", TODAY_ISO_DATE)
_log.info(" Log file: %s", log_path)
_log.info("=" * 66)
log_step("Run started", date=TODAY_ISO_DATE, log=str(log_path))
driver = None
checkpoint_path = PROJECT_ROOT / "Output" / TODAY_ISO_DATE / "checkpoint.json"
try:
_validate_deployment_layout()
cfg: AppConfig = read_app_config()
log_step("Config loaded", facilities="from input.xlsx", planner_url=(cfg.planner_url or "")[:120])
_agent_log(
"B",
"src/planner_download.py:run_download_workflow",
"Config loaded",
{
"planner_url_prefix": (cfg.planner_url or "")[:120],
"proxy_set": bool(cfg.proxy_ip and cfg.proxy_port),
"pdf_folder_set": bool(cfg.pdf_folder_path.strip()),
},
)
missing = missing_onedrive_credentials(
cfg.onedrive_username,
cfg.onedrive_password,
cfg.microsoft_totp_secret,
)
if missing:
raise ValueError(credentials_setup_message(missing))
validate_totp_secret(cfg.microsoft_totp_secret)
configure_http_download_proxy(
cfg.proxy_ip,
cfg.proxy_port,
cfg.proxy_username,
str(cfg.proxy_password) if cfg.proxy_password else None,
)
# Always download into Output/<date>/Original/<facility>/ to keep originals.
output_root = PROJECT_ROOT / "Output"
removed_shots = _remove_screenshots_under_output(output_root)
if removed_shots:
log_step("Removed legacy screenshot folders", count=removed_shots)
original_root = output_root / TODAY_ISO_DATE / "Original"
failure_log_dir = output_root / TODAY_ISO_DATE / "logs" / "failures"
set_default_download_directory(original_root, create=False)
facilities = read_facility_list()
facilities = _filter_facilities(
facilities,
facility_limit=facility_limit,
facility_name=facility_name,
)
start_index = 0
completed: set[str] = set()
if resume:
ckpt = load_checkpoint(checkpoint_path)
start_index = int(ckpt.get("last_completed_index", 0) or 0)
completed = {str(x) for x in (ckpt.get("completed_facilities", []) or [])}
if start_index > 0 or completed:
_log.info(
"Resuming after facility index %d (%d remaining); %d completed cached",
start_index,
max(0, len(facilities) - start_index),
len(completed),
)
else:
clear_checkpoint(checkpoint_path)
if start_index >= len(facilities):
_log.warning("Checkpoint past end of facility list — starting from beginning")
start_index = 0
clear_checkpoint(checkpoint_path)
log_step("Start browser (Chromium remote debug)")
_log.info("Starting Chromium (Chrome-compatible)...")
if not start_chrome_debug(
cfg.proxy_ip,
cfg.proxy_port,
cfg.proxy_username,
cfg.proxy_password,
):
_agent_log(
"C",
"src/planner_download.py:run_download_workflow",
"Browser failed to start",
{},
)
raise RuntimeError("Browser failed to start.")
_agent_log(
"C",
"src/planner_download.py:run_download_workflow",
"Browser started",
{},
)
driver = attach_browser(create_download_dir=False)
if driver is None:
_agent_log(
"C",
"src/planner_download.py:run_download_workflow",
"Browser attach failed",
{},
)
raise RuntimeError("Browser attach failed.")
log_step("Browser attached")
_log.info("Browser attached.")
consolidate_browser_to_single_tab(driver)
if cfg.proxy_username and cfg.proxy_password:
log_step("Dismiss proxy auth popup")
dismiss_proxy_auth_popup(
driver,
cfg.proxy_username.strip(),
str(cfg.proxy_password),
wait_secs=8.0,
max_attempts=3,
)
time.sleep(2)
log_step("Login and open Planner")
login_and_open_planner(
driver,
cfg.onedrive_username,
cfg.onedrive_password,
cfg.microsoft_totp_secret,
planner_url=cfg.planner_url,
)
log_url(driver, "after_login")
_close_blank_tabs(driver)
_agent_log(
"D",
"src/planner_download.py:run_download_workflow",
"Planner login returned",
{"url": (driver.current_url or "")[:160]},
)
results: dict[str, dict[str, object]] = {}
failures: list[tuple[str, str, str]] = []
total_files = 0
for idx, fac in enumerate(facilities, start=1):
if idx <= start_index:
continue
if fac in completed:
_log.info("Skipping already completed facility: %s", fac)
continue
log_step("Facility loop", index=f"{idx}/{len(facilities)}", facility=fac)
_log.info("Facility %d/%d: %s", idx, len(facilities), fac)
dest = original_root / _sanitize_folder_name(fac)
_agent_log(
"E",
"src/planner_download.py:facility_loop",
"Facility start",
{"index": idx, "facility": fac[:80]},
)
try:
facility_result: FacilityRunResult = process_facility(
driver,
fac,
dest,
screenshot_on_error=screenshot_on_error,
failure_log_dir=failure_log_dir if screenshot_on_error else None,
)
original_paths = facility_result.downloaded
file_count = len(original_paths)
comment = ""
if original_paths:
workable_paths, workable_root = _build_workable_files(
facility_name=fac,
downloaded=original_paths,
output_root=output_root,
today_iso=TODAY_ISO_DATE,
)
else:
workable_paths = []
workable_root = output_root / TODAY_ISO_DATE / "Workable"
comment = "0 files downloaded"
_remove_empty_directory(dest)
results[fac] = {
"original": original_paths,
"workable": workable_paths,
"comment": comment,
"tasks_total": facility_result.tasks_total,
"tasks_with_downloads": facility_result.tasks_with_downloads,
"tasks_with_attachments": facility_result.tasks_with_attachments,
"tasks_skipped_no_attachment": facility_result.tasks_skipped_no_attachment,
"status_changed": facility_result.status_changed,
"status_via_task_details": facility_result.status_via_task_details,
}
total_files += file_count
log_step(
"Facility complete",
facility=fac,
files=file_count,
tasks_with_downloads=str(facility_result.tasks_with_downloads),
status_changed=f"{facility_result.status_changed}/{facility_result.tasks_with_downloads}",
via_task_details=str(facility_result.status_via_task_details),
original=str(dest),
workable=str(workable_root),
)
_log.info(
"%s: %d original file(s), status %d/%d tasks (%d via task details) -> %s ; workable -> %s",
fac,
file_count,
facility_result.status_changed,
facility_result.tasks_with_downloads,
facility_result.status_via_task_details,
dest,
workable_root,
)
completed.add(fac)
log_step("Checkpoint saved", facility=fac, index=str(idx))
save_checkpoint(
checkpoint_path,
last_completed_index=idx,
facility_name=fac,
completed_facilities=sorted(completed),
)
except Exception as exc:
err_msg = str(exc)
status = _failure_status(exc)
log_step("Facility failed", facility=fac, error=err_msg[:200], status=status)
_log.error("%s: %s", fac, err_msg)
if screenshot_on_error and driver is not None:
save_failure_screenshot(driver, fac, failure_log_dir=failure_log_dir)
failures.append((fac, err_msg, status))
results[fac] = {
"original": [],
"workable": [],
"comment": err_msg[:500],
"tasks_total": 0,
"tasks_with_downloads": 0,
"tasks_with_attachments": 0,
"tasks_skipped_no_attachment": 0,
"status_changed": 0,
"status_via_task_details": 0,
}
save_checkpoint(
checkpoint_path,
last_completed_index=idx,
facility_name=fac,
completed_facilities=sorted(completed),
)
summary_path = (
PROJECT_ROOT / "Output" / TODAY_ISO_DATE / "Summary"
/ f"Download_Summary_{TODAY_ISO_DATE}.xlsx"
)
_write_download_summary(
results,
failures,
summary_path,
planner_url=cfg.planner_url,
run_timestamp=run_timestamp,
)
_prune_empty_date_output_dirs(output_root, TODAY_ISO_DATE)
clear_checkpoint(checkpoint_path)
log_step("Run complete", files=total_files, failures=len(failures), summary=str(summary_path))
_log.info("=" * 66)
_log.info(
" DOWNLOAD COMPLETE — %d file(s), %d failure(s)",
total_files,
len(failures),
)
_log.info("=" * 66)
except Exception as exc:
_log.critical("Run failed: %s", exc)
traceback.print_exc()
err_lower = str(exc).lower()
is_proxy_error = any(
kw in err_lower
for kw in ("proxy", "407", "auth", "intercepted", "credential", "popup")
)
if is_proxy_error or keep_browser:
user_msg = (
"An error occurred during OneDrive/Planner download.\n\n"
"The browser has been left open for inspection.\n\n"
f"Error:\n{str(exc)[:300]}"
)
dialog_title = "Download Error — Browser Left Open"
else:
user_msg = (
"An error occurred during OneDrive/Planner download.\n\n"
f"Error:\n{str(exc)[:300]}\n\n"
f"Details: Output\\{TODAY_ISO_DATE}\\logs\\download.log"
)
dialog_title = "Download Error"
_show_error_dialog(dialog_title, user_msg)
_log.info("PROCESS ENDED WITH ERROR")
_quit_browser(driver, keep_browser=keep_browser, is_proxy_error=is_proxy_error)
return
if not keep_browser:
if driver is not None:
_log.info("Closing browser...")
_quit_browser(driver, keep_browser=False)
driver = None
else:
_log.info("Browser left open (--keep-browser)")
_log.info("PROCESS ENDED")
def build_arg_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description="Download face-sheet PDFs from OneDrive / Planner.",
)
parser.add_argument(
"--limit",
type=int,
default=0,
help="Process only the first N facilities (e.g. --limit 2 for a dry-run).",
)
parser.add_argument(
"--facility",
type=str,
default="",
help="Process a single facility by exact name (case-insensitive match).",
)
parser.add_argument(
"--resume",
action="store_true",
help="Resume from last completed facility (checkpoint file).",
)
parser.add_argument(
"--keep-browser",
action="store_true",
help="Do not close the browser when the run finishes.",
)
parser.add_argument(
"--screenshot",
action="store_true",
help="Save failure screenshots to Output/<date>/logs/failures/ only.",
)
parser.add_argument(
"--no-screenshot",
action="store_true",
help=argparse.SUPPRESS,
)
return parser
+40
View File
@@ -0,0 +1,40 @@
"""Step-by-step run logging for MISF automation (every major UI action)."""
from __future__ import annotations
import logging
from typing import Any
_RUN = logging.getLogger("misf.run")
def log_step(message: str, **fields: Any) -> None:
"""
Log one automation step. Use for every meaningful UI/browser action.
Examples:
log_step("Open My plans -> Shared")
log_step("Filter facility", facility="Amaya", rows=2)
"""
extras = []
for key, value in fields.items():
if value is None:
continue
text = str(value).strip()
if not text:
continue
# Avoid huge URLs in logs
if len(text) > 240:
text = text[:237] + "..."
extras.append(f"{key}={text}")
suffix = f" | {' | '.join(extras)}" if extras else ""
_RUN.info("[STEP] %s%s", message, suffix)
def log_url(driver, label: str = "current") -> None:
"""Log current browser URL when driver is available."""
try:
url = (driver.current_url or "")[:240]
if url:
log_step(f"URL ({label})", url=url)
except Exception:
pass
+33
View File
@@ -0,0 +1,33 @@
"""Microsoft Authenticator TOTP secret sanitization and validation."""
from __future__ import annotations
import re
_BASE32_CHARS = frozenset("ABCDEFGHIJKLMNOPQRSTUVWXYZ234567")
def sanitize_totp_secret(raw: str) -> str:
"""Normalize Base32 secret: strip separators, uppercase, map 0→O and 1→I."""
secret = re.sub(r"[\s\-]", "", (raw or "").strip()).upper().rstrip("=")
return secret.translate(str.maketrans({"0": "O", "1": "I"}))
def invalid_totp_characters(secret: str) -> list[str]:
return sorted({c for c in secret if c not in _BASE32_CHARS})
def validate_totp_secret(raw: str) -> str:
"""
Return sanitized secret or raise ValueError with a clear message.
"""
sanitized = sanitize_totp_secret(raw)
if not sanitized:
raise ValueError("Microsoft TOTP Secret is empty (check input.xlsx or .env).")
invalid = invalid_totp_characters(sanitized)
if invalid:
raise ValueError(
"Microsoft TOTP Secret is not valid Base32. "
f"Invalid character(s): {', '.join(invalid[:10])}. "
"Fix the secret (Base32 uses A-Z and 2-7 only)."
)
return sanitized
+54
View File
@@ -0,0 +1,54 @@
"""Tests for multi-attachment URL collection."""
from __future__ import annotations
from onedrive_planner import (
_attachment_tab_count,
_merge_attachment_hrefs,
_should_mark_task_in_progress,
)
SAMPLE_A = (
"https://netorg393890.sharepoint.com/sites/ArborHills/Shared%20Documents/"
"RICHARDSON%20ARBRHLS%205.26.26.pdf"
)
SAMPLE_B = (
"https://netorg393890.sharepoint.com/sites/ArborHills/Shared%20Documents/"
"RICHARDSON%20ARBRHLS%206.4.26.pdf"
)
class TestAttachmentMulti:
def test_merge_keeps_distinct_pdf_urls(self):
merged = _merge_attachment_hrefs(
[SAMPLE_A],
[SAMPLE_B],
[SAMPLE_A],
)
assert len(merged) == 2
assert SAMPLE_A in merged
assert SAMPLE_B in merged
def test_merge_dedupes_same_filename(self):
merged = _merge_attachment_hrefs([SAMPLE_A], [SAMPLE_A + "?download=1"])
assert len(merged) == 1
def test_mark_status_only_after_all_files(self, tmp_path):
one = tmp_path / "a.pdf"
two = tmp_path / "b.pdf"
one.write_bytes(b"x")
two.write_bytes(b"y")
assert not _should_mark_task_in_progress([one], 2)
assert _should_mark_task_in_progress([one, two], 2)
assert not _should_mark_task_in_progress([], 2)
assert _should_mark_task_in_progress([one], 0)
def test_attachment_tab_count_parses_number(self):
from unittest.mock import MagicMock
tab = MagicMock()
tab.is_displayed.return_value = True
tab.text = "Attachments (2)"
tab.get_attribute.return_value = ""
driver = MagicMock()
driver.find_elements.return_value = [tab]
assert _attachment_tab_count(driver) == 2
+100
View File
@@ -0,0 +1,100 @@
"""Unit tests for SharePoint URL extraction from attachment rows."""
from __future__ import annotations
from unittest.mock import MagicMock
import base64
from onedrive_planner import (
_attachment_filename_key,
_direct_download_url,
_download_href_via_browser_fetch,
_extract_sharepoint_url,
_is_pdf_bytes,
_resolve_attachment_click_target,
_sharepoint_download_candidates,
_url_from_attachment_element,
_viewer_preview_url,
)
SAMPLE = (
"https://netorg393890.sharepoint.com/sites/AnberryNursingandRehabilitationCenter/"
"Shared%20Documents/ANBERRY%20FS%205.29.26.pdf"
)
class TestAttachmentUrl:
def test_extract_from_aria_label(self):
assert _extract_sharepoint_url(SAMPLE) == SAMPLE
def test_direct_download_transform(self):
dl = _direct_download_url(SAMPLE)
assert "download=1" in dl
def test_viewer_preview_url(self):
viewer = _viewer_preview_url(SAMPLE + "?web=1")
assert "web=1" in viewer
assert "download=1" not in viewer
viewer2 = _viewer_preview_url(SAMPLE)
assert "web=1" in viewer2
def test_url_from_span_element(self):
span = MagicMock()
span.get_attribute.side_effect = lambda name: (
SAMPLE if name == "aria-label" else None
)
span.text = ""
span.find_elements.return_value = []
assert _url_from_attachment_element(span) == SAMPLE
def test_sharepoint_download_candidates_include_layouts(self):
cands = _sharepoint_download_candidates(SAMPLE)
assert any("download.aspx" in c for c in cands)
assert any("download=1" in c for c in cands)
def test_is_pdf_bytes(self):
assert _is_pdf_bytes(b"%PDF-1.4 fake content " + b"x" * 6000)
assert not _is_pdf_bytes(b"%PDF-1.4 tiny " + b"x" * 100)
assert not _is_pdf_bytes(b"<html>viewer</html>")
def test_resolve_click_target_prefers_href_span(self):
span = MagicMock()
span.is_displayed.return_value = True
span.tag_name = "span"
span.get_attribute.side_effect = lambda name: (
f"JACKSON ARBRHLS 6.3.26.pdf {SAMPLE}" if name == "aria-label" else None
)
span.text = ""
span.find_elements.return_value = []
li = MagicMock()
li.tag_name = "li"
li.is_displayed.return_value = True
li.get_attribute.return_value = None
li.text = "JACKSON ARBRHLS 6.3.26.pdf"
li.find_elements.return_value = [span]
assert _resolve_attachment_click_target(li) is span
assert "6.3.26.pdf" in _attachment_filename_key(span)
def test_browser_fetch_saves_pdf(self, tmp_path):
driver = MagicMock()
pdf = b"%PDF-1.4 " + b"x" * 6000
driver.execute_async_script.return_value = {
"ok": True,
"b64": base64.b64encode(pdf).decode("ascii"),
"ct": "application/pdf",
"size": len(pdf),
}
driver.current_window_handle = "h1"
driver.window_handles = ["h1"]
def _exec_script(script, *args):
if "window.open" in str(script):
driver.window_handles = ["h1", "h2"]
driver.execute_script.side_effect = _exec_script
out = _download_href_via_browser_fetch(driver, tmp_path, SAMPLE, set())
assert len(out) == 1
assert out[0].exists()
assert _is_pdf_bytes(out[0].read_bytes())
+55
View File
@@ -0,0 +1,55 @@
"""Tests for board column detection helpers."""
from __future__ import annotations
from unittest.mock import MagicMock
from selenium.common.exceptions import InvalidSelectorException
from onedrive_planner import (
_FIRST_COLUMN_CLUSTER_WIDTH_PX,
_element_in_board_container,
_parent_element,
)
def test_first_column_cluster_width_reasonable():
assert _FIRST_COLUMN_CLUSTER_WIDTH_PX >= 150
def test_parent_element_stops_at_html_document_error():
"""ChromeDriver rejects .. on <html> with InvalidSelectorException."""
html = MagicMock()
html.tag_name = "html"
html.find_element.side_effect = InvalidSelectorException(
'The result of the xpath expression ".." is: [object HTMLDocument].'
)
assert _parent_element(html) is None
def test_parent_element_returns_parent_for_body():
body = MagicMock()
body.tag_name = "body"
parent = MagicMock()
parent.tag_name = "html"
body.find_element.return_value = parent
assert _parent_element(body) is parent
def test_element_in_board_container_does_not_crash_at_root():
"""Containment walk must not raise when ancestors hit document root."""
child = MagicMock()
child.tag_name = "div"
container = MagicMock()
container.tag_name = "div"
html = MagicMock()
html.tag_name = "html"
html.find_element.side_effect = InvalidSelectorException("document root")
body = MagicMock()
body.tag_name = "body"
body.find_element.return_value = html
child.find_element.return_value = body
assert _element_in_board_container(child, container) is False
+147
View File
@@ -0,0 +1,147 @@
"""Unit tests for Chromium-only browser resolution."""
from __future__ import annotations
from pathlib import Path
from unittest.mock import MagicMock, patch
from chrome import (
PROJECT_ROOT,
_allow_google_chrome_fallback,
_browser_runs_in_background,
_bundled_chromedriver_for,
_google_chrome_fallback_paths,
_installed_chromium_paths,
_portable_chromium_executable,
_prefer_installed_chromium,
capture_fullscreen_screenshot,
find_chromium_executable,
)
def test_installed_paths_exclude_google():
installed = _installed_chromium_paths()
chrome_fb = _google_chrome_fallback_paths()
assert not any("Google" in p for p in installed)
assert any("Google" in p for p in chrome_fb)
def test_portable_chrome_win_preferred_over_installed():
portable = str(PROJECT_ROOT / "chrome-win" / "chrome-win" / "chrome.exe")
installed = r"C:\Program Files\Chromium\Application\chrome.exe"
def fake_isfile(path: str) -> bool:
return path in {portable, installed}
env = {"USERPROFILE": r"C:\Users\test", "LOCALAPPDATA": r"C:\Users\test\AppData\Local"}
with patch("chrome._portable_chromium_executable", return_value=portable):
with patch("chrome.os.path.isfile", side_effect=fake_isfile):
with patch.dict("os.environ", env, clear=True):
assert find_chromium_executable() == portable
def test_installed_used_when_portable_missing():
installed = r"C:\Program Files\Chromium\Application\chrome.exe"
def fake_isfile(path: str) -> bool:
return path == installed
env = {"USERPROFILE": r"C:\Users\test", "LOCALAPPDATA": r"C:\Users\test\AppData\Local"}
with patch("chrome._portable_chromium_executable", return_value=None):
with patch("chrome.os.path.isfile", side_effect=fake_isfile):
with patch.dict("os.environ", env, clear=True):
assert find_chromium_executable() == installed
def test_misf_use_installed_chrome_skips_portable():
portable = str(PROJECT_ROOT / "chrome-win" / "chrome-win" / "chrome.exe")
installed = r"C:\Program Files\Chromium\Application\chrome.exe"
def fake_isfile(path: str) -> bool:
return path in {portable, installed}
env = {
"MISF_USE_INSTALLED_CHROME": "1",
"USERPROFILE": r"C:\Users\test",
"LOCALAPPDATA": r"C:\Users\test\AppData\Local",
}
with patch("chrome._portable_chromium_executable", return_value=portable):
with patch("chrome.os.path.isfile", side_effect=fake_isfile):
with patch.dict("os.environ", env, clear=True):
assert find_chromium_executable() == installed
def test_find_chromium_can_use_google_with_allow_flag():
google_path = r"C:\Program Files\Google\Chrome\Application\chrome.exe"
env = {
"MISF_ALLOW_CHROME": "1",
"USERPROFILE": r"C:\Users\test",
"LOCALAPPDATA": r"C:\Users\test\AppData\Local",
}
with patch("chrome._portable_chromium_executable", return_value=None):
with patch("chrome.os.path.isfile", side_effect=lambda p: p == google_path):
with patch.dict("os.environ", env, clear=True):
assert find_chromium_executable() == google_path
def test_bundled_chromedriver_next_to_portable_chrome():
browser = PROJECT_ROOT / "chrome-win" / "chrome-win" / "chrome.exe"
if not browser.is_file():
return
driver = _bundled_chromedriver_for(str(browser))
assert driver is not None
assert driver.endswith("chromedriver.exe")
def test_portable_chromium_executable_on_disk():
exe = _portable_chromium_executable()
expected = PROJECT_ROOT / "chrome-win" / "chrome-win" / "chrome.exe"
if expected.is_file():
assert exe == str(expected)
def test_foreground_default_background_opt_in():
with patch.dict("os.environ", {}, clear=True):
assert _browser_runs_in_background() is False
with patch.dict("os.environ", {"MISF_BROWSER_BACKGROUND": "1"}, clear=True):
assert _browser_runs_in_background() is True
def test_allow_chrome_flag():
with patch.dict("os.environ", {}, clear=True):
assert _allow_google_chrome_fallback() is False
with patch.dict("os.environ", {"MISF_ALLOW_CHROME": "yes"}, clear=True):
assert _allow_google_chrome_fallback() is True
def test_prefer_installed_flag():
with patch.dict("os.environ", {}, clear=True):
assert _prefer_installed_chromium() is False
with patch.dict("os.environ", {"MISF_USE_INSTALLED_CHROME": "1"}, clear=True):
assert _prefer_installed_chromium() is True
def test_fullscreen_screenshot_maximizes_and_uses_desktop_when_not_black(tmp_path):
driver = MagicMock()
shot = tmp_path / "shot.png"
shot.write_bytes(b"\xff" * 100)
with patch.dict("os.environ", {}, clear=True):
with patch("chrome.capture_desktop_screenshot", return_value=True) as cap:
with patch("chrome._is_image_mostly_black", return_value=False):
assert capture_fullscreen_screenshot(driver, shot) is True
cap.assert_called_once_with(shot)
driver.maximize_window.assert_called()
def test_fullscreen_screenshot_falls_back_when_desktop_is_black(tmp_path):
driver = MagicMock()
shot = tmp_path / "shot.png"
shot.write_bytes(b"\x00" * 100)
with patch.dict("os.environ", {}, clear=True):
with patch("chrome.capture_desktop_screenshot", return_value=True):
with patch("chrome._is_image_mostly_black", return_value=True):
with patch("chrome._capture_browser_with_timestamp_bar", return_value=True) as fb:
assert capture_fullscreen_screenshot(driver, shot) is True
fb.assert_called_once()
+48
View File
@@ -0,0 +1,48 @@
"""Config and facility list tests (no browser)."""
from __future__ import annotations
import sys
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parent.parent
SRC = ROOT / "src"
sys.path.insert(0, str(SRC))
from config import ( # noqa: E402
read_app_config,
read_facility_list,
read_onedrive_config,
)
from onedrive_planner import _sanitize_folder_name, _xpath_literal # noqa: E402
@pytest.fixture
def example_config() -> Path:
path = ROOT / "Input" / "input.xlsx.example"
assert path.exists(), f"Missing template: {path}"
return path
def test_helpers() -> None:
assert _sanitize_folder_name("Test/Facility") == "Test_Facility"
assert _xpath_literal("Plain Name") == "'Plain Name'"
assert "concat" in _xpath_literal('O\'Brien "East" Wing')
def test_read_facility_list_skips_placeholder(example_config: Path) -> None:
facilities = read_facility_list(example_config, warn_if_not_expected_count=False)
assert facilities
assert not any("REPLACE_WITH" in f for f in facilities)
def test_read_app_config_planner_url(example_config: Path) -> None:
cfg = read_app_config(example_config)
assert "planner.cloud.microsoft" in cfg.planner_url
def test_read_onedrive_config_tuple_length(example_config: Path) -> None:
row = read_onedrive_config(example_config)
assert len(row) == 10
assert "planner.cloud.microsoft" in row[9]
+36
View File
@@ -0,0 +1,36 @@
"""Smoke tests for OneDrive download config (no browser)."""
from __future__ import annotations
import sys
from pathlib import Path
ROOT = Path(__file__).resolve().parent.parent
SRC = ROOT / "src"
sys.path.insert(0, str(SRC))
from config import CONFIG_FILE, read_app_config, read_facility_list # noqa: E402
from onedrive_planner import _sanitize_folder_name, _xpath_literal # noqa: E402
def test_helpers() -> None:
assert _sanitize_folder_name("Test/Facility") == "Test_Facility"
assert _xpath_literal("Plain Name") == "'Plain Name'"
assert _xpath_literal("O'Brien Care") == '"O\'Brien Care"'
assert "concat" in _xpath_literal('O\'Brien "East" Wing')
print("OK: helper functions")
def test_config_loads() -> None:
path = CONFIG_FILE if CONFIG_FILE.exists() else ROOT / "Input" / "input.xlsx.example"
assert path.exists(), f"Missing config: {path}"
facilities = read_facility_list(path)
assert len(facilities) >= 1
cfg = read_app_config(path)
assert cfg.planner_url
print(f"OK: {len(facilities)} facilities from {path.name}")
if __name__ == "__main__":
test_helpers()
test_config_loads()
print("All smoke tests passed.")
+105
View File
@@ -0,0 +1,105 @@
"""Tests for download summary Excel columns."""
from __future__ import annotations
from pathlib import Path
import pandas as pd
from planner_download import (
_build_workable_files,
_format_status_changed,
_success_status,
_write_download_summary,
)
def test_success_status_labels():
assert _success_status(0, 0) == "No tasks"
assert _success_status(0, 3, tasks_skipped_no_attachment=3) == "No attachments"
assert _success_status(1, 3, tasks_with_downloads=1, tasks_with_attachments=1) == "Success"
assert _success_status(2, 4, tasks_with_downloads=2, tasks_with_attachments=2) == "Success"
assert _success_status(1, 4, tasks_with_downloads=1, tasks_with_attachments=2) == "Partial"
assert _success_status(0, 2, tasks_with_attachments=2) == "Download failed"
def test_format_status_changed():
assert _format_status_changed(0, 0) == "N/A"
assert _format_status_changed(2, 2, via_task_details=2) == "Yes - 2/2 tasks (2 via task details)"
assert _format_status_changed(3, 1, via_task_details=1) == "Partial - 1/3 tasks (1 via task details)"
assert _format_status_changed(1, 0) == "No - 0/1 tasks"
assert _format_status_changed(2, 1, via_task_details=1) == "Partial - 1/2 tasks (1 via task details)"
def test_write_download_summary_columns(tmp_path: Path, monkeypatch):
monkeypatch.setattr("planner_download.PROJECT_ROOT", tmp_path)
results = {
"Alpha": {
"original": [tmp_path / "a.pdf", tmp_path / "b.pdf"],
"tasks_total": 4,
"tasks_with_attachments": 2,
"tasks_skipped_no_attachment": 2,
"tasks_with_downloads": 2,
"status_changed": 1,
"status_via_task_details": 1,
},
"Beta": {
"original": [],
"tasks_total": 2,
"tasks_with_attachments": 0,
"tasks_skipped_no_attachment": 2,
"tasks_with_downloads": 0,
"status_changed": 0,
"status_via_task_details": 0,
},
}
failures = [("Gamma", "not found", "Plan not found")]
summary_path = tmp_path / "Summary" / "Download_Summary_2026-06-05.xlsx"
_write_download_summary(
results,
failures,
summary_path,
planner_url="https://example.com",
run_timestamp="2026-06-05 12:00:00",
)
df = pd.read_excel(summary_path, keep_default_na=False)
assert list(df.columns) == [
"Run Timestamp",
"Facility Name",
"Files Downloaded",
"Status",
"Status Changed",
]
alpha = df[df["Facility Name"] == "Alpha"].iloc[0]
assert alpha["Status"] == "Success"
assert "1/2 tasks" in alpha["Status Changed"]
assert "via task details" in alpha["Status Changed"]
beta = df[df["Facility Name"] == "Beta"].iloc[0]
assert beta["Status"] == "No attachments"
assert beta["Status Changed"] == "N/A"
gamma = df[df["Facility Name"] == "Gamma"].iloc[0]
assert gamma["Status"] == "Plan not found"
assert gamma["Status Changed"] == "N/A"
def test_build_workable_files_mirrors_to_medicare(tmp_path: Path):
src_dir = tmp_path / "original" / "FacA"
src_dir.mkdir(parents=True)
pdf = src_dir / "doc.pdf"
pdf.write_bytes(b"%PDF-1.4")
paths, workable_root = _build_workable_files(
facility_name="Fac A",
downloaded=[pdf],
output_root=tmp_path / "Output",
today_iso="2026-06-16",
)
medicare_root = tmp_path / "Output" / "2026-06-16" / "Workable_Medicare"
assert len(paths) == 1
assert paths[0].parent == workable_root
assert paths[0].is_file()
assert medicare_root.is_dir()
medicare_files = list(medicare_root.glob("*.pdf"))
assert len(medicare_files) == 1
assert medicare_files[0].name == paths[0].name
assert medicare_files[0].read_bytes() == pdf.read_bytes()
+55
View File
@@ -0,0 +1,55 @@
"""Unit tests for Shared plans facility search helpers."""
from __future__ import annotations
from onedrive_planner import (
FACILITY_NOT_APPEAR_MSG,
_facility_search_terms,
_row_matches_facility_search,
)
def test_facility_search_terms_full_then_short():
terms = _facility_search_terms("Golden Hill Post Acute Care")
assert terms == [
"Golden Hill Post Acute Care",
"Golden Hill Post",
"Golden Hill",
]
def test_facility_search_terms_two_words_only():
terms = _facility_search_terms("Arbor Hills")
assert terms == ["Arbor Hills"]
def test_facility_search_terms_dedupes():
terms = _facility_search_terms("Golden Hill")
assert terms == ["Golden Hill"]
def test_row_matches_full_name_in_plan_title():
assert _row_matches_facility_search(
"Golden Hill Subacute and Rehab",
"Golden Hill Post Acute Care",
"Golden Hill",
)
def test_row_matches_exact():
assert _row_matches_facility_search(
"Arbor Hills",
"Arbor Hills",
"Arbor Hills",
)
def test_row_matches_rejects_unrelated():
assert not _row_matches_facility_search(
"Sunrise Manor",
"Golden Hill Post Acute Care",
"Golden Hill",
)
def test_not_appear_message_constant():
assert "shared plans" in FACILITY_NOT_APPEAR_MSG.lower()
+202
View File
@@ -0,0 +1,202 @@
"""Unit tests for Planner Grid view helpers."""
from __future__ import annotations
from unittest.mock import MagicMock, patch
from selenium.webdriver.common.by import By
from onedrive_planner import (
_clean_grid_task_title,
_grid_row_status,
_grid_row_task_name,
_is_attachments_empty,
_is_panel_placeholder_text,
_is_status_sorted_descending,
_parse_status_blob,
_sort_descending_menu_selected,
)
def _grid_row(*, status_value: str = "", status_text: str = "", task_name: str = ""):
row = MagicMock()
row.is_displayed.return_value = True
row.text = task_name
status_btn = MagicMock()
status_btn.is_displayed.return_value = True
status_btn.get_attribute.side_effect = lambda name: (
status_value if name == "value" else ""
)
status_btn.text = status_text
name_cell = MagicMock()
name_cell.is_displayed.return_value = True
name_cell.get_attribute.return_value = f"Task Name {task_name}"
name_span = MagicMock()
name_span.is_displayed.return_value = True
name_span.text = task_name
def find_elements(by, xpath):
xp = xpath or ""
if "combobox" in xp:
return [status_btn]
if "gridcell" in xp and "span" not in xp:
return [name_cell]
if "span" in xp:
return [name_span]
return []
row.find_elements.side_effect = find_elements
name_cell.find_elements.side_effect = lambda by, xpath: [name_span]
return row
def test_grid_row_status_not_started():
row = _grid_row(status_value="notStarted", status_text="Not started")
assert _grid_row_status(row) == "not started"
def test_grid_row_status_reads_status_column_not_bucket():
row = MagicMock()
row.is_displayed.return_value = True
bucket_btn = MagicMock()
bucket_btn.is_displayed.return_value = True
bucket_btn.get_attribute.side_effect = lambda name: "todo" if name == "value" else ""
bucket_btn.text = "To do"
status_btn = MagicMock()
status_btn.is_displayed.return_value = True
status_btn.get_attribute.side_effect = lambda name: "notStarted" if name == "value" else ""
status_btn.text = "Not started"
status_cell = MagicMock()
status_cell.is_displayed.return_value = True
status_cell.get_attribute.return_value = "Status Not started"
status_cell.text = "Not started"
status_cell.find_elements.side_effect = lambda by, xpath: (
[status_btn] if "combobox" in xpath else []
)
bucket_cell = MagicMock()
bucket_cell.is_displayed.return_value = True
bucket_cell.get_attribute.return_value = "Bucket To do"
bucket_cell.text = "To do"
bucket_cell.find_elements.side_effect = lambda by, xpath: (
[bucket_btn] if "combobox" in xpath else []
)
def find_elements(by, xpath):
xp = xpath or ""
if "gridcell" in xp:
return [status_cell, bucket_cell]
if "combobox" in xp:
return [bucket_btn, status_btn]
return []
row.find_elements.side_effect = find_elements
assert _grid_row_status(row) == "not started"
def test_grid_row_status_in_progress():
row = _grid_row(status_value="inProgress", status_text="In progress")
assert _grid_row_status(row) == "in progress"
def test_grid_row_task_name_from_aria():
row = _grid_row(task_name="Missing Insurance Info")
assert _grid_row_task_name(row) == "Missing Insurance Info"
def test_clean_grid_task_title_strips_a11y_hint():
raw = "Missing Insurance Info. Press Alt + I to open task details, or pre"
assert _clean_grid_task_title(raw) == "Missing Insurance Info"
def test_grid_row_task_name_strips_a11y_from_span():
row = MagicMock()
row.is_displayed.return_value = True
row.text = ""
name_cell = MagicMock()
name_cell.is_displayed.return_value = True
name_cell.get_attribute.return_value = ""
name_span = MagicMock()
name_span.is_displayed.return_value = True
name_span.text = "6.10 priority :). Press Alt + I to open task details"
def find_elements(by, xpath):
xp = xpath or ""
if "combobox" in xp:
return []
if "gridcell" in xp and "span" not in xp:
return [name_cell]
if "span" in xp:
return [name_span]
return []
row.find_elements.side_effect = find_elements
name_cell.find_elements.side_effect = lambda by, xpath: [name_span]
assert _grid_row_task_name(row) == "6.10 priority :)"
def test_is_panel_placeholder_text():
assert _is_panel_placeholder_text("Type a description or add notes here")
assert not _is_panel_placeholder_text("Missing Insurance Info")
def test_is_status_sorted_descending():
header = MagicMock()
header.get_attribute.side_effect = lambda name: (
"descending" if name == "aria-sort" else ""
)
assert _is_status_sorted_descending(header) is True
header.get_attribute.side_effect = lambda name: (
"ascending" if name == "aria-sort" else ""
)
assert _is_status_sorted_descending(header) is False
header.get_attribute.side_effect = lambda name: "" if name == "aria-sort" else ""
assert _is_status_sorted_descending(header) is False
def test_sort_descending_menu_selected():
driver = MagicMock()
checked_btn = MagicMock()
checked_btn.is_displayed.return_value = True
driver.find_elements.side_effect = lambda by, xpath: (
[checked_btn]
if "aria-checked='true'" in xpath
else []
)
assert _sort_descending_menu_selected(driver) is True
driver.find_elements.side_effect = lambda by, xpath: []
assert _sort_descending_menu_selected(driver) is False
def test_parse_status_blob():
assert _parse_status_blob("Not started") == "not started"
assert _parse_status_blob("Status Completed") == "completed"
assert _parse_status_blob("To do") is None
def test_is_attachments_empty_detects_empty_panel():
driver = MagicMock()
panel = MagicMock()
panel.is_displayed.return_value = True
panel.text = "Select, upload, or add a link to a file or website\nAdd attachment"
driver.find_elements.return_value = [panel]
with patch("onedrive_planner._attachment_tab_count", return_value=0), patch(
"onedrive_planner._discover_listed_attachment_names", return_value=[]
), patch("onedrive_planner._list_attachment_rows", return_value=[]):
assert _is_attachments_empty(driver) is True
def test_is_attachments_empty_false_when_tab_count():
driver = MagicMock()
with patch("onedrive_planner._attachment_tab_count", return_value=2):
assert _is_attachments_empty(driver) is False
+64
View File
@@ -0,0 +1,64 @@
"""Unit tests for Microsoft login page detection."""
from __future__ import annotations
from unittest.mock import MagicMock
from selenium.webdriver.common.by import By
from onedrive_planner import (
_email_form_visible,
_needs_microsoft_login,
_password_form_visible,
_planner_session_ready,
)
def _driver(*, url: str = "", elements: dict | None = None):
drv = MagicMock()
drv.current_url = url
elements = elements or {}
def find_elements(by, sel):
key = (by, sel)
return elements.get(key, [])
drv.find_elements.side_effect = find_elements
return drv
def _input(displayed: bool = True):
el = MagicMock()
el.is_displayed.return_value = displayed
return el
class TestLoginHelpers:
def test_password_only_login_detected(self):
drv = _driver(
url="https://login.microsoftonline.com/common/oauth2/v2.0/authorize",
elements={
(By.CSS_SELECTOR, "input[type='password']"): [_input()],
},
)
assert _password_form_visible(drv)
assert not _email_form_visible(drv)
assert _needs_microsoft_login(drv)
assert not _planner_session_ready(drv)
def test_planner_ready_when_no_login_prompt(self):
drv = _driver(
url="https://planner.cloud.microsoft/webui/mytasks/assignedtome/view/grid",
elements={},
)
assert not _needs_microsoft_login(drv)
assert _planner_session_ready(drv)
def test_planner_url_with_password_still_needs_login(self):
drv = _driver(
url="https://planner.cloud.microsoft/webui/mytasks/assignedtome/view/grid",
elements={
(By.CSS_SELECTOR, "input[type='password']"): [_input()],
},
)
assert _needs_microsoft_login(drv)
assert not _planner_session_ready(drv)
+68
View File
@@ -0,0 +1,68 @@
"""Unit tests for Planner left-nav helpers."""
from __future__ import annotations
from unittest.mock import MagicMock
from selenium.webdriver.common.by import By
from onedrive_planner import (
_left_nav_pane_collapsed,
_my_plans_nav_visible,
)
def _nav_toggle(*, expanded: str = "false", displayed: bool = True):
btn = MagicMock()
btn.is_displayed.return_value = displayed
btn.get_attribute.side_effect = lambda name: expanded if name == "aria-expanded" else ""
return btn
def _nav_link(*, displayed: bool = True, enabled: bool = True):
el = MagicMock()
el.is_displayed.return_value = displayed
el.is_enabled.return_value = enabled
return el
def _mock_driver(*, elements: dict):
drv = MagicMock()
drv.find_elements.side_effect = lambda by, sel: elements.get((by, sel), [])
return drv
class TestNavHelpers:
def test_collapsed_when_aria_expanded_false(self):
drv = _mock_driver(
elements={
(By.XPATH, "//button[@aria-label='Left navigation pane']"): [
_nav_toggle(expanded="false")
],
}
)
assert _left_nav_pane_collapsed(drv)
def test_not_collapsed_when_aria_expanded_true(self):
drv = _mock_driver(
elements={
(By.XPATH, "//button[@aria-label='Left navigation pane']"): [
_nav_toggle(expanded="true")
],
}
)
assert not _left_nav_pane_collapsed(drv)
def test_my_plans_visible(self):
drv = _mock_driver(
elements={
(
By.XPATH,
"//*[self::span or self::div][normalize-space()='My plans']",
): [_nav_link()],
}
)
assert _my_plans_nav_visible(drv)
def test_my_plans_hidden(self):
drv = _mock_driver(elements={})
assert not _my_plans_nav_visible(drv)
+23
View File
@@ -0,0 +1,23 @@
"""Tests for Planner tab URL classification."""
from __future__ import annotations
from onedrive_planner import _is_facility_plan_url, _is_planner_hub_url
def test_facility_plan_url():
assert _is_facility_plan_url(
"https://planner.cloud.microsoft/webui/plan/PID_ABC/view/board?tid=x"
)
assert not _is_facility_plan_url(
"https://planner.cloud.microsoft/webui/mytasks/assignedtome/view/grid?tid=x"
)
def test_planner_hub_url():
assert _is_planner_hub_url(
"https://planner.cloud.microsoft/webui/mytasks/assignedtome/view/grid"
)
assert not _is_planner_hub_url(
"https://planner.cloud.microsoft/webui/plan/PID_ABC/view/board"
)
assert not _is_planner_hub_url("about:blank")
+13
View File
@@ -0,0 +1,13 @@
"""Tests for step logging helper."""
from __future__ import annotations
import logging
from run_log import log_step
def test_log_step_formats_message(caplog):
caplog.set_level(logging.INFO, logger="misf.run")
log_step("Open facility", facility="Amaya", rows=2)
assert any("[STEP] Open facility" in r.message for r in caplog.records)
assert any("facility=Amaya" in r.message for r in caplog.records)
+306
View File
@@ -0,0 +1,306 @@
"""Unit tests for board task filtering (not started + attachment)."""
from __future__ import annotations
from unittest.mock import MagicMock
import pytest
from selenium.common.exceptions import NoSuchElementException
from selenium.webdriver.common.by import By
from onedrive_planner import (
_card_has_attachment,
_is_active_bucket_title,
_is_completed_tasks_header,
_is_in_active_planner_column_group,
_is_in_completed_tasks_section,
_is_probable_single_task_card,
_qualifies_not_started_with_attachment,
_task_card_matches_title,
_task_has_unchecked_complete_checkbox,
_task_is_in_progress,
_task_is_not_started,
)
def _card(
*,
text: str = "",
aria: str = "",
attachment_child: bool = False,
checkbox_checked: bool = False,
progress_status: str | None = None,
):
el = MagicMock()
el.tag_name = "div"
el.text = text
el.is_displayed.return_value = True
el.get_attribute.return_value = aria
textbox = MagicMock()
textbox.is_displayed.return_value = True
textbox.text = text
textbox.get_attribute.return_value = ""
attach = MagicMock()
attach.is_displayed.return_value = True
attach.text = ""
attach.get_attribute.return_value = "1 attachment"
checkbox = MagicMock()
checkbox.is_displayed.return_value = True
checkbox.text = ""
checkbox.get_attribute.side_effect = lambda name: (
"true" if checkbox_checked else "false"
if name == "aria-checked"
else "Complete task"
if name == "aria-label"
else ""
)
progress_btn = None
if progress_status == "in_progress":
progress_btn = MagicMock()
progress_btn.is_displayed.return_value = True
progress_btn.get_attribute.side_effect = lambda name: (
"In progress" if name == "aria-label" else ""
)
elif progress_status == "not_started":
progress_btn = MagicMock()
progress_btn.is_displayed.return_value = True
progress_btn.get_attribute.side_effect = lambda name: (
"Not started" if name == "aria-label" else ""
)
def find_element(by, xpath):
if xpath == "./..":
raise NoSuchElementException
raise NoSuchElementException
el.find_element.side_effect = find_element
def _find_elements(by, xpath):
xp = xpath or ""
if xpath == "./..":
raise NoSuchElementException
if "textbox" in xp:
return [textbox] if text else []
if "attachment-indicator" in xp:
return [attach] if attachment_child else []
if "role='checkbox'" in xp or '@role="checkbox"' in xp:
if "@aria-checked='true'" in xp or 'aria-checked="true"' in xp:
return [checkbox] if checkbox_checked else []
if "@aria-checked='false'" in xp or 'aria-checked="false"' in xp:
return [] if checkbox_checked else [checkbox]
if "aria-checked='false'" in xp and "|" in xp:
return [] if checkbox_checked else [checkbox]
return [checkbox]
if progress_btn is not None and "button" in xp and "aria-label" in xp:
if progress_status == "in_progress" and "In progress" in xp:
return [progress_btn]
if progress_status == "not_started" and "Not started" in xp:
return [progress_btn]
return []
if "Attachment" in xp or "Attach" in xp or "attachment" in xp:
return [attach] if attachment_child else []
if "completed tasks" in xp.lower():
return []
return []
el.find_elements.side_effect = _find_elements
return el
class TestTaskFilter:
def test_not_started_with_paperclip_no_label_text(self):
card = _card(text="5.29.26", aria="Task card", attachment_child=True)
assert _card_has_attachment(card)
assert _task_is_not_started(card)
assert not _task_is_in_progress(card)
def test_skips_in_progress(self):
card = _card(text="Patient", aria="In progress, 1 attachment", attachment_child=True)
assert _task_is_in_progress(card)
assert not _task_is_not_started(card)
def test_skips_half_blue_in_progress_button(self):
"""Planner half-blue ring: button aria-label='In progress' (not the complete checkbox)."""
card = _card(
text="Abaite, Stephen 1/30/54-Need Insurance",
attachment_child=True,
progress_status="in_progress",
)
from onedrive_planner import _task_has_in_progress_status_button
assert _task_has_in_progress_status_button(card)
assert _task_is_in_progress(card)
assert not _task_is_not_started(card)
assert not _qualifies_not_started_with_attachment(card, from_indicator=True)
def test_processes_not_started_without_progress_button(self):
card = _card(
text="Richardson, Steve 1/21/56-Need Insurance",
attachment_child=True,
)
from onedrive_planner import _task_has_in_progress_status_button
assert not _task_has_in_progress_status_button(card)
assert _task_is_not_started(card)
assert _qualifies_not_started_with_attachment(card)
def test_skips_mixed_checkbox_in_progress(self):
card = _card(text="6.1.26", attachment_child=True)
checkbox = card.find_elements(By.XPATH, ".//label[@role='checkbox']")[0]
checkbox.get_attribute.side_effect = lambda name: (
"mixed"
if name == "aria-checked"
else "Complete task"
if name == "aria-label"
else ""
)
from onedrive_planner import _task_is_in_progress_scoped
assert _task_is_in_progress_scoped(card)
assert not _task_is_not_started(card)
def test_skips_completed(self):
card = _card(text="Done", aria="Completed", attachment_child=True, checkbox_checked=True)
assert not _task_is_not_started(card)
def test_checkbox_unchecked_is_not_started(self):
card = _card(text="5.29.26", attachment_child=True, checkbox_checked=False)
assert _task_has_unchecked_complete_checkbox(card)
assert _task_is_not_started(card)
def test_not_started_progress_ring_checkbox_qualifies(self):
"""Planner progress ring uses role=checkbox aria-label='Not started'."""
card = _card(text="6.3.26", attachment_child=True, progress_status="not_started")
checkbox = card.find_elements(By.XPATH, ".//label[@role='checkbox']")[0]
checkbox.get_attribute.side_effect = lambda name: (
"false"
if name == "aria-checked"
else "Not started"
if name == "aria-label"
else ""
)
from onedrive_planner import _filter_reject_reason, _task_is_strictly_not_started
assert _task_has_unchecked_complete_checkbox(card)
assert _task_is_strictly_not_started(card)
assert _filter_reject_reason(card, from_indicator=True) is None
assert _qualifies_not_started_with_attachment(card, from_indicator=True)
def test_planner_button_complete_checkbox_is_not_started(self):
"""Planner 2024+ uses button[role=checkbox] aria-label='Complete task'."""
card = _card(text="6.8.26", attachment_child=True, checkbox_checked=False)
checkbox = card.find_elements(
By.XPATH,
".//label[@role='checkbox'] | .//button[@role='checkbox']",
)[0]
checkbox.get_attribute.side_effect = lambda name: (
"false"
if name == "aria-checked"
else "Complete task"
if name == "aria-label"
else ""
)
from onedrive_planner import _filter_reject_reason, _task_has_unchecked_complete_checkbox
assert _task_has_unchecked_complete_checkbox(card)
assert _filter_reject_reason(card, from_indicator=True) is None
def test_in_progress_rejected_on_board_without_opening(self):
"""Half-blue In progress ring on the card — skip at board scan."""
card = _card(
text="Endersby, Mary",
attachment_child=True,
progress_status="in_progress",
)
from onedrive_planner import (
_filter_reject_reason,
_task_card_in_progress_on_board,
_task_is_strictly_not_started,
)
assert _task_card_in_progress_on_board(card)
assert not _task_is_strictly_not_started(card)
assert _filter_reject_reason(card, from_indicator=True) == "in_progress"
assert _filter_reject_reason(card, from_indicator=False) == "in_progress"
assert not _qualifies_not_started_with_attachment(card, from_indicator=True)
def test_requires_attachment(self):
card = _card(text="5.29.26", aria="Not started")
assert not _card_has_attachment(card)
def test_completed_header_matcher(self):
assert _is_completed_tasks_header("Completed tasks")
assert _is_completed_tasks_header("Completed tasks 37")
assert not _is_completed_tasks_header("To do 5.29.26 Completed tasks section")
def test_indicator_qualifies_without_board_completed_noise(self):
ind = _card(text="6.3.26", aria="1 attachment", attachment_child=True)
assert _is_probable_single_task_card(ind)
assert _qualifies_not_started_with_attachment(ind, from_indicator=True)
def test_unbucketed_active_group_not_completed_section(self):
"""Tasks in taskBoardColumnGroup-unbucketed must not match completed subsection."""
from selenium.webdriver.common.by import By
indicator = MagicMock()
active_group = MagicMock()
active_group.get_attribute.side_effect = lambda name: (
"taskBoardColumnGroup-unbucketed" if name == "data-testid" else ""
)
indicator.get_attribute.return_value = ""
def _find_elements(by, xp):
if by == By.XPATH and "taskBoardColumnGroup-" in xp:
if "contains(@data-testid,'-completed')" in xp:
return []
return [active_group]
return []
indicator.find_elements.side_effect = _find_elements
assert _is_in_active_planner_column_group(indicator)
assert not _is_in_completed_tasks_section(indicator)
def test_active_bucket_titles(self):
assert _is_active_bucket_title("To do")
assert _is_active_bucket_title("TO DO")
assert _is_active_bucket_title("New Facesheets")
assert _is_active_bucket_title("New Facesheets 12")
assert _is_active_bucket_title("New Facesheet")
assert _is_active_bucket_title("Heather gets facesheets")
assert _is_active_bucket_title("No bucket")
assert not _is_active_bucket_title("Completed tasks")
assert not _is_active_bucket_title("Completed tasks 875")
def test_bucket_column_detected(self):
col = _card(
text="To do",
aria="To do, Use tab key to access a task card, left and right arrow",
attachment_child=True,
)
from onedrive_planner import _is_planner_bucket_column
assert _is_planner_bucket_column(col)
def test_weak_title_key_ignores_paperclip_count(self):
from onedrive_planner import _is_weak_task_title_key
assert _is_weak_task_title_key("3")
assert _is_weak_task_title_key("")
assert not _is_weak_task_title_key("6.3.26 Missing Insurance Info")
def test_board_wrapper_not_single_task_card(self):
board = _card(text="To do\n6.3.26\nCompleted tasks 37", attachment_child=True)
board.find_elements.side_effect = lambda by, xpath: (
[MagicMock(is_displayed=lambda: True)] * 3
if "attachment-indicator" in xpath
else [MagicMock(is_displayed=lambda: True)] * 3
if "checkbox" in xpath
else []
)
assert not _is_probable_single_task_card(board)
def test_task_card_matches_title(self):
card = _card(text="5.29.26\n1 attachment", attachment_child=True)
assert _task_card_matches_title(card, "5.29.26")
assert not _task_card_matches_title(card, "Alvarado, Shante")
+99
View File
@@ -0,0 +1,99 @@
"""Unit tests for task-details panel status detection (scoped to panel, not board)."""
from __future__ import annotations
from unittest.mock import MagicMock
from selenium.webdriver.common.by import By
from onedrive_planner import (
_panel_progress_status_label,
_task_details_shows_in_progress,
_task_details_shows_not_started,
)
def _panel(*, status_aria: str):
panel = MagicMock()
panel.is_displayed.return_value = True
status_btn = MagicMock()
status_btn.is_displayed.return_value = True
status_btn.get_attribute.side_effect = lambda name: (
status_aria if name == "aria-label" else ""
)
title_box = MagicMock()
title_box.is_displayed.return_value = True
title_box.text = "6.5.26"
def find_elements(by, xpath):
xp = xpath or ""
if "Not started" in xp or "In progress" in xp or "Completed" in xp:
return [status_btn]
if "textbox" in xp:
return [title_box]
if "Status" in xp:
return []
return []
panel.find_elements.side_effect = find_elements
return panel
def _driver_with_panel(panel, *, board_in_progress: bool = False):
drv = MagicMock()
board_btn = MagicMock()
board_btn.is_displayed.return_value = board_in_progress
board_btn.get_attribute.side_effect = lambda name: (
"In progress" if name == "aria-label" else ""
)
def find_elements(by, xpath):
xp = xpath or ""
if "@role='dialog'" in xp or "TaskDetails" in xp or "Close task" in xp:
return [panel]
if "In progress" in xp and xp.startswith("//button"):
return [board_btn] if board_in_progress else []
if "Close" in xp:
return []
return []
drv.find_elements.side_effect = find_elements
return drv
class TestTaskPanelStatus:
def test_panel_not_started(self):
panel = _panel(status_aria="Not started")
assert _panel_progress_status_label(panel) == "not started"
def test_panel_in_progress(self):
panel = _panel(status_aria="In progress")
assert _panel_progress_status_label(panel) == "in progress"
def test_details_not_started_ignores_board_buttons(self):
panel = _panel(status_aria="Not started")
drv = _driver_with_panel(panel, board_in_progress=True)
assert _task_details_shows_not_started(drv)
assert not _task_details_shows_in_progress(drv)
def test_details_in_progress_from_panel_only(self):
panel = _panel(status_aria="In progress")
drv = _driver_with_panel(panel, board_in_progress=False)
assert _task_details_shows_in_progress(drv)
assert not _task_details_shows_not_started(drv)
def test_panel_in_progress_from_combobox_value(self):
panel = MagicMock()
panel.is_displayed.return_value = True
combo = MagicMock()
combo.is_displayed.return_value = True
combo.get_attribute.side_effect = lambda name: "inProgress" if name == "value" else ""
combo.text = ""
def find_elements(by, xpath):
xp = xpath or ""
if "combobox" in xp or "ComboBox" in xp:
return [combo]
return []
panel.find_elements.side_effect = find_elements
assert _panel_progress_status_label(panel) == "in progress"
+29
View File
@@ -0,0 +1,29 @@
"""Unit tests for TOTP secret sanitization."""
from __future__ import annotations
import sys
from pathlib import Path
import pytest
SRC = Path(__file__).resolve().parent.parent / "src"
sys.path.insert(0, str(SRC))
from totp_utils import sanitize_totp_secret, validate_totp_secret # noqa: E402
def test_sanitize_strips_separators() -> None:
assert sanitize_totp_secret("jbsw y3dp-ehpk3pxp") == "JBSWY3DPEHPK3PXP"
def test_sanitize_maps_zero_one() -> None:
assert sanitize_totp_secret("JBSW03DP") == "JBSWO3DP"
def test_validate_rejects_invalid_chars() -> None:
with pytest.raises(ValueError, match="not valid Base32"):
validate_totp_secret("not-valid-secret!")
def test_validate_accepts_standard_secret() -> None:
assert validate_totp_secret("JBSWY3DPEHPK3PXP") == "JBSWY3DPEHPK3PXP"
+149
View File
@@ -0,0 +1,149 @@
"""UiPath migration helper tests and pilot output structure checks."""
from __future__ import annotations
import json
from datetime import datetime
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parent.parent
UIPATH = ROOT / "uipath" / "MISF_Planner_Download"
HELPERS_CS = UIPATH / "Helpers"
def _read_cs(name: str) -> str:
return (HELPERS_CS / name).read_text(encoding="utf-8")
class TestUiPathProjectStructure:
def test_project_json_exists(self):
assert (UIPATH / "project.json").is_file()
def test_main_workflows_exist(self):
for name in (
"Main.xaml",
"01_ReadConfig.xaml",
"02_LoginPlanner.xaml",
"03_ProcessFacility.xaml",
"04_DownloadTaskAttachments.xaml",
"05_SetTaskInProgress.xaml",
"06_WriteSummary.xaml",
):
assert (UIPATH / name).is_file(), f"missing {name}"
def test_helpers_exist(self):
assert (UIPATH / "Helpers" / "GenerateTotp.xaml").is_file()
assert (UIPATH / "Helpers" / "SanitizeFolderName.xaml").is_file()
assert (UIPATH / "Helpers" / "TotpGenerator.cs").is_file()
def test_selectors_documented(self):
assert (UIPATH / "Selectors" / "PlannerSelectors.json").is_file()
assert (UIPATH / "Selectors" / "IFRAME_SETUP.md").is_file()
data = json.loads((UIPATH / "Selectors" / "PlannerSelectors.json").read_text())
assert "planner_iframe" in data
assert "my_plans" in data["planner_iframe"]
class TestHelperLogicMirrorsPython:
def test_sanitize_folder_name(self):
src = _read_cs("MisfHelpers.cs")
assert "SanitizeFolderName" in src
assert "InvalidFolderChars" in src
def test_totp_sanitize_matches_python(self):
src = _read_cs("MisfHelpers.cs")
assert 'Replace("0", "O")' in src
assert 'Replace("1", "I")' in src
def test_download_url_transform(self):
src = _read_cs("MisfHelpers.cs")
assert "web=1" in src
assert "download=1" in src
def test_default_planner_url(self):
src = _read_cs("MisfConstants.cs")
assert "planner.cloud.microsoft" in src
assert "0c15ad2c-19dd-4ded-9cd1-20d344386414" in src
class TestPilotOutputLayout:
"""Compare output folder layout with Python bot expectations."""
@pytest.fixture
def today_iso(self) -> str:
return datetime.now().strftime("%Y-%m-%d")
def test_output_directories_can_be_created(self, today_iso: str, tmp_path: Path):
original = tmp_path / "Output" / today_iso / "Original" / "Test_Facility"
workable = tmp_path / "Output" / today_iso / "Workable"
summary = tmp_path / "Output" / today_iso / "Summary"
for d in (original, workable, summary):
d.mkdir(parents=True, exist_ok=True)
assert original.is_dir()
assert workable.is_dir()
assert summary.is_dir()
def test_checkpoint_json_shape(self, tmp_path: Path):
ckpt = {
"last_completed_index": 2,
"last_facility": "Amaya",
"completed_facilities": ["Amaya", "Brighton Place San Diego"],
}
path = tmp_path / "checkpoint.json"
path.write_text(json.dumps(ckpt), encoding="utf-8")
loaded = json.loads(path.read_text())
assert loaded["last_completed_index"] == 2
assert isinstance(loaded["completed_facilities"], list)
class TestMainXamlExpressions:
def test_main_has_facility_limit_argument(self):
main = (UIPATH / "Main.xaml").read_text(encoding="utf-8")
assert "in_FacilityLimit" in main
assert "FilterFacilities" in main
def test_login_invokes_generate_totp(self):
login = (UIPATH / "02_LoginPlanner.xaml").read_text(encoding="utf-8")
assert "GenerateTotp.xaml" in login
assert "loginfmt" in login
def test_iframe_prefix_in_process_facility(self):
proc = (UIPATH / "03_ProcessFacility.xaml").read_text(encoding="utf-8")
assert "IFRAME" in proc
assert "My plans" in proc
def test_attachments_tab_download_flow(self):
py = (ROOT / "src" / "onedrive_planner.py").read_text(encoding="utf-8")
assert "_open_attachments_view" in py
assert "_ensure_board_scan_context" in py
assert "attachment-indicator" in py
assert "_card_has_attachment" in py
assert "_task_is_not_started" in py
assert "_open_task_via_attachment" in py
assert "_click_exact_attachment_name" in py
assert "_collect_attachment_hrefs" in py
assert "_collect_attachment_download_urls" in py
assert "_download_href_via_http" in py
assert "_download_attachment_row" in py
assert "_list_attachment_rows" in py
assert "_close_extra_download_tabs" in py
assert "_download_href_with_fallback" in py
assert "_download_one_href_browser_new_tab" in py
assert "HTTP only last" in py or "HTTP last" in py
assert "configure_http_download_proxy" in py
assert "_url_from_attachment_element" in py
assert "_scroll_attachments_panel" in py
dl = (UIPATH / "04_DownloadTaskAttachments.xaml").read_text(encoding="utf-8")
assert "Attachments" in dl
assert "sharepoint" in dl.lower()
def test_in_progress_task_details_only(self):
proc = (UIPATH / "05_SetTaskInProgress.xaml").read_text(encoding="utf-8")
assert "Task details" in proc
assert "In progress" in proc
py = (ROOT / "src" / "onedrive_planner.py").read_text(encoding="utf-8")
assert "_set_task_in_progress_from_task_details" in py
assert "_mark_task_in_progress" in py
assert "task details only" in py
assert "_set_task_in_progress_from_board(driver, card" not in py