Jarian Cottingham
a7936b8b11
- #12: Remove duplicate logging.basicConfig() from 10 modules - #15: Remove redundant import re in rebuild_database.py - #17: rglob('*') → rglob('*.html/json/txt/xml/md') for speed - #18: Dockerfile individual COPY → glob COPY *.py/*.json + .dockerignore - #19: Remove deprecated docker-compose version field - #20: Pin requirements.txt versions (flask, requests, etc.) - #22: SIGALRM → threading.Timer for multi-threaded safety - #23: AP regex parsing → BeautifulSoup selectors
387 lines
12 KiB
Python
387 lines
12 KiB
Python
#!/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'<title[^>]*>403[^<]*Forbidden</title>',
|
|
r'<title[^>]*>401[^<]*Unauthorized</title>',
|
|
r'<h1[^>]*>403</h1>',
|
|
r'<h1[^>]*>401</h1>',
|
|
]
|
|
|
|
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 '<title>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 "" |