crawl4ai_documentation_downloader

import asyncio
import os
import logging
import xml.etree.ElementTree as ET
import requests
from typing import Dict, List, Optional
from crawl4ai import AsyncWebCrawler, BrowserConfig, CrawlerRunConfig, CacheMode
from crawl4ai.markdown_generation_strategy import DefaultMarkdownGenerator

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler("crawl_log.txt"),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

# Define URL to filename mapping
URL_TO_FILENAME_MAP = {
    "https://docs.crawl4ai.com/": "00_Home_00_Home.md",
    "https://docs.crawl4ai.com/basic/installation/": "01_Setup & Installation_01_Installation.md",
    "https://docs.crawl4ai.com/core/docker-deployment/": "01_Setup & Installation_02_Docker Deployment.md",
    "https://docs.crawl4ai.com/core/quickstart/": "02_Quick Start_01_Quick Start.md",
    "https://docs.crawl4ai.com/blog/": "03_Blog & Changelog_01_Blog Home.md",
    "https://docs.crawl4ai.com/blog/releases/0.4.0/": "03_Blog & Changelog_02_Changelog v0.4.0.md",
    "https://docs.crawl4ai.com/blog/releases/0.4.1/": "03_Blog & Changelog_03_Changelog v0.4.1.md",
    "https://docs.crawl4ai.com/blog/releases/0.4.2/": "03_Blog & Changelog_04_Changelog v0.4.2.md",
    "https://docs.crawl4ai.com/blog/releases/0.5.0/": "03_Blog & Changelog_05_Changelog v0.5.0.md",
    "https://docs.crawl4ai.com/blog/releases/v0.4.3b1/": "03_Blog & Changelog_06_Changelog v0.4.3b1.md",
    "https://docs.crawl4ai.com/core/cli/": "04_Core_01_Command Line Interface.md",
    "https://docs.crawl4ai.com/core/deep-crawling/": "04_Core_02_Deep Crawling.md",
    "https://docs.crawl4ai.com/core/simple-crawling/": "04_Core_03_Simple Crawling.md",
    "https://docs.crawl4ai.com/core/crawler-result/": "04_Core_04_Crawler Result.md",
    "https://docs.crawl4ai.com/core/browser-crawler-config/": "04_Core_05_Browser & Crawler Config.md",
    "https://docs.crawl4ai.com/core/markdown-generation/": "04_Core_06_Markdown Generation.md",
    "https://docs.crawl4ai.com/core/fit-markdown/": "04_Core_07_Fit Markdown.md",
    "https://docs.crawl4ai.com/core/page-interaction/": "04_Core_08_Page Interaction.md",
    "https://docs.crawl4ai.com/core/content-selection/": "04_Core_09_Content Selection.md",
    "https://docs.crawl4ai.com/core/cache-modes/": "04_Core_10_Cache Modes.md",
    "https://docs.crawl4ai.com/core/local-files/": "04_Core_11_Local Files.md",
    "https://docs.crawl4ai.com/core/link-media/": "04_Core_12_Link & Media.md",
    "https://docs.crawl4ai.com/advanced/advanced-features/": "05_Advanced_01_Overview.md",
    "https://docs.crawl4ai.com/advanced/file-downloading/": "05_Advanced_02_File Downloading.md",
    "https://docs.crawl4ai.com/advanced/lazy-loading/": "05_Advanced_03_Lazy Loading.md",
    "https://docs.crawl4ai.com/advanced/hooks-auth/": "05_Advanced_04_Hooks & Auth.md",
    "https://docs.crawl4ai.com/advanced/proxy-security/": "05_Advanced_05_Proxy & Security.md",
    "https://docs.crawl4ai.com/advanced/session-management/": "05_Advanced_06_Session Management.md",
    "https://docs.crawl4ai.com/advanced/multi-url-crawling/": "05_Advanced_07_Multi-URL Crawling.md",
    "https://docs.crawl4ai.com/advanced/crawl-dispatcher/": "05_Advanced_08_Crawl Dispatcher.md",
    "https://docs.crawl4ai.com/advanced/identity-based-crawling/": "05_Advanced_09_Identity Based Crawling.md",
    "https://docs.crawl4ai.com/advanced/ssl-certificate/": "05_Advanced_10_SSL Certificate.md",
    "https://docs.crawl4ai.com/extraction/no-llm-strategies/": "06_Extraction_01_No LLM Strategies.md",
    "https://docs.crawl4ai.com/extraction/llm-strategies/": "06_Extraction_02_LLM Strategies.md",
    "https://docs.crawl4ai.com/extraction/clustring-strategies/": "06_Extraction_03_Clustering Strategies.md",
    "https://docs.crawl4ai.com/extraction/chunking/": "06_Extraction_04_Chunking.md",
    "https://docs.crawl4ai.com/api/async-webcrawler/": "07_API Reference_01_AsyncWebCrawler.md",
    "https://docs.crawl4ai.com/api/arun/": "07_API Reference_02_arun.md",
    "https://docs.crawl4ai.com/api/arun_many/": "07_API Reference_03_arun_many.md",
    "https://docs.crawl4ai.com/api/parameters/": "07_API Reference_04_Parameters.md",
    "https://docs.crawl4ai.com/api/crawl-result/": "07_API Reference_05_CrawlResult.md",
    "https://docs.crawl4ai.com/api/strategies/": "07_API Reference_06_Strategies.md",
}

