biz-bud/examples/firecrawl_enhanced_usage.py

"""Example usage of enhanced Firecrawl API endpoints."""

import asyncio
from typing import Any, cast

from bb_tools.api_clients.firecrawl import (
    CrawlOptions,
    ExtractOptions,
    FirecrawlApp,
    FirecrawlOptions,
    MapOptions,
    SearchOptions,
)


async def example_map_website():
    """Demonstrate using the map endpoint to discover URLs."""
    async with FirecrawlApp() as app:
        # Map a website to discover all URLs
        map_options = MapOptions(
            limit=50,
            include_subdomains=False,
            search="documentation",  # Optional: filter URLs containing "documentation"
        )

        urls = await app.map_website("https://example.com", options=map_options)
        print(f"Discovered {len(urls)} URLs")
        for url in urls[:5]:
            print(f"  - {url}")


async def example_crawl_website():
    """Demonstrate using the crawl endpoint for deep website crawling."""
    async with FirecrawlApp() as app:
        # Crawl a website with depth control
        crawl_options = CrawlOptions(
            limit=20,
            max_depth=2,
            include_paths=[r"/docs/.*", r"/api/.*"],
            exclude_paths=[r".*\.pdf$", r".*/archive/.*"],
            scrape_options=FirecrawlOptions(
                formats=["markdown", "links"],
                only_main_content=True,
            ),
        )

        result = await app.crawl_website(
            "https://example.com",
            options=crawl_options,
            wait_for_completion=True,
        )

        if isinstance(result, dict) and "data" in result:
            data = result["data"]
            if isinstance(data, list):
                print(f"Crawled {len(data)} pages")
                for page in data[:3]:
                    if isinstance(page, dict):
                        metadata = page.get("metadata", {})
                        title = (
                            metadata.get("title", "N/A") if isinstance(metadata, dict) else "N/A"
                        )
                        content = page.get("content", "")
                        print(f"  - Title: {title}")
                        if isinstance(content, str):
                            print(f"    Content preview: {content[:100]}...")


async def example_search_and_scrape():
    """Demonstrate using the search endpoint to search and scrape results."""
    async with FirecrawlApp() as app:
        # Search the web and scrape results
        search_options = SearchOptions(
            limit=5,
            tbs="qdr:w",  # Last week
            location="US",
            scrape_options=FirecrawlOptions(
                formats=["markdown"],
                only_main_content=True,
            ),
        )

        results = await app.search("RAG implementation best practices", options=search_options)
        print(f"Found and scraped {len(results)} search results")

        for i, result in enumerate(results):
            if result:
                print(f"\n{i + 1}. {result.get('title', 'No title')}")
                print(f"   URL: {result.get('url', 'No URL')}")
                markdown = result.get("markdown")
                if markdown and isinstance(markdown, str):
                    print(f"   Content preview: {markdown[:200]}...")


async def example_extract_structured_data():
    """Demonstrate using the extract endpoint for AI-powered extraction."""
    async with FirecrawlApp() as app:
        # Extract structured data from multiple URLs
        urls = [
            "https://example.com/company/about",
            "https://example.com/company/team",
            "https://example.com/company/careers",
        ]

        # Option 1: Using a prompt
        extract_options = ExtractOptions(
            prompt="Extract company information including: company name, founded year, number of employees, main products/services, and key team members with their roles.",
        )

        result = await app.extract(urls, options=extract_options)
        if result.get("success"):
            print("Extracted company information:")
            print(result.get("data", {}))

        # Option 2: Using a schema
        schema_options = ExtractOptions(
            extract_schema={
                "type": "object",
                "properties": {
                    "company_name": {"type": "string"},
                    "founded_year": {"type": "integer"},
                    "employees": {"type": "integer"},
                    "products": {"type": "array", "items": {"type": "string"}},
                    "team_members": {
                        "type": "array",
                        "items": {
                            "type": "object",
                            "properties": {
                                "name": {"type": "string"},
                                "role": {"type": "string"},
                            },
                        },
                    },
                },
            }
        )

        structured_result = await app.extract(urls, options=schema_options)
        if structured_result.get("success"):
            print("\nStructured extraction result:")
            print(structured_result.get("data", {}))


async def example_rag_integration():
    """Demonstrate using Firecrawl for RAG pipeline."""
    async with FirecrawlApp() as app:
        base_url = "https://docs.example.com"

        # Step 1: Map the documentation site
        print("Step 1: Discovering documentation pages...")
        map_options = MapOptions(limit=100, sitemap_only=True)
        all_urls = await app.map_website(base_url, options=map_options)

        # Step 2: Crawl and extract content
        print(f"\nStep 2: Crawling {len(all_urls)} pages...")
        crawl_options = CrawlOptions(
            limit=50,
            scrape_options=FirecrawlOptions(
                formats=["markdown"],
                only_main_content=True,
                exclude_tags=["nav", "footer", "header"],
            ),
        )

        crawl_result = await app.crawl_website(base_url, options=crawl_options)

        # Step 3: Process for RAG
        if isinstance(crawl_result, dict) and "data" in crawl_result:
            data = crawl_result["data"]
            if isinstance(data, list):
                print(f"\nStep 3: Processing {len(data)} pages for RAG...")
                documents = []
                for page in data:
                    if isinstance(page, dict) and page.get("markdown"):
                        page_metadata = page.get("metadata", {})
                        if isinstance(page_metadata, dict):
                            # Cast to Any to work around pyrefly type inference
                            metadata_dict = cast("Any", page_metadata)
                            documents.append(
                                {
                                    "content": page["markdown"],
                                    "metadata": {
                                        "source": base_url,
                                        "title": metadata_dict.get("title", ""),
                                        "description": metadata_dict.get("description", ""),
                                    },
                                }
                            )

                print(f"Ready to index {len(documents)} documents into vector store")
                return documents


async def main():
    """Run all examples."""
    print("=== Firecrawl Enhanced API Examples ===\n")

    # Uncomment the examples you want to run:

    # await example_map_website()
    # await example_crawl_website()
    # await example_search_and_scrape()
    # await example_extract_structured_data()
    # await example_rag_integration()

    print("\nNote: Set FIRECRAWL_API_KEY environment variable before running!")


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