`scrapers/tmdb-fetch-worker.js`

238 lines
#!/usr/bin/env node
'use strict';
// Phase B — long-running worker that drains asim_ingest_stubs by hitting the
// TMDB detail endpoints and writing into asim_movies / asim_people /
// asim_credits. Picks the next highest-popularity queued stub each iteration.
//
// Rate limit: 30 req/sec safe under TMDB's 50/sec cap (leaves headroom for
// other consumers + retries). Adaptive backoff on 429.
//
// Run modes:
//   node tmdb-fetch-worker.js              # forever (until killed)
//   node tmdb-fetch-worker.js --batch=500  # process N stubs then exit
//   node tmdb-fetch-worker.js --source=tmdb-movie  # only one source
//
// Requires TMDB_API_KEY in .env. Without it the worker fast-exits with a
// log line so the YOLO loop can surface to Steve.

require('dotenv').config();
const { Pool } = require('pg');

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

const args = process.argv.slice(2).reduce((m, a) => {
  const [k, v] = a.replace(/^--/, '').split('=');
  m[k] = v ?? true;
  return m;
}, {});
const BATCH = args.batch ? parseInt(args.batch, 10) : Infinity;
const ONLY_SOURCE = args.source || null;
const REQ_PER_SEC = parseInt(args.rps || '30', 10);
const MIN_GAP_MS = Math.ceil(1000 / REQ_PER_SEC);
let lastReqAt = 0;
let fetched = 0;
let upserted = 0;
let failed = 0;

async function rateGate() {
  const now = Date.now();
  const wait = MIN_GAP_MS - (now - lastReqAt);
  if (wait > 0) await new Promise(r => setTimeout(r, wait));
  lastReqAt = Date.now();
}

async function tmdbGet(pathPart) {
  await rateGate();
  const sep = pathPart.includes('?') ? '&' : '?';
  const url = `https://api.themoviedb.org/3${pathPart}${sep}api_key=${API_KEY}`;
  const r = await fetch(url);
  if (r.status === 429) {
    const retry = parseInt(r.headers.get('retry-after') || '2', 10) * 1000;
    console.log(`[429] backing off ${retry}ms`);
    await new Promise(res => setTimeout(res, retry));
    return tmdbGet(pathPart);
  }
  if (r.status === 404) return null;
  if (!r.ok) throw new Error(`TMDB ${r.status}: ${pathPart}`);
  return r.json();
}

async function upsertMovie(d) {
  if (!d || !d.id) return false;
  const release_year = d.release_date ? parseInt(d.release_date.slice(0, 4), 10) : null;
  const posterUrl = d.poster_path ? `https://image.tmdb.org/t/p/w500${d.poster_path}` : null;
  const backdropUrl = d.backdrop_path ? `https://image.tmdb.org/t/p/w1280${d.backdrop_path}` : null;
  await POOL.query(`
    INSERT INTO asim_movies (
      tmdb_id, imdb_id, title, original_title, release_year, release_date,
      overview, tagline, runtime_min, genres, original_language, homepage,
      status, budget, revenue, vote_average, vote_count, popularity,
      collection_id, collection_name,
      poster_source, poster_url, backdrop_source, backdrop_url, updated_at
    ) VALUES (
      $1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,
      $21,$22,$23,$24,NOW()
    )
    ON CONFLICT (tmdb_id) DO UPDATE SET
      imdb_id = COALESCE(EXCLUDED.imdb_id, asim_movies.imdb_id),
      title = EXCLUDED.title,
      original_title = EXCLUDED.original_title,
      release_year = EXCLUDED.release_year,
      release_date = EXCLUDED.release_date,
      overview = EXCLUDED.overview,
      tagline = EXCLUDED.tagline,
      runtime_min = EXCLUDED.runtime_min,
      genres = EXCLUDED.genres,
      original_language = EXCLUDED.original_language,
      homepage = EXCLUDED.homepage,
      status = EXCLUDED.status,
      budget = EXCLUDED.budget,
      revenue = EXCLUDED.revenue,
      vote_average = EXCLUDED.vote_average,
      vote_count = EXCLUDED.vote_count,
      popularity = EXCLUDED.popularity,
      collection_id = EXCLUDED.collection_id,
      collection_name = EXCLUDED.collection_name,
      poster_url = EXCLUDED.poster_url,
      backdrop_url = EXCLUDED.backdrop_url,
      updated_at = NOW()
  `, [
    d.id, d.imdb_id || null, d.title, d.original_title || null, release_year,
    d.release_date || null, d.overview || null, d.tagline || null,
    d.runtime || null, (d.genres || []).map(g => g.name), d.original_language || null,
    d.homepage || null, d.status || null, d.budget || null, d.revenue || null,
    d.vote_average || null, d.vote_count || null, d.popularity || null,
    d.belongs_to_collection?.id || null, d.belongs_to_collection?.name || null,
    posterUrl ? 'tmdb-attribution' : 'made-with-ai', posterUrl,
    backdropUrl ? 'tmdb-attribution' : 'made-with-ai', backdropUrl,
  ]);
  return true;
}

