Crawl4ai

#!/usr/bin/env python3
"""
Basic Crawl4AI crawler template
Usage: python basic_crawler.py <url>
"""

import asyncio
import sys

# Version check
MIN_CRAWL4AI_VERSION = "0.7.4"
try:
    from crawl4ai.__version__ import __version__
    from packaging import version
    if version.parse(__version__) < version.parse(MIN_CRAWL4AI_VERSION):
        print(f"⚠️  Warning: Crawl4AI {MIN_CRAWL4AI_VERSION}+ recommended (you have {__version__})")
except ImportError:
    print(f"ℹ️  Crawl4AI {MIN_CRAWL4AI_VERSION}+ required")

from crawl4ai import AsyncWebCrawler, BrowserConfig, CrawlerRunConfig, CacheMode

async def crawl_basic(url: str):
    """Basic crawling with markdown output"""

    # Configure browser
    browser_config = BrowserConfig(
        headless=True,
        viewport_width=1920,
        viewport_height=1080
    )

    # Configure crawler
    crawler_config = CrawlerRunConfig(
        cache_mode=CacheMode.BYPASS,
        remove_overlay_elements=True,
        wait_for_images=True,
        screenshot=True
    )

    async with AsyncWebCrawler(config=browser_config) as crawler:
        result = await crawler.arun(
            url=url,
            config=crawler_config
        )

        if result.success:
            print(f"✅ Crawled: {result.url}")
            print(f"   Title: {result.metadata.get('title', 'N/A')}")
            print(f"   Links found: {len(result.links.get('internal', []))} internal, {len(result.links.get('external', []))} external")
            print(f"   Media found: {len(result.media.get('images', []))} images, {len(result.media.get('videos', []))} videos")
            print(f"   Content length: {len(result.markdown)} chars")

            # Save markdown
            with open("output.md", "w") as f:
                f.write(result.markdown)
            print("📄 Saved to output.md")

            # Save screenshot if available
            if result.screenshot:
                # Check if screenshot is base64 string or bytes
                if isinstance(result.screenshot, str):
                    import base64
                    screenshot_data = base64.b64decode(result.screenshot)
                else:
                    screenshot_data = result.screenshot
                with open("screenshot.png", "wb") as f:
                    f.write(screenshot_data)
                print("📸 Saved screenshot.png")
        else:
            print(f"❌ Failed: {result.error_message}")

        return result

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python basic_crawler.py <url>")
        sys.exit(1)

    url = sys.argv[1]
    asyncio.run(crawl_basic(url))

#!/usr/bin/env python3
"""
Crawl4AI batch/multi-URL crawler with concurrent processing
Usage: python batch_crawler.py urls.txt [--max-concurrent 5]
"""

import asyncio
import sys
import json
from pathlib import Path
from typing import List, Dict, Any

# Version check
MIN_CRAWL4AI_VERSION = "0.7.4"
try:
    from crawl4ai.__version__ import __version__
    from packaging import version
    if version.parse(__version__) < version.parse(MIN_CRAWL4AI_VERSION):
        print(f"⚠️  Warning: Crawl4AI {MIN_CRAWL4AI_VERSION}+ recommended (you have {__version__})")
except ImportError:
    print(f"ℹ️  Crawl4AI {MIN_CRAWL4AI_VERSION}+ required")

from crawl4ai import AsyncWebCrawler, BrowserConfig, CrawlerRunConfig, CacheMode

async def crawl_batch(urls: List[str], max_concurrent: int = 5):
    """
    Crawl multiple URLs efficiently with concurrent processing
    """
    print(f"🚀 Starting batch crawl of {len(urls)} URLs (max {max_concurrent} concurrent)")

    # Configure browser for efficiency
    browser_config = BrowserConfig(
        headless=True,
        viewport_width=1280,
        viewport_height=800,
        verbose=False
    )

    # Configure crawler
    crawler_config = CrawlerRunConfig(
        cache_mode=CacheMode.BYPASS,
        remove_overlay_elements=True,
        wait_for="css:body",
        page_timeout=30000,  # 30 seconds timeout per page
        screenshot=False  # Disable screenshots for batch processing
    )

    results = []
    failed = []

    async with AsyncWebCrawler(config=browser_config) as crawler:
        # Use arun_many for efficient batch processing
        batch_results = await crawler.arun_many(
            urls=urls,
            config=crawler_config,
            max_concurrent=max_concurrent
        )

        for result in batch_results:
            if result.success:
                results.append({
                    "url": result.url,
                    "title": result.metadata.get("title", ""),
                    "description": result.metadata.get("description", ""),
                    "content_length": len(result.markdown),
                    "links_count": len(result.links.get("internal", [])) + len(result.links.get("external", [])),
                    "images_count": len(result.media.get("images", [])),
                })
                print(f"✅ {result.url}")
            else:
                failed.append({
                    "url": result.url,
                    "error": result.error_message
                })
                print(f"❌ {result.url}: {result.error_message}")

    # Save results
    output = {
        "success_count": len(results),
        "failed_count": len(failed),
        "results": results,
        "failed": failed
    }

    with open("batch_results.json", "w") as f:
        json.dump(output, f, indent=2)

    # Save individual markdown files
    markdown_dir = Path("batch_markdown")
    markdown_dir.mkdir(exist_ok=True)

    for i, result in enumerate(batch_results):
        if result.success:
            # Create safe filename from URL
            safe_name = result.url.replace("https://", "").replace("http://", "")
            safe_name = "".join(c if c.isalnum() or c in "-_" else "_" for c in safe_name)[:100]

            file_path = markdown_dir / f"{i:03d}_{safe_name}.md"
            with open(file_path, "w") as f:
                f.write(f"# {result.metadata.get('title', result.url)}\n\n")
                f.write(f"URL: {result.url}\n\n")
                f.write(result.markdown)

    print(f"\n📊 Batch Crawl Complete:")
    print(f"   ✅ Success: {len(results)}")
    print(f"   ❌ Failed: {len(failed)}")
    print(f"   💾 Results saved to: batch_results.json")
    print(f"   📁 Markdown files saved to: {markdown_dir}/")

    return output

