| from __future__ import annotations |
|
|
| import base64 |
| import hashlib |
| import math |
| import os |
| import shutil |
| import subprocess |
| from collections import OrderedDict |
| from datetime import datetime |
|
|
| import ffmpeg |
| from PIL import Image |
|
|
| from shared.ffmpeg_setup import download_ffmpeg |
| from shared.utils.audio_video import get_mp4_audio_codec_settings |
| from shared.utils.video_codecs import get_video_container_extension, get_video_encode_args |
| from shared.utils.utils import get_video_frame, get_video_info |
| VIDEO_EXTENSIONS = {".mp4", ".mkv", ".mov", ".webm", ".avi"} |
| _THUMB_DATA_URL_CACHE: OrderedDict[str, str] = OrderedDict() |
| _THUMB_CACHE_MAX = 64 |
|
|
|
|
| def _get_ffmpeg_path() -> str: |
| download_ffmpeg() |
| ffmpeg_path = os.environ.get("FFMPEG_BINARY", "") or shutil.which("ffmpeg") |
| if not ffmpeg_path: |
| raise RuntimeError("ffmpeg is not available.") |
| return ffmpeg_path |
|
|
|
|
| def _run_ffmpeg(args: list[str]) -> None: |
| result = subprocess.run([_get_ffmpeg_path(), "-y", "-v", "error", *args], capture_output=True, text=True) |
| if result.returncode != 0: |
| raise RuntimeError((result.stderr or result.stdout or "ffmpeg command failed").strip()) |
|
|
|
|
| def _format_timestamp(seconds: float | int | None) -> str | None: |
| if seconds is None: |
| return None |
| seconds = float(seconds) |
| if seconds < 0: |
| raise ValueError("Time values must be >= 0.") |
| return f"{seconds:.6f}".rstrip("0").rstrip(".") |
|
|
|
|
| def _resolve_segment_args(start_time: float | int | None = None, end_time: float | int | None = None, duration: float | int | None = None) -> tuple[str | None, str | None]: |
| if end_time is not None and duration is not None: |
| raise ValueError("Specify either end_time or duration, not both.") |
| start_str = _format_timestamp(0 if start_time is None else start_time) |
| end_str = _format_timestamp(end_time) |
| duration_str = _format_timestamp(duration) |
| if end_str is not None and float(end_str) <= float(start_str): |
| raise ValueError("end_time must be greater than start_time.") |
| if duration_str is not None and float(duration_str) <= 0: |
| raise ValueError("duration must be > 0.") |
| return start_str, end_str if duration_str is None else duration_str |
|
|
|
|
| def get_audio_standalone_extension(codec_key: str | None) -> str: |
| codec_key = str(codec_key or "wav").strip().lower() or "wav" |
| if codec_key == "mp3": |
| codec_key = "mp3_192" |
| return ".wav" if codec_key == "wav" else ".mp3" |
|
|
|
|
| def _get_mp4_audio_encode_args(codec_key: str | None) -> list[str]: |
| settings = get_mp4_audio_codec_settings(codec_key) |
| args = ["-c:a", settings["codec"]] |
| if settings.get("bitrate"): |
| args += ["-b:a", settings["bitrate"]] |
| return args |
|
|
|
|
| def _get_standalone_audio_encode_args(codec_key: str | None) -> list[str]: |
| codec_key = str(codec_key or "wav").strip().lower() or "wav" |
| if codec_key == "mp3": |
| codec_key = "mp3_192" |
| if codec_key == "wav": |
| return ["-c:a", "pcm_s16le"] |
| bitrate = {"mp3_128": "128k", "mp3_192": "192k", "mp3_320": "320k"}.get(codec_key, "192k") |
| return ["-c:a", "libmp3lame", "-b:a", bitrate] |
|
|
|
|
| def has_video_extension(path: str) -> bool: |
| return os.path.splitext(str(path or "").strip())[1].lower() in VIDEO_EXTENSIONS |
|
|
|
|
| def get_video_thumbnail_data_url(video_path: str) -> str: |
| video_path = os.path.abspath(os.path.normpath(str(video_path or "").strip())) |
| if len(video_path) == 0 or not os.path.isfile(video_path) or not has_video_extension(video_path): |
| return "" |
| stat = os.stat(video_path) |
| cache_key = hashlib.sha1(f"{os.path.abspath(video_path)}|{stat.st_mtime_ns}|{stat.st_size}".encode("utf-8")).hexdigest() |
| cached = _THUMB_DATA_URL_CACHE.get(cache_key, "") |
| if len(cached) > 0: |
| _THUMB_DATA_URL_CACHE.move_to_end(cache_key) |
| return cached |
| result = subprocess.run( |
| [ |
| _get_ffmpeg_path(), |
| "-v", |
| "error", |
| "-i", |
| video_path, |
| "-frames:v", |
| "1", |
| "-vf", |
| "scale=320:-1:force_original_aspect_ratio=decrease", |
| "-q:v", |
| "4", |
| "-f", |
| "image2pipe", |
| "-vcodec", |
| "mjpeg", |
| "-", |
| ], |
| capture_output=True, |
| ) |
| if result.returncode != 0 or len(result.stdout) == 0: |
| raise RuntimeError((result.stderr.decode("utf-8", errors="ignore").strip() or "ffmpeg thumbnail extraction failed").strip()) |
| data_url = f"data:image/jpeg;base64,{base64.b64encode(result.stdout).decode('ascii')}" |
| _THUMB_DATA_URL_CACHE[cache_key] = data_url |
| while len(_THUMB_DATA_URL_CACHE) > _THUMB_CACHE_MAX: |
| _THUMB_DATA_URL_CACHE.popitem(last=False) |
| return data_url |
|
|
|
|
| def merge_videos(first_video: str, second_video: str, output_path: str | None = None, *, video_codec: str | None = None, video_container: str | None = None, audio_codec: str | None = None) -> str: |
| first_video = os.path.normpath(str(first_video or "").strip()) |
| second_video = os.path.normpath(str(second_video or "").strip()) |
| if not os.path.isfile(first_video): |
| raise FileNotFoundError(f"First video not found: {first_video}") |
| if not os.path.isfile(second_video): |
| raise FileNotFoundError(f"Second video not found: {second_video}") |
| first_info = _probe_video_stream(first_video) |
| width, height = int(first_info["width"]), int(first_info["height"]) |
| first_has_audio = _has_audio_stream(first_video) |
| second_has_audio = _has_audio_stream(second_video) |
| output_path = output_path or _default_merged_output_path(first_video, second_video) |
| os.makedirs(os.path.dirname(output_path) or ".", exist_ok=True) |
| filters = [ |
| f"[0:v]scale={width}:{height}:force_original_aspect_ratio=decrease,pad={width}:{height}:(ow-iw)/2:(oh-ih)/2:color=black,setsar=1[v0]", |
| f"[1:v]scale={width}:{height}:force_original_aspect_ratio=decrease,pad={width}:{height}:(ow-iw)/2:(oh-ih)/2:color=black,setsar=1[v1]", |
| ] |
| maps = ["-map", "[v]"] |
| if first_has_audio and second_has_audio: |
| filters.append("[v0][0:a][v1][1:a]concat=n=2:v=1:a=1[v][a]") |
| maps += ["-map", "[a]", "-c:a", "aac", "-b:a", "192k"] |
| else: |
| filters.append("[v0][v1]concat=n=2:v=1:a=0[v]") |
| cmd = [ |
| "-i", |
| first_video, |
| "-i", |
| second_video, |
| "-filter_complex", |
| ";".join(filters), |
| *maps, |
| *get_video_encode_args(video_codec, video_container), |
| *(["-movflags", "+faststart"] if str(video_container or "mp4").strip().lower() == "mp4" else []), |
| output_path, |
| ] |
| if first_has_audio and second_has_audio: |
| insert_at = cmd.index(output_path) |
| cmd[insert_at:insert_at] = _get_mp4_audio_encode_args(audio_codec) |
| _run_ffmpeg(cmd) |
| return output_path |
|
|
|
|
| def extract_video(source_path: str, output_path: str, start_time: float | int = 0, end_time: float | int | None = None, duration: float | int | None = None, *, video_codec: str | None = None, video_container: str | None = None, audio_codec: str | None = None) -> str: |
| source_path = os.path.normpath(str(source_path or "").strip()) |
| output_path = os.path.normpath(str(output_path or "").strip()) |
| if not os.path.isfile(source_path): |
| raise FileNotFoundError(f"Video not found: {source_path}") |
| start_str, end_or_duration_str = _resolve_segment_args(start_time, end_time, duration) |
| os.makedirs(os.path.dirname(output_path) or ".", exist_ok=True) |
| cmd = ["-ss", start_str, "-i", source_path] |
| if duration is not None: |
| cmd += ["-t", end_or_duration_str] |
| elif end_time is not None: |
| cmd += ["-to", end_or_duration_str] |
| cmd += ["-map", "0:v:0", "-map", "0:a?"] |
| cmd += get_video_encode_args(video_codec, video_container) |
| cmd += _get_mp4_audio_encode_args(audio_codec) |
| if str(video_container or "mp4").strip().lower() == "mp4": |
| cmd += ["-movflags", "+faststart"] |
| cmd += [output_path] |
| _run_ffmpeg(cmd) |
| return output_path |
|
|
|
|
| def extract_audio(source_path: str, output_path: str, start_time: float | int | None = None, end_time: float | int | None = None, duration: float | int | None = None, audio_track_no: int | None = None, *, audio_codec: str | None = None) -> str: |
| source_path = os.path.normpath(str(source_path or "").strip()) |
| output_path = os.path.normpath(str(output_path or "").strip()) |
| if not os.path.isfile(source_path): |
| raise FileNotFoundError(f"Media not found: {source_path}") |
| probe = ffmpeg.probe(source_path) |
| audio_streams = [stream for stream in probe.get("streams", []) if str(stream.get("codec_type", "")).strip().lower() == "audio"] |
| if len(audio_streams) == 0: |
| raise RuntimeError(f"No audio stream found in {source_path}") |
| audio_track_no = 1 if audio_track_no is None else int(audio_track_no) |
| if audio_track_no <= 0: |
| raise ValueError("audio_track_no must be >= 1.") |
| if audio_track_no > len(audio_streams): |
| raise ValueError(f"audio_track_no must be between 1 and {len(audio_streams)}.") |
| start_str, end_or_duration_str = _resolve_segment_args(start_time, end_time, duration) |
| os.makedirs(os.path.dirname(output_path) or ".", exist_ok=True) |
| cmd = [] |
| if start_str is not None: |
| cmd += ["-ss", start_str] |
| cmd += ["-i", source_path] |
| if duration is not None: |
| cmd += ["-t", end_or_duration_str] |
| elif end_time is not None: |
| cmd += ["-to", end_or_duration_str] |
| cmd += ["-map", f"0:a:{audio_track_no - 1}", "-vn", *_get_standalone_audio_encode_args(audio_codec), output_path] |
| _run_ffmpeg(cmd) |
| return output_path |
|
|
|
|
| def extract_video_frame(source_path: str, output_path: str, *, frame_no: int | None = None, time_seconds: float | int | None = None) -> str: |
| source_path = os.path.normpath(str(source_path or "").strip()) |
| output_path = os.path.normpath(str(output_path or "").strip()) |
| if not os.path.isfile(source_path): |
| raise FileNotFoundError(f"Video not found: {source_path}") |
| if time_seconds is None and frame_no is None: |
| raise ValueError("frame_no or time_seconds is required.") |
| os.makedirs(os.path.dirname(output_path) or ".", exist_ok=True) |
| frame_no = resolve_video_frame_no(source_path, frame_no=frame_no, time_seconds=time_seconds) |
| image = get_video_frame(source_path, frame_no, return_last_if_missing=True, return_PIL=True).convert("RGB") |
| image.save(output_path) |
| if not os.path.isfile(output_path): |
| raise RuntimeError("Frame extraction did not produce an output image.") |
| return output_path |
|
|
|
|
| def _resolve_crop_anchor(crop_anchor: str | None) -> tuple[str, str]: |
| normalized = str(crop_anchor or "center").strip().lower().replace("-", "_").replace(" ", "_") or "center" |
| anchor_aliases = { |
| "centre": "center", |
| "middle": "center", |
| "center_left": "left", |
| "left_center": "left", |
| "center_right": "right", |
| "right_center": "right", |
| "top_center": "top", |
| "center_top": "top", |
| "bottom_center": "bottom", |
| "center_bottom": "bottom", |
| } |
| normalized = anchor_aliases.get(normalized, normalized) |
| if normalized in {"center", "left", "right", "top", "bottom"}: |
| return { |
| "center": ("center", "center"), |
| "left": ("left", "center"), |
| "right": ("right", "center"), |
| "top": ("center", "top"), |
| "bottom": ("center", "bottom"), |
| }[normalized] |
| corner_anchors = { |
| "top_left": ("left", "top"), |
| "top_right": ("right", "top"), |
| "bottom_left": ("left", "bottom"), |
| "bottom_right": ("right", "bottom"), |
| } |
| if normalized in corner_anchors: |
| return corner_anchors[normalized] |
| raise ValueError("crop_anchor must be one of center, left, right, top, bottom, top_left, top_right, bottom_left, or bottom_right.") |
|
|
|
|
| def _resolve_resize_crop_geometry( |
| source_width: int, |
| source_height: int, |
| *, |
| crop_left: float = 0, |
| crop_top: float = 0, |
| crop_right: float = 0, |
| crop_bottom: float = 0, |
| crop_unit: str = "pixels", |
| target_width: int | None = None, |
| target_height: int | None = None, |
| preserve_aspect_ratio: bool = True, |
| crop_anchor: str = "center", |
| ) -> tuple[int, int, int, int]: |
| crop_unit = str(crop_unit or "pixels").strip().lower() or "pixels" |
| if crop_unit not in {"pixels", "percent"}: |
| raise ValueError("crop_unit must be 'pixels' or 'percent'.") |
| align_x, align_y = _resolve_crop_anchor(crop_anchor) |
|
|
| def resolve_crop(value: float, total: int) -> int: |
| value = float(value or 0) |
| if value < 0: |
| raise ValueError("Crop values must be >= 0.") |
| return round(total * value / 100.0) if crop_unit == "percent" else round(value) |
|
|
| left = resolve_crop(crop_left, source_width) |
| right = resolve_crop(crop_right, source_width) |
| top = resolve_crop(crop_top, source_height) |
| bottom = resolve_crop(crop_bottom, source_height) |
| cropped_width = source_width - left - right |
| cropped_height = source_height - top - bottom |
| if cropped_width <= 0 or cropped_height <= 0: |
| raise ValueError("Crop values remove the whole frame.") |
| if not preserve_aspect_ratio or target_width is None or target_height is None: |
| return left, top, cropped_width, cropped_height |
| if target_width <= 0 or target_height <= 0: |
| raise ValueError("width and height must be > 0 when provided.") |
| target_ratio = float(target_width) / float(target_height) |
| current_ratio = float(cropped_width) / float(cropped_height) |
| if math.isclose(current_ratio, target_ratio, rel_tol=0.0, abs_tol=1e-6): |
| return left, top, cropped_width, cropped_height |
| if current_ratio > target_ratio: |
| adjusted_width = min(cropped_width, max(1, int(round(cropped_height * target_ratio)))) |
| trim = cropped_width - adjusted_width |
| left += 0 if align_x == "left" else trim if align_x == "right" else trim // 2 |
| cropped_width = adjusted_width |
| else: |
| adjusted_height = min(cropped_height, max(1, int(round(cropped_width / target_ratio)))) |
| trim = cropped_height - adjusted_height |
| top += 0 if align_y == "top" else trim if align_y == "bottom" else trim // 2 |
| cropped_height = adjusted_height |
| if cropped_width <= 0 or cropped_height <= 0: |
| raise ValueError("Unable to preserve aspect ratio with the requested crop.") |
| return left, top, cropped_width, cropped_height |
|
|
|
|
| def mute_video(source_path: str, output_path: str) -> str: |
| source_path = os.path.normpath(str(source_path or "").strip()) |
| output_path = os.path.normpath(str(output_path or "").strip()) |
| if not os.path.isfile(source_path): |
| raise FileNotFoundError(f"Video not found: {source_path}") |
| os.makedirs(os.path.dirname(output_path) or ".", exist_ok=True) |
| _run_ffmpeg(["-i", source_path, "-map", "0:v:0", "-c:v", "copy", "-an", output_path]) |
| return output_path |
|
|
|
|
| def replace_audio(video_path: str, audio_path: str, output_path: str, *, audio_codec: str | None = None) -> str: |
| video_path = os.path.normpath(str(video_path or "").strip()) |
| audio_path = os.path.normpath(str(audio_path or "").strip()) |
| output_path = os.path.normpath(str(output_path or "").strip()) |
| if not os.path.isfile(video_path): |
| raise FileNotFoundError(f"Video not found: {video_path}") |
| if not os.path.isfile(audio_path): |
| raise FileNotFoundError(f"Audio not found: {audio_path}") |
| if not _has_audio_stream(audio_path): |
| raise RuntimeError(f"No audio stream found in {audio_path}") |
| os.makedirs(os.path.dirname(output_path) or ".", exist_ok=True) |
| _run_ffmpeg(["-i", video_path, "-i", audio_path, "-map", "0:v:0", "-map", "1:a:0", "-c:v", "copy", *_get_mp4_audio_encode_args(audio_codec), "-shortest", output_path]) |
| return output_path |
|
|
|
|
| def resize_crop_video(source_path: str, output_path: str, *, width: int | None = None, height: int | None = None, crop_left: float = 0, crop_top: float = 0, crop_right: float = 0, crop_bottom: float = 0, crop_unit: str = "pixels", preserve_aspect_ratio: bool = True, crop_anchor: str = "center", video_codec: str | None = None, video_container: str | None = None, audio_codec: str | None = None) -> str: |
| source_path = os.path.normpath(str(source_path or "").strip()) |
| output_path = os.path.normpath(str(output_path or "").strip()) |
| if not os.path.isfile(source_path): |
| raise FileNotFoundError(f"Video not found: {source_path}") |
| stream = _probe_video_stream(source_path) |
| source_width = int(stream["width"]) |
| source_height = int(stream["height"]) |
| left, top, cropped_width, cropped_height = _resolve_resize_crop_geometry( |
| source_width, |
| source_height, |
| crop_left=crop_left, |
| crop_top=crop_top, |
| crop_right=crop_right, |
| crop_bottom=crop_bottom, |
| crop_unit=crop_unit, |
| target_width=width, |
| target_height=height, |
| preserve_aspect_ratio=preserve_aspect_ratio, |
| crop_anchor=crop_anchor, |
| ) |
| filters = [f"crop={cropped_width}:{cropped_height}:{left}:{top}"] |
| if width is not None or height is not None: |
| scale_w = int(width) if width is not None else -1 |
| scale_h = int(height) if height is not None else -1 |
| if width is not None and int(width) <= 0 or height is not None and int(height) <= 0: |
| raise ValueError("width and height must be > 0 when provided.") |
| filters.append(f"scale={scale_w}:{scale_h}") |
| os.makedirs(os.path.dirname(output_path) or ".", exist_ok=True) |
| cmd = ["-i", source_path, "-vf", ",".join(filters), "-map", "0:v:0", "-map", "0:a?"] |
| cmd += get_video_encode_args(video_codec, video_container) |
| cmd += _get_mp4_audio_encode_args(audio_codec) |
| if str(video_container or "mp4").strip().lower() == "mp4": |
| cmd += ["-movflags", "+faststart"] |
| cmd += [output_path] |
| _run_ffmpeg(cmd) |
| return output_path |
|
|
|
|
| def resize_crop_image(source_path: str, output_path: str, *, width: int | None = None, height: int | None = None, crop_left: float = 0, crop_top: float = 0, crop_right: float = 0, crop_bottom: float = 0, crop_unit: str = "pixels", preserve_aspect_ratio: bool = True, crop_anchor: str = "center") -> str: |
| source_path = os.path.normpath(str(source_path or "").strip()) |
| output_path = os.path.normpath(str(output_path or "").strip()) |
| if not os.path.isfile(source_path): |
| raise FileNotFoundError(f"Image not found: {source_path}") |
| with Image.open(source_path) as image: |
| source_width, source_height = image.size |
| left, top, cropped_width, cropped_height = _resolve_resize_crop_geometry( |
| source_width, |
| source_height, |
| crop_left=crop_left, |
| crop_top=crop_top, |
| crop_right=crop_right, |
| crop_bottom=crop_bottom, |
| crop_unit=crop_unit, |
| target_width=width, |
| target_height=height, |
| preserve_aspect_ratio=preserve_aspect_ratio, |
| crop_anchor=crop_anchor, |
| ) |
| output_image = image.crop((left, top, left + cropped_width, top + cropped_height)) |
| if width is not None or height is not None: |
| if width is not None and int(width) <= 0 or height is not None and int(height) <= 0: |
| raise ValueError("width and height must be > 0 when provided.") |
| target_width = int(width) if width is not None else max(1, round(cropped_width * (int(height) / cropped_height))) |
| target_height = int(height) if height is not None else max(1, round(cropped_height * (int(width) / cropped_width))) |
| output_image = output_image.resize((target_width, target_height), Image.Resampling.LANCZOS) |
| os.makedirs(os.path.dirname(output_path) or ".", exist_ok=True) |
| output_ext = os.path.splitext(output_path)[1].lower() |
| save_kwargs = {"quality": 95} if output_ext in {".jpg", ".jpeg", ".webp"} else {} |
| if output_ext in {".jpg", ".jpeg"} and output_image.mode not in {"RGB", "L"}: |
| output_image = output_image.convert("RGB") |
| output_image.save(output_path, **save_kwargs) |
| return output_path |
|
|
|
|
| def get_media_duration(path: str) -> float | None: |
| probe = ffmpeg.probe(path) |
| format_info = probe.get("format", {}) or {} |
| duration = format_info.get("duration", None) |
| try: |
| return None if duration is None else float(duration) |
| except Exception: |
| return None |
|
|
|
|
| def get_precise_video_fps(path: str) -> float | None: |
| try: |
| stream = _probe_video_stream(path) |
| except Exception: |
| return None |
| for key in ("avg_frame_rate", "r_frame_rate"): |
| value = str(stream.get(key, "") or "").strip() |
| if len(value) == 0 or value in {"0/0", "N/A"}: |
| continue |
| if "/" in value: |
| num, den = value.split("/", 1) |
| try: |
| num = float(num) |
| den = float(den) |
| if den != 0: |
| return num / den |
| except Exception: |
| continue |
| else: |
| try: |
| return float(value) |
| except Exception: |
| continue |
| return None |
|
|
|
|
| def resolve_video_frame_no(path: str, *, frame_no: int | None = None, time_seconds: float | int | None = None) -> int: |
| if time_seconds is None and frame_no is None: |
| raise ValueError("frame_no or time_seconds is required.") |
| fps, _width, _height, frame_count = get_video_info(path) |
| precise_fps = get_precise_video_fps(path) |
| effective_fps = float(precise_fps) if precise_fps is not None and precise_fps > 0 else float(fps) |
| max_frame = max(0, int(frame_count) - 1) |
| if frame_no is None: |
| if effective_fps <= 0: |
| frame_no = 0 |
| else: |
| frame_no = int(math.floor(max(0.0, float(time_seconds or 0.0)) * effective_fps + 1e-9)) |
| frame_no = int(frame_no) |
| if frame_no < 0: |
| frame_no = 0 |
| if frame_no > max_frame: |
| frame_no = max_frame |
| return frame_no |
|
|
|
|
| def has_audio_stream(path: str) -> bool: |
| return _has_audio_stream(path) |
|
|
|
|
| def has_video_stream(path: str) -> bool: |
| try: |
| _probe_video_stream(path) |
| return True |
| except Exception: |
| return False |
|
|
|
|
| def _probe_video_stream(video_path: str) -> dict: |
| probe = ffmpeg.probe(video_path) |
| for stream in probe.get("streams", []): |
| if stream.get("codec_type") == "video": |
| return stream |
| raise RuntimeError(f"No video stream found in {video_path}") |
|
|
|
|
| def _has_audio_stream(video_path: str) -> bool: |
| probe = ffmpeg.probe(video_path) |
| return any(stream.get("codec_type") == "audio" for stream in probe.get("streams", [])) |
|
|
|
|
| def _default_merged_output_path(first_video: str, second_video: str) -> str: |
| first_base = os.path.splitext(os.path.basename(first_video))[0] |
| second_base = os.path.splitext(os.path.basename(second_video))[0] |
| output_dir = os.path.dirname(first_video) or "outputs" |
| timestamp = datetime.now().strftime("%Y-%m-%d-%Hh%Mm%Ss") |
| return os.path.join(output_dir, f"{timestamp}_merged_{first_base}_{second_base}.mp4") |
|
|
