`scrapers/ca_dre_pm.mjs`

157 lines
#!/usr/bin/env node
/* ca_dre_pm.mjs — Property Management providers for the RENTV /services directory.
 *
 * CA property managers must hold a DRE broker license, so they live in the usre `firm`
 * registry (214k CA/AZ DRE firms) — we just have to find them. Feed-first + $0:
 *   1. keyword pre-filter usre.firm for PM-named firms in served SoCal/AZ metros,
 *   2. LOCAL Ollama (qwen2.5, $0) classifies each firm's SEGMENT (Commercial / Residential /
 *      HOA-Community / Mixed) and drops false matches (software/consulting-only),
 *   3. upsert the confirmed operators into realestate.rentv_licensed_targets as role
 *      'Property Manager' (local mirror; the /services page reads this DB live).
 *
 * HONESTY: public DRE license records = targeting only, not a send list. No phone (the firm
 * table has none); website only when the source provides it. source_url deep-links to the
 * official CA DRE public license lookup. Prod (Kamatera) load stays gated.
 *
 * Run:  node scrapers/ca_dre_pm.mjs           (dry-run: classify + report, no DB write)
 *       node scrapers/ca_dre_pm.mjs --apply    (upsert into local realestate DB)
 * Cost: $0 (local Ollama). */

import { createRequire } from 'node:module';
const pg = createRequire('/Users/macstudio3/Projects/rentv/')('pg'); // reuse rentv's pg install

const APPLY = process.argv.includes('--apply');
const OLLAMA = process.env.OLLAMA_URL || 'http://127.0.0.1:11434';
const MODEL = process.env.OLLAMA_MODEL || 'qwen2.5:latest';
const CONC = 3;

// Comprehensive SoCal + AZ city → RENTV market map (unmapped cities are EXCLUDED — we only
// claim the metros we serve; NorCal PM firms must not leak in as "SoCal (other)").
const CITY_MARKET = {};
const add = (m, cities) => cities.forEach(c => { CITY_MARKET[c] = m; });
add('Greater LA', ['los angeles','beverly hills','santa monica','century city','pasadena','glendale','el segundo','culver city','burbank','long beach','torrance','hollywood','west hollywood','sherman oaks','encino','van nuys','woodland hills','studio city','marina del rey','manhattan beach','redondo beach','hermosa beach','santa clarita','valencia','calabasas','tarzana','northridge','downey','whittier','arcadia','monrovia','el monte','pomona','west covina','carson','gardena','hawthorne','inglewood','san pedro','playa vista','brentwood','westlake village','agoura hills','malibu','claremont','glendora','covina','alhambra','monterey park','cerritos','lakewood','bellflower','norwalk','santa fe springs','city of industry','commerce','vernon','signal hill']);
add('Orange County', ['irvine','newport beach','costa mesa','anaheim','santa ana','orange','huntington beach','tustin','fullerton','laguna beach','laguna niguel','laguna hills','mission viejo','lake forest','aliso viejo','san clemente','dana point','yorba linda','brea','placentia','garden grove','westminster','fountain valley','buena park','cypress','rancho santa margarita','ladera ranch']);
add('San Diego', ['san diego','carlsbad','la jolla','del mar','escondido','chula vista','oceanside','vista','encinitas','solana beach','poway','rancho bernardo','national city','el cajon','la mesa','coronado','san marcos']);
add('Inland Empire', ['riverside','ontario','san bernardino','rancho cucamonga','corona','temecula','murrieta','fontana','moreno valley','redlands','chino','chino hills','upland','rialto','hemet','menifee','eastvale','jurupa valley','norco','lake elsinore','perris','colton','loma linda','yucaipa']);
add('Ventura', ['ventura','oxnard','thousand oaks','simi valley','camarillo','moorpark','newbury park','ojai','port hueneme','fillmore','santa paula']);

function marketOf(city, state) {
  const c = String(city || '').trim().toLowerCase();
  if (CITY_MARKET[c]) return CITY_MARKET[c];
  if (String(state || '').toUpperCase() === 'AZ') {
    if (/phoenix|scottsdale|tempe|mesa|chandler|gilbert|glendale|peoria|surprise|maricopa/.test(c)) return 'Arizona';
  }
  return null;
}

