""" Auto-process Hudl exchanges where the opponent sent video. Wizard flow per exchange: 1. Open sidebar row ("They sent you video.") 2. Click "Add to Video Library" 3. Modal: "Add without scheduled event" 4. Modal: "Game Footage" 5. Modal: "Add to Video Library" (confirm) 6. Modal: "No, Thanks" (Hudl Assist) """ import asyncio import re import time from playwright.async_api import Page from app.automation.helpers import human_delay from app.automation.tasks.popup import close_hudl_popups from app.automation.tasks.exchange import ( _ensure_exchange_list_visible, _open_active_exchange_row, _parse_active_exchange_li, open_exchange, ) INCOMING_VIDEO_PATTERNS = ( re.compile(r"they\s+sent\s+you\s+video", re.I), re.compile(r"their\s+video\s+is\s+in\s+your\s+videos", re.I), ) def _is_incoming_video_message(message: str) -> bool: text = (message or "").strip() if not text: return False return any(pattern.search(text) for pattern in INCOMING_VIDEO_PATTERNS) async def _wait_and_click_text( page: Page, patterns: tuple[str, ...], *, timeout_s: float = 20.0, ) -> bool: """Click the first visible element whose text matches any pattern.""" deadline = time.monotonic() + timeout_s while time.monotonic() < deadline: for pattern in patterns: regex = re.compile(pattern, re.I) for locator_factory in ( lambda p=regex: page.get_by_role("button", name=p), lambda p=regex: page.get_by_role("link", name=p), lambda p=regex: page.get_by_text(p), ): try: loc = locator_factory().first if await loc.count() > 0 and await loc.is_visible(): await loc.scroll_into_view_if_needed() await loc.click(timeout=5000) await human_delay(0.8, 1.4) return True except Exception: continue await asyncio.sleep(0.25) return False async def _click_add_to_video_library_header(page: Page) -> bool: """Click the blue 'Add to Video Library' button in exchange details header.""" selectors = ( ".ex-details-header .ex-actions .ex-action-add", ".ex-details .ex-actions button:has-text('Add to Video Library')", "button:has-text('Add to Video Library')", ) for selector in selectors: btn = page.locator(selector).first try: if await btn.count() > 0 and await btn.is_visible(): await btn.scroll_into_view_if_needed() await btn.click(timeout=5000) await human_delay(1.0, 1.6) return True except Exception: continue return await _wait_and_click_text(page, ("Add to Video Library",)) async def _complete_add_to_library_wizard(page: Page) -> list[str]: """Run modal steps after 'Add to Video Library' was clicked.""" steps_completed: list[str] = [] if await _wait_and_click_text( page, ( r"Add without scheduled event", r"Add without scheduler event", ), timeout_s=15.0, ): steps_completed.append("add_without_scheduled_event") if await _wait_and_click_text(page, (r"Game Footage",), timeout_s=15.0): steps_completed.append("game_footage") confirm_selectors = ( "[role='dialog'] button:has-text('Add to Video Library')", ".uni-modal button:has-text('Add to Video Library')", "footer button:has-text('Add to Video Library')", ) confirmed = False for selector in confirm_selectors: btn = page.locator(selector).first try: if await btn.count() > 0 and await btn.is_visible(): await btn.click(timeout=5000) await human_delay(1.0, 1.6) confirmed = True break except Exception: continue if not confirmed: confirmed = await _wait_and_click_text( page, (r"^Add to Video Library$",), timeout_s=10.0, ) if confirmed: steps_completed.append("confirm_add_to_library") if await _wait_and_click_text( page, (r"No, Thanks", r"No thanks"), timeout_s=15.0, ): steps_completed.append("hudl_assist_declined") await close_hudl_popups(page) return steps_completed async def _process_single_incoming_exchange( page: Page, row: dict, index: int, ) -> dict: school_name = row.get("name") or "" try: list_locator = await _ensure_exchange_list_visible(page, full_scroll=True) li = list_locator.nth(index) opened_url = await _open_active_exchange_row(page, li, school_name) details_pane = page.locator("div.ex-details").first await details_pane.wait_for(state="visible", timeout=15000) await human_delay(0.8, 1.2) if not await _click_add_to_video_library_header(page): return { "status": "error", "name": school_name, "message": "Add to Video Library button not found.", "url": page.url, } wizard_steps = await _complete_add_to_library_wizard(page) return { "status": "ok", "name": school_name, "message": row.get("process") or "", "href": row.get("href") or opened_url, "wizard_steps": wizard_steps, "url": page.url, } except Exception as exc: return { "status": "error", "name": school_name, "message": str(exc), "url": page.url, } async def process_incoming_video_exchanges(page: Page) -> dict: """ Find all exchanges with incoming video messages and add them to the library. Re-scans the sidebar after each item because indices shift once processed. """ if "exchange" not in page.url.lower(): await open_exchange(page) else: await close_hudl_popups(page) processed: list[dict] = [] errors: list[dict] = [] max_passes = 50 for _ in range(max_passes): list_locator = await _ensure_exchange_list_visible(page, full_scroll=True) count = await list_locator.count() next_item: tuple[int, dict] | None = None for i in range(count): row = await _parse_active_exchange_li(list_locator.nth(i), i) if not row: continue if _is_incoming_video_message(row.get("process") or ""): next_item = (i, row) break if not next_item: break index, row = next_item result = await _process_single_incoming_exchange(page, row, index) if result.get("status") == "ok": processed.append(result) else: errors.append(result) await close_hudl_popups(page) await human_delay(0.5, 1.0) if not processed and not errors: return { "status": "ok", "action": "process_incoming_video_exchanges", "processed_count": 0, "processed": [], "errors": [], "message": "No incoming video exchanges found.", } return { "status": "ok", "action": "process_incoming_video_exchanges", "processed_count": len(processed), "error_count": len(errors), "processed": processed, "errors": errors, "message": ( f"Processed {len(processed)} incoming video exchange(s)" + (f"; {len(errors)} failed." if errors else ".") ), }