async def crawl_with_extraction(urls: List[str], schema_file: str = None):
    """
    Batch crawl with structured data extraction
    """
    from crawl4ai.extraction_strategy import JsonCssExtractionStrategy

    schema = None
    if schema_file and Path(schema_file).exists():
        with open(schema_file) as f:
            schema = json.load(f)
        print(f"📋 Using extraction schema from: {schema_file}")
    else:
        # Default schema for general content
        schema = {
            "name": "content",
            "selector": "body",
            "fields": [
                {"name": "headings", "selector": "h1, h2, h3", "type": "text", "all": True},
                {"name": "paragraphs", "selector": "p", "type": "text", "all": True},
                {"name": "links", "selector": "a[href]", "type": "attribute", "attribute": "href", "all": True}
            ]
        }

    extraction_strategy = JsonCssExtractionStrategy(schema=schema)

    crawler_config = CrawlerRunConfig(
        extraction_strategy=extraction_strategy,
        cache_mode=CacheMode.BYPASS
    )

    extracted_data = []

    async with AsyncWebCrawler() as crawler:
        results = await crawler.arun_many(
            urls=urls,
            config=crawler_config,
            max_concurrent=5
        )

        for result in results:
            if result.success and result.extracted_content:
                try:
                    data = json.loads(result.extracted_content)
                    extracted_data.append({
                        "url": result.url,
                        "data": data
                    })
                    print(f"✅ Extracted from: {result.url}")
                except json.JSONDecodeError:
                    print(f"⚠️ Failed to parse JSON from: {result.url}")

    # Save extracted data
    with open("batch_extracted.json", "w") as f:
        json.dump(extracted_data, f, indent=2)

    print(f"\n💾 Extracted data saved to: batch_extracted.json")
    return extracted_data

def load_urls(source: str) -> List[str]:
    """Load URLs from file or string"""
    if Path(source).exists():
        with open(source) as f:
            urls = [line.strip() for line in f if line.strip() and not line.startswith("#")]
    else:
        # Treat as comma-separated URLs
        urls = [url.strip() for url in source.split(",") if url.strip()]

    return urls

async def main():
    if len(sys.argv) < 2:
        print("""
Crawl4AI Batch Crawler

Usage:
    # Crawl URLs from file
    python batch_crawler.py urls.txt [--max-concurrent 5]

    # Crawl with extraction
    python batch_crawler.py urls.txt --extract [schema.json]

    # Crawl comma-separated URLs
    python batch_crawler.py "https://example.com,https://example.org"

Options:
    --max-concurrent N    Max concurrent crawls (default: 5)
    --extract [schema]    Extract structured data using schema

Example urls.txt:
    https://example.com
    https://example.org
    # Comments are ignored
    https://another-site.com
""")
        sys.exit(1)

    source = sys.argv[1]
    urls = load_urls(source)

    if not urls:
        print("❌ No URLs found")
        sys.exit(1)

    print(f"📋 Loaded {len(urls)} URLs")

    # Parse options
    max_concurrent = 5
    extract_mode = False
    schema_file = None

    for i, arg in enumerate(sys.argv[2:], 2):
        if arg == "--max-concurrent" and i + 1 < len(sys.argv):
            max_concurrent = int(sys.argv[i + 1])
        elif arg == "--extract":
            extract_mode = True
            if i + 1 < len(sys.argv) and not sys.argv[i + 1].startswith("--"):
                schema_file = sys.argv[i + 1]

    if extract_mode:
        await crawl_with_extraction(urls, schema_file)
    else:
        await crawl_batch(urls, max_concurrent)

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

#!/usr/bin/env python3
"""
Crawl4AI extraction pipeline - Three approaches:
1. Generate schema with LLM (one-time) then use CSS extraction (most efficient)
2. Manual CSS/JSON schema extraction
3. Direct LLM extraction (for complex/irregular content)

Usage examples:
  Generate schema: python extraction_pipeline.py --generate-schema <url> "<instruction>"
  Use generated schema: python extraction_pipeline.py --use-schema <url> schema.json
  Manual CSS: python extraction_pipeline.py --css <url> "<css_selector>"
  Direct LLM: python extraction_pipeline.py --llm <url> "<instruction>"
"""

import asyncio
import sys
import json
from pathlib import Path

# Version check
MIN_CRAWL4AI_VERSION = "0.7.4"
try:
    from crawl4ai.__version__ import __version__
    from packaging import version
    if version.parse(__version__) < version.parse(MIN_CRAWL4AI_VERSION):
        print(f"⚠️  Warning: Crawl4AI {MIN_CRAWL4AI_VERSION}+ recommended (you have {__version__})")
except ImportError:
    print(f"ℹ️  Crawl4AI {MIN_CRAWL4AI_VERSION}+ required")

from crawl4ai import AsyncWebCrawler, BrowserConfig, CrawlerRunConfig
from crawl4ai.extraction_strategy import (
    LLMExtractionStrategy,
    JsonCssExtractionStrategy,
    CosineStrategy
)

# =============================================================================
# APPROACH 1: Generate Schema (Most Efficient for Repetitive Patterns)
# =============================================================================

async def generate_schema(url: str, instruction: str, output_file: str = "generated_schema.json"):
    """
    Step 1: Generate a reusable schema using LLM (one-time cost)
    Best for: E-commerce sites, blogs, news sites with repetitive patterns
    """
    print("🔍 Generating extraction schema using LLM...")

    browser_config = BrowserConfig(headless=True)

    # Use LLM to analyze the page structure and generate schema
    extraction_strategy = LLMExtractionStrategy(
        provider="openai/gpt-4o-mini",  # Can use any LLM provider
        instruction=f"""
        Analyze this webpage and generate a CSS/JSON extraction schema.
        Task: {instruction}

        Return a JSON schema with CSS selectors that can extract the required data.
        Format:
        {{
            "name": "items",
            "selector": "main_container_selector",
            "fields": [
                {{"name": "field1", "selector": "css_selector", "type": "text"}},
                {{"name": "field2", "selector": "css_selector", "type": "link"}},
                // more fields...
            ]
        }}

        Make selectors as specific as possible to avoid false matches.
        """
    )

    crawler_config = CrawlerRunConfig(
        extraction_strategy=extraction_strategy,
        wait_for="css:body",
        remove_overlay_elements=True
    )

    async with AsyncWebCrawler(config=browser_config) as crawler:
        result = await crawler.arun(url=url, config=crawler_config)

        if result.success and result.extracted_content:
            try:
                # Parse and save the generated schema
                schema = json.loads(result.extracted_content)

                # Validate and enhance schema
                if "name" not in schema:
                    schema["name"] = "items"
                if "fields" not in schema:
                    print("⚠️ Generated schema missing fields, using fallback")
                    schema = {
                        "name": "items",
                        "baseSelector": "div.item, article, .product",
                        "fields": [
                            {"name": "title", "selector": "h1, h2, h3", "type": "text"},
                            {"name": "description", "selector": "p", "type": "text"},
                            {"name": "link", "selector": "a", "type": "attribute", "attribute": "href"}
                        ]
                    }

                # Save schema
                with open(output_file, "w") as f:
                    json.dump(schema, f, indent=2)

                print(f"✅ Schema generated and saved to: {output_file}")
                print(f"📋 Schema structure:")
                print(json.dumps(schema, indent=2))

                return schema

            except json.JSONDecodeError as e:
                print(f"❌ Failed to parse generated schema: {e}")
                print("Raw output:", result.extracted_content[:500])
                return None
        else:
            print(f"❌ Failed to generate schema: {result.error_message if result else 'Unknown error'}")
            return None

