#!/usr/bin/env python3 """Content Extractor for NewsArchiver - Phase 2.2 Extracts article text and metadata from HTML using Trafilatura with BeautifulSoup fallback for JavaScript-heavy sites. """ import logging import sys import time from dataclasses import dataclass from datetime import datetime from pathlib import Path from typing import Optional try: import requests except ImportError: print("WARNING: requests not installed. URL fetching may not work.") print("Install with: pip install requests") requests = None try: import trafilatura from trafilatura import extract, extract_metadata except ImportError: print("ERROR: trafilatura is required. Install with: pip install trafilatura") sys.exit(1) try: from bs4 import BeautifulSoup except ImportError: print("WARNING: beautifulsoup4 not installed. Some fallback features may not work.") print("Install with: pip install beautifulsoup4") BeautifulSoup = None try: from playwright.sync_api import sync_playwright PLAYWRIGHT_AVAILABLE = True except ImportError: PLAYWRIGHT_AVAILABLE = False SCRIPT_DIR = Path(__file__).parent ARCHIVE_DIR = SCRIPT_DIR / 'archival_data' ARCHIVE_DIR.mkdir(exist_ok=True) _last_request_time = 0.0 logger = logging.getLogger(__name__) @dataclass class ArticleData: """Structured article data.""" url: str title: Optional[str] = None author: Optional[str] = None publish_date: Optional[str] = None content_text: Optional[str] = None content_html: Optional[str] = None raw_html: Optional[str] = None archive_file_path: Optional[str] = None tags: Optional[list] = None language: Optional[str] = None metadata: Optional[dict] = None extraction_method: Optional[str] = None error: Optional[str] = None guid: Optional[str] = None id: Optional[int] = None source_name: Optional[str] = None def fetch_url(url: str, timeout: int = 30) -> str: """Download HTML from URL with rate limiting. Args: url: URL to fetch timeout: Request timeout in seconds Returns: HTML string or empty string on failure """ global _last_request_time if requests is None: logger.error("requests library not available") return "" try: delay = 0.5 - (time.time() - _last_request_time) if delay > 0: time.sleep(delay) _last_request_time = time.time() headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36' } response = requests.get(url, timeout=timeout, headers=headers) response.raise_for_status() text = response.text # Check if response is an error page (only check for actual HTTP errors) if response.status_code >= 400: logger.warning("Server returned error for %s (status: %d)", url, response.status_code) return "" # Check for common error patterns if 'access denied' in text.lower() or 'forbidden' in text.lower(): logger.warning("Server returned access denied for %s", url) return "" return text except Exception as e: logger.error("Failed to fetch URL %s: %s", url, str(e)) return "" def extract_content(html: str) -> dict: """Extract article content and metadata from HTML using Trafilatura. Args: html: Raw HTML string (or plain text) Returns: Dictionary with extracted content and metadata """ result = { 'success': False, 'content_text': None, 'content_html': None, 'title': None, 'author': None, 'date': None, 'tags': None, 'language': None, 'metadata': None, 'extraction_method': None, 'error': None } if not html or not html.strip(): result['error'] = 'Empty HTML content' return result # Check if this is an error page (HTTP status codes, not the word "error" in content) # Only check for error patterns that appear in actual error pages (not in article content) # Look for specific error page patterns with proper HTML structure import re error_patterns = [ r']*>403[^<]*Forbidden', r']*>401[^<]*Unauthorized', r']*>403', r']*>401', ] html_lower = html.lower() for pattern in error_patterns: if re.search(pattern, html_lower): result['error'] = 'HTML contains error page content' result['success'] = False return result # Check for common error indicators in the HTML body # These should only trigger if we see them in context (like a 403/401 status indicator) if '403' in html_lower or '<title>401' in html_lower: result['error'] = 'HTML contains error page content' result['success'] = False return result has_html_tags = '<' in html and '>' in html try: metadata = extract_metadata(html) if metadata: result['title'] = metadata.title result['author'] = metadata.author result['date'] = metadata.date result['tags'] = metadata.tags if hasattr(metadata, 'tags') else None result['language'] = metadata.language if hasattr(metadata, 'to_dict'): result['metadata'] = metadata.to_dict() elif hasattr(metadata, '__dict__'): result['metadata'] = metadata.__dict__ content = extract( html, include_comments=False, include_tables=True, no_fallback=True ) if content: result['content_text'] = content result['content_html'] = html result['extraction_method'] = 'trafilatura' result['success'] = True elif not has_html_tags: result['content_text'] = html.strip() result['content_html'] = f'<html><body>{html}</body></html>' result['extraction_method'] = 'plain_text' result['success'] = True else: result['content_text'] = None result['content_html'] = html result['extraction_method'] = 'trafilatura_empty' result['success'] = False result['error'] = 'Content extraction failed - no article content found' if result['success']: logger.debug("Content extracted using %s", result['extraction_method']) except Exception as e: result['error'] = f"Trafilatura extraction failed: {str(e)}" logger.warning(result['error']) if not has_html_tags: result['content_text'] = html.strip() result['content_html'] = f'<html><body>{html}</body></html>' result['extraction_method'] = 'plain_text_fallback' result['success'] = True elif BeautifulSoup: fallback_result = _extract_with_beautifulsoup(html) if fallback_result['content_text']: result.update(fallback_result) result['extraction_method'] = 'beautifulsoup_fallback' return result def _extract_with_beautifulsoup(html: str) -> dict: """Fallback extraction using BeautifulSoup.""" result = { 'content_text': None, 'content_html': None, 'title': None, 'author': None, 'date': None, 'tags': None, 'language': None, 'metadata': None } try: soup = BeautifulSoup(html, 'html.parser') title = soup.find('title') if title: result['title'] = title.get_text(strip=True) meta_author = soup.find('meta', attrs={'name': 'author'}) if meta_author: result['author'] = meta_author.get('content', '').strip() meta_date = soup.find('meta', attrs={'name': 'date'}) if meta_date: result['date'] = meta_date.get('content', '').strip() meta_language = soup.find('meta', attrs={'name': 'language'}) if meta_language: result['language'] = meta_language.get('content', '').strip() for tag in ['article', 'main', 'div']: content_tags = soup.find_all(tag) if content_tags: result['content_text'] = ' '.join( tag.get_text(strip=True, separator=' ') for tag in content_tags ) if result['content_text']: break if not result['content_text']: result['content_text'] = soup.get_text(strip=True, separator=' ') result['content_text'] = result['content_text'][:100000] result['content_html'] = str(soup) except Exception as e: result['error'] = f"BeautifulSoup extraction failed: {str(e)}" logger.warning(result['error']) return result def _extract_with_playwright(url: str) -> Optional[str]: """Extract HTML from JavaScript-heavy site using Playwright.""" if not PLAYWRIGHT_AVAILABLE: logger.warning("Playwright not available. Install with: pip install playwright") return None try: with sync_playwright() as p: browser = p.chromium.launch(headless=True) page = browser.new_page() page.goto(url, wait_until='networkidle', timeout=60000) page_content = page.content() browser.close() logger.debug("HTML extracted using Playwright") return page_content except Exception as e: logger.error("Playwright extraction failed: %s", str(e)) return None def parse_article_from_html(html: str, url: str) -> ArticleData: """Parse article and return structured data. Args: html: Raw HTML string url: Original URL for reference Returns: ArticleData with extracted content and metadata """ extraction_result = extract_content(html) guid = extraction_result.get('metadata', {}).get('url', url) if extraction_result.get('metadata') else url article = ArticleData( url=url, title=extraction_result.get('title'), author=extraction_result.get('author'), publish_date=extraction_result.get('date'), content_text=extraction_result.get('content_text'), content_html=extraction_result.get('content_html'), raw_html=html, tags=extraction_result.get('tags'), language=extraction_result.get('language'), metadata=extraction_result.get('metadata'), extraction_method=extraction_result.get('extraction_method'), guid=guid ) if not extraction_result.get('success'): article.error = extraction_result.get('error') return article def get_html_from_url(url: str, extract_content: bool = True) -> str: """Download HTML from URL using Trafilatura with Playwright fallback. Args: url: URL to fetch extract_content: If True, try to extract main content (kept for backwards compatibility) Returns: Full HTML string or empty string on failure """ try: downloaded = fetch_url(url) if downloaded: logger.debug("Full HTML fetched from %s", url[:60]) return downloaded except Exception as e: logger.error("Failed to fetch URL %s: %s", url, str(e)) if PLAYWRIGHT_AVAILABLE: logger.info("Trying Playwright fallback for %s", url[:60]) playwright_html = _extract_with_playwright(url) if playwright_html: logger.debug("Full HTML from Playwright for %s", url[:60]) if extract_content: extracted = extract( playwright_html, include_comments=False, include_tables=True, no_fallback=True ) if extracted: logger.debug("Playwright content extracted from %s", url[:60]) return extracted logger.debug("Full HTML from Playwright for %s", url[:60]) return playwright_html return ""