Spaces:
Running on Zero
Running on Zero
| import gradio as gr | |
| import pandas as pd | |
| import asyncio | |
| import re | |
| from playwright.async_api import async_playwright | |
| import requests | |
| from bs4 import BeautifulSoup | |
| import tempfile | |
| from spaces import GPU | |
| import os, subprocess, sys | |
| # ---------- Dummy GPU function (ZeroGPU requirement) ---------- | |
| def dummy_gpu(): | |
| return "GPU OK" | |
| # ----------------------------------------------------------------- | |
| # ---------- Ensure Playwright Chromium is installed ---------- | |
| chromium_executable = "/home/user/.cache/ms-playwright/chromium_headless_shell-1228/chrome-headless-shell-linux64/chrome-headless-shell" | |
| if not os.path.exists(chromium_executable): | |
| print("Downloading Playwright Chromium...") | |
| subprocess.run([sys.executable, "-m", "playwright", "install", "chromium"], check=True) | |
| print("Chromium download complete.") | |
| # ----------------------------------------------------------- | |
| HEADERS = { | |
| "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" | |
| } | |
| # ------------------ Scraping functions ------------------ | |
| async def search_google_maps(page, query, max_results): | |
| try: | |
| # Load the page and wait until the main content appears | |
| await page.goto("https://www.google.com/maps", wait_until="load", timeout=40000) | |
| # Wait a moment for dynamic content | |
| await page.wait_for_timeout(2000) | |
| # Accept cookies if the popup appears | |
| try: | |
| await page.click('button[aria-label="Accept all"]', timeout=3000) | |
| except: | |
| pass | |
| # Wait for the search box to be present (longer timeout) | |
| await page.wait_for_selector('input#searchboxinput', state="attached", timeout=30000) | |
| search_box = page.locator('input#searchboxinput') | |
| await search_box.fill(query) | |
| await search_box.press("Enter") | |
| # Wait for the results feed to appear | |
| await page.wait_for_selector('div[role="feed"]', timeout=20000) | |
| feed = page.locator('div[role="feed"]') | |
| prev = 0 | |
| while True: | |
| items = await feed.locator('div[role="article"]').all() | |
| if len(items) >= max_results or len(items) == prev: | |
| break | |
| if items: | |
| await items[-1].scroll_into_view_if_needed() | |
| await page.wait_for_timeout(2000) | |
| prev = len(items) | |
| return (await feed.locator('div[role="article"]').all())[:max_results] | |
| except Exception as e: | |
| print(f"Error during search: {e}") | |
| return [] | |
| # (extract_details, scrape_website remain unchanged) | |
| async def extract_details(page, article): | |
| await article.click() | |
| await page.wait_for_selector('div[aria-label*="Information for"]', timeout=10000) | |
| await page.wait_for_timeout(2000) | |
| details = {} | |
| try: | |
| el = page.locator('h1.fontHeadlineLarge') | |
| details['name'] = await el.inner_text() if await el.count() > 0 else '' | |
| except: | |
| details['name'] = '' | |
| try: | |
| el = page.locator('button[data-item-id="address"]') | |
| details['address'] = await el.inner_text() if await el.count() > 0 else '' | |
| except: | |
| details['address'] = '' | |
| try: | |
| el = page.locator('button[data-item-id*="phone"]') | |
| details['phone'] = await el.inner_text() if await el.count() > 0 else '' | |
| except: | |
| details['phone'] = '' | |
| try: | |
| el = page.locator('a[data-item-id*="authority"]') | |
| details['website'] = await el.get_attribute('href') if await el.count() > 0 else '' | |
| except: | |
| details['website'] = '' | |
| try: | |
| el = page.locator('div.fontBodyMedium > span[aria-hidden="true"]') | |
| details['rating'] = await el.inner_text() if await el.count() > 0 else '' | |
| except: | |
| details['rating'] = '' | |
| try: | |
| el = page.locator('button[aria-label*="reviews"]') | |
| txt = await el.inner_text() if await el.count() > 0 else '' | |
| details['reviews'] = txt.split('(')[-1].split(')')[0] if '(' in txt else '' | |
| except: | |
| details['reviews'] = '' | |
| try: | |
| el = page.locator('div[aria-label*="Opening hours"]') | |
| details['hours'] = await el.inner_text() if await el.count() > 0 else '' | |
| except: | |
| details['hours'] = '' | |
| try: | |
| el = page.locator('button[jsaction*="pane.rating.category"]') | |
| details['category'] = await el.inner_text() if await el.count() > 0 else '' | |
| except: | |
| details['category'] = '' | |
| details['maps_url'] = page.url | |
| try: | |
| back_btn = page.locator('button[aria-label="Back"]') | |
| if await back_btn.count() > 0: | |
| await back_btn.click() | |
| await page.wait_for_timeout(1000) | |
| except: | |
| pass | |
| return details | |
| def scrape_website(url): | |
| res = {"emails": [], "phones": [], "socials": []} | |
| if not url: | |
| return res | |
| try: | |
| r = requests.get(url, headers=HEADERS, timeout=10) | |
| if r.status_code != 200: | |
| return res | |
| soup = BeautifulSoup(r.text, "html.parser") | |
| text = soup.get_text() | |
| for a in soup.find_all("a", href=True): | |
| h = a["href"] | |
| if h.startswith("mailto:"): | |
| e = h.replace("mailto:", "").split("?")[0].strip() | |
| if e and e not in res["emails"]: | |
| res["emails"].append(e) | |
| emails = re.findall(r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}", text) | |
| for e in emails: | |
| if e not in res["emails"]: | |
| res["emails"].append(e) | |
| phones = re.findall(r"(\+?\d{1,3}[-.\s]?)?\(?\d{1,4}\)?[-.\s]?\d{1,4}[-.\s]?\d{1,9}", text) | |
| res["phones"] = list(set(phones))[:5] | |
| socials = [] | |
| for a in soup.find_all("a", href=True): | |
| h = a["href"].lower() | |
| if any(s in h for s in ["facebook.com", "twitter.com", "linkedin.com", "instagram.com"]): | |
| socials.append(a["href"]) | |
| res["socials"] = list(set(socials))[:5] | |
| except: | |
| pass | |
| return res | |
| def run_scrape(query, country, city, max_results): | |
| full_query = query | |
| if city: | |
| full_query += f" in {city}" | |
| if country: | |
| full_query += f", {country}" | |
| async def _async(): | |
| async with async_playwright() as p: | |
| browser = await p.chromium.launch( | |
| headless=True, | |
| args=[ | |
| '--disable-blink-features=AutomationControlled', | |
| '--no-sandbox', | |
| '--disable-dev-shm-usage', | |
| ] | |
| ) | |
| context = await browser.new_context( | |
| user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", | |
| viewport={"width": 1280, "height": 800} | |
| ) | |
| page = await context.new_page() | |
| articles = await search_google_maps(page, full_query, max_results) | |
| results = [] | |
| for art in articles: | |
| try: | |
| d = await extract_details(page, art) | |
| contacts = scrape_website(d.get('website', '')) | |
| results.append({ | |
| "Name": d.get('name', ''), | |
| "Address": d.get('address', ''), | |
| "Phone (Maps)": d.get('phone', ''), | |
| "Website": d.get('website', ''), | |
| "Rating": d.get('rating', ''), | |
| "Reviews": d.get('reviews', ''), | |
| "Opening Hours": d.get('hours', ''), | |
| "Category": d.get('category', ''), | |
| "Maps URL": d.get('maps_url', ''), | |
| "Emails (site)": ", ".join(contacts['emails']), | |
| "Extra Phones (site)": ", ".join(contacts['phones']), | |
| "Socials": ", ".join(contacts['socials']) | |
| }) | |
| except Exception: | |
| pass | |
| await page.wait_for_timeout(1000) | |
| await context.close() | |
| await browser.close() | |
| return results | |
| data = asyncio.run(_async()) | |
| if not data: | |
| return pd.DataFrame(), None, "No results found. Try a different query or fewer results." | |
| df = pd.DataFrame(data) | |
| tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".xlsx") | |
| with pd.ExcelWriter(tmp.name, engine="openpyxl") as writer: | |
| df.to_excel(writer, index=False, sheet_name="Results") | |
| excel_path = tmp.name | |
| tsv = df.to_csv(sep='\t', index=False) | |
| return df, excel_path, tsv | |
| # ------------------ Gradio UI ------------------ | |
| with gr.Blocks(title="Google Maps Scraper") as demo: | |
| gr.Markdown("# 🗺️ Google Maps Business Scraper + Contact Finder") | |
| gr.Markdown("Enter a search query. Use small numbers (5‑10) to avoid timeouts.") | |
| with gr.Row(): | |
| query = gr.Textbox(label="Search Query", value="parking businesses") | |
| country = gr.Textbox(label="Country (optional)", value="UK") | |
| city = gr.Textbox(label="City (optional)", value="London") | |
| max_results = gr.Slider(minimum=5, maximum=30, value=10, step=1, label="Max Results") | |
| scrape_btn = gr.Button("Start Scraping", variant="primary") | |
| dataframe = gr.Dataframe(label="Results", interactive=False) | |
| download_btn = gr.DownloadButton(label="📥 Download Excel", visible=False) | |
| tsv_output = gr.Textbox(label="📋 TSV (Copy manually)", visible=False, lines=10) | |
| def handle_scrape(query, country, city, max_results): | |
| df, excel_path, tsv = run_scrape(query, country, city, max_results) | |
| show = bool(excel_path) | |
| return ( | |
| df, | |
| gr.update(value=excel_path, visible=show), | |
| gr.update(value=tsv, visible=show), | |
| ) | |
| scrape_btn.click( | |
| fn=handle_scrape, | |
| inputs=[query, country, city, max_results], | |
| outputs=[dataframe, download_btn, tsv_output] | |
| ) | |
| if __name__ == "__main__": | |
| demo.launch(ssr_mode=False) |