jhonny-editor/code/engine/scanner/transcricao_da_timeline.py

from dataclasses import asdict, dataclass
import hashlib
import json
from pathlib import Path
from typing import Any, Callable

from ..integracoes.midia import ExtracaoDeAudio
from .modelos import PalavraDeTranscricao, SegmentoDeTranscricao


@dataclass(frozen=True)
class TranscricaoDoClipe:
    identificador_do_clipe: str
    arquivo: str
    inicio_na_timeline: float
    fim_na_timeline: float
    segmentos: list[SegmentoDeTranscricao]
    intervalo_na_origem: tuple[float, float] | None = None

    @property
    def texto(self) -> str:
        return " ".join(s.texto for s in self.segmentos if s.texto).strip()


class TranscricaoDaTimeline:
    """Transcreve cada arquivo uma vez e projeta o resultado nos cortes.

    A primeira versão suporta apenas mapeamento linear em velocidade normal.
    Os timestamps retornados pelo provider são sempre relativos ao arquivo
    original; somente a cópia materializada para cada clipe recebe timestamps
    da timeline.
    """

    def __init__(self, provider: Any, extrator: ExtracaoDeAudio | None = None,
                 diretoria_de_trabalho: str | Path = ".transcricao",
                 analisador_de_emocao: Any | None = None,
                 diarizador: Any | None = None) -> None:
        self.provider = provider
        self.extrator = extrator or ExtracaoDeAudio()
        self.diretoria_de_trabalho = Path(diretoria_de_trabalho)
        self.analisador_de_emocao = analisador_de_emocao
        self.diarizador = diarizador

    def executar(self, timeline: Any) -> list[TranscricaoDoClipe]:
        clipes: list[Any] = []
        for faixa in timeline.faixas:
            for clipe in faixa.clipes:
                if not clipe.arquivo or clipe.offline:
                    continue
                clipes.append(clipe)

        transcricoes: dict[str, list[SegmentoDeTranscricao]] = {}
        for clipe in clipes:
            chave = self._chave_do_arquivo(clipe.arquivo)
            if chave not in transcricoes:
                transcricoes[chave] = self._transcrever_arquivo(clipe.arquivo, chave)

        resultados: list[TranscricaoDoClipe] = []
        vistos: set[tuple[str, float, float, float, float]] = set()
        for clipe in clipes:
            intervalo = clipe.intervalo_na_timeline
            origem = clipe.intervalo_na_origem
            inicio_origem = origem.inicio if origem else 0.0
            fim_origem = origem.fim if origem else float("inf")
            chave = (self._chave_do_arquivo(clipe.arquivo), inicio_origem, fim_origem,
                     intervalo.inicio, intervalo.fim)
            # Vídeo e áudio vinculados podem representar o mesmo corte.
            if chave in vistos:
                continue
            vistos.add(chave)
            segmentos = []
            for segmento in transcricoes[chave[0]]:
                fim = min(segmento.fim, fim_origem)
                inicio = max(segmento.inicio, inicio_origem)
                if inicio < fim:
                    palavras = tuple(
                        PalavraDeTranscricao(
                            palavra.texto,
                            intervalo.inicio + max(palavra.inicio, inicio_origem) - inicio_origem,
                            intervalo.inicio + min(palavra.fim, fim_origem) - inicio_origem,
                            palavra.confianca,
                            palavra.falante,
                        )
                        for palavra in segmento.palavras
                        if palavra.inicio < fim_origem and palavra.fim > inicio_origem
                    )
                    segmentos.append(SegmentoDeTranscricao(
                        intervalo.inicio + inicio - inicio_origem,
                        intervalo.inicio + fim - inicio_origem,
                        segmento.texto, segmento.confianca, palavras,
                        segmento.emocao, segmento.confianca_emocao,
                        segmento.caracteristicas_acusticas, segmento.falante,
                        segmento.voz_aparente, segmento.confianca_voz))
            resultados.append(TranscricaoDoClipe(clipe.identificador, clipe.arquivo,
                intervalo.inicio, intervalo.fim, segmentos,
                (inicio_origem, fim_origem) if origem else None))
        return resultados

    def _chave_do_arquivo(self, arquivo: str) -> str:
        caminho = Path(arquivo).expanduser().resolve()
        digest = hashlib.sha256()
        with caminho.open("rb") as conteudo:
            for bloco in iter(lambda: conteudo.read(1024 * 1024), b""):
                digest.update(bloco)
        return digest.hexdigest()

    def _transcrever_arquivo(self, arquivo: str, chave: str) -> list[SegmentoDeTranscricao]:
        pasta = self.diretoria_de_trabalho / "arquivos" / chave
        nome_arquivo = Path(arquivo).stem
        modelo = self._nome_do_modelo()
        prefixo = f"transcricao-{self._nome_seguro(nome_arquivo)}-{self._nome_seguro(modelo)}"
        cache = pasta / f"{prefixo}.json"
        if cache.is_file():
            dados = json.loads(cache.read_text(encoding="utf-8"))
            if all("palavras" in item and
                   (self.analisador_de_emocao is None or
                    ("emocao" in item and "voz_aparente" in item))
                   for item in dados):
                return [self._segmento_do_json(item) for item in dados]
        partes = self.extrator.extrair(arquivo, pasta / "audio")
        segmentos: list[SegmentoDeTranscricao] = []
        for parte in partes:
            falas = self._diarizar(parte.caminho)
            for segmento in self.provider.transcrever(type("Audio", (), {"arquivo": str(parte.caminho)})()):
                analise = self._analisar_emocao(parte.caminho, segmento.inicio, segmento.fim)
                falante = self._falante_em(falas, segmento.inicio, segmento.fim)
                segmentos.append(SegmentoDeTranscricao(parte.inicio + segmento.inicio,
                    parte.inicio + segmento.fim, segmento.texto, segmento.confianca,
                    tuple(PalavraDeTranscricao(p.texto, parte.inicio + p.inicio,
                                                parte.inicio + p.fim, p.confianca,
                                                self._falante_em(falas, p.inicio, p.fim))
                          for p in segmento.palavras),
                    analise.get("emocao"), analise.get("confianca"),
                    dict(analise.get("caracteristicas_acusticas", {})), falante,
                    analise.get("voz_aparente"), analise.get("confianca_voz")))
        pasta.mkdir(parents=True, exist_ok=True)
        cache.write_text(json.dumps([asdict(item) for item in segmentos], ensure_ascii=False, indent=2), encoding="utf-8")
        (pasta / f"{prefixo}.txt").write_text(
            " ".join(item.texto for item in segmentos if item.texto).strip() + "\n",
            encoding="utf-8",
        )
        return segmentos

    @staticmethod
    def _segmento_do_json(item: dict[str, Any]) -> SegmentoDeTranscricao:
        palavras = tuple(PalavraDeTranscricao(str(p["texto"]), float(p["inicio"]),
                                               float(p["fim"]), p.get("confianca"),
                                               p.get("falante"))
                         for p in item.get("palavras", []))
        return SegmentoDeTranscricao(float(item["inicio"]), float(item["fim"]),
            str(item["texto"]), item.get("confianca"), palavras,
            item.get("emocao"), item.get("confianca_emocao"),
            dict(item.get("caracteristicas_acusticas", {})), item.get("falante"),
            item.get("voz_aparente"), item.get("confianca_voz"))

    def _analisar_emocao(self, arquivo: Path, inicio: float, fim: float) -> dict[str, Any]:
        if self.analisador_de_emocao is None:
            return {}
        return dict(self.analisador_de_emocao.analisar(arquivo, inicio, fim))

    def _diarizar(self, arquivo: Path) -> list[tuple[float, float, str]]:
        if self.diarizador is None:
            return []
        return list(self.diarizador.analisar(arquivo))

    @staticmethod
    def _falante_em(falas: list[tuple[float, float, str]], inicio: float, fim: float) -> str | None:
        sobreposicoes = [(min(fim, saida) - max(inicio, entrada), falante)
                         for entrada, saida, falante in falas
                         if entrada < fim and saida > inicio]
        return max(sobreposicoes, default=(0.0, None))[1]

    def _nome_do_modelo(self) -> str:
        """Obtém o nome público do modelo sem acoplar o scanner ao provider."""
        modelo = getattr(self.provider, "nome_do_modelo", None)
        if modelo:
            return str(modelo)
        modelo = getattr(self.provider, "modelo", None)
        if modelo:
            return Path(str(modelo)).name
        return self.provider.__class__.__name__.lower()

    @staticmethod
    def _nome_seguro(valor: str) -> str:
        return "".join(caractere if caractere.isalnum() or caractere in "._-" else "_"
                       for caractere in valor).strip("._") or "arquivo"

    @staticmethod
    def gerar_relatorio(resultados: list[TranscricaoDoClipe], destino: str | Path) -> Path:
        caminho = Path(destino)
        dados = [{**asdict(item), "segmentos": [asdict(s) for s in item.segmentos], "texto": item.texto}
                 for item in resultados]
        caminho.write_text(json.dumps(dados, ensure_ascii=False, indent=2), encoding="utf-8")
        return caminho