Jarian Cottingham
1271f0b21b
- Remove agent/agent.md (dev-time agent context dumps), .DS_Store, committed venv configs (pyvenv.cfg), 0-byte runtime cache - Remove hardcoded /home/userpath from cron_scraper feed lookup - Replace ad-hoc test_implementation.py with pytest tests/test_scraper_cache.py - ruff clean (33 fixes: bare excepts, unused Config, whitespace) - Root pyproject.toml (activates shared Gitea CI), MIT LICENSE, README Tests
294 lines
11 KiB
Python
294 lines
11 KiB
Python
"""
|
|
Article processor for handling the processing of articles from the scraper directory.
|
|
Manages batching, processing, and integration with the fact extraction system.
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import time
|
|
from datetime import datetime
|
|
from typing import List, Tuple
|
|
|
|
from cache_manager import CacheManager
|
|
from config import BATCH_SIZE, CACHE_FILE
|
|
from fact_extractor import FactExtractor
|
|
from metrics_collector import metrics_collector
|
|
|
|
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:
|
|
# Debug: Check if directory exists
|
|
logger.info(f"Checking for articles in: {scraper_dir}")
|
|
if not os.path.exists(scraper_dir):
|
|
logger.warning(f"Scraper directory does not exist: {scraper_dir}")
|
|
# Try alternative paths
|
|
alternative_paths = [
|
|
"../scraper/articles",
|
|
"/scraper/articles",
|
|
"/app/articles",
|
|
"/articles",
|
|
]
|
|
for alt_path in alternative_paths:
|
|
if os.path.exists(alt_path):
|
|
logger.info(
|
|
f"Found articles directory at alternative path: {alt_path}"
|
|
)
|
|
scraper_dir = alt_path
|
|
break
|
|
else:
|
|
logger.error("No valid articles directory found")
|
|
return []
|
|
|
|
# Log cache state before scanning
|
|
cache_stats = self.cache_manager.get_cache_stats()
|
|
logger.info(
|
|
f"Cache state before scanning: {cache_stats['processed_files']} files marked as processed"
|
|
)
|
|
|
|
logger.info("Directory exists, walking through files...")
|
|
file_count = 0
|
|
article_file_count = 0
|
|
already_processed_count = 0
|
|
extension_counts = {}
|
|
|
|
for root, dirs, files in os.walk(scraper_dir):
|
|
for file in files:
|
|
file_count += 1
|
|
|
|
# Track file extensions for debugging
|
|
ext = (
|
|
os.path.splitext(file)[1].lower()
|
|
if "." in file
|
|
else "(no extension)"
|
|
)
|
|
extension_counts[ext] = extension_counts.get(ext, 0) + 1
|
|
|
|
# All files in this directory are guaranteed to be article files
|
|
article_file_count += 1
|
|
file_path = os.path.join(root, file)
|
|
if not self.cache_manager.is_processed(file_path):
|
|
unprocessed_articles.append((file_path, file))
|
|
else:
|
|
already_processed_count += 1
|
|
|
|
# Log detailed breakdown
|
|
logger.info(f"File extension breakdown: {extension_counts}")
|
|
logger.info(
|
|
f"Scanned {file_count} total files, {article_file_count} are article files"
|
|
)
|
|
logger.info(
|
|
f"Already processed (in cache): {already_processed_count}, Unprocessed: {len(unprocessed_articles)}"
|
|
)
|
|
|
|
if (
|
|
article_file_count > 0
|
|
and len(unprocessed_articles) == 0
|
|
and already_processed_count == article_file_count
|
|
):
|
|
logger.warning(
|
|
f"All {article_file_count} article files are marked as processed in cache. "
|
|
f"If cache should be empty, check cache file: {self.cache_manager.cache_file}"
|
|
)
|
|
|
|
logger.info(f"Found {len(unprocessed_articles)} unprocessed articles to process")
|
|
return unprocessed_articles
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error finding unprocessed articles: {e}")
|
|
logger.error(f"Error type: {type(e).__name__}")
|
|
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:
|
|
logger.debug(f"Processing article file: {filename}")
|
|
logger.debug(f"File path: {file_path}")
|
|
|
|
with open(file_path, "r", encoding="utf-8") as f:
|
|
article_data = json.load(f)
|
|
|
|
logger.debug(f"Loaded article data for: {filename}")
|
|
logger.debug(f"Article title: {article_data.get('title', 'No title')}")
|
|
logger.debug(f"Article content length: {len(article_data.get('original_content', ''))}")
|
|
|
|
# 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 json.JSONDecodeError as e:
|
|
logger.error(f"JSON decode error processing article {filename}: {e}")
|
|
logger.error(f"File path: {file_path}")
|
|
metrics_collector.increment_articles_failed()
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"Error processing article {filename}: {e}")
|
|
logger.error(f"Error type: {type(e).__name__}")
|
|
logger.error(f"File path: {file_path}")
|
|
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")
|
|
logger.debug(f"Batch contents: {[filename for _, filename in articles_batch]}")
|
|
|
|
for file_path, filename in articles_batch:
|
|
try:
|
|
logger.debug(f"Processing individual article: {filename}")
|
|
facts = self.process_article_file(file_path, filename)
|
|
if facts:
|
|
successful += 1
|
|
metrics_collector.increment_articles_processed()
|
|
logger.debug(f"Successfully processed: {filename}")
|
|
else:
|
|
failed += 1
|
|
metrics_collector.increment_articles_failed()
|
|
logger.warning(f"Failed to process: {filename}")
|
|
except Exception as e:
|
|
logger.error(f"Error processing batch item {filename}: {e}")
|
|
logger.error(f"Error type: {type(e).__name__}")
|
|
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]
|
|
logger.info(f"Processing batch {i//self.batch_size + 1} with {len(batch)} articles")
|
|
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}")
|
|
logger.info(f"Processing stats: {stats}")
|
|
|
|
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
|
|
"""
|
|
logger.info("Starting real-time processing of new articles")
|
|
return self.process_all_articles(scraper_dir)
|