""" 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.")