import json import subprocess from dataclasses import dataclass from fractions import Fraction from pathlib import Path from typing import Any from ...scanner.modelos import ErroDeAnalise @dataclass(frozen=True) class MetadadosDoArquivo: caminho: Path formato: str | None = None duracao: float | None = None codec_de_video: str | None = None codec_de_audio: str | None = None largura: int | None = None altura: int | None = None taxa_de_quadros: float | None = None canais_de_audio: int | None = None taxa_de_amostragem: int | None = None tamanho_em_bytes: int | None = None orientacao: str | None = None timecode: str | None = None class ExtracaoDeMetadados: """Extrai metadados técnicos via ffprobe, com cache por caminho normalizado.""" def __init__(self, ffprobe: str = "ffprobe") -> None: self.ffprobe = ffprobe self._cache: dict[Path, MetadadosDoArquivo] = {} def extrair(self, arquivo: str | Path) -> MetadadosDoArquivo: caminho = Path(arquivo).expanduser().resolve(strict=False) if not caminho.is_file(): raise ErroDeAnalise("arquivo_inacessivel", "Arquivo de mídia não encontrado.", str(caminho)) if caminho in self._cache: return self._cache[caminho] try: processo = subprocess.run( [self.ffprobe, "-v", "error", "-print_format", "json", "-show_format", "-show_streams", str(caminho)], check=True, capture_output=True, text=True, ) dados = json.loads(processo.stdout) except (OSError, subprocess.SubprocessError, json.JSONDecodeError) as exc: raise ErroDeAnalise("metadados_indisponiveis", f"Não foi possível ler os metadados: {exc}", str(caminho)) from exc metadados = self._converter(caminho, dados) self._cache[caminho] = metadados return metadados def limpar_cache(self) -> None: self._cache.clear() @classmethod def _converter(cls, caminho: Path, dados: dict[str, Any]) -> MetadadosDoArquivo: formato = dados.get("format") or {} streams = dados.get("streams") or [] video = next((item for item in streams if item.get("codec_type") == "video"), {}) audio = next((item for item in streams if item.get("codec_type") == "audio"), {}) duracao = cls._float(formato.get("duration")) if duracao is None: duracao = cls._float(video.get("duration") or audio.get("duration")) return MetadadosDoArquivo( caminho=caminho, formato=cls._format_name(formato.get("format_name")), duracao=duracao, codec_de_video=video.get("codec_name") or None, codec_de_audio=audio.get("codec_name") or None, largura=cls._int(video.get("width")), altura=cls._int(video.get("height")), taxa_de_quadros=cls._fps(video.get("avg_frame_rate") or video.get("r_frame_rate")), canais_de_audio=cls._int(audio.get("channels")), taxa_de_amostragem=cls._int(audio.get("sample_rate")), tamanho_em_bytes=cls._int(formato.get("size")) or caminho.stat().st_size, orientacao=video.get("side_data_list", [{}])[0].get("rotation") if video.get("side_data_list") else None, timecode=(formato.get("tags") or {}).get("timecode") or (video.get("tags") or {}).get("timecode"), ) @staticmethod def _float(valor: Any) -> float | None: try: return None if valor in (None, "", "N/A") else float(valor) except (TypeError, ValueError): return None @staticmethod def _int(valor: Any) -> int | None: try: return None if valor in (None, "", "N/A") else int(valor) except (TypeError, ValueError): return None @staticmethod def _fps(valor: Any) -> float | None: if valor in (None, "", "0/0", "N/A"): return None try: return float(Fraction(str(valor))) except (ValueError, ZeroDivisionError): return None @staticmethod def _format_name(valor: Any) -> str | None: return str(valor).split(",", 1)[0] if valor else None