| 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) |
|
|
