ChatGPT Scraping - How to Vibe Scrape with ChatGPT

import requests
from bs4 import BeautifulSoup

def get_refurbished_iphone_urls():
    url = "<https://www.apple.com/shop/refurbished/iphone>"
    headers = {"User-Agent": "Mozilla/5.0"}
    response = requests.get(url, headers=headers)

    if response.status_code != 200:
        print("Failed to retrieve page")
        return []

    soup = BeautifulSoup(response.text, "html.parser")
    product_links = []

    for link in soup.find_all("a", href=True):
        href = link["href"]
        if "/shop/product/" in href:
            full_url = f"<https://www.apple.com>{href}"
            product_links.append(full_url)

    return product_links

# Example usage
if __name__ == "__main__":
    urls = get_refurbished_iphone_urls()
    print(urls)

import asyncio
import aiohttp
import pandas as pd
from bs4 import BeautifulSoup

async def fetch_page(session, url):
    """Fetches the HTML content of a given URL."""
    async with session.get(url) as response:
        return await response.text()

async def scrape_iphone_details(url):
    """Scrapes details of an iPhone from the given Apple Refurbished Store product URL."""
    async with aiohttp.ClientSession() as session:
        html = await fetch_page(session, url)
        soup = BeautifulSoup(html, 'html.parser')

        try:
            phone_name = soup.find("h1", class_="product-title").text.strip()
            price = soup.find("span", class_="as-price-currentprice").text.strip()
            stock_status = "In Stock" if "Add to Bag" in html else "Out of Stock"

            details_section = soup.find("div", class_="tech-specs")
            details_list = details_section.find_all("li") if details_section else []

            details = {}
            for item in details_list:
                text = item.text.strip()
                if "display" in text.lower():
                    details["Screen Size"] = text.split(":")[-1].strip()
                elif "dimensions" in text.lower():
                    details["Dimensions"] = text.split(":")[-1].strip()
                elif "battery" in text.lower():
                    details["Battery Capacity"] = text.split(":")[-1].strip()
                elif "5g" in text.lower():
                    details["5G Availability"] = "Yes" if "5G" in text else "No"
                elif "usb-c" in text.lower():
                    details["USB-C Charging"] = "Yes" if "USB-C" in text else "No"
                elif "release year" in text.lower():
                    details["Year of Release"] = text.split(":")[-1].strip()

            return {
                "Input URL": url,
                "Phone Model Name": phone_name,
                "Operating System": "iOS (Check Apple site for current version)",
                "Screen Size": details.get("Screen Size", "Unknown"),
                "Dimensions": details.get("Dimensions", "Unknown"),
                "Battery Capacity": details.get("Battery Capacity", "Unknown"),
                "USB-C Charging": details.get("USB-C Charging", "No"),
                "5G Availability": details.get("5G Availability", "Unknown"),
                "Price": price,
                "Stock Status": stock_status,
                "Year of Release": details.get("Year of Release", "Unknown"),
            }

        except Exception as e:
            return {"Error": f"Failed to scrape {url} - {str(e)}"}

async def scrape_multiple_iphones(urls):
    """Scrapes details of multiple iPhones in parallel using asyncio."""
    tasks = [scrape_iphone_details(url) for url in urls]
    return await asyncio.gather(*tasks)

def main(urls):
    """Runs the scraper on a list of URLs and sorts the results by increasing screen size."""
    results = asyncio.run(scrape_multiple_iphones(urls))
    df = pd.DataFrame(results)
    df["Screen Size"] = pd.to_numeric(df["Screen Size"].str.replace('"', ''), errors='coerce')
    df = df.sort_values(by="Screen Size", ascending=True)
    print(df)

# Example usage
urls = [
    "<https://www.apple.com/shop/product/FLML3LL/A/refurbished-iphone-13-128gb-midnight-unlocked>",
    "<https://www.apple.com/shop/product/FLMJ3LL/A/refurbished-iphone-13-256gb-starlight-unlocked>"
]
main(urls)

import asyncio
import aiohttp
import pandas as pd
from bs4 import BeautifulSoup
import re

