import os import requests import subprocess def download_and_trim_video(url: str, download_path: str, start_time: float, end_time: float) -> str: """ Downloads a video from the given CDN URL and saves it to download_path. Args: url (str): The URL of the video. download_path (str): Full file path (including filename) to save the video. start_time (float): Start time in seconds for trimming. end_time (float): End time in seconds for trimming. Returns: str: The file path of the downloaded and trimmed video. Raises: requests.RequestException: If the download fails. IOError: If the file cannot be written. subprocess.CalledProcessError: If ffmpeg trimming fails. """ print(f"Downloading video from {url} to {download_path}") try: # Ensure directory exists os.makedirs(os.path.dirname(download_path), exist_ok=True) # Stream the download to avoid loading large files into memory response = requests.get(url, stream=True, timeout=60) # Longer timeout for videos response.raise_for_status() # Check for HTTP errors with open(download_path, "wb") as f: for chunk in response.iter_content(chunk_size=8192): if chunk: # filter out keep-alive new chunks f.write(chunk) print("Video downloaded successfully.", download_path) # Trim the video using ffmpeg trimmed_path = download_path.replace(".mp4", "_trimmed.mp4") duration = end_time - start_time # Build ffmpeg command command = [ "ffmpeg", "-y", # Overwrite output file if it exists "-i", download_path, # Input file "-ss", str(start_time), # Start time "-t", str(duration), # Duration "-c:v", "libx264", # Video codec "-c:a", "aac", # Audio codec "-preset", "medium", # Encoding preset "-crf", "23", # Constant Rate Factor (quality) trimmed_path, # Output file ] # Execute ffmpeg command subprocess.run(command, check=True, capture_output=True) # Remove original downloaded file os.remove(download_path) return trimmed_path except requests.RequestException as e: print(f"ERROR: Failed to download video from {url}: {e}") raise # Re-raise except IOError as e: print(f"ERROR: Failed to write video to {download_path}: {e}") raise # Re-raise except subprocess.CalledProcessError as e: print(f"ERROR: Failed to trim video: {e}") print(f"STDERR: {e.stderr.decode()}") raise # Re-raise