149 lines
5.2 KiB
Python
149 lines
5.2 KiB
Python
"""
|
|
video_bg.py — Pexels API → скачать видео-клип для фона
|
|
|
|
Если Pexels не доступен → генерируем чёрный видео-файл.
|
|
"""
|
|
|
|
import os
|
|
import subprocess
|
|
import requests
|
|
from pathlib import Path
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv(os.path.expanduser("~/.openclaw/.env"))
|
|
|
|
FFMPEG = os.environ.get("FFMPEG_BIN", os.path.expanduser("~/bin/ffmpeg-7.0.2-amd64-static/ffmpeg"))
|
|
FFPROBE = os.environ.get("FFPROBE_BIN", os.path.expanduser("~/bin/ffmpeg-7.0.2-amd64-static/ffprobe"))
|
|
|
|
|
|
def search_pexels(query: str, per_page: int = 3) -> list[dict]:
|
|
"""Поиск видео на Pexels. Возвращает список видео-объектов."""
|
|
api_key = os.environ.get("PEXELS_API_KEY", "")
|
|
if not api_key:
|
|
print("[video_bg] PEXELS_API_KEY не задан, пропускаем поиск.")
|
|
return []
|
|
|
|
url = "https://api.pexels.com/videos/search"
|
|
params = {
|
|
"query": query,
|
|
"per_page": per_page,
|
|
"orientation": "landscape",
|
|
"size": "medium",
|
|
}
|
|
headers = {"Authorization": api_key}
|
|
|
|
try:
|
|
resp = requests.get(url, params=params, headers=headers, timeout=15)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
except Exception as e:
|
|
print(f"[video_bg] Ошибка Pexels API: {e}")
|
|
return []
|
|
|
|
return data.get("videos", [])
|
|
|
|
|
|
def pick_video(video_list: list[dict]) -> str | None:
|
|
"""Выбрать URL лучшего видео (HD 1280x720)."""
|
|
for v in video_list:
|
|
files = v.get("video_files", [])
|
|
for f in files:
|
|
# Предпочитаем 720p landscape
|
|
width = f.get("width", 0)
|
|
if width == 1280:
|
|
return f.get("file")
|
|
# Если нет 720p — берём первый ≥1280
|
|
for f in sorted(files, key=lambda x: x.get("width", 0), reverse=True):
|
|
if f.get("width", 0) >= 1280:
|
|
return f.get("file")
|
|
# Если вообще ничего ≥1280 — первый попавшийся
|
|
if video_list and video_list[0].get("video_files"):
|
|
return video_list[0]["video_files"][0].get("file")
|
|
return None
|
|
|
|
|
|
def download_url(url: str, dest: str) -> str:
|
|
"""Скачать файл по URL в dest."""
|
|
print(f"[video_bg] Скачиваем {url[:80]}…")
|
|
resp = requests.get(url, stream=True, timeout=60)
|
|
resp.raise_for_status()
|
|
with open(dest, "wb") as f:
|
|
for chunk in resp.iter_content(chunk_size=8192):
|
|
f.write(chunk)
|
|
return dest
|
|
|
|
|
|
def get_duration(path: str) -> float:
|
|
"""Длительность видео в секундах через ffprobe."""
|
|
cmd = [
|
|
FFPROBE, "-v", "quiet", "-show_entries", "format=duration",
|
|
"-of", "csv=p=0", path
|
|
]
|
|
out = subprocess.check_output(cmd, text=True).strip()
|
|
return float(out)
|
|
|
|
|
|
def create_black_video(duration: float, dest: str, width: int = 1280, height: int = 720):
|
|
"""Создать чёрное видео заданной длительности."""
|
|
print(f"[video_bg] Создаём чёрный фон {duration:.1f}s…")
|
|
cmd = [
|
|
FFMPEG, "-f", "lavfi", "-i", f"color=c=black:s={width}x{height}:r=30:d={duration}",
|
|
"-c:v", "libx264", "-pix_fmt", "yuv420p", "-y", dest
|
|
]
|
|
subprocess.run(cmd, check=True, capture_output=True)
|
|
return dest
|
|
|
|
|
|
def loop_video(source: str, target_duration: float, dest: str):
|
|
"""Зациклить видео до нужной длительности."""
|
|
src_dur = get_duration(source)
|
|
if src_dur >= target_duration:
|
|
# Обрезать лишнее
|
|
cmd = [
|
|
FFMPEG, "-i", source, "-t", str(target_duration),
|
|
"-c:v", "libx264", "-pix_fmt", "yuv420p", "-an", "-y", dest
|
|
]
|
|
else:
|
|
# Зациклить через stream_loop
|
|
loops = int(target_duration / src_dur) + 1
|
|
cmd = [
|
|
FFMPEG, "-stream_loop", str(loops - 1), "-i", source,
|
|
"-t", str(target_duration),
|
|
"-c:v", "libx264", "-pix_fmt", "yuv420p", "-an", "-y", dest
|
|
]
|
|
subprocess.run(cmd, check=True, capture_output=True)
|
|
return dest
|
|
|
|
|
|
def get_bg_video(search_query: str, audio_duration: float,
|
|
output_dir: str) -> str:
|
|
"""
|
|
Основной entry point.
|
|
|
|
1. Ищет видео на Pexels по query
|
|
2. Скачивает и зацикливает до audio_duration
|
|
3. Если Pexels недоступен — создаёт чёрный фон
|
|
|
|
Возвращает путь к видео-файлу.
|
|
"""
|
|
os.makedirs(output_dir, exist_ok=True)
|
|
|
|
raw_path = os.path.join(output_dir, "raw_bg.mp4")
|
|
bg_path = os.path.join(output_dir, "background.mp4")
|
|
|
|
videos = search_pexels(search_query)
|
|
url = pick_video(videos)
|
|
|
|
if url:
|
|
try:
|
|
download_url(url, raw_path)
|
|
loop_video(raw_path, audio_duration, bg_path)
|
|
print(f"[video_bg] Видео-фон готов: {bg_path}")
|
|
return bg_path
|
|
except Exception as e:
|
|
print(f"[video_bg] Ошибка скачивания: {e}")
|
|
|
|
# Fallback: чёрный фон
|
|
create_black_video(audio_duration, bg_path)
|
|
return bg_path
|