diff --git a/docs/plans/backfill-acervo-falas-impl.md b/docs/plans/backfill-acervo-falas-impl.md new file mode 100644 index 00000000..d882d9a0 --- /dev/null +++ b/docs/plans/backfill-acervo-falas-impl.md @@ -0,0 +1,248 @@ +# Impl: Backfill 2011–2026 + import de produção do acervo de falas + +Status: em execução +Atualizado em: 2026-09-13 +Issue: #957 +Intenção: docs/plans/backfill-acervo-falas.md +Appetite restante: herdado (~1–2 dias + tempo de máquina) + +## Leitura da intenção + +**Outcome.** As 54ª–57ª legislaturas processadas pelo pipeline do C153, gravadas no banco de produção (`teqo_1313`), com relatório honesto de cobertura por legislatura: total, com trecho, com vídeo, com segmentos, sem trecho, fallback e falhas — e links do VOD validados numa amostra. Idempotência comprovada (reexecução não recria, não re-transcreve). O valor é o histórico: "o Solla sempre defendeu isso" só existe com 2011–2023 no catálogo. + +**O que NÃO negociar.** + +- Guardrail de escrita: só `speech`/`speechSegment` (+ rels de município); nenhum objeto de mídia no S3; crédito CC BY 4.0 mantido no runbook/changelog. +- Idempotência por `sourceKey` (C153) — reexecutar é seguro e é o mecanismo de resume. +- Relatório sem maquiar falha como sucesso: `withoutExcerpt`/`failed` aparecem por legislatura, não somem no total. +- Plenário apenas; nada de UI; nenhuma migration/schema novo (se surgir necessidade, justificar no PR). +- Produção é dado real: o caminho é o runbook, com guard explícito antes de qualquer escrita. + +**O que reavaliar.** + +- _"Publicar antes ou depois do C154"_: o C154 **não está em main** (changelog até 2026-09-13 só tem C153). A importação não depende do C154 (dados não quebram nada), mas a validação visual do acervo é do C154; se ele não estiver no deploy de main da Fase 3, isso fica registrado como pendência, não como bloqueio (o aceite deste item é dado + cobertura). +- _"54ª incluída"_: sim — 1 discurso, custo ~zero, e é exatamente o caso "fonte antiga sem trecho" que o relatório precisa mostrar (trechos por orador só existem de 2015 em diante, C152). +- _"Fallback YouTube para sessões sem trecho"_: o C153 já grava `youtubeUrl` do evento; "fallback" aqui é **contabilizar** esses casos na cobertura (`fallbackYoutube`), não transcrever outra fonte — isso continua fora de escopo. +- _Orçamento do D8_: os ~777 são "novos além da prova local da 57ª"; em produção o banco parte vazio, então as ~1.011 passam pelo pipeline (a 57ª re-transcreve). O relatório mede e o custo declarado é teto, não promessa. + +## Abordagem recomendada + +Evoluir o dono do pipeline (`scripts/import-camara-speeches.mjs` + `scripts/lib/*`) com três modos novos — `--all`, `--coverage`, `--verify-links` —, reusar os módulos existentes e operar no homeserver contra o banco de produção **depois** do deploy que aplica a migration do C153 e **antes** do merge do PR, com a prova viva anexada (precedente OPS79). + +**Reuso (depth check):** `withPayloadTransaction`/`upsertSpeechBundle` (`src/utilities/speech/speechImport.ts`) para a escrita; `camaraSpeeches.mjs` (puro) e `camaraFetch.mjs` (HTTP/VOD/ASR) para o pipeline; `assertLocalDatabase` + `isTruthyEnv`/`dieWithLabel`/`ensureCachedDownload` de `scripts/lib/cli.mjs`; `recover-media.mjs` como molde de guard (`MEDIA_RECOVER_CONFIRM`), eco de alvo e artefato; `ops79:migrate` como molde de confirmação (`OPS79_MIGRATE_CONFIRM`) e de operação pré-merge; `docs/ops/teqo-1313-deploy.md` §OPS79/§OPS84 como molde de seção de runbook; `drizzleResultRows` + SQL agregado (precedente `supporter/supporterListOverviewAggregate.ts`) para a coverage. + +```mermaid +flowchart TB + subgraph F1["Fase 1 — código (branch C155)"] + A["import-camara-speeches.mjs
--all · --coverage · --verify-links
CAMARA_IMPORT_CONFIRM"] + B["camaraSpeeches.mjs puro
aggregateBackfillRuns · selectLinkSample"] + C["speechCoverage.ts
SQL agregado read-only"] + A --> B + A --> C + end + subgraph F0["Fase 0 — pré-requisito de schema"] + D["deploy manual de main
verify full, staging, aprovacao
aplica migrations C153"] + end + subgraph F3["Fase 3 — homeserver (tmux + tee)"] + E["~/teqo-backfill no SHA do branch C155
env teqo-1313 + proxy 5433"] + F["camara:import --all
54 a 57, ~1.011, 10 a 15h"] + G["--coverage"] + H["--verify-links 3"] + end + I[("producao teqo_1313
speech · speech_segment")] + K["vod.camara.leg.br / dadosabertos"] + J["PR C155 (Closes #957)
resumo + artefatos anexados"] + D --> E --> F + A -.-> E + F -->|upsert transacional| I + F --> G --> H + G -->|read-only| I + H -->|GET Range| K + F -->|API eventos/VOD/ASR| K + F --> J + G --> J + H --> J +``` + +| Opções | Recomendação | Rejeitadas | +| --- | --- | --- | +| **A)** Evoluir o dono: `--all`/`--coverage`/`--verify-links` no import do C153 + utility de coverage + docs | **A** — um CLI, um artefato, caches em memória, agregação pura testável; knip e convenções de script continuam com um dono | **B)** script novo de orquestração + script novo de coverage (segundo CLI/knip/drift, dois lugares para o mesmo bug); **C)** operar só com loop shell no runbook (sem relatório por legislatura como artefato, operador vira o agregador) | + +### Decisões de engenharia + +#### D1 — Modo `--all` no dono do pipeline (barato; a decisão que importa é não criar twin) + +- **Opções:** A) `--all` no `import-camara-speeches.mjs`, loop 54→57 num processo, relatório combinado com `legislatures[]` | B) script novo `backfill-camara-speeches.mjs` chamando o import por legislatura | C) loop shell no runbook concatenando relatórios à mão. +- **Recomendação:** A — o dono do pipeline é o import; um processo preserva os caches `eventsByDate`/`eventPages` entre legislaturas, emite **um** JSON com totais combinados e bucket por legislatura, e mantém `--legislature` como retry direcionado. `--all` é mutuamente exclusivo com `--date` e com `--legislature` explícito; `--limit`/`--skip-transcribe`/`--reclassify` valem por legislatura. Checkpoint do JSON a cada legislatura (perda limitada a uma legislatura se a rede cair); falha dura em uma legislatura registra `aborted` e segue para a próxima, com exit 1 no fim. +- **Rejeitadas:** B — segundo CLI para o mesmo pipeline = drift de flags, mais um entry em `knip.json` (`exports:error`) e duas cópias das mesmas regras; C — sem artefato por legislatura (aceite pede), sem totais combinados e sem caches. + +#### D2 — Relatório = métricas do run + cobertura do DB (caro: contrato do artefato; nomes de campo são fill-in) + +- **Opções:** A) ao final do `--all`, anexar `coverage` lida do banco por legislatura + modo read-only `--coverage` (sem rede, sem escrita no banco) | B) só o relatório do run | C) script separado `camara-coverage.mjs`. +- **Recomendação:** A — run report conta o que a execução **fez**; coverage conta o que o banco **tem**. Reexecução resumida/parcial torna o run report não representativo do estado, e o aceite pede o retrato do acervo ("quantos têm vídeo, quantos caíram no fallback, o que ficou de fora e por quê"). `getSpeechCoverage(payload)` em `src/utilities/speech/speechCoverage.ts` com **uma** query agregada (`speech LEFT JOIN speech_segment`, `COUNT(*) FILTER`) via `drizzleResultRows`; o `--coverage` reusa a mesma função e é a verificação pós-run em produção. +- **Rejeitadas:** B — não distingue "pulou porque já tinha" de "não tem", não responde fallback; C — twin do script (mesmo problema do D1-B) e o modo read-only ficaria fora do dono. + +#### D3 — Guard de escrita em produção (caro de reverter: semântica fail-closed) + +- **Opções:** A) `CAMARA_IMPORT_CONFIRM=1` exigido em todo modo que escreve quando `NODE_ENV === 'production'` **ou** host do DB não-local **ou** `ALLOW_REMOTE_DB` setado; `--coverage`/`--verify-links` isentos (read-only); `--skip-transcribe` **não** isenta (escreve metadados) | B) flag CLI nova (`--yes`/`--force`) | C) confiar só no `assertLocalDatabase` + `ALLOW_REMOTE_DB` | D) exigir a flag sempre, inclusive local. +- **Recomendação:** A — precedente literal de `MEDIA_RECOVER_CONFIRM` (`scripts/recover-media.mjs:68,79-85`) e `OPS79_MIGRATE_CONFIRM` (`scripts/migrate-signature-orphan.mjs:46,54-60`). O guard roda **antes** de `getPayload` (nenhuma conexão, nenhuma escrita), ecoa o alvo (host do DB, modo, range) como o recover-media, e o predicado puro `requiresWriteConfirm({ nodeEnv, databaseUrl, allowRemoteDb })` mora em `scripts/lib/cli.mjs` (junto de `LOCAL_HOSTS`, onde a convenção manda) e é unit-testado. O `assertLocalDatabase` permanece como cinto para host remoto explícito, com o hint atualizado para apontar o runbook §C155. +- **Rejeitadas:** B — flag por invocação não sobrevive a wrapper/alias e não descreve o alvo; o repo já tem o vocabulário de env flag; C — o furo nomeado: `LOCAL_HOSTS` inclui `postgres` (host do env file do homeserver) e o runbook usa o proxy socat em `127.0.0.1:5433` — um run acidental em produção **passa silencioso**; D — quebra o DX local (C153 usa `pnpm camara:import` sem cerimônia) sem ganho: o risco é o alvo remoto, não o comando. + +#### D4 — Onde e quando o backfill roda (caro: escreve dado real; a ordem é irreversível sem rollback) + +- **Opções:** A) homeserver, checkout separado `~/teqo-backfill` no SHA do branch C155, **depois** de um deploy manual de main (aplica a migration do C153) e **antes** do merge do PR, com relatório/prova no PR (precedente OPS79) | B) rodar local (`teqo_wt155`) e copiar tabelas para produção | C) rodar só local e adiar produção | D) rodar do `~/teqo-deploy`. +- **Recomendação:** A — produção está sem as tabelas `speech*` (migration para no batch `20260912_224719_add_activity_public_event`) e sem a revisão C153; o deploy de main é pré-requisito de schema; o checkout separado não contamina o workspace do runner (que segue `main` e é gerenciado pelo deploy); operação antes do merge é o precedente OPS79 — o código do operador não precisa estar publicado para escrever dados. Sequência do runbook: deploy → `git fetch`/checkout do SHA → `pnpm install` → env → tmux. +- **Rejeitadas:** B — ids de `municipality`/rels divergem entre `teqo_wt155` e `teqo_1313` (o upsert resolve slug→id **por banco**), cópia cross-DB artesanal de 3 tabelas sem transação e ASR pago duas vezes; C — falha o aceite "dados disponíveis em produção"; D — o workspace do runner é do deploy; misturar branch de feature contamina o próximo dispatch. + +#### D5 — Validação de links do VOD (barato; tamanho/forma da amostra é fill-in com gatilho) + +- **Opções:** A) read-only `--verify-links `: n discursos por legislatura (amostragem determinística por índice, estável por `speechAt`+id), playback e download, GET com `Range: bytes=0-1023` e fallback HEAD em 405/501, timeout curto; não-200 = **warning** no artefato; 0 links verificados = exit 1 | B) curl manual no runbook | C) não validar | D) validar os 1.011×2 links. +- **Recomendação:** A — reproduzível, vira artefato e cabe no aceite com `n=3` (12 discursos, ~24 probes). Warning em vez de falha dura porque fonte antiga legitimamente não tem trecho e 404 isolado não invalida o acervo; o fail-closed em "0 verificados" evita verde vazio com rede caída. Implementa `probeVodLink(url)` em `camaraFetch.mjs` e `selectLinkSample(rows, n)` puro em `camaraSpeeches.mjs`. +- **Rejeitadas:** B — não reproduzível e não agrega ao relatório; C — aceite pede explicitamente; D — 2.022 requests sem ganho sobre amostra por legislatura. + +#### D6 — Artefato do relatório (barato; forma de guarda é fill-in) + +- **Opções:** A) JSONs em `data/camara/reports` (gitignored, `.gitignore:77`) + resumo no changelog + seção nova no runbook + anexo do JSON à Issue #957; **nada commitado** | B) commitar o JSON (ou recorte) no repo | C) só changelog | D) relatório em serviço externo. +- **Recomendação:** A — `data/camara` é gitignored por contrato; o run completo tem `speeches[]` de ~1.011 entradas (grande e regenerável); o runbook é dono do como-fazer, o changelog do o-que-aconteceu. O resumo de coverage (pequeno) vai no corpo do PR; o JSON completo fica no homeserver (`/srv/hdd/teqo-backfill/camara/reports`) e anexado à Issue (upload web). Arquivos: `import-all-.json`, `coverage-.json`, `verify-links-.json`. +- **Rejeitadas:** B — ruído de repo, `format`/knip e conflito garantido a cada reexecução; C — perde detalhe por legislatura/falhas; D — dependência externa sem necessidade. + +#### D7 — Testes proporcionais (barato; o corte é fill-in) + +- **Opções:** A) unit dos puros (`aggregateBackfillRuns`, `selectLinkSample`, `requiresWriteConfirm`/`isLocalDatabaseUrl`) + int de `speechCoverage` com fixtures | B) e2e/CLI com rede | C) nenhum teste novo (o script não tinha) | D) extrair e testar `parseArgs` inteiro. +- **Recomendação:** A — a aritmética do relatório e a amostragem são puras e falham em silêncio sem teste; o guard é segurança e vira predicado puro em `scripts/lib/cli.mjs`, unit-testado ao lado de `cliEnvFlags.unit.spec.ts`; a coverage é SQL agregado, provado por int com fixtures cobrindo legislaturas diferentes, com/sem trecho, com/sem segmento e fallback YouTube (`tests/int/speechCoverage.int.spec.ts`, padrão do `speechImport.int.spec.ts`). Somar aos 31 unit do módulo Câmara. +- **Rejeitadas:** B — flaky, custo real de ASR e sem retorno sobre o unit; C — agregação pode somar errado sem ninguém ver; D — extrair a CLI inteira para testar parsing trivial é desproporcional (fill-in futuro se `parseArgs` crescer). + +#### D8 — Forma da execução longa (barato; wall clock é física) + +- **Opções:** A) um processo `--all` sequencial, desatendido em tmux/nohup com `tee`, checkpoint por legislatura, reexecução idempotente como resume | B) paralelizar discursos | C) uma sessão por legislatura | D) chunk por `--date`/ano. +- **Recomendação:** A — ~1.011 × ~46s ≈ 13h de parede (10–15h com overhead), sem rate-limit storm contra a Câmara/Deep Infra; caches em memória e um artefato. Reexecutar `--all` (ou `--legislature N`) pula ASR/LLM do que já está no banco (`segmentCount > 0` e `classifiedBy === 'llm'`). Orçamento declarado: ASR ≈ US$0,45–0,60 (Whisper large-v3 a US$0,00045/min; C152 mediu US$0,0006–0,0021/discurso) + LLM ≈ US$0,07, medidos no relatório. A 57ª re-transcreve em produção (banco vazio; segmentos locais não são transferidos — D4-B). +- **Rejeitadas:** B — concorrência sem necessidade, retry storms e relatório de falhas confuso; C — perde caches e agregação combinada, operador vira orquestrador; D — fragmenta o relatório sem mudar custo. + +### Componentes / mudanças + +| Arquivo | Mudança | +| --- | --- | +| `scripts/import-camara-speeches.mjs` (dono) | Flags `--all` (mutex com `--date`/`--legislature` explícito) e modos `--coverage`/`--verify-links `; loop 54→57 com checkpoint JSON por legislatura; `coverage` anexada ao relatório final; guard `CAMARA_IMPORT_CONFIRM` (D3) antes de `getPayload`; eco do alvo; falha por legislatura não aborta o run inteiro; `HELP` atualizado; hint do `assertLocalDatabase` aponta o runbook §C155 | +| `scripts/lib/camaraSpeeches.mjs` (puro) | `aggregateBackfillRuns(runs)` (somas de `totals`/`asr`/`llm`/`elapsedMs`, por legislatura e combinado) e `selectLinkSample(rows, n)` (amostragem determinística) | +| `scripts/lib/camaraFetch.mjs` (HTTP) | `probeVodLink(url)` — GET `Range: bytes=0-1023`, fallback HEAD, status/content-type/bytes, erro de rede = warning | +| `scripts/lib/cli.mjs` (CLI) | `isLocalDatabaseUrl(url)` e `requiresWriteConfirm({ nodeEnv, databaseUrl, allowRemoteDb })` puros; `assert-local-database.mjs` passa a consumir `isLocalDatabaseUrl` (refactor mínimo, semântica intacta) | +| `src/utilities/speech/speechCoverage.ts` (novo) | `getSpeechCoverage(payload)` — query agregada por legislatura: `total`, `withExcerpt` (`audio_id`), `withVideo` (`vod_playback_url`), `withSegments` (`speech_segment` > 0), `withoutExcerpt`, `withoutSegments`, `fallbackYoutube` (`audio_id` nulo e `youtube_url` presente) + totais | +| `tests/unit/camaraSpeeches.unit.spec.ts` | + unit de `aggregateBackfillRuns`/`selectLinkSample` | +| `tests/unit/cliEnvFlags.unit.spec.ts` | + unit de `requiresWriteConfirm`/`isLocalDatabaseUrl` | +| `tests/int/speechCoverage.int.spec.ts` (novo) | fixtures de `speech`/`speechSegment` cobrindo os buckets; cleanup no `afterAll` | +| `docs/ops/teqo-1313-deploy.md` | Seção `## C155 — backfill do acervo de falas em produção`: pré-requisito (deploy com migration C153), comandos, resultado e rollback | +| `docs/changelog/2026-09-13-c155.md` (novo) | Entrada curta com os números medidos (preenchida na Fase 4) | + +Sem migration, sem schema, sem access, sem UI, sem `package.json` (o `pnpm camara:import` já existe). Nenhuma escrita fora de `speech`/`speechSegment`/`speech_rels`. + +### Dados → forma (relatório) + +**Coverage (por legislatura + totais):** + +| Campo | Definição | Responde | +| --- | --- | --- | +| `total` | discursos da legislatura no banco | tamanho do acervo | +| `withExcerpt` | `audio_id` não nulo | trecho identificado na página do evento | +| `withVideo` | `vod_playback_url` não nulo | tem vídeo (VOD) | +| `withSegments` | ≥1 `speechSegment` | tem transcrição minutada | +| `withoutExcerpt` | `audio_id` nulo | o que ficou de fora | +| `withoutSegments` | 0 segmentos | pendência de transcrição | +| `fallbackYoutube` | `audio_id` nulo e `youtube_url` presente | "quantos caíram no fallback" | + +**`--all` (JSON):** `{ runAt, mode: 'all', options, legislatures: [{ legislature, range, totals, asr, llm, elapsedMs, aborted, speeches[], failures[] }], totals/asr/llm/elapsedMs combinados, coverage }`. O modo `--legislature`/`--date` mantém o shape do C153 e ganha `coverage` no fim (compatível com o artefato existente). + +**Impressão (stdout):** bloco por legislatura (`listados · processados (criados/atualizados) · com trecho/sem trecho · com segmentos · ASR min e US$ · LLM tokens e US$ · tempo · falhas`), linha combinada, tabela de coverage e falhas listadas com `speechAt`/`stage`/mensagem. + +**`--verify-links` (JSON):** `{ runAt, sample: { perLegislature, total }, checked: [{ legislature, speechId, kind: 'playback'|'download', url, status, ok, note }], ok, warnings, elapsedMs }`. Não-200 entra em `warnings`; `checked.length === 0` → exit 1. + +**Destino:** resumo de coverage + verify-links no corpo do PR e no changelog; JSON completo do run anexado à Issue #957; nada em git; runbook registra os números e o rollback. + +## Fases verificáveis + +**Fase 0 — Pré-flight (0,5h, zero escrita).** +Confirmar no homeserver: revision do container `teqo-1313`; `payload_migrations` com `20260913_001112_add_speech_catalog` e `20260913_001200_add_speech_segment_trgm_index`; tabelas `speech`/`speech_segment`/`speech_rels` existindo e vazias. Checar C154 em main (registrar se ausente). Garantir `DEEPINFRA_API_KEY` (em `~/stack/.env` ou adicionar a `~/stack/teqo-1313.env`, chmod 600) — o modo de escrita sem `--skip-transcribe` passa a falhar cedo sem a chave. Definir `--out /srv/hdd/teqo-backfill/camara` (394GB; não encher `/`). +_Prova:_ saída dos dois checks de migration/tabela + chave presente (sem eco). + +**Fase 1 — Código (0,5–1 dia).** +Implementar D1–D3, D5, D7. Smoke local em `teqo_wt155`: `pnpm camara:import --all --limit 2` (7 discursos, ASR real, ~6–8 min) → relatório com 4 buckets + coverage; reexecutar o mesmo comando → **0 criados, 0 ASR** (idempotência viva); `--coverage` sem rede; `--verify-links 1`. +_Quota:_ sem migration; ~120 linhas somadas entre script/lib/utility; `pnpm gate:fast` verde; unit novos verdes; int de coverage verde. + +**Fase 2 — Docs (0,5h).** +Seção do runbook (pré-requisito, comandos, rollback) + changelog stub (números reais entram na Fase 4). +_Quota:_ ≤ ~80 linhas de runbook; `pnpm format`/lint verdes. + +**Fase 3 — Ops no homeserver (≈1h preparo + 10–15h máquina + ≈1h validação).** +3.0 Deploy manual de main (`workflow_dispatch`: verify full → staging → aprovação → produção) — aplica a migration do C153. +3.1 `~/teqo-backfill`: clone/fetch de `github.com/fsolla/teqo.git`, checkout do SHA do branch C155, `pnpm install`. +3.2 Env: `set -a; source ~/stack/teqo-1313.env; set +a`; `export DATABASE_URL="${DATABASE_URL/@postgres:5432/@127.0.0.1:5433}"` (proxy `teqo-1313-build-proxy`); `export CAMARA_IMPORT_CONFIRM=1`; tmux + `tee ~/c155-backfill-.log`. +3.3 Smoke de produção com idempotência viva: `--all --limit 1` (4 discursos) e repetir → segunda passada **0 criados, 0 ASR**; coverage = 4. +3.4 `--all` completo (checkpoint por legislatura; se abortar, retomar com `--all` ou `--legislature N`). +3.5 `--coverage` e `--verify-links 3` (12 discursos, ~24 probes). +3.6 Coletar artefatos e o log. +_Prova:_ coverage total = 1.011 (54ª=1, 55ª=362, 56ª=414, 57ª=234); warnings de links documentados; log do tmux arquivado. + +**Fase 4 — Fechamento (≈1h).** +Atualizar changelog com números medidos; runbook com resultado + rollback; anexar JSONs à Issue #957; resumo no corpo do PR. Gates: `pnpm gate:fast`, `pnpm knip`, cycles, `pnpm typecheck`, `test:unit`, `test:int`; e2e esperado `none` (script/docs fora dos prefixos de risco — `ci-scope` decide; se cair em `unmapped-risk`, adicionar mapping). Push + PR `Closes #957` + auto-merge. +_Quota:_ 1 PR; prova: check `checks` verde. + +## Ajustes descobertos na execução + +- **Fase 3.0 cumprida por outro deploy.** A C154 (`3e062749`) mergeou em `main` e o deploy manual de 2026-09-13 publicou `424ee311` em produção — o schema do C153 (`speech`/`speech_segment` + migrations `20260913_001112`/`001200`) e o `searchText` da C154 já estão no `teqo_1313`. Nenhum deploy novo é pré-requisito do backfill; o workspace `~/teqo-deploy` também já está no SHA novo. +- **Rede da workstation para a Câmara instável** (hangs/HTTP 500 na API em 2026-09-13 à tarde; o homeserver responde 200 em ~2s). O smoke local da Fase 1 vira o smoke de produção da Fase 3.3 (mesmo código, rede confiável); unit/int locais continuam sendo o gate de lógica. +- **Paginação do import ganhou backoff** (`fetchJsonWithBackoff`, 5 tentativas com espera exponencial até 30s) porque a API devolveu 500 na página 4 da 55ª durante o smoke — o comportamento de abortar só a legislatura e seguir continua, agora com mais paciência. +- **Cache/mídia do run de produção** em `--out /srv/hdd/teqo-backfill/camara` (67 GB livres em `/`, 394 GB no HDD). + +## Rabbit holes / Não escopo (engenharia) + +- Comissões/audiências e segunda fonte de transcrição (YouTube próprio) — **não**; `youtubeUrl` do evento já é metadado, não vira pipeline. +- Espelhar mídia no S3, gerar clipes, corrigir sessão sem trecho à mão — **não**. +- Scheduler/recorrência — **não**; execução manual documentada. +- Paralelizar o pipeline ou otimizar wall clock — **não** (D8-A). +- Copiar ASR/segmentos do worktree local para produção ou reconciliar ids entre DBs — **não** (D4-B). +- Migration/schema novo, access novo, UI — **não**; se algo exigir, justificar no PR antes de fazer. +- Commitar JSON de relatório — **não** (D6). +- Extrair `parseArgs` para teste — **não** agora (D7-D). + +## Riscos e mitigação + +| Risco | Mitigação | +| --- | --- | +| Migration C153 não aplicada em produção (fato: batch 52) → escrita falha | Fase 3.0 deploy de main **antes** de qualquer run; Fase 0 confere migration + tabelas | +| `DEEPINFRA_API_KEY` ausente no homeserver (fato) | Adicionar ao env file na Fase 0; o modo de escrita sem `--skip-transcribe` falha cedo e claro | +| Guard furado pelo proxy `127.0.0.1`/host `postgres` | D3: confirm por `NODE_ENV=production` + host + `ALLOW_REMOTE_DB`; eco do alvo; smoke de produção com re-run antes do run completo | +| Queda de rede/SSH durante 10–15h | tmux + `tee`; checkpoint por legislatura; resume idempotente (`--all` ou `--legislature N`); falha de uma legislatura não derruba as demais | +| Disco `/` enche com MP4/HTML cacheados | `--out /srv/hdd/teqo-backfill/camara` (394GB livres); cache de eventos no mesmo destino | +| Sessões antigas sem trecho/HTML fora do padrão (54ª inteira, 55ª parcial) | Não é falha: vira `withoutExcerpt`/`withoutSegments`/`fallbackYoutube` no coverage e pendência documentada; nenhuma correção manual | +| Custo acima do orçamento | Relatório mede ASR/LLM por legislatura; teto declarado ~US$0,7; `--skip-transcribe` disponível para re-runs de metadados | +| Memória do processo (caches `eventsByDate`/`eventPages`) em 8c/15GB | Monitorar no run; fallback: rodar por `--legislature` (o checkpoint já é natural) | +| Escrita concorrente com o app em produção | Escrita confinada a `speech`/`speechSegment`/rels, transacional, sem locks de outras tabelas; janela de baixa atividade | +| Reexecução acidental pós-run (tempo/custo) | Idempotência + coverage + guard; re-run é seguro, mas exige a flag e custa wall clock | +| C154 não está em main | Dados ficam prontos; validação visual registrada como pendência do C154, não bloqueio deste item | +| Colisão de `sourceKey` (vista na 57ª) | Já resolvida no C153 por hash de conteúdo; o relatório marca `suffixedKey` | + +## Aceite de engenharia (checklist) + +- [ ] `--all` processa 54→57 num processo, com `legislatures[]` + totais combinados + `coverage`; mutex com `--date`/`--legislature` explícito; checkpoint por legislatura. +- [ ] Idempotência comprovada: reexecução de `--all --limit 1` em produção e do smoke local resultam em 0 criados e 0 chamadas ASR. +- [ ] `--coverage` é read-only no banco e sem rede; imprime por legislatura (total, com trecho, com vídeo, com segmentos, sem trecho, sem segmentos, fallback YouTube) + totais. +- [ ] `CAMARA_IMPORT_CONFIRM=1` exigido para escrita com `NODE_ENV=production`, host remoto ou `ALLOW_REMOTE_DB`; fail-closed antes de `getPayload`; `--coverage`/`--verify-links` isentos; `--skip-transcribe` não isento; alvo ecoado. +- [ ] `--verify-links ` amostra n por legislatura, GET Range com fallback HEAD, warnings não fatais, 0 verificados = exit 1. +- [ ] Nenhuma escrita fora de `speech`/`speechSegment`/`speech_rels`; nenhum objeto no S3; CC BY 4.0 no runbook/changelog. +- [ ] Unit novos (agregação, amostragem, guard) e int de `speechCoverage` com fixtures, verdes. +- [ ] `pnpm gate:fast` + knip + cycles + typecheck + unit + int verdes; e2e conforme `ci-scope` (esperado `none`; `unmapped-risk` tratado). +- [ ] Produção com 1.011 discursos (54ª=1, 55ª=362, 56ª=414, 57ª=234) e migration C153 aplicada; coverage pós-run bate com o relatório do run. +- [ ] `--verify-links 3` na amostra sem falha dura; warnings e sessões sem trecho documentados como pendência. +- [ ] Relatórios anexados à Issue #957; resumo no corpo do PR; `docs/changelog/2026-09-13-c155.md` e seção C155 no runbook com resultado e rollback (`DELETE` em `speech_segment`/`speech_rels`/`speech`, ou re-run idempotente). +- [ ] Nenhuma migration nova; nenhum arquivo de `data/camara/` commitado; `Closes #957` no PR com auto-merge armado. + +## Self-score decision-quality + +| Item | Nota | Justificativa | +| --- | --- | --- | +| Decisões caras (guard de escrita, sequência da operação, contrato do relatório/coverage) decididas explicitamente | 5 | D3/D4/D2 têm opções, recomendação e rejeitadas; o barato (nomes de campo, n da amostra) ficou como fill-in com gatilho | +| Reuso dos donos existentes (depth check) | 5 | `import-camara-speeches.mjs`, `camaraSpeeches`/`camaraFetch`, `cli.mjs`, `speechImport`, `recover-media`/OPS79 como molde, `drizzleResultRows` — sem twin | +| Formato Opções/Recomendação/Rejeitadas em todas as decisões | 5 | D1–D8 no formato obrigatório, com rejeitadas justificadas (não decorativas) | +| Fases verificáveis com quota e prova por fase | 4 | Cada fase tem quota e prova; a estimativa de wall clock (10–15h) é probabilística e a fase 3 depende do deploy manual | +| Riscos com mitigação e rollback idempotente | 4 | 12 riscos mapeados com mitigação; rollback por DELETE/re-run documentado, mas depende de operador humano no homeserver | + +Média: **4,6** (≥4). diff --git a/scripts/assert-local-database.mjs b/scripts/assert-local-database.mjs index acdb658b..a22555fb 100644 --- a/scripts/assert-local-database.mjs +++ b/scripts/assert-local-database.mjs @@ -4,9 +4,10 @@ */ import { + databaseHostname, dieWithLabel, + isLocalDatabaseUrl, isRemoteDbOverrideSet, - LOCAL_HOSTS, ALLOW_REMOTE_DB_FLAG as OVERRIDE_FLAG, } from './lib/cli.mjs' @@ -27,14 +28,10 @@ export const assertLocalDatabase = (label, usageHint = '') => { return } - let host - try { - host = new URL(databaseUrl).hostname - } catch { - die('DATABASE_URL is not a valid connection string.') - } + const host = databaseHostname(databaseUrl) + if (host === null) die('DATABASE_URL is not a valid connection string.') - if (!LOCAL_HOSTS.has(host)) { + if (!isLocalDatabaseUrl(databaseUrl)) { die( `DATABASE_URL points at a non-local host ("${host}").\n` + (usageHint ? `${usageHint}\n\n` : '') + diff --git a/scripts/import-camara-speeches.mjs b/scripts/import-camara-speeches.mjs index 7738a319..b8a0ab8d 100644 --- a/scripts/import-camara-speeches.mjs +++ b/scripts/import-camara-speeches.mjs @@ -1,21 +1,28 @@ /** - * C153 import — populates the speech catalog from the Câmara sources: - * open-data speeches → event on that date → speaker excerpt on the event page - * → async VOD → Deep Infra Whisper segments (minutagem) + facet classification - * (gazetteer + LLM) → idempotent upsert into `speech`/`speechSegment`. + * C153 import + C155 backfill — populates the speech catalog from the Câmara + * sources: open-data speeches → event on that date → speaker excerpt on the + * event page → async VOD → Deep Infra Whisper segments (minutagem) + facet + * classification (gazetteer + LLM) → idempotent upsert into + * `speech`/`speechSegment`. * * Re-running is safe: records are matched by `sourceKey`, metadata is * refreshed, VOD/ASR is skipped when the excerpt identity and segments are * already stored, and a `manual` facet curation is never overwritten. * - * Media and reports land under `data/camara/` (gitignored). The local-DB guard - * refuses a non-local DATABASE_URL; production is C155. + * C155 adds the full-archive backfill (`--all`, one process, per-legislature + * report + DB coverage) and the read-only modes `--coverage` and + * `--verify-links `. Media and reports land under `data/camara/` + * (gitignored). The local-DB guard refuses a non-local DATABASE_URL, and any + * write run targeting production (or a remote/override DB) additionally + * requires `CAMARA_IMPORT_CONFIRM=1`. * * Usage: * pnpm camara:import --legislature 57 * pnpm camara:import --date 2023-02-07 --limit 3 - * pnpm camara:import --legislature 57 --skip-transcribe - * pnpm camara:import --legislature 57 --reclassify + * pnpm camara:import --all + * pnpm camara:import --all --limit 2 # smoke + * pnpm camara:import --coverage + * pnpm camara:import --verify-links 3 */ import { mkdir, readFile, writeFile } from 'node:fs/promises' import { dirname, join } from 'node:path' @@ -30,6 +37,7 @@ import { eventsUrl, getJson, getText, + probeVodLink, resolveVod, speechesUrl, transcribeSpeechAudio, @@ -37,6 +45,7 @@ import { import { LEGISLATURE_RANGES, SOLLA_DEPUTY_ID, + aggregateBackfillRuns, legislatureForDate, matchExcerpt, parseDurationToSeconds, @@ -44,12 +53,20 @@ import { parseOfficialKeywords, parsePresidingOfficerTransitions, resolvePresidingOfficer, + selectLinkSample, selectSpeechEvents, speechContentHash, speechDate, speechSourceKey, } from './lib/camaraSpeeches.mjs' -import { dieWithLabel, ensureCachedDownload, loadCliEnv } from './lib/cli.mjs' +import { + databaseHostname, + dieWithLabel, + ensureCachedDownload, + isTruthyEnv, + loadCliEnv, + requiresWriteConfirm, +} from './lib/cli.mjs' loadCliEnv() @@ -59,6 +76,9 @@ const config = (await import('../src/payload.config.ts')).default const { findSpeechImportState, upsertSpeechBundle } = await import('../src/utilities/speech/speechImport.ts') const { classifySpeech } = await import('../src/utilities/speech/speechClassifier.ts') +const { getSpeechCoverage } = await import('../src/utilities/speech/speechCoverage.ts') + +const WRITE_CONFIRM_FLAG = 'CAMARA_IMPORT_CONFIRM' const DEFAULT_OUT_DIR = 'data/camara' const MAX_EVENT_CANDIDATES = 5 @@ -76,13 +96,16 @@ const EVENT_TYPE_PRIORITY = [ const parseArgs = (argv) => { const options = { - legislature: 57, + legislature: null, + all: false, date: null, limit: null, speaker: 'Jorge Solla', out: DEFAULT_OUT_DIR, skipTranscribe: false, reclassify: false, + coverage: false, + verifyLinks: null, help: false, } for (let index = 0; index < argv.length; index += 1) { @@ -93,6 +116,9 @@ const parseArgs = (argv) => { return next } if (arg === '--help' || arg === '-h') options.help = true + else if (arg === '--all') options.all = true + else if (arg === '--coverage') options.coverage = true + else if (arg === '--verify-links') options.verifyLinks = Number(value()) else if (arg === '--skip-transcribe') options.skipTranscribe = true else if (arg === '--reclassify') options.reclassify = true else if (arg === '--legislature') options.legislature = Number(value()) @@ -102,34 +128,69 @@ const parseArgs = (argv) => { else if (arg === '--out') options.out = value() else die(`argumento desconhecido: ${arg}`) } - if (!options.help && !options.date && !LEGISLATURE_RANGES[options.legislature]) { + if (options.help) return options + + const readOnlyModes = [options.coverage, options.verifyLinks !== null].filter(Boolean).length + const importModes = [options.all, options.date !== null, options.legislature !== null].filter( + Boolean, + ).length + if (readOnlyModes > 1) die('--coverage e --verify-links são modos separados.') + if (readOnlyModes > 0 && importModes > 0) { + die('--coverage/--verify-links não combinam com --all/--date/--legislature.') + } + if ( + readOnlyModes > 0 && + (options.limit !== null || options.skipTranscribe || options.reclassify) + ) { + die('--limit/--skip-transcribe/--reclassify só valem para os modos de import.') + } + if (options.all && options.date !== null) die('--all e --date são mutuamente exclusivos.') + if (options.all && options.legislature !== null) { + die('--all e --legislature são mutuamente exclusivos.') + } + if ( + options.verifyLinks !== null && + (!Number.isInteger(options.verifyLinks) || options.verifyLinks < 1) + ) { + die('--verify-links exige um inteiro >= 1.') + } + if (importModes === 0 && readOnlyModes === 0) options.legislature = 57 + if (options.legislature !== null && !LEGISLATURE_RANGES[options.legislature]) { die('informe --legislature <54|55|56|57> ou --date (veja --help).') } - if (!options.help && options.date && !/^\d{4}-\d{2}-\d{2}$/.test(options.date)) { + if (options.date !== null && !/^\d{4}-\d{2}-\d{2}$/.test(options.date)) { die('--date deve ser YYYY-MM-DD.') } if (options.limit !== null && (!Number.isInteger(options.limit) || options.limit < 1)) { die('--limit deve ser >= 1.') } - if (!options.help && options.out.includes('..')) { + if (options.out.includes('..')) { die('--out não pode escapar do diretório do repo (sem "..").') } return options } const HELP = ` -Uso: pnpm camara:import [--legislature 54|55|56|57] [--date YYYY-MM-DD] [opções] +Uso: pnpm camara:import [--all | --legislature 54|55|56|57 | --date YYYY-MM-DD | --coverage | --verify-links ] [opções] -Opções: - --legislature 54|55|56|57 (default 57; ignorado se --date) - --date discursos de um único dia - --limit quantos discursos processar (default: todos) +Modos de import (escrevem no acervo): + --all processa as 54ª–57ª em sequência (relatório por legislatura) + --legislature 54|55|56|57 (default 57 quando nenhum modo é dado) + --date discursos de um único dia +Modos read-only: + --coverage imprime a cobertura do banco por legislatura (sem rede/escrita) + --verify-links amostra n discursos por legislatura e testa os links do VOD + +Opções de import: + --limit quantos discursos processar por legislatura (default: todos) --speaker nome do orador a casar no evento (default "Jorge Solla") --out diretório de cache/artefatos (default data/camara) --skip-transcribe não chama a Deep Infra (preserva os segmentos existentes) --reclassify refaz a classificação mesmo já tendo facetas da LLM --help esta ajuda +Escrita em alvo não-local ou com NODE_ENV=production exige CAMARA_IMPORT_CONFIRM=1 +(runbook: docs/ops/teqo-1313-deploy.md §C155). A página do evento é cacheada em /events; apague-a para reler a Câmara. ` @@ -165,14 +226,41 @@ async function loadEventPage(eventId, options) { return html } +/** + * The open-data API throws transient 500s and hangs on deep pages (seen + * 2026-09-13 on the 55ª page 4). `getText`'s own 3 attempts are too impatient + * for a 1.011-speech backfill, so pagination owns a bounded backoff; a page + * that never recovers aborts only its legislature (`--all` continues and the + * operator resumes with `--legislature `). + */ +async function fetchJsonWithBackoff(url, { attempts = 5, timeoutMs = 45_000 } = {}) { + let lastError + for (let attempt = 1; attempt <= attempts; attempt += 1) { + try { + return JSON.parse(await getText(url, { attempts: 1, timeoutMs })) + } catch (error) { + lastError = error + if (attempt < attempts) { + const waitMs = Math.min(30_000, 2 ** attempt * 1_000) + console.log( + `[camara:import] ${error?.message ?? error} — retry ${attempt}/${attempts - 1} em ${waitMs / 1000}s`, + ) + await sleep(waitMs) + } + } + } + throw lastError +} + async function fetchSpeeches(deputyId, from, to) { const speeches = [] for (let page = 1; ; page += 1) { - const body = await getJson(`${speechesUrl(deputyId, from, to)}&pagina=${page}`) + const body = await fetchJsonWithBackoff(`${speechesUrl(deputyId, from, to)}&pagina=${page}`) const batch = body.dados ?? [] speeches.push(...batch) const hasNext = (body.links ?? []).some((link) => link?.rel === 'next') if (!hasNext || batch.length === 0) break + await sleep(500) } return speeches } @@ -405,7 +493,7 @@ async function processSpeech(payload, speech, options, run) { const printReport = (run) => { const { totals, asr, llm, failures } = run - console.log('\n=== Relatório do import C153 ===') + console.log('\n=== Relatório do import (camara:import) ===') console.log( `discursos: ${totals.listed} listados, ${totals.processed} processados ` + `(${totals.created} criados, ${totals.updated} atualizados)`, @@ -429,6 +517,62 @@ const printReport = (run) => { } } +const pad = (value, width) => String(value).padStart(width) +const padEnd = (value, width) => String(value).padEnd(width) + +const printCoverage = (coverage) => { + console.log('\n=== Cobertura do acervo (por legislatura) ===') + console.log( + `${padEnd('leg', 5)}${pad('total', 7)}${pad('trecho', 8)}${pad('vídeo', 7)}` + + `${pad('segm.', 7)}${pad('sem trecho', 11)}${pad('sem segm.', 10)}${pad('fallback YT', 12)}`, + ) + for (const row of coverage.byLegislature) { + console.log( + `${padEnd(row.legislature ?? '—', 5)}${pad(row.total, 7)}${pad(row.withExcerpt, 8)}` + + `${pad(row.withVideo, 7)}${pad(row.withSegments, 7)}${pad(row.withoutExcerpt, 11)}` + + `${pad(row.withoutSegments, 10)}${pad(row.fallbackYoutube, 12)}`, + ) + } + const { totals } = coverage + console.log( + `${padEnd('total', 5)}${pad(totals.total, 7)}${pad(totals.withExcerpt, 8)}` + + `${pad(totals.withVideo, 7)}${pad(totals.withSegments, 7)}${pad(totals.withoutExcerpt, 11)}` + + `${pad(totals.withoutSegments, 10)}${pad(totals.fallbackYoutube, 12)}`, + ) +} + +const printBackfillReport = (combined) => { + console.log('\n=== Relatório do backfill (54ª–57ª) ===') + for (const run of combined.legislatures) { + const { totals, asr } = run + console.log( + `${run.legislature}ª: ${totals.listed} listados, ${totals.processed} processados ` + + `(${totals.created} criados, ${totals.updated} atualizados) · ` + + `${totals.withExcerpt} com trecho / ${totals.withoutExcerpt} sem · ` + + `${totals.withSegments} com segmentos · ` + + `ASR ${(asr.audioSeconds / 60).toFixed(1)} min ` + + `~US$ ${((asr.audioSeconds * DEEPINFRA_COST_PER_MINUTE_USD) / 60).toFixed(4)} · ` + + `${(run.elapsedMs / 1000).toFixed(0)}s${run.aborted ? ' · ABORTADA' : ''}`, + ) + } + const { totals, asr, llm } = combined + console.log( + `combinado: ${totals.processed} processados (${totals.created} criados, ` + + `${totals.updated} atualizados) · ${totals.withSegments} com segmentos · ` + + `ASR ${(asr.audioSeconds / 60).toFixed(1)} min ~US$ ` + + `${((asr.audioSeconds * DEEPINFRA_COST_PER_MINUTE_USD) / 60).toFixed(4)} · ` + + `LLM ${llm.used}/${llm.calls} ~US$ ${llm.estimatedCostUsd.toFixed(4)}`, + ) + console.log( + `tempo total: ${(combined.elapsedMs / 1000).toFixed(1)}s; falhas: ${combined.failures.length}`, + ) + for (const failure of combined.failures) { + console.log( + ` ✗ [${failure.legislature ?? '-'}] ${failure.speechAt} [${failure.stage}] ${failure.message}`, + ) + } +} + const speechLine = (report) => { const parts = [ report.created ? 'criado' : 'atualizado', @@ -444,59 +588,54 @@ const speechLine = (report) => { } // --------------------------------------------------------------------------- -// Main +// Runs // --------------------------------------------------------------------------- -async function main() { - const options = parseArgs(process.argv.slice(2)) - if (options.help) { - console.log(HELP) - process.exit(0) - } +const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms)) - assertLocalDatabase('camara:import', 'O import C153 escreve no banco local; produção é C155.') - const payload = await getPayload({ config }) +const stamp = (runAt) => runAt.replace(/[:.]/g, '-') - const from = options.date ?? LEGISLATURE_RANGES[options.legislature][0] - const to = options.date ?? LEGISLATURE_RANGES[options.legislature][1] - const startedAt = Date.now() - console.log( - `[camara:import] deputado ${SOLLA_DEPUTY_ID} | ${from}..${to} | speaker "${options.speaker}"`, - ) +const writeReport = async (options, filename, report) => { + const reportPath = join(options.out, 'reports', filename) + await mkdir(dirname(reportPath), { recursive: true }) + await writeFile(reportPath, JSON.stringify(report, null, 2)) + return reportPath +} - const speeches = await fetchSpeeches(SOLLA_DEPUTY_ID, from, to) - const targets = options.limit === null ? speeches : speeches.slice(0, options.limit) - console.log(`[camara:import] ${speeches.length} discursos na API; processando ${targets.length}`) - if (targets.length === 0) die(`nenhum discurso em ${from}..${to}.`) +const runOptions = (options) => ({ + legislature: options.all ? null : options.legislature, + all: options.all, + date: options.date, + limit: options.limit, + speaker: options.speaker, + skipTranscribe: options.skipTranscribe, + reclassify: options.reclassify, +}) - const run = { - runAt: new Date().toISOString(), - range: { from, to }, - options: { - legislature: options.date ? null : options.legislature, - date: options.date, - limit: options.limit, - speaker: options.speaker, - skipTranscribe: options.skipTranscribe, - reclassify: options.reclassify, - }, - totals: { - listed: speeches.length, - processed: 0, - created: 0, - updated: 0, - withExcerpt: 0, - withoutExcerpt: 0, - withSegments: 0, - failed: 0, - }, - asr: { calls: 0, audioSeconds: 0, elapsedMs: 0, failures: 0 }, - llm: { calls: 0, used: 0, failed: 0, totalTokens: 0, estimatedCostUsd: 0, sampleError: null }, - elapsedMs: 0, - speeches: [], - failures: [], - } +const createRun = (options, { legislature, from, to }) => ({ + runAt: new Date().toISOString(), + mode: options.date ? 'date' : 'legislature', + legislature, + range: { from, to }, + options: runOptions(options), + totals: { + listed: 0, + processed: 0, + created: 0, + updated: 0, + withExcerpt: 0, + withoutExcerpt: 0, + withSegments: 0, + failed: 0, + }, + asr: { calls: 0, audioSeconds: 0, elapsedMs: 0, failures: 0 }, + llm: { calls: 0, used: 0, failed: 0, totalTokens: 0, estimatedCostUsd: 0, sampleError: null }, + elapsedMs: 0, + speeches: [], + failures: [], +}) +const processTargets = async (payload, options, targets, run) => { for (const [index, speech] of targets.entries()) { const report = await processSpeech(payload, speech, options, run) run.totals.processed += 1 @@ -506,14 +645,234 @@ async function main() { } console.log(`[${index + 1}/${targets.length}] ${report.speechAt} — ${speechLine(report)}`) } +} +const runSingleRange = async (payload, options) => { + const from = options.date ?? LEGISLATURE_RANGES[options.legislature][0] + const to = options.date ?? LEGISLATURE_RANGES[options.legislature][1] + const run = createRun(options, { + legislature: options.date ? null : options.legislature, + from, + to, + }) + const startedAt = Date.now() + console.log( + `[camara:import] deputado ${SOLLA_DEPUTY_ID} | ${from}..${to} | speaker "${options.speaker}"`, + ) + + const speeches = await fetchSpeeches(SOLLA_DEPUTY_ID, from, to) + const targets = options.limit === null ? speeches : speeches.slice(0, options.limit) + console.log(`[camara:import] ${speeches.length} discursos na API; processando ${targets.length}`) + if (targets.length === 0) die(`nenhum discurso em ${from}..${to}.`) + run.totals.listed = speeches.length + + await processTargets(payload, options, targets, run) run.elapsedMs = Date.now() - startedAt + run.coverage = await getSpeechCoverage(payload) printReport(run) + printCoverage(run.coverage) - const reportPath = join(options.out, 'reports', `import-${run.runAt.replace(/[:.]/g, '-')}.json`) - await mkdir(dirname(reportPath), { recursive: true }) - await writeFile(reportPath, JSON.stringify(run, null, 2)) + const reportPath = await writeReport(options, `import-${stamp(run.runAt)}.json`, run) console.log(`\n[camara:import] relatório JSON: ${reportPath}`) +} + +const runBackfill = async (payload, options) => { + const startedAt = Date.now() + const combined = { + runAt: new Date().toISOString(), + mode: 'all', + options: runOptions(options), + legislatures: [], + totals: null, + asr: null, + llm: null, + elapsedMs: 0, + failures: [], + coverage: null, + } + const legislatures = Object.keys(LEGISLATURE_RANGES) + .map(Number) + .sort((left, right) => left - right) + + for (const legislature of legislatures) { + const [from, to] = LEGISLATURE_RANGES[legislature] + const run = createRun(options, { legislature, from, to }) + const rangeStartedAt = Date.now() + console.log(`\n[camara:import] === legislatura ${legislature}ª (${from}..${to}) ===`) + try { + const speeches = await fetchSpeeches(SOLLA_DEPUTY_ID, from, to) + const targets = options.limit === null ? speeches : speeches.slice(0, options.limit) + console.log( + `[camara:import] ${speeches.length} discursos na API; processando ${targets.length}`, + ) + run.totals.listed = speeches.length + await processTargets(payload, options, targets, run) + } catch (error) { + run.aborted = true + run.failures.push({ + sourceKey: null, + speechAt: null, + stage: 'legislature', + message: error?.message ?? String(error), + }) + console.error( + `[camara:import] legislatura ${legislature}ª abortada: ${error?.message ?? error}`, + ) + } + run.elapsedMs = Date.now() - rangeStartedAt + combined.legislatures.push(run) + combined.failures.push(...run.failures.map((failure) => ({ legislature, ...failure }))) + + const partial = aggregateBackfillRuns(combined.legislatures) + combined.totals = partial.totals + combined.asr = partial.asr + combined.llm = partial.llm + combined.elapsedMs = Date.now() - startedAt + // Checkpoint after each legislature: a network drop loses at most one. + await writeReport(options, `backfill-${stamp(combined.runAt)}.json`, combined) + } + + combined.coverage = await getSpeechCoverage(payload) + printBackfillReport(combined) + printCoverage(combined.coverage) + + const reportPath = await writeReport(options, `backfill-${stamp(combined.runAt)}.json`, combined) + console.log(`\n[camara:import] relatório JSON: ${reportPath}`) + if (combined.legislatures.some((run) => run.aborted)) { + die('legislatura(s) abortada(s) — reexecute --all ou --legislature para completar.') + } +} + +const runCoverage = async (payload, options) => { + const startedAt = Date.now() + const coverage = await getSpeechCoverage(payload) + printCoverage(coverage) + const report = { + runAt: new Date().toISOString(), + mode: 'coverage', + options: runOptions(options), + coverage, + elapsedMs: Date.now() - startedAt, + } + const reportPath = await writeReport(options, `coverage-${stamp(report.runAt)}.json`, report) + console.log(`\n[camara:import] relatório JSON: ${reportPath}`) +} + +const runVerifyLinks = async (payload, options) => { + const startedAt = Date.now() + const found = await payload.find({ + collection: 'speech', + where: { or: [{ vodPlaybackUrl: { exists: true } }, { vodDownloadUrl: { exists: true } }] }, + pagination: false, + depth: 0, + select: { legislature: true, speechAt: true, vodPlaybackUrl: true, vodDownloadUrl: true }, + // Intentional bypass: the import CLI is a trusted actor with no session. + overrideAccess: true, + }) + const sample = selectLinkSample(found.docs, options.verifyLinks) + console.log( + `[camara:import] amostra: ${sample.length} discursos (${options.verifyLinks} por legislatura)`, + ) + + const checked = [] + for (const row of sample) { + const links = [ + ['playback', row.vodPlaybackUrl], + ['download', row.vodDownloadUrl], + ] + for (const [kind, url] of links) { + if (!url) continue + const probe = await probeVodLink(url) + checked.push({ legislature: row.legislature, speechId: row.id, kind, ...probe }) + console.log( + `[camara:import] ${row.legislature ?? '—'} #${row.id} ${kind}: ` + + `${probe.ok ? 'ok' : 'WARN'} ${probe.note}`, + ) + await sleep(250) + } + } + + const warnings = checked.filter((entry) => !entry.ok).length + const report = { + runAt: new Date().toISOString(), + mode: 'verify-links', + options: runOptions(options), + sample: { perLegislature: options.verifyLinks, total: sample.length }, + checked, + ok: checked.length - warnings, + warnings, + elapsedMs: Date.now() - startedAt, + } + console.log( + `[camara:import] links: ${report.ok}/${checked.length} ok` + + `${warnings > 0 ? `, ${warnings} warning(s)` : ''}`, + ) + const reportPath = await writeReport(options, `verify-links-${stamp(report.runAt)}.json`, report) + console.log(`\n[camara:import] relatório JSON: ${reportPath}`) + if (checked.length === 0) die('nenhum link de VOD verificado — amostra vazia (banco sem links?).') +} + +// --------------------------------------------------------------------------- +// Main +// --------------------------------------------------------------------------- + +const databaseTarget = () => { + const host = databaseHostname(process.env.DATABASE_URL) + if (host === null) return '(DATABASE_URL ausente ou inválida)' + try { + return `${host}${new URL(process.env.DATABASE_URL).pathname}` + } catch { + return host + } +} + +const modeLabel = (options) => { + if (options.coverage) return 'coverage (read-only)' + if (options.verifyLinks !== null) return `verify-links n=${options.verifyLinks} (read-only)` + if (options.all) return 'backfill 54ª–57ª' + if (options.date) return `date ${options.date}` + return `legislature ${options.legislature}` +} + +/** + * C155 write guard: the homeserver env file sets `NODE_ENV=production` and the + * runbook rewrites the DB host to the local proxy, so the host check alone + * cannot see production — the explicit flag is what makes the write deliberate. + */ +const assertWriteAllowed = () => { + if (!requiresWriteConfirm()) return + if (isTruthyEnv(process.env[WRITE_CONFIRM_FLAG])) return + die( + `alvo de escrita não-local/produção detectado (${databaseTarget()}).\n` + + `Confirme a intenção com:\n ${WRITE_CONFIRM_FLAG}=1 pnpm camara:import …\n` + + `Runbook: docs/ops/teqo-1313-deploy.md §C155.`, + ) +} + +async function main() { + const options = parseArgs(process.argv.slice(2)) + if (options.help) { + console.log(HELP) + process.exit(0) + } + + const readOnly = options.coverage || options.verifyLinks !== null + if (readOnly) { + if (!process.env.DATABASE_URL) die('DATABASE_URL is not set. Refusing to continue.') + } else { + assertLocalDatabase( + 'camara:import', + 'O import escreve no acervo; produção exige CAMARA_IMPORT_CONFIRM=1 (runbook §C155).', + ) + assertWriteAllowed() + } + console.log(`[camara:import] alvo: ${databaseTarget()} | modo: ${modeLabel(options)}`) + + const payload = await getPayload({ config }) + if (options.coverage) await runCoverage(payload, options) + else if (options.verifyLinks !== null) await runVerifyLinks(payload, options) + else if (options.all) await runBackfill(payload, options) + else await runSingleRange(payload, options) process.exit(0) } diff --git a/scripts/lib/camaraFetch.mjs b/scripts/lib/camaraFetch.mjs index e7028607..1df20e21 100644 --- a/scripts/lib/camaraFetch.mjs +++ b/scripts/lib/camaraFetch.mjs @@ -84,6 +84,42 @@ export async function resolveVod(eventId, excerpt) { return { url, status: lastStatus, attempts: VOD_POLL_MAX_ATTEMPTS } } +/** + * Lightweight reachability probe of one stored VOD link (C155 sample check): + * GET with a 1 KiB `Range` so the MP4 body never streams, falling back to HEAD + * when the server refuses the ranged GET (405/501). Never throws — a network + * error becomes an `ok: false` result so one dead link cannot abort the report. + * + * @param {string | null | undefined} url + * @returns {Promise<{ url: string | null, status: number | null, ok: boolean, note: string }>} + */ +export async function probeVodLink(url, { timeoutMs = 20_000 } = {}) { + if (!url) return { url: null, status: null, ok: false, note: 'sem link' } + + const request = async (method) => { + const response = await fetch(url, { + method, + headers: { + 'User-Agent': CAMARA_USER_AGENT, + ...(method === 'GET' ? { Range: 'bytes=0-1023' } : {}), + }, + signal: AbortSignal.timeout(timeoutMs), + }) + const contentType = response.headers.get('content-type') + const note = `HTTP ${response.status}${contentType ? ` ${contentType}` : ''}` + await response.body?.cancel().catch(() => undefined) + return { url, status: response.status, ok: response.ok, note } + } + + try { + const first = await request('GET') + if (first.status === 405 || first.status === 501) return await request('HEAD') + return first + } catch (error) { + return { url, status: null, ok: false, note: error?.message ?? String(error) } + } +} + export const DEEPINFRA_COST_PER_MINUTE_USD = 0.00045 /** diff --git a/scripts/lib/camaraSpeeches.mjs b/scripts/lib/camaraSpeeches.mjs index b2cdebe2..90fadca6 100644 --- a/scripts/lib/camaraSpeeches.mjs +++ b/scripts/lib/camaraSpeeches.mjs @@ -398,3 +398,86 @@ export const normalizeTranscription = (json) => { segments, } } + +/** Numeric blocks of one run that `aggregateBackfillRuns` folds together. */ +const BACKFILL_TOTAL_KEYS = [ + 'listed', + 'processed', + 'created', + 'updated', + 'withExcerpt', + 'withoutExcerpt', + 'withSegments', + 'failed', +] +const BACKFILL_ASR_KEYS = ['calls', 'audioSeconds', 'elapsedMs', 'failures'] +const BACKFILL_LLM_KEYS = ['calls', 'used', 'failed', 'totalTokens', 'estimatedCostUsd'] + +/** + * Folds the per-legislature runs of a `--all` backfill into the combined + * block. Pure arithmetic — the run report must not depend on network/DB to be + * summed, so it is unit-tested with hand-written run objects. + * + * @param {Array<{ totals?: Record, asr?: Record, llm?: Record, elapsedMs?: unknown }> | null | undefined} runs + * @returns {{ totals: Record, asr: Record, llm: Record, elapsedMs: number }} + */ +export const aggregateBackfillRuns = (runs) => { + const list = Array.isArray(runs) ? runs : [] + const sumBy = (group, keys) => + Object.fromEntries( + keys.map((key) => [ + key, + list.reduce((total, run) => total + (Number(run?.[group]?.[key]) || 0), 0), + ]), + ) + return { + totals: sumBy('totals', BACKFILL_TOTAL_KEYS), + asr: sumBy('asr', BACKFILL_ASR_KEYS), + llm: { + ...sumBy('llm', BACKFILL_LLM_KEYS), + sampleError: list.find((run) => run?.llm?.sampleError)?.llm?.sampleError ?? null, + }, + elapsedMs: list.reduce((total, run) => total + (Number(run?.elapsedMs) || 0), 0), + } +} + +/** + * Deterministic VOD-link sample for the `--verify-links` mode: up to + * `perLegislature` speeches per legislature that actually carry a VOD link, + * spread evenly over the legislature ordered by `speechAt` + id. Same DB state + * → same sample, so the verification is reproducible and reportable. + * + * @param {Array<{ id: number, legislature?: string | null, speechAt?: string | null, vodPlaybackUrl?: string | null, vodDownloadUrl?: string | null }>} rows + * @param {number} perLegislature + * @returns {typeof rows} + */ +export const selectLinkSample = (rows, perLegislature) => { + const limit = Math.max(1, Math.trunc(Number(perLegislature) || 1)) + const byLegislature = new Map() + for (const row of Array.isArray(rows) ? rows : []) { + if (!row?.vodPlaybackUrl && !row?.vodDownloadUrl) continue + const key = String(row.legislature ?? '') + byLegislature.set(key, [...(byLegislature.get(key) ?? []), row]) + } + + const sample = [] + for (const key of [...byLegislature.keys()].sort()) { + const ordered = [...byLegislature.get(key)].sort( + (left, right) => + String(left.speechAt ?? '').localeCompare(String(right.speechAt ?? '')) || + Number(left.id) - Number(right.id), + ) + const indexes = new Set() + if (limit >= ordered.length) { + for (let index = 0; index < ordered.length; index += 1) indexes.add(index) + } else { + for (let index = 0; index < limit; index += 1) { + indexes.add(Math.floor((index * ordered.length) / limit)) + } + } + for (const index of [...indexes].sort((left, right) => left - right)) { + sample.push(ordered[index]) + } + } + return sample +} diff --git a/scripts/lib/cli.mjs b/scripts/lib/cli.mjs index 51df0618..533591b8 100644 --- a/scripts/lib/cli.mjs +++ b/scripts/lib/cli.mjs @@ -108,6 +108,47 @@ export const isTruthyEnv = (value) => value === 'true' || value === '1' export const isRemoteDbOverrideSet = () => isTruthyEnv(process.env[ALLOW_REMOTE_DB_FLAG]) +/** + * Hostname of a connection string, or null when it is missing/unparseable. + * + * @param {unknown} databaseUrl + * @returns {string | null} + */ +export const databaseHostname = (databaseUrl) => { + try { + return new URL(String(databaseUrl ?? '')).hostname + } catch { + return null + } +} + +/** + * True when the connection string targets one of the LOCAL_HOSTS. + * + * @param {unknown} databaseUrl + * @returns {boolean} + */ +export const isLocalDatabaseUrl = (databaseUrl) => { + const host = databaseHostname(databaseUrl) + return host !== null && LOCAL_HOSTS.has(host) +} +/** + * C155 write guard: any write run whose target is not provably local — or that + * runs under `NODE_ENV=production` (the homeserver env file) or with the + * `ALLOW_REMOTE_DB` escape — requires an explicit intent flag. The proxy socat + * used on the homeserver rewrites the host to 127.0.0.1, so the host check + * alone cannot see production; NODE_ENV is the honest signal there. Read-only + * modes never call this. + * + * @param {{ nodeEnv?: string | null, databaseUrl?: string | null, allowRemoteDb?: boolean }} [options] + * @returns {boolean} + */ +export const requiresWriteConfirm = ({ + nodeEnv = process.env.NODE_ENV, + databaseUrl = process.env.DATABASE_URL, + allowRemoteDb = isRemoteDbOverrideSet(), +} = {}) => nodeEnv === 'production' || allowRemoteDb || !isLocalDatabaseUrl(databaseUrl) + /** * Port for `next dev` (OPS40). Next's CLI resolves its port via commander's * `.env('PORT')` BEFORE `@next/env` loads `.env.local`, so a `PORT` written by diff --git a/src/utilities/speech/speechCoverage.ts b/src/utilities/speech/speechCoverage.ts new file mode 100644 index 00000000..a2bd9499 --- /dev/null +++ b/src/utilities/speech/speechCoverage.ts @@ -0,0 +1,124 @@ +import 'server-only' + +import { sql } from '@payloadcms/db-postgres' +import type { Payload, PayloadRequest } from 'payload' + +import { drizzleResultRows } from '@/utilities/drizzleBulk' +import { getPostgresTransactionDatabase } from '@/utilities/postgresTransactionLocks' + +/** + * C155 backfill coverage: what the catalog actually HAS, per legislature. + * The run report says what one execution did; this says what is stored, so a + * resumed/partial run cannot be mistaken for the final state. Read-only, one + * aggregate query (`--coverage` in the import CLI). + */ + +export type SpeechLegislatureCoverage = { + legislature: string | null + total: number + withExcerpt: number + withVideo: number + withSegments: number + withoutExcerpt: number + withoutSegments: number + fallbackYoutube: number +} + +export type SpeechCoverage = { + byLegislature: SpeechLegislatureCoverage[] + totals: Omit +} + +type CoverageRequest = { transactionID?: PayloadRequest['transactionID'] } + +type PostgresDb = { + execute: (query: ReturnType) => Promise +} + +const COVERAGE_QUERY = sql` + SELECT + s."legislature" AS "legislature", + COUNT(*) AS "total", + COUNT(*) FILTER (WHERE s."audio_id" IS NOT NULL) AS "with_excerpt", + COUNT(*) FILTER (WHERE s."vod_playback_url" IS NOT NULL) AS "with_video", + COUNT(*) FILTER (WHERE EXISTS ( + SELECT 1 FROM "speech_segment" seg WHERE seg."speech_id" = s."id" + )) AS "with_segments", + COUNT(*) FILTER (WHERE s."audio_id" IS NULL AND s."youtube_url" IS NOT NULL) AS "fallback_youtube" + FROM "speech" s + GROUP BY s."legislature" + ORDER BY s."legislature" NULLS LAST +` + +const toCount = (value: unknown): number => { + const count = Number(value ?? 0) + return Number.isFinite(count) ? count : 0 +} + +const rowToCoverage = (row: Record): SpeechLegislatureCoverage => { + const total = toCount(row.total) + const withExcerpt = toCount(row.with_excerpt) + const withSegments = toCount(row.with_segments) + return { + legislature: + row.legislature === null || row.legislature === undefined ? null : String(row.legislature), + total, + withExcerpt, + withVideo: toCount(row.with_video), + withSegments, + withoutExcerpt: total - withExcerpt, + withoutSegments: total - withSegments, + fallbackYoutube: toCount(row.fallback_youtube), + } +} + +const sumCoverage = ( + rows: SpeechLegislatureCoverage[], +): Omit => + rows.reduce( + (totals, row) => ({ + total: totals.total + row.total, + withExcerpt: totals.withExcerpt + row.withExcerpt, + withVideo: totals.withVideo + row.withVideo, + withSegments: totals.withSegments + row.withSegments, + withoutExcerpt: totals.withoutExcerpt + row.withoutExcerpt, + withoutSegments: totals.withoutSegments + row.withoutSegments, + fallbackYoutube: totals.fallbackYoutube + row.fallbackYoutube, + }), + { + total: 0, + withExcerpt: 0, + withVideo: 0, + withSegments: 0, + withoutExcerpt: 0, + withoutSegments: 0, + fallbackYoutube: 0, + }, + ) + +/** + * Per-legislature coverage of the speech catalog. Pass a `req` with an active + * transaction (int tests) to run the aggregate on that transaction's session; + * the CLI calls it without one and reads the pooled connection. + */ +export const getSpeechCoverage = async ( + payload: Payload, + req?: CoverageRequest, +): Promise => { + if (payload.db.name !== 'postgres') { + throw new Error('A cobertura do acervo exige o adaptador PostgreSQL.') + } + + let database: PostgresDb | undefined + if (req?.transactionID) { + database = await getPostgresTransactionDatabase(payload, req) + } else { + database = (payload.db as unknown as { drizzle?: PostgresDb }).drizzle + } + if (!database || typeof database.execute !== 'function') { + throw new Error('A sessão PostgreSQL da cobertura do acervo não está disponível.') + } + + const rows = drizzleResultRows(await database.execute(COVERAGE_QUERY)).map(rowToCoverage) + return { byLegislature: rows, totals: sumCoverage(rows) } +} diff --git a/tests/int/speechCoverage.int.spec.ts b/tests/int/speechCoverage.int.spec.ts new file mode 100644 index 00000000..4b982814 --- /dev/null +++ b/tests/int/speechCoverage.int.spec.ts @@ -0,0 +1,148 @@ +// @vitest-environment node + +import { sql } from '@payloadcms/db-postgres' +import { randomUUID } from 'node:crypto' +import type { Payload, RequiredDataFromCollectionSlug } from 'payload' +import { getPayload } from 'payload' +import { beforeAll, describe, expect, it } from 'vitest' + +import config from '@/payload.config' +import { getPostgresTransactionDatabase } from '@/utilities/postgresTransactionLocks' +import { + getSpeechCoverage, + type SpeechCoverage, + type SpeechLegislatureCoverage, +} from '@/utilities/speech/speechCoverage' + +let payload: Payload + +type SpeechCreateData = RequiredDataFromCollectionSlug<'speech'> + +const createSpeech = ( + transactionID: number | string, + data: SpeechCreateData, +): Promise<{ id: number }> => + payload.create({ + collection: 'speech', + data, + depth: 0, + req: { transactionID }, + // Intentional bypass: the spec seeds its own rows. + overrideAccess: true, + }) + +const bucketOf = (coverage: SpeechCoverage, legislature: string): SpeechLegislatureCoverage => { + const found = coverage.byLegislature.find((row) => row.legislature === legislature) + return ( + found ?? { + legislature, + total: 0, + withExcerpt: 0, + withVideo: 0, + withSegments: 0, + withoutExcerpt: 0, + withoutSegments: 0, + fallbackYoutube: 0, + } + ) +} + +const delta = ( + before: SpeechCoverage, + after: SpeechCoverage, + legislature: string, +): Omit => { + const from = bucketOf(before, legislature) + const to = bucketOf(after, legislature) + return { + total: to.total - from.total, + withExcerpt: to.withExcerpt - from.withExcerpt, + withVideo: to.withVideo - from.withVideo, + withSegments: to.withSegments - from.withSegments, + withoutExcerpt: to.withoutExcerpt - from.withoutExcerpt, + withoutSegments: to.withoutSegments - from.withoutSegments, + fallbackYoutube: to.fallbackYoutube - from.fallbackYoutube, + } +} + +describe('getSpeechCoverage (C155)', () => { + beforeAll(async () => { + payload = await getPayload({ config: await config }) + }) + + it('counts the buckets per legislature and derives the complements', async () => { + const transactionID = await payload.db.beginTransaction() + if (transactionID === null) throw new Error('Não foi possível abrir a transação do spec.') + try { + const database = await getPostgresTransactionDatabase(payload, { transactionID }) + // Fixed snapshot: the delta between the two reads cannot be moved by + // parallel specs committing their own fixtures. + await database.execute(sql`SET TRANSACTION ISOLATION LEVEL REPEATABLE READ`) + const before = await getSpeechCoverage(payload, { transactionID }) + + const withSegments = await createSpeech(transactionID, { + sourceKey: `coverage-${randomUUID()}`, + speechAt: '2012-05-01T10:00', + year: 2012, + legislature: '54', + classifiedBy: 'gazetteer', + audioId: 11, + excerptTMs: 1_000, + vodPlaybackUrl: 'https://vod/a', + vodDownloadUrl: 'https://vod/a.mp4', + }) + await payload.create({ + collection: 'speechSegment', + data: { + speech: withSegments.id, + order: 1, + startSeconds: 0, + endSeconds: 2, + text: 'a saúde pública baiana', + searchText: 'a saude publica baiana', + }, + depth: 0, + req: { transactionID }, + // Intentional bypass: the spec seeds its own rows. + overrideAccess: true, + }) + // Excerpt identified, VOD absent, no segments (the C153 "sem segmentos" gap). + await createSpeech(transactionID, { + sourceKey: `coverage-${randomUUID()}`, + speechAt: '2012-06-01T10:00', + year: 2012, + legislature: '54', + classifiedBy: 'gazetteer', + audioId: 12, + excerptTMs: 2_000, + }) + // No excerpt at all, but the event has a YouTube record (fallback bucket). + await createSpeech(transactionID, { + sourceKey: `coverage-${randomUUID()}`, + speechAt: '2012-07-01T10:00', + year: 2012, + legislature: '54', + classifiedBy: 'gazetteer', + youtubeUrl: 'https://www.youtube.com/watch?v=x', + }) + + const after = await getSpeechCoverage(payload, { transactionID }) + expect(delta(before, after, '54')).toEqual({ + total: 3, + withExcerpt: 2, + withVideo: 1, + withSegments: 1, + withoutExcerpt: 1, + withoutSegments: 2, + fallbackYoutube: 1, + }) + + const total = after.byLegislature.reduce((sum, row) => sum + row.total, 0) + expect(after.totals.total).toBe(total) + expect(after.totals.withExcerpt + after.totals.withoutExcerpt).toBe(total) + expect(after.totals.withSegments + after.totals.withoutSegments).toBe(total) + } finally { + await payload.db.rollbackTransaction(transactionID) + } + }) +}) diff --git a/tests/unit/camaraSpeeches.unit.spec.ts b/tests/unit/camaraSpeeches.unit.spec.ts index 1411d647..ff17b671 100644 --- a/tests/unit/camaraSpeeches.unit.spec.ts +++ b/tests/unit/camaraSpeeches.unit.spec.ts @@ -3,6 +3,7 @@ import { describe, expect, it } from 'vitest' import { + aggregateBackfillRuns, buildVodUrl, clockToSeconds, legislatureForDate, @@ -15,6 +16,7 @@ import { parsePresidingOfficerTransitions, parseVodStatus, resolvePresidingOfficer, + selectLinkSample, selectSpeechEvents, speechContentHash, speechDate, @@ -476,3 +478,139 @@ describe('normalizeTranscription', () => { expect(result.segments).toEqual([{ start: 0, end: 1, text: 'seg' }]) }) }) + +describe('aggregateBackfillRuns', () => { + it('sums the numeric blocks of every legislature and keeps the first LLM sample error', () => { + const combined = aggregateBackfillRuns([ + { + totals: { + listed: 10, + processed: 10, + created: 8, + updated: 2, + withExcerpt: 9, + withoutExcerpt: 1, + withSegments: 8, + failed: 1, + }, + asr: { calls: 8, audioSeconds: 480, elapsedMs: 9000, failures: 0 }, + llm: { + calls: 10, + used: 10, + failed: 0, + totalTokens: 1000, + estimatedCostUsd: 0.01, + sampleError: null, + }, + elapsedMs: 12_000, + }, + { + totals: { + listed: 5, + processed: 5, + created: 0, + updated: 5, + withExcerpt: 5, + withoutExcerpt: 0, + withSegments: 5, + failed: 0, + }, + asr: { calls: 0, audioSeconds: 0, elapsedMs: 0, failures: 0 }, + llm: { + calls: 5, + used: 4, + failed: 1, + totalTokens: 400, + estimatedCostUsd: 0.004, + sampleError: 'timeout', + }, + elapsedMs: 3_000, + }, + ]) + + expect(combined.totals).toEqual({ + listed: 15, + processed: 15, + created: 8, + updated: 7, + withExcerpt: 14, + withoutExcerpt: 1, + withSegments: 13, + failed: 1, + }) + expect(combined.asr).toEqual({ calls: 8, audioSeconds: 480, elapsedMs: 9000, failures: 0 }) + expect(combined.llm.calls).toBe(15) + expect(combined.llm.used).toBe(14) + expect(combined.llm.failed).toBe(1) + expect(combined.llm.totalTokens).toBe(1400) + expect(combined.llm.estimatedCostUsd).toBeCloseTo(0.014, 6) + expect(combined.llm.sampleError).toBe('timeout') + expect(combined.elapsedMs).toBe(15_000) + }) + + it('returns zeroed blocks for an empty/absent run list', () => { + expect(aggregateBackfillRuns([]).totals.processed).toBe(0) + expect(aggregateBackfillRuns(undefined).asr.audioSeconds).toBe(0) + expect(aggregateBackfillRuns(undefined).llm.sampleError).toBeNull() + expect(aggregateBackfillRuns(undefined).elapsedMs).toBe(0) + }) +}) + +describe('selectLinkSample', () => { + const rows = [ + { + id: 1, + legislature: '57', + speechAt: '2023-03-01T10:00', + vodPlaybackUrl: 'https://vod/1', + vodDownloadUrl: 'https://vod/1.mp4', + }, + { + id: 2, + legislature: '57', + speechAt: '2023-02-01T10:00', + vodPlaybackUrl: null, + vodDownloadUrl: null, + }, + { + id: 3, + legislature: '57', + speechAt: '2023-04-01T10:00', + vodPlaybackUrl: 'https://vod/3', + vodDownloadUrl: null, + }, + { + id: 4, + legislature: '56', + speechAt: '2021-02-01T10:00', + vodPlaybackUrl: 'https://vod/4', + vodDownloadUrl: 'https://vod/4.mp4', + }, + ] + + it('picks up to n linked speeches per legislature, skipping link-less rows', () => { + expect( + selectLinkSample(rows, 1) + .map((row) => row.id) + .sort(), + ).toEqual([1, 4]) + expect( + selectLinkSample(rows, 5) + .map((row) => row.id) + .sort(), + ).toEqual([1, 3, 4]) + }) + + it('is deterministic for the same rows', () => { + expect(selectLinkSample(rows, 2)).toEqual(selectLinkSample(rows, 2)) + }) + + it('returns an empty sample when nothing carries a link', () => { + expect( + selectLinkSample( + [{ id: 9, legislature: '55', speechAt: null, vodPlaybackUrl: null, vodDownloadUrl: null }], + 3, + ), + ).toEqual([]) + }) +}) diff --git a/tests/unit/cliEnvFlags.unit.spec.ts b/tests/unit/cliEnvFlags.unit.spec.ts index e8109b36..12ea481a 100644 --- a/tests/unit/cliEnvFlags.unit.spec.ts +++ b/tests/unit/cliEnvFlags.unit.spec.ts @@ -2,7 +2,14 @@ import { afterEach, describe, expect, it } from 'vitest' -import { ALLOW_REMOTE_DB_FLAG, isRemoteDbOverrideSet, isTruthyEnv } from '../../scripts/lib/cli.mjs' +import { + ALLOW_REMOTE_DB_FLAG, + databaseHostname, + isLocalDatabaseUrl, + isRemoteDbOverrideSet, + isTruthyEnv, + requiresWriteConfirm, +} from '../../scripts/lib/cli.mjs' describe('isTruthyEnv (OPS52-media-guard — one spelling of the true/1 intent-flag semantics)', () => { it('accepts exactly true and 1', () => { @@ -42,3 +49,75 @@ describe('isRemoteDbOverrideSet (behavior preserved through the delegation refac expect(isRemoteDbOverrideSet()).toBe(false) }) }) + +describe('databaseHostname / isLocalDatabaseUrl (C155)', () => { + it('reads the hostname and refuses missing/invalid connection strings', () => { + expect(databaseHostname('postgresql://teqo:teqo@localhost:5432/teqo_wt155')).toBe('localhost') + expect(databaseHostname('postgresql://user:pw@postgres:5432/teqo_1313')).toBe('postgres') + expect(databaseHostname(undefined)).toBeNull() + expect(databaseHostname('not a url')).toBeNull() + }) + + it('accepts the local allowlist and refuses every other host', () => { + expect(isLocalDatabaseUrl('postgresql://teqo:teqo@localhost:5432/teqo_wt155')).toBe(true) + expect(isLocalDatabaseUrl('postgresql://teqo:teqo@127.0.0.1:5433/teqo_1313')).toBe(true) + expect(isLocalDatabaseUrl('postgresql://teqo:teqo@postgres:5432/teqo_1313')).toBe(true) + expect(isLocalDatabaseUrl('postgresql://teqo:teqo@db.example.com:5432/teqo_1313')).toBe(false) + expect(isLocalDatabaseUrl(undefined)).toBe(false) + }) +}) + +describe('requiresWriteConfirm (C155 write guard)', () => { + it('stays quiet for a local target in dev/test', () => { + expect( + requiresWriteConfirm({ + nodeEnv: 'test', + databaseUrl: 'postgresql://teqo:teqo@localhost:5432/teqo_wt155', + allowRemoteDb: false, + }), + ).toBe(false) + }) + + it('demands confirmation under NODE_ENV=production even through the local proxy', () => { + expect( + requiresWriteConfirm({ + nodeEnv: 'production', + databaseUrl: 'postgresql://teqo:teqo@127.0.0.1:5433/teqo_1313', + allowRemoteDb: false, + }), + ).toBe(true) + }) + + it('demands confirmation for a remote host or the ALLOW_REMOTE_DB escape', () => { + expect( + requiresWriteConfirm({ + nodeEnv: 'development', + databaseUrl: 'postgresql://teqo:teqo@db.example.com:5432/teqo_1313', + allowRemoteDb: false, + }), + ).toBe(true) + expect( + requiresWriteConfirm({ + nodeEnv: 'development', + databaseUrl: 'postgresql://teqo:teqo@localhost:5432/teqo_wt155', + allowRemoteDb: true, + }), + ).toBe(true) + }) + + it('fails closed on a missing/unparseable URL and keeps the exact-match env semantics', () => { + // `null`/empty/`'not a url'` never reach LOCAL_HOSTS; `undefined` would + // fall back to the real process env by design (the callers' default). + expect(requiresWriteConfirm({ nodeEnv: 'development', databaseUrl: null })).toBe(true) + expect(requiresWriteConfirm({ nodeEnv: 'development', databaseUrl: '' })).toBe(true) + expect(requiresWriteConfirm({ nodeEnv: 'development', databaseUrl: 'not a url' })).toBe(true) + // `Production` (capitalized) is not the env file's value — no case-folding. + expect( + requiresWriteConfirm({ + nodeEnv: 'Production', + databaseUrl: 'postgresql://teqo:teqo@localhost:5432/teqo_wt155', + allowRemoteDb: false, + }), + ).toBe(false) + }) +})