Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
274 lines
8.5 KiB
Python
274 lines
8.5 KiB
Python
"""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
|