async def fetch_page(session, url):
    """Fetches the HTML content of a given URL asynchronously."""
    async with session.get(url) as response:
        return await response.text()

async def scrape_iphone_details(url):
    """Scrapes details of an iPhone from the given Apple Refurbished Store product URL."""
    async with aiohttp.ClientSession() as session:
        html = await fetch_page(session, url)
        soup = BeautifulSoup(html, 'html.parser')

        try:
            phone_name = soup.find("h1", class_="product-title").text.strip() if soup.find("h1", class_="product-title") else "Unknown Model"
            price = soup.find("span", class_="as-price-currentprice").text.strip() if soup.find("span", class_="as-price-currentprice") else "Unknown Price"
            stock_status = "In Stock" if "Add to Bag" in html else "Out of Stock"

            # Find the section that contains technical details
            details_section = soup.find("div", class_="tech-specs")
            details_list = details_section.find_all("li") if details_section else []

            # Extract details dynamically
            details = {
                "Screen Size": "Unknown",
                "Dimensions": "Unknown",
                "Battery Capacity": "Unknown",
                "5G Availability": "Unknown",
                "USB-C Charging": "No",
                "Year of Release": "Unknown"
            }

            for item in details_list:
                text = item.text.strip().lower()

                if "display" in text or "screen" in text:
                    match = re.search(r'(\\d+(\\.\\d+)?)\\s*inch', text)
                    details["Screen Size"] = match.group(1) + '"' if match else "Unknown"

                if "dimensions" in text:
                    details["Dimensions"] = text.split(":")[-1].strip()

                if "battery" in text:
                    match = re.search(r'(\\d{3,4})\\s*mAh', text)
                    details["Battery Capacity"] = match.group(1) + " mAh" if match else "Unknown"

                if "5g" in text:
                    details["5G Availability"] = "Yes" if "5g" in text else "No"

                if "usb-c" in text:
                    details["USB-C Charging"] = "Yes"

                if "release year" in text:
                    match = re.search(r'(\\d{4})', text)
                    details["Year of Release"] = match.group(1) if match else "Unknown"

            return {
                "Input URL": url,
                "Phone Model Name": phone_name,
                "Operating System": "iOS (Check Apple site for current version)",
                "Screen Size": details["Screen Size"],
                "Dimensions": details["Dimensions"],
                "Battery Capacity": details["Battery Capacity"],
                "USB-C Charging": details["USB-C Charging"],
                "5G Availability": details["5G Availability"],
                "Price": price,
                "Stock Status": stock_status,
                "Year of Release": details["Year of Release"],
            }

        except Exception as e:
            return {"Error": f"Failed to scrape {url} - {str(e)}"}

async def scrape_multiple_iphones(urls):
    """Scrapes details of multiple iPhones in parallel using asyncio."""
    tasks = [scrape_iphone_details(url) for url in urls]
    return await asyncio.gather(*tasks)

def main(urls):
    """Runs the scraper on a list of URLs and sorts the results by increasing screen size."""
    results = asyncio.run(scrape_multiple_iphones(urls))
    df = pd.DataFrame(results)

    # Convert screen size to numeric for sorting
    df["Screen Size"] = pd.to_numeric(df["Screen Size"].str.replace('"', ''), errors='coerce')
    df = df.sort_values(by="Screen Size", ascending=True)

    print(df)

# Example usage
urls = [
    "<https://www.apple.com/shop/product/FLML3LL/A/refurbished-iphone-13-128gb-midnight-unlocked>",
    "<https://www.apple.com/shop/product/FLMJ3LL/A/refurbished-iphone-13-256gb-starlight-unlocked>"
]
main(urls)

...
html = await fetch_page(session, url)
soup = BeautifulSoup(html, 'html.parser')

try:
    # Extract JSON-LD script containing structured data
    json_ld_script = soup.find("script", type="application/ld+json")
    product_data = json.loads(json_ld_script.string) if json_ld_script else {}

    # Extract product name, price, and stock status from JSON-LD
    phone_name = product_data.get("name", "Unknown Model")
    price = product_data.get("offers", {}).get("price", "Unknown Price")
    stock_status = product_data.get("offers", {}).get("availability", "").split("/")[-1]
    stock_status = "In Stock" if stock_status.lower() == "instock" else "Out of Stock"