async def use_generated_schema(url: str, schema_file: str):
    """
    Step 2: Use the generated schema for fast, repeated extractions
    No LLM calls needed - pure CSS extraction
    """
    print(f"📂 Loading schema from: {schema_file}")

    try:
        with open(schema_file, "r") as f:
            schema = json.load(f)
    except FileNotFoundError:
        print(f"❌ Schema file not found: {schema_file}")
        print("💡 Generate a schema first using: python extraction_pipeline.py --generate-schema <url> \"<instruction>\"")
        return None

    print("🚀 Extracting data using generated schema (no LLM calls)...")

    extraction_strategy = JsonCssExtractionStrategy(
        schema=schema,
        verbose=True
    )

    crawler_config = CrawlerRunConfig(
        extraction_strategy=extraction_strategy,
        wait_for="css:body"
    )

    async with AsyncWebCrawler() as crawler:
        result = await crawler.arun(url=url, config=crawler_config)

        if result.success and result.extracted_content:
            data = json.loads(result.extracted_content)
            items = data.get(schema.get("name", "items"), [])

            print(f"✅ Extracted {len(items)} items using schema")

            # Save results
            with open("extracted_data.json", "w") as f:
                json.dump(data, f, indent=2)
            print("💾 Saved to extracted_data.json")

            # Show sample
            if items:
                print("\n📋 Sample (first item):")
                print(json.dumps(items[0], indent=2))

            return data
        else:
            print(f"❌ Extraction failed: {result.error_message if result else 'Unknown error'}")
            return None

# =============================================================================
# APPROACH 2: Manual Schema Definition
# =============================================================================

async def extract_with_manual_schema(url: str, schema: dict = None):
    """
    Use a manually defined CSS/JSON schema
    Best for: When you know the exact structure of the website
    """

    if not schema:
        # Example schema for general content extraction
        schema = {
            "name": "content",
            "baseSelector": "body",  # Changed from 'selector' to 'baseSelector'
            "fields": [
                {"name": "title", "selector": "h1", "type": "text"},
                {"name": "paragraphs", "selector": "p", "type": "text", "all": True},
                {"name": "links", "selector": "a", "type": "attribute", "attribute": "href", "all": True}
            ]
        }

    print("📐 Using manual CSS/JSON schema for extraction...")

    extraction_strategy = JsonCssExtractionStrategy(
        schema=schema,
        verbose=True
    )

    crawler_config = CrawlerRunConfig(
        extraction_strategy=extraction_strategy
    )

    async with AsyncWebCrawler() as crawler:
        result = await crawler.arun(url=url, config=crawler_config)

        if result.success and result.extracted_content:
            data = json.loads(result.extracted_content)
            # Handle both list and dict formats
            if isinstance(data, list):
                items = data
            else:
                items = data.get(schema["name"], [])

            print(f"✅ Extracted {len(items)} items using manual schema")

            with open("manual_extracted.json", "w") as f:
                json.dump(data, f, indent=2)
            print("💾 Saved to manual_extracted.json")

            return data
        else:
            print(f"❌ Extraction failed")
            return None

# =============================================================================
# APPROACH 3: Direct LLM Extraction
# =============================================================================

async def extract_with_llm(url: str, instruction: str):
    """
    Direct LLM extraction - uses LLM for every request
    Best for: Complex, irregular content or one-time extractions
    Note: Most expensive approach, use sparingly
    """
    print("🤖 Using direct LLM extraction...")

    browser_config = BrowserConfig(headless=True)

    extraction_strategy = LLMExtractionStrategy(
        provider="openai/gpt-4o-mini",  # Can change to ollama/llama3, anthropic/claude, etc.
        instruction=instruction,
        schema={
            "type": "object",
            "properties": {
                "items": {
                    "type": "array",
                    "items": {"type": "object"}
                },
                "summary": {"type": "string"}
            }
        }
    )

    crawler_config = CrawlerRunConfig(
        extraction_strategy=extraction_strategy,
        wait_for="css:body",
        remove_overlay_elements=True
    )

    async with AsyncWebCrawler(config=browser_config) as crawler:
        result = await crawler.arun(url=url, config=crawler_config)

        if result.success and result.extracted_content:
            try:
                data = json.loads(result.extracted_content)
                items = data.get('items', [])

                print(f"✅ LLM extracted {len(items)} items")
                print(f"📝 Summary: {data.get('summary', 'N/A')}")

                with open("llm_extracted.json", "w") as f:
                    json.dump(data, f, indent=2)
                print("💾 Saved to llm_extracted.json")

                if items:
                    print("\n📋 Sample (first item):")
                    print(json.dumps(items[0], indent=2))

                return data
            except json.JSONDecodeError:
                print("⚠️ Could not parse LLM output as JSON")
                print(result.extracted_content[:500])
                return None
        else:
            print(f"❌ LLM extraction failed")
            return None

# =============================================================================
# Main CLI Interface
# =============================================================================

