`agents/datasource-agent/skills/collect-statements.js`

320 lines
/**
 * Collect Statements Skill
 *
 * Collects press releases and statements from non-profit organizations.
 * Queries the organizations table for orgs with websites, attempts to fetch
 * /press, /news, /blog, /press-releases pages, and uses Gemini to extract
 * and analyze statement data.
 */

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

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

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';

const PRESS_PATHS = ['/press', '/news', '/blog', '/press-releases', '/media', '/newsroom', '/updates', '/statements'];

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

/**
 * Fetch a page with timeout and return text content.
 *
 * @param {string} url - URL to fetch
 * @returns {Promise<string|null>} HTML text or null
 */
async function fetchPage(url) {
  const controller = new AbortController();
  const timeout = setTimeout(() => controller.abort(), 12000);
  try {
    const res = await fetch(url, {
      signal: controller.signal,
      headers: {
        'User-Agent': 'Norma-DataSource-Agent/1.0 (research; non-commercial)',
        'Accept': 'text/html,application/xhtml+xml,*/*',
      },
      redirect: 'follow',
    });
    clearTimeout(timeout);
    if (!res.ok) return null;
    const text = await res.text();
    // Only return if it looks like HTML with some content
    if (text.length < 200) return null;
    return text;
  } catch (err) {
    clearTimeout(timeout);
    return null;
  }
}

/**
 * Extract article/post links from an HTML page using regex.
 * Looks for common patterns: anchor tags with /press/, /news/, /blog/ in href.
 *
 * @param {string} html - HTML content
 * @param {string} baseUrl - Base URL for resolving relative links
 * @returns {Array<{ title: string, url: string }>}
 */
function extractArticleLinks(html, baseUrl) {
  const links = [];
  const seen = new Set();

  // Match anchor tags with href
  const linkRegex = /<a\s[^>]*href=["']([^"'#]+)["'][^>]*>([\s\S]*?)<\/a>/gi;
  let match;

  while ((match = linkRegex.exec(html)) !== null) {
    let href = match[1].trim();
    let title = match[2].replace(/<[^>]*>/g, '').trim();

    // Skip empty, javascript, mailto, and anchor-only links
    if (!href || href.startsWith('javascript:') || href.startsWith('mailto:') || href === '/') continue;

    // Resolve relative URLs
    if (href.startsWith('/')) {
      try {
        const base = new URL(baseUrl);
        href = base.origin + href;
      } catch (e) {
        continue;
      }
    } else if (!href.startsWith('http')) {
      continue;
    }

    // Filter for likely article/press release links
    const lowerHref = href.toLowerCase();
    const isArticle =
      lowerHref.includes('/press') ||
      lowerHref.includes('/news') ||
      lowerHref.includes('/blog') ||
      lowerHref.includes('/statement') ||
      lowerHref.includes('/release') ||
      lowerHref.includes('/update') ||
      lowerHref.includes('/article') ||
      lowerHref.includes('/post/') ||
      /\/\d{4}\/\d{2}\//.test(lowerHref); // date-based URLs

    if (isArticle && title.length >= 10 && !seen.has(href)) {
      seen.add(href);
      links.push({ title: title.substring(0, 300), url: href });
    }
  }

  return links.slice(0, 20);
}

/**
 * Use Gemini to analyze extracted statements/articles.
 *
 * @param {string} orgName - Organization name
 * @param {Array<{ title: string, url: string }>} articles - Article links found
 * @returns {Promise<Array<{ title: string, summary: string, topics: string[], sentiment: string, url: string }>>}
 */
async function geminiAnalyzeStatements(orgName, articles) {
  if (!articles || articles.length === 0) return [];

  const articleList = articles.slice(0, 10).map((a, i) => (i + 1) + '. "' + a.title + '" - ' + a.url).join('\n');

  try {
    const prompt = 'You are analyzing press releases and statements from the non-profit organization "' + orgName + '".\n\nHere are the article titles and URLs found on their website:\n' + articleList + '\n\nFor each article that appears to be a substantive statement or press release (skip navigation links, category pages, etc.), extract:\n- title: the article title\n- summary: a 1-2 sentence summary based on the title\n- topics: array of relevant topic tags\n- sentiment: positive, negative, neutral, or mixed\n- url: the article URL\n\nFocus especially on articles related to: nonprofit advocacy, education policy, consumer protection, economic justice, civic engagement, social impact.\n\nRespond in EXACTLY this JSON format (no markdown, no code fences):\n{"statements": [{"title": "...", "summary": "...", "topics": ["..."], "sentiment": "neutral", "url": "..."}]}\n\nReturn an empty array if none appear to be real statements.';

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

    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 Array.isArray(parsed.statements) ? parsed.statements : [];
    }

    return [];
  } catch (err) {
    console.error('[' + AGENT + '] Gemini statement analysis error for ' + orgName + ':', err.message);
    return [];
  }
}

