PYTHON scrape.py

#!/usr/bin/env python3
"""
Google Maps Lead Generator — find businesses without websites.

Usage:
    python3 scrape.py --query "plumbers" --location "Austin TX"
    python3 scrape.py --query "restaurants" --location "Round Rock TX" --limit 30
    python3 scrape.py -o leads.csv --query "auto repair" --location "Cedar Park TX"

Output: CSV/JSON with name, phone, rating, address, Google Maps URL
        (only businesses that have NO website listed)
"""

import argparse
import asyncio
import csv
import json
import re
import sys
from pathlib import Path
from urllib.parse import quote

from playwright.async_api import async_playwright

SCRIPT_DIR = Path(__file__).parent

async def scrape(query: str, location: str, max_results: int = 20, headless: bool = True) -> list[dict]:
    """Scrape Google Maps search results and return businesses without websites."""

    # Load JS extractor
    with open(SCRIPT_DIR / "extract.js", encoding="utf-8") as f:
        extract_js = f.read()

    pw = await async_playwright().start()
    browser = await pw.firefox.launch(headless=headless)
    context = await browser.new_context(
        viewport={"width": 1920, "height": 1080},
        locale="en-US",
        timezone_id="America/Chicago",
    )

    # Bypass consent dialog
    await context.add_cookies([
        {
            "name": "CONSENT",
            "value": "YES+cb.20240101-01-p0.en+FX+430",
            "domain": ".google.com",
            "path": "/",
        }
    ])

    # Stealth
    await context.add_init_script("""
        Object.defineProperty(navigator, 'webdriver', {get: () => undefined});
        const orig = window.navigator.permissions.query;
        window.navigator.permissions.query = (params) =>
            params.name === 'notifications'
                ? Promise.resolve({state: Notification.permission})
                : orig(params);
    """)

    page = await context.new_page()

    # Block heavy resources
    async def block_resources(route):
        if route.request.resource_type in {"font", "media"}:
            await route.abort()
        else:
            await route.continue_()
    await page.route("**/*", block_resources)

    url = f"https://www.google.com/maps/search/{quote(f'{query}+{location}')}"
    print(f"[*] Searching: {query} in {location}")
    print(f"[*] URL: {url}")

    await page.goto(url, wait_until="domcontentloaded", timeout=30000)
    print("[*] Waiting for results to load...")
    await page.wait_for_timeout(3000)

    # Extract all business listings
    all_results = await page.evaluate(extract_js)

    # Filter: keep only businesses WITHOUT real websites
    leads = [r for r in all_results if not r.get("website")]

    print(f"\n[+] Total businesses: {len(all_results)}")
    print(f"[+] Without websites: {len(leads)}")

    await browser.close()
    await pw.stop()

    return leads[:max_results]


def save_csv(leads: list[dict], path: str):
    """Save leads to CSV."""
    fields = ["name", "phone", "rating", "reviews", "address", "maps_url"]
    with open(path, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=fields, extrasaction="ignore")
        writer.writeheader()
        writer.writerows(leads)
    print(f"[+] Saved {len(leads)} leads to {path}")


def save_json(leads: list[dict], path: str):
    """Save leads to JSON."""
    with open(path, "w", encoding="utf-8") as f:
        json.dump(leads, f, indent=2, ensure_ascii=False)
    print(f"[+] Saved {len(leads)} leads to {path}")


def main():
    parser = argparse.ArgumentParser(description="Find businesses without websites on Google Maps")
    parser.add_argument("--query", "-q", required=True, help='Business type: "plumbers"')
    parser.add_argument("--location", "-l", required=True, help='Location: "Austin TX"')
    parser.add_argument("--limit", "-n", type=int, default=20, help="Max results (default: 20)")
    parser.add_argument("--output", "-o", default=None, help="Output file path")
    parser.add_argument("--json", dest="as_json", action="store_true", help="Output as JSON")
    parser.add_argument("--visible", action="store_true", help="Show browser window")
    args = parser.parse_args()

    leads = asyncio.run(scrape(
        query=args.query,
        location=args.location,
        max_results=args.limit,
        headless=not args.visible,
    ))

    if not leads:
        print("\n[!] No leads found (all results have websites or Google blocked us)")
        sys.exit(0)

    # Print results
    print(f"\n{'='*60}")
    print(f"  LEADS: {args.query} near {args.location}")
    print(f"  {len(leads)} businesses without websites")
    print(f"{'='*60}\n")

    for i, lead in enumerate(leads, 1):
        rating = f"⭐ {lead['rating']}" if lead['rating'] else "no rating"
        reviews = f"({lead['reviews']})" if lead['reviews'] else ""
        print(f"  {i}. {lead['name']}")
        print(f"     {rating} {reviews}")
        print(f"     📍 {lead['address'] or 'N/A'}")
        print(f"     📞 {lead['phone'] or 'N/A'}")
        print(f"     🔗 {lead['maps_url']}")
        print()

    # Save
    if not args.output:
        safe_query = args.query.replace(" ", "_")
        safe_loc = args.location.replace(" ", "_").lower()
        ext = "json" if args.as_json else "csv"
        args.output = f"leads_{safe_query}_{safe_loc}.{ext}"

    if args.as_json:
        save_json(leads, args.output)
    else:
        save_csv(leads, args.output)


if __name__ == "__main__":
    main()