NewsArchiverV2/rss_processor.py

#!/usr/bin/env python3
"""RSS Feed Processor for NewsArchiver - Phase 2.1"""

import argparse
import json
import logging
import sys
import time
import urllib.error
import urllib.request
from datetime import datetime
from pathlib import Path
from typing import List, Optional

try:
    import feedparser
    from feedparser import FeedParserDict
except ImportError:
    print("ERROR: feedparser is required. Install with: pip install feedparser")
    sys.exit(1)

try:
    import sqlite3
except ImportError:
    print("ERROR: sqlite3 is required (should be built-in)")
    sys.exit(1)

SCRIPT_DIR = Path(__file__).parent
ARCHIVE_DIR = SCRIPT_DIR / 'archival_data'
ARCHIVE_DIR.mkdir(exist_ok=True)

logger = logging.getLogger(__name__)


def init_db(db_path: Path = ARCHIVE_DIR / 'cache.db') -> None:
    """Initialize SQLite database for caching processed articles."""
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS articles (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            source_name TEXT NOT NULL,
            article_url TEXT NOT NULL UNIQUE,
            article_guid TEXT,
            title TEXT,
            author TEXT,
            publish_date TEXT,
            content_text TEXT,
            content_html TEXT,
            archive_file_path TEXT,
            metadata_file_path TEXT,
            status TEXT DEFAULT 'pending',
            error_message TEXT,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    ''')
    
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS processing_log (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            source_name TEXT,
            timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            action TEXT,
            status TEXT,
            message TEXT
        )
    ''')
    
    cursor.execute('''
        CREATE INDEX IF NOT EXISTS idx_articles_source_url 
        ON articles(source_name, article_url)
    ''')
    
    cursor.execute('''
        CREATE INDEX IF NOT EXISTS idx_articles_status 
        ON articles(status)
    ''')
    
    conn.commit()
    conn.close()
    logger.debug("Database initialized: %s", db_path)


def get_db_connection(db_path: Path = ARCHIVE_DIR / 'cache.db'):
    """Get database connection."""
    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row
    return conn


def is_duplicate(article_url: str, source_name: str, db_path: Path = ARCHIVE_DIR / 'cache.db') -> bool:
    """Check if article already in cache."""
    conn = get_db_connection(db_path)
    cursor = conn.cursor()
    
    cursor.execute(
        'SELECT 1 FROM articles WHERE source_name = ? AND article_url = ?',
        (source_name, article_url)
    )
    result = cursor.fetchone()
    conn.close()
    
    return result is not None


def add_to_cache(
    article_url: str,
    source_name: str,
    timestamp: datetime,
    article_guid: str = None,
    title: str = None,
    author: str = None,
    publish_date: str = None,
    db_path: Path = ARCHIVE_DIR / 'cache.db'
) -> bool:
    """Add article to cache."""
    conn = get_db_connection(db_path)
    cursor = conn.cursor()
    
    try:
        cursor.execute('''
            INSERT OR IGNORE INTO articles 
            (source_name, article_url, article_guid, title, author, publish_date, status)
            VALUES (?, ?, ?, ?, ?, ?, ?)
        ''', (
            source_name,
            article_url,
            article_guid,
            title,
            author,
            publish_date,
            'pending'
        ))
        conn.commit()
        conn.close()
        return True
    except sqlite3.IntegrityError:
        conn.close()
        return False


def save_article(
    source_name: str,
    article_data: dict,
    db_path: Path = ARCHIVE_DIR / 'cache.db'
) -> str:
    """Save article to storage and update cache."""
    conn = get_db_connection(db_path)
    cursor = conn.cursor()
    
    article_url = article_data.get('link', '')
    article_guid = article_data.get('id', article_url)
    title = article_data.get('title', '')
    author = article_data.get('author', '')
    publish_date = article_data.get('published', '')
    summary = article_data.get('summary', '')
    
    cursor.execute('''
        INSERT OR REPLACE INTO articles 
        (source_name, article_url, article_guid, title, author, publish_date, content_text, status)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?)
    ''', (
        source_name,
        article_url,
        article_guid,
        title,
        author,
        publish_date,
        summary,
        'archived'
    ))
    
    conn.commit()
    conn.close()
    return article_url


