Lubna Fathima
291507c2f7
- Add ecw_download.py and onedrive_planner_http.py for ECW and HTTP-based Planner access - Add build_exe.bat for packaging - Restart Chrome automatically on crash mid-run in planner_download.py - Detect and recover from Planner "something went wrong" error overlays - Improve board-load waiting, sort-descending menu matching, and screenshot timestamp rendering in chrome.py/onedrive_planner.py
680 lines
27 KiB
Python
680 lines
27 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
ECW-server VOB / face-sheet downloader.
|
|
|
|
Auth flow
|
|
---------
|
|
Playwright opens the ECW login page, fills credentials + OTP if required,
|
|
waits for a successful session, then harvests all cookies and closes the
|
|
browser. An httpx.Client is built from those cookies and used for every
|
|
subsequent request — no browser involved after login.
|
|
|
|
Document / download flow (pure HTTP after login)
|
|
--------------------------------------------------
|
|
1. GET the ECW document-list endpoint → HTML response
|
|
2. Parse the HTML to find per-facility <a href> download links
|
|
3. Download each file directly into Output/<date>/Original/<Facility>/
|
|
4. Copy to Workable/ and Workable_Medicare/ with the standard naming
|
|
DD-MM-YYYY_<Facility Name>[_N].pdf
|
|
5. Write Download_Summary_ECW_<date>.xlsx
|
|
|
|
Config keys in Input/input.xlsx Sheet1
|
|
----------------------------------------
|
|
ECW URL https://ecw.yourdomain.com (required)
|
|
ECW Username user@domain.com (optional — falls back to OneDrive Username)
|
|
ECW Password secret (optional — falls back to OneDrive Password)
|
|
ECW TOTP Secret BASE32SECRET (optional — for ECW-specific 2FA)
|
|
Proxy IP / Proxy Port / Proxy Username / Proxy Password (reused from existing config)
|
|
|
|
Usage
|
|
-----
|
|
python ecw_download.py
|
|
python ecw_download.py --limit 5
|
|
python ecw_download.py --facility "Arbor Hills"
|
|
python ecw_download.py --date 2026-06-19
|
|
python ecw_download.py --resume
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import logging
|
|
import re
|
|
import shutil
|
|
import sys
|
|
import time
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from urllib.parse import urljoin, urlparse
|
|
|
|
ROOT = Path(__file__).resolve().parent
|
|
SRC = ROOT / "src"
|
|
if str(SRC) not in sys.path:
|
|
sys.path.insert(0, str(SRC))
|
|
|
|
import httpx
|
|
from bs4 import BeautifulSoup
|
|
|
|
from checkpoint import clear_checkpoint, load_checkpoint, save_checkpoint
|
|
from config import (
|
|
PROJECT_ROOT,
|
|
TODAY_ISO_DATE,
|
|
AppConfig,
|
|
read_app_config,
|
|
read_facility_list,
|
|
_read_config_key_value_sheet,
|
|
)
|
|
from logging_setup import setup_logging
|
|
from run_log import log_step
|
|
|
|
_log = logging.getLogger(__name__)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ECW-specific config (on top of the existing AppConfig)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _read_ecw_config() -> dict[str, str]:
|
|
"""Read ECW-specific keys from the same input.xlsx Sheet1."""
|
|
try:
|
|
kv = _read_config_key_value_sheet()
|
|
except Exception:
|
|
kv = {}
|
|
cfg = read_app_config()
|
|
return {
|
|
# ECW server base URL e.g. "https://ecw.yourdomain.com"
|
|
"ecw_url": kv.get("ECW URL", "").strip(),
|
|
# ECW credentials — fall back to the OneDrive credentials if not set
|
|
"ecw_username": kv.get("ECW Username", "").strip() or cfg.onedrive_username,
|
|
"ecw_password": kv.get("ECW Password", "").strip() or cfg.onedrive_password,
|
|
# Optional separate TOTP for ECW; falls back to the Microsoft TOTP secret
|
|
"ecw_totp_secret": kv.get("ECW TOTP Secret", "").strip() or cfg.microsoft_totp_secret,
|
|
# Proxy (reused from existing config)
|
|
"proxy_ip": cfg.proxy_ip,
|
|
"proxy_port": cfg.proxy_port,
|
|
"proxy_username": cfg.proxy_username,
|
|
"proxy_password": str(cfg.proxy_password) if cfg.proxy_password else "",
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Step 1 — Playwright login (UI only, closes browser when done)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _generate_totp(secret: str) -> str:
|
|
import pyotp
|
|
return pyotp.TOTP(secret).now()
|
|
|
|
|
|
def login_and_harvest_cookies(ecw_cfg: dict[str, str]) -> dict[str, str]:
|
|
"""
|
|
Open the ECW login page in a headless Playwright browser, authenticate,
|
|
then return all cookies as {name: value} and close the browser.
|
|
|
|
Only this function uses a browser — everything after it is pure HTTP.
|
|
|
|
TODO (fill in once the login page HTML is shared):
|
|
- Update the CSS selectors for the username / password / submit fields
|
|
- Update the success condition (URL prefix or element that appears after login)
|
|
- Add OTP step if ECW shows a 2-FA prompt
|
|
"""
|
|
from playwright.sync_api import sync_playwright, TimeoutError as PWTimeout
|
|
|
|
base_url = ecw_cfg["ecw_url"]
|
|
username = ecw_cfg["ecw_username"]
|
|
password = ecw_cfg["ecw_password"]
|
|
totp_secret = ecw_cfg["ecw_totp_secret"]
|
|
|
|
# Build proxy dict for Playwright if configured
|
|
pw_proxy = None
|
|
if ecw_cfg.get("proxy_ip") and ecw_cfg.get("proxy_port"):
|
|
proxy_server = f"http://{ecw_cfg['proxy_ip']}:{ecw_cfg['proxy_port']}"
|
|
pw_proxy = {"server": proxy_server}
|
|
if ecw_cfg.get("proxy_username"):
|
|
pw_proxy["username"] = ecw_cfg["proxy_username"]
|
|
pw_proxy["password"] = ecw_cfg.get("proxy_password", "")
|
|
|
|
log_step("ECW login via Playwright — opening browser")
|
|
cookies: dict[str, str] = {}
|
|
|
|
with sync_playwright() as pw:
|
|
browser = pw.chromium.launch(
|
|
headless=True,
|
|
args=["--disable-blink-features=AutomationControlled"],
|
|
proxy=pw_proxy,
|
|
)
|
|
ctx = browser.new_context(
|
|
ignore_https_errors=True,
|
|
viewport={"width": 1280, "height": 900},
|
|
)
|
|
page = ctx.new_page()
|
|
|
|
# ── Navigate to login page ────────────────────────────────────────
|
|
login_url = base_url # TODO: update if login page is at a sub-path
|
|
log_step("ECW navigate to login", url=login_url[:80])
|
|
page.goto(login_url, wait_until="domcontentloaded", timeout=30_000)
|
|
|
|
# ── Fill credentials ──────────────────────────────────────────────
|
|
# TODO: update selectors to match the actual ECW login form
|
|
#
|
|
# Common patterns:
|
|
# input[name="username"] / input[name="j_username"] / #userId
|
|
# input[name="password"] / input[name="j_password"] / #userPass
|
|
# button[type="submit"] / input[type="submit"] / #login-btn
|
|
#
|
|
USERNAME_SELECTOR = "input[name='username']" # TODO
|
|
PASSWORD_SELECTOR = "input[name='password']" # TODO
|
|
SUBMIT_SELECTOR = "button[type='submit']" # TODO
|
|
|
|
page.wait_for_selector(USERNAME_SELECTOR, timeout=15_000)
|
|
page.fill(USERNAME_SELECTOR, username)
|
|
page.fill(PASSWORD_SELECTOR, password)
|
|
log_step("ECW credentials filled")
|
|
page.click(SUBMIT_SELECTOR)
|
|
|
|
# ── Handle OTP / 2-FA prompt (if ECW requires it) ────────────────
|
|
# TODO: update selector and logic based on the actual 2-FA screen
|
|
OTP_SELECTOR = "input[name='otp']" # TODO — set to "" to skip
|
|
if OTP_SELECTOR:
|
|
try:
|
|
page.wait_for_selector(OTP_SELECTOR, timeout=8_000)
|
|
otp = _generate_totp(totp_secret) if totp_secret else ""
|
|
if otp:
|
|
page.fill(OTP_SELECTOR, otp)
|
|
page.click(SUBMIT_SELECTOR)
|
|
log_step("ECW OTP submitted")
|
|
except PWTimeout:
|
|
log_step("No OTP prompt detected — continuing")
|
|
|
|
# ── Wait for successful login ─────────────────────────────────────
|
|
# TODO: update the success condition.
|
|
# Option A — wait for URL to change away from the login page:
|
|
# page.wait_for_url("**/dashboard**", timeout=30_000)
|
|
# Option B — wait for a specific element that only appears after login:
|
|
# page.wait_for_selector("#mainContent", timeout=30_000)
|
|
#
|
|
try:
|
|
# Placeholder: just wait for network idle
|
|
page.wait_for_load_state("networkidle", timeout=30_000)
|
|
log_step("ECW login appears successful", url=page.url[:80])
|
|
except PWTimeout:
|
|
_log.warning("ECW login wait timed out — harvesting cookies anyway")
|
|
|
|
# ── Harvest cookies ───────────────────────────────────────────────
|
|
for c in ctx.cookies():
|
|
cookies[c["name"]] = c["value"]
|
|
log_step("ECW cookies harvested", count=len(cookies))
|
|
|
|
browser.close()
|
|
|
|
if not cookies:
|
|
raise RuntimeError("ECW login produced no cookies — check credentials and selectors.")
|
|
|
|
return cookies
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Step 2 — Build httpx client from harvested cookies
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def build_http_client(ecw_cfg: dict[str, str], cookies: dict[str, str]) -> httpx.Client:
|
|
"""
|
|
Create a persistent httpx.Client with the ECW session cookies baked in.
|
|
All subsequent requests use this client — no browser involved.
|
|
"""
|
|
base_url = ecw_cfg["ecw_url"]
|
|
proxies = None
|
|
if ecw_cfg.get("proxy_ip") and ecw_cfg.get("proxy_port"):
|
|
proxy_url = f"http://{ecw_cfg['proxy_ip']}:{ecw_cfg['proxy_port']}"
|
|
if ecw_cfg.get("proxy_username"):
|
|
proxy_url = (
|
|
f"http://{ecw_cfg['proxy_username']}:{ecw_cfg['proxy_password']}"
|
|
f"@{ecw_cfg['proxy_ip']}:{ecw_cfg['proxy_port']}"
|
|
)
|
|
proxies = {"http://": proxy_url, "https://": proxy_url}
|
|
|
|
client = httpx.Client(
|
|
base_url=base_url,
|
|
follow_redirects=True,
|
|
timeout=60,
|
|
verify=False, # set to True in production if ECW has a valid cert
|
|
**({"proxies": proxies} if proxies else {}),
|
|
)
|
|
for name, value in cookies.items():
|
|
client.cookies.set(name, value)
|
|
|
|
log_step("httpx client ready", base_url=base_url[:60], cookies=len(cookies))
|
|
return client
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Step 3 — Fetch document list (pure HTTP)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def fetch_document_list(client: httpx.Client, run_date: str) -> str:
|
|
"""
|
|
GET (or POST) the ECW document-list endpoint for run_date and return raw HTML.
|
|
|
|
TODO: fill in once the HTML body / network tab is shared:
|
|
- ECW_DOC_LIST_PATH — the path relative to ECW_BASE_URL
|
|
- query params or POST body the server expects
|
|
- any additional headers (Referer, Accept, X-Requested-With, etc.)
|
|
"""
|
|
ECW_DOC_LIST_PATH = "" # TODO e.g. "/ecw/rd/Documents/ViewDocuments.jsp"
|
|
|
|
if not ECW_DOC_LIST_PATH:
|
|
raise NotImplementedError(
|
|
"ECW document list URL not configured.\n"
|
|
"Share the HTML body / browser Network tab and this will be filled in."
|
|
)
|
|
|
|
# TODO: adjust params / method once the request format is known
|
|
# Example GET:
|
|
# resp = client.get(ECW_DOC_LIST_PATH, params={"fromDate": run_date, "toDate": run_date})
|
|
#
|
|
# Example POST:
|
|
# resp = client.post(ECW_DOC_LIST_PATH, data={"date": run_date, "type": "VOB"})
|
|
#
|
|
resp = client.get(ECW_DOC_LIST_PATH, params={"date": run_date})
|
|
resp.raise_for_status()
|
|
log_step("Document list fetched", status=resp.status_code, bytes=len(resp.text))
|
|
return resp.text
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Step 4 — Parse HTML response (pure Python / BeautifulSoup)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def parse_facility_files(html: str, facilities: list[str]) -> dict[str, list[str]]:
|
|
"""
|
|
Parse the ECW document-list HTML and return:
|
|
{ facility_name: [absolute_download_url, ...] }
|
|
|
|
Only facilities in the configured list are returned.
|
|
|
|
TODO: update selectors once the HTML body is shared.
|
|
The two commented-out patterns below cover the most common ECW layouts.
|
|
"""
|
|
soup = BeautifulSoup(html, "html.parser")
|
|
facility_lower = {f.strip().lower(): f for f in facilities}
|
|
result: dict[str, list[str]] = {f: [] for f in facilities}
|
|
|
|
# ── Pattern A — one section / card per facility ───────────────────────
|
|
# for section in soup.select(".facility-section"): # TODO selector
|
|
# name_el = section.select_one(".facility-name") # TODO selector
|
|
# if not name_el:
|
|
# continue
|
|
# ecw_name = name_el.get_text(strip=True)
|
|
# matched = _match_facility(ecw_name, facility_lower)
|
|
# if not matched:
|
|
# continue
|
|
# for a in section.select("a[href]"):
|
|
# href = a.get("href", "")
|
|
# if _is_file_link(href):
|
|
# result[matched].append(_abs_url(client.base_url, href))
|
|
|
|
# ── Pattern B — flat table, each row = one file ───────────────────────
|
|
# for row in soup.select("table tbody tr"): # TODO selector
|
|
# cells = row.find_all("td")
|
|
# if len(cells) < 2:
|
|
# continue
|
|
# ecw_name = cells[0].get_text(strip=True) # TODO column index
|
|
# a = cells[-1].find("a", href=True) # TODO column index
|
|
# if not a:
|
|
# continue
|
|
# matched = _match_facility(ecw_name, facility_lower)
|
|
# if not matched:
|
|
# continue
|
|
# href = a["href"]
|
|
# if _is_file_link(href):
|
|
# result[matched].append(_abs_url(client.base_url, href))
|
|
|
|
_log.warning(
|
|
"parse_facility_files: selectors not yet configured — "
|
|
"share the HTML body so this can be filled in."
|
|
)
|
|
return result
|
|
|
|
|
|
def _match_facility(ecw_name: str, facility_lower: dict[str, str]) -> str | None:
|
|
"""Fuzzy match an ECW display name to a known facility name."""
|
|
key = ecw_name.strip().lower()
|
|
if key in facility_lower:
|
|
return facility_lower[key]
|
|
# Longest prefix overlap wins
|
|
candidates = [
|
|
(k, v) for k, v in facility_lower.items()
|
|
if key.startswith(k) or k.startswith(key)
|
|
]
|
|
return max(candidates, key=lambda t: len(t[0]))[1] if candidates else None
|
|
|
|
|
|
def _is_file_link(href: str) -> bool:
|
|
low = href.lower()
|
|
return (
|
|
any(low.endswith(ext) for ext in (".pdf", ".xlsx", ".xls", ".doc", ".docx"))
|
|
or any(kw in low for kw in ("download", "getfile", "getdoc", "document", "attachment"))
|
|
)
|
|
|
|
|
|
def _abs_url(base_url: str, href: str) -> str:
|
|
"""Return an absolute URL, joining with base_url if href is relative."""
|
|
if href.startswith(("http://", "https://")):
|
|
return href
|
|
return urljoin(base_url, href)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Step 5 — Download a file (pure HTTP)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def download_file(client: httpx.Client, url: str, dest_folder: Path) -> Path | None:
|
|
"""Download one file from url into dest_folder. Returns saved Path or None."""
|
|
try:
|
|
resp = client.get(url, follow_redirects=True)
|
|
resp.raise_for_status()
|
|
filename = _filename_from_response(resp, url)
|
|
dest_folder.mkdir(parents=True, exist_ok=True)
|
|
dest = _unique_path(dest_folder, filename)
|
|
dest.write_bytes(resp.content)
|
|
log_step("Downloaded", url=url[-60:], file=dest.name, bytes=len(resp.content))
|
|
return dest
|
|
except Exception as exc:
|
|
_log.warning("Download failed (%s): %s", url[-60:], exc)
|
|
return None
|
|
|
|
|
|
def _filename_from_response(resp: httpx.Response, fallback_url: str) -> str:
|
|
cd = resp.headers.get("content-disposition", "")
|
|
m = re.search(r'filename\*?=["\']?(?:UTF-8\'\')?([^"\';\r\n]+)', cd, re.IGNORECASE)
|
|
if m:
|
|
return m.group(1).strip().strip('"\'')
|
|
path = urlparse(fallback_url).path
|
|
name = path.split("/")[-1]
|
|
return name or "file.pdf"
|
|
|
|
|
|
def _unique_path(folder: Path, filename: str) -> Path:
|
|
candidate = folder / filename
|
|
if not candidate.exists():
|
|
return candidate
|
|
stem, suffix = Path(filename).stem, Path(filename).suffix
|
|
for i in range(1, 9999):
|
|
candidate = folder / f"{stem}_{i}{suffix}"
|
|
if not candidate.exists():
|
|
return candidate
|
|
return folder / filename
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Workable copy (identical logic to planner_download._build_workable_files)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _sanitize_folder_name(name: str) -> str:
|
|
return re.sub(r'[<>:"/\\|?*]', "_", name.strip()) or "unknown_facility"
|
|
|
|
|
|
def _unique_workable_path(folder: Path, base: str, ext: str) -> Path:
|
|
candidate = folder / f"{base}{ext}"
|
|
if not candidate.exists():
|
|
return candidate
|
|
for i in range(1, 9999):
|
|
candidate = folder / f"{base}_{i}{ext}"
|
|
if not candidate.exists():
|
|
return candidate
|
|
return folder / f"{base}{ext}"
|
|
|
|
|
|
def build_workable_files(
|
|
*,
|
|
facility_name: str,
|
|
downloaded: list[Path],
|
|
output_root: Path,
|
|
today_iso: str,
|
|
) -> tuple[list[Path], Path]:
|
|
"""Copy originals to Workable/ (DD-MM-YYYY_FacilityName[_N].pdf) and mirror to Workable_Medicare/."""
|
|
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_path(workable_root, base, ext)
|
|
shutil.copy2(src, dst)
|
|
workable_paths.append(dst)
|
|
|
|
med_dst = medicare_root / dst.name
|
|
if med_dst.exists():
|
|
med_dst = _unique_workable_path(medicare_root, dst.stem, ext)
|
|
shutil.copy2(dst, med_dst)
|
|
|
|
_log.info("Workable (%d files) -> %s | Medicare -> %s",
|
|
len(workable_paths), workable_root, medicare_root)
|
|
return workable_paths, workable_root
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Summary Excel (same columns as the Planner bot)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def write_summary(
|
|
results: dict[str, dict],
|
|
failures: list[tuple[str, str]],
|
|
summary_path: Path,
|
|
run_timestamp: str,
|
|
) -> None:
|
|
try:
|
|
import pandas as pd
|
|
except ImportError:
|
|
_log.warning("pandas not available — skipping summary Excel")
|
|
return
|
|
|
|
rows = []
|
|
failure_names = {f for f, _ in failures}
|
|
for facility, payload in results.items():
|
|
files = len(payload.get("downloaded", []))
|
|
rows.append({
|
|
"Run Timestamp": run_timestamp,
|
|
"Facility Name": facility,
|
|
"Files Downloaded": files,
|
|
"Status": "Success" if files else "No files",
|
|
"Status Changed": "N/A (ECW)",
|
|
})
|
|
for facility, err in failures:
|
|
rows.append({
|
|
"Run Timestamp": run_timestamp,
|
|
"Facility Name": facility,
|
|
"Files Downloaded": 0,
|
|
"Status": "FAILED",
|
|
"Status Changed": "N/A",
|
|
})
|
|
|
|
summary_path.parent.mkdir(parents=True, exist_ok=True)
|
|
import pandas as pd
|
|
pd.DataFrame(rows).to_excel(summary_path, index=False, sheet_name="Download")
|
|
_log.info("Summary -> %s", summary_path)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Main workflow
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def run_ecw_workflow(
|
|
*,
|
|
facility_limit: int = 0,
|
|
facility_name: str | None = None,
|
|
run_date: str | None = None,
|
|
resume: bool = False,
|
|
) -> None:
|
|
run_timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
today_iso = run_date or TODAY_ISO_DATE
|
|
log_path = setup_logging(PROJECT_ROOT)
|
|
|
|
_log.info("=" * 66)
|
|
_log.info(" MISF ECW Download — Start")
|
|
_log.info(" Run date : %s", today_iso)
|
|
_log.info(" Log : %s", log_path)
|
|
_log.info("=" * 66)
|
|
log_step("ECW run started", date=today_iso)
|
|
|
|
ecw_cfg = _read_ecw_config()
|
|
if not ecw_cfg["ecw_url"]:
|
|
raise ValueError(
|
|
"ECW URL not configured. "
|
|
"Add 'ECW URL' to Input/input.xlsx Sheet1."
|
|
)
|
|
|
|
output_root = PROJECT_ROOT / "Output"
|
|
original_root = output_root / today_iso / "Original"
|
|
checkpoint_path = output_root / today_iso / "checkpoint_ecw.json"
|
|
summary_path = (
|
|
output_root / today_iso / "Summary"
|
|
/ f"Download_Summary_ECW_{today_iso}.xlsx"
|
|
)
|
|
|
|
# ── Facility list ─────────────────────────────────────────────────────
|
|
facilities: list[str] = read_facility_list()
|
|
if facility_name:
|
|
facilities = [f for f in facilities if f.lower() == facility_name.lower()]
|
|
if not facilities:
|
|
_log.error("Facility '%s' not found in list.", facility_name)
|
|
return
|
|
if facility_limit > 0:
|
|
facilities = facilities[:facility_limit]
|
|
|
|
# ── Resume / checkpoint ───────────────────────────────────────────────
|
|
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 [])}
|
|
_log.info("Resuming from index %d; %d already completed", start_index, len(completed))
|
|
else:
|
|
clear_checkpoint(checkpoint_path)
|
|
|
|
# ── Step 1 — Login via Playwright (UI only) ───────────────────────────
|
|
cookies = login_and_harvest_cookies(ecw_cfg)
|
|
|
|
# ── Step 2 — Build httpx client (all remaining requests are HTTP) ─────
|
|
client = build_http_client(ecw_cfg, cookies)
|
|
|
|
# ── Step 3 — Fetch document list ──────────────────────────────────────
|
|
log_step("Fetching ECW document list", date=today_iso)
|
|
html = fetch_document_list(client, today_iso)
|
|
|
|
# ── Step 4 — Parse HTML for per-facility download links ───────────────
|
|
facility_files = parse_facility_files(html, facilities)
|
|
total_found = sum(len(v) for v in facility_files.values())
|
|
log_step("Files parsed from document list", total=total_found)
|
|
|
|
results: dict[str, dict] = {}
|
|
failures: list[tuple[str, str]] = []
|
|
total_dl = 0
|
|
|
|
for idx, fac in enumerate(facilities, start=1):
|
|
if idx <= start_index:
|
|
continue
|
|
if fac in completed:
|
|
_log.info("Skipping already completed: %s", fac)
|
|
continue
|
|
|
|
log_step("Facility", index=f"{idx}/{len(facilities)}", facility=fac)
|
|
urls = facility_files.get(fac, [])
|
|
|
|
if not urls:
|
|
_log.info("%s: no files found in ECW document list", fac)
|
|
results[fac] = {"downloaded": [], "workable": []}
|
|
completed.add(fac)
|
|
save_checkpoint(checkpoint_path, last_completed_index=idx,
|
|
facility_name=fac, completed_facilities=sorted(completed))
|
|
continue
|
|
|
|
dest_folder = original_root / _sanitize_folder_name(fac)
|
|
downloaded: list[Path] = []
|
|
try:
|
|
for url in urls:
|
|
path = download_file(client, url, dest_folder)
|
|
if path:
|
|
downloaded.append(path)
|
|
|
|
workable_paths, workable_root = build_workable_files(
|
|
facility_name=fac,
|
|
downloaded=downloaded,
|
|
output_root=output_root,
|
|
today_iso=today_iso,
|
|
)
|
|
total_dl += len(downloaded)
|
|
results[fac] = {"downloaded": downloaded, "workable": workable_paths}
|
|
_log.info("%s: %d downloaded, %d workable", fac, len(downloaded), len(workable_paths))
|
|
completed.add(fac)
|
|
save_checkpoint(checkpoint_path, last_completed_index=idx,
|
|
facility_name=fac, completed_facilities=sorted(completed))
|
|
|
|
except Exception as exc:
|
|
_log.error("%s: %s", fac, exc)
|
|
failures.append((fac, str(exc)))
|
|
results[fac] = {"downloaded": [], "workable": []}
|
|
save_checkpoint(checkpoint_path, last_completed_index=idx,
|
|
facility_name=fac, completed_facilities=sorted(completed))
|
|
|
|
write_summary(results, failures, summary_path, run_timestamp)
|
|
clear_checkpoint(checkpoint_path)
|
|
|
|
_log.info("=" * 66)
|
|
_log.info(" ECW DOWNLOAD COMPLETE — %d file(s), %d failure(s)", total_dl, len(failures))
|
|
_log.info("=" * 66)
|
|
log_step("ECW run complete", files=total_dl, failures=len(failures))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _build_parser() -> argparse.ArgumentParser:
|
|
p = argparse.ArgumentParser(description="MISF ECW face-sheet downloader")
|
|
p.add_argument("--limit", type=int, default=0, metavar="N",
|
|
help="Process only first N facilities")
|
|
p.add_argument("--facility", default="", metavar="NAME",
|
|
help="Process a single facility by exact name")
|
|
p.add_argument("--date", default="", metavar="YYYY-MM-DD",
|
|
help="Override run date (default: today)")
|
|
p.add_argument("--resume", action="store_true",
|
|
help="Resume from last checkpoint")
|
|
return p
|
|
|
|
|
|
def main() -> int:
|
|
args = _build_parser().parse_args()
|
|
try:
|
|
run_ecw_workflow(
|
|
facility_limit=max(0, args.limit),
|
|
facility_name=args.facility.strip() or None,
|
|
run_date=args.date.strip() or None,
|
|
resume=args.resume,
|
|
)
|
|
return 0
|
|
except KeyboardInterrupt:
|
|
print("\nCancelled.")
|
|
return 130
|
|
except Exception:
|
|
import traceback
|
|
traceback.print_exc()
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|