Revisar quadros com a visão do Claude e uma rubrica escrita

{
  "versao": "2026-10-12",
  "idioma_do_texto_na_tela": "pt-BR",
  "limiar_confianca": 0.85,
  "nota_limiar": "Regra de encaminhamento do exercício. A confiança declarada pelo modelo não é probabilidade calibrada.",
  "paleta": ["#1F2937", "#F59E0B", "#FFFFFF"],
  "categorias": [
    {
      "id": "legenda_cortada",
      "criterio": "Alguma letra da legenda toca ou ultrapassa a borda do quadro.",
      "gravidade": "alta",
      "ajuste": "Quebrar a linha ou reduzir o corpo da legenda na composição."
    },
    {
      "id": "texto_ilegivel",
      "criterio": "Texto na tela com contraste baixo, pequeno demais para celular ou sobre área movimentada.",
      "gravidade": "alta",
      "ajuste": "Aumentar corpo e contraste ou pôr uma faixa sólida atrás do texto."
    },
    {
      "id": "assunto_cortado",
      "criterio": "O prato ou a mão em ação sai do quadro quando deveria aparecer inteiro.",
      "gravidade": "media",
      "ajuste": "Reenquadrar o plano na composição ou escolher outro trecho do bruto."
    },
    {
      "id": "tela_preta",
      "criterio": "Quadro preto ou quase preto fora da abertura e do fechamento.",
      "gravidade": "alta",
      "ajuste": "Conferir o tempo de entrada da cena e o arquivo de origem no ponto marcado."
    },
    {
      "id": "quebra_de_paleta",
      "criterio": "Texto, fundo ou marca em cor fora da paleta declarada.",
      "gravidade": "baixa",
      "ajuste": "Trocar a cor pela da paleta declarada na composição."
    },
    {
      "id": "grafia_errada",
      "criterio": "Palavra do texto na tela sem acento ou com grafia errada em português.",
      "gravidade": "media",
      "ajuste": "Corrigir o texto no roteiro e na composição, nunca só no quadro."
    }
  ],
  "proibicoes": [
    "Não identificar pessoas pelo rosto.",
    "Não afirmar se a imagem foi gerada por IA.",
    "Não avaliar áudio, movimento ou sincronismo a partir de imagens paradas."
  ]
}

"""scripts/revisar_quadros.py: envia quadros rotulados ao Claude e recebe o QA num formulário fixo.

Uso, na raiz video-ia: python3 scripts/revisar_quadros.py qa/frames/qa11-fps
Requer: pip install anthropic, ANTHROPIC_API_KEY no ambiente e config/rubrica.json.
Triagem barata: QA_MODELO=claude-haiku-5-5 python3 scripts/revisar_quadros.py <pasta>
"""
import base64
import json
import math
import os
import subprocess
import sys
from pathlib import Path

import anthropic

