feat: criado repositório jhonny-editor no Gitea

criado repositório jhonny-editor no Gitea
adicionado script admin/deploy.command com commit automático
atualizado admin/DEV-NOTES.md com template limpo

Resumo:
- 32 arquivos alterados
- 16 novos
- 15 modificados
- 0 removidos

 16 files changed, 638 insertions(+), 61 deletions(-)

Arquivos:
  - code/engine/skills/boas-praticas-oo.md -> .agents/skills/boas-praticas-oo/SKILL.md
  - AGENTS.md
  - code/cep-plugin/index.html
  - code/cep-plugin/main.js
  - code/cep-plugin/styles.css
  - code/engine/integracoes/visual/apple_vision.py
  - code/engine/integracoes/visual/apple_vision_runner.swift
  - code/engine/integracoes/visual/extrator_ffmpeg.py
  - code/engine/scanner/__init__.py
  - code/engine/scanner/configuracao_visual.py
  - code/engine/scanner/coordenacao/__init__.py
  - code/engine/scanner/retakes/adaptador_de_falas.py
  - code/engine/scanner/retakes/agrupador_de_takes.py
  - code/engine/scanner/retakes/detector_de_reinicios.py
  - code/engine/scanner/retakes/modelos_de_retakes.py
  - code/engine/scanner/visual.py
  - .jhonny/
  - .retakes/
  - CODING_STANDARDS.md
  - code/engine/executar_retakes.py
  - code/engine/persistencia/
  - code/engine/scanner/retakes/__init__.py
  - code/engine/scanner/retakes/carregador_de_artefatos.py
  - code/engine/scanner/retakes/classificador_de_retakes.py
  - code/engine/scanner/retakes/coordenador_de_retakes.py
  - code/engine/scanner/retakes/deteccao_de_retakes.py
  - code/engine/scanner/retakes/gerador_de_evidencias.py
  - code/engine/scanner/retakes/repositorio_de_retakes.py
  - code/engine/testes/test_consultas_de_persistencia.py
  - code/engine/testes/test_deteccao_de_retakes.py
  - code/engine/testes/test_persistencia_sqlite.py
  - code/relatorios/audio/arquivos/92f97877259a86a8095b1eecafdefe357212a12247d04b78df60e0c0ae38b2ea/
