from dataclasses import dataclass from playwright.async_api import Page from app.automation.helpers import human_delay from app.automation.session_store import session_store class LoginVerificationError(Exception): """Raised when Hudl login verification fails.""" @dataclass(frozen=True) class HudlCredentials: email: str password: str login_url: str def resolve_credentials(credentials: HudlCredentials | None = None) -> HudlCredentials: if credentials is not None: return credentials from app.services.account_service import load_active_credentials return load_active_credentials() CANONICAL_HUDL_LOGIN_URL = "https://identity.hudl.com/u/login/identifier" HUDL_HOME_URL = "https://www.hudl.com/home" def canonical_hudl_login_url(login_url: str) -> str: """Use a stable identity URL — stored state/query params expire and cause iss redirects.""" url = (login_url or CANONICAL_HUDL_LOGIN_URL).strip() if "identity.hudl.com" in url.lower(): return url.split("?")[0] return CANONICAL_HUDL_LOGIN_URL async def _has_hudl_app_shell(page: Page) -> bool: """True when logged-in Hudl app chrome is visible.""" from app.automation.selectors import PRIMARY_NAVBAR try: nav = page.locator(PRIMARY_NAVBAR).first if await nav.count() > 0: return await nav.is_visible() except Exception: pass return False async def _is_hudl_iss_landing(url: str) -> bool: lower = url.lower() return "www.hudl.com" in lower and ( "iss=" in lower or lower.rstrip("/").endswith("hudl.com") or lower.rstrip("/").endswith("www.hudl.com") ) async def is_login_page(page: Page) -> bool: """True when Hudl is asking for email/password.""" if "identity.hudl.com" in page.url.lower(): return await page.locator('input[type="email"]').count() > 0 return ( await page.locator('input[type="email"]').count() > 0 and await page.locator('input[type="password"]').count() > 0 ) async def is_hudl_logged_in(page: Page) -> bool: """ True when already inside Hudl app (e.g. redirected to /home). Not on identity login screen. """ if await is_login_page(page): return False url = page.url.lower() if "hudl.com" not in url: return False if "identity.hudl.com" in url: return False if await _has_hudl_app_shell(page): return True if any(path in url for path in ("/home", "/app/", "/library", "/exchanges", "/teams")): return True return False async def _resolve_login_entry_page( page: Page, login_url: str, *, allow_session_skip: bool, ) -> str: """ Navigate until we reach the identity login form or an authenticated Hudl page. Returns: "login_form" | "logged_in" """ urls_to_try = [] for candidate in (login_url, canonical_hudl_login_url(login_url), CANONICAL_HUDL_LOGIN_URL): if candidate not in urls_to_try: urls_to_try.append(candidate) for url in urls_to_try: max_nav_attempts = 3 for attempt in range(1, max_nav_attempts + 1): try: await page.goto(url, wait_until="domcontentloaded") break except Exception as exc: print(f"Navigation to login URL failed on attempt {attempt}: {exc}") if attempt == max_nav_attempts: raise exc await human_delay(2, 3) await human_delay(2, 4) await close_popup_if_visible(page) if await is_login_page(page): return "login_form" if await is_hudl_logged_in(page): if allow_session_skip: return "logged_in" # Verification must show the login form — try canonical URL next. continue if await _is_hudl_iss_landing(page.url): # OIDC callback (e.g. /?iss=identity.hudl.com) — wait, then try /home or login. await human_delay(2, 3) if await is_hudl_logged_in(page): if allow_session_skip: return "logged_in" try: await page.goto(HUDL_HOME_URL, wait_until="domcontentloaded") await human_delay(2, 3) await close_popup_if_visible(page) except Exception as exc: print(f"Navigation to Hudl home after iss landing failed: {exc}") if await is_login_page(page): return "login_form" if await is_hudl_logged_in(page) and allow_session_skip: return "logged_in" if await is_login_page(page): return "login_form" if await is_hudl_logged_in(page) and allow_session_skip: return "logged_in" raise LoginVerificationError( f"Could not reach Hudl login form or home (current page: {page.url})." ) async def close_popup_if_visible(page: Page) -> None: from app.automation.tasks.popup import close_hudl_popups await close_hudl_popups(page) async def _attach_storage_warning( result: dict, page: Page, *, force_refresh: bool = False ) -> dict: if not force_refresh: cached = session_store.get_storage_warning() if cached: result["storage_warning"] = cached return result try: from app.automation.tasks.library import fetch_hudl_storage_usage storage = await fetch_hudl_storage_usage(page) session_store.set_storage_warning(storage) result["storage_warning"] = storage except Exception as exc: print(f"Non-critical storage check error: {exc}") result["storage_warning"] = {"available": False, "show_popup": False} return result async def _skip_login_response(page: Page, reason: str) -> dict: await close_popup_if_visible(page) session_store.mark_logged_in(page.url) result = { "status": "already_logged_in", "url": page.url, "message": reason, } return await _attach_storage_warning(result, page) async def _login_failure_reason(page: Page) -> str: url = page.url.lower() try: body_text = (await page.locator("body").inner_text(timeout=5000)).lower() except Exception: body_text = "" if any( token in body_text for token in ( "multi-factor", "verification code", "authenticator", "two-factor", "2-step", "enter the code", ) ): return ( "Handshake requires MFA or a verification code. " "Automated login cannot complete for this account." ) if any(token in body_text for token in ("captcha", "not a robot", "security check")): return "Handshake presented a CAPTCHA or security check. Try again later." if any(token in body_text for token in ("locked", "suspended", "blocked", "disabled")): return "Handshake account may be locked, suspended, or blocked." for selector in ( '[role="alert"]', '[data-error="true"]', ".error", ".uni-notice__content", 'div[class*="error"]', ): try: loc = page.locator(selector).first if await loc.count() > 0: text = (await loc.text_content() or "").strip() if text and len(text) < 300: return f"Handshake login error: {text}\nPlease check your credentials and try again." except Exception: continue if "identity.hudl.com" in url and await is_login_page(page): return "Login failed — still on login page. Check email and password." return ( f"Login failed — could not reach Handshake home after submitting credentials " f"(current page: {page.url})." ) async def attempt_hudl_login( page: Page, creds: HudlCredentials, *, force: bool = False, allow_session_skip: bool = True, ) -> dict: """Perform Hudl login with the given credentials.""" if not creds.email or not creds.password or not creds.login_url: raise ValueError( "Handshake credentials are incomplete. Provide email, password, and login URL." ) if force: session_store.mark_logged_out() if allow_session_skip and not force and await is_hudl_logged_in(page): return await _skip_login_response( page, "Already on Handshake — no login needed." ) entry_login_url = ( canonical_hudl_login_url(creds.login_url) if force else creds.login_url ) entry_state = await _resolve_login_entry_page( page, entry_login_url, allow_session_skip=allow_session_skip, ) if entry_state == "logged_in": return await _skip_login_response( page, "Redirected to Handshake home — session valid, login skipped.", ) if not await is_login_page(page): raise LoginVerificationError( f"Unexpected page after login URL: {page.url}. Not login or Handshake home." ) await page.locator('input[type="email"]').fill(creds.email) await human_delay() await page.locator('button[type="submit"]').click() await human_delay(2, 4) await page.locator('input[type="password"]').fill(creds.password) await human_delay() await page.locator('button[type="submit"]').click() await human_delay(4, 6) await close_popup_if_visible(page) if await is_login_page(page): session_store.mark_logged_out() raise LoginVerificationError(await _login_failure_reason(page)) if not await is_hudl_logged_in(page): try: await page.goto(HUDL_HOME_URL, wait_until="domcontentloaded") await human_delay(2, 3) await close_popup_if_visible(page) except Exception as exc: print(f"Post-login navigation to Handshake home failed: {exc}") if await is_login_page(page) or not await is_hudl_logged_in(page): session_store.mark_logged_out() raise LoginVerificationError(await _login_failure_reason(page)) session_store.mark_logged_in(page.url) try: await fetch_hudl_account_name(page) except Exception as e: print(f"Non-critical name fetch error: {e}") result = { "status": "logged_in", "url": page.url, "message": "Login successful. Session saved for next time.", } return await _attach_storage_warning(result, page, force_refresh=True) async def ensure_hudl_session(page: Page, credentials: HudlCredentials | None = None) -> dict: """ Ensure an active Hudl session without navigating to /home. Used on hot paths (exchanges, notifications) to avoid slow detours. """ creds = resolve_credentials(credentials) if await is_hudl_logged_in(page): return await _skip_login_response(page, "Session active — login skipped.") return await attempt_hudl_login( page, creds, force=False, allow_session_skip=True, ) async def verify_credentials(page: Page, creds: HudlCredentials) -> dict: """ Clears any saved browser session so verification always tests the submitted email/password (not cached cookies), then runs the same login flow used by production automation. """ from app.automation.browser_manager import browser_manager await browser_manager.clear_session() page = await browser_manager.start() result = await attempt_hudl_login( page, creds, force=True, allow_session_skip=False, ) return { "verified": True, "verification_method": "playwright_live_login", "status": result.get("status", "logged_in"), "message": "credentials verified successfully via Playwright live login.", } async def run( page: Page, force: bool = False, credentials: HudlCredentials | None = None, ) -> dict: """ POST /hudl/login: - Opens login URL; if cookies redirect to hudl/home → skip (do nothing). - If login form shows → email/password login. - Other APIs can run after this without logging in again. """ creds = resolve_credentials(credentials) return await attempt_hudl_login( page, creds, force=force, allow_session_skip=True, ) async def fetch_hudl_account_name(page: Page) -> str: """Fetch the logged-in user's display name from Hudl's top-right user menu. Falls back to the selected team name only if the user name can't be found. """ from app.automation.selectors import ( PRIMARY_NAVBAR, TEAM_SWITCHER_SELECTED_NAME, USER_ACCOUNT_NAME_SELECTORS, ) try: if "hudl.com/home" not in page.url: await page.goto("https://www.hudl.com/home", wait_until="domcontentloaded") await human_delay(1.5, 2.5) await page.locator(PRIMARY_NAVBAR).first.wait_for(state="visible", timeout=15000) for selector in USER_ACCOUNT_NAME_SELECTORS: try: loc = page.locator(selector).first if await loc.count() == 0: continue name = (await loc.text_content() or "").strip() if name: session_store.set_account_holder_name(name) return name except Exception: continue span = page.locator(TEAM_SWITCHER_SELECTED_NAME).first await span.wait_for(state="visible", timeout=10000) name = await span.text_content() if name: name = name.strip() session_store.set_account_holder_name(name) return name except Exception as e: print(f"Non-critical: Failed to extract account holder's name: {e}") return ""