jhonny-editor/code/engine/integracoes/midia/extracao_de_metadados.py

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