This is a bit of code I have developed to use with the Crawl4ai python package (GitHub - unclecode/crawl4ai: 🚀🤖 Crawl4AI: Open-source LLM Friendly Web Crawler & Scraper). It works well for crawling sitemaps.xml, just give it the link to the sitemap you want to crawl.
You can get any sites sitemap.xml by looking in the robots.txt file (Example: cnn.com/robots.txt). At some point I'll dump this on Github but wanted to share sooner than later. Use at your own risk.
✅ Shows progress: X/Y URLs completed
✅ Retries failed URLs only once
✅ Logs failed URLs separately
✅ Writes clean Markdown output
✅ Respects request delays
✅ Logs failed URLs to logfile.txt
✅ Streams results into multiple files (max 20MB each, this is the file limit for uploads to chatgpt)
Change these values in the code below to fit your needs.
SITEMAP_URL = "https://www.cnn.com/sitemap.xml" # Change this to your sitemap URL
MAX_DEPTH = 10 # Limit recursion depth
BATCH_SIZE = 1 # Number of concurrent crawls
REQUEST_DELAY = 1 # Delay between requests (seconds)
MAX_FILE_SIZE_MB = 20 # Max file size before creating a new one
OUTPUT_DIR = "cnn" # Directory to store multiple output files
RETRY_LIMIT = 1 # Retry failed URLs once
LOG_FILE = os.path.join(OUTPUT_DIR, "crawler_log.txt") # Log file for general logging
ERROR_LOG_FILE = os.path.join(OUTPUT_DIR, "logfile.txt") # Log file for failed URLs
import asyncio
import json
import os
import xml.etree.ElementTree as ET
from urllib.parse import urljoin, urlparse
import aiohttp
from aiofiles import open as aio_open
from crawl4ai import AsyncWebCrawler, CrawlerRunConfig, CacheMode
from crawl4ai.content_filter_strategy import PruningContentFilter
from crawl4ai.markdown_generation_strategy import DefaultMarkdownGenerator
# Configuration
SITEMAP_URL = "https://www.cnn.com/sitemap.xml" # Change this to your sitemap URL
MAX_DEPTH = 10 # Limit recursion depth
BATCH_SIZE = 1 # Number of concurrent crawls
REQUEST_DELAY = 1 # Delay between requests (seconds)
MAX_FILE_SIZE_MB = 20 # Max file size before creating a new one
OUTPUT_DIR = "cnn" # Directory to store multiple output files
RETRY_LIMIT = 1 # Retry failed URLs once
LOG_FILE = os.path.join(OUTPUT_DIR, "crawler_log.txt") # Log file for general logging
ERROR_LOG_FILE = os.path.join(OUTPUT_DIR, "logfile.txt") # Log file for failed URLs
# Ensure output directory exists
os.makedirs(OUTPUT_DIR, exist_ok=True)
async def log_message(message, file_path=LOG_FILE):
"""Log messages to a log file and print them to the console."""
async with aio_open(file_path, "a", encoding="utf-8") as f:
await f.write(message + "\n")
print(message)
async def fetch_sitemap(sitemap_url):
"""Fetch and parse sitemap.xml to extract all URLs."""
try:
async with aiohttp.ClientSession() as session:
async with session.get(sitemap_url) as response:
if response.status == 200:
xml_content = await response.text()
root = ET.fromstring(xml_content)
urls = [elem.text for elem in root.findall(".//{http://www.sitemaps.org/schemas/sitemap/0.9}loc")]
if not urls:
await log_message("❌ No URLs found in the sitemap.")
return urls
else:
await log_message(f"❌ Failed to fetch sitemap: HTTP {response.status}")
return []
except Exception as e:
await log_message(f"❌ Error fetching sitemap: {str(e)}")
return []
async def get_file_size(file_path):
"""Returns the file size in MB."""
if os.path.exists(file_path):
return os.path.getsize(file_path) / (1024 * 1024) # Convert bytes to MB
return 0
async def get_new_file_path(file_prefix, extension):
"""Generates a new file path when the current file exceeds the max size."""
index = 1
while True:
file_path = os.path.join(OUTPUT_DIR, f"{file_prefix}_{index}.{extension}")
if not os.path.exists(file_path) or await get_file_size(file_path) < MAX_FILE_SIZE_MB:
return file_path
index += 1
async def write_to_file(data, file_prefix, extension):
"""Writes a single JSON object as a line to a file, ensuring size limit."""
file_path = await get_new_file_path(file_prefix, extension)
async with aio_open(file_path, "a", encoding="utf-8") as f:
await f.write(json.dumps(data, ensure_ascii=False) + "\n")
async def write_to_txt(data, file_prefix):
"""Writes extracted content to a TXT file while managing file size."""
file_path = await get_new_file_path(file_prefix, "txt")
async with aio_open(file_path, "a", encoding="utf-8") as f:
await f.write(f"URL: {data['url']}\nTitle: {data['title']}\nContent:\n{data['content']}\n\n{'='*80}\n\n")
async def write_failed_url(url):
"""Logs failed URLs to a separate error log file."""
async with aio_open(ERROR_LOG_FILE, "a", encoding="utf-8") as f:
await f.write(url + "\n")
async def crawl_url(url, depth, semaphore, visited_urls, queue, total_urls, completed_urls, retry_count=0):
"""Crawls a single URL, handles retries, logs failed URLs, and extracts child links."""
async with semaphore:
await asyncio.sleep(REQUEST_DELAY) # Rate limiting
run_config = CrawlerRunConfig(
cache_mode=CacheMode.BYPASS,
markdown_generator=DefaultMarkdownGenerator(
content_filter=PruningContentFilter(threshold=0.5, threshold_type="fixed")
),
stream=True,
remove_overlay_elements=True,
exclude_social_media_links=True,
process_iframes=True,
)
async with AsyncWebCrawler() as crawler:
try:
result = await crawler.arun(url=url, config=run_config)
if result.success:
data = {
"url": result.url,
"title": result.markdown_v2.raw_markdown.split("\n")[0] if result.markdown_v2.raw_markdown else "No Title",
"content": result.markdown_v2.fit_markdown,
}
# Save extracted data
await write_to_file(data, "sitemap_data", "jsonl")
await write_to_txt(data, "sitemap_data")
completed_urls[0] += 1 # Increment completed count
await log_message(f"✅ {completed_urls[0]}/{total_urls} - Successfully crawled: {url}")
# Extract and queue child pages
for link in result.links.get("internal", []):
href = link["href"]
absolute_url = urljoin(url, href) # Convert to absolute URL
if absolute_url not in visited_urls:
queue.append((absolute_url, depth + 1))
else:
await log_message(f"⚠️ Failed to extract content from: {url}")
except Exception as e:
if retry_count < RETRY_LIMIT:
await log_message(f"🔄 Retrying {url} (Attempt {retry_count + 1}/{RETRY_LIMIT}) due to error: {str(e)}")
await crawl_url(url, depth, semaphore, visited_urls, queue, total_urls, completed_urls, retry_count + 1)
else:
await log_message(f"❌ Skipping {url} after {RETRY_LIMIT} failed attempts.")
await write_failed_url(url)
async def crawl_sitemap_urls(urls, max_depth=MAX_DEPTH, batch_size=BATCH_SIZE):
"""Crawls all URLs from the sitemap and follows child links up to max depth."""
if not urls:
await log_message("❌ No URLs to crawl. Exiting.")
return
total_urls = len(urls) # Total number of URLs to process
completed_urls = [0] # Mutable count of completed URLs
visited_urls = set()
queue = [(url, 0) for url in urls]
semaphore = asyncio.Semaphore(batch_size) # Concurrency control
while queue:
tasks = []
batch = queue[:batch_size]
queue = queue[batch_size:]
for url, depth in batch:
if url in visited_urls or depth >= max_depth:
continue
visited_urls.add(url)
tasks.append(crawl_url(url, depth, semaphore, visited_urls, queue, total_urls, completed_urls))
await asyncio.gather(*tasks)
async def main():
# Clear previous logs
async with aio_open(LOG_FILE, "w") as f:
await f.write("")
async with aio_open(ERROR_LOG_FILE, "w") as f:
await f.write("")
# Fetch URLs from the sitemap
urls = await fetch_sitemap(SITEMAP_URL)
if not urls:
await log_message("❌ Exiting: No valid URLs found in the sitemap.")
return
await log_message(f"✅ Found {len(urls)} pages in the sitemap. Starting crawl...")
# Start crawling
await crawl_sitemap_urls(urls)
await log_message(f"✅ Crawling complete! Files stored in {OUTPUT_DIR}")
# Execute
asyncio.run(main())
