payman / src /ingestion /url_processor.py
satyamdev404's picture
Upload 31 files
e0aa230 verified
"""
URL Processor Module
This module is responsible for crawling and extracting content from provided URLs,
including nested documents and links with complete web scraping functionality.
Technologies: BeautifulSoup, requests, trafilatura
"""
import logging
import time
import re
from datetime import datetime
from typing import Dict, List, Any, Optional, Set
from urllib.parse import urlparse, urljoin, urlunparse
from urllib.robotparser import RobotFileParser
# Import web scraping libraries
try:
import requests
from bs4 import BeautifulSoup
import trafilatura
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
except ImportError as e:
logging.warning(f"Some web scraping libraries are not installed: {e}")
from utils.error_handler import URLProcessingError, error_handler, ErrorType
class URLProcessor:
"""
Processes URLs to extract content from web pages and linked documents with full functionality.
Features:
- Web page content extraction with trafilatura
- Recursive link following with depth control
- Rate limiting and retry logic
- Robots.txt respect
- Multiple content type handling
"""
def __init__(self, config: Optional[Dict[str, Any]] = None):
"""
Initialize the URLProcessor with configuration.
Args:
config: Configuration dictionary with processing parameters
"""
self.config = config or {}
self.logger = logging.getLogger(__name__)
# Configuration settings
self.max_depth = self.config.get("max_depth", 1)
self.follow_links = self.config.get("follow_links", True)
self.max_pages = self.config.get("max_pages", 10)
self.timeout = self.config.get("timeout", 10)
self.user_agent = self.config.get("user_agent", "RAG-AI-Bot/1.0")
self.respect_robots_txt = self.config.get("respect_robots_txt", True)
self.rate_limit_delay = self.config.get("rate_limit_delay", 1.0)
# Retry configuration
self.max_retries = 3
self.backoff_factor = 0.3
# Track visited URLs and robots.txt cache
self.visited_urls: Set[str] = set()
self.robots_cache: Dict[str, RobotFileParser] = {}
self.last_request_time: Dict[str, float] = {}
# Setup session with retry strategy
self.session = self._setup_session()
def _setup_session(self) -> requests.Session:
"""
Setup requests session with retry strategy and headers.
Returns:
Configured requests session
"""
session = requests.Session()
# Retry strategy
retry_strategy = Retry(
total=self.max_retries,
backoff_factor=self.backoff_factor,
status_forcelist=[429, 500, 502, 503, 504],
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("http://", adapter)
session.mount("https://", adapter)
# 🏷Default headers
session.headers.update(
{
"User-Agent": self.user_agent,
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "en-US,en;q=0.5",
"Accept-Encoding": "gzip, deflate",
"Connection": "keep-alive",
}
)
return session
@error_handler(ErrorType.URL_PROCESSING)
def process_url(self, url: str, depth: int = 0) -> Dict[str, Any]:
"""
Process a URL and extract its content with full functionality.
Args:
url: The URL to process
depth: Current crawling depth
Returns:
Dictionary containing extracted text and metadata
"""
# Validation checks
if not url or not self._is_valid_url(url):
raise URLProcessingError(f"Invalid URL: {url}", url)
if depth > self.max_depth:
self.logger.info(f"🛑 Max depth reached for: {url}")
return {}
if len(self.visited_urls) >= self.max_pages:
self.logger.info(f"🛑 Max pages limit reached")
return {}
if url in self.visited_urls:
self.logger.info(f"Already visited: {url}")
return {}
# Check robots.txt if enabled
if self.respect_robots_txt and not self._can_fetch(url):
self.logger.info(f"Robots.txt disallows: {url}")
return {}
self.visited_urls.add(url)
self.logger.info(f"Processing URL: {url} (depth: {depth})")
try:
# Rate limiting
self._apply_rate_limit(url)
# Fetch and extract content
content, metadata = self._extract_content(url)
if not content:
self.logger.warning(f"No content extracted from: {url}")
return {}
result = {
"content": content,
"metadata": metadata,
"source": url,
"depth": depth,
"linked_documents": [],
"document_type": "webpage",
"crawl_stats": {
"max_depth_configured": self.max_depth,
"follow_links_enabled": self.follow_links,
"current_depth": depth,
},
}
# Follow links if configured and not at max depth
if (
self.follow_links
and depth < self.max_depth
and len(self.visited_urls) < self.max_pages
):
links = self._extract_links(url, content)
self.logger.info(f" Found {len(links)} links on {url}")
for link in links[:5]: # Limit links per page
try:
linked_content = self.process_url(link, depth + 1)
if linked_content:
result["linked_documents"].append(linked_content)
except Exception as e:
self.logger.warning(
f"Failed to process linked URL {link}: {str(e)}"
)
continue
return result
except Exception as e:
raise URLProcessingError(f"Error processing URL: {str(e)}", url)
def process_batch(self, urls: List[str]) -> List[Dict[str, Any]]:
"""
Process multiple URLs in batch.
Args:
urls: List of URLs to process
Returns:
List of processed URL results
"""
results = []
self.logger.info(f"Processing batch of {len(urls)} URLs")
for i, url in enumerate(urls):
try:
result = self.process_url(url)
if result:
results.append(result)
self.logger.info(f"Processed {i+1}/{len(urls)}: {url}")
except Exception as e:
self.logger.error(f"❌ Failed to process {url}: {str(e)}")
continue
return results
def _is_valid_url(self, url: str) -> bool:
"""
Validate URL format and scheme.
Args:
url: URL to validate
Returns:
True if URL is valid
"""
try:
parsed = urlparse(url)
return bool(parsed.netloc) and parsed.scheme in ["http", "https"]
except Exception:
return False
def _can_fetch(self, url: str) -> bool:
"""
Check if URL can be fetched according to robots.txt.
Args:
url: URL to check
Returns:
True if URL can be fetched
"""
try:
parsed_url = urlparse(url)
base_url = f"{parsed_url.scheme}://{parsed_url.netloc}"
if base_url not in self.robots_cache:
robots_url = urljoin(base_url, "/robots.txt")
rp = RobotFileParser()
rp.set_url(robots_url)
try:
rp.read()
self.robots_cache[base_url] = rp
except Exception:
# If robots.txt can't be fetched, assume allowed
return True
return self.robots_cache[base_url].can_fetch(self.user_agent, url)
except Exception:
# If robots.txt check fails, assume allowed
return True
def _apply_rate_limit(self, url: str) -> None:
"""
Apply rate limiting between requests to the same domain.
Args:
url: URL being processed
"""
domain = urlparse(url).netloc
current_time = time.time()
if domain in self.last_request_time:
time_since_last = current_time - self.last_request_time[domain]
if time_since_last < self.rate_limit_delay:
sleep_time = self.rate_limit_delay - time_since_last
self.logger.info(
f"Rate limiting: sleeping {sleep_time:.1f}s for {domain}"
)
time.sleep(sleep_time)
self.last_request_time[domain] = time.time()
def _extract_content(self, url: str) -> tuple:
"""
Extract content from a web page using trafilatura with BeautifulSoup fallback.
Args:
url: The URL to extract content from
Returns:
Tuple of (content, metadata)
"""
self.logger.info(f"Extracting content from: {url}")
try:
# Fetch the page
response = self.session.get(url, timeout=self.timeout)
response.raise_for_status()
# Basic metadata
metadata = {
"url": url,
"status_code": response.status_code,
"content_type": response.headers.get("content-type", ""),
"content_length": len(response.content),
"extracted_time": datetime.now().isoformat(),
"encoding": response.encoding or "utf-8",
}
# Check content type
content_type = response.headers.get("content-type", "").lower()
if "application/pdf" in content_type:
return self._handle_pdf_url(response, metadata)
elif "text/html" not in content_type and "text/plain" not in content_type:
self.logger.warning(f"Unsupported content type: {content_type}")
return "", metadata
# Primary method: trafilatura (best for content extraction)
try:
content = trafilatura.extract(
response.text,
include_comments=False,
include_tables=True,
include_formatting=False,
favor_precision=True,
)
if content and len(content.strip()) > 50: # Minimum content threshold
# Extract additional metadata with trafilatura
metadata_extracted = trafilatura.extract_metadata(response.text)
if metadata_extracted:
metadata.update(
{
"title": metadata_extracted.title or "",
"author": metadata_extracted.author or "",
"description": metadata_extracted.description or "",
"sitename": metadata_extracted.sitename or "",
"date": metadata_extracted.date or "",
}
)
metadata.update(
{
"extraction_method": "trafilatura",
"word_count": len(content.split()),
"character_count": len(content),
}
)
return content.strip(), metadata
except Exception as e:
self.logger.warning(f"Trafilatura failed: {str(e)}")
# Fallback method: BeautifulSoup
return self._extract_with_beautifulsoup(response.text, metadata)
except requests.RequestException as e:
raise URLProcessingError(f"Failed to fetch URL: {str(e)}", url)
except Exception as e:
raise URLProcessingError(f"Content extraction failed: {str(e)}", url)
def _extract_with_beautifulsoup(self, html: str, metadata: Dict[str, Any]) -> tuple:
"""
Fallback content extraction using BeautifulSoup.
Args:
html: HTML content
metadata: Existing metadata dictionary
Returns:
Tuple of (content, metadata)
"""
try:
soup = BeautifulSoup(html, "html.parser")
# Extract metadata
title_tag = soup.find("title")
if title_tag:
metadata["title"] = title_tag.get_text().strip()
# Meta tags
for meta in soup.find_all("meta"):
name = meta.get("name", "").lower()
content = meta.get("content", "")
if name == "description":
metadata["description"] = content
elif name == "author":
metadata["author"] = content
# Remove unwanted elements
for element in soup(
["script", "style", "nav", "header", "footer", "aside"]
):
element.decompose()
# Extract main content
content_selectors = [
"main",
"article",
".content",
"#content",
".post",
".entry",
]
content = ""
for selector in content_selectors:
content_elem = soup.select_one(selector)
if content_elem:
content = content_elem.get_text(separator="\n", strip=True)
break
# Fallback to body if no main content found
if not content:
body = soup.find("body")
if body:
content = body.get_text(separator="\n", strip=True)
# Clean and validate content
content = re.sub(r"\n\s*\n", "\n\n", content) # Clean multiple newlines
content = content.strip()
metadata.update(
{
"extraction_method": "beautifulsoup",
"word_count": len(content.split()),
"character_count": len(content),
}
)
return content, metadata
except Exception as e:
self.logger.error(f"❌ BeautifulSoup extraction failed: {str(e)}")
return "", metadata
def _handle_pdf_url(
self, response: requests.Response, metadata: Dict[str, Any]
) -> tuple:
"""
📄 Handle PDF content from URL.
Args:
response: HTTP response containing PDF
metadata: Existing metadata
Returns:
Tuple of (content, metadata)
"""
self.logger.info("📄 Detected PDF content, extracting text...")
try:
# Save PDF temporarily and process with document processor
import tempfile
import os
from .document_processor import DocumentProcessor
with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as tmp_file:
tmp_file.write(response.content)
tmp_file.flush()
# Process PDF
doc_processor = DocumentProcessor(self.config)
result = doc_processor.process_document(tmp_file.name)
# Cleanup
os.unlink(tmp_file.name)
metadata.update(
{
"document_type": "pdf_from_url",
"extraction_method": "document_processor",
}
)
metadata.update(result.get("metadata", {}))
return result.get("content", ""), metadata
except Exception as e:
self.logger.error(f"❌ PDF extraction failed: {str(e)}")
return "", metadata
def _extract_links(self, url: str, content: str) -> List[str]:
"""
Extract links from a web page.
Args:
url: The source URL
content: Page content (for context)
Returns:
List of discovered URLs
"""
self.logger.info(f" Extracting links from: {url}")
try:
response = self.session.get(url, timeout=self.timeout)
soup = BeautifulSoup(response.text, "html.parser")
links = []
base_domain = urlparse(url).netloc
for a_tag in soup.find_all("a", href=True):
href = a_tag.get("href")
if not href:
continue
# Convert relative URLs to absolute
absolute_url = urljoin(url, href)
# Filter links
if self._should_follow_link(absolute_url, base_domain):
links.append(absolute_url)
# 🎯 Remove duplicates and limit
unique_links = list(dict.fromkeys(links)) # Preserve order
return unique_links[:20] # Limit to prevent explosion
except Exception as e:
self.logger.error(f"❌ Link extraction failed: {str(e)}")
return []
def _should_follow_link(self, url: str, base_domain: str) -> bool:
"""
Determine if a link should be followed.
Args:
url: URL to check
base_domain: Base domain of the source page
Returns:
True if link should be followed
"""
try:
parsed = urlparse(url)
# Skip non-HTTP(S) links
if parsed.scheme not in ["http", "https"]:
return False
# Skip already visited
if url in self.visited_urls:
return False
# Skip file downloads (basic check)
path = parsed.path.lower()
skip_extensions = [
".pdf",
".doc",
".docx",
".zip",
".exe",
".dmg",
".jpg",
".png",
".gif",
]
if any(path.endswith(ext) for ext in skip_extensions):
return False
# Skip fragments and query-heavy URLs
if parsed.fragment or len(parsed.query) > 100:
return False
# Prefer same domain (but allow subdomains)
link_domain = parsed.netloc
if not (
link_domain == base_domain or link_domain.endswith("." + base_domain)
):
return False
return True
except Exception:
return False
def reset(self):
"""Reset the processor state, clearing visited URLs and caches."""
self.visited_urls.clear()
self.robots_cache.clear()
self.last_request_time.clear()
self.logger.info("URL processor state reset")
def get_statistics(self) -> Dict[str, Any]:
"""
Get processing statistics.
Returns:
Dictionary with processing statistics
"""
return {
"urls_processed": len(self.visited_urls),
"domains_cached": len(self.robots_cache),
"rate_limited_domains": len(self.last_request_time),
"max_pages_limit": self.max_pages,
"max_depth_limit": self.max_depth,
}