This commit is contained in:
João Henrique
2026-09-08 17:13:19 -04:00
parent b9bf3b2863
commit ee181360c8
40 changed files with 3674 additions and 61 deletions
+92
View File
@@ -0,0 +1,92 @@
"""Submódulo de detecção de retakes do scanner.
Analisa a transcrição já existente para identificar possíveis retakes —
repetições, reinícios e tomadas alternativas da mesma fala — gerando dados
estruturados para a revisão na interface. Não corta, não apaga e não decide
qual tomada usar.
"""
from .agrupador_de_takes import AgrupadorDeTakes
from .analisador_de_intervalos import AnalisadorDeIntervalos
from .classificador_de_retakes import (
CONTINUIDADE_DE_FALA,
POSSIVEL_RETAKE,
REPETICAO_DE_FRASE,
REPETICAO_DE_INICIO,
REPETICAO_PARCIAL,
RETAKE_CONFIRMADO,
SEM_INDICIO,
ClassificadorDeRetakes,
RegrasDeClassificacao,
)
from .carregador_de_artefatos import (
ArtefatoDeTranscricaoInvalido,
CarregadorDeArtefatosDeAudio,
)
from .comparador_de_falas import ComparadorDeFalas
from .comparador_semantico import ComparadorSemantico
from .coordenador_de_retakes import CoordenadorDeRetakes, ResultadoDaAnaliseDeRetakes
from .deteccao_de_retakes import DeteccaoDeRetakes
from .detector_de_reinicios import DetectorDeReinicios
from .gerador_de_evidencias import GeradorDeEvidencias
from .modelos_de_retakes import (
AgrupamentoDeFalas,
ClassificacaoDeRetake,
EvidenciaDeRetake,
Fala,
GrupoDeRetake,
ParAgrupado,
ResultadoDaComparacao,
ResultadoDoIntervalo,
SinalDeReinicio,
STATUS_CONFIRMADO,
STATUS_DE_REVISAO_VALIDOS,
STATUS_IGNORADO,
STATUS_PENDENTE,
STATUS_REJEITADO,
TomadaDeRetake,
)
from .repositorio_de_retakes import (
GrupoDeRetakeInexistente,
RepositorioDeRetakes,
)
__all__ = [
"AgrupadorDeTakes",
"AgrupamentoDeFalas",
"AnalisadorDeIntervalos",
"ArtefatoDeTranscricaoInvalido",
"CarregadorDeArtefatosDeAudio",
"ClassificacaoDeRetake",
"ClassificadorDeRetakes",
"ComparadorDeFalas",
"ComparadorSemantico",
"CONTINUIDADE_DE_FALA",
"CoordenadorDeRetakes",
"DeteccaoDeRetakes",
"DetectorDeReinicios",
"EvidenciaDeRetake",
"Fala",
"GeradorDeEvidencias",
"GrupoDeRetake",
"GrupoDeRetakeInexistente",
"ParAgrupado",
"POSSIVEL_RETAKE",
"RegrasDeClassificacao",
"REPETICAO_DE_FRASE",
"REPETICAO_DE_INICIO",
"REPETICAO_PARCIAL",
"RepositorioDeRetakes",
"ResultadoDaAnaliseDeRetakes",
"ResultadoDaComparacao",
"ResultadoDoIntervalo",
"RETAKE_CONFIRMADO",
"SEM_INDICIO",
"SinalDeReinicio",
"STATUS_CONFIRMADO",
"STATUS_DE_REVISAO_VALIDOS",
"STATUS_IGNORADO",
"STATUS_PENDENTE",
"STATUS_REJEITADO",
"TomadaDeRetake",
]
@@ -69,6 +69,11 @@ class AdaptadorDeFalas:
self.video_id = video_id
self.faixa_id = faixa_id
@staticmethod
def normalizar(texto: str) -> str:
"""Normaliza um texto (minúsculas, sem pontuação, sem espaços duplos)."""
return _normalizar(texto)
def converter(
self,
transcricoes: Iterable[TranscricaoDoClipe],
@@ -84,20 +89,59 @@ class AdaptadorDeFalas:
for segmento in transcricao.segmentos:
if _e_ruido(segmento.texto):
continue
falas.append(Fala(
id=f"{segmento_id}:{segmento.inicio:.3f}",
video_id=video_id,
faixa_id=faixa_id,
segmento_id=segmento_id,
ordem=-1, # preenchida após a ordenação
inicio=segmento.inicio,
fim=segmento.fim,
texto=segmento.texto,
texto_normalizado=_normalizar(segmento.texto),
palavras=segmento.palavras,
falante=segmento.falante,
))
falas.append(self._fala_de_segmento(segmento, segmento_id, video_id, faixa_id))
return self._ordenar(falas)
def converter_de_segmentos(
self,
segmentos_por_clipe: Iterable[tuple[str, Iterable[object]]],
video_id: str | None = None,
faixa_id: str | None = None,
) -> list[Fala]:
"""Converte ``(segmento_id, [SegmentoDeTranscricao])`` em falas.
Usado quando a transcrição chega no ``ContextoDeAnalise`` do scanner
já como segmentos individuais (padrão do pipeline), em vez de
``TranscricaoDoClipe``.
"""
video_id = video_id or self.video_id or "video"
faixa_id = faixa_id or self.faixa_id or "faixa"
falas: list[Fala] = []
for segmento_id, segmentos in segmentos_por_clipe:
for segmento in segmentos:
if _e_ruido(getattr(segmento, "texto", "")):
continue
falas.append(self._fala_de_segmento(segmento, segmento_id, video_id, faixa_id))
return self._ordenar(falas)
@staticmethod
def _fala_de_segmento(
segmento: object,
segmento_id: str,
video_id: str,
faixa_id: str,
) -> Fala:
texto = str(getattr(segmento, "texto", "")).strip()
inicio = float(getattr(segmento, "inicio", 0.0))
fim = float(getattr(segmento, "fim", inicio))
return Fala(
id=f"{segmento_id}:{inicio:.3f}",
video_id=video_id,
faixa_id=faixa_id,
segmento_id=segmento_id,
ordem=-1, # preenchida na ordenação
inicio=inicio,
fim=fim,
texto=texto,
texto_normalizado=_normalizar(texto),
palavras=getattr(segmento, "palavras", ()) or (),
falante=getattr(segmento, "falante", None),
)
@staticmethod
def _ordenar(falas: list[Fala]) -> list[Fala]:
falas.sort(key=lambda item: (item.inicio, item.fim))
for indice, fala in enumerate(falas):
falas[indice] = Fala(
@@ -18,7 +18,6 @@ from .modelos_de_retakes import (
AgrupamentoDeFalas,
Fala,
ParAgrupado,
ResultadoDoIntervalo,
SinalDeReinicio,
)
@@ -34,6 +33,10 @@ _JANELA_PADRAO = 5
_ZONA_DE_DUVIDA_INFERIOR = 0.60
_ZONA_DE_DUVIDA_SUPERIOR = 0.85
# Janela após o início da segunda tentativa em que um corte visual ainda é
# considerado "próximo" do possível retake (evidência fraca).
_JANELA_VISUAL_PROXIMA = 3.0
class AgrupadorDeTakes:
"""Gera grupos candidatos conectando falas semelhantes e próximas."""
@@ -52,8 +55,13 @@ class AgrupadorDeTakes:
self.detector_de_reinicios = detector_de_reinicios
self.janela = janela
def agrupar(self, falas: Iterable[Fala]) -> list[AgrupamentoDeFalas]:
def agrupar(
self,
falas: Iterable[Fala],
limites_de_cena: Iterable[float] | None = None,
) -> list[AgrupamentoDeFalas]:
falas = sorted(falas, key=lambda item: (item.ordem, item.inicio))
limites = sorted(limites_de_cena) if limites_de_cena else []
sinais = {sinal.fala_id: sinal for sinal in self.detector_de_reinicios.detectar(falas)}
arestas: list[tuple[int, int, ParAgrupado]] = []
@@ -63,7 +71,7 @@ class AgrupadorDeTakes:
fala_b = falas[jindice]
if fala_a.faixa_id != fala_b.faixa_id:
continue
par = self._avaliar_par(fala_a, fala_b, sinais)
par = self._avaliar_par(fala_a, fala_b, sinais, limites)
if par is not None:
arestas.append((indice, jindice, par))
@@ -74,6 +82,7 @@ class AgrupadorDeTakes:
fala_a: Fala,
fala_b: Fala,
sinais: dict[str, SinalDeReinicio],
limites_de_cena: list[float] | None = None,
) -> ParAgrupado | None:
comparacao = self.comparador.comparar(fala_a, fala_b)
intervalo = self.analisador_de_intervalos.analisar(fala_a, fala_b)
@@ -94,10 +103,25 @@ class AgrupadorDeTakes:
and (tem_reinicio or intervalo.indicio_de_nova_tentativa))
)
if candidato:
mudanca_visual = self._tem_mudanca_visual_proxima(
fala_a, fala_b, limites_de_cena or [])
return ParAgrupado(fala_a, fala_b, comparacao, intervalo,
round(semantica, 4))
round(semantica, 4), mudanca_visual)
return None
@staticmethod
def _tem_mudanca_visual_proxima(
fala_a: Fala,
fala_b: Fala,
limites: list[float],
) -> bool:
"""Verdadeiro quando há um corte visual na transição entre as tentativas."""
janela = _JANELA_VISUAL_PROXIMA
for limite in limites:
if fala_a.fim <= limite <= fala_b.inicio + janela:
return True
return False
@staticmethod
def _agrupar_por_arestas(
falas: list[Fala],
@@ -0,0 +1,60 @@
"""Carrega as transcrições produzidas pelo scanner para análise de retakes.
O painel não retranscreve vídeos: os artefatos ``<caso>-audio-faixa-N.json``
gerados pelo scanner já contêm os segmentos de fala por clipe. Esta classe
converte esses artefatos em falas prontas para o ``CoordenadorDeRetakes``,
sem duplicar a lógica de transcrição.
"""
from __future__ import annotations
import json
from pathlib import Path
from types import SimpleNamespace
from .adaptador_de_falas import AdaptadorDeFalas
from .modelos_de_retakes import Fala
class ArtefatoDeTranscricaoInvalido(ValueError):
"""Erro levantado quando um artefato do scanner não tem transcrição útil."""
class CarregadorDeArtefatosDeAudio:
"""Lê os artefatos de transcrição do scanner e devolve falas ordenadas."""
def __init__(self, adaptador: AdaptadorDeFalas | None = None) -> None:
self.adaptador = adaptador or AdaptadorDeFalas()
def carregar(self, caminhos_dos_artefatos: list[str | Path],
video_id: str) -> list[Fala]:
"""Converte os artefatos informados em falas prontas para análise.
Artefatos sem transcrição concluída são ignorados. Quando nenhum
artefato contiver segmentos, levanta ``ArtefatoDeTranscricaoInvalido``
para que a tela possa avisar o usuário sem executar a análise.
"""
falas: list[Fala] = []
for caminho in caminhos_dos_artefatos:
falas.extend(self._falas_do_artefato(Path(caminho), video_id))
if not falas:
raise ArtefatoDeTranscricaoInvalido(
"Nenhum artefato de transcrição contém segmentos de fala. "
"Execute o Scanner com transcrição antes de analisar retakes.")
return falas
def _falas_do_artefato(self, caminho: Path,
video_id: str) -> list[Fala]:
"""Extrai as falas de um único artefato ``-audio-faixa-N.json``."""
if not caminho.is_file():
return []
dados = json.loads(caminho.read_text(encoding="utf-8"))
if dados.get("status") != "concluida":
return []
pares: list[tuple[str, list[object]]] = []
for item in dados.get("segmentos_por_clipe", []):
segmentos = [SimpleNamespace(**segmento)
for segmento in item.get("segmentos", [])]
if segmentos:
pares.append((str(item.get("clipe", "clipe")), segmentos))
return self.adaptador.converter_de_segmentos(pares, video_id=video_id)
@@ -0,0 +1,99 @@
"""Classificação dos grupos candidatos em tipos de retake.
Transforma as métricas (similaridade textual, semântica, proximidade
temporal, sinal de reinício e repetição do início) em um tipo e um nível de
confiança, usando regras configuráveis. Nenhum valor isolado decide; a
classificação combina os sinais.
"""
from __future__ import annotations
from dataclasses import dataclass
from .modelos_de_retakes import AgrupamentoDeFalas, ClassificacaoDeRetake
SEM_INDICIO = "sem_indicio_de_retake"
RETAKE_CONFIRMADO = "retake_confirmado"
POSSIVEL_RETAKE = "possivel_retake"
REPETICAO_DE_FRASE = "repeticao_de_frase"
REPETICAO_DE_INICIO = "repeticao_de_inicio"
REPETICAO_PARCIAL = "repeticao_parcial"
CONTINUIDADE_DE_FALA = "continuidade_de_fala"
@dataclass(frozen=True)
class RegrasDeClassificacao:
"""Limiares e pesos configuráveis para a classificação."""
similaridade_textual_minima: float = 0.60
similaridade_para_confirmar: float = 0.85
proximidade_minima: float = 0.50
semantica_para_confirmar: float = 0.85
peso_textual: float = 0.50
peso_semantica: float = 0.25
peso_proximidade: float = 0.15
peso_reinicio: float = 0.10
# Evidência fraca: reforça a confiança, nunca decide o tipo sozinha.
peso_visual_fraca: float = 0.03
class ClassificadorDeRetakes:
"""Atribui tipo e confiança a um ``AgrupamentoDeFalas``."""
def __init__(self, regras: RegrasDeClassificacao | None = None) -> None:
self.regras = regras or RegrasDeClassificacao()
def classificar(self, agrupamento: AgrupamentoDeFalas) -> ClassificacaoDeRetake:
if not agrupamento.pares:
return ClassificacaoDeRetake(SEM_INDICIO, 0.0)
par = max(agrupamento.pares,
key=lambda item: item.comparacao.similaridade_final)
texto = par.comparacao.similaridade_final
semantica = par.similaridade_semantica
proximidade = par.intervalo.proximidade_temporal
reproducao_inicio = par.comparacao.similaridade_do_inicio >= 0.9
tem_reinicio = bool(agrupamento.sinais_de_reinicio)
r = self.regras
confianca = (
r.peso_textual * texto
+ r.peso_semantica * semantica
+ r.peso_proximidade * proximidade
+ r.peso_reinicio * (1.0 if tem_reinicio else 0.0)
)
# A análise visual é reforço (evidência fraca): soma um pequeno peso à
# confiança sem jamais alterar o tipo de retake.
if par.mudanca_visual_proxima:
confianca += r.peso_visual_fraca
# Repetições completas e muito semelhantes de frase.
if texto >= r.similaridade_para_confirmar:
tipo = RETAKE_CONFIRMADO
if reproducao_inicio:
tipo = RETAKE_CONFIRMADO
elif (
texto >= r.similaridade_textual_minima
and semantica >= r.semantica_para_confirmar
and tem_reinicio
):
tipo = RETAKE_CONFIRMADO
elif (
texto >= r.similaridade_textual_minima
and proximidade >= r.proximidade_minima
):
tipo = POSSIVEL_RETAKE
elif reproducao_inicio and tem_reinicio:
tipo = REPETICAO_DE_INICIO
elif semantica >= r.semantica_para_confirmar:
tipo = REPETICAO_PARCIAL
else:
tipo = POSSIVEL_RETAKE
if tipo == RETAKE_CONFIRMADO:
confianca = min(1.0, max(confianca, texto * 0.9 + 0.05 * (1 if tem_reinicio else 0)))
else:
confianca = min(1.0, confianca)
return ClassificacaoDeRetake(tipo, round(confianca, 3))
@@ -0,0 +1,203 @@
"""Coordenador do módulo de detecção de retakes.
Responsável por organizar o fluxo completo, delegando cada etapa a uma das
classes do submódulo. Não implementa as regras de detecção — apenas chama
os colaboradores na ordem correta e monta o resultado.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Iterable
from ..transcricao_da_timeline import TranscricaoDoClipe
from .adaptador_de_falas import AdaptadorDeFalas
from .agrupador_de_takes import AgrupadorDeTakes
from .analisador_de_intervalos import AnalisadorDeIntervalos
from .classificador_de_retakes import ClassificadorDeRetakes
from .comparador_de_falas import ComparadorDeFalas
from .comparador_semantico import ComparadorSemantico
from .detector_de_reinicios import DetectorDeReinicios
from .gerador_de_evidencias import GeradorDeEvidencias
from .modelos_de_retakes import (
Fala,
GrupoDeRetake,
TomadaDeRetake,
gerar_id_do_grupo,
)
from .repositorio_de_retakes import RepositorioDeRetakes
def _limites_de_cena(cenas_visuais: Iterable[object] | None) -> list[float]:
"""Extrai os pontos de corte visual do contexto do scanner.
Cada cena visual tem um ``inicio``; o início de uma cena que não é a
primeira representa uma mudança visual. Objetos sem ``inicio`` são
ignorados de forma segura.
"""
if not cenas_visuais:
return []
limites: set[float] = set()
for cena in cenas_visuais:
inicio = getattr(cena, "inicio", None)
if inicio is not None and float(inicio) > 0:
limites.add(float(inicio))
return sorted(limites)
@dataclass
class ResultadoDaAnaliseDeRetakes:
"""Saída estruturada da análise de retakes de um vídeo."""
video_id: str
grupos: list[GrupoDeRetake]
def para_dict(self) -> dict[str, Any]:
return {
"video_id": self.video_id,
"grupos_de_retakes": [
{
"grupo_id": grupo.id,
"tipo": grupo.tipo,
"confianca": grupo.confianca,
"tomada_principal_id": grupo.tomada_principal_id,
"tomadas": [
{
"ordem": tomada.ordem,
"fala_id": tomada.fala_id,
"segmento_id": tomada.segmento_id,
"inicio": tomada.inicio,
"fim": tomada.fim,
"texto": tomada.texto,
"similaridade_com_anterior": tomada.similaridade_com_anterior,
"confianca": tomada.confianca,
}
for tomada in grupo.tomadas
],
"evidencias": [
{
"tipo": evidencia.tipo,
"descricao": evidencia.descricao,
"valor": evidencia.valor,
}
for evidencia in grupo.evidencias
],
}
for grupo in self.grupos
],
}
class CoordenadorDeRetakes:
"""Organiza a análise completa de retakes de um vídeo."""
def __init__(
self,
detector_de_reinicios: DetectorDeReinicios | None = None,
comparador_de_falas: ComparadorDeFalas | None = None,
comparador_semantico: ComparadorSemantico | None = None,
agrupador_de_takes: AgrupadorDeTakes | None = None,
analisador_de_intervalos: AnalisadorDeIntervalos | None = None,
classificador_de_retakes: ClassificadorDeRetakes | None = None,
gerador_de_evidencias: GeradorDeEvidencias | None = None,
repositorio_de_retakes: RepositorioDeRetakes | None = None,
adaptador_de_falas: AdaptadorDeFalas | None = None,
) -> None:
self.comparador_de_falas = comparador_de_falas or ComparadorDeFalas()
self.comparador_semantico = comparador_semantico or ComparadorSemantico()
self.detector_de_reinicios = detector_de_reinicios or DetectorDeReinicios(
self.comparador_de_falas)
self.analisador_de_intervalos = analisador_de_intervalos or AnalisadorDeIntervalos()
self.agrupador_de_takes = agrupador_de_takes or AgrupadorDeTakes(
self.comparador_de_falas,
self.comparador_semantico,
self.analisador_de_intervalos,
self.detector_de_reinicios,
)
self.classificador_de_retakes = classificador_de_retakes or ClassificadorDeRetakes()
self.gerador_de_evidencias = gerador_de_evidencias or GeradorDeEvidencias()
self.repositorio_de_retakes = repositorio_de_retakes or RepositorioDeRetakes()
self.adaptador_de_falas = adaptador_de_falas or AdaptadorDeFalas()
def analisar(
self,
video_id: str,
transcricoes: Iterable[TranscricaoDoClipe] | None = None,
falas: Iterable[Fala] | None = None,
faixa_id: str | None = None,
cenas_visuais: Iterable[object] | None = None,
) -> ResultadoDaAnaliseDeRetakes:
"""Executa o pipeline completo e retorna os grupos de retakes.
Se ``falas`` for fornecido, usa diretamente; caso contrário converte
os ``TranscricaoDoClipe`` recebidos via ``adaptador_de_falas``.
``cenas_visuais`` é opcional: quando informado, os cortes visuais são
usados como evidência fraca (reforço), nunca como regra obrigatória.
"""
if falas is None:
transcricoes = list(transcricoes or [])
falas = self.adaptador_de_falas.converter(transcricoes, video_id, faixa_id)
falas = list(falas)
limites_de_cena = _limites_de_cena(cenas_visuais)
versao_visual = "visual" if limites_de_cena else "sem_visual"
# Idempotência: se já analisado com as mesmas versões, devolve o cache.
# A ``versao_visual`` no hash garante que ativar/desativar a análise
# visual força uma nova análise, em vez de reaproveitar a antiga (§10).
if self.repositorio_de_retakes.ja_analisado(video_id, versao_visual):
grupos = self.repositorio_de_retakes.carregar(video_id)
return ResultadoDaAnaliseDeRetakes(video_id, grupos)
agrupamentos = self.agrupador_de_takes.agrupar(falas, limites_de_cena)
grupos: list[GrupoDeRetake] = []
for agrupamento in agrupamentos:
classificacao = self.classificador_de_retakes.classificar(agrupamento)
evidencias = self.gerador_de_evidencias.gerar(agrupamento, classificacao)
tomadas = self._montar_tomadas(agrupamento, classificacao.confianca)
grupos.append(GrupoDeRetake(
id=gerar_id_do_grupo(),
video_id=video_id,
faixa_id=self._faixa_do_grupo(agrupamento),
tipo=classificacao.tipo,
confianca=classificacao.confianca,
tomadas=tomadas,
evidencias=evidencias,
))
self.repositorio_de_retakes.registrar(video_id, grupos, versao_visual)
return ResultadoDaAnaliseDeRetakes(video_id, grupos)
@staticmethod
def _faixa_do_grupo(agrupamento) -> str:
if agrupamento.pares:
return agrupamento.pares[0].fala_a.faixa_id
return ""
@staticmethod
def _montar_tomadas(agrupamento, confianca_do_grupo: float) -> list[TomadaDeRetake]:
# Mapa das falas e da similaridade de cada uma com a anterior.
falas: dict[str, Fala] = {}
similaridade_com_anterior: dict[str, float] = {}
for par in agrupamento.pares:
falas[par.fala_a.id] = par.fala_a
falas[par.fala_b.id] = par.fala_b
similaridade_com_anterior[par.fala_b.id] = par.comparacao.similaridade_final
tomadas: list[TomadaDeRetake] = []
for ordem, fala_id in enumerate(agrupamento.fala_ids):
if fala_id not in falas:
continue
fala = falas[fala_id]
tomadas.append(TomadaDeRetake(
ordem=ordem,
fala_id=fala.id,
segmento_id=fala.segmento_id,
inicio=fala.inicio,
fim=fala.fim,
texto=fala.texto,
similaridade_com_anterior=similaridade_com_anterior.get(fala_id, 0.0),
confianca=confianca_do_grupo,
))
return tomadas
self.adaptador_de_falas = adaptador_de_falas or AdaptadorDeFalas()
@@ -0,0 +1,59 @@
"""Adaptador que expõe a detecção de retakes como um ``Analisador`` do scanner.
Permite plugar o ``CoordenadorDeRetakes`` no ``PipelineDoScanner`` como uma
etapa entre a detecção de cenas/eventos e a persistência, reutilizando o
``ContextoDeAnalise`` que já carrega as transcrições.
"""
from __future__ import annotations
from typing import Any
from ...dominio import Timeline
from .adaptador_de_falas import AdaptadorDeFalas
from .coordenador_de_retakes import CoordenadorDeRetakes
class DeteccaoDeRetakes:
nome = "deteccao_de_retakes"
def __init__(
self,
coordenador: CoordenadorDeRetakes | None = None,
video_id: str | None = None,
adaptador: AdaptadorDeFalas | None = None,
) -> None:
self.coordenador = coordenador or CoordenadorDeRetakes()
self.video_id = video_id
self.adaptador = adaptador or AdaptadorDeFalas()
def executar(self, contexto: Any) -> Any:
"""Consome as transcrições do contexto e preenche ``contexto.retakes``."""
if contexto is None:
return contexto
transcricoes = getattr(contexto, "transcricoes", None) or {}
if not transcricoes:
return contexto
# ``contexto.transcricoes`` é um dicionário ``identificador_clipe ->
# list[SegmentoDeTranscricao]``. Ordenamos por chave para processar as
# falas na ordem da timeline.
itens = [
(segmento_id, transcricoes[segmento_id])
for segmento_id in sorted(transcricoes)
]
timeline: Timeline | None = contexto.timeline
video_id = self.video_id or (timeline.identificador if timeline else "video")
falas = self.adaptador.converter_de_segmentos(itens, video_id=video_id)
# Cenas visuais do contexto entram como reforço (evidência fraca).
cenas_visuais = getattr(contexto, "cenas_visuais", None) or []
resultado = self.coordenador.analisar(video_id, falas=falas, cenas_visuais=cenas_visuais)
contexto.retakes = {
"status": "concluida",
"resultado": resultado.para_dict(),
"grupos": resultado.grupos,
}
return contexto
@@ -78,9 +78,18 @@ class DetectorDeReinicios:
if not evidencias:
return None
tem_marcador = any("marcador" in e for e in evidencias)
tem_repeticao = any("repetição" in e for e in evidencias)
if tem_marcador:
tipo = "reinicio_explicito"
elif tem_repeticao:
tipo = "repeticao_do_inicio"
else:
tipo = "reinicio_pos_pausa"
return SinalDeReinicio(
fala_id=fala.id,
tipo="repeticao_do_inicio" if any("repetição" in e for e in evidencias) else "reinicio_pos_pausa",
tipo=tipo,
intensidade=round(min(1.0, max(intensidades)), 3),
evidencias=evidencias,
)
@@ -0,0 +1,87 @@
"""Geração de evidências legíveis para cada grupo de retakes.
Produz uma lista estruturada ``{tipo, descricao, valor}`` que explica por
que duas falas foram consideradas retake — pronto para a interface do
painel descrever os motivos ao usuário.
"""
from __future__ import annotations
from .modelos_de_retakes import AgrupamentoDeFalas, ClassificacaoDeRetake, EvidenciaDeRetake
class GeradorDeEvidencias:
"""Monta as evidências de um grupo a partir do agrupamento e da categoria."""
def gerar(
self,
agrupamento: AgrupamentoDeFalas,
classificacao: ClassificacaoDeRetake,
) -> list[EvidenciaDeRetake]:
evidencias: list[EvidenciaDeRetake] = []
if not agrupamento.pares:
return evidencias
par = max(agrupamento.pares,
key=lambda item: item.comparacao.similaridade_final)
comp = par.comparacao
estado = "É um retake provável." if classificacao.tipo == "retake_confirmado" else (
"Pode ser um retake." if classificacao.tipo == "possivel_retake" else "Repetição detectada.")
evidencias.append(EvidenciaDeRetake(
tipo="similaridade_textual",
descricao=(
f"As duas falas possuem {comp.similaridade_jaccard * 100:.0f}% "
f"de palavras em comum e {comp.similaridade_final * 100:.0f}% de "
f"similaridade textual geral."
),
valor=comp.similaridade_final,
))
if par.similaridade_semantica > 0:
evidencias.append(EvidenciaDeRetake(
tipo="similaridade_semantica",
descricao=f"Semanticamente, as falas têm {par.similaridade_semantica * 100:.0f}% de afinidade.",
valor=par.similaridade_semantica,
))
if comp.similaridade_do_inicio >= 0.8:
evidencias.append(EvidenciaDeRetake(
tipo="repeticao_do_inicio",
descricao="As duas falas começam com as mesmas palavras.",
valor=comp.similaridade_do_inicio,
))
evidencias.append(EvidenciaDeRetake(
tipo="proximidade_temporal",
descricao=(
f"A segunda tentativa começa {par.intervalo.intervalo_em_segundos:.1f} "
f"segundos após a primeira."
),
valor=par.intervalo.proximidade_temporal,
))
for sinal in agrupamento.sinais_de_reinicio:
for detalhe in sinal.evidencias:
evidencias.append(EvidenciaDeRetake(
tipo="reinicio",
descricao=detalhe,
valor=sinal.intensidade,
))
if par.mudanca_visual_proxima:
evidencias.append(EvidenciaDeRetake(
tipo="mudanca_visual_proxima",
descricao=(
"Há um corte visual próximo: as tentativas podem estar em "
"tomadas diferentes, reforçando a hipótese de retake."
),
valor=1.0,
))
evidencias.append(EvidenciaDeRetake(
tipo="classificacao",
descricao=f"{estado} Confiança de {classificacao.confianca * 100:.0f}%.",
valor=classificacao.confianca,
))
return evidencias
@@ -115,6 +115,7 @@ class ParAgrupado:
comparacao: ResultadoDaComparacao
intervalo: ResultadoDoIntervalo
similaridade_semantica: float = 0.0
mudanca_visual_proxima: bool = False
@dataclass(frozen=True)
@@ -126,6 +127,24 @@ class AgrupamentoDeFalas:
sinais_de_reinicio: tuple[SinalDeReinicio, ...] = ()
@dataclass(frozen=True)
class ClassificacaoDeRetake:
"""Resultado da classificação de um grupo de retakes."""
tipo: str
confianca: float
# Estados possíveis da revisão manual de um grupo (decisão do usuário na
# tela de retakes; a detecção automática nunca altera o estado de revisão).
STATUS_PENDENTE = "pendente_de_revisao"
STATUS_CONFIRMADO = "confirmado"
STATUS_REJEITADO = "nao_e_retake"
STATUS_IGNORADO = "ignorado"
STATUS_DE_REVISAO_VALIDOS = frozenset(
{STATUS_PENDENTE, STATUS_CONFIRMADO, STATUS_REJEITADO, STATUS_IGNORADO})
@dataclass(frozen=True)
class GrupoDeRetake:
"""Agrupa as tomadas que representam tentativas da mesma fala."""
@@ -138,6 +157,7 @@ class GrupoDeRetake:
tomadas: list[TomadaDeRetake] = field(default_factory=list)
evidencias: list[EvidenciaDeRetake] = field(default_factory=list)
criado_em: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
status_de_revisao: str = STATUS_PENDENTE
def __post_init__(self) -> None:
if not 0.0 <= self.confianca <= 1.0:
@@ -0,0 +1,179 @@
"""Persistência dos grupos de retakes.
Na primeira versão, grava os grupos em um arquivo JSON idempotente, no
mesmo espírito do cache de transcrição da engine. O `RepositorioDeRetakes`
define o contrato (``registrar``/``carregar``/``ja_analisado``) para que a
camada de banco de dados (SQLite/Postgres, prevista na Etapa 8 da
arquitetura) possa ser plugada depois sem alterar o coordenador.
"""
from __future__ import annotations
import hashlib
import json
from dataclasses import asdict
from pathlib import Path
from typing import Iterable
from .modelos_de_retakes import (
STATUS_DE_REVISAO_VALIDOS,
GrupoDeRetake,
)
# Versões da transição e das regras para a idempotência.
VERSAO_REGRA_FALAS = 1
VERSAO_REGRA_TEXTO = 1
VERSAO_MODELO_PADRAO = "lexico"
class GrupoDeRetakeInexistente(LookupError):
"""Erro levantado quando um grupo de retakes não existe no repositório."""
def __init__(self, video_id: str, grupo_id: str) -> None:
super().__init__(
f"Grupo {grupo_id} não encontrado para o vídeo {video_id}.")
self.video_id = video_id
self.grupo_id = grupo_id
class RepositorioDeRetakes:
"""Persiste e recupera grupos de retakes em um arquivo JSON."""
def __init__(
self,
diretoria_de_trabalho: str | Path = ".retakes",
versao_transcricao: str = "1",
versao_regras: str = f"regras:{VERSAO_REGRA_FALAS}.{VERSAO_REGRA_TEXTO}",
versao_modelo: str = VERSAO_MODELO_PADRAO,
) -> None:
self.diretoria_de_trabalho = Path(diretoria_de_trabalho)
self.versao_transcricao = versao_transcricao
self.versao_regras = versao_regras
self.versao_modelo = versao_modelo
def _caminho(self, video_id: str) -> Path:
return self.diretoria_de_trabalho / f"retakes-{self._seguro(video_id)}.json"
@staticmethod
def _seguro(valor: str) -> str:
return "".join(c if c.isalnum() or c in "._-" else "_" for c in valor) or "video"
def _hash_de_identificacao(self, video_id: str, versao_visual: str = "sem_visual") -> str:
origem = "|".join([video_id, self.versao_transcricao, self.versao_regras,
self.versao_modelo, versao_visual])
return hashlib.sha256(origem.encode("utf-8")).hexdigest()[:12]
def ja_analisado(self, video_id: str, versao_visual: str = "sem_visual") -> bool:
"""Verdadeiro se o vídeo já foi analisado com exatamente estas versões.
A ``versao_visual`` separa análises com e sem cenas visuais, de modo
que uma mudança de configuração visual força uma nova análise (§10).
"""
caminho = self._caminho(video_id)
if not caminho.is_file():
return False
try:
dados = json.loads(caminho.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
return False
return dados.get("_hash") == self._hash_de_identificacao(video_id, versao_visual)
def registrar(
self,
video_id: str,
grupos: Iterable[GrupoDeRetake],
versao_visual: str = "sem_visual",
) -> Path:
"""Grava os grupos de forma idempotente e retorna o arquivo gerado."""
self.diretoria_de_trabalho.mkdir(parents=True, exist_ok=True)
dados = {
"video_id": video_id,
"versao_transcricao": self.versao_transcricao,
"versao_regras": self.versao_regras,
"versao_modelo": self.versao_modelo,
"versao_visual": versao_visual,
"_hash": self._hash_de_identificacao(video_id, versao_visual),
"grupos_de_retakes": [asdict(grupo) for grupo in grupos],
}
caminho = self._caminho(video_id)
caminho.write_text(json.dumps(dados, ensure_ascii=False, indent=2),
encoding="utf-8")
return caminho
def carregar(self, video_id: str) -> list[GrupoDeRetake]:
"""Carrega os grupos já persistidos para o vídeo."""
caminho = self._caminho(video_id)
if not caminho.is_file():
return []
dados = json.loads(caminho.read_text(encoding="utf-8"))
grupos: list[GrupoDeRetake] = []
for item in dados.get("grupos_de_retakes", []):
tomadas = item.get("tomadas", [])
evidencias = item.get("evidencias", [])
grupos.append(GrupoDeRetake(
id=item["id"],
video_id=item.get("video_id", video_id),
faixa_id=item.get("faixa_id", ""),
tipo=item["tipo"],
confianca=item["confianca"],
tomadas=[self._tomada_do_dict(t) for t in tomadas],
evidencias=[self._evidencia_do_dict(e) for e in evidencias],
criado_em=item.get("criado_em", ""),
status_de_revisao=item.get("status_de_revisao", "pendente_de_revisao"),
))
return grupos
def atualizar_status_de_revisao(
self,
video_id: str,
grupo_id: str,
novo_status: str,
) -> GrupoDeRetake:
"""Registra a decisão manual do usuário sobre um grupo de retakes.
A revisão não altera a detecção automática: apenas grava o estado
escolhido (confirmado, não é retake, ignorado ou pendente) no arquivo
já existente do vídeo, preservando os demais grupos.
Levanta ``ValueError`` se o status não for válido e
``GrupoDeRetakeInexistente`` se o grupo não existir.
"""
if novo_status not in STATUS_DE_REVISAO_VALIDOS:
validos = ", ".join(sorted(STATUS_DE_REVISAO_VALIDOS))
raise ValueError(f"Status de revisão inválido: {novo_status!r}. "
f"Válidos: {validos}.")
caminho = self._caminho(video_id)
if not caminho.is_file():
raise GrupoDeRetakeInexistente(video_id, grupo_id)
dados = json.loads(caminho.read_text(encoding="utf-8"))
grupos = dados.get("grupos_de_retakes", [])
alvo = next((g for g in grupos if g.get("id") == grupo_id), None)
if alvo is None:
raise GrupoDeRetakeInexistente(video_id, grupo_id)
alvo["status_de_revisao"] = novo_status
caminho.write_text(json.dumps(dados, ensure_ascii=False, indent=2),
encoding="utf-8")
return self.carregar(video_id)[grupos.index(alvo)]
@staticmethod
def _tomada_do_dict(item: dict) -> object:
from .modelos_de_retakes import TomadaDeRetake
return TomadaDeRetake(
ordem=item["ordem"],
fala_id=item["fala_id"],
segmento_id=item.get("segmento_id", ""),
inicio=item["inicio"],
fim=item["fim"],
texto=item.get("texto", ""),
similaridade_com_anterior=item.get("similaridade_com_anterior", 0.0),
confianca=item.get("confianca", 0.0),
)
@staticmethod
def _evidencia_do_dict(item: dict) -> object:
from .modelos_de_retakes import EvidenciaDeRetake
return EvidenciaDeRetake(
tipo=item["tipo"],
descricao=item.get("descricao", ""),
valor=item.get("valor"),
)