"""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.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: return re.sub(r"[^A-Za-z0-9._-]+", "-", nome.strip()).strip(".-") or "timeline" def executar(entrada: Path, saida: Path) -> Path: 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. repositorio_de_analises.registrar_evidencias_visuais( timeline.identificador, clipe.identificador, resultado.evidencias) repositorio_de_analises.registrar_cenas( timeline.identificador, resultado.cenas, clipe.identificador) except Exception as erro: 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": [{"identificador": clipe.identificador, "nome": clipe.nome, "inicio": clipe.intervalo_na_timeline.inicio, "fim": clipe.intervalo_na_timeline.fim} 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 Exception as erro: 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))