nca-toolkit / video_creator /libraries /pexels_client.py
ismdrobiul489's picture
Fix blank screens: 1:1 scene logic, 9:16 enforcement, image fallback, and ffmpeg syntax fix
3eab663
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:
selected = random.choice(long_enough_videos)
logger.info(f"Selected Pexels video ID {selected['id']} (duration: {selected['duration']}s) for query '{query}'")
return selected
# Fallback to any suitable video
selected = random.choice(suitable_videos)
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 a random photo
photo = random.choice(photos)
# 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