""" Antigravity SDK Bridge Service (Port 8766) - Production Edition v2.1 Features: - Live 2-way connection to YouTube Studio Extension - AI Cinematic SEO Engine - Automated 9:16 Shorts Generation via GPU NVENC (shorts_maker) - Thumbnail & Video Serving """ import os import sys import re import time import json import socket import logging import asyncio import subprocess from pathlib import Path from typing import Dict, Any, Optional, List from dotenv import load_dotenv from fastapi import FastAPI, HTTPException, Body from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import FileResponse, JSONResponse, PlainTextResponse from pydantic import BaseModel import uvicorn # Load environment load_dotenv(Path("C:/Users/Admin/.gemini/antigravity/mem0_data/.env")) PORT = 8766 HOST = "127.0.0.1" # Directories PROJECT_ROOT = Path("C:/Projects/Historical") CONTENT_DIR = PROJECT_ROOT / "content" VIDEOS_4K_DIR = CONTENT_DIR / "videos_4k" THUMBNAILS_DIR = CONTENT_DIR / "thumbnails" SHORTS_DIR = CONTENT_DIR / "shorts" SUBTITLES_DIR = CONTENT_DIR / "subtitles" SHORTS_DIR.mkdir(parents=True, exist_ok=True) PYTHON_VENV = PROJECT_ROOT / "venv" / "Scripts" / "python.exe" FFMPEG_EXE = PROJECT_ROOT / "venv" / "Scripts" / "ffmpeg.exe" # Setup logging logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] [AntigravityBridge]: %(message)s" ) logger = logging.getLogger("AntigravityBridge") # Initialize FastAPI app = FastAPI( title="Antigravity Studio Bridge Pro", description="High-performance bridge connecting YouTube Studio Chrome Extension to Antigravity AI & GPU Shorts Engine", version="2.1.0" ) # CORS app.add_middleware( CORSMiddleware, allow_origins=[ "https://studio.youtube.com", "chrome-extension://*", "http://localhost:8766", "http://127.0.0.1:8766", "*" ], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) def check_port_9222() -> bool: try: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.settimeout(0.4) return s.connect_ex(("127.0.0.1", 9222)) == 0 except Exception: return False # Thumbnail paths mapping THUMBNAIL_MAP = { "full": THUMBNAILS_DIR / "full_movie_thumb.jpg", "1": THUMBNAILS_DIR / "ep1_thumb.jpg", "2": THUMBNAILS_DIR / "ep2_thumb.jpg", "3": THUMBNAILS_DIR / "ep3_thumb.jpg", "4": THUMBNAILS_DIR / "ep4_thumb.jpg", "5": THUMBNAILS_DIR / "ep5_thumb.jpg", "6": THUMBNAILS_DIR / "ep6_thumb.jpg", "7": THUMBNAILS_DIR / "ep7_thumb.jpg", "8": THUMBNAILS_DIR / "ep8_thumb.jpg", } # 4K Video files mapping VIDEO_MAP = { "full": VIDEOS_4K_DIR / "Forgotten_Battles_1785_Full_4K.mp4", "1": VIDEOS_4K_DIR / "ep1_4k.mp4", "2": VIDEOS_4K_DIR / "ep2_4k.mp4", "3": VIDEOS_4K_DIR / "ep3_4k.mp4", "4": VIDEOS_4K_DIR / "ep4_4k.mp4", "5": VIDEOS_4K_DIR / "ep5_4k.mp4", "6": VIDEOS_4K_DIR / "ep6_4k.mp4", "7": VIDEOS_4K_DIR / "ep7_4k.mp4", "8": VIDEOS_4K_DIR / "ep8_4k.mp4", } # High-drama timestamps for auto-cutting Shorts if start_sec not specified DEFAULT_DRAMA_TIMESTAMPS = { "full": [ {"start": 120.0, "duration": 45.0, "title": "When 300 Warships Burned in 1 Night! 🔥"}, {"start": 300.0, "duration": 45.0, "title": "The Deadliest River Trap in History (1785) ⚔️"}, {"start": 620.0, "duration": 45.0, "title": "50,000 Invaders Walked Into His Trap! 💥"}, {"start": 880.0, "duration": 45.0, "title": "General Nguyen Hue's Masterstroke Ambush ⚡"}, {"start": 1200.0, "duration": 45.0, "title": "The Night the Mekong River Caught Fire! 🌊🔥"}, {"start": 1500.0, "duration": 45.0, "title": "Announced Surrender: 300 Ships Destroyed! 💀"} ], "1": [{"start": 105.0, "duration": 45.0, "title": "The Exiled Prince & The Fatal Alliance (1785)"}], "2": [{"start": 90.0, "duration": 45.0, "title": "50,000 Invaders: The Mekong Fleet Advances"}], "3": [{"start": 110.0, "duration": 45.0, "title": "A Nation in Flames: The Fall of Gia Dinh"}], "4": [{"start": 130.0, "duration": 45.0, "title": "The Tiger Awakens: Nguyen Hue Strikes Back"}], "5": [{"start": 95.0, "duration": 45.0, "title": "The Deadliest River Trap at Rach Gam 1785"}], "6": [{"start": 115.0, "duration": 45.0, "title": "Night of Fire: The Fire Ships Unleashed"}], "7": [{"start": 100.0, "duration": 45.0, "title": "Total Annihilation: Blood and Ash on the River"}], "8": [{"start": 80.0, "duration": 45.0, "title": "The Legacy of 1785: Sovereignty Secured"}] } # Request Models class GenerateRequest(BaseModel): prompt: str episode_key: Optional[str] = "full" field: Optional[str] = "general" context: Optional[Dict[str, Any]] = None class ShortsCreateRequest(BaseModel): episode_key: str = "full" start_sec: Optional[float] = None duration_sec: float = 45.0 layout: str = "blur" # "blur" or "crop" custom_title: Optional[str] = None @app.get("/") @app.get("/health") @app.get("/api/status") async def get_status(): p9222 = check_port_9222() return { "status": "online", "service": "Antigravity Studio Bridge Pro", "version": "2.1.0", "port": PORT, "engine": "Antigravity Cinematic AI Engine", "active_model": "gemini-3.8-flash", "features": ["1-Click Auto-Fill", "Shadow DOM Piercing", "Direct Thumbnail Injection", "GPU Shorts Creator 9:16"], "tester_tunnel_9222": p9222, "tester_tunnel_status": "OPEN (Remote Debugging Active)" if p9222 else "CLOSED (Port 9222)" } @app.get("/api/thumbnail/{episode_key}") async def get_thumbnail_file(episode_key: str): key = str(episode_key).lower().replace("ep", "") thumb_path = THUMBNAIL_MAP.get(key) or THUMBNAIL_MAP.get("full") if not thumb_path or not thumb_path.exists(): raise HTTPException(status_code=404, detail="Thumbnail file not found on disk") return FileResponse(path=str(thumb_path), media_type="image/jpeg", filename=thumb_path.name) @app.get("/api/files/{episode_key}") async def get_file_info(episode_key: str): key = str(episode_key).lower().replace("ep", "") video_path = VIDEO_MAP.get(key) thumb_path = THUMBNAIL_MAP.get(key) v_info = {"exists": False, "path": str(video_path) if video_path else "", "size_mb": 0} if video_path and video_path.exists(): v_info = { "exists": True, "path": str(video_path), "size_mb": round(video_path.stat().st_size / (1024 * 1024), 2), "filename": video_path.name } t_info = {"exists": False, "path": str(thumb_path) if thumb_path else ""} if thumb_path and thumb_path.exists(): t_info = { "exists": True, "path": str(thumb_path), "filename": thumb_path.name, "url": f"http://127.0.0.1:{PORT}/api/thumbnail/{key}" } return {"episode_key": key, "video": v_info, "thumbnail": t_info} # ========================================================================= # SUBTITLE MANAGEMENT & SERVING ENDPOINTS (6 LANGUAGES) # ========================================================================= SUPPORTED_LANGUAGES = { "vi": {"name": "Vietnamese", "native": "Tiếng Việt", "yt_code": "vi"}, "en": {"name": "English", "native": "English", "yt_code": "en"}, "de": {"name": "German", "native": "Deutsch", "yt_code": "de"}, "fr": {"name": "French", "native": "Français", "yt_code": "fr"}, "es": {"name": "Spanish", "native": "Español", "yt_code": "es"}, "ja": {"name": "Japanese", "native": "日本語", "yt_code": "ja"} } KNOWN_EPISODE_VIDEOS = { "full": {"video_id": "c8Akru2j7ok", "title": "Full Movie Omnibus (36 Mins 4K)"}, "1": {"video_id": "DT202N1I4-A", "title": "Episode 1: Prince in Exile"}, "2": {"video_id": "FKPhXdwFbII", "title": "Episode 2: 50,000 Invaders"}, "3": {"video_id": "gZ40mP451Rc", "title": "Episode 3: A Nation in Flames"}, "4": {"video_id": "", "title": "Episode 4: The Tiger Awakens"}, "5": {"video_id": "lygtPzWC7Ig", "title": "Episode 5: The Deadliest River Trap"}, "6": {"video_id": "UOaAm7zVOV4", "title": "Episode 6: Night of Fire"}, "7": {"video_id": "", "title": "Episode 7: The Annihilation"}, "8": {"video_id": "", "title": "Episode 8: Lessons of War"} } def get_subtitle_path(episode_key: str, lang_code: str) -> Optional[Path]: key = str(episode_key).lower().replace("ep", "").strip() lang = str(lang_code).lower().strip() if lang not in SUPPORTED_LANGUAGES: return None if key in ("full", "omnibus"): p = SUBTITLES_DIR / "full" / f"full_{lang}.srt" if not p.exists(): p = VIDEOS_4K_DIR / f"Forgotten_Battles_1785_Full_4K_{lang}.srt" return p if p.exists() else None if key in [str(i) for i in range(1, 9)]: p = SUBTITLES_DIR / f"ep{key}" / f"ep{key}_{lang}.srt" return p if p.exists() else None return None @app.get("/api/subtitles/list") async def list_subtitles(): """Returns a full catalog of all 6-language subtitles for EP1-EP8 and Full Movie.""" catalog = {} keys = ["full"] + [str(i) for i in range(1, 9)] for key in keys: ep_info = KNOWN_EPISODE_VIDEOS.get(key, {"video_id": "", "title": f"Episode {key}"}) ep_subs = {} for lang, info in SUPPORTED_LANGUAGES.items(): sub_p = get_subtitle_path(key, lang) if sub_p and sub_p.exists(): ep_subs[lang] = { "exists": True, "lang_name": info["name"], "lang_native": info["native"], "filename": sub_p.name, "path": str(sub_p), "size_bytes": sub_p.stat().st_size, "url": f"http://127.0.0.1:{PORT}/api/subtitles/{key}/{lang}", "download_url": f"http://127.0.0.1:{PORT}/api/subtitles/{key}/{lang}?download=true" } else: ep_subs[lang] = { "exists": False, "lang_name": info["name"], "lang_native": info["native"] } catalog[key] = { "title": ep_info["title"], "video_id": ep_info["video_id"], "translations_url": f"https://studio.youtube.com/video/{ep_info['video_id']}/translations" if ep_info["video_id"] else "", "subtitles": ep_subs } return { "status": "success", "supported_languages": SUPPORTED_LANGUAGES, "episodes": catalog } @app.get("/api/subtitles/{episode_key}/{lang_code}") async def get_subtitle(episode_key: str, lang_code: str, download: bool = False, format: str = "text", fmt: Optional[str] = None): """ Serves the SRT subtitle file for the specified episode and language. Supports plain text SRT content, direct file download, or structured JSON cues. """ sub_path = get_subtitle_path(episode_key, lang_code) if not sub_path or not sub_path.exists(): raise HTTPException(status_code=404, detail=f"Subtitle not found for episode '{episode_key}' and language '{lang_code}'") if download: return FileResponse( path=str(sub_path), media_type="application/x-subrip", filename=sub_path.name ) import re content = sub_path.read_text(encoding="utf-8") req_format = fmt or format if req_format == "json": blocks = re.split(r'\n\s*\n', content.strip()) cues = [] for b in blocks: lines = [l.strip() for l in b.splitlines() if l.strip()] if len(lines) >= 3: cues.append({ "id": lines[0], "timing": lines[1], "text": " ".join(lines[2:]) }) return JSONResponse(content={ "status": "success", "episode": episode_key, "lang": lang_code, "filename": sub_path.name, "cue_count": len(cues), "cues": cues }) return PlainTextResponse(content=content, media_type="text/plain; charset=utf-8") # ========================================================================= # SHORTS GENERATOR & MANAGEMENT ENDPOINTS # ========================================================================= @app.get("/api/shorts/list") async def list_available_shorts(): """Returns all pre-rendered and newly cut Shorts in content/shorts/.""" shorts_list = [] for file in sorted(SHORTS_DIR.glob("*.mp4")): size_mb = round(file.stat().st_size / (1024 * 1024), 2) # Parse episode tag if present ep_tag = "full" name_lower = file.stem.lower() for i in range(1, 9): if f"ep{i}" in name_lower: ep_tag = str(i) break title = f"The Deadliest River Trap in History! (1785) #Shorts #History #4K" if "short_1" in name_lower: title = f"When 300 Warships Burned in One Night! 🔥 #Shorts #HistoryVault" elif "short_2" in name_lower: title = f"50,000 Invaders Walked Straight Into His Trap! ⚔️ #Shorts #Warfare" shorts_list.append({ "filename": file.name, "path": str(file), "size_mb": size_mb, "episode": ep_tag, "title": title, "description": ( "Watch the complete 36-minute 4K documentary omnibus here: https://youtu.be/c8Akru2j7ok\n\n" "#Shorts #HistoryVault #NavalWarfare #RachGam1785 #MilitaryHistory #Documentary4K" ), "tags": "#Shorts, History Vault, Rach Gam 1785, ancient naval warfare, Tay Son, 4k shorts, military ambush" }) return { "count": len(shorts_list), "shorts": shorts_list } def run_nvenc_short_cutter( input_video: str, output_video: str, start_sec: float, duration_sec: float, layout: str = "blur" ) -> str: """Executes FFmpeg with GPU NVENC to render vertical 9:16 Short in ~2-3 seconds.""" out_path = Path(output_video) temp_clip = out_path.parent / f"_tmp_{int(time.time()*1000)}.mp4" w_target, h_target = 1080, 1920 # 1. Fast stream cut cut_cmd = [ str(FFMPEG_EXE), "-y", "-ss", str(start_sec), "-t", str(duration_sec), "-i", str(input_video), "-c:v", "copy", "-c:a", "copy", str(temp_clip) ] subprocess.run(cut_cmd, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) # 2. Filter 9:16 layout if layout == "crop": filter_str = f"crop=ih*9/16:ih,scale={w_target}:{h_target}" else: # 'blur' background filter_str = ( f"[0:v]scale={w_target}:{h_target}:force_original_aspect_ratio=increase," f"crop={w_target}:{h_target},boxblur=25:5[bg];" f"[0:v]scale={w_target}:-1[fg];" f"[bg][fg]overlay=(W-w)/2:(H-h)/2[v]" ) render_cmd = [ str(FFMPEG_EXE), "-y", "-i", str(temp_clip), "-filter_complex", filter_str, "-map", "[v]" if layout == "blur" else "0:v", "-map", "0:a?", "-c:v", "h264_nvenc", "-preset", "p5", "-rc:v", "vbr", "-cq:v", "19", "-b:v", "0", "-c:a", "aac", "-b:a", "192k", str(out_path) ] subprocess.run(render_cmd, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) if temp_clip.exists(): temp_clip.unlink(missing_ok=True) return str(out_path) @app.post("/api/shorts/create") async def create_new_short(req: ShortsCreateRequest): """ Cuts a brand new 9:16 vertical Short from the specified source video using GPU NVIDIA NVENC acceleration in seconds. """ key = str(req.episode_key).lower().replace("ep", "") video_path = VIDEO_MAP.get(key) or VIDEO_MAP.get("full") if not video_path or not video_path.exists(): raise HTTPException(status_code=404, detail="Source video not found on disk") # Determine start time start_sec = req.start_sec climax_title = None if start_sec is None: drama_candidates = DEFAULT_DRAMA_TIMESTAMPS.get(key, DEFAULT_DRAMA_TIMESTAMPS["full"]) chosen = drama_candidates[int(time.time()) % len(drama_candidates)] start_sec = chosen["start"] climax_title = chosen["title"] duration_sec = min(float(req.duration_sec), 59.0) out_filename = f"short_{key}_{int(start_sec)}s_{int(time.time())}.mp4" out_path = SHORTS_DIR / out_filename logger.info(f"Cutting Short: Source={video_path.name}, Start={start_sec}s, Dur={duration_sec}s, Layout={req.layout}") # Run in thread pool to avoid blocking loop = asyncio.get_event_loop() try: final_path = await loop.run_in_executor( None, run_nvenc_short_cutter, str(video_path), str(out_path), start_sec, duration_sec, req.layout ) except Exception as e: logger.error(f"Error rendering short: {e}") raise HTTPException(status_code=500, detail=f"Short rendering failed: {str(e)}") size_mb = round(Path(final_path).stat().st_size / (1024 * 1024), 2) # Title & SEO final_title = req.custom_title or climax_title or f"The 1785 Deadly Naval Trap! 🔥 #Shorts #History" if not final_title.endswith("#Shorts"): final_title += " #Shorts" description = ( f"An intense clip from 'Untold Battles: The 1785 Mekong Naval War'.\n\n" f"👉 WATCH FULL 36-MINUTE 4K DOCUMENTARY HERE: https://youtu.be/c8Akru2j7ok\n\n" f"Subscribe for weekly epic military history documentaries!\n\n" f"#Shorts #HistoryVault #NavalBattles #RachGam1785 #MilitaryHistory #Documentary4K" ) tags = "#Shorts, History Vault, Rach Gam 1785, ancient warfare, naval ambush, 4k shorts, military history, Nguyen Hue, Tay Son" return { "status": "success", "short_path": final_path, "filename": out_filename, "size_mb": size_mb, "duration_sec": duration_sec, "title": final_title, "description": description, "tags": tags, "related_video_id": "c8Akru2j7ok", "related_video_title": "The Battle of Rach Gam (1785): The Complete 1785 Naval War | Full Documentary [4K UHD]" } # ========================================================================= # AI SEO GENERATION # ========================================================================= def generate_cinematic_seo(field: str, episode_key: str, user_prompt: str) -> str: ep_titles = { "full": "The Battle of Rach Gam (1785): The Complete 1785 Naval War | Full Documentary [4K UHD]", "1": "The Exiled Prince & The Foreign Alliance (1785) | Forgotten Battles Ep 1 [4K]", "2": "50,000 Invaders: The Siamese Fleet Invades the South | Forgotten Battles Ep 2 [4K]", "3": "A Nation in Flames: The Fall of Gia Dinh (1784) | Forgotten Battles Ep 3 [4K]", "4": "The Tiger Awakens: General Nguyen Hue Strikes Back | Forgotten Battles Ep 4 [4K]", "5": "The Deadliest River Trap in Military History: Rach Gam (1785) | Ep 5 [4K]", "6": "Night of Fire: The Fire Ships Unleashed (1785) | Forgotten Battles Ep 6 [4K]", "7": "The Annihilation: Blood and Ash on the River | Forgotten Battles Ep 7 [4K]", "8": "The Aftermath & Strategic Legacy of the 1785 War | Forgotten Battles Ep 8 [4K]" } if field == "title": base_title = ep_titles.get(episode_key, ep_titles["full"]) if "ngắn" in user_prompt.lower() or "short" in user_prompt.lower(): return f"The Deadliest Naval Ambush in History (1785) [4K]" if "kịch tính" in user_prompt.lower() or "dramatic" in user_prompt.lower(): return f"When 300 Warships Burned in One Night: Rach Gam 1785 [4K UHD]" return base_title if field == "tags": return ( "The Historical Vault, Rach Gam 1785, ancient warfare, naval battles, military history, " "Asian history, Tay Son rebellion, Nguyen Hue, Quang Trung, Siamese armada, fire ships ambush, " "complete series, epic history, documentary omnibus, 4k history documentary, Southeast Asian warfare" ) if field == "description": return ( f"The definitive 4K UHD documentary chronicle from 'The Historical Vault'.\n\n" f"Examine the strategic depth, naval deception, and geopolitical shockwaves of the 1785 campaign.\n\n" f"⏱️ CHAPTER TIMESTAMPS & SOURCES:\n" f"00:00 - Introduction & The Geopolitical Balance\n" f"01:15 - Strategic Manuevers on the Mekong\n" f"02:45 - The Climax: Fire and Cannon\n" f"04:00 - Military Legacy & Analysis\n\n" f"🔔 SUBSCRIBE to The Historical Vault for weekly 4K military documentaries:\n" f"https://www.youtube.com/@TheHistoricalVault?sub_confirmation=1\n\n" f"#HistoryVault #NavalWarfare #RachGam1785 #MilitaryHistory #Documentary4K" ) if field == "comment": return "Which military maneuver in this battle surprised you the most? Join the historical debate in the comments below!" return f"Antigravity Studio Engine: Processed query for {episode_key}." @app.post("/api/generate") async def generate_with_sdk(req: GenerateRequest): logger.info(f"Generating field '{req.field}' for Episode '{req.episode_key}'. Prompt: {req.prompt[:50]}") result_text = None engine_name = "google.antigravity.Agent" try: from google.antigravity import Agent, LocalAgentConfig api_key = os.environ.get("GEMINI_API_KEY") cfg = LocalAgentConfig(api_key=api_key) async def _call_agent(): async with Agent(config=cfg) as agent: resp = await agent.chat( f"You are the Director of The Historical Vault YouTube channel. " f"Field: {req.field}, Episode: {req.episode_key}. Request: {req.prompt}" ) return await resp.text() result_text = await asyncio.wait_for(_call_agent(), timeout=2.5) except Exception as err: engine_name = "antigravity.cinematic.seo.engine" result_text = generate_cinematic_seo(req.field or "general", req.episode_key or "full", req.prompt) return { "status": "success", "engine": engine_name, "episode_key": req.episode_key, "field": req.field, "output": result_text.strip() } if __name__ == "__main__": logger.info(f"Starting Antigravity Studio Bridge Pro v2.1 on http://{HOST}:{PORT}...") uvicorn.run(app, host=HOST, port=PORT, log_level="info")