mirror of
https://github.com/fsolla/teqo.git
synced 2026-11-13 06:43:06 +00:00
refactor(C155): simplify pós-review — backoff no camaraFetch, runLegislature e ajustes de guard/relatório
This commit is contained in:
Francisco Solla
parent
a41ee0d11d
commit
4f5f32dfef
8 changed files with 117 additions and 75 deletions
|
|
@ -1 +1 @@
|
|||
**C155 — Backfill 2011–2026 + import de produção do acervo de falas (2026-09-13):** o catálogo do C153 ganhou o pipeline de acervo completo. O `pnpm camara:import` ganhou três modos: `--all` (54ª–57ª num processo, com relatório por legislatura e checkpoint por legislatura), `--coverage` (read-only, retrato do banco por legislatura: total, com trecho, com vídeo, com segmentos, sem trecho, sem segmentos, fallback YouTube) e `--verify-links <n>` (amostra determinística por legislatura com GET Range/HEAD nos links de playback/download do VOD, warning não-fatal). O guard de escrita `CAMARA_IMPORT_CONFIRM=1` (predicado puro `requiresWriteConfirm`/`isLocalDatabaseUrl` em `scripts/lib/cli.mjs`) fail-closa antes de conectar quando o alvo não é provadamente local — `NODE_ENV=production` (env file do homeserver), host remoto ou `ALLOW_REMOTE_DB`; `--coverage`/`--verify-links` são isentos. A paginação da API ganhou backoff (`fetchJsonWithBackoff`, 5 tentativas exponenciais) porque a listagem da 55ª devolveu 500 persistente na página 4. Cobertura via `getSpeechCoverage` (`src/utilities/speech/speechCoverage.ts`, query agregada `COUNT(*) FILTER` + `drizzleResultRows`, aceita sessão transacional para os int tests). Runbook novo em `docs/ops/teqo-1313-deploy.md` §C155 (pré-requisitos, comandos com o proxy 5433, rollback por legislatura). Testes: unit de `aggregateBackfillRuns`/`selectLinkSample`/`requiresWriteConfirm` + int `speechCoverage` (deltas exatos por bucket sob `REPEATABLE READ`). Prova viva em produção (`teqo_1313`, deploy `424ee311` já com o schema C153/C154): smoke idempotente (`--all --limit 1` duas vezes → 0 criados/0 ASR/0 LLM na segunda) e backfill completo com relatório por legislatura. **Resultado:** _preenchido após o run_.
|
||||
**C155 — Backfill 2011–2026 + import de produção do acervo de falas (2026-09-13):** o catálogo do C153 ganhou o pipeline de acervo completo. O `pnpm camara:import` ganhou três modos: `--all` (54ª–57ª num processo, com relatório por legislatura e checkpoint por legislatura), `--coverage` (read-only, retrato do banco por legislatura: total, com trecho, com vídeo, com segmentos, sem trecho, sem segmentos, fallback YouTube) e `--verify-links <n>` (amostra determinística por legislatura com GET Range/HEAD nos links de playback/download do VOD, warning não-fatal). O guard de escrita `CAMARA_IMPORT_CONFIRM=1` (predicado puro `requiresWriteConfirm`/`isLocalDatabaseUrl` em `scripts/lib/cli.mjs`) fail-closa antes de conectar quando o alvo não é provadamente local — `NODE_ENV=production` (env file do homeserver), host remoto ou `ALLOW_REMOTE_DB`; `--coverage`/`--verify-links` são isentos. A paginação da API ganhou backoff (`fetchJsonWithBackoff`, 5 tentativas exponenciais) porque a listagem da 55ª devolveu 500 persistente na página 4. Cobertura via `getSpeechCoverage` (`src/utilities/speech/speechCoverage.ts`, query agregada `COUNT(*) FILTER` + `drizzleResultRows`, aceita sessão transacional para os int tests). Runbook novo em `docs/ops/teqo-1313-deploy.md` §C155 (pré-requisitos, comandos com o proxy 5433, rollback por legislatura). Testes: unit de `aggregateBackfillRuns`/`selectLinkSample`/`requiresWriteConfirm` + int `speechCoverage` (deltas exatos por bucket sob `REPEATABLE READ`). Prova viva em produção (`teqo_1313`, deploy `424ee311` já com o schema C153/C154): smoke idempotente (`--all --limit 1` duas vezes → 0 criados/0 ASR/0 LLM na segunda) e backfill completo em execução no homeserver; a 55ª será retomada com `--legislature 55` (a listagem da API responde 500 na página 4). **Resultado:** _a preencher no fechamento do PR._
|
||||
|
|
|
|||
|
|
@ -103,7 +103,7 @@ flowchart TB
|
|||
#### 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-<runAt>.json`, `coverage-<runAt>.json`, `verify-links-<runAt>.json`.
|
||||
- **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/backups/teqo-camara/reports`) e anexado à Issue (upload web). Arquivos: `backfill-<runAt>.json`, `coverage-<runAt>.json`, `verify-links-<runAt>.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)
|
||||
|
|
|
|||
|
|
@ -36,6 +36,7 @@ import {
|
|||
eventPageUrl,
|
||||
eventsUrl,
|
||||
getJson,
|
||||
getJsonWithBackoff,
|
||||
getText,
|
||||
probeVodLink,
|
||||
resolveVod,
|
||||
|
|
@ -226,36 +227,12 @@ 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 <n>`).
|
||||
*/
|
||||
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 fetchJsonWithBackoff(`${speechesUrl(deputyId, from, to)}&pagina=${page}`)
|
||||
const body = await getJsonWithBackoff(`${speechesUrl(deputyId, from, to)}&pagina=${page}`, {
|
||||
label: 'camara:import',
|
||||
})
|
||||
const batch = body.dados ?? []
|
||||
speeches.push(...batch)
|
||||
const hasNext = (body.links ?? []).some((link) => link?.rel === 'next')
|
||||
|
|
@ -552,7 +529,9 @@ const printBackfillReport = (combined) => {
|
|||
`${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' : ''}`,
|
||||
`${(run.elapsedMs / 1000).toFixed(0)}s` +
|
||||
`${run.failures.length > 0 ? ` · ${run.failures.length} falha(s)` : ''}` +
|
||||
`${run.aborted ? ' · ABORTADA' : ''}`,
|
||||
)
|
||||
}
|
||||
const { totals, asr, llm } = combined
|
||||
|
|
@ -603,7 +582,8 @@ const writeReport = async (options, filename, report) => {
|
|||
}
|
||||
|
||||
const runOptions = (options) => ({
|
||||
legislature: options.all ? null : options.legislature,
|
||||
// A `--date` run is not a legislature run — C153 shape kept.
|
||||
legislature: options.all || options.date ? null : options.legislature,
|
||||
all: options.all,
|
||||
date: options.date,
|
||||
limit: options.limit,
|
||||
|
|
@ -647,27 +627,40 @@ const processTargets = async (payload, options, targets, run) => {
|
|||
}
|
||||
}
|
||||
|
||||
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,
|
||||
})
|
||||
/**
|
||||
* One legislature end-to-end: list from the API, slice `--limit`, process,
|
||||
* close the run. Throws on a list failure or an empty range — the caller owns
|
||||
* the policy (single range dies; `--all` marks the legislature aborted and
|
||||
* continues with the next).
|
||||
*/
|
||||
const runLegislature = async (payload, options, { legislature, from, to, label }) => {
|
||||
const run = createRun(options, { legislature, from, to })
|
||||
const startedAt = Date.now()
|
||||
console.log(
|
||||
`[camara:import] deputado ${SOLLA_DEPUTY_ID} | ${from}..${to} | speaker "${options.speaker}"`,
|
||||
)
|
||||
console.log(`\n[camara:import] === ${label} (${from}..${to}) ===`)
|
||||
|
||||
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}.`)
|
||||
if (targets.length === 0) throw new Error(`nenhum discurso em ${from}..${to}.`)
|
||||
run.totals.listed = speeches.length
|
||||
|
||||
await processTargets(payload, options, targets, run)
|
||||
run.elapsedMs = Date.now() - startedAt
|
||||
return run
|
||||
}
|
||||
|
||||
const runSingleRange = async (payload, options) => {
|
||||
const from = options.date ?? LEGISLATURE_RANGES[options.legislature][0]
|
||||
const to = options.date ?? LEGISLATURE_RANGES[options.legislature][1]
|
||||
console.log(
|
||||
`[camara:import] deputado ${SOLLA_DEPUTY_ID} | ${from}..${to} | speaker "${options.speaker}"`,
|
||||
)
|
||||
const run = await runLegislature(payload, options, {
|
||||
legislature: options.date ? null : options.legislature,
|
||||
from,
|
||||
to,
|
||||
label: options.date ? `date ${options.date}` : `legislatura ${options.legislature}ª`,
|
||||
})
|
||||
run.coverage = await getSpeechCoverage(payload)
|
||||
printReport(run)
|
||||
printCoverage(run.coverage)
|
||||
|
|
@ -683,10 +676,7 @@ const runBackfill = async (payload, options) => {
|
|||
mode: 'all',
|
||||
options: runOptions(options),
|
||||
legislatures: [],
|
||||
totals: null,
|
||||
asr: null,
|
||||
llm: null,
|
||||
elapsedMs: 0,
|
||||
...aggregateBackfillRuns([]),
|
||||
failures: [],
|
||||
coverage: null,
|
||||
}
|
||||
|
|
@ -696,19 +686,19 @@ const runBackfill = async (payload, options) => {
|
|||
|
||||
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}) ===`)
|
||||
let run
|
||||
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)
|
||||
run = await runLegislature(payload, options, {
|
||||
legislature,
|
||||
from,
|
||||
to,
|
||||
label: `legislatura ${legislature}ª`,
|
||||
})
|
||||
} catch (error) {
|
||||
run = createRun(options, { legislature, from, to })
|
||||
run.aborted = true
|
||||
run.elapsedMs = Date.now() - rangeStartedAt
|
||||
run.failures.push({
|
||||
sourceKey: null,
|
||||
speechAt: null,
|
||||
|
|
@ -719,7 +709,6 @@ const runBackfill = async (payload, options) => {
|
|||
`[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 })))
|
||||
|
||||
|
|
@ -803,13 +792,15 @@ const runVerifyLinks = async (payload, options) => {
|
|||
warnings,
|
||||
elapsedMs: Date.now() - startedAt,
|
||||
}
|
||||
if (checked.length === 0) {
|
||||
die('nenhum link de VOD verificado — amostra vazia (banco sem links?).')
|
||||
}
|
||||
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?).')
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
|
|
@ -817,13 +808,11 @@ const runVerifyLinks = async (payload, options) => {
|
|||
// ---------------------------------------------------------------------------
|
||||
|
||||
const databaseTarget = () => {
|
||||
const host = databaseHostname(process.env.DATABASE_URL)
|
||||
const url = process.env.DATABASE_URL
|
||||
const host = databaseHostname(url)
|
||||
if (host === null) return '(DATABASE_URL ausente ou inválida)'
|
||||
try {
|
||||
return `${host}${new URL(process.env.DATABASE_URL).pathname}`
|
||||
} catch {
|
||||
return host
|
||||
}
|
||||
// `databaseHostname` only returns non-null when `new URL` already parsed.
|
||||
return `${host}${new URL(url).pathname}`
|
||||
}
|
||||
|
||||
const modeLabel = (options) => {
|
||||
|
|
@ -858,13 +847,15 @@ async function main() {
|
|||
|
||||
const readOnly = options.coverage || options.verifyLinks !== null
|
||||
if (readOnly) {
|
||||
if (!process.env.DATABASE_URL) die('DATABASE_URL is not set. Refusing to continue.')
|
||||
if (!process.env.DATABASE_URL) die('DATABASE_URL não definida; recusando continuar.')
|
||||
} else {
|
||||
// Confirm first: a non-local host is exactly what the flag is for; the
|
||||
// local-DB guard below still blocks a remote host without ALLOW_REMOTE_DB.
|
||||
assertWriteAllowed()
|
||||
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)}`)
|
||||
|
||||
|
|
|
|||
|
|
@ -40,6 +40,37 @@ export async function getText(url, { attempts = 3, ...options } = {}) {
|
|||
|
||||
export const getJson = async (url, options) => JSON.parse(await getText(url, options))
|
||||
|
||||
/**
|
||||
* Pagination-grade GET: the open-data API throws transient 500s and hangs on
|
||||
* deep pages (seen 2026-09-13 on the 55ª page 4), and `getText`'s own 3
|
||||
* attempts are too impatient for a 1.011-speech backfill. Bounded backoff;
|
||||
* the caller owns the failure policy (the import aborts only the legislature).
|
||||
*
|
||||
* @param {string} url
|
||||
* @param {{ attempts?: number, timeoutMs?: number, label?: string }} [options]
|
||||
*/
|
||||
export async function getJsonWithBackoff(
|
||||
url,
|
||||
{ attempts = 5, timeoutMs = 45_000, label = 'camara' } = {},
|
||||
) {
|
||||
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(
|
||||
`[${label}] ${error?.message ?? error} — retry ${attempt}/${attempts - 1} em ${waitMs / 1000}s`,
|
||||
)
|
||||
await sleep(waitMs)
|
||||
}
|
||||
}
|
||||
}
|
||||
throw lastError
|
||||
}
|
||||
|
||||
export const downloadToBuffer = async (url, { attempts = 3, timeoutMs = 120_000 } = {}) => {
|
||||
let lastError
|
||||
for (let attempt = 1; attempt <= attempts; attempt += 1) {
|
||||
|
|
|
|||
|
|
@ -447,9 +447,9 @@ export const aggregateBackfillRuns = (runs) => {
|
|||
* 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 {Array<{ id: number, legislature?: string | null, speechAt?: string | null, vodPlaybackUrl?: string | null, vodDownloadUrl?: string | null }> | null | undefined} rows
|
||||
* @param {number} perLegislature
|
||||
* @returns {typeof rows}
|
||||
* @returns {Array<{ id: number, legislature?: string | null, speechAt?: string | null, vodPlaybackUrl?: string | null, vodDownloadUrl?: string | null }>}
|
||||
*/
|
||||
export const selectLinkSample = (rows, perLegislature) => {
|
||||
const limit = Math.max(1, Math.trunc(Number(perLegislature) || 1))
|
||||
|
|
|
|||
|
|
@ -132,6 +132,7 @@ 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
|
||||
|
|
|
|||
|
|
@ -4,7 +4,10 @@ import { sql } from '@payloadcms/db-postgres'
|
|||
import type { Payload, PayloadRequest } from 'payload'
|
||||
|
||||
import { drizzleResultRows } from '@/utilities/drizzleBulk'
|
||||
import { getPostgresTransactionDatabase } from '@/utilities/postgresTransactionLocks'
|
||||
import {
|
||||
getPostgresTransactionDatabase,
|
||||
type PostgresTransactionDatabase,
|
||||
} from '@/utilities/postgresTransactionLocks'
|
||||
|
||||
/**
|
||||
* C155 backfill coverage: what the catalog actually HAS, per legislature.
|
||||
|
|
@ -31,10 +34,6 @@ export type SpeechCoverage = {
|
|||
|
||||
type CoverageRequest = { transactionID?: PayloadRequest['transactionID'] }
|
||||
|
||||
type PostgresDb = {
|
||||
execute: (query: ReturnType<typeof sql>) => Promise<unknown>
|
||||
}
|
||||
|
||||
const COVERAGE_QUERY = sql`
|
||||
SELECT
|
||||
s."legislature" AS "legislature",
|
||||
|
|
@ -109,11 +108,11 @@ export const getSpeechCoverage = async (
|
|||
throw new Error('A cobertura do acervo exige o adaptador PostgreSQL.')
|
||||
}
|
||||
|
||||
let database: PostgresDb | undefined
|
||||
let database: PostgresTransactionDatabase | undefined
|
||||
if (req?.transactionID) {
|
||||
database = await getPostgresTransactionDatabase(payload, req)
|
||||
} else {
|
||||
database = (payload.db as unknown as { drizzle?: PostgresDb }).drizzle
|
||||
database = (payload.db as unknown as { drizzle?: PostgresTransactionDatabase }).drizzle
|
||||
}
|
||||
if (!database || typeof database.execute !== 'function') {
|
||||
throw new Error('A sessão PostgreSQL da cobertura do acervo não está disponível.')
|
||||
|
|
|
|||
|
|
@ -601,6 +601,26 @@ describe('selectLinkSample', () => {
|
|||
).toEqual([1, 3, 4])
|
||||
})
|
||||
|
||||
it('spreads the picks over a legislature larger than n', () => {
|
||||
const many = Array.from({ length: 6 }, (_, index) => ({
|
||||
id: index + 1,
|
||||
legislature: '55',
|
||||
speechAt: `2016-0${index + 1}-01T10:00`,
|
||||
vodPlaybackUrl: `https://vod/${index + 1}`,
|
||||
vodDownloadUrl: null,
|
||||
}))
|
||||
expect(
|
||||
selectLinkSample(many, 2)
|
||||
.map((row) => row.id)
|
||||
.sort(),
|
||||
).toEqual([1, 4])
|
||||
expect(
|
||||
selectLinkSample(many, 3)
|
||||
.map((row) => row.id)
|
||||
.sort(),
|
||||
).toEqual([1, 3, 5])
|
||||
})
|
||||
|
||||
it('is deterministic for the same rows', () => {
|
||||
expect(selectLinkSample(rows, 2)).toEqual(selectLinkSample(rows, 2))
|
||||
})
|
||||
|
|
|
|||
Loading…
Reference in a new issue