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]] = {} chaves_por_arquivo: dict[str, str] = {} for clipe in clipes: chave = chaves_por_arquivo.get(clipe.arquivo) if chave is None: chave = self._chave_do_arquivo(clipe.arquivo) chaves_por_arquivo[clipe.arquivo] = chave if chave not in transcricoes: transcricoes[chave] = self._transcrever_arquivo(clipe.arquivo, chave) resultados: list[TranscricaoDoClipe] = [] 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 = (chaves_por_arquivo[clipe.arquivo], inicio_origem, fim_origem, intervalo.inicio, intervalo.fim) 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