async function upsertPerson(d) {
  if (!d || !d.id) return false;
  const birthYear = d.birthday ? parseInt(d.birthday.slice(0, 4), 10) : null;
  const deathYear = d.deathday ? parseInt(d.deathday.slice(0, 4), 10) : null;
  const headshot = d.profile_path ? `https://image.tmdb.org/t/p/w500${d.profile_path}` : null;
  await POOL.query(`
    INSERT INTO asim_people (
      tmdb_id, imdb_id, full_name, also_known_as,
      birth_year, birth_date, death_year, death_date, birth_place,
      primary_profession, bio,
      headshot_source, headshot_url, updated_at
    ) VALUES (
      $1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,NOW()
    )
    ON CONFLICT (tmdb_id) DO UPDATE SET
      imdb_id = COALESCE(EXCLUDED.imdb_id, asim_people.imdb_id),
      full_name = EXCLUDED.full_name,
      also_known_as = EXCLUDED.also_known_as,
      birth_year = EXCLUDED.birth_year,
      birth_date = EXCLUDED.birth_date,
      death_year = EXCLUDED.death_year,
      death_date = EXCLUDED.death_date,
      birth_place = EXCLUDED.birth_place,
      primary_profession = EXCLUDED.primary_profession,
      bio = EXCLUDED.bio,
      headshot_url = COALESCE(EXCLUDED.headshot_url, asim_people.headshot_url),
      updated_at = NOW()
  `, [
    d.id, d.imdb_id || null, d.name, d.also_known_as || null,
    birthYear, d.birthday || null, deathYear, d.deathday || null, d.place_of_birth || null,
    [d.known_for_department || null].filter(Boolean), d.biography || null,
    headshot ? 'tmdb-attribution' : 'made-with-ai', headshot,
  ]);
  return true;
}

// TMDB endpoints per source
const ENDPOINT = {
  'tmdb-movie':  (id) => `/movie/${id}?language=en-US&append_to_response=external_ids`,
  'tmdb-tv':     (id) => `/tv/${id}?language=en-US&append_to_response=external_ids`,
  'tmdb-person': (id) => `/person/${id}?language=en-US&append_to_response=external_ids`,
  'tmdb-collection': (id) => `/collection/${id}`,
  'tmdb-company':    (id) => `/company/${id}`,
  'tmdb-network':    (id) => `/network/${id}`,
  'tmdb-keyword':    (id) => `/keyword/${id}`,
};

async function processOne(stub) {
  const ep = ENDPOINT[stub.source];
  if (!ep) {
    await POOL.query(`UPDATE asim_ingest_stubs SET status='skipped', last_error='unknown source' WHERE id=$1`, [stub.id]);
    return;
  }
  try {
    const d = await tmdbGet(ep(stub.source_id));
    if (!d) {
      await POOL.query(`UPDATE asim_ingest_stubs SET status='gone', fetched_at=NOW() WHERE id=$1`, [stub.id]);
      return;
    }
    if (stub.source === 'tmdb-movie') await upsertMovie(d);
    else if (stub.source === 'tmdb-person') await upsertPerson(d);
    // TV / collection / company / network / keyword — schemas land in tick 4+;
    // for now mark them done so the queue advances.
    await POOL.query(`UPDATE asim_ingest_stubs SET status='done', fetched_at=NOW() WHERE id=$1`, [stub.id]);
    upserted++;
  } catch (e) {
    failed++;
    await POOL.query(
      `UPDATE asim_ingest_stubs SET status=CASE WHEN attempts >= 3 THEN 'failed' ELSE 'queued' END,
         attempts = attempts + 1, last_error=$2 WHERE id=$1`,
      [stub.id, String(e.message).slice(0, 400)]
    );
  }
  fetched++;
}

async function nextBatch(n) {
  const params = [n];
  let where = `WHERE status='queued'`;
  if (ONLY_SOURCE) { params.unshift(ONLY_SOURCE); where = `WHERE source=$1 AND status='queued'`; }
  const sql = `
    UPDATE asim_ingest_stubs SET status='in_flight'
      WHERE id IN (
        SELECT id FROM asim_ingest_stubs ${where}
          ORDER BY popularity DESC NULLS LAST, id ASC LIMIT $${params.length}
          FOR UPDATE SKIP LOCKED
      )
    RETURNING id, source, source_id, popularity
  `;
  const r = await POOL.query(sql, params);
  return r.rows;
}

async function main() {
  if (!API_KEY) {
    console.error('[tmdb-fetch] TMDB_API_KEY not set in .env — exiting.');
    console.error('  Register a free key at https://www.themoviedb.org/settings/api');
    console.error('  Then add to ~/Projects/asseeninmovies/.env and re-run.');
    process.exit(2);
  }
  const startedAt = Date.now();
  console.log(`[tmdb-fetch] starting · rps=${REQ_PER_SEC} · batch=${BATCH === Infinity ? '∞' : BATCH} · source=${ONLY_SOURCE || 'any'}`);
  while (fetched < BATCH) {
    const stubs = await nextBatch(Math.min(40, BATCH - fetched));
    if (stubs.length === 0) {
      console.log('[tmdb-fetch] queue empty — sleeping 60s then re-checking');
      await new Promise(r => setTimeout(r, 60000));
      continue;
    }
    // Run with bounded parallelism (rate-gate handles the actual throttle).
    await Promise.all(stubs.map(processOne));
    const elapsedSec = (Date.now() - startedAt) / 1000;
    const rate = (fetched / elapsedSec).toFixed(1);
    if (fetched % 200 === 0 || fetched < 50) {
      console.log(`[tmdb-fetch] ${fetched} fetched · ${upserted} upserted · ${failed} failed · ${rate}/s`);
    }
  }
  console.log(`[tmdb-fetch] batch done · ${fetched} fetched · ${upserted} upserted · ${failed} failed`);
  await POOL.end();
}

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