philandClaude
25086fc01b
- Modular architecture with separate modules for scraping, parsing, security, validation, and caching - Comprehensive security measures including HTML sanitization, rate limiting, and input validation - Robust error handling with custom exceptions and retry logic - HTTP caching with ETags and Last-Modified headers for efficiency - Pre-compiled regex patterns for improved performance - Comprehensive test suite with 66 tests covering all major functionality - Docker support for containerized deployment - Configuration management with environment variable support - Working parser that successfully extracts 32 articles from Warhammer Community 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
112 lines
4.0 KiB
Python
112 lines
4.0 KiB
Python
"""Web scraping functionality using Playwright."""
|
|
|
|
import time
|
|
import logging
|
|
from playwright.sync_api import sync_playwright
|
|
from typing import Optional
|
|
|
|
from .config import Config
|
|
from .exceptions import NetworkError, PageLoadError, ContentSizeError
|
|
from .retry_utils import retry_on_exception, PLAYWRIGHT_RETRY_CONFIG
|
|
from .cache import ContentCache
|
|
from .security import wait_for_rate_limit
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Global cache instance
|
|
_cache = ContentCache()
|
|
|
|
|
|
def load_page_with_retry(url: str, use_cache: bool = True) -> str:
|
|
"""Load page content with caching and retry logic for network errors."""
|
|
logger.info(f"Loading page: {url}")
|
|
|
|
# Check cache first if enabled
|
|
if use_cache:
|
|
# Check if content has changed using conditional requests
|
|
change_check = _cache.check_if_content_changed(url)
|
|
if change_check and change_check['status'] == 'not_modified':
|
|
cached_content = _cache.get_cached_content(url)
|
|
if cached_content:
|
|
logger.info("Using cached content (not modified)")
|
|
return cached_content
|
|
|
|
# Check for valid cached content
|
|
cached_content = _cache.get_cached_content(url)
|
|
if cached_content:
|
|
logger.info("Using cached content")
|
|
return cached_content
|
|
|
|
# Load fresh content
|
|
html = _load_page_fresh(url)
|
|
|
|
# Cache the content if caching is enabled
|
|
if use_cache:
|
|
_cache.cache_content(url, html)
|
|
|
|
return html
|
|
|
|
|
|
@retry_on_exception((NetworkError, PageLoadError), PLAYWRIGHT_RETRY_CONFIG)
|
|
def _load_page_fresh(url: str) -> str:
|
|
"""Load fresh page content using Playwright."""
|
|
logger.info(f"Loading fresh content from: {url}")
|
|
|
|
# Apply rate limiting before making request
|
|
wait_for_rate_limit()
|
|
|
|
try:
|
|
with sync_playwright() as p:
|
|
browser = p.chromium.launch(headless=True)
|
|
page = browser.new_page()
|
|
|
|
# Set a longer timeout for loading the page
|
|
page.set_default_navigation_timeout(Config.PAGE_TIMEOUT_MS)
|
|
|
|
try:
|
|
# Load the page
|
|
page.goto(url, wait_until="networkidle")
|
|
|
|
# Simulate scrolling to load more content
|
|
logger.info(f"Scrolling page {Config.MAX_SCROLL_ITERATIONS} times to load content")
|
|
for i in range(Config.MAX_SCROLL_ITERATIONS):
|
|
logger.debug(f"Scroll iteration {i + 1}/{Config.MAX_SCROLL_ITERATIONS}")
|
|
page.evaluate("window.scrollBy(0, document.body.scrollHeight)")
|
|
time.sleep(Config.SCROLL_DELAY_SECONDS)
|
|
|
|
# Get the fully rendered HTML content
|
|
html = page.content()
|
|
|
|
# Check content size for security
|
|
if len(html) > Config.MAX_CONTENT_SIZE:
|
|
error_msg = f"Content size {len(html)} exceeds maximum {Config.MAX_CONTENT_SIZE}"
|
|
logger.error(error_msg)
|
|
raise ContentSizeError(error_msg)
|
|
|
|
logger.info(f"Page loaded successfully, content size: {len(html)} bytes")
|
|
return html
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to load page content: {e}")
|
|
if "timeout" in str(e).lower() or "network" in str(e).lower():
|
|
raise NetworkError(f"Network error loading page: {e}")
|
|
else:
|
|
raise PageLoadError(f"Page load error: {e}")
|
|
finally:
|
|
browser.close()
|
|
|
|
except Exception as e:
|
|
if isinstance(e, (NetworkError, PageLoadError, ContentSizeError)):
|
|
raise
|
|
logger.error(f"Unexpected error in Playwright: {e}")
|
|
raise PageLoadError(f"Playwright error: {e}")
|
|
|
|
|
|
def clear_cache() -> None:
|
|
"""Clear the content cache."""
|
|
_cache.clear_cache()
|
|
|
|
|
|
def get_cache_info() -> dict:
|
|
"""Get information about the cache."""
|
|
return _cache.get_cache_info() |