Spaces:
Sleeping
Sleeping
File size: 6,835 Bytes
09d041b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 | """Sitemap parsing utilities for extracting URLs from domains."""
import requests
import xml.etree.ElementTree as ET
from typing import List, Optional, Set
from urllib.parse import urljoin
import time
import logging
logger = logging.getLogger(__name__)
def get_urls_from_sitemap(
domain: str,
max_urls: Optional[int] = None,
timeout: int = 10,
max_depth: int = 2
) -> List[str]:
"""
Extract URLs from a domain's sitemap.xml.
Args:
domain: Domain name (e.g., 'google.com' or 'www.google.com')
max_urls: Maximum number of URLs to return (None for all)
timeout: Request timeout in seconds
max_depth: Maximum depth for nested sitemaps (sitemap index files)
Returns:
List of URLs found in the sitemap
"""
urls: Set[str] = set()
# Try common sitemap locations
sitemap_urls = _get_sitemap_urls(domain)
for sitemap_url in sitemap_urls:
try:
logger.info(f"Fetching sitemap: {sitemap_url}")
response = requests.get(sitemap_url, timeout=timeout, headers={
'User-Agent': 'Mozilla/5.0 (compatible; URLCollector/1.0)'
})
if response.status_code == 200:
extracted = _parse_sitemap(
response.content,
max_urls - len(urls) if max_urls else None,
timeout,
max_depth
)
urls.update(extracted)
logger.info(f"Found {len(extracted)} URLs from {sitemap_url}")
if max_urls and len(urls) >= max_urls:
break
else:
logger.debug(f"Failed to fetch {sitemap_url}: {response.status_code}")
except requests.RequestException as e:
logger.debug(f"Error fetching {sitemap_url}: {e}")
continue
except Exception as e:
logger.warning(f"Unexpected error parsing {sitemap_url}: {e}")
continue
result = list(urls)
if max_urls:
result = result[:max_urls]
return result
def _get_sitemap_urls(domain: str) -> List[str]:
"""
Generate possible sitemap URLs for a domain.
Args:
domain: Domain name
Returns:
List of potential sitemap URLs to try
"""
# Remove any protocol if present
domain = domain.replace('http://', '').replace('https://', '').rstrip('/')
# Try both with and without www
domains_to_try = [domain]
if not domain.startswith('www.'):
domains_to_try.append(f'www.{domain}')
else:
domains_to_try.append(domain.replace('www.', '', 1))
sitemap_urls = []
for d in domains_to_try:
# Try HTTPS first, then HTTP
sitemap_urls.extend([
f'https://{d}/sitemap.xml',
f'https://{d}/sitemap_index.xml',
f'https://{d}/sitemap',
f'http://{d}/sitemap.xml',
])
return sitemap_urls
def _parse_sitemap(
content: bytes,
max_urls: Optional[int] = None,
timeout: int = 10,
max_depth: int = 2,
current_depth: int = 0
) -> Set[str]:
"""
Parse sitemap XML content and extract URLs.
Handles both regular sitemaps and sitemap index files.
Args:
content: XML content as bytes
max_urls: Maximum URLs to extract
timeout: Request timeout for nested sitemaps
max_depth: Maximum recursion depth for sitemap indexes
current_depth: Current recursion depth
Returns:
Set of URLs found in the sitemap
"""
urls: Set[str] = set()
try:
root = ET.fromstring(content)
# Define XML namespaces
namespaces = {
'sm': 'http://www.sitemaps.org/schemas/sitemap/0.9',
'image': 'http://www.google.com/schemas/sitemap-image/1.1',
'news': 'http://www.google.com/schemas/sitemap-news/0.9'
}
# Check if this is a sitemap index (contains references to other sitemaps)
sitemap_refs = root.findall('.//sm:sitemap/sm:loc', namespaces)
if sitemap_refs and current_depth < max_depth:
# This is a sitemap index - fetch referenced sitemaps
logger.info(f"Found sitemap index with {len(sitemap_refs)} sitemaps")
for sitemap_loc in sitemap_refs:
if max_urls and len(urls) >= max_urls:
break
sitemap_url = sitemap_loc.text
if sitemap_url:
try:
response = requests.get(sitemap_url, timeout=timeout, headers={
'User-Agent': 'Mozilla/5.0 (compatible; URLCollector/1.0)'
})
if response.status_code == 200:
nested_urls = _parse_sitemap(
response.content,
max_urls - len(urls) if max_urls else None,
timeout,
max_depth,
current_depth + 1
)
urls.update(nested_urls)
time.sleep(0.1) # Small delay to be polite
except Exception as e:
logger.debug(f"Error fetching nested sitemap {sitemap_url}: {e}")
continue
# Extract regular URL entries
url_entries = root.findall('.//sm:url/sm:loc', namespaces)
for loc in url_entries:
if max_urls and len(urls) >= max_urls:
break
if loc.text:
urls.add(loc.text)
except ET.ParseError as e:
logger.warning(f"XML parse error: {e}")
except Exception as e:
logger.warning(f"Error parsing sitemap: {e}")
return urls
def extract_urls_from_domains(
domains: List[str],
max_urls_per_domain: int = 10,
timeout: int = 10,
delay_between_domains: float = 0.5
) -> dict:
"""
Extract URLs from multiple domains using their sitemaps.
Args:
domains: List of domain names
max_urls_per_domain: Maximum URLs to extract per domain
timeout: Request timeout in seconds
delay_between_domains: Delay in seconds between domain requests
Returns:
Dictionary mapping domain to list of extracted URLs
"""
results = {}
for i, domain in enumerate(domains):
logger.info(f"Processing domain {i+1}/{len(domains)}: {domain}")
urls = get_urls_from_sitemap(
domain,
max_urls=max_urls_per_domain,
timeout=timeout
)
results[domain] = urls
if i < len(domains) - 1: # Don't sleep after the last domain
time.sleep(delay_between_domains)
return results |