LLM-Scraper-Notebook.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "9fc17c25",
   "metadata": {},
   "source": [
    "## Libraries to be imported \n",
    "\n",
    "1. **os** -> os is used to interact with system files and securely handle sensitive information \n",
    "<br><br>\n",
    "2. **time** -> used for sleeping the code \n",
    "<br><br>\n",
    "3. **selenium** -> This library is used for web automation. In this code we will use it for Automating browser                        action ,Handling dynamic web contents and Extracting the RAW HTML file.\n",
    "<br><br>\n",
    "4. **json** -> request , get , fetch \n",
    "<br><br>\n",
    "5. **transformers** -> the LLM and main worker."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2c2bc9e1",
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import time\n",
    "from selenium import webdriver\n",
    "from selenium.webdriver.common.by import By # Use to locate elements on the page , \n",
    "                                            # It replicates the getElementByID function of javascript \n",
    "from selenium.webdriver.common.keys import Keys\n",
    "from selenium.common.exceptions import NoSuchElementException\n",
    "import json\n",
    "from transformers import pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f7f59f32",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Initialize the web driver , we will be using chrome \n",
    "# Also if your chrome has adblock enabled it will also get rid of unnecessary scripts \n",
    "\n",
    "driver = webdriver.Chrome()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "442db092",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Function to check if user is already logged in\n",
    "\n",
    "def is_user_logged_in():\n",
    "    try:\n",
    "        # Check for user profile information or sign-out button\n",
    "        driver.find_element(By.XPATH, \"//span[@class='nav-line-3']\")\n",
    "        return True\n",
    "    except NoSuchElementException:\n",
    "        return False"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5413ef29",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Function to login to Amazon \n",
    "\n",
    "def amazon_login( username , password ):\n",
    "    \n",
    "    print(\" Logging you in ... \")\n",
    "    print(\" ... \")\n",
    "    \n",
    "    # Driver go to the website \n",
    "    driver.get(\"https://www.amazon.com/-/es/ap/signin?openid.pape.max_auth_age=3600&openid.return_to=https://www.amazon.com/myh/households?language=es&openid.identity=http://specs.openid.net/auth/2.0/identifier_select&openid.assoc_handle=usflex&openid.mode=checkid_setup&language=en_US&openid.claimed_id=http://specs.openid.net/auth/2.0/identifier_select&openid.ns=http://specs.openid.net/auth/2.0\")\n",
    "    # driver.get(\"https://www.amazon.com/ap/signin\")\n",
    "    \n",
    "    # wait for the page to load \n",
    "    time.sleep(2)\n",
    "    \n",
    "    # if user is already logged in\n",
    "    if is_user_logged_in():\n",
    "        print(\"You are already logged in.\")\n",
    "        return\n",
    "    \n",
    "    \n",
    "    # Check if CAPTCHA input is required\n",
    "    captcha_required = False\n",
    "    try:\n",
    "        captcha_input = driver.find_element(By.ID, \"captchacharacters\")\n",
    "        captcha_required = True\n",
    "    except NoSuchElementException:\n",
    "        pass\n",
    "\n",
    "    if captcha_required:\n",
    "        # Pause execution and prompt the user to solve the CAPTCHA manually\n",
    "        input(\"Please solve the CAPTCHA manually, then press Enter to continue...\")\n",
    "        # Once the CAPTCHA is solved, press Enter to proceed\n",
    "        captcha_input.send_keys(Keys.ENTER)\n",
    "        time.sleep(5)  # Add a delay to allow the page to load after submitting CAPTCHA\n",
    "    \n",
    "    #############################################################################\n",
    "    ########  Please read the README.md if you are stuck at this point  #########\n",
    "    #############################################################################\n",
    "        \n",
    "    # Takes in username and perform the actions required \n",
    "    username_input = driver.find_element(By.ID, \"ap_email\")\n",
    "    username_input.send_keys(username)\n",
    "    driver.find_element(By.ID, \"continue\").click()\n",
    "    \n",
    "    # wait for the page to load \n",
    "    time.sleep(2)\n",
    "    \n",
    "    # Takes in password and perform the actions required \n",
    "    password_input = driver.find_element(By.ID, \"ap_password\")\n",
    "    password_input.send_keys(password)\n",
    "    driver.find_element(By.ID, \"signInSubmit\").click()\n",
    "    \n",
    "    # wait for the page to load \n",
    "    time.sleep(5)\n",
    "    \n",
    "    \n",
    "    # Check if OTP input is required\n",
    "    otp_input_required = False\n",
    "    try:\n",
    "        otp_input = driver.find_element(By.ID, \"auth-mfa-otpcode\")\n",
    "        otp_input_required = True\n",
    "    except NoSuchElementException:\n",
    "        pass\n",
    "\n",
    "    if otp_input_required:\n",
    "        otp = input(\"Enter the OTP received on your email/phone: \")\n",
    "        otp_input.send_keys(otp)\n",
    "        \n",
    "        driver.find_element(By.ID, \"auth-signin-button\").click()\n",
    "        \n",
    "        time.sleep(5)\n",
    "        \n",
    "        \n",
    "    print(\" Logging complete ... \")\n",
    "    print(\" ... \")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "aac4015a",
   "metadata": {},
   "outputs": [],
   "source": [
    "# To handle unexpected error during login we can make use of LLM \n",
    "# Here we are using free to use model which is availaible in Hugging face's transformers library \n",
    "\n",
    "def handle_unexpected_elements(page_source):\n",
    "    \n",
    "    prompt_text = f\"You are navigating an order details page and encountered the following HTML: {page_source}. What actions should you take to continue extracting order details?\"\n",
    "    \n",
    "    # Loading the free-to-use GPT-Neo model\n",
    "    generator = pipeline(\"text-generation\", model=\"EleutherAI/gpt-neo-2.7B\")\n",
    "    response = generator(prompt_text, max_length=150, num_return_sequences=1)\n",
    "    \n",
    "    # Extract and return the generated response\n",
    "    actions = response[0]['generated_text'].strip()\n",
    "    return actions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8d44255b",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Function to get the RAW HTML file\n",
    "# It takes in the link to order page and return the RAW HTML file\n",
    "\n",
    "def get_source_page(driver, order_link):\n",
    "    \n",
    "    driver.get(order_link)\n",
    "    \n",
    "    # wait for the page to load \n",
    "    time.sleep(3)\n",
    "    \n",
    "    page_source = driver.page_source\n",
    "    \n",
    "    # Use GPT-Neo to handle any unexpected elements\n",
    "    actions = handle_unexpected_elements(page_source)\n",
    "    \n",
    "    if \"click the 'Close' button\" in actions:\n",
    "        \n",
    "        close_button = driver.find_element(By.XPATH, \"//button[text()='Close']\")\n",
    "        close_button.click()\n",
    "        \n",
    "        time.sleep(2)\n",
    "    \n",
    "    # Save the page HTML\n",
    "    order_html = driver.page_source\n",
    "    return order_html"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2bb14384",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Function to navigate through all the orders , get the HTML files for those , \n",
    "# and write in another file to give it to LLM \n",
    "\n",
    "def navigate_and_fetch_all_orders():\n",
    "    \n",
    "    print(\" Getting your order history ... \")\n",
    "    print(\" ... \")\n",
    "    \n",
    "    driver.get(\"https://www.amazon.com/gp/your-account/order-history\")\n",
    "    time.sleep(5)\n",
    "\n",
    "    all_order_html = \"\"  # Initialize an empty string to store all order HTML\n",
    "    \n",
    "    \n",
    "    orders = driver.find_elements(By.CLASS_NAME, \"order\")\n",
    "    for order in orders:\n",
    "        order_link = order.find_element(By.CSS_SELECTOR, \"a.a-link-normal\").get_attribute(\"href\")\n",
    "        order_html = fetch_order_details_with_llm(driver, order_link)\n",
    "        all_order_html += order_html  # Append current order HTML to the string\n",
    "\n",
    "        time.sleep(3)\n",
    "\n",
    "        ########################################################################################\n",
    "        # Decomment this if there are several pages , by default it only select the shown page # \n",
    "        ########################################################################################\n",
    "\n",
    "#     while True:\n",
    "#         orders = driver.find_elements(By.CLASS_NAME, \"order\")\n",
    "#         for order in orders:\n",
    "#             order_link = order.find_element(By.CSS_SELECTOR, \"a.a-link-normal\").get_attribute(\"href\")\n",
    "#             order_html = fetch_order_details_with_llm(driver, order_link)\n",
    "#             all_order_html += order_html  # Append current order HTML to the string\n",
    "\n",
    "#             time.sleep(3)\n",
    "\n",
    "#         # Check if there's a next page\n",
    "#         next_button = driver.find_element(By.CSS_SELECTOR, \".a-pagination li.a-last a\")\n",
    "#         if \"disabled\" in next_button.get_attribute(\"class\"):\n",
    "#             break  # Exit loop if there's no next page\n",
    "\n",
    "#         # Click on the next page button\n",
    "#         next_button.click()\n",
    "#         time.sleep(5)  # Add a delay to ensure the page is loaded\n",
    "\n",
    "    # Write all order HTML to a single file\n",
    "    with open(\"all_orders.html\", \"w\", encoding=\"utf-8\") as file:\n",
    "        file.write(all_order_html)\n",
    "        \n",
    "    print(\" Saved the RAW files ... \")\n",
    "    print(\" ... \")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fd214ab8",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Function to take in HTML file as input , give it to LLM and return the required output \n",
    "\n",
    "def extract_order_details( html_content ):\n",
    "    prompt_text = f\"Extract order details such as order number, product names, quantities, prices, and delivery status from the following HTML: {html_content}\"\n",
    "    \n",
    "    print(\" Using LLM fetching your data ... \")\n",
    "    print(\" ... \")\n",
    "    \n",
    "    # Use the LLM\n",
    "    generator = pipeline(\"text-generation\", model=\"EleutherAI/gpt-neo-2.7B\")\n",
    "    response = generator(prompt_text, max_length=300, num_return_sequences=1)\n",
    "    \n",
    "    # Extract and return the generated order details\n",
    "    order_details = response[0]['generated_text'].strip()\n",
    "    return order_details"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e0fdbb39",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Function to save order details in required format in another file \n",
    "\n",
    "def save_order_details():\n",
    "    \n",
    "    with open(\"all_orders.html\", \"r\", encoding=\"utf-8\") as html_file:\n",
    "        html_content = html_file.read()\n",
    "\n",
    "    order_data = extract_order_details(html_content)\n",
    "\n",
    "    with open(\"orders_data.json\", \"w\", encoding=\"utf-8\") as json_file:\n",
    "        json.dump(order_data, json_file, indent=4)\n",
    "        \n",
    "    print(\" Succesfully saved ... \")\n",
    "    print(\" ... \")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a61a8539",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Main execution flow\n",
    "\n",
    "USERNAME = input(\"Enter your Amazon username: \")\n",
    "PASSWORD = input(\"Enter your Amazon password: \")\n",
    "\n",
    "# Decomment this for sensitive information\n",
    "\n",
    "# # Load sensitive information from environment variables\n",
    "# AMAZON_USERNAME = os.getenv(\"AMAZON_USERNAME\")\n",
    "# AMAZON_PASSWORD = os.getenv(\"AMAZON_PASSWORD\")\n",
    "\n",
    "amazon_login(USERNAME, PASSWORD)\n",
    "navigate_and_fetch_all_orders()\n",
    "save_order_details()\n",
    "\n",
    "# Close the Driver\n",
    "\n",
    "driver.quit()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e11bcd30",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f2f55296",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "79055672",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}