async def main():
    if len(sys.argv) < 3:
        print("""
Crawl4AI Extraction Pipeline - Three Approaches

1️⃣  GENERATE & USE SCHEMA (Most Efficient for Repetitive Patterns):
    Step 1: Generate schema (one-time LLM cost)
    python extraction_pipeline.py --generate-schema <url> "<what to extract>"

    Step 2: Use schema for fast extraction (no LLM)
    python extraction_pipeline.py --use-schema <url> generated_schema.json

2️⃣  MANUAL SCHEMA (When You Know the Structure):
    python extraction_pipeline.py --manual <url>
    (Edit the schema in the script for your needs)

3️⃣  DIRECT LLM (For Complex/Irregular Content):
    python extraction_pipeline.py --llm <url> "<extraction instruction>"

Examples:
    # E-commerce products
    python extraction_pipeline.py --generate-schema https://shop.com "Extract all products with name, price, image"
    python extraction_pipeline.py --use-schema https://shop.com generated_schema.json

    # News articles
    python extraction_pipeline.py --generate-schema https://news.com "Extract headlines, dates, and summaries"

    # Complex content
    python extraction_pipeline.py --llm https://complex-site.com "Extract financial data and quarterly reports"
""")
        sys.exit(1)

    mode = sys.argv[1]
    url = sys.argv[2]

    if mode == "--generate-schema":
        if len(sys.argv) < 4:
            print("Error: Missing extraction instruction")
            print("Usage: python extraction_pipeline.py --generate-schema <url> \"<instruction>\"")
            sys.exit(1)
        instruction = sys.argv[3]
        output_file = sys.argv[4] if len(sys.argv) > 4 else "generated_schema.json"
        await generate_schema(url, instruction, output_file)

    elif mode == "--use-schema":
        if len(sys.argv) < 4:
            print("Error: Missing schema file")
            print("Usage: python extraction_pipeline.py --use-schema <url> <schema.json>")
            sys.exit(1)
        schema_file = sys.argv[3]
        await use_generated_schema(url, schema_file)

    elif mode == "--manual":
        await extract_with_manual_schema(url)

    elif mode == "--llm":
        if len(sys.argv) < 4:
            print("Error: Missing extraction instruction")
            print("Usage: python extraction_pipeline.py --llm <url> \"<instruction>\"")
            sys.exit(1)
        instruction = sys.argv[3]
        await extract_with_llm(url, instruction)

    else:
        print(f"Unknown mode: {mode}")
        print("Use --generate-schema, --use-schema, --manual, or --llm")
        sys.exit(1)

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

#!/usr/bin/env python3
"""
Google Search Scraper using Crawl4AI
Usage: python google_search.py "<search query>" [max_results]

Example: python google_search.py "2026年Go语言展望" 20
"""

import asyncio
import sys
import json
import urllib.parse
from typing import List, Dict

try:
    from crawl4ai.__version__ import __version__
    from packaging import version
    MIN_CRAWL4AI_VERSION = "0.7.4"
    if version.parse(__version__) < version.parse(MIN_CRAWL4AI_VERSION):
        print(f"⚠️  Warning: Crawl4AI {MIN_CRAWL4AI_VERSION}+ recommended (you have {__version__})")
except ImportError:
    print(f"ℹ️  Crawl4AI {MIN_CRAWL4AI_VERSION}+ required")

from crawl4ai import AsyncWebCrawler, BrowserConfig, CrawlerRunConfig, CacheMode
from crawl4ai.extraction_strategy import JsonCssExtractionStrategy, LLMExtractionStrategy


async def search_google_css(query: str, max_results: int = 20) -> List[Dict]:
    """
    使用 CSS 选择器策略提取 Google 搜索结果（最快，无需 LLM）
    """

    # 构建搜索 URL
    encoded_query = urllib.parse.quote(query)
    search_url = f"https://www.google.com/search?q={encoded_query}&num={max_results}"

    print(f"🔍 Searching: {query}")
    print(f"📊 Max results: {max_results}")
    print(f"🌐 URL: {search_url}")

    # 定义 Google 搜索结果的 CSS schema
    # Google 的 HTML 结构会变化，这里使用常用的选择器
    schema = {
        "name": "search_results",
        "baseSelector": "div.g, div[data-hveid], div.tF2Cxc, div.yuRUbf",
        "fields": [
            {
                "name": "title",
                "selector": "h3, h3.LC20lb, div[role='heading']",
                "type": "text"
            },
            {
                "name": "link",
                "selector": "a",
                "type": "attribute",
                "attribute": "href"
            },
            {
                "name": "description",
                "selector": "div.VwiC3b, div.s, div.ITZIwc, span.aCOpRe",
                "type": "text"
            },
            {
                "name": "site_name",
                "selector": "div.NJo7tc, span.VuuXrf, cite",
                "type": "text"
            }
        ]
    }

    browser_config = BrowserConfig(
        headless=True,
        viewport_width=1920,
        viewport_height=1080,
        user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
    )

    crawler_config = CrawlerRunConfig(
        extraction_strategy=JsonCssExtractionStrategy(schema=schema, verbose=True),
        cache_mode=CacheMode.BYPASS,
        wait_for="css:div.g, div.search, body",
        page_timeout=30000,
        js_code=[
            # 等待页面加载完成
            "const waitFor = (ms) => new Promise(resolve => setTimeout(resolve, ms));",
            "await waitFor(2000);"
        ]
    )

    async with AsyncWebCrawler(config=browser_config) as crawler:
        result = await crawler.arun(url=search_url, config=crawler_config)

        if result.success:
            print("✅ Successfully fetched search results")

            if result.extracted_content:
                try:
                    data = json.loads(result.extracted_content)
                    # 处理列表和字典两种格式
                    if isinstance(data, list):
                        results = data
                    else:
                        results = data.get("search_results", data.get("results", []))

                    # 过滤掉空结果和无效结果
                    seen = set()
                    valid_results = []
                    for r in results:
                        if r.get("title") and r.get("link"):
                            # 清理 URL（Google 有时会在 URL 前加 /url?q=）
                            link = r["link"]
                            if link.startswith("/url?q="):
                                from urllib.parse import urlparse, parse_qs
                                parsed = urlparse(link)
                                link = parse_qs(parsed.query).get("q", [link])[0]
                            r["link"] = link

                            # 使用 URL 作为唯一标识去重
                            if link not in seen:
                                seen.add(link)
                                valid_results.append(r)

                    print(f"📋 Extracted {len(valid_results)} valid results")
                    return valid_results[:max_results]

                except json.JSONDecodeError as e:
                    print(f"❌ Failed to parse extracted content: {e}")
                    print("Raw output:", result.extracted_content[:500] if result.extracted_content else "None")
                    return []
            else:
                print("⚠️ No extracted content, trying alternative method...")
                return await search_google_llm_fallback(query, max_results)
        else:
            print(f"❌ Failed: {result.error_message}")
            print("Trying fallback method...")
            return await search_google_llm_fallback(query, max_results)


