File size: 2,760 Bytes
1425afc | 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 | from __future__ import annotations
import json
import mimetypes
import subprocess
from dataclasses import asdict
from pathlib import Path
from renderer.core.models import AssetMetadata
from renderer.core.utils import read_json, write_json
class AssetProbe:
def __init__(self, cache_path: Path) -> None:
self.cache_path = cache_path
self.cache: dict[str, dict] = read_json(cache_path, {})
def probe(self, path: str | Path) -> AssetMetadata:
media = Path(path)
stat = media.stat()
key = str(media.resolve())
cached = self.cache.get(key)
if cached and cached.get("size_bytes") == stat.st_size and cached.get("mtime") == stat.st_mtime:
return AssetMetadata(**cached)
raw = self._ffprobe(media)
streams = raw.get("streams", [])
fmt = raw.get("format", {})
video_stream = next((s for s in streams if s.get("codec_type") == "video"), {})
audio_stream = next((s for s in streams if s.get("codec_type") == "audio"), {})
metadata = AssetMetadata(
path=str(media),
mime_type=mimetypes.guess_type(str(media))[0] or "application/octet-stream",
size_bytes=stat.st_size,
mtime=stat.st_mtime,
duration=float(fmt.get("duration") or video_stream.get("duration") or audio_stream.get("duration") or 0),
width=_int_or_none(video_stream.get("width")),
height=_int_or_none(video_stream.get("height")),
fps=_parse_fps(video_stream.get("avg_frame_rate") or video_stream.get("r_frame_rate")),
bitrate=_int_or_none(fmt.get("bit_rate")),
video_codec=video_stream.get("codec_name"),
audio_codec=audio_stream.get("codec_name"),
has_audio=bool(audio_stream),
streams=streams,
)
self.cache[key] = asdict(metadata)
write_json(self.cache_path, self.cache)
return metadata
@staticmethod
def _ffprobe(path: Path) -> dict:
command = [
"ffprobe",
"-v",
"error",
"-show_format",
"-show_streams",
"-print_format",
"json",
str(path),
]
result = subprocess.run(command, check=True, capture_output=True, text=True)
return json.loads(result.stdout or "{}")
def _parse_fps(value: str | None) -> float | None:
if not value or value == "0/0":
return None
if "/" in value:
num, den = value.split("/", 1)
den_f = float(den)
return float(num) / den_f if den_f else None
return float(value)
def _int_or_none(value: object) -> int | None:
if value in (None, ""):
return None
return int(value)
|