feat: copiada a skill editar-por-voz e seus critérios para as skil

- copiada a skill editar-por-voz e seus critérios para as skills do projeto
- adaptada a skill editar-por-voz para usar o banco SQLite como fonte oficial
- corrigida a cópia do JSON do tipo de vídeo para priorizar clipboard Unicode
- criado carregador único do contexto de edição por voz a partir do SQLite
- criada entrada para registrar planos no SQLite e reaproveitar o mesmo plano na aplicação
- habilitado carregamento do plano editorial diretamente do SQLite na interface
- corrigido o bloqueio visual da etapa de carregamento do plano salvo
- configurado carregamento automático do plano salvo após selecionar a sequência
- documentada a sequência como raiz de retomada do fluxo de edição
- criada a estrutura local para análise de músicas instrumentais com Essentia

Resumo:
- 22 arquivos alterados
- 11 novos
- 11 modificados
- 0 removidos

 11 files changed, 270 insertions(+), 34 deletions(-)

Arquivos:
  - .jhonny/analises.db
  - CONTEXT.md
  - code/cep-plugin/index.html
  - code/cep-plugin/main.js
  - code/engine/README.md
  - code/engine/aplicar_plano_de_edicao.py
  - code/engine/integracoes/audio/__init__.py
  - code/engine/persistencia/consultas.py
  - code/engine/requirements-audio.txt
  - code/engine/testes/test_consultas_de_persistencia.py
  - code/tests/cep-tipos-video-encoding.test.ts
  - code/.jhonny/
  - code/engine/carregar_plano_de_edicao.py
  - code/engine/integracoes/audio/contratos.py
  - code/engine/integracoes/audio/modelos_de_analise_musical.py
  - code/engine/integracoes/audio/provider_de_analise_musical_essentia.py
  - code/engine/preparar_edicao_por_voz.py
  - code/engine/registrar_plano_de_edicao.py
  - code/engine/testes/test_analisador_de_musica_essentia.py
  - code/engine/testes/test_carregar_plano_de_edicao.py
  - code/engine/testes/test_registrar_plano_de_edicao.py
  - code/plugins/premiere-pro/skills/editar-por-voz/
