NCAkit / modules /video_creator /services /libraries /pixabay_client.py
ismdrobiul489's picture
Add Pixabay video API: dual search (Pexels + Pixabay), optional both APIs, 9:16 portrait filtering
cce7cd9
"""
Pixabay Video Client
Fetches background videos from Pixabay API
"""
import logging
import random
import requests
from typing import List, Optional
logger = logging.getLogger(__name__)
class PixabayClient:
"""Client for Pixabay API to fetch background videos"""
def __init__(self, api_key: str):
"""
Initialize Pixabay client
Args:
api_key: Pixabay API key
"""
self.api_key = api_key
self.base_url = "https://pixabay.com/api/videos/"
self.joker_terms = ["nature", "abstract", "space", "ocean"]
def find_video(
self,
search_terms: List[str],
duration: float,
exclude_ids: Optional[List[int]] = None,
orientation: str = "portrait"
) -> Optional[dict]:
"""
Find a suitable video from Pixabay
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, or None
"""
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"Pixabay: 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
return None
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 Pixabay for: {query} ({orientation})")
response = requests.get(
self.base_url,
params={
"key": self.api_key,
"q": query,
"per_page": 30,
"safesearch": "true",
"video_type": "film"
},
timeout=15
)
if response.status_code != 200:
logger.warning(f"Pixabay API error: {response.status_code}")
return None
data = response.json()
videos = data.get("hits", [])
if not videos:
logger.debug(f"Pixabay: No videos found for query: {query}")
return None
# Filter suitable videos
suitable_videos = []
for video in videos:
if video["id"] in exclude_ids:
continue
video_files = video.get("videos", {})
if not video_files:
continue
# Get large or medium quality
large = video_files.get("large", {})
medium = video_files.get("medium", {})
# Prefer large, fallback to medium
selected_file = large if large.get("url") else medium
if not selected_file.get("url"):
continue
width = selected_file.get("width", 0)
height = selected_file.get("height", 0)
# Filter by orientation
if orientation == "portrait":
# Portrait: height > width (9:16)
if height <= width:
continue
else:
# Landscape: width > height (16:9)
if width <= height:
continue
video_duration = video.get("duration", 0)
suitable_videos.append({
"id": video["id"],
"url": selected_file["url"],
"duration": video_duration,
"width": width,
"height": height
})
if not suitable_videos:
logger.debug(f"Pixabay: No {orientation} videos found for: {query}")
return None
# Filter by duration if possible
duration_threshold = min(min_duration * 0.5, 15)
long_enough_videos = [v for v in suitable_videos if v["duration"] >= duration_threshold]
if long_enough_videos:
# Select FIRST (most relevant) video
selected = long_enough_videos[0]
logger.info(f"Pixabay: Selected video ID {selected['id']} (duration: {selected['duration']}s) for '{query}'")
return selected
# Fallback to first suitable video
selected = suitable_videos[0]
logger.info(f"Pixabay: Selected video ID {selected['id']} ({selected['duration']}s) for '{query}' (fallback)")
return selected
except Exception as e:
logger.error(f"Pixabay search error: {e}")
return None