feat: Implement AI processor with fact extraction capabilities for articles from scraper

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"]

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.
+"""

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)

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")

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"

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

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()

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()

ai_processor/requirements.txt

@@ -0,0 +1,8 @@
+# AI Processor Requirements
+requests
+prometheus-client
+chromadb
+python-dotenv
+
+# For logging and monitoring
+logging