`agents/onboard-agent/skills/discover.js`

287 lines
/**
 * OnBoard Discover Skill
 *
 * Finds non-profit organizations matching trending pulse topics by:
 * 1. Querying pulse_topics for critical/high urgency items
 * 2. Searching ProPublica Nonprofit Explorer for each topic
 * 3. Cross-referencing against existing organizations to avoid duplicates
 * 4. Scoring each match: issue_overlap 40% + geo_overlap 30% + AI_assessment 30%
 * 5. Inserting new discoveries into onboard_discoveries
 */

const { query } = require('../../shared/db');
const { logAction } = require('../../shared/audit-logger');
const { searchNonprofits } = require('../lib/propublica');
const { getIssueAreas, issueOverlapScore } = require('../lib/ntee-matcher');
const fetch = require('node-fetch');

const AGENT = 'onboard-agent';
const PLATFORM = 'propublica';

const GEMINI_API_KEY = '${GOOGLE_API_KEY}';
const GEMINI_URL = `https://generativelanguage.googleapis.com/v1beta/models/gemini-2.0-flash:generateContent?key=${GEMINI_API_KEY}`;

/**
 * Call Gemini 2.0 Flash for AI scoring/assessment.
 *
 * @param {string} prompt - Prompt text
 * @returns {Promise<string>} Response text
 */
async function geminiScore(prompt) {
  try {
    const res = await fetch(GEMINI_URL, {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({
        contents: [{ parts: [{ text: prompt }] }],
        generationConfig: { temperature: 0.2, maxOutputTokens: 500 },
      }),
    });

    const data = await res.json();
    return data?.candidates?.[0]?.content?.parts?.[0]?.text || '';
  } catch (err) {
    console.error(`[${AGENT}] Gemini score error:`, err.message);
    return '';
  }
}

/**
 * Extract keywords from a pulse topic name and description.
 *
 * @param {Object} topic - Pulse topic row
 * @returns {string[]} Array of search keywords
 */
function extractKeywords(topic) {
  const text = `${topic.name || ''} ${topic.description || ''}`.toLowerCase();
  // Remove common stop words, split on whitespace and punctuation
  const stopWords = new Set(['the', 'a', 'an', 'is', 'are', 'was', 'were', 'be', 'been',
    'being', 'have', 'has', 'had', 'do', 'does', 'did', 'will', 'would', 'could',
    'should', 'may', 'might', 'can', 'shall', 'to', 'of', 'in', 'for', 'on', 'with',
    'at', 'by', 'from', 'as', 'into', 'through', 'during', 'before', 'after',
    'and', 'but', 'or', 'nor', 'not', 'so', 'yet', 'both', 'either', 'neither',
    'this', 'that', 'these', 'those', 'it', 'its', 'they', 'them', 'their',
    'about', 'up', 'out', 'than', 'also', 'just', 'more', 'very']);

  const words = text.replace(/[^a-z0-9\s]/g, ' ').split(/\s+/).filter(w => w.length > 2 && !stopWords.has(w));
  // Return unique keywords, max 5
  return [...new Set(words)].slice(0, 5);
}

/**
 * Check if an organization already exists in the DB by name similarity.
 *
 * @param {string} orgName - Organization name from ProPublica
 * @param {string|null} ein - EIN if available
 * @returns {Promise<Object|null>} Existing org row or null
 */
async function findExistingOrg(orgName, ein) {
  // First try exact EIN match
  if (ein) {
    const { rows } = await query(
      `SELECT id, name, ein FROM organizations WHERE ein = $1 LIMIT 1`,
      [String(ein)]
    );
    if (rows.length > 0) return rows[0];
  }

  // Then try text search on name
  const { rows } = await query(
    `SELECT id, name, ein FROM organizations
     WHERE to_tsvector('english', name) @@ plainto_tsquery('english', $1)
     LIMIT 5`,
    [orgName]
  );

  // Check for close match
  const nameLower = orgName.toLowerCase();
  for (const row of rows) {
    const rowLower = row.name.toLowerCase();
    if (rowLower === nameLower || rowLower.includes(nameLower) || nameLower.includes(rowLower)) {
      return row;
    }
  }

  return null;
}

/**
 * @param {Object} body - Request body
 * @param {number} [body.limit=10] - Max topics to process
 * @param {boolean} [body.skip_ai=false] - Skip AI assessment (faster, for testing)
 * @returns {Promise<Object>}
 */