async def search_google_llm_fallback(query: str, max_results: int = 20) -> List[Dict]:
    """
    使用 LLM 作为备选方案提取搜索结果
    注意：这需要配置 LLM API 密钥
    """

    print("🤖 Using LLM fallback extraction...")

    encoded_query = urllib.parse.quote(query)
    search_url = f"https://www.google.com/search?q={encoded_query}&num={max_results}"

    # 尝试使用简单的 LLM 提取
    extraction_strategy = LLMExtractionStrategy(
        provider="openai/gpt-4o-mini",
        instruction=f"""
        Extract the top {max_results} search results from this Google search page for "{query}".

        For each search result, extract:
        1. Title - the blue link text
        2. Link - the URL (clean the URL, remove /url?q= prefix if present)
        3. Description - the gray text snippet below the title
        4. Site name - the green text showing the website name

        Return as JSON with a "results" array containing objects with these fields.
        Skip any ads or sponsored content.
        """
    )

    crawler_config = CrawlerRunConfig(
        extraction_strategy=extraction_strategy,
        cache_mode=CacheMode.BYPASS,
        page_timeout=30000
    )

    async with AsyncWebCrawler() as crawler:
        result = await crawler.arun(url=search_url, config=crawler_config)

        if result.success and result.extracted_content:
            try:
                data = json.loads(result.extracted_content)
                return data.get("results", [])
            except json.JSONDecodeError:
                print("⚠️ LLM output could not be parsed as JSON")
                return []
        else:
            print(f"❌ Fallback also failed: {result.error_message}")
            return []


async def search_google_with_html_parsing(query: str, max_results: int = 20) -> List[Dict]:
    """
    直接解析 HTML 作为最后的备选方案
    """

    print("🔧 Using direct HTML parsing...")

    encoded_query = urllib.parse.quote(query)
    search_url = f"https://www.google.com/search?q={encoded_query}&num={max_results}"

    browser_config = BrowserConfig(
        headless=True,
        viewport_width=1920,
        viewport_height=1080,
        user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
    )

    crawler_config = CrawlerRunConfig(
        cache_mode=CacheMode.BYPASS,
        wait_for="css:body",
        page_timeout=30000,
        js_code=[
            "const waitFor = (ms) => new Promise(resolve => setTimeout(resolve, ms));",
            "await waitFor(3000);"
        ]
    )

    async with AsyncWebCrawler(config=browser_config) as crawler:
        result = await crawler.arun(url=search_url, config=crawler_config)

        if result.success and result.html:
            from bs4 import BeautifulSoup

            soup = BeautifulSoup(result.html, 'html.parser')
            results = []

            # Google 搜索结果通常在 div.g 中
            for div in soup.select('div.g, div.tF2Cxc'):
                try:
                    # 提取标题
                    title_elem = div.select_one('h3')
                    title = title_elem.get_text() if title_elem else ""

                    # 提取链接
                    link_elem = div.select_one('a')
                    link = link_elem.get('href', '') if link_elem else ""

                    # 清理 Google 重定向链接
                    if link.startswith('/url?q='):
                        from urllib.parse import urlparse, parse_qs, unquote
                        parsed = urlparse(link)
                        link = unquote(parse_qs(parsed.query).get('q', [link])[0])

                    # 提取描述
                    desc_elem = div.select_one('div.VwiC3b, div.s, span.aCOpRe')
                    description = desc_elem.get_text() if desc_elem else ""

                    # 提取网站名称
                    site_elem = div.select_one('div.NJo7tc, span.VuuXrf, cite')
                    site_name = site_elem.get_text() if site_elem else ""

                    if title and link and not link.startswith('#'):
                        results.append({
                            "title": title.strip(),
                            "link": link.strip(),
                            "description": description.strip(),
                            "site_name": site_name.strip()
                        })

                        if len(results) >= max_results:
                            break

                except Exception as e:
                    continue

            print(f"📋 Parsed {len(results)} results from HTML")
            return results
        else:
            print(f"❌ HTML parsing failed")
            return []


async def main():
    if len(sys.argv) < 2:
        print("Usage: python google_search.py \"<search query>\" [max_results]")
        print("Example: python google_search.py \"2026年Go语言展望\" 20")
        sys.exit(1)

    query = sys.argv[1]
    max_results = int(sys.argv[2]) if len(sys.argv) > 2 else 20

    # 方法1: CSS 提取
    results = await search_google_css(query, max_results)

    # 如果 CSS 提取失败，尝试 HTML 解析
    if not results:
        results = await search_google_with_html_parsing(query, max_results)

    # 输出结果
    if results:
        output = {
            "query": query,
            "total_results": len(results),
            "results": results
        }

        print("\n" + "="*60)
        print(f"✅ Successfully extracted {len(results)} search results")
        print("="*60)

        # 保存到文件
        output_file = "google_search_results.json"
        with open(output_file, "w", encoding="utf-8") as f:
            json.dump(output, f, ensure_ascii=False, indent=2)

        print(f"\n💾 Results saved to: {output_file}")
        print("\n📋 Preview (first 3 results):")
        print(json.dumps(results[:3], ensure_ascii=False, indent=2))

        # 打印完整 JSON 到 stdout
        print("\n" + "="*60)
        print("FULL JSON OUTPUT:")
        print("="*60)
        print(json.dumps(output, ensure_ascii=False, indent=2))
    else:
        print("❌ No results extracted. Please check:")
        print("  1. Your internet connection")
        print("  2. Whether Google is blocking the request (try with headless=False)")
        print("  3. The CSS selectors (Google might have changed their HTML)")


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