...

except Exception as e:
    print(f"❌ ERROR while scraping {url} ❌")
    traceback.print_exc()  # Print full error traceback for debugging
    return None  # Return None so failed entries are ignored

❌ ERROR while scraping <https://www.apple.com/shop/product/FLML3LL/A/refurbished-iphone-13-128gb-midnight-unlocked> ❌
Traceback (most recent call last):
  File "/home/kd/scrapingbee/chatgpt-apple-scraper/get_data.py", line 40, in scrape_iphone_details
    price = product_data.get("offers", {}).get("price", "Unknown Price")
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
AttributeError: 'list' object has no attribute 'get'
Traceback (most recent call last):
  File "/home/kd/scrapingbee/chatgpt-apple-scraper/get_data.py", line 124, in <module>
    main(urls)
  File "/home/kd/scrapingbee/chatgpt-apple-scraper/get_data.py", line 115, in main
    df["Screen Size"] = pd.to_numeric(df["Screen Size"].str.replace('"', ''), errors='coerce')
                                      ~~^^^^^^^^^^^^^^^
  File "/home/kd/.cache/pypoetry/virtualenvs/chatgpt-apple-scraper-C8P38fdX-py3.11/lib/python3.11/site-packages/pandas/core/frame.py", line 4102, in __getitem__
    indexer = self.columns.get_loc(key)
              ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/kd/.cache/pypoetry/virtualenvs/chatgpt-apple-scraper-C8P38fdX-py3.11/lib/python3.11/site-packages/pandas/core/indexes/range.py", line 417, in get_loc
    raise KeyError(key)
KeyError: 'Screen Size'

import asyncio
import aiohttp
import pandas as pd
from bs4 import BeautifulSoup
import re
import json
import traceback

async def fetch_page(session, url):
    """Fetches the HTML content of a given URL asynchronously."""
    async with session.get(url) as response:
        return await response.text()

async def scrape_iphone_details(url):
    """Scrapes details of an iPhone from the given Apple Refurbished Store product URL."""
    async with aiohttp.ClientSession() as session:
        html = await fetch_page(session, url)
        soup = BeautifulSoup(html, 'html.parser')

        try:
            # Extract JSON-LD script containing "Product" data
            json_ld_scripts = soup.find_all("script", type="application/ld+json")
            product_data = {}

            for script in json_ld_scripts:
                try:
                    data = json.loads(script.string)
                    if isinstance(data, list):  # JSON-LD sometimes is an array
                        for item in data:
                            if item.get("@type") == "Product":
                                product_data = item
                                break
                    elif data.get("@type") == "Product":
                        product_data = data
                except json.JSONDecodeError:
                    continue  # Skip invalid JSON sections

            # Extract product details
            phone_name = product_data.get("name", "Unknown Model")

            # Fix for the 'list' object issue
            offers = product_data.get("offers", [])
            if isinstance(offers, list) and len(offers) > 0:
                price = offers[0].get("price", "Unknown Price")
                stock_status = offers[0].get("availability", "").split("/")[-1]
            else:
                price = "Unknown Price"
                stock_status = "Unknown"

            stock_status = "In Stock" if stock_status.lower() == "instock" else "Out of Stock"

            # Extract technical specifications from "TechSpecs-panel"
            details_section = soup.find("div", class_="TechSpecs-panel")
            details_divs = details_section.find_all("div") if details_section else []

            details = {
                "Screen Size": "Unknown",
                "Dimensions": "Unknown",
                "Battery Capacity": "Unknown",
                "5G Availability": "Unknown",
                "USB-C Charging": "No",
                "Year of Release": "Unknown"
            }

            for div in details_divs:
                text = div.get_text(strip=True).lower()

                if "display" in text or "screen" in text:
                    match = re.search(r'(\\d+(\\.\\d+)?)\\s*inch', text)
                    details["Screen Size"] = match.group(1) + '"' if match else "Unknown"

                if "dimensions" in text:
                    details["Dimensions"] = text.split(":")[-1].strip()

                if "battery" in text:
                    match = re.search(r'(\\d{3,4})\\s*mAh', text)
                    details["Battery Capacity"] = match.group(1) + " mAh" if match else "Unknown"

                if "5g" in text:
                    details["5G Availability"] = "Yes" if "5g" in text else "No"

                if "usb-c" in text:
                    details["USB-C Charging"] = "Yes"

                if "release year" in text:
                    match = re.search(r'(\\d{4})', text)
                    details["Year of Release"] = match.group(1) if match else "Unknown"

            return {
                "Input URL": url,
                "Phone Model Name": phone_name,
                "Operating System": "iOS (Check Apple site for current version)",
                "Screen Size": details["Screen Size"],
                "Dimensions": details["Dimensions"],
                "Battery Capacity": details["Battery Capacity"],
                "USB-C Charging": details["USB-C Charging"],
                "5G Availability": details["5G Availability"],
                "Price": f"${price}" if price != "Unknown Price" else price,
                "Stock Status": stock_status,
                "Year of Release": details["Year of Release"],
            }

        except Exception as e:
            print(f"❌ ERROR while scraping {url} ❌")
            traceback.print_exc()
            return None  # Ignore failed scrapes

