Implement enhanced cache system with batch processing and two-phase processing approach

ai_processor/ai_processor.py

@@ -7,29 +7,89 @@ import time
 # Simplified AI processor for fact extraction
 # This version focuses on the core fact extraction functionality
 
-LOCAL_AI_SERVICE_URL = os.getenv("AI_SERVICE_URL") 
+# AI Service endpoint
+AI_SERVER_HOST = os.getenv("AI_SERVER_HOST", "example.com")
+AI_SERVER_PORT = os.getenv("AI_SERVER_PORT", "4000")
+AI_SERVER_URL = f"http://{AI_SERVER_HOST}:{AI_SERVER_PORT}"
 
 def process_article_content(article_content, filename, source):
     """
-    Process article content and extract key facts
+    Process article content and extract key facts using the centralized AI service
     This is the core fact extraction function
     """
     try:
-        # Simple fact extraction - in a real implementation this would call the AI service
+        # Use the gpt-oss model for fact extraction as specified
-        # with a proper prompt for fact extraction
+        extraction_url = f"{AI_SERVER_URL}/v1/chat/completions"
         
-        # For now, we'll create a basic structure
+        # 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: {filename}
+        Article Content: {article_content[:2000]}...
+        
+        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:
+        {{
+            "filename": "{filename}",
+            "source": "{source}",
+            "original_content": "{article_content[:1000]}...",
+            "extracted_facts": {{
+                "summary": "brief summary",
+                "key_entities": ["entity1", "entity2"],
+                "financial_impact": "positive/negative/neutral",
+                "main_topic": "main topic",
+                "key_dates": ["date1", "date2"],
+                "main_points": ["point1", "point2", "point3"]
+            }},
+            "processed_at": "{datetime.datetime.now().isoformat()}"
+        }}
+        """
+        
+        # Call the AI service with gpt-oss model
+        response = requests.post(
+            extraction_url,
+            json={
+                "model": "gpt-oss",
+                "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
+            },
+            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
             facts = {
                 "filename": filename,
                 "source": source,
-            "original_content": article_content,
+                "original_content": article_content[:1000] + "..." if len(article_content) > 1000 else article_content,
                 "extracted_facts": {
                     "summary": article_content[:200] + "..." if len(article_content) > 200 else article_content,
                     "key_entities": ["Sample Company", "Sample Person"],
                     "financial_impact": "neutral",
                     "main_topic": "Business/Financial News",
                     "key_dates": ["2026"],
-                "tickers_mentioned": []
+                    "main_points": ["Sample point 1", "Sample point 2"]
                 },
                 "processed_at": datetime.datetime.now().isoformat()
             }
@@ -46,7 +106,8 @@ def main_fact_extraction_loop():
     print("Starting fact extraction loop...")
     
     # Retrieve the current archive of pulled articles
-    articles_folder = os.path.join("/app/articles")
+    # Use the correct path for the scraper articles directory
+    articles_folder = os.path.join("articles")
     if not os.path.exists(articles_folder):
         print(f"Articles folder {articles_folder} does not exist. Please check the path.")
         return
@@ -57,13 +118,15 @@ def main_fact_extraction_loop():
     processed_count = 0
     failed_count = 0
     
-    for newspaper in os.listdir(articles_folder):
+    # Walk through all subdirectories in articles folder
-        newspaper_path = os.path.join(articles_folder, newspaper)
+    for root, dirs, files in os.walk(articles_folder):
-        if not os.path.isdir(newspaper_path):
+        for filename in files:
+            # Only process text files (not the cache file)
+            if filename == "processed_articles_cache.json":
                 continue
                 
-        for filename in os.listdir(newspaper_path):
+            file_path = os.path.join(root, filename)
-            file_path = os.path.join(newspaper_path, filename)
+            # Create output path in the output directory
             output_path = os.path.join("output", f"{filename}.json")
             
             # Skip if already processed

embedding/advanced_embedder.py

@@ -54,10 +54,10 @@ except Exception as e:
     # For now, let's continue but log the error
     pass
 
-# AI Server configuration
+# AI Server configuration - using the centralized endpoint
 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}/v1/embeddings"
+AI_SERVER_URL = f"http://{AI_SERVER_HOST}:{AI_SERVER_PORT}"
 
 # Cache file for tracking processed articles
 CACHE_FILE = os.getenv("CACHE_FILE", "processed_articles_cache.json")
@@ -66,18 +66,22 @@ CACHE_FILE = os.getenv("CACHE_FILE", "processed_articles_cache.json")
 OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "placeholder-key")
 OPENAI_BASE_URL = os.getenv("OPENAI_BASE_URL", f"http://{AI_SERVER_HOST}:{AI_SERVER_PORT}")
 
+# Batch processing configuration
+BATCH_SIZE = int(os.getenv("EMBEDDING_BATCH_SIZE", "50"))
+
 def get_embedding(text):
     """
