"""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, } 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 ) 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).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} 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))