João Henrique
dd341e7d4c
- Instalado essentia-tensorflow no venv whisperx-transcricao (usado pelo painel CEP), corrigindo erro "A classificação musical precisa do suporte TensorFlow do Essentia" no catálogo de trilhas. - Ligada a classificação de emoção da fala (ProviderDeEmocaoHuggingFace) ao scanner via nova flag `classificar_emocao`, e adicionado emoção/confiança como critério de desempate secundário no zoom da skill editar-por-voz. - Zoom do caminho Python (aplicar_plano_de_edicao.py) passou a usar adjustment layer numa faixa de vídeo própria, em vez de escalar o clipe inteiro via set_clip_properties — mesma técnica já usada pelo executor .mjs do painel CEP. - Adicionado campo "Carregar por ID" na aba Editar Vídeo do painel CEP (step 2), para carregar um plano do SQLite direto pelo plano_id quando ele não está associado a nenhum video_id conhecido (ex.: plano de um vídeo que este banco nunca escaneou). - Trocado o campo de ID numérico por uma lista suspensa (novo listar_planos_de_edicao.py + RepositorioDePlanos.listar_todos_os_planos) que mostra data/hora e a intenção registrada pela skill em cada plano salvo, para escolher sem decorar o id. Resumo: - 15 arquivos alterados - 2 novos - 13 modificados - 0 removidos 13 files changed, 451 insertions(+), 65 deletions(-) Arquivos: - .jhonny/analises.db - code/cep-plugin/index.html - code/cep-plugin/main.js - code/engine/editor/aplicador_de_plano_de_edicao.py - code/engine/editor/escrita/escrita_no_editor.py - code/engine/executar_scanner.py - code/engine/persistencia/repositorio_de_planos.py - code/engine/scanner/configuracao_visual.py - code/engine/testes/duplos_de_premiere.py - code/engine/testes/test_aplicador_de_plano_de_edicao.py - code/engine/testes/test_escrita_no_editor.py - code/plugins/premiere-pro/skills/editar-por-voz/criterios/01-leitura-do-json.md - code/plugins/premiere-pro/skills/editar-por-voz/criterios/05-zoom.md - code/engine/listar_planos_de_edicao.py - code/engine/testes/test_configuracao_visual_do_scanner.py
348 lines
20 KiB
Python
348 lines
20 KiB
Python
"""Entrada local do painel CEP para uma execução configurada do Scanner."""
|
|
|
|
import argparse
|
|
import json
|
|
import math
|
|
import os
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
import re
|
|
from types import SimpleNamespace
|
|
|
|
# O painel chama este arquivo diretamente, portanto o Python inclui apenas
|
|
# ``code/engine`` no caminho de importação. O pacote público fica em ``code``.
|
|
CAMINHO_DO_CODIGO = Path(__file__).resolve().parent.parent
|
|
if str(CAMINHO_DO_CODIGO) not in sys.path:
|
|
sys.path.insert(0, str(CAMINHO_DO_CODIGO))
|
|
|
|
|
|
def _encontrar_caminho_da_midia() -> str:
|
|
"""Localiza a pasta com ``ffmpeg``/``ffprobe`` para o PATH dos subprocessos.
|
|
|
|
O Premiere é aberto pelo Finder/LaunchServices (sem shell de login), então o
|
|
PATH herdado pelo painel não costuma trazer ``/opt/homebrew/bin``. Essa
|
|
função devolve o primeiro caminho conhecido onde essas ferramentas existem,
|
|
ou string vazia quando nenhum é encontrado.
|
|
"""
|
|
for candidato in ("/opt/homebrew/bin", "/usr/local/bin"):
|
|
if Path(candidato, "ffprobe").is_file() or Path(candidato, "ffmpeg").is_file():
|
|
return candidato
|
|
return ""
|
|
|
|
|
|
# O mesmo ajuste de PATH já usado em ``admin/deploy.command`` e
|
|
# ``admin/install.command``: sem ele, a extração de frames e a transcrição
|
|
# falham com ``No such file or directory: 'ffprobe'``.
|
|
CAMINHO_DA_MIDIA = _encontrar_caminho_da_midia()
|
|
if CAMINHO_DA_MIDIA:
|
|
os.environ["PATH"] = CAMINHO_DA_MIDIA + os.pathsep + os.environ.get("PATH", "")
|
|
|
|
from engine.integracoes.premiere.conversores import ConversorDeTimeline
|
|
from engine.integracoes.midia import ExtracaoDeAudio
|
|
from engine.dominio import Clipe
|
|
from engine.scanner import ConfiguracaoVisualDoScanner, criar_detector_apple_vision, gerar_relatorio_visual
|
|
from engine.scanner.transcricao_da_timeline import TranscricaoDaTimeline
|
|
from engine.persistencia import RepositorioDeAnalisesSQLite, RepositorioDeTimelineSQLite
|
|
|
|
|
|
MODELO_DIARIZACAO_PADRAO = "pyannote/speaker-diarization-community-1"
|
|
|
|
# O padrão do banco de análises é ``.jhonny/analises.db`` relativo ao cwd, mas o
|
|
# painel roda o scanner com cwd ``code``. Fixamos o caminho absoluto na raiz do
|
|
# projeto para garantir que a gravação sempre caia no mesmo banco.
|
|
CAMINHO_DO_BANCO = CAMINHO_DO_CODIGO.parent / ".jhonny" / "analises.db"
|
|
|
|
|
|
def nome_seguro(nome: str) -> str:
|
|
"""Converte o nome de uma sequência em nome de pasta seguro no filesystem.
|
|
|
|
Parâmetros:
|
|
nome: Nome original da sequência, como vem do Premiere.
|
|
|
|
Retorna:
|
|
O nome sem caracteres problemáticos, ou "timeline" quando nada sobra.
|
|
"""
|
|
return re.sub(r"[^A-Za-z0-9._-]+", "-", nome.strip()).strip(".-") or "timeline"
|
|
|
|
|
|
def serializar_clipe_para_relatorio(clipe: Clipe) -> dict[str, object]:
|
|
"""Preserva no relatório os dados que identificam a mídia e sua origem.
|
|
|
|
As chaves ``sourceFile``, ``inPoint`` e ``outPoint`` mantêm o contrato
|
|
externo usado pelo painel CEP e pelo bridge do Premiere.
|
|
"""
|
|
intervalo_na_origem = clipe.intervalo_na_origem
|
|
return {
|
|
"identificador": clipe.identificador,
|
|
"nome": clipe.nome,
|
|
"inicio": clipe.intervalo_na_timeline.inicio,
|
|
"fim": clipe.intervalo_na_timeline.fim,
|
|
"sourceFile": clipe.arquivo,
|
|
"inPoint": intervalo_na_origem.inicio if intervalo_na_origem else None,
|
|
"outPoint": intervalo_na_origem.fim if intervalo_na_origem else None,
|
|
}
|
|
|
|
|
|
def executar(entrada: Path, saida: Path) -> Path:
|
|
"""Executa uma análise do Scanner conforme o pedido do painel.
|
|
|
|
Persiste a timeline, as evidências visuais, as cenas e a transcrição no
|
|
banco de análises, e grava os relatórios JSON que o painel exibe.
|
|
|
|
Parâmetros:
|
|
entrada: Arquivo JSON com o perfil e a timeline a analisar.
|
|
saida: Caminho de referência para a pasta onde o relatório será
|
|
escrito; o nome final é derivado do nome da sequência.
|
|
|
|
Retorna:
|
|
O caminho do relatório JSON gerado.
|
|
"""
|
|
pedido = json.loads(entrada.read_text(encoding="utf-8"))
|
|
configuracao = ConfiguracaoVisualDoScanner.de_dict(pedido["perfil"])
|
|
timeline = ConversorDeTimeline().converter(pedido["timeline"])
|
|
analisar_imagem = configuracao.modo_de_analise in {"imagem", "ambos"}
|
|
analisar_som = configuracao.modo_de_analise in {"som", "ambos"}
|
|
clipes = [clipe for faixa in timeline.faixas if analisar_imagem and faixa.tipo == "video"
|
|
and faixa.indice in configuracao.faixas_de_video for clipe in faixa.clipes]
|
|
total_estimado = sum(max(1, math.ceil((clipe.intervalo_na_origem or clipe.intervalo_na_timeline).duracao /
|
|
configuracao.intervalo_em_segundos)) + 1 for clipe in clipes) or 1
|
|
concluidos = 0
|
|
|
|
def emitir(etapa: str, percentual: float, clipe: str = "") -> None:
|
|
print(json.dumps({"evento": "progresso", "etapa": etapa, "percentual": round(percentual, 1),
|
|
"clipe": clipe}, ensure_ascii=False), flush=True)
|
|
|
|
# Persistência: timeline/estrutura + análise visual + transcrição vão para o
|
|
# banco SQLite do projeto (mesmo dos relatórios gerados).
|
|
repositorio_de_timeline = RepositorioDeTimelineSQLite(CAMINHO_DO_BANCO)
|
|
repositorio_de_analises = RepositorioDeAnalisesSQLite(CAMINHO_DO_BANCO)
|
|
emitir("Persistindo timeline", 0)
|
|
repositorio_de_timeline.registrar_timeline(timeline)
|
|
|
|
resultados = []
|
|
for clipe in clipes:
|
|
peso = max(1, math.ceil((clipe.intervalo_na_origem or clipe.intervalo_na_timeline).duracao /
|
|
configuracao.intervalo_em_segundos)) + 1
|
|
inicio_do_clipe = concluidos
|
|
|
|
def progresso_atual(atual: int, total: int, nome: str = clipe.nome,
|
|
base: int = inicio_do_clipe, peso_do_clipe: int = peso) -> None:
|
|
emitir("Analisando frames", 100 * (base + peso_do_clipe * atual / max(total, 1)) / total_estimado, nome)
|
|
|
|
detector = criar_detector_apple_vision(Path(tempfile.gettempdir()) / "premiere-mcp-scanner",
|
|
configuracao=configuracao)
|
|
detector.ao_progresso = progresso_atual
|
|
emitir("Extraindo frames", 100 * inicio_do_clipe / total_estimado, clipe.nome)
|
|
try:
|
|
resultado = detector.detectar(clipe)
|
|
arquivo = Path(tempfile.gettempdir()) / f"scanner-{clipe.identificador}.json"
|
|
gerar_relatorio_visual(clipe, resultado, arquivo, configuracao.intervalo_em_segundos)
|
|
resultados.append(json.loads(arquivo.read_text(encoding="utf-8")))
|
|
# Grava no banco os fatos visuais e as cenas detectadas deste clipe.
|
|
# Substitui em vez de acrescentar: reanalisar o mesmo clipe tem de
|
|
# deixar uma cópia só de cada evidência, não empilhar execuções.
|
|
repositorio_de_analises.substituir_evidencias_visuais(
|
|
timeline.identificador, clipe.identificador, resultado.evidencias)
|
|
repositorio_de_analises.substituir_cenas(
|
|
timeline.identificador, resultado.cenas, clipe.identificador)
|
|
except (OSError, RuntimeError, ValueError, json.JSONDecodeError) as erro:
|
|
# Falha esperada de mídia/detector (arquivo ilegível, subprocesso do
|
|
# Apple Vision indisponível, resposta malformada): não deve descartar
|
|
# os demais clipes. Erros de programação (TypeError, AttributeError
|
|
# etc.) propagam normalmente para não serem mascarados como isso.
|
|
resultados.append({"clipe": {"identificador": clipe.identificador, "nome": clipe.nome,
|
|
"arquivo_original": clipe.arquivo}, "erro": str(erro)})
|
|
concluidos += peso
|
|
faixas_de_audio = [faixa for faixa in timeline.faixas
|
|
if analisar_som and faixa.tipo == "audio"
|
|
and faixa.indice in configuracao.faixas_de_audio]
|
|
audio = [{"indice": faixa.indice, "nome": faixa.nome,
|
|
"clipes": [serializar_clipe_para_relatorio(clipe) for clipe in faixa.clipes]}
|
|
for faixa in faixas_de_audio]
|
|
caso = nome_seguro(timeline.nome)
|
|
pasta = saida.parent / caso
|
|
pasta.mkdir(parents=True, exist_ok=True)
|
|
saida = pasta / f"{caso}-resultado.json"
|
|
detalhado = pasta / "detalhado"
|
|
detalhado.mkdir(parents=True, exist_ok=True)
|
|
geometria = []
|
|
for resultado in resultados:
|
|
for frame in resultado.get("observacoes_por_frame", []):
|
|
fatos = [item for item in frame["evidencias"] if item["tipo"] in {"rosto", "pessoa", "pose", "mao"}]
|
|
if fatos:
|
|
geometria.append({"clipe": resultado["clipe"]["identificador"], "timestamp_na_origem": frame["timestamp_na_origem"], "evidencias": fatos})
|
|
frame["evidencias"] = [item for item in frame["evidencias"] if item not in fatos]
|
|
(detalhado / "geometria.jsonl").write_text("".join(json.dumps(item, ensure_ascii=False) + "\n" for item in geometria), encoding="utf-8")
|
|
video_arquivo = f"{caso}-video.json"
|
|
(pasta / video_arquivo).write_text(json.dumps({"clipes": resultados}, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
audio_arquivos = []
|
|
transcricoes_por_clipe: dict[str, list[dict]] = {}
|
|
transcricao_configurada = pedido["perfil"].get("transcricao", {})
|
|
transcricao_configurada = {
|
|
**transcricao_configurada,
|
|
"metricas_de_fala": configuracao.metricas_de_fala,
|
|
"classificar_emocao": configuracao.classificar_emocao,
|
|
}
|
|
if faixas_de_audio and transcricao_configurada.get("caminho_modelo"):
|
|
try:
|
|
from engine.integracoes.whisper import ProviderDeTranscricaoLocal
|
|
from engine.integracoes.audio import AnalisadorDeMetricasDeVoz
|
|
from engine.dominio import Timeline
|
|
emitir("Carregando modelo Whisper", 5)
|
|
provider = ProviderDeTranscricaoLocal(transcricao_configurada["caminho_modelo"],
|
|
idioma=transcricao_configurada.get("idioma", "pt"),
|
|
ao_progresso=lambda percentual: emitir(
|
|
"Transcrevendo áudio", 5 + percentual * 0.8))
|
|
analisador_de_metricas_de_voz = (
|
|
AnalisadorDeMetricasDeVoz() if configuracao.metricas_de_fala else None
|
|
)
|
|
analisador_de_emocao = None
|
|
if configuracao.classificar_emocao:
|
|
from engine.integracoes.huggingface import ProviderDeEmocaoHuggingFace
|
|
analisador_de_emocao = ProviderDeEmocaoHuggingFace()
|
|
diarizador = None
|
|
if transcricao_configurada.get("diarizacao"):
|
|
if not os.environ.get("HF_TOKEN"):
|
|
transcricao_configurada = {**transcricao_configurada,
|
|
"aviso_diarizacao": "Token do Hugging Face não configurado."}
|
|
else:
|
|
from engine.integracoes.huggingface import ProviderDeDiarizacaoHuggingFace
|
|
modelo_diarizacao = os.environ.get("HF_DIARIZATION_MODEL", MODELO_DIARIZACAO_PADRAO)
|
|
diarizador = ProviderDeDiarizacaoHuggingFace(modelo_diarizacao)
|
|
transcricao_configurada = {**transcricao_configurada,
|
|
"modelo_diarizacao": modelo_diarizacao}
|
|
emitir("Transcrevendo áudio", 5)
|
|
transcricoes = TranscricaoDaTimeline(provider, diretoria_de_trabalho=pasta / ".cache-audio",
|
|
diarizador=diarizador,
|
|
analisador_de_metricas_de_voz=analisador_de_metricas_de_voz,
|
|
analisador_de_emocao=analisador_de_emocao).executar(
|
|
Timeline(timeline.identificador, timeline.nome, faixas=faixas_de_audio))
|
|
# Persiste a transcrição (segmentos + falas) no banco de análises.
|
|
repositorio_de_analises.substituir_transcricoes(timeline.identificador, transcricoes)
|
|
for transcricao in transcricoes:
|
|
transcricoes_por_clipe[transcricao.identificador_do_clipe] = [
|
|
{"inicio": item.inicio, "fim": item.fim, "texto": item.texto,
|
|
"confianca": item.confianca, "falante": item.falante,
|
|
"caracteristicas_acusticas": item.caracteristicas_acusticas,
|
|
"emocao": item.emocao, "confianca_emocao": item.confianca_emocao,
|
|
"voz_aparente": item.voz_aparente, "confianca_voz": item.confianca_voz}
|
|
for item in transcricao.segmentos]
|
|
except (ImportError, OSError, RuntimeError, ValueError) as erro:
|
|
# Dependência ausente (ex.: soundfile), modelo/áudio ilegível, ou
|
|
# falha de runtime do Whisper/pyannote: registra e segue sem
|
|
# transcrição. Erros de programação propagam normalmente.
|
|
transcricao_configurada = {**transcricao_configurada, "erro": str(erro)}
|
|
for faixa in audio:
|
|
arquivo = f"{caso}-audio-faixa-{faixa['indice'] + 1}.json"
|
|
segmentos = [{"clipe": clipe["identificador"], "segmentos": transcricoes_por_clipe.get(clipe["identificador"], [])}
|
|
for clipe in faixa["clipes"]]
|
|
(pasta / arquivo).write_text(json.dumps({"faixa": faixa, "transcricao": transcricao_configurada,
|
|
"segmentos_por_clipe": segmentos,
|
|
"status": "concluida" if transcricoes_por_clipe else "indisponivel"},
|
|
ensure_ascii=False, indent=2), encoding="utf-8")
|
|
audio_arquivos.append(arquivo)
|
|
saida.write_text(json.dumps({"versao": 1, "perfil": pedido["perfil"],
|
|
"sequencia": {"id": timeline.identificador, "nome": timeline.nome},
|
|
"artefatos": {"audio": audio_arquivos, "video": video_arquivo,
|
|
"geometria": "detalhado/geometria.jsonl"}}, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
emitir("Relatório JSON gerado", 100)
|
|
return saida
|
|
|
|
|
|
def executar_complemento(entrada: Path, relatorio: Path, acao: str) -> Path:
|
|
"""Processa locutores ou métricas usando a transcrição já persistida."""
|
|
pedido = json.loads(entrada.read_text(encoding="utf-8"))
|
|
dados_relatorio = json.loads(relatorio.read_text(encoding="utf-8"))
|
|
timeline = ConversorDeTimeline().converter(pedido["timeline"])
|
|
clipes = {clipe.identificador: clipe for faixa in timeline.faixas for clipe in faixa.clipes}
|
|
partes_por_arquivo: dict[str, list] = {}
|
|
diarizacao_por_parte: dict[str, list] = {}
|
|
if acao == "diarizacao":
|
|
if not os.environ.get("HF_TOKEN"):
|
|
raise RuntimeError("Token do Hugging Face não configurado para identificar locutores.")
|
|
from engine.integracoes.huggingface import ProviderDeDiarizacaoHuggingFace
|
|
diarizador = ProviderDeDiarizacaoHuggingFace(os.environ.get("HF_DIARIZATION_MODEL", MODELO_DIARIZACAO_PADRAO))
|
|
analisador = None
|
|
elif acao == "metricas_de_fala":
|
|
from engine.integracoes.audio import AnalisadorDeMetricasDeVoz
|
|
diarizador = None
|
|
analisador = AnalisadorDeMetricasDeVoz()
|
|
else:
|
|
raise ValueError(f"Complemento desconhecido: {acao}")
|
|
|
|
arquivos_audio = dados_relatorio["artefatos"].get("audio", [])
|
|
repositorio_de_analises = RepositorioDeAnalisesSQLite(CAMINHO_DO_BANCO)
|
|
total = max(1, len(arquivos_audio))
|
|
for indice_arquivo, nome_arquivo in enumerate(arquivos_audio, 1):
|
|
caminho_relatorio = relatorio.parent / nome_arquivo
|
|
dados_audio = json.loads(caminho_relatorio.read_text(encoding="utf-8"))
|
|
for grupo in dados_audio.get("segmentos_por_clipe", []):
|
|
clipe = clipes.get(grupo.get("clipe"))
|
|
if not clipe or not clipe.arquivo:
|
|
continue
|
|
caminho_origem = str(clipe.arquivo)
|
|
if caminho_origem not in partes_por_arquivo:
|
|
partes_por_arquivo[caminho_origem] = ExtracaoDeAudio().extrair(
|
|
caminho_origem, relatorio.parent / ".cache-audio-complementar" / str(abs(hash(caminho_origem))))
|
|
partes = partes_por_arquivo[caminho_origem]
|
|
inicio_origem = clipe.intervalo_na_origem.inicio if clipe.intervalo_na_origem else 0.0
|
|
fim_anterior = None
|
|
for segmento in grupo.get("segmentos", []):
|
|
inicio = inicio_origem + segmento["inicio"] - clipe.intervalo_na_timeline.inicio
|
|
fim = inicio_origem + segmento["fim"] - clipe.intervalo_na_timeline.inicio
|
|
parte = next((item for item in partes if item.inicio <= inicio <= item.fim), partes[-1])
|
|
inicio_local = max(0.0, inicio - parte.inicio)
|
|
fim_local = max(inicio_local, min(fim, parte.fim) - parte.inicio)
|
|
chave = str(parte.caminho)
|
|
if acao == "diarizacao":
|
|
if chave not in diarizacao_por_parte:
|
|
diarizacao_por_parte[chave] = diarizador.analisar(parte.caminho)
|
|
falas = diarizacao_por_parte[chave]
|
|
sobreposicoes = [(min(fim_local, saida) - max(inicio_local, entrada), falante)
|
|
for entrada, saida, falante in falas
|
|
if entrada < fim_local and saida > inicio_local]
|
|
segmento["falante"] = max(sobreposicoes, default=(0.0, None))[1]
|
|
repositorio_de_analises.atualizar_enriquecimento_de_transcricao(
|
|
timeline.identificador,
|
|
clipe.identificador,
|
|
segmento["inicio"],
|
|
segmento["fim"],
|
|
falante=segmento.get("falante"),
|
|
)
|
|
else:
|
|
palavras = [SimpleNamespace(inicio=0.0, fim=0.0)
|
|
for _ in segmento.get("texto", "").split()]
|
|
segmento["caracteristicas_acusticas"] = analisador.analisar(
|
|
parte.caminho, inicio_local, fim_local, palavras,
|
|
segmento["inicio"], fim_anterior)
|
|
repositorio_de_analises.atualizar_enriquecimento_de_transcricao(
|
|
timeline.identificador,
|
|
clipe.identificador,
|
|
segmento["inicio"],
|
|
segmento["fim"],
|
|
caracteristicas_acusticas=segmento["caracteristicas_acusticas"],
|
|
)
|
|
fim_anterior = segmento["fim"]
|
|
dados_audio.setdefault("transcricao", {})["diarizacao" if acao == "diarizacao" else "metricas_de_fala"] = True
|
|
caminho_relatorio.write_text(json.dumps(dados_audio, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
print(json.dumps({"evento": "progresso", "etapa": "Complementando transcrição",
|
|
"percentual": round(100 * indice_arquivo / total, 1)}, ensure_ascii=False), flush=True)
|
|
dados_relatorio.setdefault("perfil", {}).setdefault("transcricao", {})[
|
|
"diarizacao" if acao == "diarizacao" else "metricas_de_fala"] = True
|
|
dados_relatorio.write_text(json.dumps(dados_relatorio, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
return relatorio
|
|
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("entrada", type=Path)
|
|
parser.add_argument("saida", type=Path)
|
|
parser.add_argument("--complemento", choices=("diarizacao", "metricas_de_fala"))
|
|
args = parser.parse_args()
|
|
if args.complemento:
|
|
caminho = executar_complemento(args.entrada, args.saida, args.complemento)
|
|
else:
|
|
caminho = executar(args.entrada, args.saida)
|
|
print(json.dumps({"evento": "concluido", "arquivo": str(caminho)}, ensure_ascii=False))
|