{
  "schema_version": "2.0",
  "meta": {
    "generated_at": "2026-01-17T08:12:47.744Z",
    "slug": "smallnest-crawl4ai",
    "source_url": "https://github.com/smallnest/crawl4ai-skill/tree/master/",
    "source_ref": "master",
    "model": "claude",
    "analysis_version": "3.0.0",
    "source_type": "community",
    "content_hash": "e932d043202cdef88ff04d7cabf4dfbbc3944549a1bb62b8784d9c55ccbe6722",
    "tree_hash": "ee4b9f54d4f8a1552135ac8f1e85434daa89b91daeed746b345aeae1adb27700"
  },
  "skill": {
    "name": "crawl4ai",
    "description": "This skill should be used when users need to scrape websites, extract structured data, handle JavaScript-heavy pages, crawl multiple URLs, or build automated web data pipelines. Includes optimized extraction patterns with schema generation for efficient, LLM-free extraction.",
    "summary": "This skill should be used when users need to scrape websites, extract structured data, handle JavaSc...",
    "icon": "🕷️",
    "version": "0.7.4",
    "author": "smallnest",
    "license": "MIT",
    "category": "data",
    "tags": [
      "web-scraping",
      "crawler",
      "data-extraction",
      "async"
    ],
    "supported_tools": [
      "claude",
      "codex",
      "claude-code"
    ],
    "risk_factors": [
      "filesystem",
      "network",
      "scripts"
    ]
  },
  "security_audit": {
    "risk_level": "low",
    "is_blocked": false,
    "safe_to_publish": true,
    "summary": "Static analysis flagged 2290 issues but 99% are false positives from markdown documentation. Actual Python code shows legitimate web crawler functionality with user-controlled URLs, explicit credential configuration, and standard file output operations. No hidden data exfiltration or malicious patterns found.",
    "risk_factor_evidence": [
      {
        "factor": "filesystem",
        "evidence": [
          {
            "file": "scripts/basic_crawler.py",
            "line_start": 54,
            "line_end": 67
          },
          {
            "file": "scripts/extraction_pipeline.py",
            "line_start": 103,
            "line_end": 103
          },
          {
            "file": "scripts/google_search.py",
            "line_start": 300,
            "line_end": 300
          }
        ]
      },
      {
        "factor": "network",
        "evidence": [
          {
            "file": "scripts/google_search.py",
            "line_start": 35,
            "line_end": 35
          },
          {
            "file": "scripts/basic_crawler.py",
            "line_start": 41,
            "line_end": 44
          }
        ]
      },
      {
        "factor": "scripts",
        "evidence": [
          {
            "file": "tests/run_all_tests.py",
            "line_start": 15,
            "line_end": 18
          }
        ]
      }
    ],
    "critical_findings": [],
    "high_findings": [],
    "medium_findings": [],
    "low_findings": [],
    "dangerous_patterns": [],
    "files_scanned": 16,
    "total_lines": 9145,
    "audit_model": "claude",
    "audited_at": "2026-01-17T08:12:47.744Z"
  },
  "content": {
    "user_title": "Scrape websites and extract structured data",
    "value_statement": "Crawl4AI enables efficient web scraping with JavaScript support, schema-based extraction, and flexible output formats. Users can extract data without LLM calls for cost-effective automation or use LLM-powered extraction for complex content.",
    "seo_keywords": [
      "crawl4ai",
      "web scraping",
      "web crawler",
      "data extraction",
      "Claude",
      "Codex",
      "Claude Code",
      "async crawler",
      "markdown generation",
      "structured data"
    ],
    "actual_capabilities": [
      "Crawl single or multiple URLs with JavaScript support",
      "Extract structured data using CSS selectors or LLM",
      "Generate clean markdown output from web pages",
      "Handle authenticated sessions and proxy configurations",
      "Process dynamic content with configurable wait conditions"
    ],
    "limitations": [
      "Requires external crawl4ai package installation",
      "LLM extraction requires API keys and costs money",
      "Rate limits and bot detection may affect some sites"
    ],
    "use_cases": [
      {
        "target_user": "Data engineers",
        "title": "Build data pipelines",
        "description": "Extract structured data from websites for analytics and reporting workflows."
      },
      {
        "target_user": "Developers",
        "title": "Document websites",
        "description": "Convert documentation sites to markdown for offline reading or migration."
      },
      {
        "target_user": "Researchers",
        "title": "Aggregate web content",
        "description": "Collect and filter content from multiple sources for research analysis."
      }
    ],
    "prompt_templates": [
      {
        "title": "Basic crawl",
        "scenario": "Get page content",
        "prompt": "Crawl this URL and return the main content as markdown: https://example.com"
      },
      {
        "title": "Extract data",
        "scenario": "Structured extraction",
        "prompt": "Extract product names, prices, and links from this e-commerce page using CSS selectors."
      },
      {
        "title": "Handle JavaScript",
        "scenario": "Dynamic content",
        "prompt": "Crawl this JavaScript-heavy page and wait for the dynamic content to load before extracting."
      },
      {
        "title": "Batch processing",
        "scenario": "Multiple URLs",
        "prompt": "Crawl these three URLs in parallel and extract the main headlines from each: https://news1.com, https://news2.com, https://news3.com"
      }
    ],
    "output_examples": [
      {
        "input": "Crawl https://docs.python.org/3/ and extract the installation instructions",
        "output": [
          "## Installation Instructions",
          "- Download Python from python.org",
          "- Run the installer",
          "- Add Python to PATH",
          "Source: https://docs.python.org/3/"
        ]
      },
      {
        "input": "Extract all article titles and links from a blog listing page",
        "output": [
          "Extracted 15 articles:",
          "- 'Getting Started with Python' → https://blog.example.com/python-start",
          "- 'Advanced Patterns' → https://blog.example.com/advanced",
          "- 'Best Practices' → https://blog.example.com/best-practices"
        ]
      },
      {
        "input": "Crawl a dynamic page with infinite scroll",
        "output": [
          "Waited 3 seconds for content to load",
          "Found 50 product cards",
          "Extracted names, prices, and images for all products"
        ]
      }
    ],
    "best_practices": [
      "Use schema-based CSS extraction for repetitive sites to avoid LLM costs",
      "Set appropriate timeouts and wait conditions for JavaScript-heavy pages",
      "Respect rate limits and use caching during development to reduce load"
    ],
    "anti_patterns": [
      "Using LLM extraction when CSS selectors would work (higher cost)",
      "Crawling without proper timeout settings (may hang indefinitely)",
      "Ignoring rate limits on target sites (may get blocked)"
    ],
    "faq": [
      {
        "question": "What is crawl4ai?",
        "answer": "A web crawling and data extraction library with CLI and Python SDK support."
      },
      {
        "question": "Do I need to install anything?",
        "answer": "Yes, run: pip install crawl4ai and crawl4ai-setup"
      },
      {
        "question": "Can I extract data without LLM?",
        "answer": "Yes, use CSS selector-based extraction which is faster and free."
      },
      {
        "question": "Does it handle JavaScript pages?",
        "answer": "Yes, it uses a browser and can wait for dynamic content."
      },
      {
        "question": "What output formats supported?",
        "answer": "Markdown, JSON, HTML, and extracted structured data."
      },
      {
        "question": "How to handle authentication?",
        "answer": "Configure session_id and provide credentials in browser config."
      }
    ]
  },
  "file_structure": [
    {
      "name": "references",
      "type": "dir",
      "path": "references",
      "children": [
        {
          "name": "cli-guide.md",
          "type": "file",
          "path": "references/cli-guide.md",
          "lines": 359
        },
        {
          "name": "complete-sdk-reference.md",
          "type": "file",
          "path": "references/complete-sdk-reference.md",
          "lines": 5927
        },
        {
          "name": "sdk-guide.md",
          "type": "file",
          "path": "references/sdk-guide.md",
          "lines": 391
        }
      ]
    },
    {
      "name": "scripts",
      "type": "dir",
      "path": "scripts",
      "children": [
        {
          "name": "basic_crawler.py",
          "type": "file",
          "path": "scripts/basic_crawler.py",
          "lines": 81
        },
        {
          "name": "batch_crawler.py",
          "type": "file",
          "path": "scripts/batch_crawler.py",
          "lines": 237
        },
        {
          "name": "extraction_pipeline.py",
          "type": "file",
          "path": "scripts/extraction_pipeline.py",
          "lines": 363
        },
        {
          "name": "google_search.py",
          "type": "file",
          "path": "scripts/google_search.py",
          "lines": 321
        }
      ]
    },
    {
      "name": "tests",
      "type": "dir",
      "path": "tests",
      "children": [
        {
          "name": "README.md",
          "type": "file",
          "path": "tests/README.md",
          "lines": 49
        },
        {
          "name": "run_all_tests.py",
          "type": "file",
          "path": "tests/run_all_tests.py",
          "lines": 63
        },
        {
          "name": "test_advanced_patterns.py",
          "type": "file",
          "path": "tests/test_advanced_patterns.py",
          "lines": 73
        },
        {
          "name": "test_basic_crawling.py",
          "type": "file",
          "path": "tests/test_basic_crawling.py",
          "lines": 50
        },
        {
          "name": "test_data_extraction.py",
          "type": "file",
          "path": "tests/test_data_extraction.py",
          "lines": 63
        },
        {
          "name": "test_markdown_generation.py",
          "type": "file",
          "path": "tests/test_markdown_generation.py",
          "lines": 88
        }
      ]
    },
    {
      "name": "README.md",
      "type": "file",
      "path": "README.md",
      "lines": 298
    },
    {
      "name": "SKILL.md",
      "type": "file",
      "path": "SKILL.md",
      "lines": 404
    }
  ]
}