/**
 * Insert a statement into the database.
 *
 * @param {string} orgId - Organization UUID
 * @param {string} orgName - Organization name
 * @param {Object} statement - Extracted statement data
 * @returns {Promise<Object|null>}
 */
async function insertStatement(orgId, orgName, statement) {
  const pool = getPool();
  try {
    const { rows } = await pool.query(
      'INSERT INTO nonprofit_statements\n        (org_id, org_name, statement_title, statement_body, statement_url, category, topics, sentiment, source_url)\n       VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)\n       RETURNING id, statement_title',
      [
        orgId || null,
        orgName,
        statement.title || 'Untitled Statement',
        statement.summary || '',
        statement.url || null,
        'press-release',
        statement.topics || [],
        statement.sentiment || 'neutral',
        statement.url || null,
      ]
    );
    return rows[0] || null;
  } catch (err) {
    // Skip duplicates silently
    if (err.code === '23505') return null;
    console.error('[' + AGENT + '] DB insert error for statement "' + (statement.title || '') + '":', err.message);
    return null;
  }
}

/**
 * Main collect-statements skill handler.
 *
 * @param {Object} body - Request body
 * @param {number} [body.max_orgs=20] - Max organizations to check
 * @param {boolean} [body.analyze=true] - Whether to AI-analyze found articles
 * @returns {Promise<Object>} Collection results
 */
module.exports = async function collectStatements(body = {}) {
  const maxOrgs = body.max_orgs || 20;
  const shouldAnalyze = body.analyze !== false;

  const pool = getPool();

  // Get organizations with websites
  let orgs;
  try {
    const { rows } = await pool.query(
      "SELECT id, name, website FROM organizations WHERE website IS NOT NULL AND website != '' ORDER BY updated_at DESC LIMIT $1",
      [maxOrgs]
    );
    orgs = rows;
  } catch (err) {
    console.error('[' + AGENT + '] DB query error for organizations:', err.message);
    orgs = [];
  }

  let orgsChecked = 0;
  let statementsFound = 0;
  let pagesChecked = 0;
  const statementsByOrg = [];

  for (const org of orgs) {
    orgsChecked++;
    const orgStatements = [];
    let baseUrl = org.website;

    // Normalize the URL
    if (!baseUrl.startsWith('http')) {
      baseUrl = 'https://' + baseUrl;
    }
    // Remove trailing slash
    baseUrl = baseUrl.replace(/\/+$/, '');

    console.log('[' + AGENT + '] Checking ' + org.name + ' (' + baseUrl + ')...');

    // Try each press path
    for (const path of PRESS_PATHS) {
      const pageUrl = baseUrl + path;
      pagesChecked++;

      const html = await fetchPage(pageUrl);
      if (!html) continue;

      const articleLinks = extractArticleLinks(html, baseUrl);
      if (articleLinks.length === 0) continue;

      console.log('[' + AGENT + ']   Found ' + articleLinks.length + ' links at ' + path);

      if (shouldAnalyze) {
        const analyzed = await geminiAnalyzeStatements(org.name, articleLinks);
        // Delay for Gemini rate limits
        await new Promise(function(r) { setTimeout(r, 400); });

        for (const stmt of analyzed) {
          const inserted = await insertStatement(org.id, org.name, stmt);
          if (inserted) {
            statementsFound++;
            orgStatements.push({
              title: inserted.statement_title,
              topics: stmt.topics,
              sentiment: stmt.sentiment,
            });
          }
        }
      } else {
        // Without analysis, insert raw links
        for (const link of articleLinks.slice(0, 5)) {
          const inserted = await insertStatement(org.id, org.name, {
            title: link.title,
            summary: '',
            url: link.url,
            topics: [],
            sentiment: 'neutral',
          });
          if (inserted) {
            statementsFound++;
            orgStatements.push({ title: inserted.statement_title });
          }
        }
      }

      // Only scrape the first successful path per org to be polite
      break;
    }

    if (orgStatements.length > 0) {
      statementsByOrg.push({
        org: org.name,
        count: orgStatements.length,
        statements: orgStatements.slice(0, 3),
      });
    }
  }

  await logAction({
    agent: AGENT,
    actionType: 'collect',
    platform: PLATFORM,
    content: 'Checked ' + orgsChecked + ' orgs, found ' + statementsFound + ' statements',
    responseData: {
      orgs_checked: orgsChecked,
      statements_found: statementsFound,
      pages_checked: pagesChecked,
      orgs_with_statements: statementsByOrg.length,
    },
    status: 'success',
  });

  return {
    orgs_checked: orgsChecked,
    statements_found: statementsFound,
    pages_checked: pagesChecked,
    by_org: statementsByOrg,
    fetched_at: new Date().toISOString(),
  };
};