async function classify(name) {
  const prompt = `A California DRE-licensed real estate firm is named "${name}". Classify it as a property-management provider. Reply ONLY JSON {"is_pm": true|false, "segment": "Commercial"|"Residential"|"HOA/Community"|"Mixed", "descriptor": "<=8 word phrase"}. is_pm=false ONLY if the name clearly is NOT an operating property manager (e.g. software, pure consulting, a law firm). segment = the property type they manage; use "Mixed" if unclear. descriptor = a short professional label, e.g. "Commercial property & asset management".`;
  try {
    const r = await fetch(OLLAMA + '/api/generate', {
      method: 'POST', headers: { 'content-type': 'application/json' },
      body: JSON.stringify({ model: MODEL, prompt, stream: false, format: 'json', options: { temperature: 0.2, num_predict: 120 } }),
    });
    const j = await r.json();
    const o = JSON.parse(String(j.response || '{}'));
    return { is_pm: o.is_pm !== false, segment: ['Commercial', 'Residential', 'HOA/Community', 'Mixed'].includes(o.segment) ? o.segment : 'Mixed', descriptor: String(o.descriptor || '').slice(0, 60) };
  } catch (e) { return { is_pm: true, segment: 'Mixed', descriptor: 'Property management' }; } // fail-open to keep the firm, generic label
}

async function mapPool(items, fn, conc) {
  const out = new Array(items.length); let i = 0;
  await Promise.all(Array.from({ length: conc }, async () => {
    while (i < items.length) { const idx = i++; out[idx] = await fn(items[idx], idx); }
  }));
  return out;
}

// Fetch the REAL DRE license status + expiration from the public record (Cody must-fix: never
// claim a status we haven't verified). The constructed pplinfo URL resolves to the right firm.
async function verifyDre(licenseNo) {
  try {
    const r = await fetch('https://www2.dre.ca.gov/PublicASP/pplinfo.asp?License_id=' + encodeURIComponent(licenseNo),
      { headers: { 'User-Agent': 'Mozilla/5.0 (RENTV directory verifier)' }, signal: AbortSignal.timeout(15000) });
    if (!r.ok) return { status: null, exp: null };
    let t = await r.text();
    t = t.replace(/<[^>]+>/g, ' ').replace(/&nbsp;/g, ' ').replace(/\s+/g, ' ');
    const notFound = /no record|invalid|not found/i.test(t) && !/property|LICENSED|EXPIRED/i.test(t);
    if (notFound) return { status: 'Not found', exp: null };
    const sm = t.match(/License Status\s*:?\s*(LICENSED|EXPIRED|CANCEL[A-Z]*|SURRENDER[A-Z]*|REVOK[A-Z]*|SUSPEND[A-Z]*)/i);
    const em = t.match(/Expiration Date\s*:?\s*(\d{2}\/\d{2}\/\d{2,4})/);
    let status = sm ? sm[1].replace(/^LICENSED.*/i, 'Licensed') : null;
    if (status) status = status.charAt(0).toUpperCase() + status.slice(1).toLowerCase();
    // Expired-by-date guard even if the page says Licensed.
    if (em) { const p = em[1].split('/'); let yr = +p[2]; if (yr < 100) yr += 2000; const d = new Date(Date.UTC(yr, +p[0] - 1, +p[1])); if (!isNaN(d) && d.getTime() < Date.UTC(2026, 7, 6)) status = 'Expired'; }
    return { status, exp: em ? em[1] : null };
  } catch (e) { return { status: null, exp: null }; }
}