-    Get embedding using the OpenAI-compatible server
+    Get embedding using the OpenAI-compatible server with qwen3:8b model
     """
     try:
+        embedding_url = f"{AI_SERVER_URL}/v1/embeddings"
         response = requests.post(
-            AI_SERVER_URL,
+            embedding_url,
             json={
                 "input": text,
                 "model": "qwen3:8b"
             },
-            timeout=30
+            timeout=60
         )
         response.raise_for_status()
         embedding = response.json()['data'][0]['embedding']
@@ -88,16 +92,19 @@ def get_embedding(text):
 
 def extract_facts_from_article(article_content, title):
     """
-    Extract structured facts from article content using LLM with proper prompting
+    Extract structured facts from article content using the centralized AI service with gpt-oss model
     """
     try:
+        # Use the centralized AI endpoint for fact extraction
+        extraction_url = f"{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[:1000]}...
+        Article Content: {article_content[:2000]}...
         
         Extract the following information:
         1. Main topic/subject
@@ -108,7 +115,7 @@ def extract_facts_from_article(article_content, title):
         
         Format the response as a JSON object with these fields:
         {{
-            "title": "article title",
+            "title": "{title}",
             "summary": "brief summary",
             "main_topic": "main topic",
             "key_entities": ["entity1", "entity2"],
@@ -118,17 +125,32 @@ def extract_facts_from_article(article_content, title):
         }}
         """
         
-        # For now, using the existing embedding approach - in a real implementation
+        # Call the AI service with gpt-oss model for fact extraction
-        # this would call the AI server with a proper prompt
+        response = requests.post(
-        # response = requests.post(AI_SERVER_URL, json={
+            extraction_url,
-        #     "model": "gpt-4",
+            json={
-        #     "messages": [
+                "model": "gpt-oss",
-        #         {"role": "system", "content": "You are a helpful assistant that extracts structured facts from articles."},
+                "messages": [
-        #         {"role": "user", "content": prompt}
+                    {"role": "system", "content": "You are a helpful assistant that extracts structured facts from articles."},
-        #     ]
+                    {"role": "user", "content": prompt}
-        # })
+                ],
+                "temperature": 0.3,
+                "max_tokens": 1000
+            },
+            timeout=60
+        )
         
-        # Simplified version for now - in production this would be a proper LLM call
+        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
             facts = {
                 "title": title,
                 "summary": article_content[:200] + "..." if len(article_content) > 200 else article_content,
@@ -248,7 +270,7 @@ def embed_and_store_facts(facts, facts_collection, articles_collection):
 
 def main():
     """
-    Main embedding pipeline function
+    Main embedding pipeline function with batch processing
     """
     logger.info("Starting advanced embedding pipeline")
     
@@ -267,17 +289,31 @@ def main():
     # Track processing time
     start_time = datetime.datetime.now()
     
-    # Walk through all subdirectories in scraper articles
+    # Collect all articles to process
+    articles_to_process = []
     for root, dirs, files in os.walk(scraper_articles_dir):
         for file in files:
             if file.endswith('.json'):
                 file_path = os.path.join(root, file)
-                
                 # Check if already processed
-                if file_path in processed_cache:
+                if file_path not in processed_cache:
-                    logger.info(f"Article {file} already processed, skipping.")
+                    articles_to_process.append((file_path, file))
-                    continue
     
+    logger.info(f"Found {len(articles_to_process)} articles to process in batches of {BATCH_SIZE}")
+    
+    # Process articles in batches
+    total_processed = 0
+    total_failed = 0
+    
+    for i in range(0, len(articles_to_process), BATCH_SIZE):
+        batch = articles_to_process[i:i + BATCH_SIZE]
+        logger.info(f"Processing batch {i//BATCH_SIZE + 1} with {len(batch)} articles")
+        
+        batch_processed = 0
+        batch_failed = 0
+        
+        for file_path, file in batch:
+            try:
                 # Process the article
                 facts = process_article_file(file_path)
                 if facts:
@@ -289,29 +325,57 @@ def main():
                     )
                     
                     if success:
+                        # Update cache with detailed status tracking
                         processed_cache[file_path] = {
                             "processed_date": datetime.datetime.now().isoformat(),
-                            "status": "completed"
+                            "status": "fact_extracted",
+                            "embedding_status": "pending",
+                            "last_updated": datetime.datetime.now().isoformat()
                         }
                         articles_processed_total.inc()
                         logger.info(f"Successfully processed {file}")
+                        batch_processed += 1
+                        total_processed += 1
                     else:
                         articles_failed_total.inc()
-                        logger.error(f"Failed to process {file}")
+                        logger.error(f"Failed to embed {file}")
+                        batch_failed += 1
+                        total_failed += 1
+                else:
+                    logger.error(f"Failed to extract facts for {file}")
+                    batch_failed += 1
+                    total_failed += 1
                     
