#!/usr/bin/env python3 """ Generate and upload 10 Christian corrido tumbao concert videos from Grok to Google Drive. Uses Playwright connected via CDP to the existing Chrome session. """ from playwright.sync_api import sync_playwright import subprocess import time import sys # 10 prompts for Christian corrido tumbao concert - consistent character, crowd, aerial shots, 16:9 horizontal PROMPTS = [ "A charismatic singer performs on a massive outdoor stage with purple and gold lights, thousands of fans singing along with raised hands, aerial drone shot revealing the enormous crowd, Christian corrido tumbado aesthetic, cinematic wide shot, 16x9 horizontal format", "An artist in a cowboy hat leads a massive crowd in prayer and worship at a huge open-air concert, golden hour lighting, aerial view showing tens of thousands of people, Christian corrido tumbao mood, epic cinematic scene, 16x9 horizontal", "Lead singer with a guitar performs an emotional corrido while the crowd holds up phone lights creating a sea of stars, dramatic stage lighting in blue and gold, stadium packed with fans, aerial tracking shot, Christian urban worship atmosphere, 16x9 horizontal", "A Mexican-American corrido artist commands the stage with confetti cannons firing, massive crowd going wild with excitement, aerial drone shot circling the venue, stage with cross-shaped LED screen behind the performer, Christian celebration, cinematic, 16x9 horizontal", "Street corrido performer sings on a truck bed stage at a massive outdoor festival, crowd of thousands dancing and singing along, sunset sky with dramatic clouds, aerial wide shot of the festival grounds, Christian tumbao vibes, cinematic, 16x9 horizontal", "Singer in a leather jacket performs intimate moment during concert while 50,000 people wave lighters and phones, dramatic red and gold stage lighting, aerial pan shot revealing enormous crowd stretching to the horizon, Christian corrido worship, 16x9 horizontal", "Young corrido artist performs on stage with fog machines and dramatic spotlights, crowd members lifted on shoulders singing every word, aerial drone shot diving through the crowd toward the stage, stadium concert atmosphere, Christian faith celebration, 16x9 horizontal", "Lead vocalist with full band plays an upbeat Christian corrido, massive dance floor filled with young crowd, confetti and pyrotechnics on stage, aerial establishing shot of packed arena with light shows, tumbao rhythm atmosphere, cinematic, 16x9 horizontal", "An artist performs an acoustic corrido in a dramatic church converted into concert venue, crowd of thousands with hands raised, dramatic stained glass window as backdrop, aerial shot slowly rotating above the packed floor, Christian worship and celebration, 16x9 horizontal", "Final concert scene with artist conducting the massive crowd in a spiritual anthem, every phone flashlight on creating ocean of light, dramatic fireworks from stage, aerial wide shot showing city-block-sized crowd, Christian corrido tumbado finale, cinematic grandeur, 16x9 horizontal" ] DRIVE_FOLDER_ID = "1NITLT29n0cqNMuHwEpIDx99EUE0hgLHP" VIDEO_DIR = "/home/claw/videos" DONE_FILE = "/tmp/grok-gen-state.json" def get_grok_page(page): """Navigate to Grok Imagine page.""" page.goto("https://grok.com/imagine") page.wait_for_timeout(3000) return page def set_video_mode_16x9(page): """Configure video mode, 720p, 10s, 16:9.""" # Click Video radio video_radio = page.get_by_role("radio", name="Video") if video_radio.count() > 0: video_radio.click() page.wait_for_timeout(500) # Click 720p radio p720 = page.get_by_role("radio", name="720p") if p720.count() > 0: p720.click() page.wait_for_timeout(200) # Click 10s radio s10 = page.get_by_role("radio", name="10s") if s10.count() > 0: s10.click() page.wait_for_timeout(200) # Click Relación de aspecto and look for 16:9 aspect_btn = page.get_by_role("button", name="Relación de aspecto") if aspect_btn.count() > 0: aspect_btn.click() page.wait_for_timeout(1000) # Try to find 16:9 option opts = page.locator('[role=option], [role=menuitem], button').all() for opt in opts: txt = opt.inner_text()[:30] if '16' in txt or '9' in txt or 'horizontal' in txt.lower(): opt.click() page.wait_for_timeout(500) break def wait_for_generation(page, timeout=120): """Wait for video generation to complete.""" start = time.time() while time.time() - start < timeout: # Check for video player or download button download_btn = page.get_by_role("button", name="Descargar") if download_btn.count() > 0: return True progress = page.locator('[role=progressbar], .progress, [class*=progress]').all() page.wait_for_timeout(3000) elapsed = int(time.time() - start) print(f" [{elapsed}s] waiting for generation...", flush=True) return False def main(): with sync_playwright() as p: # Connect to existing Chrome via CDP cdp = p.chromium.connect_over_cdp("http://localhost:9223") print("Connected to Chrome via CDP") # Get or create page if cdp.contexts and cdp.contexts[0].pages: page = cdp.contexts[0].pages[0] else: page = cdp.contexts[0].new_page() # Start HTTP server for video downloads import threading import http.server import os class QuietHandler(http.server.SimpleHTTPRequestHandler): def log_message(self, format, *args): pass # suppress logging os.chdir(VIDEO_DIR) server = http.server.HTTPServer(("localhost", 8888), QuietHandler) t = threading.Thread(target=server.serve_forever) t.daemon = True t.start() print(f"HTTP server started on port 8888 serving {VIDEO_DIR}") # Navigate to Grok page.goto("https://grok.com/imagine") page.wait_for_timeout(3000) print("On Grok Imagine page") all_video_urls = [] for i, prompt in enumerate(PROMPTS): print(f"\n--- Video {i+1}/10 ---") print(f"Prompt: {prompt[:80]}...") # Reload Grok Imagine fresh for each video page.goto("https://grok.com/imagine") page.wait_for_timeout(3000) # Set video mode set_video_mode_16x9(page) # Find the contenteditable prompt input and type prompt_input = page.locator("div[contenteditable=true]").first prompt_input.click() prompt_input.fill(prompt) page.wait_for_timeout(500) # Press Enter to generate page.keyboard.press("Enter") print(" Generating...") # Wait for video if wait_for_generation(page): print(" ✅ Video ready!") # Get video URL video_el = page.locator("video").first if video_el.count() > 0: src = video_el.get_attribute("src") if src: all_video_urls.append((i+1, src)) print(f" Video URL: {src[:80]}") else: print(f" ❌ Generation timed out for video {i+1}") # Small delay between generations page.wait_for_timeout(2000) print(f"\n=== Downloaded {len(all_video_urls)} video URLs ===") # Now download and upload each for idx, (num, url) in enumerate(all_video_urls): local_file = f"{VIDEO_DIR}/corrido_tumba_{num:02d}.mp4" # Download if not exists if not os.path.exists(local_file) or os.path.getsize(local_file) < 1000: print(f"Downloading video {num} from {url[:60]}...") result = subprocess.run( ["curl", "-L", "-A", "Mozilla/5.0", "-o", local_file, url], capture_output=True, text=True, timeout=60 ) size = os.path.getsize(local_file) if os.path.exists(local_file) else 0 print(f" Downloaded: {size} bytes") else: print(f"Video {num} already exists: {os.path.getsize(local_file)} bytes") # Upload all 10 to Drive print("\n=== Uploading to Google Drive ===") page.goto(f"https://drive.google.com/drive/folders/{DRIVE_FOLDER_ID}") page.wait_for_timeout(2000) for num in range(1, 11): local_file = f"{VIDEO_DIR}/corrido_tumba_{num:02d}.mp4" if os.path.exists(local_file) and os.path.getsize(local_file) > 1000: print(f"Uploading corrido_tumba_{num:02d}.mp4...") # Click Nuevo > Subir archivo page.get_by_role("button", name="Nuevo").click() page.wait_for_timeout(1000) page.get_by_role("menuitem", name="Subir archivo").click() page.wait_for_timeout(1000) # Set file file_input = page.locator("input[type=file]").last file_input.set_input_files([local_file]) page.wait_for_timeout(8000) # wait for upload print(f" ✅ Video {num} uploaded") else: print(f" ⚠️ Video {num} not found locally") print("\n=== ALL DONE ===") server.shutdown() cdp.close() if __name__ == "__main__": main()