def fetch_rss_feed(rss_url: str, timeout: int = 30) -> FeedParserDict:
    """Fetch and parse RSS feed."""
    logger.info("Fetching RSS feed: %s", rss_url[:50] + "..." if len(rss_url) > 50 else rss_url)
    
    try:
        req = urllib.request.Request(
            rss_url,
            headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'}
        )
        with urllib.request.urlopen(req, timeout=timeout) as response:
            feed_content = response.read()
        
        feed = feedparser.parse(feed_content)
        
        if feed.bozo:
            logger.warning("Feed parsing completed with warnings: %s", feed.bozo)
        
        entry_count = len(feed.entries)
        logger.info("Feed parsed: %d entries", entry_count)
        
        return feed
    except urllib.error.HTTPError as e:
        logger.error("HTTP error fetching RSS feed %s: %s", rss_url, str(e.code))
        raise
    except urllib.error.URLError as e:
        logger.error("URL error fetching RSS feed %s: %s", rss_url, str(e.reason))
        raise
    except Exception as e:
        logger.error("Failed to fetch RSS feed %s: %s", rss_url, str(e))
        raise


def decode_google_news_url(google_url: str, entry: dict = None) -> Optional[str]:
    """Decode Google News encrypted URL to actual article URL.
    
    Google News RSS uses encrypted URLs like:
    https://news.google.com/rss/articles/CBMioAFB... which need to be decoded.
    
    Args:
        google_url: Google News encrypted URL
        entry: Full RSS entry for additional context
        
    Returns:
        Decoded article URL or None if not a Google News URL
    """
    if 'news.google.com' not in google_url:
        return google_url
    
    try:
        import base64
        import urllib.parse
        
        if '/rss/articles/' in google_url:
            parts = google_url.split('/rss/articles/')
            if len(parts) >= 2:
                encoded = parts[1]
                if encoded.startswith('CBM'):
                    encoded = encoded[3:]
                    padding = (4 - len(encoded) % 4) % 4
                    encoded += '=' * padding
                    try:
                        decoded = base64.urlsafe_b64decode(encoded).decode('utf-8')
                        logger.debug("Decoded Google News URL: %s -> %s", google_url[:60], decoded[:60])
                        return decoded
                    except Exception:
                        pass
        
        if '/articles/' in google_url:
            parts = google_url.split('/articles/')
            if len(parts) >= 2:
                encoded = parts[1]
                if encoded.startswith('CBM'):
                    encoded = encoded[3:]
                    padding = (4 - len(encoded) % 4) % 4
                    encoded += '=' * padding
                    try:
                        decoded = base64.urlsafe_b64decode(encoded).decode('utf-8')
                        logger.debug("Decoded Google News URL: %s -> %s", google_url[:60], decoded[:60])
                        return decoded
                    except Exception:
                        pass
        
        if 'url=' in google_url:
            parsed = urllib.parse.urlparse(google_url)
            params = urllib.parse.parse_qs(parsed.query)
            if 'url' in params:
                return params['url'][0]
        
        logger.debug("Could not decode Google News URL: %s", google_url[:60])
        
        if entry and 'source' in entry:
            source = entry.get('source', {})
            if isinstance(source, dict) and 'href' in source:
                source_href = source['href']
                logger.debug("Using source URL as fallback: %s", source_href)
                return source_href
        
        return google_url
        
    except Exception as e:
        logger.debug("Failed to decode Google News URL %s: %s", google_url[:60], str(e))
        return google_url


