app.py

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) ----------
@GPU
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)