def get_markdown_content(result) -> Optional[str]:
    """Extract markdown content from the crawl result, handling different versions/formats."""
    # Try different approaches to get markdown content

    # 1. Direct markdown string
    if hasattr(result, 'markdown') and isinstance(result.markdown, str):
        return result.markdown

    # 2. MarkdownGenerationResult in result.markdown
    if hasattr(result, 'markdown') and hasattr(result.markdown, 'raw_markdown'):
        return result.markdown.raw_markdown

    # 3. MarkdownGenerationResult in result.markdown_v2
    if hasattr(result, 'markdown_v2') and hasattr(result.markdown_v2, 'raw_markdown'):
        return result.markdown_v2.raw_markdown

    # 4. Try fit_markdown if raw_markdown is not available
    if hasattr(result, 'markdown_v2') and hasattr(result.markdown_v2, 'fit_markdown'):
        return result.markdown_v2.fit_markdown

    # 5. Last resort: use the HTML content if no markdown was generated
    if result.html:
        logger.warning(f"Using raw HTML for {result.url} as markdown was not available")
        return f"# Content from {result.url}\n\n(Raw HTML was converted to plain text)\n\n{result.cleaned_html or result.html}"

    # No content found
    return None

async def download_sitemap(sitemap_url: str) -> List[str]:
    """Download the sitemap.xml and extract all URLs."""
    try:
        # Request with appropriate headers
        headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
        }

        logger.info(f"Downloading sitemap from {sitemap_url}")
        response = requests.get(sitemap_url, headers=headers, timeout=30)
        response.raise_for_status()

        # Parse the XML
        root = ET.fromstring(response.content)

        # Extract URLs (handle namespaces in XML)
        urls = []

        # Try different namespace approaches (sitemap XML can vary)
        namespaces = {
            'default': 'http://www.sitemaps.org/schemas/sitemap/0.9',
            'ns': 'http://www.sitemaps.org/schemas/sitemap/0.9'
        }

        for prefix, uri in namespaces.items():
            if prefix == 'default':
                xpath = './/loc'
            else:
                xpath = './/{' + uri + '}loc'

            locations = root.findall(xpath)
            if locations:
                urls = [loc.text for loc in locations]
                break

        if not urls:
            # Fallback: try simple parsing without namespaces
            urls = [element.text for element in root.findall(".//loc")]

        logger.info(f"Successfully extracted {len(urls)} URLs from sitemap")
        return urls
    except requests.RequestException as e:
        logger.error(f"Failed to download sitemap: {e}")
        raise
    except ET.ParseError as e:
        logger.error(f"Failed to parse sitemap XML: {e}")
        raise

def get_filename_for_url(url: str) -> Optional[str]:
    """Get the filename for a given URL based on the mapping."""
    # Remove trailing slash for consistency
    url_normalized = url.rstrip('/')

    # Direct lookup from the mapping
    for map_url, filename in URL_TO_FILENAME_MAP.items():
        map_url_normalized = map_url.rstrip('/')
        if url_normalized == map_url_normalized:
            return filename

    logger.warning(f"No filename mapping found for URL: {url}")
    return None

