GoogleMapDataExtractor.py

# MalkiRestaurantExtractor.py
# pip install selenium pandas webdriver-manager
import time
import csv
import os
from typing import List
import urllib.parse
import pandas as pd
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.common.by import By
from webdriver_manager.chrome import ChromeDriverManager

# ---- Config ----
HEADLESS = False          # set True to run silently, False to watch the browser
SCROLL_ROUNDS = 6
PLACE_VISIT_DELAY = 3

#  search bussinesses by entering businss type + city name, for example "small restaurants in Raleigh, North Carolina"
CITY_LIST = [
    "small restaurants in Raleigh, North Carolina",
#    you can add more cities here, for example:
#    "small restaurants in Charlotte, North Carolina",
#    also you change business type, for example "cafes", "bakeries", "italian restaurants", etc.
]
OUTPUT_CSV = "US_Small_Restaurants.csv"

# ---- Helper functions ----
def start_driver(headless=True):
    options = Options()
    if headless:
        options.add_argument("--headless=new")
    options.add_argument("--window-size=1920,1080")
    options.add_argument("--disable-gpu")
    options.add_argument("--no-sandbox")
    options.add_argument("--disable-dev-shm-usage")
    options.add_argument("user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                         "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/142.0.7444.60 Safari/537.36")

    service = Service(ChromeDriverManager().install())
    driver = webdriver.Chrome(service=service, options=options)
    driver.implicitly_wait(5)
    return driver


def collect_listings_from_search(driver, query):
    encoded = urllib.parse.quote_plus(query)
    url = f"https://www.google.com/maps/search/{encoded}"
    driver.get(url)
    time.sleep(5)
    listings = []

    try:
        scrollable = driver.find_element(By.CSS_SELECTOR, 'div[role="feed"]')
        for _ in range(SCROLL_ROUNDS):
            driver.execute_script("arguments[0].scrollTop = arguments[0].scrollHeight", scrollable)
            time.sleep(2)
    except Exception:
        pass

    anchors = driver.find_elements(By.CSS_SELECTOR, 'a[href*="/place/"], a.hfpxzc')
    seen = set()

    for a in anchors:
        href = a.get_attribute("href")
        name = a.get_attribute("aria-label") or a.get_attribute("title") or ""
        if not href or href in seen or not name.strip():
            continue
        seen.add(href)

        address = ""
        try:
            parent = a.find_element(By.XPATH, "..")
            address_el = parent.find_element(By.CSS_SELECTOR, "div.W4Efsd span:nth-of-type(2)")
            address = address_el.text
        except Exception:
            pass

        listings.append({"name": name.strip(), "maps_url": href, "address": address})
    return listings


def visit_place_and_extract(driver, maps_url):
    data = {"website": ""}
    try:
        driver.get(maps_url)
        time.sleep(PLACE_VISIT_DELAY)

        website_el = None
        candidates = driver.find_elements(
            By.XPATH,
            "//a[contains(@href,'http') and (contains(., 'Website') or contains(@aria-label,'Website'))]"
        )
        if not candidates:
            candidates = driver.find_elements(By.CSS_SELECTOR, "a[data-tooltip*='Website'], a[aria-label*='Website']")

        for el in candidates:
            href = el.get_attribute("href")
            if href and href.startswith("http"):
                website_el = href
                break
        if website_el:
            data["website"] = website_el
    except Exception:
        pass

    return data


def setup_csv_writer(filename):
    """Create a CSV writer and write header if file is new/empty."""
    file_exists = os.path.isfile(filename) and os.path.getsize(filename) > 0
    f = open(filename, 'a', newline='', encoding='utf-8')
    writer = csv.DictWriter(f, fieldnames=["Restaurant Name", "Website", "Address"])
    if not file_exists:
        writer.writeheader()
    return f, writer


# ---- Main ----
def main():
    driver = start_driver(headless=HEADLESS)
    seen_names = set()               # to avoid duplicates in real time
    csv_file, csv_writer = setup_csv_writer(OUTPUT_CSV)
    total_processed = 0

    try:
        for city in CITY_LIST:
            print(f"\n🔍 Searching: {city}")
            listings = collect_listings_from_search(driver, city)
            print(f"📋 Found {len(listings)} results")

            for i, item in enumerate(listings):
                # Skip if we already have this restaurant name
                if item["name"] in seen_names:
                    print(f"⏭️ [{i+1}/{len(listings)}] Skipping duplicate: {item['name']}")
                    continue

                print(f"📌 [{i+1}/{len(listings)}] Extracting: {item['name']}")
                details = visit_place_and_extract(driver, item["maps_url"])

                row = {
                    "Restaurant Name": item["name"],
                    "Website": details.get("website", ""),
                    "Address": item["address"]
                }

                # Write immediately to CSV and flush to disk
                csv_writer.writerow(row)
                csv_file.flush()
                seen_names.add(item["name"])
                total_processed += 1

                # Also show live output on console
                print(f"   ✅ Saved -> {row['Website'] or 'no website'}")

                time.sleep(2)

            time.sleep(5)

    except KeyboardInterrupt:
        print("\n⚠️ Script interrupted by user. Partial data saved.")
    except Exception as e:
        print(f"\n❌ Unexpected error: {e}")
    finally:
        csv_file.close()
        driver.quit()
        print(f"\n🎉 Done! Saved {total_processed} restaurants (unique) to {OUTPUT_CSV}")

    # Optional: load with pandas for final report
    if os.path.exists(OUTPUT_CSV):
        df = pd.read_csv(OUTPUT_CSV)
        print("\n📊 Preview of saved data:")
        print(df.head(10).to_string(index=False))


if __name__ == "__main__":
    main()