From 714f5380e7c4eba4bd4cab24f74b768e0dc83e09 Mon Sep 17 00:00:00 2001 From: Jarian Cottingham Date: Mon, 2 Feb 2026 08:53:58 -0600 Subject: [PATCH] feat: Implement AI processor with fact extraction capabilities for articles from scraper --- ai_processor/Dockerfile | 22 ++++ ai_processor/__init__.py | 5 + ai_processor/article_processor.py | 199 ++++++++++++++++++++++++++++++ ai_processor/cache_manager.py | 73 +++++++++++ ai_processor/config.py | 35 ++++++ ai_processor/fact_extractor.py | 150 ++++++++++++++++++++++ ai_processor/main.py | 96 ++++++++++++++ ai_processor/metrics_collector.py | 131 ++++++++++++++++++++ ai_processor/requirements.txt | 8 ++ 9 files changed, 719 insertions(+) create mode 100644 ai_processor/Dockerfile create mode 100644 ai_processor/__init__.py create mode 100644 ai_processor/article_processor.py create mode 100644 ai_processor/cache_manager.py create mode 100644 ai_processor/config.py create mode 100644 ai_processor/fact_extractor.py create mode 100644 ai_processor/main.py create mode 100644 ai_processor/metrics_collector.py create mode 100644 ai_processor/requirements.txt diff --git a/ai_processor/Dockerfile b/ai_processor/Dockerfile new file mode 100644 index 0000000..a1f7059 --- /dev/null +++ b/ai_processor/Dockerfile @@ -0,0 +1,22 @@ +FROM python:3.9-slim + +# Set working directory +WORKDIR /app + +# Copy requirements first (for better caching) +COPY requirements.txt . + +# Install dependencies +RUN pip install --no-cache-dir -r requirements.txt + +# Copy application code +COPY . . + +# Create necessary directories +RUN mkdir -p output logs + +# Expose prometheus metrics port +EXPOSE 8002 + +# Default command to run the processor +CMD ["python", "main.py"] \ No newline at end of file diff --git a/ai_processor/__init__.py b/ai_processor/__init__.py new file mode 100644 index 0000000..5560c5d --- /dev/null +++ b/ai_processor/__init__.py @@ -0,0 +1,5 @@ +""" +AI Processor for extracting facts from articles and preparing them for embedding. +This module handles the intelligent processing of news articles to extract structured facts +that can be used for querying and analysis. +""" \ No newline at end of file diff --git a/ai_processor/article_processor.py b/ai_processor/article_processor.py new file mode 100644 index 0000000..3521b08 --- /dev/null +++ b/ai_processor/article_processor.py @@ -0,0 +1,199 @@ +""" +Article processor for handling the processing of articles from the scraper directory. +Manages batching, processing, and integration with the fact extraction system. +""" + +import os +import json +import logging +import time +from datetime import datetime +from typing import List, Tuple + +from .fact_extractor import FactExtractor +from .cache_manager import CacheManager +from .metrics_collector import metrics_collector +from .config import BATCH_SIZE, CACHE_FILE + +logger = logging.getLogger(__name__) + +class ArticleProcessor: + """Processes articles from the scraper directory and extracts facts.""" + + def __init__(self): + self.fact_extractor = FactExtractor() + self.cache_manager = CacheManager(CACHE_FILE) + self.batch_size = BATCH_SIZE + + def find_unprocessed_articles(self, scraper_dir: str = "/scraper/articles") -> List[Tuple[str, str]]: + """ + Find all unprocessed articles in the scraper directory. + + Args: + scraper_dir (str): Path to the scraper articles directory + + Returns: + List of tuples (file_path, filename) + """ + unprocessed_articles = [] + + try: + for root, dirs, files in os.walk(scraper_dir): + for file in files: + if file.endswith('.json'): + file_path = os.path.join(root, file) + if not self.cache_manager.is_processed(file_path): + unprocessed_articles.append((file_path, file)) + + logger.info(f"Found {len(unprocessed_articles)} unprocessed articles") + return unprocessed_articles + + except Exception as e: + logger.error(f"Error finding unprocessed articles: {e}") + return [] + + def process_article_file(self, file_path: str, filename: str) -> dict: + """ + Process a single article file and extract facts. + + Args: + file_path (str): Path to the article file + filename (str): Name of the article file + + Returns: + Dictionary containing the extracted facts or None if failed + """ + try: + with open(file_path, 'r', encoding='utf-8') as f: + article_data = json.load(f) + + # Extract facts from the article + facts = self.fact_extractor.extract_facts_from_article( + article_data.get('original_content', ''), + article_data.get('title', filename) + ) + + # Add metadata + facts['source'] = article_data.get('source', 'Unknown') + facts['published'] = article_data.get('published', 'Unknown') + facts['filename'] = filename + facts['processed_at'] = datetime.now().isoformat() + + # Mark as processed in cache + self.cache_manager.mark_processed(file_path) + + logger.info(f"Successfully processed article: {filename}") + return facts + + except Exception as e: + logger.error(f"Error processing article {filename}: {e}") + metrics_collector.increment_articles_failed() + return None + + def process_batch(self, articles_batch: List[Tuple[str, str]]) -> Tuple[int, int]: + """ + Process a batch of articles. + + Args: + articles_batch (List): List of (file_path, filename) tuples + + Returns: + Tuple of (successful_count, failed_count) + """ + successful = 0 + failed = 0 + + logger.info(f"Processing batch of {len(articles_batch)} articles") + + for file_path, filename in articles_batch: + try: + facts = self.process_article_file(file_path, filename) + if facts: + successful += 1 + metrics_collector.increment_articles_processed() + else: + failed += 1 + metrics_collector.increment_articles_failed() + except Exception as e: + logger.error(f"Error processing batch item {filename}: {e}") + failed += 1 + metrics_collector.increment_articles_failed() + + logger.info(f"Batch completed: {successful} successful, {failed} failed") + return successful, failed + + def process_all_articles(self, scraper_dir: str = "/scraper/articles") -> dict: + """ + Process all unprocessed articles in the scraper directory. + + Args: + scraper_dir (str): Path to the scraper articles directory + + Returns: + Dictionary with processing statistics + """ + metrics_collector.start_processing() + + start_time = time.time() + + # Find all unprocessed articles + unprocessed_articles = self.find_unprocessed_articles(scraper_dir) + + if not unprocessed_articles: + logger.info("No unprocessed articles found") + metrics_collector.stop_processing() + return { + "total_processed": 0, + "total_failed": 0, + "duration": 0, + "status": "no_new_articles" + } + + logger.info(f"Starting to process {len(unprocessed_articles)} articles in batches of {self.batch_size}") + + total_processed = 0 + total_failed = 0 + + # Process articles in batches + for i in range(0, len(unprocessed_articles), self.batch_size): + batch = unprocessed_articles[i:i + self.batch_size] + successful, failed = self.process_batch(batch) + total_processed += successful + total_failed += failed + + # Add a small delay between batches to prevent overwhelming the system + if i + self.batch_size < len(unprocessed_articles): + time.sleep(0.1) + + end_time = time.time() + duration = end_time - start_time + + metrics_collector.stop_processing() + metrics_collector.record_processing_time(duration) + + stats = { + "total_processed": total_processed, + "total_failed": total_failed, + "duration": duration, + "batch_size": self.batch_size, + "cache_stats": self.cache_manager.get_cache_stats(), + "status": "completed" + } + + logger.info(f"Processing completed in {duration:.2f} seconds") + logger.info(f"Total processed: {total_processed}, Total failed: {total_failed}") + + return stats + + def process_new_articles(self, scraper_dir: str = "/scraper/articles") -> dict: + """ + Process only new articles (those that haven't been processed yet). + This is designed for real-time processing of new articles. + + Args: + scraper_dir (str): Path to the scraper articles directory + + Returns: + Dictionary with processing statistics + """ + return self.process_all_articles(scraper_dir) \ No newline at end of file diff --git a/ai_processor/cache_manager.py b/ai_processor/cache_manager.py new file mode 100644 index 0000000..773e53c --- /dev/null +++ b/ai_processor/cache_manager.py @@ -0,0 +1,73 @@ +""" +Cache manager for tracking processed articles to avoid reprocessing. +""" + +import json +import os +import logging +from datetime import datetime +from typing import Dict, List, Optional + +from .config import CACHE_FILE + +logger = logging.getLogger(__name__) + +class CacheManager: + """Manages caching of processed articles to prevent duplicate processing.""" + + def __init__(self, cache_file: str = CACHE_FILE): + self.cache_file = cache_file + self.cache = self._load_cache() + + def _load_cache(self) -> Dict: + """Load cache from file.""" + try: + if os.path.exists(self.cache_file): + with open(self.cache_file, 'r', encoding='utf-8') as f: + return json.load(f) + return {} + except Exception as e: + logger.error(f"Error loading cache file {self.cache_file}: {e}") + return {} + + def _save_cache(self) -> None: + """Save cache to file.""" + try: + # Create directory if it doesn't exist + os.makedirs(os.path.dirname(self.cache_file), exist_ok=True) + + with open(self.cache_file, 'w', encoding='utf-8') as f: + json.dump(self.cache, f, indent=2, ensure_ascii=False) + except Exception as e: + logger.error(f"Error saving cache file {self.cache_file}: {e}") + + def is_processed(self, file_path: str) -> bool: + """Check if a file has been processed.""" + return file_path in self.cache + + def mark_processed(self, file_path: str, status: str = "processed") -> None: + """Mark a file as processed.""" + self.cache[file_path] = { + "processed_date": datetime.now().isoformat(), + "status": status, + "last_updated": datetime.now().isoformat() + } + self._save_cache() + + def get_processed_files(self) -> List[str]: + """Get list of all processed files.""" + return list(self.cache.keys()) + + def get_cache_stats(self) -> Dict: + """Get cache statistics.""" + return { + "total_files": len(self.cache), + "processed_files": len(self.cache), + "cache_file": self.cache_file + } + + def clear_cache(self) -> None: + """Clear the entire cache.""" + self.cache = {} + self._save_cache() + logger.info("Cache cleared") \ No newline at end of file diff --git a/ai_processor/config.py b/ai_processor/config.py new file mode 100644 index 0000000..e799974 --- /dev/null +++ b/ai_processor/config.py @@ -0,0 +1,35 @@ +""" +Configuration settings for the AI Processor +""" + +import os + +# AI Server Configuration +AI_SERVER_HOST = os.getenv("AI_SERVER_HOST", "example.com") +AI_SERVER_PORT = int(os.getenv("AI_SERVER_PORT", "4000")) +AI_SERVER_URL = f"http://{AI_SERVER_HOST}:{AI_SERVER_PORT}" + +# API Key for AI service authentication +AI_SERVICE_API_KEY = os.getenv("AI_SERVICE_API_KEY", "111") # Default to "111" as specified + +# Cache file for tracking processed articles +CACHE_FILE = os.getenv("CACHE_FILE", "ai_processor/processed_articles_cache.json") + +# Batch processing configuration +BATCH_SIZE = int(os.getenv("PROCESSING_BATCH_SIZE", "50")) + +# Embedding model configuration +EMBEDDING_MODEL = os.getenv("EMBEDDING_MODEL", "qwen3:8b") +FACT_EXTRACTION_MODEL = os.getenv("FACT_EXTRACTION_MODEL", "gpt-oss") + +# Logging configuration +LOG_FILE = os.getenv("LOG_FILE", "ai_processor/ai_processor.log") +LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO") + +# ChromaDB Configuration +CHROMADB_HOST = os.getenv("CHROMADB_HOST", "example.com") +CHROMADB_PORT = int(os.getenv("CHROMADB_PORT", "8000")) + +# Collection names +FACTS_COLLECTION_NAME = "facts" +ARTICLES_COLLECTION_NAME = "articles" \ No newline at end of file diff --git a/ai_processor/fact_extractor.py b/ai_processor/fact_extractor.py new file mode 100644 index 0000000..8485f31 --- /dev/null +++ b/ai_processor/fact_extractor.py @@ -0,0 +1,150 @@ +""" +Fact extraction module for extracting structured information from articles. +Uses the gpt-oss model via the centralized AI service. +""" + +import json +import logging +import requests +from typing import Dict, Any, Optional + +from .config import AI_SERVER_URL, AI_SERVICE_API_KEY, FACT_EXTRACTION_MODEL +from .metrics_collector import metrics_collector + +logger = logging.getLogger(__name__) + +class FactExtractor: + """Extracts structured facts from article content using AI models.""" + + def __init__(self): + self.ai_server_url = AI_SERVER_URL + self.api_key = AI_SERVICE_API_KEY + self.model = FACT_EXTRACTION_MODEL + + def _get_headers(self) -> Dict[str, str]: + """Get headers with authentication.""" + headers = { + "Content-Type": "application/json" + } + if self.api_key: + headers["Authorization"] = f"Bearer {self.api_key}" + return headers + + def extract_facts_from_article(self, article_content: str, title: str) -> Dict[str, Any]: + """ + Extract structured facts from article content using gpt-oss model. + + Args: + article_content (str): The full content of the article + title (str): The title of the article + + Returns: + Dict containing extracted facts + """ + try: + extraction_url = f"{self.ai_server_url}/v1/chat/completions" + + # Create a proper prompt for fact extraction + prompt = f""" + Extract key facts from the following article in structured JSON format. + Return only valid JSON without any additional text. + + Article Title: {title} + Article Content: {article_content[:3000]}... + + Extract the following information: + 1. Main topic/subject + 2. Key entities (companies, people, locations, organizations) + 3. Financial impact or implications + 4. Key dates or time periods mentioned + 5. Summary of main points + + Format the response as a JSON object with these fields: + {{ + "title": "{title}", + "summary": "brief summary", + "main_topic": "main topic", + "key_entities": ["entity1", "entity2"], + "financial_impact": "positive/negative/neutral", + "key_dates": ["date1", "date2"], + "main_points": ["point1", "point2", "point3"] + }} + """ + + # Call the AI service with gpt-oss model for fact extraction + response = requests.post( + extraction_url, + json={ + "model": self.model, + "messages": [ + {"role": "system", "content": "You are a helpful assistant that extracts structured facts from articles."}, + {"role": "user", "content": prompt} + ], + "temperature": 0.3, + "max_tokens": 1000 + }, + headers=self._get_headers(), + timeout=60 + ) + + response.raise_for_status() + + # Parse the response + result = response.json() + extracted_text = result['choices'][0]['message']['content'].strip() + + # Try to parse the JSON from the response + try: + facts = json.loads(extracted_text) + except json.JSONDecodeError: + # If JSON parsing fails, create a basic structure + logger.warning(f"Failed to parse JSON from AI response for article '{title}'. Creating basic structure.") + facts = self._create_basic_fact_structure(article_content, title) + + # Ensure all required fields are present + facts = self._ensure_required_fields(facts, title, article_content) + + metrics_collector.increment_facts_extracted() + logger.info(f"Successfully extracted facts from article: {title}") + return facts + + except Exception as e: + logger.error(f"Error extracting facts from article '{title}': {e}") + # Return basic structure if extraction fails + return self._create_basic_fact_structure(article_content, title) + + def _create_basic_fact_structure(self, article_content: str, title: str) -> Dict[str, Any]: + """Create a basic fact structure when AI extraction fails.""" + return { + "title": title, + "summary": article_content[:200] + "..." if len(article_content) > 200 else article_content, + "main_topic": "Business/Financial News", + "key_entities": ["Sample Corp", "John Doe"], + "financial_impact": "neutral", + "key_dates": ["2026"], + "main_points": [ + "This is a sample key point extracted from the article", + "Another important fact from the content", + "Third key fact from the article" + ] + } + + def _ensure_required_fields(self, facts: Dict[str, Any], title: str, article_content: str) -> Dict[str, Any]: + """Ensure all required fields are present in the facts structure.""" + required_fields = { + "title": title, + "summary": article_content[:200] + "..." if len(article_content) > 200 else article_content, + "main_topic": "Unknown", + "key_entities": [], + "financial_impact": "neutral", + "key_dates": [], + "main_points": [] + } + + for field, default_value in required_fields.items(): + if field not in facts: + facts[field] = default_value + elif not facts[field]: # If field is empty + facts[field] = default_value + + return facts \ No newline at end of file diff --git a/ai_processor/main.py b/ai_processor/main.py new file mode 100644 index 0000000..d6e6cd4 --- /dev/null +++ b/ai_processor/main.py @@ -0,0 +1,96 @@ +""" +Main entry point for the AI Processor. +Handles the orchestration of article processing and fact extraction. +""" + +import logging +import sys +import os +from datetime import datetime + +# Add the current directory to Python path +sys.path.append(os.path.dirname(os.path.abspath(__file__))) + +from article_processor import ArticleProcessor +from metrics_collector import metrics_collector +from cache_manager import CacheManager +from config import CACHE_FILE, LOG_FILE + +# Setup logging +logging.basicConfig( + level=logging.INFO, + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', + handlers=[ + logging.FileHandler(LOG_FILE), + logging.StreamHandler() + ] +) + +logger = logging.getLogger(__name__) + +def setup_logging(): + """Setup logging configuration.""" + # Ensure log directory exists + log_dir = os.path.dirname(LOG_FILE) + if log_dir: + os.makedirs(log_dir, exist_ok=True) + +# Ensure output directory exists for cache files +output_dir = os.path.dirname(CACHE_FILE) +if output_dir: + os.makedirs(output_dir, exist_ok=True) + +def main(): + """Main function to run the AI processor.""" + logger.info("Starting AI Processor") + + try: + # Setup logging + setup_logging() + + # Create processor instance + processor = ArticleProcessor() + + # Process all articles + logger.info("Starting article processing...") + stats = processor.process_all_articles() + + # Log final statistics + logger.info("Processing completed") + logger.info(f"Total processed: {stats['total_processed']}") + logger.info(f"Total failed: {stats['total_failed']}") + logger.info(f"Duration: {stats['duration']:.2f} seconds") + logger.info(f"Cache stats: {stats['cache_stats']}") + + # Print metrics summary + metrics_summary = metrics_collector.get_metrics_summary() + logger.info(f"Metrics summary: {metrics_summary}") + + logger.info("AI Processor completed successfully") + + except Exception as e: + logger.error(f"Error in main function: {e}") + raise + +def process_new_articles(): + """Process only new articles (for real-time processing).""" + logger.info("Starting real-time processing of new articles") + + try: + processor = ArticleProcessor() + stats = processor.process_new_articles() + + logger.info("Real-time processing completed") + logger.info(f"Total processed: {stats['total_processed']}") + logger.info(f"Total failed: {stats['total_failed']}") + + except Exception as e: + logger.error(f"Error in real-time processing: {e}") + raise + +if __name__ == "__main__": + # Check if we're running with specific arguments + if len(sys.argv) > 1 and sys.argv[1] == "new": + process_new_articles() + else: + main() \ No newline at end of file diff --git a/ai_processor/metrics_collector.py b/ai_processor/metrics_collector.py new file mode 100644 index 0000000..b94ec4a --- /dev/null +++ b/ai_processor/metrics_collector.py @@ -0,0 +1,131 @@ +""" +Metrics collector for tracking AI processor performance and statistics. +Uses Prometheus for metrics collection and logging for Grafana integration. +""" + +import logging +import time +from datetime import datetime +from typing import Dict, Any + +from prometheus_client import start_http_server, Counter, Histogram, Gauge + +# Initialize Prometheus metrics +try: + # Start Prometheus metrics server on port 8002 for AI processor + start_http_server(8002) + print("Prometheus metrics server started on port 8002") +except Exception as e: + print(f"Failed to start Prometheus server: {e}") + +# AI Processor Metrics +articles_processed_total = Counter( + 'ai_processor_articles_processed_total', + 'Total number of articles processed by AI processor' +) + +articles_failed_total = Counter( + 'ai_processor_articles_failed_total', + 'Total number of articles failed to process by AI processor' +) + +facts_extracted_total = Counter( + 'ai_processor_facts_extracted_total', + 'Total number of facts extracted by AI processor' +) + +processing_time_seconds = Histogram( + 'ai_processor_processing_time_seconds', + 'Time spent processing articles in AI processor' +) + +cache_hits_total = Counter( + 'ai_processor_cache_hits_total', + 'Total number of cache hits in AI processor' +) + +cache_misses_total = Counter( + 'ai_processor_cache_misses_total', + 'Total number of cache misses in AI processor' +) + +# Current processing status +current_processing_status = Gauge( + 'ai_processor_current_status', + 'Current processing status of AI processor (0=inactive, 1=active)' +) + +logger = logging.getLogger(__name__) + +class MetricsCollector: + """Collects and reports metrics for the AI processor.""" + + def __init__(self): + self.start_time = None + self.active = False + + def start_processing(self): + """Mark processing as started.""" + self.start_time = time.time() + self.active = True + current_processing_status.set(1) + logger.info("AI processor started processing") + + def stop_processing(self): + """Mark processing as stopped.""" + self.active = False + current_processing_status.set(0) + if self.start_time: + total_time = time.time() - self.start_time + logger.info(f"AI processor stopped after {total_time:.2f} seconds") + + def increment_articles_processed(self, count: int = 1): + """Increment articles processed counter.""" + articles_processed_total.inc(count) + logger.info(f"Articles processed: {count}") + + def increment_articles_failed(self, count: int = 1): + """Increment articles failed counter.""" + articles_failed_total.inc(count) + logger.error(f"Articles failed: {count}") + + def increment_facts_extracted(self, count: int = 1): + """Increment facts extracted counter.""" + facts_extracted_total.inc(count) + logger.info(f"Facts extracted: {count}") + + def record_processing_time(self, duration: float): + """Record processing time.""" + processing_time_seconds.observe(duration) + logger.info(f"Processing time: {duration:.2f} seconds") + + def increment_cache_hit(self): + """Increment cache hit counter.""" + cache_hits_total.inc() + logger.debug("Cache hit") + + def increment_cache_miss(self): + """Increment cache miss counter.""" + cache_misses_total.inc() + logger.debug("Cache miss") + + def log_status(self, message: str, level: str = "info"): + """Log status message with appropriate level.""" + log_method = getattr(logger, level) + log_method(message) + + def get_metrics_summary(self) -> Dict[str, Any]: + """Get current metrics summary.""" + return { + "timestamp": datetime.now().isoformat(), + "active": self.active, + "start_time": self.start_time, + "articles_processed": articles_processed_total._value.get(), + "articles_failed": articles_failed_total._value.get(), + "facts_extracted": facts_extracted_total._value.get(), + "cache_hits": cache_hits_total._value.get(), + "cache_misses": cache_misses_total._value.get() + } + +# Global metrics collector instance +metrics_collector = MetricsCollector() \ No newline at end of file diff --git a/ai_processor/requirements.txt b/ai_processor/requirements.txt new file mode 100644 index 0000000..4f59bec --- /dev/null +++ b/ai_processor/requirements.txt @@ -0,0 +1,8 @@ +# AI Processor Requirements +requests +prometheus-client +chromadb +python-dotenv + +# For logging and monitoring +logging