Flatten Bolt-Compress to repo root
This commit is contained in:
majestic
@@ -0,0 +1,342 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Optional, Set
|
||||
|
||||
from PIL import Image
|
||||
|
||||
from .models import IMAGE_EXTENSIONS, VIDEO_EXTENSIONS, MediaInfo, MediaKind
|
||||
|
||||
# Keep Pillow from warning on very large, legitimate photos.
|
||||
Image.MAX_IMAGE_PIXELS = 300_000_000
|
||||
|
||||
|
||||
class MediaError(Exception):
|
||||
"""A media file could not be read or processed."""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class FFmpegTools:
|
||||
ffmpeg: Optional[Path]
|
||||
ffprobe: Optional[Path]
|
||||
|
||||
@property
|
||||
def can_encode_video(self) -> bool:
|
||||
return self.ffmpeg is not None and self.ffprobe is not None
|
||||
|
||||
|
||||
def _creation_flags() -> int:
|
||||
if os.name == "nt":
|
||||
return getattr(subprocess, "CREATE_NO_WINDOW", 0)
|
||||
return 0
|
||||
|
||||
|
||||
def find_ffmpeg_tools() -> FFmpegTools:
|
||||
"""Locate ffmpeg/ffprobe, preferring a bundled bin folder, then PATH."""
|
||||
override = os.environ.get("BOLT_FFMPEG_DIR")
|
||||
candidates = []
|
||||
if override:
|
||||
candidates.append(Path(override))
|
||||
|
||||
app_root = Path(__file__).resolve().parent.parent
|
||||
candidates.extend([app_root / "bin", app_root])
|
||||
|
||||
exe = ".exe" if os.name == "nt" else ""
|
||||
|
||||
def locate(name: str) -> Optional[Path]:
|
||||
for directory in candidates:
|
||||
candidate = directory / f"{name}{exe}"
|
||||
if candidate.is_file():
|
||||
return candidate
|
||||
resolved = shutil.which(name)
|
||||
return Path(resolved) if resolved else None
|
||||
|
||||
return FFmpegTools(ffmpeg=locate("ffmpeg"), ffprobe=locate("ffprobe"))
|
||||
|
||||
|
||||
def classify_path(path: Path) -> Optional[MediaKind]:
|
||||
suffix = path.suffix.lower()
|
||||
if suffix in IMAGE_EXTENSIONS:
|
||||
return MediaKind.IMAGE
|
||||
if suffix in VIDEO_EXTENSIONS:
|
||||
return MediaKind.VIDEO
|
||||
return None
|
||||
|
||||
|
||||
def human_size(num_bytes: int) -> str:
|
||||
value = float(max(0, num_bytes))
|
||||
units = ("B", "KB", "MB", "GB", "TB")
|
||||
for unit in units:
|
||||
if value < 1024 or unit == units[-1]:
|
||||
if unit == "B":
|
||||
return f"{int(value)} {unit}"
|
||||
if value >= 100:
|
||||
return f"{value:.0f} {unit}"
|
||||
if value >= 10:
|
||||
return f"{value:.1f} {unit}"
|
||||
return f"{value:.2f} {unit}"
|
||||
value /= 1024
|
||||
return f"{value:.1f} TB"
|
||||
|
||||
|
||||
def format_duration(seconds: float) -> str:
|
||||
if not seconds or seconds <= 0:
|
||||
return "—"
|
||||
total = int(round(seconds))
|
||||
hours, rem = divmod(total, 3600)
|
||||
minutes, secs = divmod(rem, 60)
|
||||
if hours:
|
||||
return f"{hours}:{minutes:02d}:{secs:02d}"
|
||||
return f"{minutes}:{secs:02d}"
|
||||
|
||||
|
||||
def _display_rotation(stream: dict) -> int:
|
||||
rotation = 0
|
||||
for side_data in stream.get("side_data_list", []) or []:
|
||||
if "rotation" in side_data:
|
||||
try:
|
||||
rotation = int(round(float(side_data["rotation"])))
|
||||
except (TypeError, ValueError):
|
||||
pass
|
||||
if rotation == 0:
|
||||
tags = stream.get("tags", {}) or {}
|
||||
if "rotate" in tags:
|
||||
try:
|
||||
rotation = int(round(float(tags["rotate"])))
|
||||
except (TypeError, ValueError):
|
||||
pass
|
||||
return rotation % 360
|
||||
|
||||
|
||||
def _probe_video(path: Path, size: int, tools: FFmpegTools) -> MediaInfo:
|
||||
if tools.ffprobe is None:
|
||||
raise MediaError("ffprobe was not found. Install FFmpeg and make sure it is on PATH.")
|
||||
|
||||
cmd = [
|
||||
str(tools.ffprobe),
|
||||
"-v",
|
||||
"error",
|
||||
"-show_streams",
|
||||
"-show_format",
|
||||
"-of",
|
||||
"json",
|
||||
str(path),
|
||||
]
|
||||
try:
|
||||
result = subprocess.run(
|
||||
cmd,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
encoding="utf-8",
|
||||
errors="replace",
|
||||
timeout=30,
|
||||
creationflags=_creation_flags(),
|
||||
)
|
||||
except (OSError, subprocess.TimeoutExpired) as exc:
|
||||
raise MediaError(f"Could not inspect video: {exc}") from exc
|
||||
|
||||
if result.returncode != 0:
|
||||
message = (result.stderr or "Unreadable video file").strip().splitlines()[-1]
|
||||
raise MediaError(message[:300])
|
||||
|
||||
try:
|
||||
payload = json.loads(result.stdout or "{}")
|
||||
except json.JSONDecodeError as exc:
|
||||
raise MediaError("Could not read video metadata.") from exc
|
||||
|
||||
streams = payload.get("streams", []) or []
|
||||
stream = next(
|
||||
(s for s in streams if s.get("codec_type") == "video" and s.get("disposition", {}).get("attached_pic") != 1),
|
||||
None,
|
||||
)
|
||||
if stream is None:
|
||||
stream = next((s for s in streams if s.get("codec_type") == "video"), None)
|
||||
if stream is None:
|
||||
raise MediaError("No video stream was found in this file.")
|
||||
|
||||
try:
|
||||
width = int(stream.get("width") or 0)
|
||||
height = int(stream.get("height") or 0)
|
||||
except (TypeError, ValueError):
|
||||
width = height = 0
|
||||
|
||||
rotation = _display_rotation(stream)
|
||||
if rotation in (90, 270):
|
||||
width, height = height, width
|
||||
|
||||
duration = 0.0
|
||||
for candidate in (payload.get("format", {}).get("duration"), stream.get("duration")):
|
||||
try:
|
||||
candidate_float = float(candidate)
|
||||
if candidate_float > 0:
|
||||
duration = candidate_float
|
||||
break
|
||||
except (TypeError, ValueError):
|
||||
continue
|
||||
|
||||
format_name = str(payload.get("format", {}).get("format_name", ""))
|
||||
primary_format = format_name.split(",")[0] if format_name else path.suffix.lstrip(".")
|
||||
primary_format = primary_format.upper()
|
||||
codec = str(stream.get("codec_name", "")).upper()
|
||||
|
||||
return MediaInfo(
|
||||
path=path,
|
||||
kind=MediaKind.VIDEO,
|
||||
size=size,
|
||||
width=width,
|
||||
height=height,
|
||||
format_name=primary_format,
|
||||
duration=duration,
|
||||
codec=codec,
|
||||
)
|
||||
|
||||
|
||||
def _probe_image(path: Path, size: int) -> MediaInfo:
|
||||
try:
|
||||
with Image.open(path) as image:
|
||||
if getattr(image, "is_animated", False) and getattr(image, "n_frames", 1) > 1:
|
||||
raise MediaError("Animated images are not supported yet.")
|
||||
width, height = image.size
|
||||
try:
|
||||
orientation = image.getexif().get(0x0112, 1)
|
||||
except Exception:
|
||||
orientation = 1
|
||||
if orientation in (5, 6, 7, 8):
|
||||
width, height = height, width
|
||||
format_name = (image.format or path.suffix.lstrip(".")).upper()
|
||||
except MediaError:
|
||||
raise
|
||||
except Exception as exc:
|
||||
raise MediaError(f"Could not read image: {exc}") from exc
|
||||
|
||||
if width <= 0 or height <= 0:
|
||||
raise MediaError("The image has no readable dimensions.")
|
||||
|
||||
return MediaInfo(
|
||||
path=path,
|
||||
kind=MediaKind.IMAGE,
|
||||
size=size,
|
||||
width=width,
|
||||
height=height,
|
||||
format_name=format_name,
|
||||
)
|
||||
|
||||
|
||||
def probe_media(path: Path, tools: Optional[FFmpegTools] = None) -> MediaInfo:
|
||||
path = Path(path)
|
||||
kind = classify_path(path)
|
||||
if kind is None:
|
||||
raise MediaError("Unsupported file type.")
|
||||
try:
|
||||
size = path.stat().st_size
|
||||
except OSError as exc:
|
||||
raise MediaError(f"File is not accessible: {exc}") from exc
|
||||
|
||||
if kind == MediaKind.IMAGE:
|
||||
return _probe_image(path, size)
|
||||
return _probe_video(path, size, tools or find_ffmpeg_tools())
|
||||
|
||||
|
||||
def image_output_extension(source: Path, format_choice: str) -> str:
|
||||
if format_choice == "jpeg":
|
||||
return ".jpg"
|
||||
if format_choice == "webp":
|
||||
return ".webp"
|
||||
if format_choice == "png":
|
||||
return ".png"
|
||||
keep_map = {
|
||||
".jpg": ".jpg",
|
||||
".jpeg": ".jpg",
|
||||
".png": ".png",
|
||||
".webp": ".webp",
|
||||
".bmp": ".bmp",
|
||||
".tif": ".tif",
|
||||
".tiff": ".tif",
|
||||
}
|
||||
return keep_map.get(source.suffix.lower(), ".jpg")
|
||||
|
||||
|
||||
def video_output_extension(codec: str) -> str:
|
||||
return ".webm" if codec in {"vp9", "av1"} else ".mp4"
|
||||
|
||||
|
||||
def unique_output_path(source: Path, folder: Optional[Path], suffix: str, extension: str) -> Path:
|
||||
destination_dir = Path(folder) if folder else source.parent
|
||||
base = destination_dir / f"{source.stem}{suffix}{extension}"
|
||||
candidate = base
|
||||
counter = 2
|
||||
while candidate.exists():
|
||||
candidate = destination_dir / f"{source.stem}{suffix} ({counter}){extension}"
|
||||
counter += 1
|
||||
return candidate
|
||||
|
||||
|
||||
def _run_encoders(ffmpeg: Path) -> str:
|
||||
try:
|
||||
result = subprocess.run(
|
||||
[str(ffmpeg), "-hide_banner", "-encoders"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
encoding="utf-8",
|
||||
errors="replace",
|
||||
timeout=20,
|
||||
creationflags=_creation_flags(),
|
||||
)
|
||||
except (OSError, subprocess.TimeoutExpired):
|
||||
return ""
|
||||
return result.stdout or ""
|
||||
|
||||
|
||||
_encoder_cache: dict[str, Set[str]] = {}
|
||||
|
||||
|
||||
def available_encoders(tools: Optional[FFmpegTools] = None) -> Set[str]:
|
||||
tools = tools or find_ffmpeg_tools()
|
||||
if tools.ffmpeg is None:
|
||||
return set()
|
||||
key = str(tools.ffmpeg)
|
||||
if key not in _encoder_cache:
|
||||
encoders: Set[str] = set()
|
||||
for line in _run_encoders(tools.ffmpeg).splitlines():
|
||||
match = re.match(r"^\s*V\S*\s+(\S+)", line)
|
||||
if match:
|
||||
encoders.add(match.group(1))
|
||||
_encoder_cache[key] = encoders
|
||||
return set(_encoder_cache[key])
|
||||
|
||||
|
||||
def available_video_codecs(tools: Optional[FFmpegTools] = None) -> Set[str]:
|
||||
encoders = available_encoders(tools)
|
||||
codecs: Set[str] = set()
|
||||
if "libx264" in encoders:
|
||||
codecs.add("h264")
|
||||
if "libx265" in encoders:
|
||||
codecs.add("h265")
|
||||
if "libvpx-vp9" in encoders:
|
||||
codecs.add("vp9")
|
||||
if {"libsvtav1", "libaom-av1"} & encoders:
|
||||
codecs.add("av1")
|
||||
return codecs
|
||||
|
||||
|
||||
def compact_error(text: str, limit: int = 400) -> str:
|
||||
lines = []
|
||||
for raw in (text or "").splitlines():
|
||||
line = raw.strip()
|
||||
if not line:
|
||||
continue
|
||||
if re.match(
|
||||
r"^(frame|fps|stream_\d|bitrate|total_size|out_time|dup_frames|drop_frames|speed|progress)=",
|
||||
line,
|
||||
):
|
||||
continue
|
||||
lines.append(line)
|
||||
message = " ".join(lines) if lines else (text or "Encoding failed.").strip()
|
||||
message = re.sub(r"\s+", " ", message)
|
||||
return message[-limit:] if len(message) > limit else message
|
||||
Reference in New Issue
Block a user