`scrapers/tmdb-download-ids.js`

140 lines
#!/usr/bin/env node
'use strict';
// Phase A — download TMDB's daily ID exports (no auth required) and bulk-load
// them as queued stubs in asim_ingest_stubs. After this runs we have a complete
// checklist of every TMDB id that exists; Phase B worker chips away at it.
//
// Exports are gzipped JSONL — one line per record, fields {id, popularity, ...}.
// Files live at https://files.tmdb.org/p/exports/<type>_ids_MM_DD_YYYY.json.gz
// and roll over daily at 8AM UTC.
//
// Legal: ID exports + facts are explicitly open-use per TMDB terms. Attribution
// required on any user-facing page (handled in server.js home view).

require('dotenv').config();
const fs = require('fs');
const path = require('path');
const zlib = require('zlib');
const readline = require('readline');
const { Pool } = require('pg');

const RAW_DIR = path.join(__dirname, '..', 'data', 'raw', 'tmdb-id-exports');
fs.mkdirSync(RAW_DIR, { recursive: true });

const pool = new Pool({
  connectionString: process.env.DATABASE_URL || 'postgresql:///dw_unified?host=/tmp&user=stevestudio2',
  max: 4,
});

// TMDB rotates the export at 8AM UTC. If we're before 8AM, use yesterday's file.
function exportDateStr() {
  const d = new Date();
  if (d.getUTCHours() < 8) d.setUTCDate(d.getUTCDate() - 1);
  const MM = String(d.getUTCMonth() + 1).padStart(2, '0');
  const DD = String(d.getUTCDate()).padStart(2, '0');
  const YYYY = d.getUTCFullYear();
  return `${MM}_${DD}_${YYYY}`;
}

// Each entry: source label we use internally, TMDB export file prefix.
const EXPORTS = [
  { source: 'tmdb-movie',      file: 'movie_ids' },
  { source: 'tmdb-tv',         file: 'tv_series_ids' },
  { source: 'tmdb-person',     file: 'person_ids' },
  { source: 'tmdb-collection', file: 'collection_ids' },
  { source: 'tmdb-company',    file: 'production_company_ids' },
  { source: 'tmdb-network',    file: 'tv_network_ids' },
  { source: 'tmdb-keyword',    file: 'keyword_ids' },
];

async function downloadOne(prefix, dateStr) {
  const url = `https://files.tmdb.org/p/exports/${prefix}_${dateStr}.json.gz`;
  const dest = path.join(RAW_DIR, `${prefix}_${dateStr}.json.gz`);
  if (fs.existsSync(dest) && fs.statSync(dest).size > 1000) {
    return { url, dest, cached: true, bytes: fs.statSync(dest).size };
  }
  const r = await fetch(url);
  if (!r.ok) throw new Error(`${prefix}: HTTP ${r.status}`);
  const buf = Buffer.from(await r.arrayBuffer());
  fs.writeFileSync(dest, buf);
  return { url, dest, cached: false, bytes: buf.length };
}

async function bulkLoadStubs(source, gzPath, runId) {
  // Stream-parse the gzipped JSONL and bulk-INSERT in chunks of 5000.
  // Use ON CONFLICT DO UPDATE to refresh popularity on subsequent runs.
  const stream = fs.createReadStream(gzPath).pipe(zlib.createGunzip());
  const rl = readline.createInterface({ input: stream, crlfDelay: Infinity });

  let batch = [];
  const BATCH_SIZE = 5000;
  let total = 0;
  let inserted = 0;

  const flush = async () => {
    if (batch.length === 0) return;
    const values = batch.map((_, i) => `($1, $${i * 2 + 2}, $${i * 2 + 3})`).join(',');
    const params = [source, ...batch.flatMap(b => [String(b.id), b.popularity ?? null])];
    const sql = `
      INSERT INTO asim_ingest_stubs (source, source_id, popularity)
      VALUES ${values}
      ON CONFLICT (source, source_id) DO UPDATE
        SET popularity = COALESCE(EXCLUDED.popularity, asim_ingest_stubs.popularity)
    `;
    await pool.query(sql, params);
    inserted += batch.length;
    batch = [];
  };

  for await (const line of rl) {
    if (!line) continue;
    try {
      const obj = JSON.parse(line);
      if (typeof obj.id !== 'number') continue;
      if (obj.adult) { continue; } // skip adult by default
      batch.push({ id: obj.id, popularity: typeof obj.popularity === 'number' ? obj.popularity : null });
      total++;
      if (batch.length >= BATCH_SIZE) await flush();
      if (total % 50000 === 0) process.stdout.write(`  ${source}: ${total.toLocaleString()} parsed, ${inserted.toLocaleString()} inserted\n`);
    } catch (e) {
      // Skip malformed lines silently — TMDB files occasionally have trailing junk
    }
  }
  await flush();
  return { total, inserted };
}

async function main() {
  const dateStr = exportDateStr();
  console.log(`[tmdb-ids] using export date ${dateStr}`);

  // Open a run-row to track progress in PG.
  const runR = await pool.query(
    `INSERT INTO asim_ingest_runs (run_kind, source, notes) VALUES ('tmdb-id-export-download', $1, $2) RETURNING id`,
    ['all', `dateStr=${dateStr}`]
  );
  const runId = runR.rows[0].id;
  let grandTotal = 0;

  try {
    for (const { source, file } of EXPORTS) {
      console.log(`\n[${source}] downloading ${file}_${dateStr}.json.gz`);
      const { dest, cached, bytes } = await downloadOne(file, dateStr);
      console.log(`  ${cached ? 'cached' : 'fetched'} ${(bytes/1024/1024).toFixed(1)}MB → ${dest}`);
      const { total, inserted } = await bulkLoadStubs(source, dest, runId);
      console.log(`  ${source}: ${total.toLocaleString()} records, ${inserted.toLocaleString()} stubs upserted`);
      grandTotal += inserted;
    }
    await pool.query(`UPDATE asim_ingest_runs SET finished_at=NOW(), status='done', records_out=$1 WHERE id=$2`, [grandTotal, runId]);
    console.log(`\n[tmdb-ids] DONE — ${grandTotal.toLocaleString()} stubs across ${EXPORTS.length} sources`);
  } catch (e) {
    console.error(`[tmdb-ids] FAILED:`, e.message);
    await pool.query(`UPDATE asim_ingest_runs SET finished_at=NOW(), status='failed', notes=$1 WHERE id=$2`, [String(e.message).slice(0, 800), runId]);
    process.exitCode = 1;
  } finally {
    await pool.end();
  }
}

main();