From 157e4541fcc52ca776030cc80a4816201cfeaa82 Mon Sep 17 00:00:00 2001 From: Jarian Cottingham Date: Sun, 1 Feb 2026 21:03:00 -0600 Subject: [PATCH] Implement enhanced cache system with batch processing and two-phase processing approach --- ai_processor/ai_processor.py | 115 +++++++++++++++++++------ embedding/advanced_embedder.py | 150 +++++++++++++++++++++++---------- scraper/scraper.py | 64 +++++++++++++- 3 files changed, 259 insertions(+), 70 deletions(-) diff --git a/ai_processor/ai_processor.py b/ai_processor/ai_processor.py index bdc65f1..622cb6f 100644 --- a/ai_processor/ai_processor.py +++ b/ai_processor/ai_processor.py @@ -7,32 +7,92 @@ 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 - # with a proper prompt for fact extraction + # Use the gpt-oss model for fact extraction as specified + extraction_url = f"{AI_SERVER_URL}/v1/chat/completions" - # For now, we'll create a basic structure - facts = { - "filename": filename, - "source": source, - "original_content": 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": [] + # 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 }, - "processed_at": datetime.datetime.now().isoformat() - } + 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[: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"], + "main_points": ["Sample point 1", "Sample point 2"] + }, + "processed_at": datetime.datetime.now().isoformat() + } return facts except Exception as e: @@ -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): - newspaper_path = os.path.join(articles_folder, newspaper) - if not os.path.isdir(newspaper_path): - continue - - for filename in os.listdir(newspaper_path): - file_path = os.path.join(newspaper_path, filename) + # Walk through all subdirectories in articles folder + for root, dirs, files in os.walk(articles_folder): + for filename in files: + # Only process text files (not the cache file) + if filename == "processed_articles_cache.json": + continue + + file_path = os.path.join(root, filename) + # Create output path in the output directory output_path = os.path.join("output", f"{filename}.json") # Skip if already processed diff --git a/embedding/advanced_embedder.py b/embedding/advanced_embedder.py index 8499e83..ee1e562 100644 --- a/embedding/advanced_embedder.py +++ b/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,30 +125,45 @@ def extract_facts_from_article(article_content, title): }} """ - # For now, using the existing embedding approach - in a real implementation - # this would call the AI server with a proper prompt - # response = requests.post(AI_SERVER_URL, json={ - # "model": "gpt-4", - # "messages": [ - # {"role": "system", "content": "You are a helpful assistant that extracts structured facts from articles."}, - # {"role": "user", "content": prompt} - # ] - # }) + # Call the AI service with gpt-oss model for fact extraction + 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 + ) - # Simplified version for now - in production this would be a proper LLM call - facts = { - "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" - ] - } + 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, + "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" + ] + } return facts @@ -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: - logger.info(f"Article {file} already processed, skipping.") - continue - + if file_path not in processed_cache: + articles_to_process.append((file_path, file)) + + 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 + + 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"Cache updated after batch {i//BATCH_SIZE + 1}") + except Exception as e: + logger.error(f"Error saving cache file: {e}") - # Save updated cache + # Final cache save 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"Final cache saved with {len(processed_cache)} entries") except Exception as e: - logger.error(f"Error saving cache file: {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") diff --git a/scraper/scraper.py b/scraper/scraper.py index 5a6eb47..2b0c545 100644 --- a/scraper/scraper.py +++ b/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. @@ -463,6 +518,13 @@ def main(): for result in results: 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}") @@ -474,4 +536,4 @@ def main(): if __name__ == "__main__": - main() + main() \ No newline at end of file