-    # Save updated cache
+            except Exception as e:
+                logger.error(f"Error processing article {file}: {e}")
+                articles_failed_total.inc()
+                batch_failed += 1
+                total_failed += 1
+        
+        logger.info(f"Batch {i//BATCH_SIZE + 1} completed: {batch_processed} successful, {batch_failed} failed")
+        
+        # Save cache periodically during batch processing
         try:
             with open(CACHE_FILE, 'w', encoding='utf-8') as f:
                 json.dump(processed_cache, f, indent=2)
-        logger.info(f"Updated cache with newly processed articles. Total cached: {len(processed_cache)}")
+            logger.info(f"Cache updated after batch {i//BATCH_SIZE + 1}")
         except Exception as e:
             logger.error(f"Error saving cache file: {e}")
     
+    # Final cache save
+    try:
+        with open(CACHE_FILE, 'w', encoding='utf-8') as f:
+            json.dump(processed_cache, f, indent=2)
+        logger.info(f"Final cache saved with {len(processed_cache)} entries")
+    except Exception as e:
+        logger.error(f"Error saving final cache file: {e}")
+    
     # Calculate and log processing time
     end_time = datetime.datetime.now()
     total_time = (end_time - start_time).total_seconds()
     processing_time_seconds.observe(total_time)
     logger.info(f"Embedding pipeline completed in {total_time:.2f} seconds")
+    logger.info(f"Total processed: {total_processed}, Total failed: {total_failed}")
     
     logger.info("Embedding pipeline completed")
 

scraper/scraper.py

@@ -25,6 +25,7 @@ logger = logging.getLogger(__name__)
 FEED_FILE = os.getenv("FEED_FILE", "./rss_short_feed.json")
 MAX_FEED_WORKERS = int(os.getenv("MAX_FEED_WORKERS", "10"))
 MAX_ARTICLE_WORKERS = int(os.getenv("MAX_ARTICLE_WORKERS", "10"))
+BATCH_SIZE = int(os.getenv("BATCH_SIZE", "50"))  # Batch processing size
 
 # Ensure necessary NLTK resources are downloaded
 d = Downloader()
@@ -34,6 +35,60 @@ if not d.is_installed("punkt_tab"):
 articles = []
 
 
+# Enhanced cache system
+def load_processed_cache():
+    """Load the processed articles cache with enhanced tracking"""
+    cache_path = "articles/processed_articles_cache.json"
+    try:
+        if os.path.exists(cache_path):
+            with open(cache_path, 'r', encoding='utf-8') as f:
+                return json.load(f)
+        else:
+            return {}
+    except Exception as e:
+        logger.error(f"Error loading cache: {e}")
+        return {}
+
+def save_processed_cache(cache_data):
+    """Save the processed articles cache with enhanced tracking"""
+    cache_path = "articles/processed_articles_cache.json"
+    try:
+        with open(cache_path, 'w', encoding='utf-8') as f:
+            json.dump(cache_data, f, indent=2, ensure_ascii=False)
+        logger.info(f"Cache saved with {len(cache_data)} entries")
+    except Exception as e:
+        logger.error(f"Error saving cache: {e}")
+
+def is_article_processed(article_path, cache_data):
+    """Check if an article has been processed"""
+    return article_path in cache_data
+
+def mark_article_processed(article_path, status="completed", embedding_status="pending"):
+    """Mark an article as processed with detailed status tracking"""
+    cache_data = load_processed_cache()
+    cache_data[article_path] = {
+        "processed_date": datetime.now().isoformat(),
+        "status": status,
+        "embedding_status": embedding_status,
+        "last_updated": datetime.now().isoformat()
+    }
+    save_processed_cache(cache_data)
+    
+def get_processing_progress():
+    """Get overall processing progress"""
+    cache_data = load_processed_cache()
+    total_articles = len(cache_data)
+    completed_articles = sum(1 for data in cache_data.values() if data.get('status') == 'completed')
+    embedded_articles = sum(1 for data in cache_data.values() if data.get('embedding_status') == 'completed')
+    
+    return {
+        "total_articles": total_articles,
+        "completed_articles": completed_articles,
+        "embedded_articles": embedded_articles,
+        "completion_rate": (completed_articles / total_articles * 100) if total_articles > 0 else 0
+    }
+
+
 def load_rss_feed_sources(feed_file=FEED_FILE):
     """
     Loads the RSS feed sources from a JSON file.
@@ -464,6 +519,13 @@ def main():
                     f.write(result + "\n")
             logger.info("All articles pulled successfully.")
             
+            # Log processing progress
+            progress = get_processing_progress()
+            logger.info(f"Processing progress - Total: {progress['total_articles']}, "
+                       f"Completed: {progress['completed_articles']}, "
+                       f"Embedded: {progress['embedded_articles']}, "
+                       f"Completion rate: {progress['completion_rate']:.1f}%")
+
         except Exception as e:
             logger.error(f"Major error in main loop: {e}")
             # Continue to next iteration even if there's a major error