# O revisor padrão é o Sonnet 5.5; QA_MODELO=claude-haiku-5-5 faz a triagem.
MODELO = os.environ.get("QA_MODELO", "claude-sonnet-5-5")
# US$ por milhão de tokens de entrada, tabela de 12/10/2026 (Haiku 5.5: até 100 mil tokens de prompt).
PRECOS_ENTRADA = {"claude-sonnet-5-5": 2.0, "claude-haiku-5-5": 0.10}
PRECO_ENTRADA_POR_MILHAO = PRECOS_ENTRADA[MODELO]
MAX_IMAGENS = 40                 # política do exercício (a API aceita até 600 por requisição)
LIMITE_IMAGEM_B64 = 10_000_000   # 10 MB por imagem na API direta (5 MB no Bedrock e no Google Cloud)
LIMITE_REQUISICAO = 30_000_000   # margem abaixo dos 32 MB por requisição
TIPOS = {".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".png": "image/png", ".webp": "image/webp"}

pasta = Path(sys.argv[1] if len(sys.argv) > 1 else "qa/frames/qa11-fps")
rubrica = json.loads(Path("config/rubrica.json").read_text(encoding="utf-8"))
quadros = sorted(p for p in pasta.iterdir() if p.suffix.lower() in TIPOS)[:MAX_IMAGENS]
if not quadros:
    raise SystemExit(f"Nenhum quadro em {pasta}.")
arq_tempos = pasta / "tempos.txt"
tempos = arq_tempos.read_text().split() if arq_tempos.exists() else []


def dimensoes(caminho):
    saida = subprocess.run(
        ["ffprobe", "-v", "error", "-select_streams", "v:0",
         "-show_entries", "stream=width,height", "-of", "json", str(caminho)],
        check=True, capture_output=True, text=True,
    ).stdout
    s = json.loads(saida)["streams"][0]
    return s["width"], s["height"]


conteudo, manifesto, tokens_visuais = [], [], 0
for numero, caminho in enumerate(quadros, start=1):
    largura, altura = dimensoes(caminho)
    if len(quadros) > 20 and max(largura, altura) > 2000:
        raise SystemExit(f"{caminho}: com mais de 20 imagens, cada lado fica em até 2000 px.")
    tokens_visuais += math.ceil(largura / 28) * math.ceil(altura / 28)
    dados = base64.standard_b64encode(caminho.read_bytes()).decode("ascii")
    if len(dados) >= LIMITE_IMAGEM_B64:
        raise SystemExit(f"{caminho}: passa de 10 MB em base64.")
    segundo = tempos[numero - 1] if numero <= len(tempos) else "desconhecido"
    manifesto.append({"quadro": numero, "arquivo": str(caminho), "segundo": segundo,
                      "largura": largura, "altura": altura})
    # Rótulo ANTES de cada imagem: é assim que o relatório consegue citar o quadro certo.
    conteudo.append({"type": "text", "text": f"Quadro {numero} (aprox. {segundo} s):"})
    conteudo.append({"type": "image", "source": {
        "type": "base64", "media_type": TIPOS[caminho.suffix.lower()], "data": dados}})

ids = [c["id"] for c in rubrica["categorias"]]
conteudo.append({"type": "text", "text": (
    "Faça o QA visual somente destas imagens, seguindo a rubrica do sistema. "
    f"Responda chamando a ferramenta registrar_qa com os {len(quadros)} quadros, "
    "cada um exatamente uma vez, e problemas vazio quando não houver defeito."
)})

# Saída estruturada: um formulário com campos fixos em vez de texto livre.
ferramenta = {
    "name": "registrar_qa",
    "description": "Registra o resultado do QA visual, um item por quadro recebido.",
    "strict": True,   # a entrada segue o esquema; o enum deixa de ser só orientação
    "input_schema": {
        "type": "object",
        "properties": {"quadros": {"type": "array", "items": {
            "type": "object",
            "properties": {
                "quadro": {"type": "integer"},
                # O modo estrito não aceita minimum/maximum: a faixa 0 a 1 vai na descrição
                # e o avaliar_qa.py confere.
                "confianca": {"type": "number", "description": "Autoavaliação de 0 a 1."},
                "problemas": {"type": "array", "items": {
                    "type": "object",
                    "properties": {
                        "categoria": {"type": "string", "enum": ids},
                        "gravidade": {"type": "string", "enum": ["alta", "media", "baixa"]},
                        "evidencia": {"type": "string"},
                    },
                    "required": ["categoria", "gravidade", "evidencia"],
                    "additionalProperties": False,
                }},
            },
            "required": ["quadro", "confianca", "problemas"],
            "additionalProperties": False,
        }}},
        "required": ["quadros"],
        "additionalProperties": False,
    },
}
sistema = (
    "Você é revisor de QA visual de vídeo. Rubrica em JSON:\n"
    + json.dumps(rubrica, ensure_ascii=False)
    + "\nA confiança é uma autoavaliação de 0 a 1, não uma probabilidade calibrada."
)

tamanho = len(json.dumps(conteudo)) + len(sistema)
if tamanho >= LIMITE_REQUISICAO:
    raise SystemExit("Requisição perto dos 32 MB: divida o lote.")

Path("qa").mkdir(exist_ok=True)
Path("qa/qa11-quadros.json").write_text(
    json.dumps(manifesto, ensure_ascii=False, indent=2), encoding="utf-8")

if MODELO == "claude-haiku-5-5" and tokens_visuais > 90_000:
    raise SystemExit("Triagem perto do degrau de 100 mil tokens do Haiku 5.5: divida o lote.")
custo = tokens_visuais * PRECO_ENTRADA_POR_MILHAO / 1_000_000
print(f"{MODELO}: {len(quadros)} imagens, {tokens_visuais} tokens visuais, US$ {custo:.4f} de entrada visual.")
print("A estimativa não inclui o texto, a rubrica nem os tokens de saída.")
if input("Autoriza esta chamada paga? Digite SIM: ").strip() != "SIM":
    raise SystemExit("Chamada cancelada; quadros e manifesto preservados.")

cliente = anthropic.Anthropic()   # lê ANTHROPIC_API_KEY do ambiente
resposta = cliente.messages.create(
    model=MODELO,
    max_tokens=8000,
    system=sistema,
    tools=[ferramenta],   # tool_choice fica no padrão (auto): o Sonnet 5.5 recusa chamada forçada
    messages=[{"role": "user", "content": conteudo}],
)
print("Uso informado pela API:", resposta.usage)

if resposta.stop_reason == "max_tokens":
    raise SystemExit("Resposta truncada: encaminhe para revisão humana ou divida o lote.")
bloco = next((b for b in resposta.content if b.type == "tool_use"), None)
if bloco is not None:
    relatorio = bloco.input
else:
    # Plano B: o modelo respondeu em texto. Tenta ler JSON; se não der, revisão humana.
    texto = "".join(b.text for b in resposta.content if b.type == "text")
    try:
        relatorio = json.loads(texto)
    except json.JSONDecodeError:
        raise SystemExit("Resposta fora do formato: encaminhe para revisão humana.")

Path("qa/qa11-relatorio.json").write_text(
    json.dumps(relatorio, ensure_ascii=False, indent=2), encoding="utf-8")
print("Relatório salvo em qa/qa11-relatorio.json")

"""scripts/avaliar_qa.py: confere cobertura e formato do QA. Nunca autoriza publicação sozinho."""
import json
import math
from pathlib import Path

rubrica = json.loads(Path("config/rubrica.json").read_text(encoding="utf-8"))
esperados = json.loads(Path("qa/qa11-quadros.json").read_text(encoding="utf-8"))
relatorio = json.loads(Path("qa/qa11-relatorio.json").read_text(encoding="utf-8"))
LIMIAR = rubrica["limiar_confianca"]          # 0.85, regra do exercício
CATEGORIAS = {c["id"].lower() for c in rubrica["categorias"]}

itens = relatorio.get("quadros") if isinstance(relatorio, dict) else None
motivos = []
if not isinstance(itens, list):
    itens, motivos = [], ["Campo quadros ausente ou inválido."]

recebidos = [i.get("quadro") for i in itens if isinstance(i, dict)]
if sorted(map(str, recebidos)) != sorted(str(q["quadro"]) for q in esperados):
    motivos.append("Cobertura incompleta, quadro repetido ou número inesperado.")

for item in itens:
    if not isinstance(item, dict):
        motivos.append("Item de quadro inválido.")
        continue
    q, conf, probs = item.get("quadro"), item.get("confianca"), item.get("problemas")
    if type(conf) not in (int, float) or not math.isfinite(conf) or not 0 <= conf <= 1:
        motivos.append(f"Quadro {q}: confiança inválida.")
    elif conf < LIMIAR:
        motivos.append(f"Quadro {q}: confiança {conf} abaixo de {LIMIAR}.")
    if not isinstance(probs, list):
        motivos.append(f"Quadro {q}: lista de problemas inválida.")
        continue
    for p in probs:
        # A caixa das letras do enum não é garantida: compara em minúsculas.
        if not isinstance(p, dict) or str(p.get("categoria")).lower() not in CATEGORIAS:
            motivos.append(f"Quadro {q}: categoria fora da rubrica.")
        else:
            motivos.append(f"Quadro {q}: {p['categoria']} ({p.get('gravidade')}): {p.get('evidencia')}")

decisao = {
    "estado": "revisao_humana_prioritaria" if motivos else "sem_achados_na_amostra_aguarda_revisao_humana",
    "motivos": motivos,
    "publicacao_autorizada": False,   # sempre falso: quem autoriza é uma pessoa
    "observacao": "A amostra não comprova áudio, sincronismo nem continuidade entre quadros.",
}
Path("qa/qa11-decisao.json").write_text(
    json.dumps(decisao, ensure_ascii=False, indent=2), encoding="utf-8")
print(json.dumps(decisao, ensure_ascii=False, indent=2))

#!/usr/bin/env bash
# Rode na raiz video-ia, depois de extrair a amostra em qa/frames/qa11-fps.
set -euo pipefail

# 1. Triagem barata: o mesmo script, com o Haiku 5.5.
QA_MODELO=claude-haiku-5-5 python3 scripts/revisar_quadros.py qa/frames/qa11-fps

# 2. Guarda a triagem e copia os quadros duvidosos para qa/frames/qa11-marcados.
python3 scripts/separar_marcados.py

# 3. Revisor padrão (Sonnet 5.5) só nos marcados; depois, o validador de sempre.
python3 scripts/revisar_quadros.py qa/frames/qa11-marcados
python3 scripts/avaliar_qa.py

"""scripts/separar_marcados.py: guarda a triagem do Haiku 5.5 e separa os quadros para o Sonnet 5.5.

Rode na raiz video-ia logo depois da triagem. Não chama API nenhuma.
"""
import json
import shutil
from pathlib import Path

rubrica = json.loads(Path("config/rubrica.json").read_text(encoding="utf-8"))
LIMIAR = rubrica["limiar_confianca"]
qa = Path("qa")
destino = qa / "frames" / "qa11-marcados"
if destino.exists():
    raise SystemExit(f"{destino} já existe: arquive a triagem anterior.")

# A triagem vira registro próprio; a rodada do Sonnet grava os arquivos padrão de novo.
(qa / "qa11-relatorio.json").rename(qa / "qa11-triagem-relatorio.json")
(qa / "qa11-quadros.json").rename(qa / "qa11-triagem-quadros.json")
relatorio = json.loads((qa / "qa11-triagem-relatorio.json").read_text(encoding="utf-8"))
lista = json.loads((qa / "qa11-triagem-quadros.json").read_text(encoding="utf-8"))
manifesto = {q["quadro"]: q for q in lista}

itens = relatorio.get("quadros", []) if isinstance(relatorio, dict) else []
vistos = {i.get("quadro") for i in itens if isinstance(i, dict)}
marcados = set(manifesto) - vistos   # quadro que a triagem pulou vai para o Sonnet
for item in itens:
    if not isinstance(item, dict):
        continue
    conf = item.get("confianca")
    if item.get("problemas") or type(conf) not in (int, float) or conf < LIMIAR:
        marcados.add(item.get("quadro"))
marcados &= set(manifesto)

if not marcados:
    raise SystemExit("Nenhum quadro marcado: siga para a pessoa que assiste com áudio.")
destino.mkdir(parents=True)
tempos = []
for n in sorted(marcados):
    origem = Path(manifesto[n]["arquivo"])
    shutil.copy2(origem, destino / f"m_{n:03d}{origem.suffix}")
    tempos.append(str(manifesto[n]["segundo"]))
(destino / "tempos.txt").write_text("\n".join(tempos) + "\n", encoding="utf-8")
print(f"{len(marcados)} de {len(manifesto)} quadros marcados em {destino}.")