João Henrique
b0b8027c7e
- aprimorado o preview de falantes na tela Refinar para reproduzir os trechos do vídeo - corrigida a leitura dos timestamps do preview de falantes usando os clipes do banco Resumo: - 5 arquivos alterados - 1 novos - 4 modificados - 0 removidos 4 files changed, 142 insertions(+), 3 deletions(-) Arquivos: - .jhonny/analises.db - code/cep-plugin/main.js - code/engine/persistencia/consultas.py - code/engine/testes/test_consultas_de_persistencia.py - code/engine/consultar_preview_de_falantes.py
583 lines
24 KiB
Python
583 lines
24 KiB
Python
"""
|
|
Módulo de consultas de acesso rápido ao banco de análises.
|
|
|
|
Este módulo define ``ConsultasDeAnalises``, responsável por ler o banco de
|
|
análises por intenção (visão geral do vídeo, falas num intervalo, retakes,
|
|
evidências visuais, cenas), para que um agente que vai montar ou editar um
|
|
vídeo consulte o banco sem precisar escrever SQL nem carregar tabelas
|
|
inteiras.
|
|
|
|
Este módulo não persiste dados — apenas lê. A escrita fica a cargo de
|
|
``RepositorioDeTimelineSQLite``, ``RepositorioDeAnalisesSQLite`` e
|
|
``RepositorioDeRetakesSQLite``, no mesmo pacote.
|
|
|
|
Cada método devolve apenas os campos relevantes para a decisão daquele
|
|
nível (nunca a linha inteira da tabela, nunca sub-recursos que não foram
|
|
pedidos), para manter as respostas compactas em tokens quando consumidas
|
|
por um agente de IA.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
from .conexao import abrir_banco
|
|
|
|
|
|
class ErroDeConsultaInvalida(ValueError):
|
|
"""Representa uma consulta feita com parâmetros inválidos ou incompletos."""
|
|
|
|
|
|
class ConsultasDeAnalises:
|
|
"""
|
|
Consulta o banco de análises por intenção, devolvendo respostas compactas.
|
|
|
|
Esta classe não persiste dados e não altera o banco — todos os métodos
|
|
são operações de leitura. A escrita é responsabilidade dos
|
|
repositórios do pacote (``RepositorioDeTimelineSQLite``,
|
|
``RepositorioDeAnalisesSQLite``, ``RepositorioDeRetakesSQLite``).
|
|
|
|
Atributos:
|
|
conexao: Conexão SQLite já aberta e com o esquema aplicado.
|
|
"""
|
|
|
|
def __init__(self, banco: str | Path | sqlite3.Connection = ".jhonny/analises.db") -> None:
|
|
"""
|
|
Inicializa as consultas a partir de um caminho de banco ou conexão existente.
|
|
|
|
Parâmetros:
|
|
banco: Caminho do arquivo do banco de análises, ou uma conexão
|
|
SQLite já aberta (reaproveitada sem reabrir).
|
|
"""
|
|
self.conexao = banco if isinstance(banco, sqlite3.Connection) else abrir_banco(banco)
|
|
|
|
def consultar_resumo_do_video(self, video_id: str) -> dict | None:
|
|
"""
|
|
Consulta a visão geral de um vídeo: o ponto de partida antes de descer a qualquer detalhe.
|
|
|
|
Parâmetros:
|
|
video_id: Identificador do vídeo (mesmo id usado na timeline do Premiere).
|
|
|
|
Retorna:
|
|
Um dicionário com nome, duração, resolução e as contagens de
|
|
clipes, falas, retakes e cenas do vídeo, ou None quando o
|
|
vídeo não existe no banco.
|
|
|
|
Pode gerar:
|
|
ErroDeConsultaInvalida: quando ``video_id`` for vazio.
|
|
"""
|
|
self._validar_texto_obrigatorio("video_id", video_id)
|
|
|
|
video = self.conexao.execute(
|
|
"SELECT nome, duracao, taxa_de_quadros, largura, altura FROM videos WHERE id = ?",
|
|
(video_id,),
|
|
).fetchone()
|
|
if video is None:
|
|
return None
|
|
contagens = self.conexao.execute(
|
|
"""SELECT
|
|
(SELECT COUNT(*) FROM clipes WHERE video_id = ?) AS clipes,
|
|
(SELECT COUNT(*) FROM segmentos_de_transcricao WHERE video_id = ?) AS falas,
|
|
(SELECT COUNT(*) FROM grupos_de_retake WHERE video_id = ?) AS retakes,
|
|
(SELECT COUNT(*) FROM cenas WHERE video_id = ?) AS cenas""",
|
|
(video_id, video_id, video_id, video_id),
|
|
).fetchone()
|
|
return {
|
|
"video_id": video_id,
|
|
"nome": video["nome"],
|
|
"duracao": video["duracao"],
|
|
"taxa_de_quadros": video["taxa_de_quadros"],
|
|
"resolucao": f"{video['largura']}x{video['altura']}" if video["largura"] else None,
|
|
"total_clipes": contagens["clipes"],
|
|
"total_falas": contagens["falas"],
|
|
"total_retakes": contagens["retakes"],
|
|
"total_cenas": contagens["cenas"],
|
|
}
|
|
|
|
def consultar_status_da_analise(self, video_id: str) -> dict | None:
|
|
"""Devolve o estado compacto das análises de áudio de um vídeo."""
|
|
self._validar_texto_obrigatorio("video_id", video_id)
|
|
resumo = self.consultar_resumo_do_video(video_id)
|
|
if resumo is None:
|
|
return None
|
|
linha = self.conexao.execute(
|
|
"""SELECT
|
|
COUNT(*) AS falas,
|
|
SUM(CASE WHEN falante IS NOT NULL AND TRIM(falante) <> '' THEN 1 ELSE 0 END) AS falas_com_falante,
|
|
SUM(CASE WHEN caracteristicas_acusticas IS NOT NULL
|
|
AND caracteristicas_acusticas <> '{}'
|
|
THEN 1 ELSE 0 END) AS falas_com_metricas
|
|
FROM segmentos_de_transcricao
|
|
WHERE video_id = ?""",
|
|
(video_id,),
|
|
).fetchone()
|
|
falantes = [
|
|
item["falante"] for item in self.conexao.execute(
|
|
"""SELECT DISTINCT falante
|
|
FROM segmentos_de_transcricao
|
|
WHERE video_id = ? AND falante IS NOT NULL AND TRIM(falante) <> ''
|
|
ORDER BY falante""",
|
|
(video_id,),
|
|
).fetchall()
|
|
]
|
|
return {
|
|
**resumo,
|
|
"audio": {
|
|
"transcricao_disponivel": bool(linha["falas"]),
|
|
"total_falas": linha["falas"] or 0,
|
|
"falas_com_falante": linha["falas_com_falante"] or 0,
|
|
"falas_com_metricas": linha["falas_com_metricas"] or 0,
|
|
"diarizacao_disponivel": bool(linha["falas_com_falante"]),
|
|
"metricas_de_voz_disponiveis": bool(linha["falas_com_metricas"]),
|
|
"falantes": falantes,
|
|
},
|
|
}
|
|
|
|
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.
|
|
|
|
Os segmentos de transcrição usam segundos da timeline e normalmente
|
|
pertencem a um clipe de áudio. A consulta encontra o clipe de vídeo
|
|
correspondente e calcula os segundos na origem usando os campos
|
|
``inicio_na_timeline`` e ``inicio_na_origem`` da tabela ``clipes``.
|
|
|
|
Parâmetros:
|
|
video_id: Identificador do vídeo analisado.
|
|
|
|
Retorna:
|
|
Lista ordenada de intervalos com falante, tempos da timeline,
|
|
tempos no arquivo de origem e caminho do vídeo.
|
|
|
|
Pode gerar:
|
|
ErroDeConsultaInvalida: quando ``video_id`` estiver vazio.
|
|
"""
|
|
self._validar_texto_obrigatorio("video_id", video_id)
|
|
linhas = self.conexao.execute(
|
|
"""SELECT s.clipe_id, s.inicio, s.fim, s.falante,
|
|
a.arquivo AS arquivo_do_audio,
|
|
v.id AS clipe_de_video, v.inicio_na_timeline,
|
|
v.fim_na_timeline, v.inicio_na_origem,
|
|
v.fim_na_origem, v.arquivo AS arquivo_do_video
|
|
FROM segmentos_de_transcricao AS s
|
|
LEFT JOIN clipes AS a
|
|
ON a.video_id = s.video_id AND a.id = s.clipe_id
|
|
LEFT JOIN clipes AS v
|
|
ON v.video_id = s.video_id
|
|
AND v.inicio_na_timeline < s.fim
|
|
AND v.fim_na_timeline > s.inicio
|
|
LEFT JOIN faixas AS f
|
|
ON f.video_id = v.video_id AND f.id = v.faixa_id
|
|
AND f.tipo = 'video'
|
|
WHERE s.video_id = ?
|
|
AND s.falante IS NOT NULL
|
|
AND TRIM(s.falante) <> ''
|
|
AND (f.id IS NOT NULL OR v.id IS NULL)
|
|
ORDER BY s.inicio, s.fim,
|
|
(v.arquivo = a.arquivo) DESC, s.id""",
|
|
(video_id,),
|
|
).fetchall()
|
|
|
|
resultado = []
|
|
vistos = set()
|
|
for linha in linhas:
|
|
chave = (linha["clipe_id"], linha["inicio"], linha["fim"], linha["falante"])
|
|
if chave in vistos:
|
|
continue
|
|
vistos.add(chave)
|
|
inicio_na_timeline = float(linha["inicio"])
|
|
fim_na_timeline = float(linha["fim"])
|
|
inicio_na_origem = linha["inicio_na_origem"]
|
|
fim_na_origem = linha["fim_na_origem"]
|
|
arquivo = linha["arquivo_do_video"] or linha["arquivo_do_audio"]
|
|
if inicio_na_origem is None:
|
|
inicio_no_arquivo = inicio_na_timeline
|
|
fim_no_arquivo = fim_na_timeline
|
|
else:
|
|
deslocamento = inicio_na_timeline - float(linha["inicio_na_timeline"])
|
|
inicio_no_arquivo = float(inicio_na_origem) + deslocamento
|
|
fim_no_arquivo = inicio_no_arquivo + (fim_na_timeline - inicio_na_timeline)
|
|
if fim_na_origem is not None:
|
|
fim_no_arquivo = min(fim_no_arquivo, float(fim_na_origem))
|
|
if not arquivo or fim_no_arquivo <= inicio_no_arquivo:
|
|
continue
|
|
resultado.append({
|
|
"clipe_id": linha["clipe_id"],
|
|
"falante": linha["falante"],
|
|
"inicio_na_timeline": inicio_na_timeline,
|
|
"fim_na_timeline": fim_na_timeline,
|
|
"inicio": max(0.0, inicio_no_arquivo),
|
|
"fim": max(0.0, fim_no_arquivo),
|
|
"arquivo": arquivo,
|
|
"clipe_de_video": linha["clipe_de_video"],
|
|
})
|
|
return resultado
|
|
|
|
def listar_falas_no_intervalo(
|
|
self,
|
|
video_id: str,
|
|
inicio: float = 0.0,
|
|
fim: float | None = None,
|
|
incluir_palavras: bool = False,
|
|
) -> list[dict]:
|
|
"""
|
|
Lista as falas transcritas num intervalo da timeline, ordenadas por posição.
|
|
|
|
Não traz ``caracteristicas_acusticas`` nem palavra a palavra por
|
|
padrão — só o necessário para decidir o que cortar. Use
|
|
``incluir_palavras=True`` apenas quando o corte precisar acontecer
|
|
no meio de uma frase.
|
|
|
|
Parâmetros:
|
|
video_id: Identificador do vídeo.
|
|
inicio: Início do intervalo, em segundos na timeline (padrão: 0.0).
|
|
fim: Fim do intervalo, em segundos na timeline. Quando None,
|
|
não há limite superior.
|
|
incluir_palavras: Quando True, inclui a lista de palavras de
|
|
cada fala (com início e fim próprios). Aumenta bastante o
|
|
tamanho da resposta — usar somente quando necessário.
|
|
|
|
Retorna:
|
|
Lista de dicionários com clipe, intervalo, texto, falante e
|
|
emoção de cada fala encontrada. Lista vazia quando não há
|
|
falas no intervalo.
|
|
|
|
Pode gerar:
|
|
ErroDeConsultaInvalida: quando ``video_id`` for vazio, ``inicio``
|
|
for negativo, ou ``fim`` for anterior a ``inicio``.
|
|
"""
|
|
self._validar_texto_obrigatorio("video_id", video_id)
|
|
self._validar_intervalo(inicio, fim)
|
|
|
|
condicoes = ["video_id = ?", "fim >= ?"]
|
|
parametros: list = [video_id, inicio]
|
|
self._acrescentar_se_definido(condicoes, parametros, "inicio <= ?", fim)
|
|
|
|
linhas = self.conexao.execute(
|
|
f"""SELECT id, clipe_id, inicio, fim, texto, confianca, falante, emocao
|
|
FROM segmentos_de_transcricao
|
|
WHERE {self._clausula_where(condicoes)}
|
|
ORDER BY inicio""",
|
|
parametros,
|
|
).fetchall()
|
|
|
|
resultado = []
|
|
for linha in linhas:
|
|
item = {
|
|
"clipe_id": linha["clipe_id"],
|
|
"inicio": linha["inicio"],
|
|
"fim": linha["fim"],
|
|
"texto": linha["texto"],
|
|
"falante": linha["falante"],
|
|
"emocao": linha["emocao"],
|
|
}
|
|
if incluir_palavras:
|
|
item["palavras"] = [
|
|
{"texto": p["texto"], "inicio": p["inicio"], "fim": p["fim"]}
|
|
for p in self.conexao.execute(
|
|
"""SELECT texto, inicio, fim FROM palavras_de_transcricao
|
|
WHERE segmento_id = ? ORDER BY ordem""",
|
|
(linha["id"],),
|
|
).fetchall()
|
|
]
|
|
resultado.append(item)
|
|
return resultado
|
|
|
|
def consultar_clipe_no_instante(self, video_id: str, instante: float) -> dict | None:
|
|
"""
|
|
Consulta qual clipe da timeline cobre um instante específico.
|
|
|
|
Parâmetros:
|
|
video_id: Identificador do vídeo.
|
|
instante: Posição na timeline, em segundos.
|
|
|
|
Retorna:
|
|
Um dicionário com o clipe encontrado (id, faixa, nome,
|
|
intervalo e arquivo de origem), ou None quando nenhum clipe
|
|
cobre o instante informado.
|
|
|
|
Pode gerar:
|
|
ErroDeConsultaInvalida: quando ``video_id`` for vazio ou
|
|
``instante`` for negativo.
|
|
"""
|
|
self._validar_texto_obrigatorio("video_id", video_id)
|
|
if instante < 0:
|
|
raise ErroDeConsultaInvalida("O instante consultado não pode ser negativo.")
|
|
|
|
linha = self.conexao.execute(
|
|
"""SELECT id, faixa_id, nome, inicio_na_timeline, fim_na_timeline, arquivo
|
|
FROM clipes
|
|
WHERE video_id = ? AND inicio_na_timeline <= ? AND fim_na_timeline >= ?
|
|
LIMIT 1""",
|
|
(video_id, instante, instante),
|
|
).fetchone()
|
|
if linha is None:
|
|
return None
|
|
return {
|
|
"clipe_id": linha["id"], "faixa_id": linha["faixa_id"], "nome": linha["nome"],
|
|
"inicio": linha["inicio_na_timeline"], "fim": linha["fim_na_timeline"],
|
|
"arquivo": linha["arquivo"],
|
|
}
|
|
|
|
def listar_retakes_do_video(self, video_id: str) -> list[dict]:
|
|
"""
|
|
Lista o resumo dos grupos de retake de um vídeo.
|
|
|
|
Não traz as tomadas nem as evidências completas de cada grupo —
|
|
use ``consultar_detalhes_do_retake`` para descer a esse nível
|
|
quando o agente já tiver decidido qual grupo analisar.
|
|
|
|
Parâmetros:
|
|
video_id: Identificador do vídeo.
|
|
|
|
Retorna:
|
|
Lista de dicionários com o resumo de cada grupo (tipo,
|
|
confiança, total de tomadas, intervalo coberto e a tomada de
|
|
maior confiança). Lista vazia quando o vídeo não tem retakes.
|
|
|
|
Pode gerar:
|
|
ErroDeConsultaInvalida: quando ``video_id`` for vazio.
|
|
"""
|
|
self._validar_texto_obrigatorio("video_id", video_id)
|
|
|
|
grupos = self.conexao.execute(
|
|
"""SELECT id, tipo, confianca FROM grupos_de_retake
|
|
WHERE video_id = ? ORDER BY criado_em""",
|
|
(video_id,),
|
|
).fetchall()
|
|
resultado = []
|
|
for grupo in grupos:
|
|
extremos = self.conexao.execute(
|
|
"""SELECT COUNT(*) AS n, MIN(inicio) AS inicio, MAX(fim) AS fim
|
|
FROM tomadas_de_retake WHERE grupo_id = ?""",
|
|
(grupo["id"],),
|
|
).fetchone()
|
|
tomada_principal = self.conexao.execute(
|
|
"""SELECT fala_id, inicio, fim FROM tomadas_de_retake
|
|
WHERE grupo_id = ? ORDER BY confianca DESC LIMIT 1""",
|
|
(grupo["id"],),
|
|
).fetchone()
|
|
resultado.append({
|
|
"grupo_id": grupo["id"], "tipo": grupo["tipo"], "confianca": grupo["confianca"],
|
|
"total_tomadas": extremos["n"], "inicio": extremos["inicio"],
|
|
"fim": extremos["fim"],
|
|
"tomada_principal": (
|
|
{"fala_id": tomada_principal["fala_id"], "inicio": tomada_principal["inicio"],
|
|
"fim": tomada_principal["fim"]} if tomada_principal else None
|
|
),
|
|
})
|
|
return resultado
|
|
|
|
def consultar_detalhes_do_retake(self, grupo_id: str) -> dict | None:
|
|
"""
|
|
Consulta as tomadas e evidências completas de um grupo de retake específico.
|
|
|
|
Parâmetros:
|
|
grupo_id: Identificador do grupo de retake (ex.: ``retake_421fa7``).
|
|
|
|
Retorna:
|
|
Um dicionário com tipo, confiança, a lista completa de
|
|
tomadas (ordem, fala, intervalo, texto) e a lista completa de
|
|
evidências (tipo, descrição, valor) do grupo, ou None quando
|
|
o grupo não existe.
|
|
|
|
Pode gerar:
|
|
ErroDeConsultaInvalida: quando ``grupo_id`` for vazio.
|
|
"""
|
|
self._validar_texto_obrigatorio("grupo_id", grupo_id)
|
|
|
|
grupo = self.conexao.execute(
|
|
"SELECT id, video_id, faixa_id, tipo, confianca FROM grupos_de_retake WHERE id = ?",
|
|
(grupo_id,),
|
|
).fetchone()
|
|
if grupo is None:
|
|
return None
|
|
tomadas = [
|
|
{"ordem": t["ordem"], "fala_id": t["fala_id"], "inicio": t["inicio"], "fim": t["fim"],
|
|
"texto": t["texto"], "confianca": t["confianca"]}
|
|
for t in self.conexao.execute(
|
|
"""SELECT ordem, fala_id, inicio, fim, texto, confianca
|
|
FROM tomadas_de_retake WHERE grupo_id = ? ORDER BY ordem""",
|
|
(grupo_id,),
|
|
).fetchall()
|
|
]
|
|
evidencias = [
|
|
{"tipo": e["tipo"], "descricao": e["descricao"], "valor": e["valor"]}
|
|
for e in self.conexao.execute(
|
|
"SELECT tipo, descricao, valor FROM evidencias_de_retake WHERE grupo_id = ?",
|
|
(grupo_id,),
|
|
).fetchall()
|
|
]
|
|
return {
|
|
"grupo_id": grupo["id"], "tipo": grupo["tipo"], "confianca": grupo["confianca"],
|
|
"tomadas": tomadas, "evidencias": evidencias,
|
|
}
|
|
|
|
def listar_evidencias_visuais_do_clipe(
|
|
self,
|
|
video_id: str,
|
|
clipe_id: str,
|
|
tipo: str | None = None,
|
|
confianca_minima: float | None = None,
|
|
) -> list[dict]:
|
|
"""
|
|
Lista as evidências visuais de um clipe (qualidade, rosto, composição, tremor...).
|
|
|
|
Parâmetros:
|
|
video_id: Identificador do vídeo.
|
|
clipe_id: Identificador do clipe dentro da timeline do vídeo.
|
|
tipo: Quando informado, filtra só as evidências desse tipo
|
|
(ex.: ``"qualidade"``, ``"rosto"``). Recomendado sempre
|
|
que o agente já souber o que procura, para reduzir o
|
|
tamanho da resposta.
|
|
confianca_minima: Quando informado, descarta evidências com
|
|
confiança abaixo desse valor.
|
|
|
|
Retorna:
|
|
Lista de dicionários com tipo, intervalo, valor (payload
|
|
próprio de cada analisador), confiança e provider de cada
|
|
evidência. Lista vazia quando não há evidências que atendam
|
|
ao filtro.
|
|
|
|
Pode gerar:
|
|
ErroDeConsultaInvalida: quando ``video_id``/``clipe_id`` forem
|
|
vazios ou ``confianca_minima`` estiver fora de [0.0, 1.0].
|
|
"""
|
|
self._validar_texto_obrigatorio("video_id", video_id)
|
|
self._validar_texto_obrigatorio("clipe_id", clipe_id)
|
|
self._validar_confianca(confianca_minima)
|
|
|
|
condicoes = ["video_id = ?", "clipe_id = ?"]
|
|
parametros: list = [video_id, clipe_id]
|
|
self._acrescentar_se_definido(condicoes, parametros, "tipo = ?", tipo)
|
|
self._acrescentar_se_definido(
|
|
condicoes, parametros, "(confianca IS NULL OR confianca >= ?)", confianca_minima)
|
|
|
|
linhas = self.conexao.execute(
|
|
f"""SELECT tipo, inicio, fim, valor, confianca, provider
|
|
FROM evidencias_visuais
|
|
WHERE {self._clausula_where(condicoes)}
|
|
ORDER BY inicio""",
|
|
parametros,
|
|
).fetchall()
|
|
return [
|
|
{"tipo": linha["tipo"], "inicio": linha["inicio"], "fim": linha["fim"],
|
|
"valor": json.loads(linha["valor"]), "confianca": linha["confianca"],
|
|
"provider": linha["provider"]}
|
|
for linha in linhas
|
|
]
|
|
|
|
def listar_cenas_do_video(self, video_id: str, clipe_id: str | None = None) -> list[dict]:
|
|
"""
|
|
Lista os cortes de cena detectados no vídeo, ou só de um clipe específico.
|
|
|
|
Parâmetros:
|
|
video_id: Identificador do vídeo.
|
|
clipe_id: Quando informado, restringe o resultado às cenas
|
|
desse clipe. Quando None, traz as cenas do vídeo inteiro.
|
|
|
|
Retorna:
|
|
Lista de dicionários com clipe, intervalo e confiança de cada
|
|
cena, ordenados pela posição na timeline. Lista vazia quando
|
|
não há cenas detectadas.
|
|
|
|
Pode gerar:
|
|
ErroDeConsultaInvalida: quando ``video_id`` for vazio.
|
|
"""
|
|
self._validar_texto_obrigatorio("video_id", video_id)
|
|
|
|
condicoes = ["video_id = ?"]
|
|
parametros: list = [video_id]
|
|
self._acrescentar_se_definido(condicoes, parametros, "clipe_id = ?", clipe_id)
|
|
|
|
linhas = self.conexao.execute(
|
|
f"""SELECT clipe_id, inicio, fim, confianca FROM cenas
|
|
WHERE {self._clausula_where(condicoes)} ORDER BY inicio""",
|
|
parametros,
|
|
).fetchall()
|
|
return [
|
|
{"clipe_id": linha["clipe_id"], "inicio": linha["inicio"], "fim": linha["fim"],
|
|
"confianca": linha["confianca"]}
|
|
for linha in linhas
|
|
]
|
|
|
|
@staticmethod
|
|
def _clausula_where(condicoes: list[str]) -> str:
|
|
"""Monta a cláusula ``WHERE`` a partir das condições já validadas."""
|
|
return " AND ".join(condicoes)
|
|
|
|
@staticmethod
|
|
def _acrescentar_se_definido(
|
|
condicoes: list[str], parametros: list, fragmento_sql: str, valor: object,
|
|
) -> None:
|
|
"""
|
|
Acrescenta um filtro opcional à consulta, apenas quando o valor foi informado.
|
|
|
|
Centraliza o padrão repetido de "só filtra por este campo quando
|
|
o chamador passou um valor", usado por várias consultas deste
|
|
módulo para evitar duplicar a montagem do ``WHERE`` dinâmico.
|
|
|
|
Parâmetros:
|
|
condicoes: Lista de condições SQL já acumuladas (alterada
|
|
no local).
|
|
parametros: Lista de parâmetros posicionais já acumulados
|
|
(alterada no local).
|
|
fragmento_sql: Condição SQL com um único placeholder ``?``.
|
|
valor: Valor do filtro. Quando None, nada é acrescentado.
|
|
"""
|
|
if valor is not None:
|
|
condicoes.append(fragmento_sql)
|
|
parametros.append(valor)
|
|
|
|
@staticmethod
|
|
def _validar_texto_obrigatorio(nome_do_campo: str, valor: str) -> None:
|
|
"""
|
|
Garante que um identificador obrigatório foi informado e não é vazio.
|
|
|
|
Parâmetros:
|
|
nome_do_campo: Nome do parâmetro, usado na mensagem de erro.
|
|
valor: Valor recebido para o campo.
|
|
|
|
Pode gerar:
|
|
ErroDeConsultaInvalida: quando ``valor`` for None, vazio ou
|
|
composto só por espaços.
|
|
"""
|
|
if not valor or not str(valor).strip():
|
|
raise ErroDeConsultaInvalida(
|
|
f"O campo '{nome_do_campo}' é obrigatório e não pode ser vazio.")
|
|
|
|
@staticmethod
|
|
def _validar_intervalo(inicio: float, fim: float | None) -> None:
|
|
"""
|
|
Garante que um intervalo de tempo é consistente.
|
|
|
|
Parâmetros:
|
|
inicio: Início do intervalo, em segundos.
|
|
fim: Fim do intervalo, em segundos, ou None quando não há limite.
|
|
|
|
Pode gerar:
|
|
ErroDeConsultaInvalida: quando ``inicio`` for negativo ou
|
|
``fim`` for anterior a ``inicio``.
|
|
"""
|
|
if inicio < 0:
|
|
raise ErroDeConsultaInvalida("O início do intervalo não pode ser negativo.")
|
|
if fim is not None and fim < inicio:
|
|
raise ErroDeConsultaInvalida("O fim do intervalo não pode ser anterior ao início.")
|
|
|
|
@staticmethod
|
|
def _validar_confianca(confianca_minima: float | None) -> None:
|
|
"""
|
|
Garante que um limiar de confiança está dentro da faixa válida.
|
|
|
|
Parâmetros:
|
|
confianca_minima: Valor a validar, ou None quando não informado.
|
|
|
|
Pode gerar:
|
|
ErroDeConsultaInvalida: quando o valor estiver fora de
|
|
[0.0, 1.0].
|
|
"""
|
|
if confianca_minima is not None and not (0.0 <= confianca_minima <= 1.0):
|
|
raise ErroDeConsultaInvalida("confianca_minima deve estar entre 0.0 e 1.0.")
|