async def scrape_multiple_iphones(urls):
    """Scrapes details of multiple iPhones in parallel using asyncio."""
    tasks = [scrape_iphone_details(url) for url in urls]
    return await asyncio.gather(*tasks)

def main(urls):
    """Runs the scraper on a list of URLs and sorts the results by increasing screen size."""
    results = asyncio.run(scrape_multiple_iphones(urls))

    # Remove None values (failed scrapes)
    results = [r for r in results if r is not None]

    df = pd.DataFrame(results)

    # Fix for missing 'Screen Size' key
    if "Screen Size" in df.columns:
        df["Screen Size"] = pd.to_numeric(df["Screen Size"].str.replace('"', ''), errors='coerce')
        df = df.sort_values(by="Screen Size", ascending=True)

    print(df)

# Example usage
urls = [
    "<https://www.apple.com/shop/product/FLML3LL/A/refurbished-iphone-13-128gb-midnight-unlocked>",
    "<https://www.apple.com/shop/product/FLMJ3LL/A/refurbished-iphone-13-256gb-starlight-unlocked>"
]
main(urls)

# scraper.py
import asyncio
import aiohttp
import pandas as pd
from bs4 import BeautifulSoup
import json
import traceback

async def fetch_page(session, url):
    """Fetches the HTML content of a given URL asynchronously."""
    async with session.get(url) as response:
        return await response.text()

async def scrape_iphone_details(url):
    """Scrapes details of an iPhone from the given Apple Refurbished Store product URL."""
    async with aiohttp.ClientSession() as session:
        html = await fetch_page(session, url)
        soup = BeautifulSoup(html, 'html.parser')

        try:
            # Extract JSON-LD script containing "Product" data
            json_ld_scripts = soup.find_all("script", type="application/ld+json")
            product_data = {}

            for script in json_ld_scripts:
                try:
                    data = json.loads(script.string)
                    if isinstance(data, list):  # JSON-LD sometimes is an array
                        for item in data:
                            if item.get("@type") == "Product":
                                product_data = item
                                break
                    elif data.get("@type") == "Product":
                        product_data = data
                except json.JSONDecodeError:
                    continue  # Skip invalid JSON sections

            # Extract product details
            phone_name = product_data.get("name", "Unknown Model")

            # Extract only the model name (remove color & storage info)
            model_base_name = phone_name.split("-")[0].strip()

            # Fix for the 'list' object issue
            offers = product_data.get("offers", [])
            if isinstance(offers, list) and len(offers) > 0:
                price = offers[0].get("price", "Unknown Price")
                stock_status = offers[0].get("availability", "").split("/")[-1]
            else:
                price = "Unknown Price"
                stock_status = "Unknown"

            stock_status = "In Stock" if stock_status.lower() == "instock" else "Out of Stock"

            return {
                "Model Name": model_base_name,
                "Full Name": phone_name,
                "Input URL": url,
                "Price": price,
                "Stock Status": stock_status,
            }

        except Exception as e:
            print(f"❌ ERROR while scraping {url} ❌")
            traceback.print_exc()
            return None  # Ignore failed scrapes