def process_rss_feed(
    rss_url: str,
    source_name: str,
    output_dir: Path,
    db_path: Path = ARCHIVE_DIR / 'cache.db'
) -> List[dict]:
    """Process RSS feed and archive new articles."""
    logger.info("Processing RSS feed for source: %s", source_name)
    
    try:
        feed = fetch_rss_feed(rss_url)
    except Exception as e:
        logger.error("Failed to fetch feed: %s", str(e))
        return []
    
    new_articles = []
    skipped_count = 0
    
    for entry in feed.entries:
        article_url = entry.get('link', '')
        article_url = decode_google_news_url(article_url, entry)
        article_guid = entry.get('id', article_url)
        
        if not article_url:
            logger.warning("Skipping entry without URL")
            continue
        
        if is_duplicate(article_url, source_name, db_path):
            logger.debug("Skipping duplicate: %s", article_url[:60])
            skipped_count += 1
            continue
        
        title = entry.get('title', 'No Title')
        author = entry.get('author', entry.get('authors', [{}])[0].get('name', '') if entry.get('authors') else '')
        published = entry.get('published', entry.get('published_parsed', ''))
        summary = entry.get('summary', entry.get('description', ''))
        
        if published:
            if hasattr(published, 'tm_year'):
                publish_date = datetime(*published[:6]).isoformat()
            else:
                publish_date = published
        else:
            publish_date = datetime.now().isoformat()
        
        article_data = {
            'source_name': source_name,
            'url': article_url,
            'guid': article_guid,
            'title': title,
            'author': author,
            'publish_date': publish_date,
            'summary': summary,
            'entry': entry
        }
        
        add_to_cache(
            article_url=article_url,
            source_name=source_name,
            timestamp=datetime.now(),
            article_guid=article_guid,
            title=title,
            author=author,
            publish_date=publish_date,
            db_path=db_path
        )
        
        new_articles.append(article_data)
        logger.debug("New article: %s", title[:60])
    
    logger.info("Processed %s: %d new, %d skipped", source_name, len(new_articles), skipped_count)
    return new_articles


def process_all_feeds(
    rss_feeds_path: Path = SCRIPT_DIR / 'rss_feeds.json',
    output_dir: Path = ARCHIVE_DIR,
    db_path: Path = ARCHIVE_DIR / 'cache.db'
) -> dict:
    """Process all RSS feeds from rss_feeds.json."""
    if not rss_feeds_path.exists():
        logger.error("RSS feeds file not found: %s", rss_feeds_path)
        return {'success': False, 'error': 'File not found'}
    
    with open(rss_feeds_path, 'r', encoding='utf-8') as f:
        rss_feeds = json.load(f)
    
    results = {
        'total': 0,
        'success': 0,
        'failed': 0,
        'new_articles': 0,
        'errors': []
    }
    
    for source_name, feed_info in rss_feeds.items():
        rss_url = feed_info.get('rss_url', '')
        
        if not rss_url:
            logger.warning("No RSS URL for source: %s", source_name)
            continue
        
        if feed_info.get('disabled', False):
            logger.info("Skipping disabled source: %s (%s)", source_name, 
                       feed_info.get('disable_reason', 'No reason provided'))
            continue
        
        results['total'] += 1
        
        try:
            articles = process_rss_feed(rss_url, source_name, output_dir, db_path)
            results['success'] += 1
            results['new_articles'] += len(articles)
            logger.info("Completed %s: %d new articles", source_name, len(articles))
        except Exception as e:
            results['failed'] += 1
            results['errors'].append({
                'source': source_name,
                'url': rss_url,
                'error': str(e)
            })
            logger.error("Failed to process %s: %s", source_name, str(e))
    
    return results


def main():
    parser = argparse.ArgumentParser(description='RSS Feed Processor for NewsArchiver')
    parser.add_argument('--url', help='Single RSS URL to process')
    parser.add_argument('--source', help='Source name (required if --url provided)')
    parser.add_argument('--all', action='store_true', help='Process all feeds 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('--verbose', action='store_true', help='Enable verbose logging')
    args = parser.parse_args()
    
    if args.verbose:
        logger.setLevel(logging.DEBUG)
    
    logger.info("=" * 60)
    logger.info("RSS Feed Processor - Phase 2.1")
    logger.info("=" * 60)
    
    init_db()
    
    if args.url:
        if not args.source:
            logger.error("Source name required when using --url")
            return
        process_rss_feed(args.url, args.source, args.output)
    elif args.all:
        results = process_all_feeds(args.rss_feeds, args.output)
        print("\n" + "=" * 60)
        print("PROCESSING COMPLETE")
        print("=" * 60)
        print(f"Total feeds: {results['total']}")
        print(f"Successful: {results['success']}")
        print(f"Failed: {results['failed']}")
        print(f"New articles: {results['new_articles']}")
        if results['errors']:
            print("\nErrors:")
            for error in results['errors']:
                print(f"  - {error['source']}: {error['error']}")
        print("=" * 60)
    else:
        parser.print_help()


if __name__ == '__main__':
    main()