""" Media asset fetcher using Pexels API This module fetches B-Roll footage based on keywords extracted from: - PDF content - Video transcription - Generated script """ import os import json import requests import time from pathlib import Path from typing import Dict, Any, List, Optional try: from google import genai from google.genai import types except ImportError: genai = None class MediaFetcher: """Fetch B-Roll videos from Pexels API""" def __init__(self, api_key: Optional[str] = None, gemini_key: Optional[str] = None): self.api_key = api_key or os.getenv("PEXELS_API_KEY") if not self.api_key: raise ValueError("PEXELS_API_KEY not found in environment") self.gemini_key = gemini_key or os.getenv("GEMINI_API_KEY") self.base_url = "https://api.pexels.com/videos/search" self.output_dir = Path("public/broll") self.output_dir.mkdir(parents=True, exist_ok=True) # Translation cache self.cache_file = Path("assets/1_processed/translation_cache.json") self.translation_cache = self._load_cache() # Initialize Gemini for translation if self.gemini_key and genai: self.gemini_client = genai.Client(api_key=self.gemini_key) else: self.gemini_client = None def _load_cache(self) -> Dict[str, str]: """Load translation cache from file""" if self.cache_file.exists(): with open(self.cache_file, 'r', encoding='utf-8') as f: return json.load(f) return {} def _save_cache(self): """Save translation cache to file""" self.cache_file.parent.mkdir(parents=True, exist_ok=True) with open(self.cache_file, 'w', encoding='utf-8') as f: json.dump(self.translation_cache, f, ensure_ascii=False, indent=2) def _translate_to_english(self, japanese_keywords: List[str]) -> Dict[str, str]: """Translate Japanese keywords to English using Gemini""" translations = {} keywords_to_translate = [kw for kw in japanese_keywords if kw not in self.translation_cache] if not keywords_to_translate: return {kw: self.translation_cache[kw] for kw in japanese_keywords} if self.gemini_client: prompt = f""" Translate the following Japanese keywords to English (single words or short phrases). Return ONLY a JSON object mapping Japanese to English. Keywords: {', '.join(keywords_to_translate)} Example format: {{ "家族": "family", "老後": "retirement" }} """ try: response = self.gemini_client.models.generate_content( model='gemini-2.0-flash-exp', contents=prompt ) response_text = response.text.strip() if "```json" in response_text: json_start = response_text.find("```json") + 7 json_end = response_text.find("```", json_start) response_text = response_text[json_start:json_end].strip() elif "```" in response_text: json_start = response_text.find("```") + 3 json_end = response_text.find("```", json_start) response_text = response_text[json_start:json_end].strip() new_translations = json.loads(response_text) self.translation_cache.update(new_translations) self._save_cache() except Exception as e: print(f" ⚠️ Translation failed: {e}, using fallback...") # Fallback translations fallback = { "家族": "family", "老後": "retirement", "退職": "retirement", "貯蓄": "savings", "資産": "finance", "住宅": "housing", "教育": "education", "保険": "insurance", "年金": "pension", "投資": "investment", "将来": "future", "計画": "planning" } for kw in keywords_to_translate: self.translation_cache[kw] = fallback.get(kw, kw) else: # No Gemini, use fallback fallback = { "家族": "family", "老後": "retirement", "退職": "retirement", "貯蓄": "savings", "資産": "finance", "住宅": "housing", "教育": "education", "保険": "insurance", "年金": "pension", "投資": "investment", "将来": "future", "計画": "planning" } for kw in keywords_to_translate: self.translation_cache[kw] = fallback.get(kw, kw) return {kw: self.translation_cache.get(kw, kw) for kw in japanese_keywords} def fetch_broll(self, keywords: List[str], max_per_keyword: int = 2) -> Dict[str, List[Dict[str, Any]]]: """ Fetch B-Roll videos based on keywords Args: keywords: List of Japanese keywords max_per_keyword: Maximum videos to fetch per keyword Returns: Dictionary mapping Japanese keywords to list of video info """ print(f" 🎬 Fetching B-Roll for keywords: {', '.join(keywords)}") # Translate keywords translations = self._translate_to_english(keywords) print(f" 🌎 Translations: {translations}") results = {} for jp_keyword, en_keyword in translations.items(): # ENHANCED: Inject explicit Japanese context for every search to avoid Caucasian subjects search_query = f"Japanese {en_keyword} people Asian" print(f" 🔍 Searching Pexels for '{en_keyword}' (with strong Japanese/Asian priority)...") # Search Pexels with "Japanese" added to the query for better relevance headers = {"Authorization": self.api_key} params = { "query": search_query, "per_page": max_per_keyword, "orientation": "landscape" } try: response = requests.get(self.base_url, headers=headers, params=params) response.raise_for_status() data = response.json() videos = [] for video in data.get("videos", [])[:max_per_keyword]: # Download HD video file video_files = video.get("video_files", []) hd_file = None # Find HD version for vf in video_files: if vf.get("quality") == "hd" and vf.get("width", 0) <= 1920: hd_file = vf break if not hd_file and video_files: hd_file = video_files[0] # Fallback to first available if hd_file: # Download video video_url = hd_file["link"] filename = f"{en_keyword.title()}_{video['id']}.mp4" filepath = self.output_dir / filename if not filepath.exists(): print(f" ⬇️ Downloading {filename}...") video_response = requests.get(video_url, stream=True) with open(filepath, 'wb') as f: for chunk in video_response.iter_content(chunk_size=8192): f.write(chunk) else: print(f" ✓ Already cached: {filename}") videos.append({ "filepath": str(filepath), "duration": video.get("duration", 0), "photographer": video.get("user", {}).get("name", "Unknown") }) results[jp_keyword] = videos print(f" ✓ Found {len(videos)} videos for '{jp_keyword}'") time.sleep(0.5) # Rate limiting except requests.exceptions.RequestException as e: print(f" ⚠️ Pexels API error for '{en_keyword}': {e}") results[jp_keyword] = [] return results def _translate_keyword(self, japanese_keyword: str) -> str: """Translate Japanese keyword to English using Gemini""" # TODO: Implement translation # Use Gemini Pro for accurate translation # Cache translations to avoid repeated API calls return japanese_keyword def _download_video(self, url: str, output_path: str) -> str: """Download video from URL""" # TODO: Implement video download # Use requests to download # Show progress bar # Validate video format return output_path def fetch_broll(script: Dict[str, Any]) -> Dict[str, Any]: """Convenience function for fetching B-Roll assets""" # Extract keywords from script keywords = [] # Try to extract from video_data if available if "video_data" in script and "keywords" in script["video_data"]: keywords.extend(script["video_data"]["keywords"]) # Add default keywords if none found if not keywords: keywords = ["家族", "老後", "貯蓄"] try: fetcher = MediaFetcher() return fetcher.fetch_broll(keywords, max_per_keyword=2) except Exception as e: print(f" ⚠️ B-Roll fetching failed: {e}") return {}