#!/usr/bin/env python3 """Archive Engine for NewsArchiver - Phase 2.5 Orchestrates the archiving process: - RSS polling - Page archiving with SingleFile - Content extraction - Storage management """ import argparse import json import logging import os import sys from datetime import datetime from pathlib import Path from typing import List, Dict, Optional try: import storage_manager from rss_processor import fetch_rss_feed, is_duplicate, save_article as cache_save_article from content_extractor import parse_article_from_html, get_html_from_url from storage_manager import save_article as storage_save_article except ImportError as e: print(f"ERROR: Required module not found: {e}") sys.exit(1) try: from singlefile_archive import archive_page_with_singlefile, archive_page_with_playwright except ImportError: archive_page_with_singlefile = None archive_page_with_playwright = None try: from ap_processor import process_ap_articles except ImportError: process_ap_articles = None try: from storage_manager import get_archive_file_path_from_db except ImportError: get_archive_file_path_from_db = None SCRIPT_DIR = Path(__file__).parent.resolve() ARCHIVE_DIR = Path(os.environ.get("ARCHIVE_DIR", str(SCRIPT_DIR / "archival_data"))).resolve() ARCHIVE_DIR.mkdir(exist_ok=True) logger = logging.getLogger(__name__) def get_archive_file_path(source_name: str, article_url: str, archive_dir: Path = ARCHIVE_DIR) -> Optional[Path]: """Find archived HTML file for an article. First checks the database mapping for the archive file path. Falls back to searching HTML files if not found in database. Args: source_name: Newspaper source name article_url: Article URL archive_dir: Root archive directory Returns: Path to archived HTML file if found, None otherwise """ archive_file_path = get_archive_file_path_from_db(article_url, source_name) if archive_file_path: archive_path = Path(archive_file_path) if archive_path.exists(): logger.debug("Found archived HTML (DB) for %s: %s", article_url[:60], archive_path) return archive_path logger.debug("Archive file not found on disk: %s", archive_file_path) websites_dir = archive_dir / 'websites' if not websites_dir.exists(): return None try: source_dir = websites_dir / source_name if not source_dir.exists(): return None html_dir = source_dir / 'html' if not html_dir.exists(): return None for date_dir in sorted(html_dir.iterdir()): if not date_dir.is_dir(): continue for html_file in sorted(date_dir.glob('article_*.html')): try: with open(html_file, 'r', encoding='utf-8') as f: html_content = f.read() if article_url in html_content: logger.debug("Found archived HTML (search) for %s: %s", article_url[:60], html_file) return html_file except Exception as e: logger.debug("Error reading %s: %s", html_file, str(e)) continue except Exception as e: logger.debug("Error searching for archive: %s", str(e)) return None def extract_content_from_archive(article_url: str, source_name: str = None, archive_dir: Path = ARCHIVE_DIR) -> dict: """Extract content from web page, preferring archived HTML over live fetch. Args: article_url: URL to extract content from source_name: Newspaper source name (for locating archives) archive_dir: Root archive directory Returns: Dictionary with extracted content and metadata """ raw_html = None extraction_method = 'live_fetch' try: if source_name: archived_file = get_archive_file_path(source_name, article_url, archive_dir) if archived_file: try: raw_html = archived_file.read_text(encoding='utf-8') extraction_method = 'archive' logger.debug("Loaded archived HTML from %s", archived_file) except Exception as e: logger.warning("Failed to read archived HTML from %s: %s", archived_file, str(e)) if raw_html is None: raw_html = get_html_from_url(article_url) if not raw_html: logger.error("No HTML content retrieved for %s", article_url[:60]) return { 'success': False, 'error': 'Failed to retrieve HTML content' } article_data = parse_article_from_html(raw_html, article_url) extraction_method = article_data.extraction_method or extraction_method logger.debug("Extracted content from %s using %s", article_url[:60], extraction_method) return { 'success': True, 'article_data': article_data } except Exception as e: logger.error("Error extracting content from %s: %s", article_url, str(e)) return { 'success': False, 'error': str(e) } def archive_and_extract(article_url: str, source_name: str, archive_dir: Path) -> dict: """Archive a URL and extract content from the archive. This function first tries to archive the URL using SingleFile or Playwright. Then it extracts content from the archived HTML. Args: article_url: URL to archive and extract source_name: Newspaper source name (for directory structure) archive_dir: Root archive directory Returns: Dictionary with extraction results """ archived_file = archive_article_url(article_url, source_name, archive_dir) if not archived_file: logger.error("Failed to archive %s", article_url[:60]) return { 'success': False, 'error': 'Failed to archive article' } extraction_result = extract_content_from_archive(article_url, source_name, archive_dir) if extraction_result['success']: logger.info("Successfully archived and extracted content from %s", article_url[:60]) else: logger.error("Failed to extract content from archived %s: %s", article_url[:60], extraction_result.get('error', 'Unknown error')) return extraction_result def archive_url_with_fallback(url: str, output_path: Path) -> bool: """Archive a URL using SingleFile if available, falling back to Playwright. Args: url: URL to archive output_path: Output file path for the archived HTML Returns: True if successful, False otherwise """ if archive_page_with_singlefile: if archive_page_with_singlefile(url, output_path): return True if archive_page_with_playwright: logger.info("Falling back to Playwright archiving for %s", url[:60]) if archive_page_with_playwright(url, output_path): return True return False def archive_article_url(article_url: str, source_name: str, archive_dir: Path) -> Optional[Path]: """Archive a single article URL using SingleFile or Playwright fallback. Args: article_url: URL to archive source_name: Newspaper source name (for directory structure) archive_dir: Root archive directory Returns: Path to archived HTML file if successful, None otherwise """ try: websites_dir = archive_dir / 'websites' source_dir = websites_dir / source_name html_dir = source_dir / 'html' timestamp = datetime.now().strftime('%Y-%m-%d') date_dir = html_dir / timestamp date_dir.mkdir(parents=True, exist_ok=True) base_timestamp = int(datetime.now().timestamp()) counter = 0 article_filename = f'article_{base_timestamp}_{counter}.html' output_path = date_dir / article_filename while output_path.exists(): counter += 1 article_filename = f'article_{base_timestamp}_{counter}.html' output_path = date_dir / article_filename if archive_page_with_singlefile: if archive_page_with_singlefile(article_url, output_path): logger.info("Archived %s to %s", article_url[:60], output_path) return output_path logger.info("SingleFile failed, attempting direct fetch fallback for %s", article_url[:60]) if archive_page_with_playwright: logger.info("Falling back to Playwright for %s", article_url[:60]) if archive_page_with_playwright(article_url, output_path): return output_path logger.info("Attempting direct HTML fetch fallback for %s", article_url[:60]) direct_html = get_html_from_url(article_url) if direct_html: output_path.write_text(direct_html, encoding='utf-8') logger.info("Archived %s to %s using direct fetch", article_url[:60], output_path) return output_path logger.error("Failed to archive %s", article_url[:60]) return None except Exception as e: logger.error("Error archiving %s: %s", article_url, str(e)) return None def archive_newspaper( source_name: str, rss_url: str, output_dir: Path, dry_run: bool = False ) -> dict: """Main archiving workflow for a newspaper. Args: source_name: Newspaper source name rss_url: RSS feed URL output_dir: Output directory for archived content dry_run: If True, preview without making changes Returns: Dictionary with results summary """ storage_manager.initialize_storage() logger.info("=" * 60) logger.info("Starting newspaper archive: %s", source_name) logger.info("=" * 60) results = { 'source': source_name, 'rss_url': rss_url, 'processed': 0, 'archived': 0, 'skipped': 0, 'failed': 0, 'errors': [] } try: feed = fetch_rss_feed(rss_url) except Exception as e: logger.error("Failed to fetch RSS feed for %s: %s", source_name, str(e)) results['errors'].append({ 'action': 'fetch_rss', 'error': str(e) }) return results for item in feed.entries: article_url = item.get('link', '') if not article_url: logger.warning("Skipping entry without URL") results['failed'] += 1 continue if is_duplicate(article_url, source_name, ARCHIVE_DIR / 'cache.db'): logger.debug("Skipping duplicate: %s", article_url[:60]) results['skipped'] += 1 continue results['processed'] += 1 if dry_run: title = item.get('title', 'No Title') logger.info("[DRY-RUN] Would process: %s - %s", title[:60], article_url[:60]) results['archived'] += 1 continue extraction_result = archive_and_extract(article_url, source_name, output_dir) if not extraction_result['success']: logger.error("Failed to extract content from %s: %s", article_url[:60], extraction_result.get('error', 'Unknown error')) results['failed'] += 1 results['errors'].append({ 'url': article_url, 'action': 'extract_content', 'error': extraction_result.get('error', 'Unknown error') }) continue article_data = extraction_result['article_data'] try: storage_save_article(source_name, article_data) results['archived'] += 1 except Exception as e: logger.error("Failed to save article %s: %s", article_url[:60], str(e)) results['failed'] += 1 results['errors'].append({ 'url': article_url, 'action': 'save_article', 'error': str(e) }) logger.info("Archive complete: %s - %d processed, %d archived, %d skipped, %d failed", source_name, results['processed'], results['archived'], results['skipped'], results['failed']) return results def archive_all_sources( rss_feeds_path: Path = SCRIPT_DIR / 'rss_feeds.json', output_dir: Path = ARCHIVE_DIR, dry_run: bool = False ) -> dict: """Process all RSS feeds from rss_feeds.json. Args: rss_feeds_path: Path to RSS feeds JSON file output_dir: Output directory for archived content dry_run: If True, preview without making changes Returns: Dictionary with results summary """ if not rss_feeds_path.exists(): logger.error("RSS feeds file not found: %s", rss_feeds_path) return { 'success': False, 'error': 'File not found', 'processed': 0, 'success_count': 0, 'failed_count': 0 } with open(rss_feeds_path, 'r', encoding='utf-8') as f: rss_feeds = json.load(f) total_results = { 'sources_processed': 0, 'sources_failed': 0, 'total_articles_processed': 0, 'total_articles_archived': 0, 'total_articles_skipped': 0, 'total_articles_failed': 0, 'source_results': [], 'errors': [] } for source_name, feed_info in rss_feeds.items(): rss_url = feed_info.get('rss_url', '') feed_type = feed_info.get('feed_type', 'rss') if not rss_url: logger.warning("No RSS URL for source: %s", source_name) continue try: if feed_type == 'html' and process_ap_articles: ap_results = process_ap_articles( output_dir, ARCHIVE_DIR / 'cache.db', dry_run ) ap_results['source'] = source_name total_results['sources_processed'] += 1 total_results['total_articles_processed'] += ap_results['processed'] total_results['total_articles_archived'] += ap_results['archived'] total_results['total_articles_skipped'] += ap_results['skipped'] total_results['total_articles_failed'] += ap_results['failed'] total_results['source_results'].append(ap_results) if ap_results.get('errors'): total_results['errors'].extend(ap_results['errors']) elif feed_type == 'html' and not process_ap_articles: logger.error("HTML feed type configured but ap_processor unavailable for %s", source_name) total_results['sources_failed'] += 1 total_results['source_results'].append({ 'source': source_name, 'rss_url': rss_url, 'error': 'HTML feed processing not available' }) else: results = archive_newspaper(source_name, rss_url, output_dir, dry_run) total_results['sources_processed'] += 1 total_results['total_articles_processed'] += results['processed'] total_results['total_articles_archived'] += results['archived'] total_results['total_articles_skipped'] += results['skipped'] total_results['total_articles_failed'] += results['failed'] total_results['source_results'].append(results) if results['errors']: total_results['errors'].extend(results['errors']) except Exception as e: logger.error("Failed to process source %s: %s", source_name, str(e)) total_results['sources_failed'] += 1 total_results['source_results'].append({ 'source': source_name, 'rss_url': rss_url, 'error': str(e) }) return total_results def main(): """Main entry point.""" parser = argparse.ArgumentParser(description='Archive Engine for NewsArchiver - Phase 2.5') parser.add_argument('--source', help='Single source name to process') parser.add_argument('--rss-url', help='RSS URL (required if --source provided)') parser.add_argument('--all', action='store_true', help='Process all sources from rss_feeds.json') parser.add_argument('--rss-feeds', type=Path, default=SCRIPT_DIR / 'rss_feeds.json', help='Path to RSS feeds JSON file') parser.add_argument('--output', type=Path, default=ARCHIVE_DIR, help='Output directory for archived content') parser.add_argument('--dry-run', action='store_true', help='Preview without making changes') parser.add_argument('--verbose', action='store_true', help='Enable verbose logging') args = parser.parse_args() if args.verbose: logger.setLevel(logging.DEBUG) logger.info("=" * 60) logger.info("Archive Engine - Phase 2.5") logger.info("=" * 60) storage_manager.initialize_storage() if args.source: if not args.rss_url: logger.error("RSS URL required when using --source") return results = archive_newspaper(args.source, args.rss_url, args.output, args.dry_run) print("\n" + "=" * 60) print(f"SOURCE: {args.source}") print("=" * 60) print(f"Processed: {results['processed']}") print(f"Archived: {results['archived']}") print(f"Skipped: {results['skipped']}") print(f"Failed: {results['failed']}") if results['errors']: print("\nErrors:") for error in results['errors']: print(f" - {error.get('url', 'Unknown')}: {error.get('error', 'Unknown error')}") print("=" * 60) elif args.all: results = archive_all_sources(args.rss_feeds, args.output, args.dry_run) print("\n" + "=" * 60) print("PROCESSING COMPLETE") print("=" * 60) print(f"Sources processed: {results['sources_processed']}") print(f"Sources failed: {results['sources_failed']}") print(f"Total articles processed: {results['total_articles_processed']}") print(f"Total articles archived: {results['total_articles_archived']}") print(f"Total articles skipped: {results['total_articles_skipped']}") print(f"Total articles failed: {results['total_articles_failed']}") if results['errors']: print("\nErrors:") for error in results['errors'][:10]: url = error.get('url', 'Unknown') action = error.get('action', 'Unknown') error_msg = error.get('error', 'Unknown error') print(f" - [{action}] {url}: {error_msg}") if len(results['errors']) > 10: print(f" ... and {len(results['errors']) - 10} more errors") print("=" * 60) else: parser.print_help() if __name__ == '__main__': main()