`scrapers/import-thesetdecorator-roster.js`

218 lines
#!/usr/bin/env node
'use strict';
// Tick 2 — import thesetdecorator/data/decorator-roster.jsonl into
// asim_people (and asim_credits via stub movies in asim_movies). Each row
// is an IMDb person record enriched with credits + (sometimes) headshot,
// agency, public_email, etc. We treat thesetdecorator as a trusted source
// (Steve curates it) and tag rows with union_memberships=['iatse-44']
// when the primary_profession includes 'set_decorator'.
//
// IMDb's data here is FACTUAL (names, tconst, year). Steve's standing rule
// is no IMDb VISUAL assets — we never copy IMDb images. Headshots from this
// file came from thesetdecorator's own ingest (typically Wikimedia or
// claim-uploaded). We default headshot_source='thesetdecorator' so the
// provenance is clear.

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

const ROSTER = path.join(process.env.HOME, 'Projects', 'thesetdecorator', 'data', 'decorator-roster.jsonl');
const pool = new Pool({
  connectionString: process.env.DATABASE_URL || 'postgresql:///dw_unified?host=/tmp&user=stevestudio2',
  max: 6,
});

function professionRoles(arr) {
  // Map thesetdecorator's bucket labels to our role taxonomy + union guess.
  // (1:1 where possible; some labels split into multiple roles.)
  const map = {
    set_decorator: { role: 'set_decorator', dept: 'Art', union: 'iatse-44' },
    art_department: { role: 'art_department', dept: 'Art', union: 'iatse-44' },
    art_director: { role: 'art_director', dept: 'Art', union: 'iatse-800' },
    production_designer: { role: 'production_designer', dept: 'Art', union: 'iatse-800' },
    production_design: { role: 'production_designer', dept: 'Art', union: 'iatse-800' },
    propmaster: { role: 'propmaster', dept: 'Art', union: 'iatse-44' },
    actor: { role: 'actor', dept: 'Acting', union: 'sag-aftra' },
    actress: { role: 'actor', dept: 'Acting', union: 'sag-aftra' },
    director: { role: 'director', dept: 'Directing', union: 'dga' },
    writer: { role: 'writer', dept: 'Writing', union: 'wga' },
    producer: { role: 'producer', dept: 'Production', union: 'pga' },
    cinematographer: { role: 'cinematographer', dept: 'Camera', union: 'iatse-600' },
    editor: { role: 'editor', dept: 'Editing', union: 'iatse-700' },
    composer: { role: 'composer', dept: 'Sound', union: null },
    costume_designer: { role: 'costume_designer', dept: 'Costume & Make-Up', union: 'iatse-892' },
  };
  const roles = new Set();
  const unions = new Set();
  for (const p of (arr || [])) {
    const m = map[p];
    if (m) { roles.add(m.role); if (m.union) unions.add(m.union); }
  }
  return { roles: Array.from(roles), unions: Array.from(unions) };
}

async function upsertPerson(row) {
  const prof = professionRoles(row.primary_profession);
  const r = await pool.query(`
    INSERT INTO asim_people (
      imdb_id, full_name, birth_year, death_year,
      primary_profession, union_memberships,
      bio, agent_name, headshot_source, headshot_url,
      claimed_by_email, claimed_at,
      updated_at
    ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,NOW())
    ON CONFLICT (imdb_id) DO UPDATE SET
      full_name = EXCLUDED.full_name,
      birth_year = COALESCE(EXCLUDED.birth_year, asim_people.birth_year),
      death_year = COALESCE(EXCLUDED.death_year, asim_people.death_year),
      primary_profession = (
        SELECT array_agg(DISTINCT x) FROM unnest(asim_people.primary_profession || EXCLUDED.primary_profession) x WHERE x IS NOT NULL
      ),
      union_memberships = (
        SELECT array_agg(DISTINCT x) FROM unnest(COALESCE(asim_people.union_memberships,'{}') || EXCLUDED.union_memberships) x WHERE x IS NOT NULL
      ),
      bio = COALESCE(asim_people.bio, EXCLUDED.bio),
      agent_name = COALESCE(asim_people.agent_name, EXCLUDED.agent_name),
      headshot_url = COALESCE(asim_people.headshot_url, EXCLUDED.headshot_url),
      headshot_source = CASE WHEN asim_people.headshot_url IS NULL AND EXCLUDED.headshot_url IS NOT NULL
                             THEN EXCLUDED.headshot_source ELSE asim_people.headshot_source END,
      claimed_by_email = COALESCE(asim_people.claimed_by_email, EXCLUDED.claimed_by_email),
      claimed_at = COALESCE(asim_people.claimed_at, EXCLUDED.claimed_at),
      updated_at = NOW()
    RETURNING id
  `, [
    row.imdb_id || null,
    row.full_name,
    row.birth_year || null,
    row.death_year || null,
    prof.roles,
    prof.unions,
    row.bio || null,
    row.agency || null,
    row.headshot_url ? 'thesetdecorator' : 'made-with-ai',
    row.headshot_url || null,
    row.claimed_by || null,
    row.claimed_at ? new Date(row.claimed_at) : null,
  ]);
  const personId = r.rows[0].id;

  // Cross-ref the source so we know where this row came from.
  await pool.query(`
    INSERT INTO asim_people_sources (person_id, source, source_record_id, source_url)
    VALUES ($1, 'thesetdecorator-roster', $2, $3)
    ON CONFLICT (source, source_record_id) DO NOTHING
  `, [personId, row.id || row.imdb_id, row.imdb_url || null]);

  return personId;
}