This commit is contained in:
João Henrique
2026-09-10 13:09:01 -04:00
parent eda1b1aadc
commit a9e5b5ff58
33 changed files with 1811 additions and 34 deletions
+35
View File
@@ -11,3 +11,38 @@ assert result.valid
```
As integrações reais devem implementar os protocolos em `content_analyzer.py`, `decision_engine.py`, `edit_plan.py`, `plan_applicator.py`, `validator.py` e `persistence.py`. O pacote não cria dependências externas por padrão.
## Análise de músicas instrumentais
A análise musical local fica separada da análise de voz e segue este fluxo:
```text
caso de uso da engine
↓
AnalisadorDeMusica (contrato)
↓
AnalisadorDeMusicaInstrumentalEssentia
↓
ResultadoDaAnaliseMusical
```
Arquivos principais:
- `integracoes/audio/contratos.py`: contrato substituível do analisador;
- `integracoes/audio/modelos_de_analise_musical.py`: resultado interno em PT-BR;
- `integracoes/audio/provider_de_analise_musical_essentia.py`: adaptador local do Essentia;
- `testes/test_analisador_de_musica_essentia.py`: testes unitários sem instalar Essentia.
Exemplo de uso:
```python
from engine.integracoes.audio import AnalisadorDeMusicaInstrumentalEssentia
resultado = AnalisadorDeMusicaInstrumentalEssentia().analisar("musica.wav")
print(resultado.generos, resultado.humores, resultado.batidas_por_minuto)
```
A dependência é carregada sob demanda. Para habilitar a análise, instale as
dependências de `requirements-audio.txt`. O adaptador não persiste o resultado
nem gera a descrição narrativa; essas responsabilidades ficarão no caso de
uso e na apresentação, respectivamente.
+26 -9
View File
@@ -17,6 +17,7 @@ troca de sequência.
from __future__ import annotations
import argparse
import json
import os
import shutil
@@ -81,7 +82,13 @@ def resolver_caminho_do_node() -> str:
)
def _registrar_no_banco(caminho_do_plano: Path, resultado: object, sequencia: str) -> None:
def _registrar_no_banco(
caminho_do_plano: Path,
resultado: object,
sequencia: str,
plano_id: int | None = None,
caminho_do_banco: Path = CAMINHO_DO_BANCO,
) -> None:
"""Guarda no banco de análises o plano aplicado e o resultado da aplicação.
O registro é o único vestígio de como uma edição foi decidida: sem ele o
@@ -103,8 +110,9 @@ def _registrar_no_banco(caminho_do_plano: Path, resultado: object, sequencia: st
)
dados = json.loads(caminho_do_plano.read_text(encoding="utf-8"))
repositorio = RepositorioDePlanos(abrir_banco(CAMINHO_DO_BANCO))
plano_id = repositorio.registrar_plano(plano_de_dict(dados))
repositorio = RepositorioDePlanos(abrir_banco(caminho_do_banco))
if plano_id is None:
plano_id = repositorio.registrar_plano(plano_de_dict(dados))
aplicadas = sum(1 for item in resultado.resultados if item.sucesso)
repositorio.registrar_aplicacao(
plano_id,
@@ -117,7 +125,11 @@ def _registrar_no_banco(caminho_do_plano: Path, resultado: object, sequencia: st
file=sys.stderr)
def executar(caminho_do_plano: Path) -> dict:
def executar(
caminho_do_plano: Path,
plano_id: int | None = None,
caminho_do_banco: Path = CAMINHO_DO_BANCO,
) -> dict:
"""Lê o plano em ``caminho_do_plano`` e o aplica na sequência ativa do Premiere.
Devolve um dicionário pronto para virar JSON, com o resultado de cada
@@ -154,7 +166,10 @@ def executar(caminho_do_plano: Path) -> dict:
)
resultado = aplicador.aplicar(plano)
_registrar_no_banco(caminho_do_plano, resultado, sequencia_ativa.nome)
_registrar_no_banco(
caminho_do_plano, resultado, sequencia_ativa.nome,
plano_id=plano_id, caminho_do_banco=caminho_do_banco,
)
return {
"arquivo_de_origem": plano.arquivo_de_origem,
@@ -175,10 +190,12 @@ def executar(caminho_do_plano: Path) -> dict:
def main() -> None:
"""Ponto de entrada da linha de comando."""
if len(sys.argv) != 2:
print("Uso: python aplicar_plano_de_edicao.py <caminho_do_plano.json>", file=sys.stderr)
sys.exit(2)
resultado = executar(Path(sys.argv[1]))
argumentos = argparse.ArgumentParser()
argumentos.add_argument("plano", type=Path)
argumentos.add_argument("--plano-id", type=int)
argumentos.add_argument("--banco", type=Path, default=CAMINHO_DO_BANCO)
opcoes = argumentos.parse_args()
resultado = executar(opcoes.plano, opcoes.plano_id, opcoes.banco)
print(json.dumps(resultado, ensure_ascii=False, indent=2))
if not resultado["todas_bem_sucedidas"]:
sys.exit(1)
+80
View File
@@ -0,0 +1,80 @@
"""Entrada de linha de comando para carregar um plano salvo no SQLite."""
from __future__ import annotations
import argparse
import json
from pathlib import Path
import sys
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))
from engine.persistencia.conexao import abrir_banco
from engine.persistencia.repositorio_de_planos import RepositorioDePlanos
def carregar(
caminho_do_banco: Path,
plano_id: int | None = None,
video_id: str | None = None,
) -> dict[str, object] | None:
"""Carrega um plano existente, preferindo o mais recente do vídeo informado."""
conexao = abrir_banco(caminho_do_banco)
try:
if plano_id is None:
if not video_id:
raise ValueError("Informe plano_id ou video_id para carregar um plano.")
linha = conexao.execute(
"""SELECT id FROM planos_de_edicao
WHERE video_id = ? ORDER BY criado_em DESC, id DESC LIMIT 1""",
(video_id,),
).fetchone()
if linha is None:
return None
plano_id = int(linha["id"])
plano = RepositorioDePlanos(conexao).carregar_plano(plano_id)
if plano is None:
return None
return {
"plano_id": plano_id,
"source": plano.origem,
"actions": [
{
"kind": acao.tipo,
"start": acao.inicio,
"end": acao.fim,
"reason": acao.motivo or "Decisão registrada no plano editorial.",
**acao.parametros,
}
for acao in plano.acoes
],
"video_id": plano.video_id,
"tipo_de_video": plano.tipo_de_video,
"modelo_da_ia": plano.modelo_da_ia,
"intencao": plano.intencao,
}
finally:
conexao.close()
def principal() -> None:
"""Executa o carregamento pela linha de comando."""
argumentos = argparse.ArgumentParser()
argumentos.add_argument("--banco", type=Path, default=Path(".jhonny/analises.db"))
argumentos.add_argument("--plano-id", type=int)
argumentos.add_argument("--video-id")
opcoes = argumentos.parse_args()
try:
plano = carregar(opcoes.banco, opcoes.plano_id, opcoes.video_id)
if plano is None:
raise ValueError("Nenhum plano de edição encontrado.")
print(json.dumps({"ok": True, "plano": plano}, ensure_ascii=False))
except (OSError, ValueError) as erro:
print(json.dumps({"ok": False, "erro": str(erro)}, ensure_ascii=False))
raise SystemExit(1) from erro
if __name__ == "__main__":
principal()
+8 -2
View File
@@ -1,5 +1,11 @@
"""Integrações locais para análise de áudio."""
"""Integrações locais para análise de áudio e música."""
from .analisador_de_metricas_de_voz import AnalisadorDeMetricasDeVoz
from .modelos_de_analise_musical import ResultadoDaAnaliseMusical
from .provider_de_analise_musical_essentia import AnalisadorDeMusicaInstrumentalEssentia
__all__ = ["AnalisadorDeMetricasDeVoz"]
__all__ = [
"AnalisadorDeMetricasDeVoz",
"AnalisadorDeMusicaInstrumentalEssentia",
"ResultadoDaAnaliseMusical",
]
@@ -0,0 +1,32 @@
"""
Contratos para provedores de análise musical.
O contrato permite que a aplicação use Essentia, um dublê de teste ou outro
provedor no futuro sem conhecer detalhes de carregamento e classificação.
"""
from pathlib import Path
from typing import Protocol
from .modelos_de_analise_musical import ResultadoDaAnaliseMusical
class AnalisadorDeMusica(Protocol):
"""Define a operação necessária para analisar um arquivo musical."""
def analisar(self, arquivo: str | Path) -> ResultadoDaAnaliseMusical:
"""
Analisa um arquivo de música e devolve seus descritores internos.
Parâmetros:
arquivo: Caminho de um arquivo de áudio existente.
Retorna:
Resultado estruturado da análise musical.
Pode gerar:
FileNotFoundError: quando o arquivo não existir.
RuntimeError: quando Essentia não estiver instalado.
"""
...
@@ -0,0 +1,35 @@
"""
Modelos internos para representar análises de músicas instrumentais.
Este módulo define somente os dados produzidos pela análise. Ele não importa
Essentia, não lê arquivos e não transforma o resultado em texto para a
interface.
"""
from dataclasses import dataclass, field
from pathlib import Path
@dataclass(frozen=True)
class ResultadoDaAnaliseMusical:
"""
Representa os descritores calculados para uma música instrumental.
A classe mantém valores objetivos, como BPM e tonalidade, além das
classificações fornecidas pelos modelos disponíveis. Ela não afirma uma
interpretação definitiva da intenção artística da música.
"""
caminho_do_arquivo: Path
duracao_em_segundos: float | None = None
batidas_por_minuto: float | None = None
tonalidade: str | None = None
modo: str | None = None
intensidade: float | None = None
dancabilidade: float | None = None
generos: tuple[str, ...] = field(default_factory=tuple)
humores: tuple[str, ...] = field(default_factory=tuple)
instrumentos: tuple[str, ...] = field(default_factory=tuple)
etiquetas: tuple[str, ...] = field(default_factory=tuple)
provedor: str = "Essentia"
modelo: str | None = None
@@ -0,0 +1,142 @@
"""
Adaptador local para análise musical usando Essentia.
O adaptador concentra a dependência externa e converte seus descritores para
``ResultadoDaAnaliseMusical``. Ele não persiste resultados nem conhece a
interface do painel. As classificações de gênero, humor e instrumentos são
lidas quando os modelos de alto nível instalados retornarem essas categorias.
"""
from pathlib import Path
from typing import Any, Callable
from .modelos_de_analise_musical import ResultadoDaAnaliseMusical
class AnalisadorDeMusicaInstrumentalEssentia:
"""
Analisa arquivos de áudio localmente por meio do ``MusicExtractor``.
A criação do extrator é tardia para que a engine continue importável sem
Essentia instalada. Um extrator compatível pode ser injetado nos testes,
evitando dependência de modelos pesados durante o teste unitário.
"""
def __init__(
self,
extrator: Callable[[str], tuple[Any, Any]] | None = None,
nome_do_modelo: str | None = None,
) -> None:
"""Inicializa o adaptador sem carregar a dependência externa."""
self._extrator = extrator
self._nome_do_modelo = nome_do_modelo
def analisar(self, arquivo: str | Path) -> ResultadoDaAnaliseMusical:
"""
Analisa um arquivo de música instrumental.
Parâmetros:
arquivo: Caminho de áudio que será analisado.
Retorna:
Descritores objetivos e classificações disponíveis.
Pode gerar:
FileNotFoundError: quando o caminho não apontar para um arquivo.
RuntimeError: quando Essentia não estiver instalada.
ValueError: quando o resultado externo não possuir formato válido.
"""
caminho = Path(arquivo).expanduser().resolve(strict=False)
if not caminho.is_file():
raise FileNotFoundError(f"Arquivo de áudio não encontrado: {caminho}")
dados, _ = self._obter_extrator()(str(caminho))
if dados is None:
raise ValueError("O Essentia devolveu uma análise vazia.")
return self._converter_resultado(caminho, dados)
def _obter_extrator(self) -> Callable[[str], tuple[Any, Any]]:
if self._extrator is not None:
return self._extrator
try:
from essentia.standard import MusicExtractor
except ImportError as erro:
raise RuntimeError(
"A análise musical local precisa da biblioteca Essentia. "
"Instale as dependências de requirements-audio.txt."
) from erro
self._extrator = MusicExtractor()
return self._extrator
def _converter_resultado(
self, caminho: Path, dados: Any,
) -> ResultadoDaAnaliseMusical:
return ResultadoDaAnaliseMusical(
caminho_do_arquivo=caminho,
duracao_em_segundos=self._numero(self._buscar(dados, "metadata.audio_properties.length")),
batidas_por_minuto=self._numero(self._buscar(dados, "rhythm.bpm")),
tonalidade=self._texto(self._buscar(dados, "tonal.key_edma.key")),
modo=self._texto(self._buscar(dados, "tonal.key_edma.scale")),
intensidade=self._numero(self._buscar(dados, "lowlevel.average_loudness")),
dancabilidade=self._numero(self._buscar(dados, "rhythm.danceability")),
generos=self._etiquetas(dados, ("genre", "genres")),
humores=self._etiquetas(dados, ("mood", "moods")),
instrumentos=self._etiquetas(dados, ("instrument", "instruments")),
etiquetas=self._etiquetas(dados, ("tags", "semantic")),
modelo=self._nome_do_modelo,
)
@staticmethod
def _buscar(dados: Any, caminho: str) -> Any:
"""Busca uma chave pontuada em um resultado Essentia ou devolve None.
O ``Pool`` do Essentia devolve descritores com a chave completa, como
``rhythm.bpm``. Dublês de teste e algumas integrações podem devolvê-los
como dicionários aninhados; os dois formatos são aceitos aqui.
"""
try:
return dados[caminho]
except (KeyError, IndexError, TypeError):
pass
valor = dados
for parte in caminho.split("."):
try:
valor = valor[parte]
except (KeyError, IndexError, TypeError):
return None
return valor
@classmethod
def _etiquetas(cls, dados: Any, nomes: tuple[str, ...]) -> tuple[str, ...]:
"""Converte classificações externas em uma tupla ordenada de textos."""
encontrados: list[str] = []
for nome in nomes:
valor = cls._buscar(dados, f"highlevel.{nome}.value")
if isinstance(valor, dict):
encontrados.extend(
str(chave) for chave, pontuacao in valor.items()
if cls._numero(pontuacao) is not None and float(pontuacao) > 0
)
elif isinstance(valor, (list, tuple)):
encontrados.extend(str(item) for item in valor)
elif isinstance(valor, str):
encontrados.append(valor)
return tuple(dict.fromkeys(encontrados))
@staticmethod
def _numero(valor: Any) -> float | None:
"""Converte valores numéricos externos sem aceitar booleanos."""
if isinstance(valor, bool) or valor is None:
return None
try:
return float(valor)
except (TypeError, ValueError):
return None
@staticmethod
def _texto(valor: Any) -> str | None:
"""Converte valores textuais externos, ignorando vazios."""
if valor is None or isinstance(valor, (dict, list, tuple)):
return None
texto = str(valor).strip()
return texto or None
+99
View File
@@ -135,6 +135,105 @@ class ConsultasDeAnalises:
},
}
def consultar_contexto_da_edicao(
self,
video_id: str | None = None,
incluir_palavras: bool = True,
) -> dict | None:
"""
Carrega a edição concluída mais recente e seu contexto editorial.
Quando ``video_id`` não é informado, seleciona a edição concluída mais
recente do banco. O retorno reúne a configuração editorial, o vídeo,
os clipes, as falas e o histórico resumido de planos, para que um
agente possa iniciar uma edição sem escrever SQL nem procurar arquivos
auxiliares.
Parâmetros:
video_id: Restringe a busca a um vídeo quando informado.
incluir_palavras: Inclui palavras da transcrição para decisões
que dependem de limites precisos.
Retorna:
Contexto completo da edição mais recente, ou None quando não há
edição concluída compatível.
Pode gerar:
ErroDeConsultaInvalida: quando ``video_id`` for vazio ou a
configuração persistida não for um JSON válido.
"""
if video_id is not None:
self._validar_texto_obrigatorio("video_id", video_id)
condicao = "WHERE video_id = ?" if video_id is not None else ""
parametros = (video_id,) if video_id is not None else ()
edicao = self.conexao.execute(
f"""SELECT id, video_id, sequencia, origem, tipo_de_video,
configuracao, concluida_em
FROM edicoes_de_video
{condicao}
ORDER BY concluida_em DESC, id DESC
LIMIT 1""",
parametros,
).fetchone()
if edicao is None:
return None
try:
configuracao = json.loads(edicao["configuracao"])
except (TypeError, json.JSONDecodeError) as erro:
raise ErroDeConsultaInvalida(
f"A configuração da edição {edicao['id']} não é um JSON válido."
) from erro
if not isinstance(configuracao, dict):
raise ErroDeConsultaInvalida(
f"A configuração da edição {edicao['id']} precisa ser um objeto."
)
identificador_do_video = edicao["video_id"]
clipes = [
dict(linha) for linha in self.conexao.execute(
"""SELECT id, faixa_id, nome, inicio_na_timeline, fim_na_timeline,
inicio_na_origem, fim_na_origem, arquivo, offline
FROM clipes
WHERE video_id = ?
ORDER BY inicio_na_timeline, id""",
(identificador_do_video,),
).fetchall()
]
planos = [
dict(linha) for linha in self.conexao.execute(
"""SELECT p.id, p.origem, p.tipo_de_video, p.modelo_da_ia,
p.intencao, p.criado_em,
COUNT(a.id) AS total_de_acoes,
(SELECT COUNT(*) FROM aplicacoes_do_plano ap
WHERE ap.plano_id = p.id AND ap.sucesso = 1)
AS aplicacoes_com_sucesso
FROM planos_de_edicao p
LEFT JOIN acoes_do_plano a ON a.plano_id = p.id
WHERE p.video_id = ?
GROUP BY p.id
ORDER BY p.criado_em DESC, p.id DESC""",
(identificador_do_video,),
).fetchall()
]
return {
"edicao": {
"id": edicao["id"],
"video_id": identificador_do_video,
"sequencia": edicao["sequencia"],
"origem": edicao["origem"],
"tipo_de_video": edicao["tipo_de_video"],
"configuracao": configuracao,
"concluida_em": edicao["concluida_em"],
},
"video": self.consultar_resumo_do_video(identificador_do_video),
"analise": self.consultar_status_da_analise(identificador_do_video),
"clipes": clipes,
"falas": self.listar_falas_no_intervalo(
identificador_do_video, incluir_palavras=incluir_palavras,
),
"planos": planos,
}
def listar_intervalos_de_preview_dos_falantes(self, video_id: str) -> list[dict]:
"""
Lista os intervalos dos falantes convertidos para o arquivo de vídeo.
+76
View File
@@ -0,0 +1,76 @@
"""Prepara o contexto da edição por voz a partir do banco SQLite."""
from __future__ import annotations
import argparse
import json
from pathlib import Path
import sys
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))
from engine.persistencia.consultas import ConsultasDeAnalises
from engine.persistencia.conexao import abrir_banco
def preparar_contexto(
caminho_do_banco: Path,
video_id: str | None = None,
incluir_palavras: bool = True,
) -> dict | None:
"""
Carrega a edição mais recente e seu contexto editorial.
Parâmetros:
caminho_do_banco: Caminho do banco SQLite de análises.
video_id: Vídeo específico; quando omitido, usa a edição mais recente.
incluir_palavras: Inclui os limites das palavras da transcrição.
Retorna:
Contexto serializável para o agente de edição, ou None quando não há
edição concluída compatível.
"""
conexao = abrir_banco(caminho_do_banco)
try:
return ConsultasDeAnalises(conexao).consultar_contexto_da_edicao(
video_id=video_id,
incluir_palavras=incluir_palavras,
)
finally:
conexao.close()
def principal() -> None:
"""Imprime o contexto editorial em JSON para consumo do agente."""
argumentos = argparse.ArgumentParser(
description="Carrega a edição por voz mais recente do banco SQLite."
)
argumentos.add_argument(
"--banco", type=Path, default=Path(".jhonny/analises.db"),
help="Caminho do banco SQLite de análises.",
)
argumentos.add_argument("--video-id", help="Restringe a consulta a um vídeo.")
argumentos.add_argument(
"--sem-palavras", action="store_true",
help="Omite os intervalos palavra a palavra para reduzir a resposta.",
)
opcoes = argumentos.parse_args()
try:
contexto = preparar_contexto(
opcoes.banco,
video_id=opcoes.video_id,
incluir_palavras=not opcoes.sem_palavras,
)
if contexto is None:
print(json.dumps({"ok": False, "erro": "Nenhuma edição concluída encontrada."}, ensure_ascii=False))
raise SystemExit(1)
print(json.dumps({"ok": True, "contexto": contexto}, ensure_ascii=False))
except (OSError, ValueError) as erro:
print(json.dumps({"ok": False, "erro": str(erro)}, ensure_ascii=False))
raise SystemExit(1) from erro
if __name__ == "__main__":
principal()
+77
View File
@@ -0,0 +1,77 @@
"""Entrada de linha de comando para salvar um plano de edição no SQLite."""
from __future__ import annotations
import argparse
import json
from pathlib import Path
import sys
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))
from engine.persistencia.conexao import abrir_banco
from engine.persistencia.repositorio_de_planos import RepositorioDePlanos, plano_de_dict
def registrar(
caminho_do_plano: Path,
caminho_do_banco: Path,
video_id: str | None = None,
tipo_de_video: str | None = None,
modelo_da_ia: str | None = None,
intencao: str | None = None,
) -> int:
"""Lê o JSON devolvido pela IA e registra o plano associado à edição mais recente."""
dados = json.loads(caminho_do_plano.read_text(encoding="utf-8"))
conexao = abrir_banco(caminho_do_banco)
try:
if video_id is None:
origem = dados.get("source")
linha = conexao.execute(
"""SELECT video_id FROM edicoes_de_video
WHERE origem = ? OR sequencia = ?
ORDER BY concluida_em DESC, id DESC LIMIT 1""",
(origem, origem),
).fetchone()
video_id = linha["video_id"] if linha is not None else None
plano = plano_de_dict(
dados,
video_id=video_id,
tipo_de_video=tipo_de_video,
modelo_da_ia=modelo_da_ia,
intencao=intencao,
)
return RepositorioDePlanos(conexao).registrar_plano(plano)
finally:
conexao.close()
def principal() -> None:
"""Executa o registro do plano pela linha de comando."""
argumentos = argparse.ArgumentParser()
argumentos.add_argument("plano", type=Path)
argumentos.add_argument("--banco", type=Path, default=Path(".jhonny/analises.db"))
argumentos.add_argument("--video-id")
argumentos.add_argument("--tipo-de-video")
argumentos.add_argument("--modelo-da-ia")
argumentos.add_argument("--intencao")
opcoes = argumentos.parse_args()
try:
identificador = registrar(
opcoes.plano,
opcoes.banco,
video_id=opcoes.video_id,
tipo_de_video=opcoes.tipo_de_video,
modelo_da_ia=opcoes.modelo_da_ia,
intencao=opcoes.intencao,
)
print(json.dumps({"ok": True, "plano_id": identificador}, ensure_ascii=False))
except (OSError, ValueError) as erro:
print(json.dumps({"ok": False, "erro": str(erro)}, ensure_ascii=False))
raise SystemExit(1) from erro
if __name__ == "__main__":
principal()
+2
View File
@@ -2,3 +2,5 @@
# a diarização e a detecção de emoção (integracoes/huggingface/provider_de_diarizacao_local.py
# e provider_de_emocao_local.py importam soundfile para ler o áudio extraído).
soundfile
# Dependência opcional carregada somente pelo adaptador de análise musical.
essentia
@@ -0,0 +1,56 @@
import tempfile
import unittest
from pathlib import Path
from engine.integracoes.audio import AnalisadorDeMusicaInstrumentalEssentia
class TesteAnalisadorDeMusicaInstrumentalEssentia(unittest.TestCase):
def test_converte_descritores_e_classificacoes(self):
def extrator(_caminho: str):
return {
"metadata": {"audio_properties": {"length": 123.4}},
"rhythm": {"bpm": 92.0, "danceability": 0.61},
"tonal": {"key_edma": {"key": "A", "scale": "minor"}},
"lowlevel": {"average_loudness": 0.35},
"highlevel": {
"genre": {"value": {"ambient": 0.9, "rock": 0.0}},
"mood": {"value": ["calm", "melancholic"]},
},
}, {}
with tempfile.TemporaryDirectory() as pasta:
arquivo = Path(pasta) / "instrumental.wav"
arquivo.write_bytes(b"audio")
resultado = AnalisadorDeMusicaInstrumentalEssentia(extrator).analisar(arquivo)
self.assertEqual(resultado.batidas_por_minuto, 92.0)
self.assertEqual(resultado.tonalidade, "A")
self.assertEqual(resultado.modo, "minor")
self.assertEqual(resultado.generos, ("ambient",))
self.assertEqual(resultado.humores, ("calm", "melancholic"))
def test_rejeita_arquivo_inexistente(self):
analisador = AnalisadorDeMusicaInstrumentalEssentia(lambda _: ({}, {}))
with self.assertRaises(FileNotFoundError):
analisador.analisar("/arquivo/inexistente.wav")
def test_aceita_descritores_com_chaves_pontuadas_do_essentia(self):
def extrator(_caminho: str):
return {
"metadata.audio_properties.length": 10.0,
"rhythm.bpm": 110.0,
"rhythm.danceability": 0.42,
"tonal.key_edma.key": "D",
"tonal.key_edma.scale": "major",
"lowlevel.average_loudness": 0.22,
}, {}
with tempfile.TemporaryDirectory() as pasta:
arquivo = Path(pasta) / "instrumental.wav"
arquivo.write_bytes(b"audio")
resultado = AnalisadorDeMusicaInstrumentalEssentia(extrator).analisar(arquivo)
self.assertEqual(resultado.duracao_em_segundos, 10.0)
self.assertEqual(resultado.batidas_por_minuto, 110.0)
self.assertEqual(resultado.tonalidade, "D")
@@ -0,0 +1,37 @@
"""Testes do carregamento de planos de edição diretamente do SQLite."""
import json
import tempfile
import unittest
from pathlib import Path
from engine.carregar_plano_de_edicao import carregar
from engine.persistencia import abrir_banco
from engine.persistencia.repositorio_de_planos import RepositorioDePlanos, plano_de_dict
class TesteCarregarPlanoDeEdicao(unittest.TestCase):
"""Garante que a interface receba o plano no contrato de aplicação."""
def test_carrega_plano_mais_recente_do_video(self):
with tempfile.TemporaryDirectory() as pasta:
banco = Path(pasta) / "analises.db"
conexao = abrir_banco(banco)
repositorio = RepositorioDePlanos(conexao)
plano_id = repositorio.registrar_plano(plano_de_dict(
{"source": "Sequência", "actions": [
{"kind": "cut", "start": 1, "end": 2, "reason": "repetição"},
]}, video_id="video-1", tipo_de_video="depoimento",
))
conexao.close()
carregado = carregar(banco, video_id="video-1")
self.assertEqual(carregado["plano_id"], plano_id)
self.assertEqual(carregado["video_id"], "video-1")
self.assertEqual(carregado["actions"][0]["kind"], "cut")
self.assertEqual(carregado["actions"][0]["reason"], "repetição")
if __name__ == "__main__":
unittest.main()
@@ -15,7 +15,9 @@ from engine.dominio import Clipe, Faixa, IntervaloDeTempo, Timeline
from engine.persistencia import (
ConsultasDeAnalises,
ErroDeConsultaInvalida,
EdicaoDeVideo,
RepositorioDeAnalisesSQLite,
RepositorioDeEdicoes,
RepositorioDeRetakesSQLite,
RepositorioDeTimelineSQLite,
)
@@ -95,6 +97,24 @@ class TesteConsultasDeAnalises(unittest.TestCase):
with self.assertRaises(ErroDeConsultaInvalida):
self.consultas.consultar_resumo_do_video("")
def test_consultar_contexto_da_edicao_carrega_edicao_mais_recente_e_falas(self) -> None:
RepositorioDeEdicoes(self.consultas.conexao).registrar(EdicaoDeVideo(
video_id=VIDEO_ID,
sequencia="Sequência",
origem="/videos/entrada.mp4",
tipo_de_video="depoimento",
configuracao={"objetivo": "Gerar confiança"},
))
contexto = self.consultas.consultar_contexto_da_edicao()
self.assertIsNotNone(contexto)
self.assertEqual(contexto["edicao"]["tipo_de_video"], "depoimento")
self.assertEqual(contexto["edicao"]["configuracao"]["objetivo"], "Gerar confiança")
self.assertEqual(contexto["video"]["video_id"], VIDEO_ID)
self.assertEqual(len(contexto["falas"]), 1)
self.assertEqual(contexto["falas"][0]["texto"], "Hoje nós vamos mostrar como funciona o sistema.")
def test_consultar_status_da_analise_expoe_falantes_e_metricas(self) -> None:
status = self.consultas.consultar_status_da_analise(VIDEO_ID)
@@ -0,0 +1,43 @@
"""Testes da entrada de planos de edição no banco SQLite."""
import json
import tempfile
import unittest
from pathlib import Path
from engine.persistencia import EdicaoDeVideo, RepositorioDeEdicoes, abrir_banco
from engine.registrar_plano_de_edicao import registrar
class TesteRegistrarPlanoDeEdicao(unittest.TestCase):
"""Garante que o plano é associado ao contexto editorial mais recente."""
def test_registra_plano_e_descobre_video_pela_origem(self):
with tempfile.TemporaryDirectory() as pasta:
banco = Path(pasta) / "analises.db"
conexao = abrir_banco(banco)
RepositorioDeEdicoes(conexao).registrar(EdicaoDeVideo(
video_id="video-1", sequencia="Depoimento", origem="/v.mp4",
tipo_de_video="depoimento", configuracao={"legendas": True},
))
conexao.close()
arquivo = Path(pasta) / "plano.json"
arquivo.write_text(json.dumps({
"source": "/v.mp4",
"actions": [{"kind": "cut", "start": 1, "end": 2}],
}), encoding="utf-8")
plano_id = registrar(arquivo, banco, tipo_de_video="depoimento")
conexao = abrir_banco(banco)
linha = conexao.execute(
"SELECT video_id, tipo_de_video FROM planos_de_edicao WHERE id = ?",
(plano_id,),
).fetchone()
conexao.close()
self.assertEqual(dict(linha), {"video_id": "video-1", "tipo_de_video": "depoimento"})
if __name__ == "__main__":
unittest.main()