from __future__ import annotations import asyncio import logging import os import sys from datetime import date, datetime, time, timedelta, timezone from pathlib import Path from typing import Any from zoneinfo import ZoneInfo import httpx from config import settings from paths import DATA_DIR from services.calendar_models import CalendarMeeting, format_meeting_description from services.secret_box import decrypt_secret from services.state_store import ( clear_graph_session, clear_teams_browser_meta, get_graph_session, get_teams_browser_meta, set_graph_session, set_teams_browser_meta, ) logger = logging.getLogger(__name__) BROWSER_STATE_FILE = DATA_DIR / "teams_browser_state.json" MFA_PREVIEW_FILE = DATA_DIR / "teams_mfa_preview.png" OUTLOOK_CALENDAR_URL = "https://outlook.office.com/calendar/view/workweek" LOGIN_TIMEOUT_MS = 5 * 60 * 1000 # Number-match / Authenticator MFA can be slow; keep browser open for interaction MFA_WAIT_MS = 10 * 60 * 1000 # Public Microsoft Office client — used only for optional ROPC (password grant). # Tenants with MFA / CA usually reject this; then we fall back to browser login. DEFAULT_PUBLIC_CLIENT_ID = "d3590ed6-52b3-4102-aeff-aad2292ab01c" GRAPH_SCOPES = "offline_access Calendars.Read User.Read openid profile" # Extra Chromium flags for containers (no X11 / small /dev/shm) _DOCKER_CHROME_ARGS = [ "--disable-blink-features=AutomationControlled", "--no-sandbox", "--disable-dev-shm-usage", "--disable-gpu", ] def _running_in_docker() -> bool: if os.environ.get("PTS_IN_DOCKER", "").strip().lower() in {"1", "true", "yes"}: return True if Path("/.dockerenv").exists(): return True try: cgroup = Path("/proc/1/cgroup").read_text(encoding="utf-8", errors="ignore") if "docker" in cgroup or "containerd" in cgroup or "kubepods" in cgroup: return True except OSError: pass return False def _display_available() -> bool: return bool(os.environ.get("DISPLAY", "").strip()) def _must_run_headless() -> bool: """True only when a browser window cannot be shown at all. Docker with DISPLAY (Xvfb + noVNC) → headed OK, user interacts via :6080. Linux without DISPLAY/WAYLAND → headless only. macOS / Windows → headed works without DISPLAY. """ if _display_available() or os.environ.get("WAYLAND_DISPLAY", "").strip(): return False if _running_in_docker(): return True if sys.platform.startswith("linux"): return True return False def resolve_playwright_headless(requested: bool | None = None) -> bool: """Decide Playwright headless mode. Preference order: 1. No display possible → always headless 2. Explicit argument (login sends headless=false for MFA) 3. settings.teams_headless """ if _must_run_headless(): if requested is False: logger.warning( "Headed Playwright requested but no DISPLAY " "(Docker without Xvfb?) — forcing headless=True." ) return True if requested is not None: return bool(requested) return bool(settings.teams_headless) def playwright_launch_args() -> list[str]: # Always use sandbox-friendly flags in Docker (headed on Xvfb still needs them) if _running_in_docker() or _must_run_headless(): return list(_DOCKER_CHROME_ARGS) return ["--disable-blink-features=AutomationControlled"] def browser_view_info() -> dict[str, Any]: """How the user can see / interact with the Teams login browser.""" vnc_port = int(os.environ.get("TEAMS_VNC_PORT", "6080") or "6080") has_display = _display_available() in_docker = _running_in_docker() interactive = in_docker and has_display return { "in_docker": in_docker, "display": os.environ.get("DISPLAY") or None, "interactive": interactive, "vnc_port": vnc_port, "mfa_preview": MFA_PREVIEW_FILE.exists(), "message": ( "Docker：登入時會開啟虛擬 Chrome，請用互動畫面完成 MFA（可點、可輸入）。" if interactive else ( "本機：請在跳出的 Chromium 視窗完成 MFA。" if not in_docker else "Docker 但無 DISPLAY：無法顯示驗證畫面，請確認容器已啟動 Xvfb。" ) ), } class TeamsCalendarScraper: """Calendar via username/password: Graph API first, browser session as fallback.""" def __init__(self) -> None: self.timezone = ZoneInfo(settings.teams_calendar_timezone) @property def enabled(self) -> bool: return bool(settings.teams_calendar_enabled) @property def has_credentials(self) -> bool: return bool((settings.teams_username or "").strip() and (settings.teams_password or "").strip()) @property def has_browser_state(self) -> bool: return BROWSER_STATE_FILE.exists() and BROWSER_STATE_FILE.stat().st_size > 10 @property def has_graph_token(self) -> bool: """Unused for fetch path (crawl-only). Kept for status diagnostics.""" session = get_graph_session() or {} return bool(session.get("access_token") and session.get("refresh_token")) def status(self) -> dict[str, Any]: meta = get_teams_browser_meta() mode = meta.get("auth_mode") or ("browser_session" if self.has_browser_state else None) last_error = str(meta.get("last_error") or "") # Only treat as stale for real session-expiry signals (not old Graph errors) stale = bool( last_error and any( k in last_error.lower() for k in ("session expired", "re-login", "browser session expired", "session 過期") ) ) # Crawl mode: connected = browser cookies present connected = self.has_browser_state and not stale return { "enabled": self.enabled, "configured": self.has_credentials or self.has_browser_state, "has_credentials": self.has_credentials, "connected": connected, "username": (settings.teams_username or "").strip() or None, "last_login_at": meta.get("last_login_at"), "last_error": meta.get("last_error"), "message": meta.get("message") or "crawl-only (no Graph API)", "auth_mode": mode, "source": "teams_browser_crawl", } def format_meeting_description(self, meeting: CalendarMeeting) -> str: return format_meeting_description(meeting) def _round_hours(self, hours: float) -> float: step = settings.hour_step if hours <= 0: return 0.0 units = max(1, int(round(hours / step))) return round(units * step, 1) def _event_hours(self, start: datetime, end: datetime) -> float: duration = (end - start).total_seconds() / 3600 return self._round_hours(duration) def _parse_time(self, value: str | datetime) -> datetime: if isinstance(value, datetime): dt = value else: cleaned = str(value).replace("Z", "+00:00") dt = datetime.fromisoformat(cleaned) if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return dt def _day_window(self, day: date) -> tuple[datetime, datetime]: start = datetime.combine(day, time.min, tzinfo=self.timezone) end = start + timedelta(days=1) return start, end def _credentials(self) -> tuple[str, str]: username = (settings.teams_username or "").strip() password = decrypt_secret(settings.teams_password or "") if not username or not password: raise ValueError("Teams username/password not set. Login on the web UI first.") return username, password async def login( self, username: str | None = None, password: str | None = None, *, headless: bool | None = None, ) -> dict[str, Any]: user = (username or settings.teams_username or "").strip() pwd = password if password is not None else decrypt_secret(settings.teams_password or "") if not user or not pwd: raise ValueError("Teams username and password are required") # Crawl-only: browser login + save cookies (no Graph API / ROPC) clear_graph_session() use_headless = resolve_playwright_headless(headless) try: from playwright.async_api import async_playwright except ImportError as exc: raise ValueError( "playwright is not installed. Run: pip install playwright && playwright install chromium" ) from exc view = browser_view_info() set_teams_browser_meta( { "message": ( f"正在登入 Outlook（headless={use_headless}）。" + ( f" 請開啟 noVNC 埠 {view['vnc_port']} 完成 MFA；瀏覽器會保持開啟最多 {MFA_WAIT_MS // 60000} 分鐘。" if not use_headless and _running_in_docker() else " 請在跳出的視窗完成 MFA。" if not use_headless else " 無畫面模式：若卡住請確認 Docker 已啟用 Xvfb/noVNC。" ) ), "last_error": None, "mfa_interactive": not use_headless, "vnc_port": view.get("vnc_port"), } ) probe_count = 0 async with async_playwright() as p: launch_kwargs: dict[str, Any] = { "headless": use_headless, "channel": "chromium", "args": playwright_launch_args(), } if _running_in_docker(): launch_kwargs["env"] = { **os.environ, "DISPLAY": os.environ.get("DISPLAY", ":99"), } browser = await p.chromium.launch(**launch_kwargs) context = await browser.new_context( user_agent=( "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/122.0.0.0 Safari/537.36" ), locale="zh-TW", timezone_id=settings.teams_calendar_timezone, viewport={"width": 1280, "height": 800}, ) page = await context.new_page() try: await page.goto("https://login.microsoftonline.com/", wait_until="domcontentloaded") await self._snapshot_mfa(page, "login_start") await self._fill_microsoft_login(page, user, pwd) await page.wait_for_timeout(1500) await self._snapshot_mfa(page, "after_password") # If MFA / number-match is already on screen, do NOT navigate away # (would refresh the number). Only open Outlook when not mid-identity. if not self._on_identity_page(page.url): try: await page.goto( OUTLOOK_CALENDAR_URL, wait_until="domcontentloaded", timeout=LOGIN_TIMEOUT_MS, ) except Exception as nav_exc: # noqa: BLE001 logger.info("Outlook nav after login (MFA may be active): %s", nav_exc) else: logger.info("Identity/MFA page active — keeping page open for user: %s", page.url) await self._wait_for_outlook_ready(page) # Wait for calendar UI + cookies to settle (browser still open) await page.wait_for_timeout(5000) try: await page.wait_for_load_state("networkidle", timeout=20000) except Exception: # noqa: BLE001 pass # Verify we are not still on login if "login.microsoftonline.com" in page.url.lower(): await self._snapshot_mfa(page, "still_login") raise ValueError( "仍在 Microsoft 登入頁 — 請在互動畫面完成 MFA／密碼後再等一下。" ) # Probe calendar fetch while session is live (same browser) try: today = datetime.now(self.timezone).date() probe_start, _ = self._day_window(today) _, probe_end = self._day_window(today + timedelta(days=7)) probe_events = await self._page_fetch_calendar_events(page, probe_start, probe_end) probe_count = len(probe_events) logger.info("Login probe fetched %s events in live session", probe_count) except Exception as probe_exc: # noqa: BLE001 logger.warning("Login probe fetch failed (session still saved): %s", probe_exc) BROWSER_STATE_FILE.parent.mkdir(parents=True, exist_ok=True) await context.storage_state(path=str(BROWSER_STATE_FILE)) await self._snapshot_mfa(page, "success") except Exception as exc: # noqa: BLE001 try: await self._snapshot_mfa(page, "error") except Exception: # noqa: BLE001 pass set_teams_browser_meta( { "last_error": str(exc), "message": ( "Teams 登入失敗。請在 noVNC 互動畫面完成 MFA 後再試。" if _running_in_docker() else "Teams 登入失敗。請在瀏覽器視窗完成 MFA 後再試。" ), } ) raise ValueError(f"Teams login failed: {exc}") from exc finally: # Close only after success or hard failure — not mid-MFA await browser.close() set_teams_browser_meta( { "last_login_at": datetime.now(timezone.utc).isoformat(), "last_error": None, "message": ( f"Browser crawl session saved. Login probe: {probe_count} events." if probe_count else "Browser crawl session saved. If test is 0, re-login and complete MFA fully." ), "username": user, "auth_mode": "browser_crawl", "login_probe_events": probe_count, } ) return { "ok": True, "message": ( f"Teams 瀏覽器登入成功（爬蟲）。即時探測到 {probe_count} 筆會議。" if probe_count else "Teams 瀏覽器登入完成。若測試仍 0 筆，請再登入一次並完成 MFA。" ), "connected": True, "auth_mode": "browser_crawl", "login_probe_events": probe_count, } async def logout(self) -> None: if BROWSER_STATE_FILE.exists(): BROWSER_STATE_FILE.unlink() clear_graph_session() clear_teams_browser_meta() def _token_client_id(self) -> str: return (settings.azure_client_id or "").strip() or DEFAULT_PUBLIC_CLIENT_ID def _token_tenant(self) -> str: return (settings.azure_tenant_id or "").strip() or "organizations" async def _login_graph_ropc(self, username: str, password: str) -> dict[str, Any]: """Resource Owner Password Credentials — direct API, no browser.""" url = f"https://login.microsoftonline.com/{self._token_tenant()}/oauth2/v2.0/token" payload = { "client_id": self._token_client_id(), "scope": GRAPH_SCOPES, "username": username, "password": password, "grant_type": "password", } async with httpx.AsyncClient(timeout=60.0) as client: response = await client.post(url, data=payload) data = response.json() if response.content else {} if response.status_code >= 400: err = data.get("error_description") or data.get("error") or response.text raise ValueError(str(err)[:300]) access = data.get("access_token") if not access: raise ValueError("Graph token response missing access_token") set_graph_session( { "access_token": access, "refresh_token": data.get("refresh_token"), "scope": data.get("scope"), "expires_in": data.get("expires_in"), "token_type": data.get("token_type"), "auth_mode": "ropc", } ) return data async def _refresh_graph_token(self) -> str: session = get_graph_session() or {} refresh = session.get("refresh_token") if not refresh: raise ValueError("No Graph refresh_token") url = f"https://login.microsoftonline.com/{self._token_tenant()}/oauth2/v2.0/token" payload = { "client_id": self._token_client_id(), "scope": GRAPH_SCOPES, "refresh_token": refresh, "grant_type": "refresh_token", } async with httpx.AsyncClient(timeout=60.0) as client: response = await client.post(url, data=payload) data = response.json() if response.content else {} if response.status_code >= 400: raise ValueError(data.get("error_description") or data.get("error") or "refresh failed") access = data.get("access_token") if not access: raise ValueError("refresh missing access_token") session["access_token"] = access if data.get("refresh_token"): session["refresh_token"] = data["refresh_token"] set_graph_session(session) return access async def _graph_access_token(self) -> str: session = get_graph_session() or {} token = session.get("access_token") if token: return str(token) return await self._refresh_graph_token() @staticmethod def _on_identity_page(url: str) -> bool: u = (url or "").lower() return any( host in u for host in ( "login.microsoftonline.com", "login.live.com", "login.microsoft.com", "device.login.microsoftonline.com", "aadcdn.msauth.net", "aadcdn.msftauth.net", ) ) async def _snapshot_mfa(self, page: Any, label: str = "") -> None: """Write latest login page screenshot for UI polling (Docker MFA).""" try: MFA_PREVIEW_FILE.parent.mkdir(parents=True, exist_ok=True) await page.screenshot(path=str(MFA_PREVIEW_FILE), full_page=False, type="png") if label: set_teams_browser_meta( { **get_teams_browser_meta(), "mfa_preview_label": label, "mfa_preview_at": datetime.now(timezone.utc).isoformat(), "mfa_page_url": page.url, } ) except Exception as exc: # noqa: BLE001 logger.debug("MFA snapshot failed (%s): %s", label, exc) async def _fill_microsoft_login(self, page: Any, username: str, password: str) -> None: # Email step email = page.locator('input[type="email"], input[name="loginfmt"]') await email.first.wait_for(state="visible", timeout=30000) await email.first.fill(username) await self._snapshot_mfa(page, "email_filled") await page.locator('input[type="submit"], input[data-report-event="Signin_Submit"]').first.click() # Password step (may be skipped if SSO/MFA device) try: pwd = page.locator('input[type="password"], input[name="passwd"]') await pwd.first.wait_for(state="visible", timeout=15000) await pwd.first.fill(password) await self._snapshot_mfa(page, "password_filled") await page.locator('input[type="submit"], input[data-report-event="Signin_Submit"]').first.click() except Exception: # noqa: BLE001 logger.info("Password field not shown (SSO/MFA path)") await self._snapshot_mfa(page, "after_credentials") # Stay signed in? (may appear after MFA — also handled in wait loop) try: stay = page.locator('#idSIButton9, input[value="Yes"], input[value="是"]') await stay.first.wait_for(state="visible", timeout=8000) await stay.first.click() except Exception: # noqa: BLE001 pass async def _wait_for_outlook_ready(self, page: Any) -> None: """Keep Chromium open while user completes MFA in noVNC / local window.""" deadline = asyncio.get_event_loop().time() + MFA_WAIT_MS / 1000 last_snap = 0.0 while asyncio.get_event_loop().time() < deadline: url = page.url.lower() now = asyncio.get_event_loop().time() # Refresh screenshot ~every 1.5s so UI shows the number / OTP page if now - last_snap >= 1.5: await self._snapshot_mfa(page, "mfa_wait") last_snap = now remaining = int(deadline - now) set_teams_browser_meta( { **get_teams_browser_meta(), "message": ( f"等待 MFA／進到 Outlook… 剩餘約 {remaining} 秒。" " 請在互動畫面操作，不要關閉。" ), "mfa_page_url": page.url, } ) if "outlook.office.com" in url and "login" not in url and "login.microsoftonline" not in url: try: await page.wait_for_load_state("networkidle", timeout=15000) except Exception: # noqa: BLE001 pass return # "Stay signed in?" after MFA try: stay = page.locator('#idSIButton9, input[value="Yes"], input[value="是"]') if await stay.first.is_visible(): await stay.first.click() await self._snapshot_mfa(page, "stay_signed_in") except Exception: # noqa: BLE001 pass # MFA finished but still not on Outlook — open calendar once if not self._on_identity_page(page.url) and "outlook.office.com" not in url: try: await page.goto( OUTLOOK_CALENDAR_URL, wait_until="domcontentloaded", timeout=60000, ) await self._snapshot_mfa(page, "nav_outlook") except Exception as nav_exc: # noqa: BLE001 logger.debug("Deferred Outlook nav: %s", nav_exc) await asyncio.sleep(0.8) await self._snapshot_mfa(page, "timeout") raise TimeoutError( f"等待 Outlook 逾時（{MFA_WAIT_MS // 60000} 分鐘）。" "請在 noVNC 互動畫面完成 MFA 後再按一次登入。" ) async def fetch_meetings_for_days(self, days: list[date]) -> dict[str, list[CalendarMeeting]]: if not days: return {} if not self.enabled: return {d.isoformat(): [] for d in days} # Crawl-only path — never call Graph API (no refresh_token / MFA issues) clear_graph_session() if not self.has_browser_state: if self.has_credentials: # Docker has no X server — never force headed mode await self.login(headless=resolve_playwright_headless(None)) else: raise ValueError("Teams 未登入。請到設定頁登入 Teams（瀏覽器 / MFA）。") sorted_days = sorted(set(days)) range_start, _ = self._day_window(sorted_days[0]) _, range_end = self._day_window(sorted_days[-1]) errors: list[str] = [] raw_events: list[dict[str, Any]] = [] session_expired = False try: raw_events = await self._scrape_calendar_events(range_start, range_end) logger.info("Browser crawl returned %s raw events", len(raw_events)) except Exception as first_exc: # noqa: BLE001 msg = f"Browser crawl: {first_exc}" errors.append(msg) logger.warning("Teams crawl failed: %s", first_exc) err_l = str(first_exc).lower() session_expired = "expired" in err_l or "login" in err_l or "re-login" in err_l if session_expired and BROWSER_STATE_FILE.exists(): try: BROWSER_STATE_FILE.unlink() logger.info("Cleared expired Teams browser state") except OSError: pass # Do NOT auto re-login during fetch (MFA needs human). Ask user to login in Settings. if not raw_events and (session_expired or not self.has_browser_state): raise ValueError( "Teams 網頁 session 無效或過期。請到設定頁重新「登入 Teams」並完成 MFA，" "再按「測試抓會議」。" + (" 詳情: " + " | ".join(errors) if errors else "") ) meetings_by_day: dict[str, list[CalendarMeeting]] = {d.isoformat(): [] for d in sorted_days} allowed = set(meetings_by_day.keys()) seen_meetings: set[tuple[str, str, str]] = set() for item in raw_events: try: start = self._parse_time(item["start"]) end = self._parse_time(item["end"]) except Exception: # noqa: BLE001 continue if end <= start: continue if item.get("is_cancelled"): continue show_as = str(item.get("show_as") or "").lower() # free / workingelsewhere often still real meetings in some tenants — only drop free if show_as == "free": continue local_day = start.astimezone(self.timezone).date().isoformat() if local_day not in allowed: continue hours = self._event_hours(start, end) if hours <= 0: continue subject = str(item.get("subject") or "Meeting").strip() dedupe_key = ( subject.lower(), start.astimezone(timezone.utc).isoformat(), end.astimezone(timezone.utc).isoformat(), ) if dedupe_key in seen_meetings: continue seen_meetings.add(dedupe_key) meetings_by_day[local_day].append( CalendarMeeting( subject=subject, start=start, end=end, hours=hours, is_online=bool(item.get("is_online")), ) ) if not any(meetings_by_day.values()) and errors: set_teams_browser_meta( { **get_teams_browser_meta(), "last_error": " | ".join(errors)[:500], "message": "會議同步回傳 0 筆，可能是 session 過期或 token 無效", } ) return meetings_by_day async def list_meetings( self, start: date, end: date, *, skip_holidays: bool = True, debug: bool = False, ) -> dict[str, Any]: """Public payload for the web UI: meetings by day + totals. When skip_holidays=True (default), only Taiwan workdays are included (weekends + TW public holidays skipped), matching PTS fill range behavior. """ from services.holidays_util import is_workday, iter_workdays if end < start: start, end = end, start if skip_holidays: days = iter_workdays(start, end) calendar_span = (end - start).days + 1 skipped_days = max(0, calendar_span - len(days)) else: days = [] cursor = start while cursor <= end: days.append(cursor) cursor += timedelta(days=1) skipped_days = 0 debug_info: dict[str, Any] = { "mode": "browser_crawl_only", "graph_api": "disabled", "has_browser_state": self.has_browser_state, "has_credentials": self.has_credentials, "enabled": self.enabled, "day_count": len(days), "days": [d.isoformat() for d in days], "steps": [], } meetings_by_day: dict[str, list[CalendarMeeting]] = {} fetch_error: str | None = None try: if days: meetings_by_day = await self.fetch_meetings_for_days(days) debug_info["steps"].append("fetch_meetings_for_days: ok") else: debug_info["steps"].append("no days after holiday filter") except Exception as exc: # noqa: BLE001 fetch_error = str(exc) debug_info["steps"].append(f"fetch_meetings_for_days: ERROR {exc}") logger.exception("list_meetings fetch failed") days_out: list[dict[str, Any]] = [] total_hours = 0.0 total_count = 0 for day in days: key = day.isoformat() items = meetings_by_day.get(key, []) day_hours = round(sum(m.hours for m in items), 1) total_hours += day_hours total_count += len(items) days_out.append( { "date": key, "hours": day_hours, "count": len(items), "is_workday": is_workday(day), "meetings": [ { "subject": m.subject, "start": m.start.astimezone(self.timezone).isoformat(), "end": m.end.astimezone(self.timezone).isoformat(), "hours": m.hours, "is_online": m.is_online, "time_label": ( f"{m.start.astimezone(self.timezone).strftime('%H:%M')}" f"–{m.end.astimezone(self.timezone).strftime('%H:%M')}" ), } for m in items ], } ) meta = get_teams_browser_meta() payload: dict[str, Any] = { "ok": fetch_error is None, "enabled": self.enabled, "connected": self.status().get("connected"), "timezone": settings.teams_calendar_timezone, "start_date": start.isoformat(), "end_date": end.isoformat(), "skip_holidays": skip_holidays, "workday_count": len(days), "skipped_days": skipped_days, "total_hours": round(total_hours, 1), "total_count": total_count, "days": days_out, "synced_at": datetime.now(timezone.utc).isoformat(), "error": fetch_error, "status": self.status(), "fetch_hint": ( None if total_count else ( fetch_error or meta.get("last_error") or meta.get("message") or "若一直是 0 筆：請先「登入 Teams」完成 MFA，再按測試。" ) ), } if debug: payload["debug"] = debug_info return payload async def _scrape_calendar_events( self, range_start: datetime, range_end: datetime, ) -> list[dict[str, Any]]: try: from playwright.async_api import async_playwright except ImportError as exc: raise ValueError("playwright is not installed") from exc if not self.has_browser_state: raise ValueError("No Teams browser session — please login on Settings page") collected: list[dict[str, Any]] = [] graph_payloads: list[dict[str, Any]] = [] captured_bearers: list[str] = [] capture_errors: list[str] = [] seen_urls: list[str] = [] async with async_playwright() as p: # channel="chromium" forces Chrome's "new" headless mode (same full # browser binary as headed, just no window) instead of Playwright's # lightweight headless-shell. The shell's fingerprint (GPU/canvas, # navigator.plugins, missing APIs, etc.) diverges enough from a real # headed browser that Azure AD's sign-in risk detection was bouncing # our cookie-reuse requests back to the login page — which we then # (wrongly) treated as an expired session, deleting the saved cookies # and forcing a fresh interactive MFA login on every single fetch. browser = await p.chromium.launch( headless=resolve_playwright_headless(True), channel="chromium", args=playwright_launch_args(), ) context = await browser.new_context( storage_state=str(BROWSER_STATE_FILE), user_agent=( "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/122.0.0.0 Safari/537.36" ), locale="zh-TW", timezone_id=settings.teams_calendar_timezone, ) page = await context.new_page() def on_request(request: Any) -> None: try: auth = (request.headers or {}).get("authorization") or "" if auth.lower().startswith("bearer ") and len(auth) > 40: token = auth.split(" ", 1)[1].strip() if token not in captured_bearers: captured_bearers.append(token) except Exception: # noqa: BLE001 return async def on_response(response: Any) -> None: try: if response.status != 200: return url = response.url url_l = url.lower() if len(seen_urls) < 80: seen_urls.append(url[:180]) interesting = any( key in url_l for key in ( "calendarview", "calendar/getschedule", "calendarservice", "/me/events", "finditem", "getcalendarview", "service.svc", "outlook.office.com/api/", "graph.microsoft.com", "substrate.office.com", "calendar", ) ) if not interesting: return ctype = (response.headers.get("content-type") or "").lower() if "json" not in ctype and "javascript" not in ctype and "text" not in ctype: return try: data = await response.json() except Exception: # noqa: BLE001 return if isinstance(data, dict): graph_payloads.append(data) elif isinstance(data, list): graph_payloads.append({"value": data}) except Exception: # noqa: BLE001 return page.on("request", on_request) page.on("response", on_response) try: # Load calendar SPA date_param = range_start.date().isoformat() urls_to_try = [ f"https://outlook.office.com/calendar/view/week?startdt={date_param}", OUTLOOK_CALENDAR_URL, "https://outlook.office.com/calendar/view/month", "https://outlook.office.com/mail/", ] landed = False for cal_url in urls_to_try: try: await page.goto(cal_url, wait_until="domcontentloaded", timeout=90000) await page.wait_for_timeout(3000) url_now = page.url.lower() if "login.microsoftonline.com" in url_now or "login.live.com" in url_now: if BROWSER_STATE_FILE.exists(): try: BROWSER_STATE_FILE.unlink() except OSError: pass set_teams_browser_meta( { **get_teams_browser_meta(), "last_error": "session expired", "message": "Browser session expired. Re-login Teams in Settings (complete MFA).", "auth_mode": "browser_crawl", } ) raise ValueError( "Teams browser session expired — open Settings → login Teams (complete MFA)." ) landed = True if "calendar" in url_now or "outlook.office.com" in url_now: break except ValueError: raise except Exception as nav_exc: # noqa: BLE001 capture_errors.append(f"nav {cal_url}: {nav_exc}") if not landed: raise ValueError("Could not open Outlook with saved session") # Wait for SPA network try: await page.wait_for_load_state("networkidle", timeout=20000) except Exception: # noqa: BLE001 await page.wait_for_timeout(5000) # A) Page-context fetch (cookies + SPA origin) — most reliable crawl try: page_events = await self._page_fetch_calendar_events(page, range_start, range_end) if page_events: collected = page_events logger.info("Page-context fetch returned %s events", len(collected)) except Exception as pe: # noqa: BLE001 capture_errors.append(f"page-fetch: {pe}") logger.info("Page-context fetch failed: %s", pe) # B) Use bearer tokens the SPA itself sent (not Graph app registration) if not collected and captured_bearers: for token in captured_bearers[:3]: try: events = await self._fetch_graph_calendar_view( context, token, range_start, range_end ) if events: collected = events logger.info( "SPA-intercepted bearer returned %s events", len(collected) ) break except Exception as te: # noqa: BLE001 capture_errors.append(f"bearer: {te}") # C) Parse any intercepted JSON payloads. A single page view only # shows "this week" though — to cover a wider requested range # (e.g. the past month) we must page through every week in # [range_start, range_end] so each one fires its own calendarView # XHR that on_response() can capture. if not collected: await self._page_through_weeks(page, range_start, range_end, capture_errors) collected = self._events_from_graph_payloads(graph_payloads) if collected: logger.info("XHR crawl intercept returned %s events", len(collected)) else: capture_errors.append( f"XHR payloads={len(graph_payloads)} bearers={len(captured_bearers)}" ) logger.info( "No calendar events. sample_urls=%s", seen_urls[:15], ) await context.storage_state(path=str(BROWSER_STATE_FILE)) finally: await browser.close() if not collected and capture_errors: logger.warning("Teams crawl empty. errors=%s", capture_errors) return collected async def _jump_to_date(self, page: Any, target: date) -> bool: """Click the mini date-picker's day cell to move the main calendar view. Reloading the page with a `?startdt=` query param does NOT work — OWA ignores it on a fresh navigation and always renders "today"'s week. The mini date-picker (the small monthly calendar in the folder pane) is the only reliable way to jump the main view to an arbitrary date without a full reload. Its day buttons are labelled e.g. "1, 7 月, 2026" (day, month, year) in zh-TW. If the target date's month isn't currently rendered in the picker, we click "移至上個月" (previous month) to page the mini-calendar itself back until the day becomes visible. """ label = f"{target.day}, {target.month} 月, {target.year}" day_button = page.locator(f'button[aria-label="{label}"]') prev_month_button = page.locator('button[aria-label^="移至上個月"]') for _ in range(36): # safety cap: ~3 years of previous-month clicks try: if await day_button.count() > 0: await day_button.first.click(timeout=5000) return True except Exception: # noqa: BLE001 pass try: if await prev_month_button.count() == 0: return False await prev_month_button.first.click(timeout=5000) await page.wait_for_timeout(400) except Exception: # noqa: BLE001 return False return False async def _page_through_weeks( self, page: Any, range_start: datetime, range_end: datetime, capture_errors: list[str], ) -> None: """Walk the Outlook SPA's own calendar view across the requested range. The XHR-sniffing fallback (path C) only sees whatever calendar view is currently on screen. Without this, a request for "the past month" would silently only ever capture the current week. We click through the mini date-picker to each week in the range so every jump fires a fresh calendarView call for that week, which on_response() captures into graph_payloads — accumulated across every iteration. """ cursor = range_start.date() end_date = range_end.date() max_weeks = 14 # safety cap (~3 months) so a huge range can't hang forever visited = 0 while cursor <= end_date and visited < max_weeks: jumped = False try: jumped = await self._jump_to_date(page, cursor) except Exception as exc: # noqa: BLE001 capture_errors.append(f"date-picker-nav {cursor.isoformat()}: {exc}") if not jumped: # Fall back to a full reload — slower and often ignored by OWA, # but better than skipping the week entirely. try: week_url = ( "https://outlook.office.com/calendar/view/week" f"?startdt={cursor.isoformat()}" ) await page.goto(week_url, wait_until="domcontentloaded", timeout=60000) await page.wait_for_timeout(3000) except Exception as exc: # noqa: BLE001 capture_errors.append(f"week-nav {cursor.isoformat()}: {exc}") try: await page.wait_for_load_state("networkidle", timeout=15000) except Exception: # noqa: BLE001 await page.wait_for_timeout(3500) # Nudge the SPA so it actually re-issues the calendarView XHR for # this specific week (some builds cache/skip the fetch otherwise). try: await page.keyboard.press("ArrowRight") await page.wait_for_timeout(1000) await page.keyboard.press("ArrowLeft") await page.wait_for_timeout(1500) except Exception: # noqa: BLE001 pass cursor += timedelta(days=7) visited += 1 logger.info( "Paged through %s week(s) covering %s..%s", visited, range_start.date().isoformat(), range_end.date().isoformat(), ) async def _page_fetch_calendar_events( self, page: Any, range_start: datetime, range_end: datetime, ) -> list[dict[str, Any]]: """Run fetch() inside the Outlook page (full cookie + same-site context).""" start_s = range_start.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") end_s = range_end.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") tz = settings.teams_calendar_timezone result = await page.evaluate( """async ({ startS, endS, tz }) => { const urls = [ `https://outlook.office.com/api/v2.0/me/calendarview?startDateTime=${encodeURIComponent(startS)}&endDateTime=${encodeURIComponent(endS)}&$top=200&$orderby=Start/DateTime`, `https://outlook.office.com/api/v2.0/me/calendar/calendarView?startDateTime=${encodeURIComponent(startS)}&endDateTime=${encodeURIComponent(endS)}&$top=200`, `https://graph.microsoft.com/v1.0/me/calendarView?startDateTime=${encodeURIComponent(startS)}&endDateTime=${encodeURIComponent(endS)}&$top=200&$orderby=start/dateTime`, `https://outlook.office.com/api/v2.0/me/events?$top=100&$orderby=Start/DateTime&$filter=Start/DateTime ge '${startS}'`, ]; const out = []; for (const url of urls) { try { const r = await fetch(url, { credentials: 'include', headers: { 'Accept': 'application/json', 'Prefer': `outlook.timezone="${tz}"`, }, }); const text = await r.text(); let body = null; try { body = JSON.parse(text); } catch (e) { body = text.slice(0, 200); } out.push({ url, status: r.status, body }); if (r.status === 200 && body && (body.value || body.Value)) { return { ok: true, status: r.status, url, body }; } } catch (e) { out.push({ url, error: String(e) }); } } return { ok: false, attempts: out }; }""", {"startS": start_s, "endS": end_s, "tz": tz}, ) if not isinstance(result, dict): return [] if result.get("ok") and isinstance(result.get("body"), dict): events = self._events_from_graph_payloads([result["body"]]) logger.info("page fetch ok via %s → %s events", result.get("url"), len(events)) return events # log attempts briefly attempts = result.get("attempts") or [] for a in attempts[:6]: logger.info( "page-fetch attempt status=%s url=%s err=%s", a.get("status"), str(a.get("url") or "")[:80], a.get("error"), ) return [] async def _extract_access_token(self, page: Any) -> str | None: """Pull MSAL / ADAL access tokens from browser storage.""" try: token = await page.evaluate( """() => { function looksLikeJwt(s) { return typeof s === 'string' && s.split('.').length === 3 && s.length > 40; } function walk(obj, depth) { if (!obj || depth > 6) return null; if (typeof obj === 'string') { if (looksLikeJwt(obj)) return obj; return null; } if (typeof obj !== 'object') return null; // MSAL credential entry if (obj.secret && looksLikeJwt(obj.secret)) { const ct = String(obj.credentialType || obj.credential_type || '').toLowerCase(); if (!ct || ct.includes('access')) return obj.secret; } if (obj.access_token && looksLikeJwt(obj.access_token)) return obj.access_token; if (obj.accessToken && looksLikeJwt(obj.accessToken)) return obj.accessToken; for (const v of Object.values(obj)) { const found = walk(v, depth + 1); if (found) return found; } return null; } const stores = [localStorage, sessionStorage]; for (const store of stores) { for (let i = 0; i < store.length; i++) { const k = store.key(i); if (!k) continue; const raw = store.getItem(k); if (!raw || raw.length < 30) continue; // quick skip if (!/access|msal|token|login/i.test(k + raw.slice(0, 80))) { if (!raw.includes('eyJ')) continue; } try { const data = JSON.parse(raw); const found = walk(data, 0); if (found) return found; } catch (e) { if (looksLikeJwt(raw)) return raw; } } } return null; }""" ) return token if isinstance(token, str) and len(token) > 40 else None except Exception as exc: # noqa: BLE001 logger.info("Token extract failed: %s", exc) return None async def _fetch_outlook_rest_calendar_view( self, context: Any, range_start: datetime, range_end: datetime, ) -> list[dict[str, Any]]: """Outlook REST v2 using OWA cookies (no Graph app registration).""" from urllib.parse import quote start_s = range_start.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") end_s = range_end.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") urls = [ ( "https://outlook.office.com/api/v2.0/me/calendarview" f"?startDateTime={quote(start_s)}&endDateTime={quote(end_s)}" "&$top=200&$orderby=Start/DateTime" "&$select=Subject,Start,End,IsCancelled,IsOnlineMeeting,ShowAs" ), ( "https://outlook.office365.com/api/v2.0/me/calendarview" f"?startDateTime={quote(start_s)}&endDateTime={quote(end_s)}" "&$top=200&$orderby=Start/DateTime" ), ] headers = { "Accept": "application/json", "Prefer": f'outlook.timezone="{settings.teams_calendar_timezone}"', } for url in urls: response = await context.request.get(url, headers=headers) if response.status == 401 or response.status == 403: logger.info("Outlook REST %s → %s", url.split("?")[0], response.status) continue if response.status != 200: logger.info("Outlook REST failed %s: %s", response.status, (await response.text())[:160]) continue data = await response.json() events = self._events_from_graph_payloads([data]) if events: return events return [] async def _fetch_graph_calendar_view( self, context: Any, token: str, range_start: datetime, range_end: datetime, ) -> list[dict[str, Any]]: from urllib.parse import quote start_s = range_start.isoformat() end_s = range_end.isoformat() url = ( "https://graph.microsoft.com/v1.0/me/calendarView" f"?startDateTime={quote(start_s)}" f"&endDateTime={quote(end_s)}" "&$select=subject,start,end,isCancelled,isOnlineMeeting,showAs" "&$orderby=start/dateTime" "&$top=200" ) response = await context.request.get( url, headers={ "Authorization": f"Bearer {token}", "Prefer": f'outlook.timezone="{settings.teams_calendar_timezone}"', }, ) if response.status != 200: body = "" try: body = (await response.text())[:200] except Exception: # noqa: BLE001 pass logger.info("Graph calendarView via browser token failed: %s %s", response.status, body) return [] data = await response.json() return self._events_from_graph_payloads([data]) async def _fetch_via_graph_api( self, range_start: datetime, range_end: datetime, ) -> list[dict[str, Any]]: """Graph calendarView using stored access token (httpx).""" from urllib.parse import quote token = await self._graph_access_token() start_s = range_start.isoformat() end_s = range_end.isoformat() url = ( "https://graph.microsoft.com/v1.0/me/calendarView" f"?startDateTime={quote(start_s)}" f"&endDateTime={quote(end_s)}" "&$select=subject,start,end,isCancelled,isOnlineMeeting,showAs" "&$orderby=start/dateTime" "&$top=200" ) headers = { "Authorization": f"Bearer {token}", "Prefer": f'outlook.timezone="{settings.teams_calendar_timezone}"', } async with httpx.AsyncClient(timeout=60.0) as client: response = await client.get(url, headers=headers) if response.status_code == 401: # try refresh then once more try: token = await self._refresh_graph_token() headers["Authorization"] = f"Bearer {token}" response = await client.get(url, headers=headers) except Exception as exc: # noqa: BLE001 raise ValueError(f"Graph token expired and refresh failed: {exc}") from exc if response.status_code >= 400: raise ValueError( f"Graph calendarView HTTP {response.status_code}: {response.text[:200]}" ) data = response.json() return self._events_from_graph_payloads([data]) def _events_from_graph_payloads(self, payloads: list[dict[str, Any]]) -> list[dict[str, Any]]: events: list[dict[str, Any]] = [] seen: set[tuple[str, str, str]] = set() for payload in payloads: values = self._extract_item_list(payload) for item in values: if not isinstance(item, dict): continue start_s, end_s = self._extract_start_end(item) if not start_s or not end_s: continue subject = ( item.get("subject") or item.get("Subject") or item.get("name") or item.get("Name") or "Meeting" ) # The same calendar view can be intercepted more than once (initial # load + the ArrowRight/ArrowLeft nudge used to trigger XHRs, or the # page-fetch/bearer/XHR-intercept paths overlapping), so the identical # event shows up in multiple captured payloads — sometimes with the # SAME instant expressed in different formats (UTC "Z" vs. an # already-shifted "+08:00" local offset). Comparing raw strings would # miss those, so normalize start/end to UTC before deduping. try: start_key = self._parse_time(start_s).astimezone(timezone.utc).isoformat() end_key = self._parse_time(end_s).astimezone(timezone.utc).isoformat() except Exception: # noqa: BLE001 start_key, end_key = str(start_s), str(end_s) dedupe_key = (str(subject).strip().lower(), start_key, end_key) if dedupe_key in seen: continue seen.add(dedupe_key) events.append( { "subject": str(subject), "start": start_s, "end": end_s, "is_cancelled": bool( item.get("isCancelled") or item.get("IsCancelled") or item.get("isCancelledEvent") ), "is_online": bool( item.get("isOnlineMeeting") or item.get("IsOnlineMeeting") or item.get("onlineMeeting") or item.get("OnlineMeeting") or item.get("isOnlineMeeting") ), "show_as": str( item.get("showAs") or item.get("ShowAs") or item.get("showas") or "" ), } ) return events def _extract_item_list(self, payload: Any) -> list[Any]: if isinstance(payload, list): return payload if not isinstance(payload, dict): return [] for key in ("value", "Value", "Items", "items", "events", "Events", "Body"): nested = payload.get(key) if isinstance(nested, list): return nested if isinstance(nested, dict): # OWA Body.ResponseMessages... deeper = self._extract_item_list(nested) if deeper: return deeper # recursive shallow search for list of event-like dicts for v in payload.values(): if isinstance(v, list) and v and isinstance(v[0], dict): sample = v[0] if any(k in sample for k in ("Start", "start", "Subject", "subject", "End", "end")): return v if isinstance(v, dict): deeper = self._extract_item_list(v) if deeper: return deeper return [] def _extract_start_end(self, item: dict[str, Any]) -> tuple[str | None, str | None]: start = item.get("start") or item.get("Start") or item.get("startTime") or item.get("StartTime") end = item.get("end") or item.get("End") or item.get("endTime") or item.get("EndTime") start_s = self._normalize_time_field(start) end_s = self._normalize_time_field(end) return start_s, end_s def _normalize_time_field(self, value: Any) -> str | None: if value is None: return None if isinstance(value, str): # OWA legacy: /Date(1710000000000)/ if value.startswith("/Date("): try: ms = int(value.split("(")[1].split(")")[0].split("+")[0].split("-")[0]) return datetime.fromtimestamp(ms / 1000, tz=timezone.utc).isoformat() except Exception: # noqa: BLE001 return None return value if isinstance(value, (int, float)): # epoch ms or seconds ts = float(value) if ts > 1e12: ts = ts / 1000 return datetime.fromtimestamp(ts, tz=timezone.utc).isoformat() if isinstance(value, dict): return ( value.get("dateTime") or value.get("DateTime") or value.get("date") or value.get("Date") ) return None