#!/usr/bin/env python3
"""
Run all skill tests
"""
import subprocess
import sys
from pathlib import Path

def run_test(test_file):
    """Run a single test file"""
    print(f"\n{'='*60}")
    print(f"Running: {test_file}")
    print('='*60)

    result = subprocess.run(
        [sys.executable, test_file],
        capture_output=False
    )

    return result.returncode == 0

def main():
    test_dir = Path(__file__).parent
    test_files = [
        "test_basic_crawling.py",
        "test_markdown_generation.py",
        "test_data_extraction.py",
        "test_advanced_patterns.py"
    ]

    results = {}
    for test_file in test_files:
        test_path = test_dir / test_file
        if test_path.exists():
            results[test_file] = run_test(str(test_path))
        else:
            print(f"⚠️  Test file not found: {test_file}")
            results[test_file] = False

    # Summary
    print(f"\n{'='*60}")
    print("TEST SUMMARY")
    print('='*60)

    all_passed = True
    for test_file, passed in results.items():
        status = "✅ PASSED" if passed else "❌ FAILED"
        print(f"{status}: {test_file}")
        if not passed:
            all_passed = False

    print('='*60)

    if all_passed:
        print("\n✅ All tests passed!")
        return 0
    else:
        print("\n❌ Some tests failed!")
        return 1

