`agents/datasource-agent/skills/rate-sources.js`

264 lines
/**
 * Rate Sources Skill
 *
 * Rates unrated items across all 3 source tables (discovered_apis,
 * newspaper_frontpages, nonprofit_statements) using Gemini 2.0 Flash
 * to score relevance against current pulse_topics.
 */

const fetch = require('node-fetch');
const { logAction } = require('../../shared/audit-logger');

const AGENT = 'datasource-agent';
const PLATFORM = 'rating';

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

const DB_URL = process.env.DATABASE_URL || 'postgresql://dw_admin@127.0.0.1:5432/sdcc';

let _pool;
function getPool() {
  if (!_pool) {
    const { Pool } = require('pg');
    _pool = new Pool({ connectionString: DB_URL, max: 5 });
  }
  return _pool;
}

/**
 * Get current pulse topics for relevance scoring context.
 *
 * @returns {Promise<string[]>}
 */
async function getPulseTopics() {
  const pool = getPool();
  try {
    const { rows } = await pool.query(
      `SELECT name, category, urgency FROM pulse_topics ORDER BY created_at DESC LIMIT 20`
    );
    return rows.map((r) => `${r.name} (${r.category}, ${r.urgency})`);
  } catch (err) {
    console.error(`[${AGENT}] Error fetching pulse topics:`, err.message);
    // Fallback default topics
    return [
      'nonprofit advocacy',
      'civic engagement',
      'education policy',
      'economic inequality',
      'social impact',
      'consumer finance protection',
      'public benefits access',
    ];
  }
}

/**
 * Score an item with Gemini against current pulse topics.
 *
 * @param {string} itemName - Name/title of the item
 * @param {string} itemDescription - Description/content of the item
 * @param {string[]} pulseTopics - Current pulse topics
 * @returns {Promise<{ score: number, topics: string[] }>}
 */
async function geminiScoreItem(itemName, itemDescription, pulseTopics) {
  try {
    const topicsList = pulseTopics.join(', ');
    const prompt = `Rate the following item's relevance (0-100) to these active topics: ${topicsList}

Item: ${itemName}
Details: ${(itemDescription || 'No details available').substring(0, 500)}

Respond in EXACTLY this JSON format (no markdown, no code fences):
{"score": 50, "topics": ["matched topic 1", "matched topic 2"]}`;

    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: 300 },
      }),
    });

    const data = await res.json();
    const text = data?.candidates?.[0]?.content?.parts?.[0]?.text || '';

    const jsonMatch = text.match(/\{[\s\S]*\}/);
    if (jsonMatch) {
      const parsed = JSON.parse(jsonMatch[0]);
      return {
        score: Math.min(100, Math.max(0, Number(parsed.score) || 0)),
        topics: Array.isArray(parsed.topics) ? parsed.topics : [],
      };
    }
    return { score: 0, topics: [] };
  } catch (err) {
    console.error(`[${AGENT}] Gemini scoring error for "${itemName}":`, err.message);
    return { score: 0, topics: [] };
  }
}

/**
 * Rate unrated discovered APIs.
 *
 * @param {string[]} pulseTopics - Current pulse topics
 * @returns {Promise<number>} Number rated
 */
async function rateApis(pulseTopics) {
  const pool = getPool();
  let rated = 0;

  try {
    const { rows } = await pool.query(
      `SELECT id, name, description FROM discovered_apis
       WHERE relevance_score = 0
       ORDER BY created_at DESC
       LIMIT 20`
    );

    for (const row of rows) {
      const result = await geminiScoreItem(row.name, row.description, pulseTopics);
      await pool.query(
        `UPDATE discovered_apis SET relevance_score = $1, match_topics = $2, status = $3 WHERE id = $4`,
        [result.score, result.topics, result.score >= 50 ? 'relevant' : 'discovered', row.id]
      );
      rated++;
      await new Promise((r) => setTimeout(r, 300));
    }
  } catch (err) {
    console.error(`[${AGENT}] Error rating APIs:`, err.message);
  }

  return rated;
}