module.exports = async function discover(body = {}) {
  const topicLimit = body.limit || 10;
  const skipAi = body.skip_ai || false;

  // 1. Query pulse_topics for critical/high urgency
  const { rows: topics } = await query(
    `SELECT id, name, category, urgency, description, created_at
     FROM pulse_topics
     WHERE urgency IN ('critical', 'high')
     ORDER BY created_at DESC
     LIMIT $1`,
    [topicLimit]
  );

  if (topics.length === 0) {
    await logAction({
      agent: AGENT,
      actionType: 'discover',
      platform: PLATFORM,
      status: 'skipped',
      content: 'No critical/high urgency topics found',
    });
    return { topics_searched: 0, orgs_found: 0, new_discoveries: 0 };
  }

  let totalOrgsFound = 0;
  let totalNewDiscoveries = 0;

  // 2. For each topic, search ProPublica
  for (const topic of topics) {
    const keywords = extractKeywords(topic);
    if (keywords.length === 0) continue;

    const searchQuery = keywords.join(' ');
    const result = await searchNonprofits(searchQuery);
    const orgs = result.organizations || [];

    totalOrgsFound += orgs.length;

    // 3. Process each org result
    for (const org of orgs.slice(0, 10)) { // Limit to 10 orgs per topic
      // Cross-reference with existing organizations
      const existing = await findExistingOrg(org.name, org.ein);

      // Get issue areas from NTEE code
      const orgIssues = getIssueAreas(org.ntee_code);

      // Score: issue_overlap 40%
      const issueScore = issueOverlapScore(orgIssues, keywords) * 40;

      // Score: geo_overlap 30% (check if topic description mentions state)
      const topicText = `${topic.name} ${topic.description || ''}`.toLowerCase();
      const geoScore = org.state && topicText.includes((org.state || '').toLowerCase()) ? 30 : 0;

      // Score: AI_assessment 30%
      let aiScore = 15; // Default mid-range
      if (!skipAi) {
        try {
          const aiPrompt = `Score how well this nonprofit matches this issue topic on a scale of 0-100.
Topic: "${topic.name}" - ${topic.description || 'no description'}
Nonprofit: "${org.name}" in ${org.city || '?'}, ${org.state || '?'} (NTEE: ${org.ntee_code || 'unknown'})
Revenue: $${(org.total_revenue || 0).toLocaleString()}

Reply with ONLY a number 0-100, nothing else.`;

          const aiResult = await geminiScore(aiPrompt);
          const parsed = parseInt(aiResult.trim(), 10);
          if (!isNaN(parsed) && parsed >= 0 && parsed <= 100) {
            aiScore = (parsed / 100) * 30;
          }
        } catch (err) {
          console.error(`[${AGENT}] AI score error for ${org.name}:`, err.message);
        }
      }

      const matchScore = Math.round((issueScore + geoScore + aiScore) * 100) / 100;

      // Skip low-score matches
      if (matchScore < 10) continue;

      // 4. Insert into organizations if new
      let orgId;
      if (existing) {
        orgId = existing.id;
      } else {
        try {
          const { rows: insertedOrg } = await query(
            `INSERT INTO organizations (name, ein, type, ntee_code, city, state,
               annual_revenue, total_assets, source, source_id)
             VALUES ($1, $2, 'nonprofit', $3, $4, $5, $6, $7, 'onboard-agent', $8)
             ON CONFLICT (ein) DO UPDATE SET
               updated_at = NOW(),
               source = CASE WHEN organizations.source = 'manual' THEN organizations.source ELSE 'onboard-agent' END
             RETURNING id`,
            [
              org.name,
              org.ein ? String(org.ein) : null,
              org.ntee_code || null,
              org.city || null,
              org.state || null,
              org.total_revenue || null,
              org.total_assets || null,
              org.ein ? `propublica-${org.ein}` : null,
            ]
          );
          orgId = insertedOrg[0]?.id;
        } catch (err) {
          console.error(`[${AGENT}] Insert org error for ${org.name}:`, err.message);
          continue;
        }
      }

      if (!orgId) continue;

      // 5. Insert discovery
      try {
        // Check if discovery already exists for this org + topic
        const { rows: existingDisc } = await query(
          `SELECT id FROM onboard_discoveries
           WHERE org_id = $1 AND trigger_topic_id = $2
           LIMIT 1`,
          [orgId, topic.id]
        );

        if (existingDisc.length > 0) continue; // Skip duplicate

        await query(
          `INSERT INTO onboard_discoveries
            (org_id, trigger_topic_id, trigger_type, trigger_label, match_score, match_details, status)
           VALUES ($1, $2, 'pulse_topic', $3, $4, $5, 'discovered')`,
          [
            orgId,
            topic.id,
            topic.name,
            matchScore,
            JSON.stringify({
              issue_score: Math.round(issueScore * 100) / 100,
              geo_score: geoScore,
              ai_score: Math.round(aiScore * 100) / 100,
              org_issues: orgIssues,
              search_query: searchQuery,
              propublica_ein: org.ein,
            }),
          ]
        );

        totalNewDiscoveries++;
      } catch (err) {
        console.error(`[${AGENT}] Insert discovery error:`, err.message);
      }
    }
  }

  // Log the action
  await logAction({
    agent: AGENT,
    actionType: 'discover',
    platform: PLATFORM,
    content: `Searched ${topics.length} topics`,
    responseData: {
      topics_searched: topics.length,
      orgs_found: totalOrgsFound,
      new_discoveries: totalNewDiscoveries,
    },
    status: 'success',
  });

  return {
    topics_searched: topics.length,
    orgs_found: totalOrgsFound,
    new_discoveries: totalNewDiscoveries,
  };
};