feat: initial commit - Jhonny Editor
- Adicionado estrutura completa do projeto - Configurado MCP server para Premiere Pro - Adicionado documentação e skills - Configurado Gitignore para o projeto
This commit is contained in:
@@ -0,0 +1,7 @@
|
||||
from .coordenacao import AnalisadorDeTimeline, ContextoDeAnalise, PipelineDoScanner
|
||||
|
||||
__all__ = ["AnalisadorDeTimeline", "ContextoDeAnalise", "PipelineDoScanner"]
|
||||
from .modelos import Cena, ErroDeAnalise, Evento, ResultadoDaAnalise, SegmentoDeTranscricao, StatusDaAnalise
|
||||
from .transcricao_da_timeline import TranscricaoDaTimeline, TranscricaoDoClipe
|
||||
|
||||
__all__ = ["Cena", "ErroDeAnalise", "Evento", "ResultadoDaAnalise", "SegmentoDeTranscricao", "StatusDaAnalise", "TranscricaoDaTimeline", "TranscricaoDoClipe"]
|
||||
@@ -0,0 +1,67 @@
|
||||
from typing import Any, Protocol
|
||||
|
||||
from .modelos import Cena, SegmentoDeTranscricao
|
||||
|
||||
|
||||
class ProviderDeTranscricao(Protocol):
|
||||
def transcrever(self, clipe: Any) -> list[SegmentoDeTranscricao]: ...
|
||||
|
||||
|
||||
class ProviderDeAnaliseVisual(Protocol):
|
||||
def analisar(self, clipe: Any, quadros: list[Any] | None = None) -> dict[str, Any]: ...
|
||||
|
||||
|
||||
class TranscricaoDeAudio:
|
||||
nome = "transcricao_de_audio"
|
||||
|
||||
def __init__(self, provider: ProviderDeTranscricao) -> None:
|
||||
self.provider = provider
|
||||
|
||||
def executar(self, contexto):
|
||||
if contexto.timeline is None:
|
||||
return contexto
|
||||
for faixa in contexto.timeline.faixas:
|
||||
for clipe in faixa.clipes:
|
||||
contexto.transcricoes[clipe.identificador] = self.provider.transcrever(clipe)
|
||||
return contexto
|
||||
|
||||
|
||||
class AnaliseVisual:
|
||||
nome = "analise_visual"
|
||||
|
||||
def __init__(self, provider: ProviderDeAnaliseVisual) -> None:
|
||||
self.provider = provider
|
||||
|
||||
def executar(self, contexto):
|
||||
if contexto.timeline is None:
|
||||
return contexto
|
||||
for faixa in contexto.timeline.faixas:
|
||||
for clipe in faixa.clipes:
|
||||
contexto.caracteristicas_visuais[clipe.identificador] = self.provider.analisar(clipe)
|
||||
return contexto
|
||||
|
||||
|
||||
class DeteccaoDeCenas:
|
||||
nome = "deteccao_de_cenas"
|
||||
|
||||
def executar(self, contexto):
|
||||
cenas: list[Cena] = []
|
||||
for faixa in (contexto.timeline.faixas if contexto.timeline else []):
|
||||
for clipe in faixa.clipes:
|
||||
cenas.append(Cena(clipe.intervalo_na_timeline.inicio,
|
||||
clipe.intervalo_na_timeline.fim,
|
||||
contexto.caracteristicas_visuais.get(clipe.identificador, {}).get("confianca")))
|
||||
contexto.cenas = self._consolidar(cenas)
|
||||
return contexto
|
||||
|
||||
@staticmethod
|
||||
def _consolidar(cenas: list[Cena]) -> list[Cena]:
|
||||
resultado: list[Cena] = []
|
||||
for cena in sorted(cenas, key=lambda item: item.inicio):
|
||||
if resultado and cena.inicio <= resultado[-1].fim:
|
||||
anterior = resultado[-1]
|
||||
resultado[-1] = Cena(anterior.inicio, max(anterior.fim, cena.fim),
|
||||
anterior.confianca or cena.confianca)
|
||||
else:
|
||||
resultado.append(cena)
|
||||
return resultado
|
||||
@@ -0,0 +1,46 @@
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any, Protocol
|
||||
|
||||
from ...dominio import Timeline
|
||||
|
||||
|
||||
@dataclass
|
||||
class ContextoDeAnalise:
|
||||
timeline: Timeline | None = None
|
||||
dados_da_timeline: dict[str, Any] | None = None
|
||||
erros: list[str] = field(default_factory=list)
|
||||
avisos: list[str] = field(default_factory=list)
|
||||
analises_concluidas: list[str] = field(default_factory=list)
|
||||
transcricoes: dict[str, list[Any]] = field(default_factory=dict)
|
||||
caracteristicas_visuais: dict[str, dict[str, Any]] = field(default_factory=dict)
|
||||
cenas: list[Any] = field(default_factory=list)
|
||||
eventos: list[Any] = field(default_factory=list)
|
||||
|
||||
|
||||
class Analisador(Protocol):
|
||||
nome: str
|
||||
|
||||
def executar(self, contexto: ContextoDeAnalise) -> ContextoDeAnalise: ...
|
||||
|
||||
|
||||
class PipelineDoScanner:
|
||||
"""Executa análises em ordem, sem conhecer sua implementação."""
|
||||
|
||||
def __init__(self, analises: list[Analisador]) -> None:
|
||||
self.analises = analises
|
||||
|
||||
def executar(self, contexto: ContextoDeAnalise) -> ContextoDeAnalise:
|
||||
for analise in self.analises:
|
||||
contexto = analise.executar(contexto)
|
||||
contexto.analises_concluidas.append(analise.nome)
|
||||
return contexto
|
||||
|
||||
|
||||
class AnalisadorDeTimeline:
|
||||
"""Ponto de entrada da análise da timeline."""
|
||||
|
||||
def __init__(self, pipeline: PipelineDoScanner) -> None:
|
||||
self.pipeline = pipeline
|
||||
|
||||
def analisar(self) -> ContextoDeAnalise:
|
||||
return self.pipeline.executar(ContextoDeAnalise())
|
||||
@@ -0,0 +1,3 @@
|
||||
from .descoberta_da_timeline import DescobertaDaTimeline
|
||||
|
||||
__all__ = ["DescobertaDaTimeline"]
|
||||
@@ -0,0 +1,19 @@
|
||||
from ...integracoes.premiere.conversores import ConversorDeTimeline
|
||||
from ...integracoes.premiere.leitura import AcessoAoEditor
|
||||
from ..coordenacao import ContextoDeAnalise
|
||||
|
||||
|
||||
class DescobertaDaTimeline:
|
||||
"""Descobre e converte a timeline ativa do editor."""
|
||||
|
||||
nome = "descoberta_da_timeline"
|
||||
|
||||
def __init__(self, acesso_ao_editor: AcessoAoEditor, conversor: ConversorDeTimeline) -> None:
|
||||
self.acesso_ao_editor = acesso_ao_editor
|
||||
self.conversor = conversor
|
||||
|
||||
def executar(self, contexto: ContextoDeAnalise) -> ContextoDeAnalise:
|
||||
dados_brutos = self.acesso_ao_editor.obter_timeline_ativa()
|
||||
contexto.dados_da_timeline = dados_brutos
|
||||
contexto.timeline = self.conversor.converter(dados_brutos)
|
||||
return contexto
|
||||
@@ -0,0 +1,56 @@
|
||||
from dataclasses import dataclass, field
|
||||
from enum import Enum
|
||||
from typing import Any
|
||||
|
||||
|
||||
class StatusDaAnalise(str, Enum):
|
||||
CONCLUIDA = "concluida"
|
||||
CONCLUIDA_COM_AVISOS = "concluida_com_avisos"
|
||||
FALHOU = "falhou"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ErroDeAnalise(Exception):
|
||||
codigo: str
|
||||
mensagem: str
|
||||
caminho: str | None = None
|
||||
detalhes: dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
Exception.__init__(self, self.mensagem)
|
||||
|
||||
|
||||
@dataclass
|
||||
class ResultadoDaAnalise:
|
||||
status: StatusDaAnalise
|
||||
timeline: Any = None
|
||||
erros: list[ErroDeAnalise] = field(default_factory=list)
|
||||
avisos: list[str] = field(default_factory=list)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SegmentoDeTranscricao:
|
||||
inicio: float
|
||||
fim: float
|
||||
texto: str
|
||||
confianca: float | None = None
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if self.inicio < 0 or self.fim < self.inicio:
|
||||
raise ValueError("Intervalo de transcrição inválido.")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Cena:
|
||||
inicio: float
|
||||
fim: float
|
||||
confianca: float | None = None
|
||||
referencias: tuple[float, ...] = ()
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Evento:
|
||||
tipo: str
|
||||
inicio: float
|
||||
fim: float
|
||||
confianca: float | None = None
|
||||
@@ -0,0 +1,63 @@
|
||||
from dataclasses import asdict, dataclass
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any, Callable
|
||||
|
||||
from ..integracoes.midia import ExtracaoDeAudio
|
||||
from .modelos import SegmentoDeTranscricao
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class TranscricaoDoClipe:
|
||||
identificador_do_clipe: str
|
||||
arquivo: str
|
||||
inicio_na_timeline: float
|
||||
fim_na_timeline: float
|
||||
segmentos: list[SegmentoDeTranscricao]
|
||||
|
||||
@property
|
||||
def texto(self) -> str:
|
||||
return " ".join(s.texto for s in self.segmentos if s.texto).strip()
|
||||
|
||||
|
||||
class TranscricaoDaTimeline:
|
||||
"""Transcreve somente os intervalos efetivamente usados na timeline."""
|
||||
|
||||
def __init__(self, provider: Any, extrator: ExtracaoDeAudio | None = None,
|
||||
diretoria_de_trabalho: str | Path = ".transcricao") -> None:
|
||||
self.provider = provider
|
||||
self.extrator = extrator or ExtracaoDeAudio()
|
||||
self.diretoria_de_trabalho = Path(diretoria_de_trabalho)
|
||||
|
||||
def executar(self, timeline: Any) -> list[TranscricaoDoClipe]:
|
||||
resultados: list[TranscricaoDoClipe] = []
|
||||
for faixa in timeline.faixas:
|
||||
for clipe in faixa.clipes:
|
||||
if not clipe.arquivo or clipe.offline:
|
||||
continue
|
||||
intervalo = clipe.intervalo_na_timeline
|
||||
pasta = self.diretoria_de_trabalho / clipe.identificador
|
||||
origem = clipe.intervalo_na_origem
|
||||
partes = self.extrator.extrair_intervalo(
|
||||
clipe.arquivo, pasta, origem.inicio if origem else 0.0,
|
||||
origem.fim if origem else None,
|
||||
)
|
||||
segmentos: list[SegmentoDeTranscricao] = []
|
||||
for parte in partes:
|
||||
for segmento in self.provider.transcrever(type("Audio", (), {"arquivo": str(parte.caminho)})()):
|
||||
base_origem = origem.inicio if origem else 0.0
|
||||
inicio = max(intervalo.inicio, intervalo.inicio + (parte.inicio - base_origem) + segmento.inicio)
|
||||
fim = min(intervalo.fim, intervalo.inicio + (parte.inicio - base_origem) + segmento.fim)
|
||||
if inicio < intervalo.fim and fim > inicio:
|
||||
segmentos.append(SegmentoDeTranscricao(inicio, fim, segmento.texto, segmento.confianca))
|
||||
resultados.append(TranscricaoDoClipe(clipe.identificador, clipe.arquivo,
|
||||
intervalo.inicio, intervalo.fim, segmentos))
|
||||
return resultados
|
||||
|
||||
@staticmethod
|
||||
def gerar_relatorio(resultados: list[TranscricaoDoClipe], destino: str | Path) -> Path:
|
||||
caminho = Path(destino)
|
||||
dados = [{**asdict(item), "segmentos": [asdict(s) for s in item.segmentos], "texto": item.texto}
|
||||
for item in resultados]
|
||||
caminho.write_text(json.dumps(dados, ensure_ascii=False, indent=2), encoding="utf-8")
|
||||
return caminho
|
||||
Reference in New Issue
Block a user