Ler os eventos sem varrer

Hoje o Pictor não empurra evento para você. Não há webhook de saída: quem pergunta é o seu lado. Esta página é como fazer isso sem estourar o teto e sem perder evento no meio.

A chamada

curl -s -H "Authorization: Bearer $PICTOR_KEY" \
  "$PICTOR_API/api/analytics/events?since=2026-09-03T15:00:00Z&limit=200"

Sem since, a janela padrão é de 24 horas. A resposta vem do mais recente para o mais antigo, limitada a limit (máximo 500).

O laço que funciona

A ideia é simples: guarde o carimbo do evento mais novo que você já tem e use-o como since na próxima volta.

from datetime import datetime, timedelta, timezone
import httpx

LIMITE = 200

def puxar(http: httpx.Client, desde: datetime) -> list[dict]:
    """Todos os eventos desde `desde` — inclusive quando são mais que `LIMITE`."""
    coletados, ate = [], None

    while True:
        params = {"since": desde.isoformat(), "limit": LIMITE}
        if ate is not None:
            params["until"] = ate.isoformat()

        r = http.get("/api/analytics/events", params=params)
        r.raise_for_status()
        pagina = r.json()
        if not pagina:
            break

        coletados.extend(pagina)

        # A janela transbordou: vieram exatamente `LIMITE` linhas, então há mais
        # coisa ANTES da última que chegou. Sem este passo, tudo o que não coube
        # some em silêncio — e é assim que uma integração perde a madrugada
        # movimentada e nunca descobre.
        if len(pagina) < LIMITE:
            break
        mais_antigo = datetime.fromisoformat(pagina[-1]["ts_start"])
        ate = mais_antigo - timedelta(microseconds=1)

    return coletados


def acompanhar(http: httpx.Client, ultimo_visto: datetime) -> datetime:
    eventos = puxar(http, desde=ultimo_visto)
    for e in sorted(eventos, key=lambda e: e["ts_start"]):
        processar(e)
        ultimo_visto = datetime.fromisoformat(e["ts_start"])
    return ultimo_visto

Se uma janela devolveu exatamente limit linhas, ela transbordou. Os eventos mais antigos daquela janela ficaram de fora, e nada na resposta avisa. Caminhe para trás com until até a página vir incompleta — é o laço acima.

Filtre no servidor, não depois

min_score e review_status entram na consulta antes do limite de linhas.

# Só o que a IA marcou com confiança alta:
"…/api/analytics/events?since=…&min_score=0.9"

# Só o que o operador ainda não julgou:
"…/api/analytics/events?since=…&review_status=unreviewed"

Filtrar do seu lado, sobre a página que chegou, produz um número que parece total e não é. “12 eventos acima de 90%” viraria “12 entre os 200 mais recentes”, não 12 no período.

review_status aceita unreviewed, true_positive, false_positive e reviewed.

As fotos

Cada evento traz snapshot_url e snapshot_annotated_url (o segundo com a marcação sobre a imagem). São endereços assinados e temporários.

Quem arquiva, arquiva os bytes. Guardar o link produz um dossiê que abre hoje e não abre em seis meses — e você só descobre no dia em que alguém precisa dele.

def arquivar(evento: dict) -> None:
    url = evento.get("snapshot_annotated_url") or evento.get("snapshot_url")
    if not url:
        return
    # Baixe AGORA, no mesmo laço em que você leu o evento.
    guardar_no_seu_storage(evento["event_id"], httpx.get(url).content)

Quanto pedir, e de quanto em quanto tempo

O teto é por chave e por minuto, e um 304 custa o mesmo que um 200 — não existe requisição de graça. Faça a conta antes de escrever o laço:

Cadência	Requisições/min por chave	Observação
a cada 5 min	0,2	folgado; suficiente para relatório e dossiê
a cada 1 min	1	a escolha normal para operação
a cada 10 s	6	só se alguém realmente age em 10 segundos
a cada 1 s	60	não faça: é vigília, não acompanhamento

Uma chamada por minuto por conta é praticamente nada dentro do teto — e cobre qualquer operação de portaria. O que estoura o teto não é a frequência do laço de eventos: é diagnosticar câmera por câmera (veja Saber por que a câmera não abre).

E quando houver push

Quando existir webhook de saída, o desenho do seu lado não muda: você continua com este laço como rede de segurança, e o webhook passa a antecipá-lo. Integração que só funciona por push perde tudo o que acontece durante uma queda de rede — e não tem como saber que perdeu.

Do zero ao vídeo na tela

Quando a câmera não abre

O que aconteceu

Montar a tela da portaria

A chamada

O laço que funciona

Filtre no servidor, não depois

As fotos

Quanto pedir, e de quanto em quanto tempo

E quando houver push

​A chamada

​O laço que funciona

​Filtre no servidor, não depois

​As fotos

​Quanto pedir, e de quanto em quanto tempo

​E quando houver push

A chamada

O laço que funciona

Filtre no servidor, não depois

As fotos

Quanto pedir, e de quanto em quanto tempo

E quando houver push