async function upsertMovieStub(credit) {
  // No TMDB id yet — we stub by imdb_id (tconst) so credits can attach.
  // When tmdb-fetch-worker later hits this movie via Phase B, it merges in
  // the rich TMDB fields and the imdb_id stays as the join key for legacy
  // thesetdecorator rows.
  if (!credit.tconst) return null;
  const r = await pool.query(`
    INSERT INTO asim_movies (imdb_id, title, release_year, status, poster_source)
    VALUES ($1, $2, $3, 'imdb-stub', 'made-with-ai')
    ON CONFLICT (imdb_id) DO UPDATE SET
      title = COALESCE(asim_movies.title, EXCLUDED.title),
      release_year = COALESCE(asim_movies.release_year, EXCLUDED.release_year),
      updated_at = NOW()
    RETURNING id
  `, [credit.tconst, credit.title || credit.tconst, credit.year || null]).catch(async (e) => {
    // imdb_id doesn't have a UNIQUE constraint — fall back to SELECT-or-INSERT
    const sel = await pool.query(`SELECT id FROM asim_movies WHERE imdb_id=$1 LIMIT 1`, [credit.tconst]);
    if (sel.rows[0]) return sel;
    return pool.query(`INSERT INTO asim_movies (imdb_id, title, release_year, status, poster_source)
                       VALUES ($1,$2,$3,'imdb-stub','made-with-ai') RETURNING id`, [credit.tconst, credit.title || credit.tconst, credit.year || null]);
  });
  return r.rows[0].id;
}

async function upsertCredit(personId, movieId, credit) {
  const roleMap = {
    set_decorator: 'set_decorator',
    set_dec: 'set_decorator',
    production_designer: 'production_designer',
    art_director: 'art_director',
    propmaster: 'propmaster',
    actor: 'actor', actress: 'actor',
    director: 'director',
    writer: 'writer',
    cinematographer: 'cinematographer',
    editor: 'editor',
  };
  const role = roleMap[credit.role] || credit.role || 'unknown';
  await pool.query(`
    INSERT INTO asim_credits (person_id, movie_id, role, job_title, department, source, source_record_id)
    VALUES ($1, $2, $3, $4, $5, 'thesetdecorator-roster', $6)
    ON CONFLICT DO NOTHING
  `, [personId, movieId, role, credit.job || null, role === 'set_decorator' ? 'Art' : null, credit.tconst]);
}

// imdb_id on asim_movies needs UNIQUE for the upsert to work cleanly.
async function ensureMovieImdbUnique() {
  try {
    await pool.query(`CREATE UNIQUE INDEX IF NOT EXISTS uq_asim_movies_imdb ON asim_movies(imdb_id) WHERE imdb_id IS NOT NULL`);
  } catch (e) { /* ignore */ }
}

async function main() {
  if (!fs.existsSync(ROSTER)) {
    console.error('[roster] not found:', ROSTER);
    process.exit(1);
  }
  await ensureMovieImdbUnique();
  const rs = await pool.query(`INSERT INTO asim_ingest_runs (run_kind, source, notes) VALUES ('thesetdecorator-roster-import','thesetdecorator-roster',$1) RETURNING id`, [`file=${ROSTER}`]);
  const runId = rs.rows[0].id;

  const rl = readline.createInterface({
    input: fs.createReadStream(ROSTER),
    crlfDelay: Infinity,
  });

  let nPeople = 0;
  let nMovies = 0;
  let nCredits = 0;
  let nErrors = 0;
  const startedAt = Date.now();

  for await (const line of rl) {
    if (!line.trim()) continue;
    try {
      const row = JSON.parse(line);
      if (!row.imdb_id || !row.full_name) { nErrors++; continue; }
      const personId = await upsertPerson(row);
      nPeople++;
      for (const credit of (row.credits || [])) {
        const movieId = await upsertMovieStub(credit);
        if (movieId) {
          await upsertCredit(personId, movieId, credit);
          nCredits++;
        }
      }
      nMovies += (row.credits || []).length;
      if (nPeople % 500 === 0) {
        const elapsed = (Date.now() - startedAt) / 1000;
        const rate = (nPeople / elapsed).toFixed(1);
        console.log(`  ${nPeople} people · ${nCredits} credits · ${rate}/s`);
      }
    } catch (e) {
      nErrors++;
      if (nErrors <= 5) console.error('[parse]', e.message);
    }
  }

  await pool.query(`UPDATE asim_ingest_runs SET finished_at=NOW(), status='done', records_in=$1, records_out=$2, notes=$3 WHERE id=$4`,
    [nPeople, nCredits, `people=${nPeople} credits=${nCredits} errors=${nErrors}`, runId]);

  console.log(`\n[roster] DONE — ${nPeople.toLocaleString()} people · ${nCredits.toLocaleString()} credits · ${nErrors} errors`);
  await pool.end();
}

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