import requests import logging from typing import List, Optional from pathlib import Path import random logger = logging.getLogger(__name__) class PexelsClient: """Client for Pexels API to fetch background videos""" def __init__(self, api_key: str): """ Initialize Pexels client Args: api_key: Pexels API key """ self.api_key = api_key self.base_url = "https://api.pexels.com/videos" self.headers = {"Authorization": api_key} self.joker_terms = ["nature", "globe", "space", "ocean"] def find_video( self, search_terms: List[str], duration: float, exclude_ids: Optional[List[int]] = None, orientation: str = "portrait" ) -> dict: """ Find a suitable video from Pexels Args: search_terms: Keywords to search for duration: Required video duration in seconds exclude_ids: List of video IDs to exclude orientation: 'portrait' or 'landscape' Returns: Dict with 'id' and 'url' of the selected video """ exclude_ids = exclude_ids or [] # Try user-provided search terms first for term in search_terms: video = self._search_and_select(term, duration, exclude_ids, orientation) if video: return video # Fall back to joker terms logger.info(f"No videos found for {search_terms}, using joker terms") for term in self.joker_terms: video = self._search_and_select(term, duration, exclude_ids, orientation) if video: return video raise Exception("No suitable videos found on Pexels") def _search_and_select( self, query: str, min_duration: float, exclude_ids: List[int], orientation: str ) -> Optional[dict]: """Search for videos and select a suitable one""" try: logger.debug(f"Searching Pexels for: {query} ({orientation})") response = requests.get( f"{self.base_url}/search", headers=self.headers, params={ "query": query, "orientation": orientation, "per_page": 15, "size": "medium" # Good balance of quality and file size }, timeout=10 ) if response.status_code != 200: logger.warning(f"Pexels API error: {response.status_code}") return None data = response.json() videos = data.get("videos", []) if not videos: logger.debug(f"No videos found for query: {query}") return None # Filter suitable videos suitable_videos = [] for video in videos: if video["id"] in exclude_ids: continue # Get video file URL (HD or SD) video_files = video.get("video_files", []) if not video_files: continue # Sort by quality and find a good match video_files = sorted( video_files, key=lambda x: x.get("width", 0) * x.get("height", 0), reverse=True ) # Find appropriate quality based on orientation target_width = 1080 if orientation == "portrait" else 1920 target_height = 1920 if orientation == "portrait" else 1080 selected_file = None for vf in video_files: # Look for files close to our target resolution if vf.get("width") and vf.get("height"): if (abs(vf["width"] - target_width) < 300 and abs(vf["height"] - target_height) < 300): selected_file = vf break # Fallback to highest quality if no exact match if not selected_file and video_files: selected_file = video_files[0] if selected_file and selected_file.get("link"): suitable_videos.append({ "id": video["id"], "url": selected_file["link"], "duration": video.get("duration", 0) }) if not suitable_videos: return None # Filter by duration if possible # Try to find videos that are at least 50% of the requested duration # to avoid stitching too many tiny clips duration_threshold = min(min_duration * 0.5, 15) # Cap at 15s requirement long_enough_videos = [v for v in suitable_videos if v["duration"] >= duration_threshold] if long_enough_videos: # Select FIRST (most relevant) video instead of random selected = long_enough_videos[0] logger.info(f"Selected Pexels video ID {selected['id']} (duration: {selected['duration']}s) for query '{query}'") return selected # Fallback to first suitable video selected = suitable_videos[0] logger.info(f"Selected Pexels video ID {selected['id']} (duration: {selected['duration']}s) for query '{query}' (fallback)") return selected except Exception as e: logger.error(f"Error searching Pexels: {e}") return None def find_photo( self, query: str, orientation: str = "portrait" ) -> Optional[dict]: """ Find a suitable photo from Pexels Args: query: Search term orientation: 'portrait' or 'landscape' Returns: Dict with 'id' and 'url' of the photo """ try: logger.debug(f"Searching Pexels for photo: {query} ({orientation})") # Pexels Photo API endpoint url = "https://api.pexels.com/v1/search" response = requests.get( url, headers=self.headers, params={ "query": query, "orientation": orientation, "per_page": 15, "size": "large" }, timeout=10 ) if response.status_code != 200: logger.warning(f"Pexels Photo API error: {response.status_code}") return None data = response.json() photos = data.get("photos", []) if not photos: logger.debug(f"No photos found for query: {query}") return None # Select FIRST (most relevant) photo instead of random photo = photos[0] # Get URL (prefer original or large2x) src = photo.get("src", {}) url = src.get("original") or src.get("large2x") or src.get("large") if not url: return None logger.info(f"Selected Pexels photo ID {photo['id']} for query '{query}'") return { "id": photo["id"], "url": url, "type": "photo" } except Exception as e: logger.error(f"Error searching Pexels photos: {e}") return None