lambda_function.py

import json
import boto3
import psycopg2
from psycopg2.extras import RealDictCursor
import os
from datetime import datetime
import sys
import logging

# Add the dashboard directory to Python path
CURRENT_DIR = os.path.dirname(os.path.abspath(__file__))
sys.path.append(CURRENT_DIR)

# Import your existing services.
# ML classification is intentionally NOT imported here: it moved out of this
# Lambda into the dedicated classifier container (the ensemble is ~13GB and
# exceeds Lambda's 10GB limits). This Lambda now only ingests; the classifier
# fills strategic_intent/tone/confidence from the dashboard_medianarrative table.
from dashboard.services.mediacloud_ingestion_service import main as run_mediacloud_ingestion

logger = logging.getLogger(__name__)

# Use the table name from your ingestion script
TABLE_NAME = "dashboard_medianarrative" 

def get_db_connection():
    return psycopg2.connect(
        host=os.environ.get('DB_HOST'),
        database=os.environ.get('DB_NAME'),
        user=os.environ.get('DB_USER'),
        password=os.environ.get('DB_PASSWORD'),
        port=os.environ.get('DB_PORT', '5432')
    )

def lambda_handler(event, context):
    conn = None
    try:
        # Map Environment Variables (Ensures consistency)
        os.environ['API_KEY'] = os.environ.get('MEDIACLOUD_API_KEY', '')
        
        conn = get_db_connection()
        
        # 1. Initial State
        initial_count = get_count(conn)
        logger.info(f"Starting ingestion. Current count: {initial_count}")
        
        # 2. Run Ingestion (This calls your scraping/mediacloud logic)
        run_mediacloud_ingestion()

        # 3. Final Validation
        run_quality_validation(conn)

        final_count = get_count(conn)

        # Classification of the newly-ingested rows is handled out-of-band by the
        # classifier container (fill_missing_intents), which drains rows whose
        # strategic_intent is still null.
        return {
            'statusCode': 200,
            'body': json.dumps({
                'message': 'Success',
                'initial_count': initial_count,
                'ingested': final_count - initial_count,
                'final_count': final_count
            })
        }
        
    except Exception as e:
        logger.error(f"Lambda Failure: {str(e)}")
        return {'statusCode': 500, 'body': json.dumps({'error': str(e)})}
    finally:
        if conn:
            conn.close()

def get_count(conn):
    with conn.cursor() as cur:
        cur.execute(f"SELECT COUNT(*) FROM {TABLE_NAME}")
        return cur.fetchone()[0]

def run_quality_validation(conn):
    with conn.cursor() as cursor:
        cursor.execute(f"""
            UPDATE {TABLE_NAME} SET pseudo_kept = TRUE, pseudo_weight = 1.0
            WHERE pseudo_kept IS NULL AND article_text IS NOT NULL AND LENGTH(article_text) > 100
        """)
        conn.commit()