if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""
Test advanced patterns from SKILL.md
"""
import asyncio
from crawl4ai import AsyncWebCrawler, BrowserConfig, CrawlerRunConfig

async def test_session_management():
    """Test session management"""
    print("Testing session management...")

    async with AsyncWebCrawler() as crawler:
        session_id = "test_session"

        # First crawl with session
        config1 = CrawlerRunConfig(session_id=session_id)
        result1 = await crawler.arun("https://example.com", config=config1)

        assert result1.success, f"First crawl failed: {result1.error_message}"

        # Second crawl reusing session
        config2 = CrawlerRunConfig(session_id=session_id)
        result2 = await crawler.arun("https://example.org", config=config2)

        assert result2.success, f"Second crawl failed: {result2.error_message}"

        print(f"✅ Session management works")

async def test_proxy_config():
    """Test proxy configuration in BrowserConfig"""
    print("\nTesting proxy configuration structure...")

    # Test that proxy config is in BrowserConfig (not CrawlerRunConfig)
    browser_config = BrowserConfig(
        headless=True,
        proxy_config={
            "server": "http://proxy.example.com:8080",
            "username": "user",
            "password": "pass"
        }
    )

    print(f"✅ Proxy config structure correct (in BrowserConfig)")

async def test_batch_crawling():
    """Test arun_many for batch crawling"""
    print("\nTesting batch crawling...")

    urls = ["https://example.com", "https://example.org"]

    async with AsyncWebCrawler() as crawler:
        results = await crawler.arun_many(
            urls=urls,
            max_concurrent=2
        )

        assert len(results) == 2, f"Expected 2 results, got {len(results)}"

        for result in results:
            if result.success:
                print(f"✅ {result.url}: Success")
            else:
                print(f"⚠️ {result.url}: {result.error_message}")

async def main():
    await test_session_management()
    await test_proxy_config()
    await test_batch_crawling()

if __name__ == "__main__":
    asyncio.run(main())
    print("\n✅ All advanced pattern tests passed!")

#!/usr/bin/env python3
"""
Test basic crawling examples from SKILL.md
"""
import asyncio
from crawl4ai import AsyncWebCrawler, BrowserConfig, CrawlerRunConfig

async def test_basic_crawl():
    """Test basic crawling setup"""
    print("Testing basic crawl setup...")

    # Test from SKILL.md Section 1
    browser_config = BrowserConfig(
        headless=True,
        viewport_width=1920,
        viewport_height=1080,
        user_agent="custom-agent"
    )

    crawler_config = CrawlerRunConfig(
        page_timeout=30000,
        screenshot=True,
        remove_overlay_elements=True
    )

    async with AsyncWebCrawler(config=browser_config) as crawler:
        result = await crawler.arun(
            url="https://example.com",
            config=crawler_config
        )

        # Verify result attributes
        assert result.success, f"Crawl failed: {result.error_message}"
        assert hasattr(result, 'html'), "Missing html attribute"
        assert hasattr(result, 'markdown'), "Missing markdown attribute"
        assert hasattr(result, 'links'), "Missing links attribute"

        # Test markdown as string (StringCompatibleMarkdown)
        markdown_str = str(result.markdown)
        assert len(markdown_str) > 0, "Markdown is empty"

        print(f"✅ Success: {result.success}")
        print(f"✅ HTML length: {len(result.html)}")
        print(f"✅ Markdown length: {len(markdown_str)}")
        print(f"✅ Links found: {len(result.links)}")

if __name__ == "__main__":
    asyncio.run(test_basic_crawl())
    print("\n✅ All basic crawling tests passed!")

#!/usr/bin/env python3
"""
Test data extraction examples from SKILL.md
"""
import asyncio
import json
from crawl4ai import AsyncWebCrawler, CrawlerRunConfig
from crawl4ai.extraction_strategy import JsonCssExtractionStrategy, LLMExtractionStrategy

async def test_manual_schema_extraction():
    """Test manual CSS/JSON schema extraction"""
    print("Testing manual schema extraction...")

    # Schema from SKILL.md
    schema = {
        "name": "articles",
        "baseSelector": "body",  # Using body since example.com is simple
        "fields": [
            {"name": "title", "selector": "h1", "type": "text"},
            {"name": "paragraphs", "selector": "p", "type": "text", "all": True}
        ]
    }

    extraction_strategy = JsonCssExtractionStrategy(schema=schema)
    config = CrawlerRunConfig(extraction_strategy=extraction_strategy)

    async with AsyncWebCrawler() as crawler:
        result = await crawler.arun("https://example.com", config=config)

        assert result.success, f"Crawl failed: {result.error_message}"
        assert result.extracted_content, "No extracted content"

        data = json.loads(result.extracted_content)
        assert isinstance(data, list) or isinstance(data, dict), "Invalid extraction format"

        print(f"✅ Manual schema extraction works")
        print(f"   Extracted data type: {type(data)}")

async def test_llm_extraction():
    """Test LLM-based extraction (requires API key in env)"""
    print("\nTesting LLM extraction structure...")

    try:
        # Just test that the strategy can be created
        extraction_strategy = LLMExtractionStrategy(
            provider="openai/gpt-4o-mini",
            instruction="Extract key financial metrics"
        )

        config = CrawlerRunConfig(extraction_strategy=extraction_strategy)
        print(f"✅ LLMExtractionStrategy created successfully")

    except Exception as e:
        print(f"✅ LLMExtractionStrategy structure verified (API key not tested)")

async def main():
    await test_manual_schema_extraction()
    await test_llm_extraction()

if __name__ == "__main__":
    asyncio.run(main())
    print("\n✅ All data extraction tests passed!")

#!/usr/bin/env python3
"""
Test markdown generation examples from SKILL.md
"""
import asyncio
from crawl4ai import AsyncWebCrawler, CrawlerRunConfig
from crawl4ai.content_filter_strategy import PruningContentFilter, BM25ContentFilter
from crawl4ai.markdown_generation_strategy import DefaultMarkdownGenerator

async def test_basic_markdown():
    """Test basic markdown extraction"""
    print("Testing basic markdown extraction...")

    async with AsyncWebCrawler() as crawler:
        result = await crawler.arun("https://example.com")

        # result.markdown is StringCompatibleMarkdown
        markdown_str = str(result.markdown)
        assert len(markdown_str) > 0, "Markdown is empty"
        print(f"✅ Basic markdown length: {len(markdown_str)}")

async def test_fit_markdown_with_filters():
    """Test Fit Markdown with content filters"""
    print("\nTesting Fit Markdown with filters...")

    # Test BM25 filter
    bm25_filter = BM25ContentFilter(
        user_query="example domain",
        bm25_threshold=1.0
    )

    md_generator = DefaultMarkdownGenerator(content_filter=bm25_filter)
    config = CrawlerRunConfig(markdown_generator=md_generator)

    async with AsyncWebCrawler() as crawler:
        result = await crawler.arun("https://example.com", config=config)

        # Access both raw and fit markdown
        assert hasattr(result.markdown, 'raw_markdown'), "Missing raw_markdown attribute"
        assert hasattr(result.markdown, 'fit_markdown'), "Missing fit_markdown attribute"

        print(f"✅ Raw markdown length: {len(result.markdown.raw_markdown)}")
        print(f"✅ Fit markdown length: {len(result.markdown.fit_markdown or '')}")

async def test_pruning_filter():
    """Test Pruning filter"""
    print("\nTesting Pruning filter...")

    pruning_filter = PruningContentFilter(threshold=0.4, threshold_type="fixed")
    md_generator = DefaultMarkdownGenerator(content_filter=pruning_filter)
    config = CrawlerRunConfig(markdown_generator=md_generator)

    async with AsyncWebCrawler() as crawler:
        result = await crawler.arun("https://example.com", config=config)

        assert result.success, f"Crawl failed: {result.error_message}"
        print(f"✅ Pruning filter works")

async def test_markdown_options():
    """Test markdown generator options"""
    print("\nTesting markdown generator options...")

    generator = DefaultMarkdownGenerator(
        options={
            "ignore_links": False,
            "ignore_images": False,
            "image_alt_text": True
        }
    )

    config = CrawlerRunConfig(markdown_generator=generator)

    async with AsyncWebCrawler() as crawler:
        result = await crawler.arun("https://example.com", config=config)

        assert result.success, f"Crawl failed: {result.error_message}"
        print(f"✅ Markdown options work")

async def main():
    await test_basic_markdown()
    await test_fit_markdown_with_filters()
    await test_pruning_filter()
    await test_markdown_options()

if __name__ == "__main__":
    asyncio.run(main())
    print("\n✅ All markdown generation tests passed!")