João Henrique
400c3c93dd
- destacado no deploy o resumo que será gravado na mensagem do commit Resumo: - 8 arquivos alterados - 1 novos - 7 modificados - 0 removidos 7 files changed, 143 insertions(+), 11 deletions(-) Arquivos: - .jhonny/analises.db - admin/deploy.command - code/engine/executar_scanner.py - code/engine/integracoes/midia/__init__.py - code/engine/integracoes/visual/apple_vision.py - code/engine/scanner/configuracao_visual.py - code/engine/scanner/visual.py - code/engine/integracoes/midia/compactador_de_proxy.py
254 lines
12 KiB
Python
254 lines
12 KiB
Python
import threading
|
|
from collections.abc import Callable
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from dataclasses import dataclass, field
|
|
|
|
from ..dominio import Clipe
|
|
from ..integracoes.visual.contratos import (AnalisadorDeFrame, AnalisadorDeSequenciaDeQuadros,
|
|
DetectorDeIntervalosDeCena, ExtratorDeQuadros)
|
|
from ..integracoes.visual.modelos import QuadroDeVideo
|
|
from .modelos import Cena, CenaVisual, EvidenciaVisual
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ResultadoVisualDoClipe:
|
|
identificador_do_clipe: str
|
|
evidencias: list[EvidenciaVisual] = field(default_factory=list)
|
|
cenas: list[Cena] = field(default_factory=list)
|
|
cenas_visuais: list[CenaVisual] = field(default_factory=list)
|
|
|
|
|
|
class DetectorDeCenas:
|
|
"""Módulo profundo que orquestra frames, analisadores e cenas de um clipe.
|
|
|
|
A interface não expõe OpenCV, ONNX, MediaPipe, PySceneDetect ou Vision.
|
|
Cada adapter pode ser trocado sem alterar chamadores nem resultados do domínio.
|
|
"""
|
|
|
|
def __init__(self, extrator: ExtratorDeQuadros,
|
|
analisadores_de_frame: list[AnalisadorDeFrame] | None = None,
|
|
detectores_de_intervalo: list[DetectorDeIntervalosDeCena] | None = None,
|
|
analisadores_de_sequencia: list[AnalisadorDeSequenciaDeQuadros] | None = None,
|
|
ao_progresso: Callable[[int, int], None] | None = None,
|
|
trabalhadores_em_paralelo: int = 4) -> None:
|
|
self.extrator = extrator
|
|
self.analisadores_de_frame = analisadores_de_frame or []
|
|
self.analisadores_de_sequencia = analisadores_de_sequencia or []
|
|
self.detectores_de_intervalo = detectores_de_intervalo or []
|
|
self.ao_progresso = ao_progresso
|
|
self.trabalhadores_em_paralelo = max(1, trabalhadores_em_paralelo)
|
|
|
|
def detectar(self, clipe: Clipe) -> ResultadoVisualDoClipe:
|
|
quadros = self.extrator.extrair(clipe)
|
|
self._preparar_analisadores(quadros)
|
|
total = len(quadros) * len(self.analisadores_de_frame) + len(self.analisadores_de_sequencia)
|
|
progresso = [0]
|
|
evidencias = self._analisar_quadros(quadros, progresso, total) + self._analisar_sequencia(quadros, progresso, total)
|
|
cenas = self._detectar_intervalos(clipe, quadros)
|
|
cenas_visuais = [CenaVisual(
|
|
clipe.identificador, cena.inicio, cena.fim,
|
|
tuple(item for item in evidencias if item.inicio <= cena.fim and item.fim >= cena.inicio),
|
|
cena.referencias,
|
|
) for cena in cenas]
|
|
return ResultadoVisualDoClipe(clipe.identificador, evidencias, cenas, cenas_visuais)
|
|
|
|
def _preparar_analisadores(self, quadros: list[QuadroDeVideo]) -> None:
|
|
"""Prepara analisadores que precisam saber a quantidade de frames do clipe.
|
|
|
|
Caso especial do `AnalisadorAppleVisionNativo`: ele distribui a
|
|
interpretação editorial (Apple Intelligence / Foundation Models) em um
|
|
subconjunto dos frames para manter a cobertura de cena com custo menor.
|
|
"""
|
|
quantidade_de_quadros = len(quadros)
|
|
for analisador in self.analisadores_de_frame:
|
|
preparar = getattr(analisador, "preparar_interpretacao_por_clipe", None)
|
|
if callable(preparar):
|
|
preparar(quantidade_de_quadros)
|
|
|
|
def _analisar_quadros(self, quadros: list[QuadroDeVideo], progresso: list[int] | None = None,
|
|
total: int = 0) -> list[EvidenciaVisual]:
|
|
if not self.analisadores_de_frame or not quadros:
|
|
return []
|
|
tarefas = [(quadro, analisador) for quadro in quadros for analisador in self.analisadores_de_frame]
|
|
travamento = threading.Lock()
|
|
|
|
def processar(tarefa: tuple[QuadroDeVideo, AnalisadorDeFrame]) -> list[EvidenciaVisual]:
|
|
quadro, analisador = tarefa
|
|
evidencias = analisador.analisar(quadro)
|
|
with travamento:
|
|
self._avancar_progresso(progresso, total)
|
|
return evidencias
|
|
|
|
if self.trabalhadores_em_paralelo == 1 or len(tarefas) == 1:
|
|
resultados = [processar(tarefa) for tarefa in tarefas]
|
|
else:
|
|
with ThreadPoolExecutor(max_workers=self.trabalhadores_em_paralelo) as executor:
|
|
resultados = list(executor.map(processar, tarefas))
|
|
return [evidencia for lista in resultados for evidencia in lista]
|
|
|
|
def _analisar_sequencia(self, quadros: list[QuadroDeVideo], progresso: list[int] | None = None,
|
|
total: int = 0) -> list[EvidenciaVisual]:
|
|
resultado: list[EvidenciaVisual] = []
|
|
for analisador in self.analisadores_de_sequencia:
|
|
resultado.extend(analisador.analisar(quadros))
|
|
self._avancar_progresso(progresso, total)
|
|
return resultado
|
|
|
|
def _avancar_progresso(self, progresso: list[int] | None, total: int) -> None:
|
|
if progresso is None or total <= 0:
|
|
return
|
|
progresso[0] += 1
|
|
if self.ao_progresso:
|
|
self.ao_progresso(progresso[0], total)
|
|
|
|
def _detectar_intervalos(self, clipe: Clipe, quadros: list[QuadroDeVideo]) -> list[Cena]:
|
|
cenas: list[Cena] = []
|
|
for detector in self.detectores_de_intervalo:
|
|
cenas.extend(detector.detectar(clipe, quadros))
|
|
cenas = self._consolidar_cenas(cenas)
|
|
if not cenas and quadros:
|
|
cenas = [Cena(quadros[0].timestamp, quadros[-1].timestamp)]
|
|
return cenas
|
|
|
|
@staticmethod
|
|
def _consolidar_cenas(cenas: list[Cena]) -> list[Cena]:
|
|
resultado: list[Cena] = []
|
|
for cena in sorted(cenas, key=lambda item: (item.inicio, item.fim)):
|
|
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,
|
|
anterior.referencias + cena.referencias)
|
|
else:
|
|
resultado.append(cena)
|
|
return resultado
|
|
|
|
|
|
class AnaliseVisualDaTimeline:
|
|
"""Etapa do Scanner que guarda evidências e cenas por clipe no contexto."""
|
|
|
|
nome = "analise_visual_local"
|
|
|
|
def __init__(self, detector: DetectorDeCenas, continuar_em_falha: bool = True) -> None:
|
|
self.detector = detector
|
|
self.continuar_em_falha = continuar_em_falha
|
|
|
|
def executar(self, contexto):
|
|
if contexto.timeline is None:
|
|
return contexto
|
|
for faixa in contexto.timeline.faixas:
|
|
for clipe in faixa.clipes:
|
|
if clipe.offline or not clipe.arquivo:
|
|
continue
|
|
try:
|
|
resultado = self.detector.detectar(clipe)
|
|
except Exception as exc:
|
|
if not self.continuar_em_falha:
|
|
raise
|
|
contexto.avisos.append(
|
|
f"analise_visual_indisponivel: clipe {clipe.identificador}: {exc}"
|
|
)
|
|
contexto.caracteristicas_visuais[clipe.identificador] = {
|
|
"evidencias": [], "cenas": [], "cenas_visuais": [], "disponivel": False,
|
|
}
|
|
continue
|
|
contexto.caracteristicas_visuais[clipe.identificador] = {
|
|
"evidencias": resultado.evidencias,
|
|
"cenas": resultado.cenas,
|
|
"cenas_visuais": resultado.cenas_visuais,
|
|
"disponivel": True,
|
|
}
|
|
contexto.cenas.extend(resultado.cenas)
|
|
contexto.cenas_visuais.extend(resultado.cenas_visuais)
|
|
contexto.cenas = DetectorDeCenas._consolidar_cenas(contexto.cenas)
|
|
return contexto
|
|
|
|
|
|
def _detectores_de_cena_do_perfil(perfil: ConfiguracaoVisualDoScanner) -> list:
|
|
"""Monta os detectores de intervalo de cena ativos para o perfil.
|
|
|
|
A análise visual auxiliar (PySceneDetect) é opcional. Só entra como um
|
|
``DetectorDeIntervalosDeCena`` quando a configuração a ativa — e nunca
|
|
decide sozinho se há retake.
|
|
"""
|
|
from ..integracoes.visual import DetectorDeCenasPySceneDetect
|
|
|
|
detectores: list = []
|
|
if getattr(perfil, "usar_analise_visual", False) and getattr(perfil, "detectar_cortes", True):
|
|
detectores.append(DetectorDeCenasPySceneDetect(limiar=perfil.limiar_de_corte))
|
|
return detectores
|
|
|
|
|
|
def criar_detector_visual_local(
|
|
diretorio_de_cache: str | None = ".frames",
|
|
intervalo_em_segundos: float = 1.0,
|
|
configuracao: ConfiguracaoVisualDoScanner | None = None,
|
|
) -> DetectorDeCenas:
|
|
"""Monta o detector local padrão com uma interface curta para o Scanner."""
|
|
from .configuracao_visual import ConfiguracaoVisualDoScanner
|
|
from ..integracoes.visual import (
|
|
AnalisadorDeComposicaoOpenCV,
|
|
AnalisadorDeContinuidadeOpenCV,
|
|
AnalisadorDeQualidadeOpenCV,
|
|
ExtratorDeQuadrosOpenCV,
|
|
)
|
|
|
|
perfil = configuracao or ConfiguracaoVisualDoScanner(intervalo_em_segundos=intervalo_em_segundos)
|
|
|
|
return DetectorDeCenas(
|
|
ExtratorDeQuadrosOpenCV(
|
|
intervalo_em_segundos=perfil.intervalo_em_segundos,
|
|
diretorio_de_cache=diretorio_de_cache,
|
|
),
|
|
analisadores_de_frame=[
|
|
AnalisadorDeQualidadeOpenCV(),
|
|
AnalisadorDeComposicaoOpenCV(),
|
|
],
|
|
analisadores_de_sequencia=[AnalisadorDeContinuidadeOpenCV()],
|
|
detectores_de_intervalo=_detectores_de_cena_do_perfil(perfil),
|
|
)
|
|
|
|
|
|
def criar_detector_apple_vision(
|
|
diretorio_de_cache: str | None = ".frames",
|
|
intervalo_em_segundos: float = 1.0,
|
|
configuracao=None,
|
|
) -> DetectorDeCenas:
|
|
"""Monta a análise local de cena baseada em Vision e arquivos de origem."""
|
|
from .configuracao_visual import ConfiguracaoVisualDoScanner
|
|
from ..integracoes.visual import (AnalisadorAppleVisionNativo,
|
|
AnalisadorSequenciaAppleVisionNativo,
|
|
ExtratorDeQuadrosFFmpeg)
|
|
from ..integracoes.midia import CompactadorDeProxy, ExtratorDeQuadrosDoProxy
|
|
|
|
perfil = configuracao or ConfiguracaoVisualDoScanner(intervalo_em_segundos=intervalo_em_segundos)
|
|
|
|
# Quantidade fixa de trabalhadores para extração e análise de frames.
|
|
# Em volumes externos (ex.: /Volumes/Merongo), mais threads do que isso
|
|
# satura a E/S e trava os processos de mídia a ~0% de CPU.
|
|
QUANTIDADE_DE_TRABALHADORES = 4
|
|
|
|
extrator = ExtratorDeQuadrosFFmpeg(intervalo_em_segundos=perfil.intervalo_em_segundos,
|
|
diretorio_de_cache=diretorio_de_cache or ".frames",
|
|
resolucao_maxima=perfil.resolucao_maxima,
|
|
trabalhadores_em_paralelo=QUANTIDADE_DE_TRABALHADORES)
|
|
# Leitura via proxy compactado: para arquivos pesados, evita buscar frame
|
|
# por frame no arquivo original (um -ss por frame num MOV de vários GB).
|
|
if perfil.usar_proxy:
|
|
extrator = ExtratorDeQuadrosDoProxy(
|
|
extrator, CompactadorDeProxy(), largura_maxima=perfil.resolucao_do_proxy
|
|
)
|
|
|
|
return DetectorDeCenas(
|
|
extrator,
|
|
analisadores_de_frame=[AnalisadorAppleVisionNativo(
|
|
recursos=tuple(sorted(perfil.recursos_vision)),
|
|
interpretar_com_apple_intelligence=perfil.interpretar_cena,
|
|
maximo_de_frames_interpretados=perfil.maximo_de_frames_interpretados,
|
|
)],
|
|
analisadores_de_sequencia=[AnalisadorSequenciaAppleVisionNativo()]
|
|
if perfil.analisar_continuidade else [],
|
|
detectores_de_intervalo=_detectores_de_cena_do_perfil(perfil),
|
|
trabalhadores_em_paralelo=QUANTIDADE_DE_TRABALHADORES,
|
|
)
|