João Henrique
b03b175973
- adicionadas métricas opcionais de fala ao Scanner com análise acústica na engine Resumo: - 7 arquivos alterados - 2 novos - 5 modificados - 0 removidos 5 files changed, 43 insertions(+), 6 deletions(-) Arquivos: - code/cep-plugin/index.html - code/cep-plugin/main.js - code/engine/executar_scanner.py - code/engine/scanner/configuracao_visual.py - code/engine/scanner/transcricao_da_timeline.py - code/engine/integracoes/audio/ - code/engine/testes/test_analisador_de_metricas_de_voz.py
215 lines
10 KiB
Python
215 lines
10 KiB
Python
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,
|
|
analisador_de_metricas_de_voz: 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
|
|
self.analisador_de_metricas_de_voz = analisador_de_metricas_de_voz
|
|
|
|
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)) and
|
|
(self.analisador_de_metricas_de_voz is None or
|
|
"caracteristicas_acusticas" 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] = []
|
|
fim_anterior: float | None = None
|
|
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)
|
|
inicio_absoluto = parte.inicio + segmento.inicio
|
|
metricas_de_voz = self._analisar_metricas_de_voz(
|
|
parte.caminho, segmento.inicio, segmento.fim, segmento.palavras,
|
|
inicio_absoluto, fim_anterior,
|
|
)
|
|
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"),
|
|
metricas_de_voz or dict(analise.get("caracteristicas_acusticas", {})), falante,
|
|
analise.get("voz_aparente"), analise.get("confianca_voz")))
|
|
fim_anterior = inicio_absoluto + segmento.fim - segmento.inicio
|
|
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 _analisar_metricas_de_voz(self, arquivo: Path, inicio: float, fim: float,
|
|
palavras: tuple[PalavraDeTranscricao, ...],
|
|
inicio_absoluto: float,
|
|
fim_anterior: float | None) -> dict[str, Any]:
|
|
"""Calcula métricas acústicas somente quando foram solicitadas."""
|
|
if self.analisador_de_metricas_de_voz is None:
|
|
return {}
|
|
return dict(self.analisador_de_metricas_de_voz.analisar(
|
|
arquivo, inicio, fim, palavras, inicio_absoluto, fim_anterior))
|
|
|
|
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
|