- adicionadas métricas opcionais de fala ao Scanner com análise acústica na engine - corrigido o progresso da transcrição do Scanner durante o processamento do Whisper - refatorado o runner Swift do Apple Vision para executar requests em lote com retry CPU-only, corrigindo falhas de CVPixelBuffer/ANE/OCR - documentado o fluxo de edição de vídeo por voz integrado ao Scanner, banco e engine - formalizado o protocolo de edição por chat com perfil editorial e consultas progressivas Resumo: - 10 arquivos alterados - 5 novos - 5 modificados - 0 removidos 5 files changed, 367 insertions(+), 95 deletions(-) Arquivos: - .jhonny/analises.db - code/engine/executar_scanner.py - code/engine/integracoes/visual/apple_vision_runner.swift - code/engine/integracoes/whisper/provider_de_transcricao_local.py - code/engine/testes/test_provider_de_transcricao_local.py - .jhonny/analises.db-shm - .jhonny/analises.db-wal - apple_vision_runner - code/relatorios/teste-enquadramento-20s.json - docs/
52 lines
2.5 KiB
Python
52 lines
2.5 KiB
Python
"""Adapta o faster-whisper para a transcrição local do Scanner."""
|
|
|
|
from pathlib import Path
|
|
from typing import Any, Callable
|
|
|
|
from ...scanner.modelos import PalavraDeTranscricao, SegmentoDeTranscricao
|
|
|
|
|
|
class ProviderDeTranscricaoLocal:
|
|
"""Executa transcrição local em lote, sem transmitir a mídia."""
|
|
|
|
def __init__(self, modelo: str, dispositivo: str = "cpu", tipo_de_calculo: str = "int8",
|
|
idioma: str = "pt", tamanho_do_lote: int = 16,
|
|
ao_progresso: Callable[[float], None] | None = None) -> None:
|
|
"""Carrega o modelo local e configura o tamanho do lote de inferência."""
|
|
if tamanho_do_lote < 1:
|
|
raise ValueError("O tamanho do lote deve ser maior que zero.")
|
|
self.nome_do_modelo = Path(modelo).name
|
|
if "faster-whisper-" in modelo:
|
|
self.nome_do_modelo = modelo.split("faster-whisper-", 1)[1].split("/", 1)[0]
|
|
from faster_whisper import BatchedInferencePipeline, WhisperModel
|
|
self.modelo = WhisperModel(modelo, device=dispositivo, compute_type=tipo_de_calculo)
|
|
self.pipeline = BatchedInferencePipeline(model=self.modelo)
|
|
self.idioma = idioma
|
|
self.tamanho_do_lote = tamanho_do_lote
|
|
self.ao_progresso = ao_progresso
|
|
|
|
def transcrever(self, clipe: Any) -> list[SegmentoDeTranscricao]:
|
|
"""Transcreve o áudio com VAD, timestamps e inferência em lote."""
|
|
arquivo = Path(clipe.arquivo)
|
|
if not arquivo.is_file():
|
|
raise FileNotFoundError(f"Arquivo de áudio não encontrado: {arquivo}")
|
|
segmentos, informacoes = self.pipeline.transcribe(
|
|
str(arquivo), language=self.idioma, vad_filter=True,
|
|
word_timestamps=True, batch_size=self.tamanho_do_lote,
|
|
)
|
|
duracao = float(getattr(informacoes, "duration", 0.0) or 0.0)
|
|
resultado: list[SegmentoDeTranscricao] = []
|
|
for segmento in segmentos:
|
|
resultado.append(SegmentoDeTranscricao(
|
|
float(segmento.start), float(segmento.end), segmento.text.strip(),
|
|
None, tuple(PalavraDeTranscricao(
|
|
palavra.word.strip(), float(palavra.start), float(palavra.end),
|
|
getattr(palavra, "probability", None)
|
|
) for palavra in (segmento.words or []) if palavra.word.strip())
|
|
))
|
|
if self.ao_progresso and duracao > 0:
|
|
self.ao_progresso(min(99.0, max(0.0, float(segmento.end) / duracao * 100)))
|
|
if self.ao_progresso:
|
|
self.ao_progresso(100.0)
|
|
return resultado
|