Spaces:
Running
Running
File size: 4,493 Bytes
c109c49 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 | import subprocess
import os
import time
import shutil
import logging
import httpx
import subtitles
from security_utils import is_safe_url
logger = logging.getLogger(__name__)
# Configurable limits
FFMPEG_TIMEOUT = int(os.getenv("FFMPEG_TIMEOUT_SECONDS", "120"))
MAX_IMAGE_SIZE = 10 * 1024 * 1024 # Max 10MB image
def download_image(url: str, dest_path: str):
"""
Downloads an image safely with size limits and timeouts using HTTPX.
"""
if not is_safe_url(url):
raise ValueError(f"Forbidden URL target: {url}")
with httpx.Client(timeout=10.0, follow_redirects=True) as client:
# Stream response to check size before reading full content into memory
with client.stream("GET", url) as response:
if response.status_code != 200:
raise IOError(f"Failed to fetch image, status: {response.status_code}")
content_length = response.headers.get("Content-Length")
if content_length and int(content_length) > MAX_IMAGE_SIZE:
raise ValueError("Remote image size exceeds limit.")
total_bytes = 0
with open(dest_path, "wb") as out:
for chunk in response.iter_bytes(chunk_size=8192):
total_bytes += len(chunk)
if total_bytes > MAX_IMAGE_SIZE:
raise ValueError("Remote image size exceeds limit during streaming.")
out.write(chunk)
def render_video(audio_path, image_url, script_data, output_path, preset="horizontal"):
config = {
"vertical": (1080, 1920, 1.25),
"square": (1080, 1080, 1.15),
"horizontal": (1920, 1080, 1.0)
}
width, height, fg_scale = config.get(preset, config["horizontal"])
# Generate unique random tokens for temp files to prevent collision
rand_token = os.urandom(8).hex()
img_path = f"temp_bg_{rand_token}.jpg"
subs_path = f"temp_subs_{rand_token}.ass"
try:
# 1. Download visual assets safely
download_image(image_url, img_path)
# 2. Compile subtitles safely
subtitles.create_ass_subtitles(script_data, subs_path, width, height)
# 3. Escape subs path for FFmpeg filter argument
# FFmpeg's ass filter requires escaping backslashes and single quotes.
escaped_subs_path = subs_path.replace("\\", "\\\\").replace("'", "\\'").replace(":", "\\:")
bg_f = f"scale=400:-1,boxblur=10:5,colorchannelmixer=rr=0.5:gg=0.5:bb=0.5,zoompan=z='min(zoom+0.0005,1.2)':d=1:x='iw/2-(iw/zoom/2)':y='ih/2-(ih/zoom/2)':s={width}x{height}"
fg_f = f"scale={int(width*fg_scale)}:-1" if preset != "horizontal" else f"scale=-1:{height}"
ov_f = f"[0:v]{bg_f}[bg];[1:v]{fg_f}[fg];[bg][fg]overlay=(W-w)/2:(H-h)/2,ass='{escaped_subs_path}'"
cmd = [
"ffmpeg", "-y", "-loop", "1", "-i", img_path, "-loop", "1", "-i", img_path, "-i", audio_path,
"-filter_complex", ov_f, "-c:v", "libx264", "-preset", "ultrafast", "-crf", "20",
"-c:a", "aac", "-b:a", "192k", "-pix_fmt", "yuv420p", "-shortest", output_path
]
logger.info(f"Executing FFmpeg with timeout={FFMPEG_TIMEOUT}s...")
# Execute subprocess with timeout to prevent infinite loops / orphaned processes
result = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=FFMPEG_TIMEOUT, check=True)
logger.info("FFmpeg execution completed successfully.")
return True
except subprocess.TimeoutExpired as te:
logger.error("FFmpeg compilation timed out.")
raise RuntimeError("Video rendering timed out.") from te
except subprocess.CalledProcessError as cpe:
stderr_output = cpe.stderr.decode("utf-8", errors="replace") if cpe.stderr else ""
logger.error(f"FFmpeg failed with error code {cpe.returncode}. Stderr: {stderr_output}")
raise RuntimeError("Video rendering engine failed.") from cpe
except Exception as e:
logger.error(f"Render engine error: {e}", exc_info=True)
raise
finally:
# Guaranteed cleanup of intermediate temporary files
for p in [img_path, subs_path]:
if os.path.exists(p):
try:
os.remove(p)
except Exception as cleanup_err:
logger.warning(f"Could not remove temp file {p}: {cleanup_err}")
|