async def crawl_and_save_page(url: str, output_dir: str, crawler: AsyncWebCrawler, max_retries: int = 3) -> None:
    """Crawl a single page and save it as markdown."""
    filename = get_filename_for_url(url)
    if not filename:
        logger.warning(f"Skipping URL with no filename mapping: {url}")
        return

    output_path = os.path.join(output_dir, filename)

    for attempt in range(max_retries):
        try:
            logger.info(f"Crawling: {url} (attempt {attempt+1}/{max_retries})")

            # Configure markdown generator
            md_generator = DefaultMarkdownGenerator(
                options={"ignore_links": False, "body_width": 0}
            )

            config = CrawlerRunConfig(
                cache_mode=CacheMode.BYPASS,  # Always get fresh content
                page_timeout=60000,  # 60 seconds timeout
                word_count_threshold=5,  # Include most content blocks
                excluded_tags=['script', 'style', 'svg'],  # Exclude non-content tags
                markdown_generator=md_generator
            )

            result = await crawler.arun(url=url, config=config)

            if not result.success:
                logger.error(f"Failed to crawl {url}: {result.error_message}")
                if attempt < max_retries - 1:
                    logger.info(f"Retrying {url} in 3 seconds...")
                    await asyncio.sleep(3)  # Wait before retry
                    continue
                return

            # Get markdown content using our helper function
            markdown_content = get_markdown_content(result)

            if markdown_content:
                with open(output_path, 'w', encoding='utf-8') as f:
                    f.write(markdown_content)
                logger.info(f"Saved: {output_path}")
                break  # Success, exit retry loop
            else:
                logger.warning(f"No content found for {url}")
                if attempt < max_retries - 1:
                    logger.info(f"Retrying {url} in 3 seconds...")
                    await asyncio.sleep(3)
                    continue
                else:
                    # Last attempt failed, create a placeholder file
                    with open(output_path, 'w', encoding='utf-8') as f:
                        f.write(f"# Error retrieving content for {url}\n\nThe content could not be retrieved after {max_retries} attempts.")
                    logger.error(f"Created placeholder file for {url} after {max_retries} failed attempts")

        except Exception as e:
            logger.error(f"Error processing {url}: {str(e)}")
            if attempt < max_retries - 1:
                logger.info(f"Retrying {url} in 3 seconds...")
                await asyncio.sleep(3)
            else:
                logger.error(f"Failed to process {url} after {max_retries} attempts")
                # Create a placeholder file on complete failure
                with open(output_path, 'w', encoding='utf-8') as f:
                    f.write(f"# Error processing {url}\n\nAn error occurred: {str(e)}")
                logger.error(f"Created error placeholder file for {url}")

async def process_urls_with_semaphore(urls: List[str], output_dir: str, crawler: AsyncWebCrawler, max_concurrent: int = 5):
    """Process URLs with a semaphore to limit concurrency."""
    semaphore = asyncio.Semaphore(max_concurrent)

    async def process_with_semaphore(url):
        async with semaphore:
            await crawl_and_save_page(url, output_dir, crawler)

    # Create tasks
    tasks = [process_with_semaphore(url) for url in urls]

    # Process URLs and handle exceptions
    for task in asyncio.as_completed(tasks):
        try:
            await task
        except Exception as e:
            logger.error(f"Task error: {str(e)}")

async def main():
    # Create output directory
    output_dir = "Crawl4AI_Documentation"
    os.makedirs(output_dir, exist_ok=True)

    sitemap_url = "https://docs.crawl4ai.com/sitemap.xml"

    try:
        # Option 1: Process URLs from sitemap
        urls = await download_sitemap(sitemap_url)

        # Option 2: If sitemap fails, fallback to predefined URLs
        if not urls:
            logger.warning("No URLs found in sitemap, using predefined mapping instead")
            urls = list(URL_TO_FILENAME_MAP.keys())

        logger.info(f"Processing {len(urls)} URLs")

        # Configure browser and crawler
        browser_config = BrowserConfig(
            headless=True,  # Run in headless mode
            verbose=True,   # Enable verbose logging
            user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
        )

        # Create crawler and process URLs
        async with AsyncWebCrawler(config_browser=browser_config) as crawler:
            await process_urls_with_semaphore(urls, output_dir, crawler, max_concurrent=3)

        # Log completion status
        total_files = len(os.listdir(output_dir))
        logger.info(f"Crawling complete. {total_files} files saved to {output_dir}")

        # Verify all files were created
        missing_files = []
        for url, filename in URL_TO_FILENAME_MAP.items():
            if not os.path.exists(os.path.join(output_dir, filename)):
                missing_files.append(f"{filename} ({url})")

        if missing_files:
            logger.warning(f"Some files are missing: {len(missing_files)} files")
            for missing in missing_files:
                logger.warning(f"Missing: {missing}")
        else:
            logger.info("All expected files were successfully created")

    except Exception as e:
        logger.error(f"An error occurred during execution: {str(e)}")
        import traceback
        logger.error(traceback.format_exc())

if __name__ == "__main__":
    asyncio.run(main())