import asyncio import os import time from pathlib import Path from playwright.async_api import BrowserContext, Page, Playwright, async_playwright BROWSER_DATA_DIR = Path(__file__).resolve().parents[2] / "browser_data" # Independent tabs that can run Hudl flows in parallel (same cookies / login). LANES = ("default", "exchange", "manage_library") # Default action timeout — overlays should not burn a full 30s retry loop. DEFAULT_ACTION_TIMEOUT_MS = int(os.getenv("PLAYWRIGHT_TIMEOUT_MS", "15000")) DEFAULT_NAV_TIMEOUT_MS = int(os.getenv("PLAYWRIGHT_NAV_TIMEOUT_MS", "45000")) # slow_mo=700 made every click feel like a "timer" hang; keep a tiny delay only. DEFAULT_SLOW_MO_MS = int(os.getenv("PLAYWRIGHT_SLOW_MO_MS", "50")) def _configure_page_timeouts(page: Page) -> None: try: page.set_default_timeout(DEFAULT_ACTION_TIMEOUT_MS) page.set_default_navigation_timeout(DEFAULT_NAV_TIMEOUT_MS) except Exception: pass class BrowserManager: """ ONE Chromium profile for the FastAPI app (shared Hudl login cookies), with multiple pages ("lanes") so Exchange and Manage Library can run at the same time without stealing each other's navigation. """ def __init__(self): self._playwright: Playwright | None = None self.context: BrowserContext | None = None self._pages: dict[str, Page | None] = {lane: None for lane in LANES} self._lane_locks: dict[str, asyncio.Lock] = { lane: asyncio.Lock() for lane in LANES } self._lane_busy: dict[str, bool] = {lane: False for lane in LANES} self._boot_lock = asyncio.Lock() self.is_running = False self.idle_timeout = 300 # 5 minutes self.last_activity_time = time.time() self._loop_task: asyncio.Task | None = None @property def page(self) -> Page | None: """Backward-compatible alias for the default lane page.""" return self._pages.get("default") @property def is_busy(self) -> bool: return any(self._lane_busy.values()) @is_busy.setter def is_busy(self, value: bool) -> None: # Legacy callers toggle a global busy flag — map onto the default lane. self._lane_busy["default"] = bool(value) def set_lane_busy(self, lane: str, busy: bool) -> None: lane = self._normalize_lane(lane) self._lane_busy[lane] = busy def is_lane_busy(self, lane: str) -> bool: return self._lane_busy.get(self._normalize_lane(lane), False) def lane_lock(self, lane: str) -> asyncio.Lock: return self._lane_locks[self._normalize_lane(lane)] def _normalize_lane(self, lane: str | None) -> str: if lane in LANES: return lane return "default" async def start(self, headless: bool | None = None, lane: str = "default") -> Page: lane = self._normalize_lane(lane) async with self._boot_lock: if not self._loop_task or self._loop_task.done(): self._loop_task = asyncio.create_task(self._idle_check_loop()) if self.context is None: await self._boot_context(headless) page = await self._ensure_lane_page(lane) self.is_running = True self.record_activity() return page async def _boot_context(self, headless: bool | None) -> None: BROWSER_DATA_DIR.mkdir(parents=True, exist_ok=True) self._check_and_auto_clear_session() if headless is None: import os import sys headless_env = os.getenv("HEADLESS", "false").lower() in ("true", "1", "yes") is_linux = sys.platform.startswith("linux") has_display = "DISPLAY" in os.environ if is_linux and not has_display: headless = True else: headless = headless_env self._playwright = await async_playwright().start() self._prune_cache() self.context = await self._playwright.chromium.launch_persistent_context( user_data_dir=str(BROWSER_DATA_DIR), headless=headless, slow_mo=DEFAULT_SLOW_MO_MS if not headless else 0, viewport={"width": 1366, "height": 768}, locale="en-US", args=[ "--disk-cache-size=10485760", "--media-cache-size=10485760", "--disable-gpu-program-cache", "--disable-gpu-shader-disk-cache", "--disable-application-cache", "--disable-dev-shm-usage", "--disable-crash-reporter", "--disable-breakpad", "--no-sandbox", "--start-maximized", ], ) try: self.context.set_default_timeout(DEFAULT_ACTION_TIMEOUT_MS) self.context.set_default_navigation_timeout(DEFAULT_NAV_TIMEOUT_MS) except Exception: pass # Seed the default lane with the first Chromium tab. first = ( self.context.pages[0] if self.context.pages else await self.context.new_page() ) _configure_page_timeouts(first) self._pages["default"] = first async def block_resources(route): req = route.request url = req.url.lower() if "hudl.com" in url: await route.continue_() return if req.resource_type in ("image", "media", "font"): await route.abort() return blocked_patterns = [ "google-analytics", "mixpanel", "hotjar", "sentry.io", "facebook.net", ] if any(pat in url for pat in blocked_patterns): await route.abort() return await route.continue_() await self.context.route("**/*", block_resources) async def _ensure_lane_page(self, lane: str) -> Page: if self.context is None: raise RuntimeError("Browser context is not started") page = self._pages.get(lane) if page is not None and not page.is_closed(): _configure_page_timeouts(page) return page assigned = { id(p) for p in self._pages.values() if p is not None and not p.is_closed() } if lane == "default" and self.context.pages: # Prefer reusing an existing tab for the default lane. for existing in self.context.pages: if not existing.is_closed(): _configure_page_timeouts(existing) self._pages["default"] = existing return existing # Reclaim an unassigned tab already on Manage Library so downloads # don't start from a blank about:blank tab after a lane remap. if lane == "manage_library": for existing in self.context.pages: if existing.is_closed() or id(existing) in assigned: continue if "/manage/library/" in (existing.url or ""): _configure_page_timeouts(existing) self._pages[lane] = existing print( f"Browser lane '{lane}' reused existing Manage Library tab" ) return existing page = await self.context.new_page() _configure_page_timeouts(page) self._pages[lane] = page print(f"Browser lane '{lane}' opened on a dedicated tab") return page def _prune_cache(self) -> None: """Clear heavy non-essential Chromium cache directories before booting.""" import shutil try: if not BROWSER_DATA_DIR.exists(): return dirs_to_prune = [ BROWSER_DATA_DIR / "Default" / "Cache", BROWSER_DATA_DIR / "Default" / "Code Cache", BROWSER_DATA_DIR / "Default" / "GPUCache", BROWSER_DATA_DIR / "Default" / "Service Worker" / "CacheStorage", BROWSER_DATA_DIR / "Default" / "Service Worker" / "ScriptCache", BROWSER_DATA_DIR / "Crash Reports", ] for path in dirs_to_prune: if path.exists() and path.is_dir(): shutil.rmtree(path, ignore_errors=True) except Exception as e: print(f"Non-critical: Cache pruning failed: {e}") def _check_and_auto_clear_session(self) -> None: """Clear browser data once per day for a fresh login session.""" import shutil last_clear_file = BROWSER_DATA_DIR / "last_clear.txt" should_clear = False if not last_clear_file.exists(): should_clear = True else: try: timestamp = int(last_clear_file.read_text().strip()) if time.time() - timestamp > 86400: should_clear = True except Exception: should_clear = True if should_clear: print( "Daily session clear triggered: clearing browser data for a fresh login session." ) try: if BROWSER_DATA_DIR.exists(): for item in BROWSER_DATA_DIR.iterdir(): if item.is_dir(): shutil.rmtree(item, ignore_errors=True) else: item.unlink(missing_ok=True) except Exception as e: print(f"Failed during daily session cleanup: {e}") try: last_clear_file.write_text(str(int(time.time()))) except Exception as e: print(f"Failed to write last_clear timestamp: {e}") async def clear_session(self) -> None: """Close the browser and wipe the persistent profile.""" async with self._boot_lock: await self._shutdown_unlocked() import shutil try: if BROWSER_DATA_DIR.exists(): for item in BROWSER_DATA_DIR.iterdir(): if item.is_dir(): shutil.rmtree(item, ignore_errors=True) else: item.unlink(missing_ok=True) except Exception as e: print(f"Failed to delete browser data directory: {e}") BROWSER_DATA_DIR.mkdir(parents=True, exist_ok=True) last_clear_file = BROWSER_DATA_DIR / "last_clear.txt" try: last_clear_file.write_text(str(int(time.time()))) except Exception as e: print(f"Failed to write last_clear timestamp: {e}") async def get_page(self, lane: str = "default") -> Page: self.record_activity() lane = self._normalize_lane(lane) page = self._pages.get(lane) if page is None or page.is_closed() or self.context is None: return await self.start(lane=lane) return page async def close(self) -> None: async with self._boot_lock: await self._shutdown_unlocked() async def _shutdown_unlocked(self) -> None: if self.context: try: await self.context.close() except Exception: pass if self._playwright: try: await self._playwright.stop() except Exception: pass self.context = None self._playwright = None self._pages = {lane: None for lane in LANES} self._lane_busy = {lane: False for lane in LANES} self.is_running = False def record_activity(self) -> None: self.last_activity_time = time.time() async def _idle_check_loop(self) -> None: while True: try: await asyncio.sleep(15) if self.is_running and not self.is_busy: elapsed = time.time() - self.last_activity_time if elapsed > self.idle_timeout: print( f"Browser idle timeout reached: {elapsed:.1f}s > " f"{self.idle_timeout}s. Automatically closing browser " "to release resources..." ) await self.close() except asyncio.CancelledError: break except Exception as e: print(f"Error in idle check loop: {e}") await asyncio.sleep(5) browser_manager = BrowserManager()