async def scrape_multiple_iphones(urls):
    """Scrapes details of multiple iPhones in parallel using asyncio."""
    tasks = [scrape_iphone_details(url) for url in urls]
    return await asyncio.gather(*tasks)

def main(urls):
    """Runs the scraper on a list of URLs and filters the lowest price per model."""
    results = asyncio.run(scrape_multiple_iphones(urls))
    
    # Remove None values (failed scrapes)
    results = [r for r in results if r is not None]

    df = pd.DataFrame(results)

    # Keep only the lowest price for each model
    df = df.sort_values(by="Price").drop_duplicates(subset=["Model Name"], keep="first")

    return df

# price_monitor.py

import time
import datetime
import os
import pandas as pd
import scraper  # Import the scraper script
from get_refurbished_iphone_urls import get_refurbished_iphone_urls  # Fetches URLs dynamically

PRICE_FILE = "price_data.csv"

def load_previous_data():
    """Loads the last recorded price data from disk (if available)."""
    if os.path.exists(PRICE_FILE):
        try:
            return pd.read_csv(PRICE_FILE)
        except Exception as e:
            print(f"⚠️ Error loading previous data: {e}. Starting fresh.")
            return None
    return None

def save_new_data(df):
    """Saves the latest price data to disk."""
    df.to_csv(PRICE_FILE, index=False)
    print(f"✅ Saved latest price data to {PRICE_FILE}")

def compare_prices(old_df, new_df):
    """Compare old and new price data and return changes as a markdown string with emojis."""
    changes = []
    
    for _, new_row in new_df.iterrows():
        model_name = new_row["Model Name"]
        new_price = new_row["Price"]
        new_stock = new_row["Stock Status"]
        
        old_row = old_df[old_df["Model Name"] == model_name]
        if not old_row.empty:
            old_price = old_row.iloc[0]["Price"]
            old_stock = old_row.iloc[0]["Stock Status"]

            if new_price != old_price or new_stock != old_stock:
                price_change_emoji = "🔻" if new_price < old_price else "🔺"
                stock_change_emoji = "🟢" if new_stock == "In Stock" else "🔴"
                changes.append(f"| {model_name} | {price_change_emoji} **${old_price} → ${new_price}** | {stock_change_emoji} {old_stock} → **{new_stock}** |")
        else:
            changes.append(f"| {model_name} | 🎉 **New Listing: ${new_price}** | 🟢 **{new_stock}** |")

    if not changes:
        return None

    markdown_content = f"# 📉 Price Changes - {datetime.date.today()}\n\n"
    markdown_content += "| Model Name | Price Change | Stock Status |\n"
    markdown_content += "|------------|-------------|--------------|\n"
    markdown_content += "\n".join(changes)
    
    return markdown_content

def save_markdown(content):
    """Save the markdown content to a file with today's date."""
    filename = f"price_changes_{datetime.date.today()}.md"
    with open(filename, "w") as file:
        file.write(content)
    print(f"✅ Saved changes to {filename}")

def price_monitor():
    """Runs the price monitor indefinitely, checking for changes daily."""
    previous_data = load_previous_data()

    while True:
        print("🔄 Running daily price check...")

        # Get the latest refurbished iPhone URLs dynamically
        urls = get_refurbished_iphone_urls()
        new_data = scraper.main(urls)

        if previous_data is not None:
            changes = compare_prices(previous_data, new_data)
            if changes:
                save_markdown(changes)
            else:
                print("✅ No changes detected.")

        # Save the new data for future comparison
        save_new_data(new_data)

        # Update previous_data for the next iteration
        previous_data = new_data.copy()

        print("⏳ Sleeping for 24 hours...")
        time.sleep(86400)  # Sleep for one day

if __name__ == "__main__":
    price_monitor()