`scripts/pull-la-newspapers-ocr.ts`

174 lines
/**
 * pull-la-newspapers-ocr.ts
 *
 * Phase 2 of the LA newspaper ingest. Pulls IA `_djvu.txt` plain-text OCR
 * for issues with NULL ocr_text in news_issue. Uses the official Internet
 * Archive `ia` CLI (Python, on-policy bulk tool) with 10 parallel workers.
 * Each worker downloads → reads file → upserts to PG → deletes file.
 *
 * Source: https://archive.org/details/{identifier} via `ia download`
 *
 * Usage: npx tsx scripts/pull-la-newspapers-ocr.ts [--paper=LADailyHerald]
 *                                                  [--limit=100]
 *                                                  [--concurrency=10]
 *                                                  [--workdir=/tmp/ia-ocr]
 */
import { Pool } from 'pg';
import { spawn } from 'node:child_process';
import { readFile, mkdir, rm, stat } from 'node:fs/promises';
import { join } from 'node:path';

const pool = new Pool({
  host: process.env.PGHOST ?? '/tmp',
  database: process.env.PGDATABASE ?? 'stayclaim',
  user: process.env.PGUSER ?? process.env.USER,
  password: process.env.PGPASSWORD,
  port: parseInt(process.env.PGPORT ?? '5432', 10),
  max: 12,
});

const args = Object.fromEntries(
  process.argv.slice(2).map(a => {
    const m = a.match(/^--([^=]+)(?:=(.*))?$/);
    return m ? [m[1], m[2] ?? 'true'] : [a, 'true'];
  })
);
const PAPER       = args.paper as string | undefined;
const LIMIT       = args.limit ? parseInt(args.limit as string, 10) : undefined;
const CONCURRENCY = args.concurrency ? parseInt(args.concurrency as string, 10) : 10;
const WORKDIR     = (args.workdir as string) || '/tmp/ia-ocr';
const MAX_BYTES   = (args['skip-large'] ? parseFloat(args['skip-large'] as string) : 5) * 1024 * 1024;

type PendingRow = { id: string; source_identifier: string; paper_short: string };

async function loadQueue(): Promise<PendingRow[]> {
  const params: any[] = [];
  let where = 'ocr_text IS NULL';
  if (PAPER) { where += ` AND paper_short = $${params.length+1}`; params.push(PAPER); }
  let sql = `SELECT id, source_identifier, paper_short FROM news_issue WHERE ${where} ORDER BY issue_date NULLS LAST`;
  if (LIMIT) sql += ` LIMIT ${LIMIT}`;
  const { rows } = await pool.query<PendingRow>(sql, params);
  return rows;
}

function iaDownload(identifier: string, cwd: string): Promise<{ ok: boolean; reason?: string }> {
  return new Promise((resolve) => {
    const args = [
      'download', identifier,
      '--glob=*_djvu.txt',
      '--retries=3',
      '--no-change-timestamp',
      '-q',
    ];
    const proc = spawn('ia', args, { cwd, stdio: ['ignore', 'pipe', 'pipe'] });
    let stderr = '';
    proc.stderr.on('data', (d) => { stderr += d.toString(); });
    proc.on('close', (code) => {
      if (code === 0) resolve({ ok: true });
      else resolve({ ok: false, reason: `ia exit ${code}: ${stderr.slice(0, 200)}` });
    });
    proc.on('error', (e) => resolve({ ok: false, reason: String(e) }));
  });
}

const stats = {
  processed: 0,
  fetched: 0,
  empty: 0,
  failed: 0,
  bytes: 0,
};

async function processOne(row: PendingRow): Promise<void> {
  const subdir = join(WORKDIR, row.source_identifier);
  const txtPath = join(subdir, `${row.source_identifier}_djvu.txt`);
  try {
    const r = await iaDownload(row.source_identifier, WORKDIR);
    if (!r.ok) {
      stats.failed++;
      console.error(`  FAIL ${row.source_identifier}: ${r.reason}`);
      return;
    }
    let text: string;
    try {
      const st = await stat(txtPath);
      if (st.size > MAX_BYTES) {
        stats.empty++;
        await pool.query(`UPDATE news_issue SET ocr_text='', ocr_fetched_at=now() WHERE id=$1`, [row.id]);
        return;
      }
      text = await readFile(txtPath, 'utf8');
    } catch {
      // ia returned ok but no _djvu.txt found — issue may not have OCR
      stats.empty++;
      await pool.query(`UPDATE news_issue SET ocr_text='', ocr_fetched_at=now() WHERE id=$1`, [row.id]);
      return;
    }
    await pool.query(
      `UPDATE news_issue SET ocr_text=$1, ocr_fetched_at=now() WHERE id=$2`,
      [text, row.id]
    );
    stats.fetched++;
    stats.bytes += text.length;
  } catch (e) {
    stats.failed++;
    console.error(`  FAIL ${row.source_identifier}: ${(e as Error).message}`);
  } finally {
    stats.processed++;
    // clean up downloaded file/dir
    await rm(subdir, { recursive: true, force: true }).catch(() => {});
  }
}

async function worker(workerId: number, queue: PendingRow[]): Promise<void> {
  while (queue.length > 0) {
    const row = queue.shift();
    if (!row) break;
    await processOne(row);
  }
}

async function main() {
  await mkdir(WORKDIR, { recursive: true });
  // wipe any stragglers from a previous run
  for await (const _ of [] as any[]) {}
  const queue = await loadQueue();
  console.log(`pull-la-newspapers-ocr (ia parallel)`);
  console.log(`  paper=${PAPER ?? '(all)'}  queue=${queue.length}  concurrency=${CONCURRENCY}  workdir=${WORKDIR}`);
  if (queue.length === 0) {
    console.log('  nothing to do');
    await pool.end();
    return;
  }

  const t0 = Date.now();
  const reporter = setInterval(() => {
    const dt = (Date.now() - t0) / 1000;
    const rate = stats.processed / dt;
    const remaining = queue.length;
    const etaSec = remaining / Math.max(rate, 0.1);
    const mb = (stats.bytes / 1024 / 1024).toFixed(1);
    process.stdout.write(
      `  ${stats.processed} done (ok=${stats.fetched} empty=${stats.empty} fail=${stats.failed}) — ${rate.toFixed(2)}/s — ${mb} MB — ${remaining} left, ETA ${(etaSec/60).toFixed(1)}m   \r`
    );
  }, 2000);

  const workers = Array.from({ length: CONCURRENCY }, (_, i) => worker(i, queue));
  await Promise.all(workers);

  clearInterval(reporter);
  const dt = (Date.now() - t0) / 1000;
  console.log('');
  console.log('--- done ---');
  console.log(`  processed=${stats.processed}  fetched=${stats.fetched}  empty=${stats.empty}  failed=${stats.failed}`);
  console.log(`  elapsed=${dt.toFixed(1)}s  rate=${(stats.processed/dt).toFixed(2)}/s`);
  console.log(`  bytes=${(stats.bytes/1024/1024).toFixed(1)} MB`);
  await rm(WORKDIR, { recursive: true, force: true }).catch(() => {});
  await pool.end();
}

main().catch(e => {
  console.error('FATAL', e);
  process.exit(1);
});