crawl_and_extract_with_xai_grok.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "private_outputs": true,
      "provenance": [],
      "include_colab_link": true
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/alexfazio/firecrawl-quickstart/blob/main/web_crawler_grok_firecrawl.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "intro-section"
      },
      "source": [
        "# Building a Web Crawler with Grok-2 and Firecrawl\n",
        "\n",
        "By Alex Fazio (https://twitter.com/alxfazio)\n",
        "\n",
        "Github repo: https://github.com/alexfazio/firecrawl-cookbook\n",
        "\n",
        "This Jupyter notebook demonstrates how to combine Grok-2's language model capabilities with Firecrawl's web scraping features to build an intelligent web crawler that can extract structured data from websites.\n",
        "\n",
        "By the end of this notebook, you'll be able to:\n",
        "\n",
        "1. Set up the Grok-2 and Firecrawl environment\n",
        "2. Build a targeted web crawler that understands content\n",
        "3. Extract and process structured data from websites\n",
        "4. Export the processed content in JSON format\n",
        "\n",
        "This cookbook is designed for developers and data scientists who want to build advanced web crawlers with AI-powered content understanding."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "9rQmTgiMVk5X"
      },
      "source": [
        "## Setup\n",
        "\n",
        "First, let's install the required packages:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "zBE3KvuKVk5X"
      },
      "outputs": [],
      "source": [
        "%pip install firecrawl-py requests python-dotenv --quiet"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "q2GWsM_gVk5X"
      },
      "outputs": [],
      "source": [
        "import os\n",
        "import json\n",
        "import requests\n",
        "from dotenv import load_dotenv\n",
        "from firecrawl import FirecrawlApp"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "VqpYy-sxVk5Y"
      },
      "source": [
        "## Initialize Environment\n",
        "\n",
        "Enter your API keys securely:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "Wrec0L1sVk5Y"
      },
      "outputs": [],
      "source": [
        "from getpass import getpass\n",
        "\n",
        "# Securely get API keys\n",
        "grok_api_key = getpass(\"Enter your Grok-2 API key: \")\n",
        "firecrawl_api_key = getpass(\"Enter your Firecrawl API key: \")\n",
        "\n",
        "# Initialize FirecrawlApp\n",
        "app = FirecrawlApp(api_key=firecrawl_api_key)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "uiFw3MfjVk5Y"
      },
      "source": [
        "## Define Grok-2 API Interaction\n",
        "\n",
        "Let's create a function to handle interactions with the Grok-2 API, including comprehensive error handling and debugging information:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "PTc5bc85Vk5Y"
      },
      "outputs": [],
      "source": [
        "def grok_completion(prompt):\n",
        "    url = \"https://api.x.ai/v1/chat/completions\"\n",
        "    headers = {\n",
        "        \"Content-Type\": \"application/json\",\n",
        "        \"Authorization\": f\"Bearer {grok_api_key}\"\n",
        "    }\n",
        "    data = {\n",
        "        \"messages\": [\n",
        "            {\"role\": \"system\", \"content\": \"You are a helpful assistant.\"},\n",
        "            {\"role\": \"user\", \"content\": prompt}\n",
        "        ],\n",
        "        \"model\": \"grok-beta\",\n",
        "        \"stream\": False,\n",
        "        \"temperature\": 0\n",
        "    }\n",
        "\n",
        "    try:\n",
        "        response = requests.post(url, headers=headers, json=data)\n",
        "        print(f\"\\nAPI Response Status Code: {response.status_code}\")\n",
        "\n",
        "        if response.status_code != 200:\n",
        "            print(f\"Error Response: {response.text}\")\n",
        "            return None\n",
        "\n",
        "        response_data = response.json()\n",
        "        print(\"\\nFull API Response:\")\n",
        "        print(json.dumps(response_data, indent=2))\n",
        "\n",
        "        if 'choices' not in response_data:\n",
        "            print(\"\\nWarning: 'choices' key not found in response\")\n",
        "            print(\"Available keys:\", list(response_data.keys()))\n",
        "            return None\n",
        "\n",
        "        if not response_data['choices']:\n",
        "            print(\"\\nWarning: 'choices' array is empty\")\n",
        "            return None\n",
        "\n",
        "        choice = response_data['choices'][0]\n",
        "        if 'message' not in choice:\n",
        "            print(\"\\nWarning: 'message' key not found in first choice\")\n",
        "            print(\"Available keys in choice:\", list(choice.keys()))\n",
        "            return None\n",
        "\n",
        "        if 'content' not in choice['message']:\n",
        "            print(\"\\nWarning: 'content' key not found in message\")\n",
        "            print(\"Available keys in message:\", list(choice['message'].keys()))\n",
        "            return None\n",
        "\n",
        "        return choice['message']['content']\n",
        "\n",
        "    except requests.exceptions.RequestException as e:\n",
        "        print(f\"\\nRequest Error: {str(e)}\")\n",
        "        return None\n",
        "    except json.JSONDecodeError as e:\n",
        "        print(f\"\\nJSON Decode Error: {str(e)}\")\n",
        "        print(\"Raw Response:\", response.text)\n",
        "        return None\n",
        "    except Exception as e:\n",
        "        print(f\"\\nUnexpected Error: {str(e)}\")\n",
        "        return None"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Wecuh9TbVk5Y"
      },
      "source": [
        "## Website Crawling Functions\n",
        "\n",
        "This function combines Grok-2's understanding with Firecrawl's search capabilities to find relevant pages. It:\n",
        "\n",
        "1. Uses Grok-2 to distill the user's objective into a focused search term\n",
        "2. Enforces strict formatting rules for consistent search terms\n",
        "3. Cleans and normalizes the search output\n",
        "4. Uses Firecrawl's map endpoint to discover relevant pages\n",
        "\n",
        "The function takes a broad objective (e.g., \"Find articles about startup investments\") and converts it into an optimized search term (e.g., \"startup funding\") to ensure targeted results.\n",
        "\n",
        "Note: The function limits search terms to 2 words maximum for optimal performance with Firecrawl's search algorithm."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "cGA5X6PDVk5Y"
      },
      "outputs": [],
      "source": [
        "def find_relevant_pages(objective, url):\n",
        "    prompt = f\"\"\"Based on the objective '{objective}', provide ONLY a 1-2 word search term to locate relevant information on the website.\n",
        "\n",
        "Rules:\n",
        "- Return ONLY the search term, nothing else\n",
        "- Maximum 2 words\n",
        "- No punctuation or formatting\n",
        "- No explanatory text\"\"\"\n",
        "\n",
        "    search_term = grok_completion(prompt)\n",
        "\n",
        "    if search_term is None:\n",
        "        print(\"Failed to get search term from Grok-2 API\")\n",
        "        return []\n",
        "\n",
        "    # Clean up the search term\n",
        "    search_term = search_term.strip().replace('\"', '').replace('*', '')\n",
        "    words = search_term.split()\n",
        "    if len(words) > 2:\n",
        "        search_term = \" \".join(words[:2])\n",
        "\n",
        "    print(f\"Using search term: '{search_term}'\")\n",
        "\n",
        "    try:\n",
        "        map_result = app.map_url(url, params={\"search\": search_term})\n",
        "        return map_result.get(\"links\", [])\n",
        "    except Exception as e:\n",
        "        print(f\"Error mapping URL: {str(e)}\")\n",
        "        return []"
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "## Content Extraction and Processing\n",
        "\n",
        "This function handles the extraction and intelligent processing of content from each webpage. It:\n",
        "\n",
        "1. Scrapes content from each relevant page\n",
        "2. Uses Grok-2 to analyze the content against our objective\n",
        "3. Extracts structured data in JSON format\n",
        "4. Handles various edge cases and errors\n",
        "\n",
        "The function processes up to 3 pages and returns the first successful match, using Grok-2 to determine relevance and extract specific data points."
      ],
      "metadata": {
        "id": "XSuxErSsYH1L"
      }
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "lffHeKe-Vk5Y"
      },
      "outputs": [],
      "source": [
        "def extract_data_from_pages(links, objective):\n",
        "    for link in links[:3]:\n",
        "        try:\n",
        "            print(f\"\\nProcessing link: {link}\")\n",
        "            scrape_result = app.scrape_url(link, params={'formats': ['markdown']})\n",
        "            content = scrape_result.get('markdown', '')\n",
        "\n",
        "            if not content:\n",
        "                print(\"No content extracted from page\")\n",
        "                continue\n",
        "\n",
        "            prompt = f\"\"\"Given the following content, extract the information related to the objective '{objective}' in JSON format. If not found, reply 'Objective not met'.\n",
        "\n",
        "Content: {content}\n",
        "\n",
        "Remember:\n",
        "- Only return JSON if the objective is met.\n",
        "- Do not include any extra text or markdown formatting.\n",
        "- Do not wrap the JSON in code blocks.\n",
        "\"\"\"\n",
        "            result = grok_completion(prompt)\n",
        "\n",
        "            if result is None:\n",
        "                print(\"Failed to get response from Grok-2 API\")\n",
        "                continue\n",
        "\n",
        "            result = result.strip()\n",
        "\n",
        "            # Handle case where response is wrapped in code blocks\n",
        "            if result.startswith(\"```\") and result.endswith(\"```\"):\n",
        "                # Remove the code block markers and any language identifier\n",
        "                result = result.split(\"\\n\", 1)[1].rsplit(\"\\n\", 1)[0]\n",
        "\n",
        "            if result != \"Objective not met\":\n",
        "                try:\n",
        "                    data = json.loads(result)\n",
        "                    return data\n",
        "                except json.JSONDecodeError as e:\n",
        "                    print(f\"Error parsing JSON response: {str(e)}\")\n",
        "                    print(\"Raw response:\", result)\n",
        "                    continue\n",
        "            else:\n",
        "                print(\"Objective not met for this page\")\n",
        "\n",
        "        except Exception as e:\n",
        "            print(f\"Error processing page: {str(e)}\")\n",
        "            continue\n",
        "\n",
        "    return None"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "hT7hScXvVk5Y"
      },
      "source": [
        "## Main Execution\n",
        "\n",
        "Let's create and run the main function that ties everything together:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "C_wDgTBOVk5Y"
      },
      "outputs": [],
      "source": [
        "import pprint\n",
        "\n",
        "def main():\n",
        "    url = input(\"Enter the website URL to crawl: \")\n",
        "    objective = input(\"Enter your data extraction objective: \")\n",
        "\n",
        "    print(\"\\nFinding relevant pages...\")\n",
        "    links = find_relevant_pages(objective, url)\n",
        "\n",
        "    if not links:\n",
        "        print(\"No relevant pages found.\")\n",
        "        return\n",
        "\n",
        "    print(f\"\\nFound {len(links)} relevant pages:\")\n",
        "    for i, link in enumerate(links[:3], 1):\n",
        "        pprint.pprint(f\"{i}. {link}\")\n",
        "\n",
        "    print(\"\\nExtracting data from pages...\")\n",
        "    data = extract_data_from_pages(links, objective)\n",
        "\n",
        "    if data:\n",
        "        print(\"\\nData extracted successfully:\")\n",
        "        pprint.pprint(json.dumps(data, indent=2))\n",
        "    else:\n",
        "        print(\"Could not find data matching the objective.\")"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "TtIGh9jSVk5Z"
      },
      "outputs": [],
      "source": [
        "# Run the crawler\n",
        "main()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "PZPOttGoVk5Z"
      },
      "source": [
        "## What's Next?\n",
        "\n",
        "Now that you have a working web crawler, consider these enhancements:\n",
        "\n",
        "1. Add error handling and retries\n",
        "2. Implement concurrent processing\n",
        "3. Add content filtering and validation\n",
        "4. Create custom extraction rules\n",
        "\n",
        "The combination of Grok-2 and Firecrawl offers powerful possibilities for intelligent web scraping and content analysis.\n",
        "\n",
        "## Additional Resources\n",
        "\n",
        "- [x.ai Grok-2 API Documentation](https://api.x.ai/docs)\n",
        "- [Firecrawl Python Library Documentation](https://docs.firecrawl.dev)\n",
        "- [Example Code Repository](https://github.com/example/web-crawler)"
      ]
    }
  ]
}