/**
 * Rate unrated newspaper frontpages from today.
 *
 * @param {string[]} pulseTopics - Current pulse topics
 * @returns {Promise<number>} Number rated
 */
async function rateFrontpages(pulseTopics) {
  const pool = getPool();
  let rated = 0;

  try {
    const { rows } = await pool.query(
      `SELECT id, newspaper_name, top_headline, headlines FROM newspaper_frontpages
       WHERE relevance_score = 0 AND scraped_date = CURRENT_DATE
       ORDER BY created_at DESC
       LIMIT 20`
    );

    for (const row of rows) {
      // Build description from top headline and first few headlines
      let desc = row.top_headline || '';
      try {
        const headlines = typeof row.headlines === 'string' ? JSON.parse(row.headlines) : row.headlines;
        if (Array.isArray(headlines)) {
          desc += ' | Headlines: ' + headlines.slice(0, 5).map((h) => h.title || '').join('; ');
        }
      } catch (e) { /* ignore */ }

      const result = await geminiScoreItem(row.newspaper_name, desc, pulseTopics);
      await pool.query(
        `UPDATE newspaper_frontpages SET relevance_score = $1, match_topics = $2 WHERE id = $3`,
        [result.score, result.topics, row.id]
      );
      rated++;
      await new Promise((r) => setTimeout(r, 300));
    }
  } catch (err) {
    console.error(`[${AGENT}] Error rating frontpages:`, err.message);
  }

  return rated;
}

/**
 * Rate unrated nonprofit statements.
 *
 * @param {string[]} pulseTopics - Current pulse topics
 * @returns {Promise<number>} Number rated
 */
async function rateStatements(pulseTopics) {
  const pool = getPool();
  let rated = 0;

  try {
    const { rows } = await pool.query(
      `SELECT id, org_name, statement_title, statement_body FROM nonprofit_statements
       WHERE relevance_score = 0
       ORDER BY created_at DESC
       LIMIT 20`
    );

    for (const row of rows) {
      const desc = (row.statement_body || row.statement_title || '').substring(0, 500);
      const result = await geminiScoreItem(
        `${row.org_name}: ${row.statement_title}`,
        desc,
        pulseTopics
      );
      await pool.query(
        `UPDATE nonprofit_statements SET relevance_score = $1, match_topics = $2 WHERE id = $3`,
        [result.score, result.topics, row.id]
      );
      rated++;
      await new Promise((r) => setTimeout(r, 300));
    }
  } catch (err) {
    console.error(`[${AGENT}] Error rating statements:`, err.message);
  }

  return rated;
}

/**
 * Main rate-sources skill handler.
 *
 * @param {Object} body - Request body (unused)
 * @returns {Promise<Object>} Rating results
 */
module.exports = async function rateSources(body = {}) {
  console.log(`[${AGENT}] Rating unscored sources...`);

  // Get current pulse topics for context
  const pulseTopics = await getPulseTopics();
  console.log(`[${AGENT}] Using ${pulseTopics.length} pulse topics for scoring`);

  // Rate all three source types
  const apiCount = await rateApis(pulseTopics);
  const frontpageCount = await rateFrontpages(pulseTopics);
  const statementCount = await rateStatements(pulseTopics);

  const total = apiCount + frontpageCount + statementCount;

  await logAction({
    agent: AGENT,
    actionType: 'rate',
    platform: PLATFORM,
    content: `Rated ${total} items across 3 source types`,
    responseData: {
      apis: apiCount,
      frontpages: frontpageCount,
      statements: statementCount,
      total,
      pulse_topics_used: pulseTopics.length,
    },
    status: 'success',
  });

  return {
    rated: {
      apis: apiCount,
      frontpages: frontpageCount,
      statements: statementCount,
    },
    total_rated: total,
    pulse_topics_count: pulseTopics.length,
    rated_at: new Date().toISOString(),
  };
};