async function main() {
  const t0 = Date.now();

  // --verify: replace the placeholder status on already-loaded PM rows with the REAL DRE status.
  if (process.argv.includes('--verify')) {
    const re = new pg.Client({ host: '/tmp', database: 'realestate' }); await re.connect();
    const rows = (await re.query("SELECT license_no FROM rentv_licensed_targets WHERE source='ca_dre_pm'")).rows;
    console.log(`[ca_dre_pm --verify] verifying ${rows.length} PM firms against the live DRE record ($0)…`);
    const res = await mapPool(rows, async (r) => ({ license_no: r.license_no, ...(await verifyDre(r.license_no)) }), CONC);
    const tally = {}; let updated = 0;
    for (const x of res) {
      const st = x.status || 'Unverified'; tally[st] = (tally[st] || 0) + 1;
      await re.query('UPDATE rentv_licensed_targets SET license_status=$1, raw = coalesce(raw,\'{}\'::jsonb) || jsonb_build_object(\'dre_expiration\',$2::text,\'dre_verified\',true) WHERE source=$3 AND license_no=$4',
        [x.status, x.exp, 'ca_dre_pm', x.license_no]);
      updated++;
    }
    // Drop the non-current ones from the CRE directory (honest: only show verified-Licensed firms).
    const del = await re.query("DELETE FROM rentv_licensed_targets WHERE source='ca_dre_pm' AND (license_status IS NULL OR license_status NOT ILIKE 'Licensed')");
    await re.end();
    console.log('[ca_dre_pm --verify] status tally:', tally);
    console.log(`[ca_dre_pm --verify] ✓ updated ${updated}, removed ${del.rowCount} non-Licensed/unverified. ${((Date.now() - t0) / 1000).toFixed(1)}s · $0`);
    return;
  }
  const usre = new pg.Client({ host: '/tmp', database: 'usre' }); await usre.connect();
  const cand = (await usre.query(
    `SELECT name, hq_city, hq_state, license_no, website, agent_count FROM firm
     WHERE name ~* 'property manage|property mgmt|asset manage|commercial manage|realty manage|community manage|association manage'
       AND hq_state IN ('CA','AZ') AND license_no IS NOT NULL AND license_no <> ''`)).rows;
  await usre.end();

  // Market-filter first (only served metros), so we classify a bounded set.
  const targeted = cand.map(r => ({ ...r, market: marketOf(r.hq_city, r.hq_state) })).filter(r => r.market);
  console.log(`[ca_dre_pm] ${cand.length} PM-named DRE firms → ${targeted.length} in served metros. Classifying with LOCAL ${MODEL} ($0)…`);

  const classified = await mapPool(targeted, async (r) => ({ ...r, ...(await classify(r.name)) }), CONC);
  const kept = classified.filter(r => r.is_pm);
  const commercial = kept.filter(r => r.segment === 'Commercial' || r.segment === 'Mixed');

  const bySeg = {}; kept.forEach(r => { bySeg[r.segment] = (bySeg[r.segment] || 0) + 1; });
  const byMkt = {}; kept.forEach(r => { byMkt[r.market] = (byMkt[r.market] || 0) + 1; });
  console.log(`[ca_dre_pm] kept ${kept.length} PM firms (dropped ${classified.length - kept.length} non-PM). commercial/mixed: ${commercial.length}`);
  console.log('  by segment:', bySeg);
  console.log('  by market :', byMkt);
  console.log(`  ${((Date.now() - t0) / 1000).toFixed(1)}s · cost $0 (local)`);

  if (!APPLY) { console.log('\n(dry-run) sample:', kept.slice(0, 6).map(r => `${r.name} [${r.segment}] ${r.market}`)); console.log('re-run with --apply to upsert into local realestate DB.'); return; }

  // Load only commercial/mixed PM firms (CRE-relevant) into the local realestate mirror.
  const re = new pg.Client({ host: '/tmp', database: 'realestate' }); await re.connect();
  let n = 0;
  for (const r of commercial) {
    const url = 'https://www2.dre.ca.gov/PublicASP/pplinfo.asp?License_id=' + encodeURIComponent(r.license_no);
    await re.query(
      `INSERT INTO rentv_licensed_targets
        (source, role, entity_name, license_no, license_type, license_status, city, state, market, phone, website, commercial_flag, within_300mi, source_url, raw)
       VALUES ('ca_dre_pm','Property Manager',$1,$2,$3,'',$4,$5,$6,'',$7,true,true,$8,$9)
       ON CONFLICT (source, license_no) DO UPDATE SET
        entity_name=EXCLUDED.entity_name, license_type=EXCLUDED.license_type, city=EXCLUDED.city,
        market=EXCLUDED.market, website=EXCLUDED.website, source_url=EXCLUDED.source_url, raw=EXCLUDED.raw`,
      [r.name, r.license_no, r.descriptor || (r.segment + ' property management'), r.hq_city,
       (r.hq_state || 'CA'), r.market, r.website || '', url, JSON.stringify(r)]);
    n++;
  }
  await re.end();
  console.log(`[ca_dre_pm] ✓ upserted ${n} commercial/mixed PM firms into realestate.rentv_licensed_targets (role='Property Manager'). cost $0 (local).`);
}

main().catch(e => { console.error('[ca_dre_pm] FAILED:', e.message); process.exit(1); });