""" Google Veo Video Provider – generates I2V (Image-to-Video) via the Google Flow UI using Playwright to handle reCAPTCHA automatically. Strategy: 1. Use GoogleFlowImageProvider to generate an image and get mediaGenerationId 2. Launch Playwright browser with session cookies 3. Navigate to Flow, configure I2V settings 4. Use the mediaGenerationId as start frame 5. Submit generation and capture the response 6. Poll for completion and download the video Requires: playwright (pip install playwright && playwright install chromium) Uses the same cookies as GoogleFlowImageProvider. """ from __future__ import annotations import asyncio import json import logging import os import random import time import uuid from collections.abc import Callable from pathlib import Path from typing import Any import httpx from providers.ports import VideoProvider # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- SESSION_URL = "https://labs.google/fx/api/auth/session" GENERATE_I2V_URL = "https://aisandbox-pa.googleapis.com/v1/video:batchAsyncGenerateVideoStartImage" GENERATE_T2V_URL = "https://aisandbox-pa.googleapis.com/v1/video:batchAsyncGenerateVideoText" STATUS_URL = "https://aisandbox-pa.googleapis.com/v1/video:batchCheckAsyncVideoGenerationStatus" CREATE_PROJECT_URL = "https://labs.google/fx/api/trpc/project.createProject" VIDEO_ASPECT_RATIOS = { "landscape": "VIDEO_ASPECT_RATIO_LANDSCAPE", "portrait": "VIDEO_ASPECT_RATIO_PORTRAIT", "square": "VIDEO_ASPECT_RATIO_SQUARE", } VIDEO_MODELS = { "veo_3_1_i2v_fast": "veo_3_1_i2v_s_fast", "veo_3_1_t2v_fast": "veo_3_1_t2v_fast", "veo_2_i2v": "veo_2_0_i2v", "veo_2_t2v": "veo_2_0_t2v", } DEFAULT_HEADERS = { "Origin": "https://labs.google", "Referer": "https://labs.google/fx/tools/video-fx", "Content-Type": "application/json", } POLL_INTERVAL = 10 POLL_TIMEOUT = 300 class GoogleVeoVideoProvider(VideoProvider): """Generates I2V videos via Google Veo using Playwright for reCAPTCHA handling. Supports both: - I2V (Image-to-Video): provide start_image_media_id from GoogleFlowImageProvider - T2V (Text-to-Video): prompt only, no image """ name = "google_veo" def __init__(self): self.session_token = os.getenv("GOOGLE_FLOW_SESSION_TOKEN", "") self.csrf_token = os.getenv("GOOGLE_FLOW_CSRF_TOKEN", "") self.project_id = os.getenv("GOOGLE_FLOW_PROJECT_ID", "") aspect_key = os.getenv("GOOGLE_VEO_ASPECT_RATIO", os.getenv("GOOGLE_FLOW_ASPECT_RATIO", "portrait")) self.aspect_ratio = VIDEO_ASPECT_RATIOS.get(aspect_key, "VIDEO_ASPECT_RATIO_PORTRAIT") model_key = os.getenv("GOOGLE_VEO_MODEL", "veo_3_1_i2v_fast") self.model = VIDEO_MODELS.get(model_key, model_key) if not self.session_token: raise EnvironmentError( "GOOGLE_FLOW_SESSION_TOKEN is required. " "Extract from Chrome DevTools → Application → Cookies → " "https://labs.google → __Secure-next-auth.session-token" ) if not self.csrf_token: raise EnvironmentError( "GOOGLE_FLOW_CSRF_TOKEN is required. " "Extract from Chrome DevTools → Application → Cookies → " "https://labs.google → __Host-next-auth.csrf-token" ) self._access_token: str | None = None self.http_client = httpx.Client(timeout=120, follow_redirects=True) # ------------------------------------------------------------------ # Auth helpers # ------------------------------------------------------------------ def _cookie_string(self) -> str: return ( f"__Secure-next-auth.session-token={self.session_token}; " f"__Host-next-auth.csrf-token={self.csrf_token}" ) def _refresh_access_token(self) -> str: if self._access_token: return self._access_token resp = self.http_client.get( SESSION_URL, headers={**DEFAULT_HEADERS, "Cookie": self._cookie_string()}, ) resp.raise_for_status() data = resp.json() self._access_token = data.get("access_token") or data.get("accessToken") if not self._access_token: raise RuntimeError("No access_token — cookies may be expired.") return self._access_token def _auth_headers(self) -> dict[str, str]: token = self._refresh_access_token() return { **DEFAULT_HEADERS, "Cookie": self._cookie_string(), "Authorization": f"Bearer {token}", } def _ensure_project(self) -> str: if self.project_id: return self.project_id logger.info("Creating new Flow project for video...") resp = self.http_client.post( CREATE_PROJECT_URL, json={"json": {"projectTitle": f"veo-{int(time.time())}", "toolName": "PINHOLE"}}, headers=self._auth_headers(), ) resp.raise_for_status() data = resp.json() self.project_id = data["result"]["data"]["json"]["result"]["projectId"] logger.info("Created project: %s", self.project_id) return self.project_id # ------------------------------------------------------------------ # Playwright-based video generation (handles reCAPTCHA) # ------------------------------------------------------------------ def _generate_via_browser( self, prompt: str, start_image_media_id: str | None = None ) -> dict: """Use Playwright to submit a video generation request. The browser handles reCAPTCHA automatically. We intercept the response to get the operation name for polling. """ from playwright.sync_api import sync_playwright captured_response = None with sync_playwright() as p: browser = p.chromium.launch(headless=False) context = browser.new_context( viewport={"width": 1400, "height": 900}, locale="pt-BR", ) context.add_cookies([ { "name": "__Secure-next-auth.session-token", "value": self.session_token, "domain": "labs.google", "path": "/", "secure": True, "httpOnly": True, "sameSite": "Lax", }, { "name": "__Host-next-auth.csrf-token", "value": self.csrf_token, "domain": "labs.google", "path": "/", "secure": True, "httpOnly": True, "sameSite": "Lax", }, ]) page = context.new_page() # Intercept fetch to capture video generation response page.add_init_script(""" const _origFetch = window.fetch; window.__veoResponse = null; window.fetch = async function(...args) { const [url, opts] = args; const urlStr = typeof url === 'string' ? url : url.url; const resp = await _origFetch.apply(this, args); if (urlStr.includes('batchAsyncGenerateVideo')) { const clone = resp.clone(); try { window.__veoResponse = await clone.json(); } catch(e) {} } return resp; }; """) logger.info("Launching browser for Veo I2V generation...") page.goto( "https://labs.google/fx/tools/video-fx", wait_until="networkidle", timeout=60000, ) page.wait_for_timeout(3000) # Click "New project" try: page.locator("button:has-text('Novo projeto'), button:has-text('New project')").first.click(timeout=5000) logger.info("Clicked new project") page.wait_for_timeout(5000) except Exception: logger.warning("No 'New project' button") # Wait for editor try: page.locator("textarea").first.wait_for(state="visible", timeout=30000) except Exception: page.screenshot(path="veo_debug.png") browser.close() raise RuntimeError("Editor not loaded. Screenshot: veo_debug.png") # If we have a start image, we need to select it from the asset library # For now, we'll use T2V if no media_id, or the UI automation for I2V if start_image_media_id: logger.info("I2V mode: using mediaId %s...", start_image_media_id[:30]) # The image should already be in the user's Flow library # We need to click the "add image" button and select it # This is complex UI automation - for now, fall back to T2V logger.warning("I2V UI automation not yet implemented, using T2V") # Type prompt page.locator("textarea").first.click() page.wait_for_timeout(300) page.keyboard.type(prompt) logger.info("Typed prompt") page.wait_for_timeout(1000) # Click generate try: page.locator("button:has-text('Criar'), button:has-text('Create')").first.click(timeout=5000) logger.info("Clicked generate") except Exception: page.screenshot(path="veo_debug_no_btn.png") browser.close() raise RuntimeError("Generate button not found") # Wait for response logger.info("Waiting for generation response...") for _ in range(30): page.wait_for_timeout(2000) resp_data = page.evaluate("window.__veoResponse") if resp_data: captured_response = resp_data break if not captured_response: browser.close() raise RuntimeError("No generation response captured") browser.close() return captured_response # ------------------------------------------------------------------ # Polling # ------------------------------------------------------------------ def _poll_status(self, operation_name: str, scene_id: str) -> dict: """Poll until video generation completes.""" status = "MEDIA_GENERATION_STATUS_PENDING" start = time.time() while time.time() - start < POLL_TIMEOUT: body = { "operations": [ { "operation": {"name": operation_name}, "sceneId": scene_id, "status": status, } ] } resp = self.http_client.post( STATUS_URL, json=body, headers=self._auth_headers(), ) resp.raise_for_status() data = resp.json() ops = data.get("operations", []) if not ops: time.sleep(POLL_INTERVAL) continue op = ops[0] status = op.get("status", status) logger.info("Poll: status=%s (%.0fs)", status, time.time() - start) if status == "MEDIA_GENERATION_STATUS_COMPLETED": return op if status in ("MEDIA_GENERATION_STATUS_FAILED", "MEDIA_GENERATION_STATUS_CANCELLED"): raise RuntimeError(f"Video generation failed: {status}") time.sleep(POLL_INTERVAL) raise TimeoutError(f"Video generation timed out after {POLL_TIMEOUT}s") # ------------------------------------------------------------------ # Public API # ------------------------------------------------------------------ async def generate( self, image_path: Path, video_prompt: str, output_path: Path, on_progress: Callable[[int], None] | None = None, ) -> Path: """Async wrapper for image-to-video generation. Maps ABC signature to internal sync method. Note: image_path is ignored (Google Veo uses prompt-based generation). video_prompt is passed as prompt to _generate_sync. """ return await asyncio.to_thread( self._generate_sync, video_prompt, output_path, ) def _generate_sync( self, prompt: str, output_path: str | Path, start_image_media_id: str | None = None, seed: int | None = None, ) -> Path: """Sync implementation of generate.""" output_path = Path(output_path) output_path.parent.mkdir(parents=True, exist_ok=True) mode = "I2V" if start_image_media_id else "T2V" logger.info("Generating %s video via Google Veo → %s", mode, output_path) # Use browser to handle reCAPTCHA gen_response = self._generate_via_browser(prompt, start_image_media_id) # Extract operation info ops = gen_response.get("operations", []) if not ops: raise RuntimeError(f"No operations: {json.dumps(gen_response)[:500]}") # Take first operation (we generate 1 video) op = ops[0] operation_name = op.get("operation", {}).get("name", "") scene_id = op.get("sceneId", "") if not operation_name: raise RuntimeError(f"No operation name: {json.dumps(op)[:500]}") logger.info("Video generation started: %s", operation_name) # Poll for completion completed_op = self._poll_status(operation_name, scene_id) # Download video video_url = ( completed_op.get("fifeVideoUrl") or completed_op.get("videoUrl") or completed_op.get("result", {}).get("fifeVideoUrl") ) encoded_video = ( completed_op.get("encodedVideo") or completed_op.get("result", {}).get("encodedVideo") ) if encoded_video: import base64 video_data = base64.b64decode(encoded_video) elif video_url: logger.info("Downloading video...") dl_resp = httpx.get(video_url, timeout=120, follow_redirects=True) dl_resp.raise_for_status() video_data = dl_resp.content else: raise RuntimeError( f"No video URL/data. Keys: {list(completed_op.keys())}" ) with open(output_path, "wb") as f: f.write(video_data) logger.info("Video saved: %s (%d bytes)", output_path, len(video_data)) return output_path def close(self): self.http_client.close() def __enter__(self): return self def __exit__(self, *args): self.close()