deploy Maps-backed ProspectOS workspace

api/server.py

@@ -1,9 +1,10 @@
 #!/usr/bin/env python3
-"""Small browser-only prospect discovery service.
+"""Personal ProspectOS service backed by Google Maps Scraper.
 
-The service deliberately has one live source (Google in a headless browser),
-keeps a local catalog for deduplication, and exposes only review actions.
-There is no email scraping, outreach, CRM automation, or source registry.
+The app owns the small local catalogue and review workflow. Google Maps
+Scraper runs as a separate Playwright service and is called through its
+asynchronous REST API. A browser-search implementation remains available as
+an explicit fallback (DISCOVERY_BACKEND=browser) for local testing.
 """
 from __future__ import annotations
 
@@ -14,14 +15,20 @@ import re
 import sqlite3
 import threading
 import time
+import urllib.error
+import urllib.request
 from datetime import datetime, timezone
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
-from typing import Any
+from typing import Any, Callable
 from urllib.parse import parse_qs, quote_plus, unquote, urlparse
 
 DB_PATH = os.environ.get("PROSPECT_DB", "/data/prospects.db")
-MAX_RESULTS = 25
+MAX_RESULTS = 50
+GMAPS_POLL_SECONDS = float(os.environ.get("GMAPS_POLL_SECONDS", "2"))
+GMAPS_REQUEST_TIMEOUT = int(os.environ.get("GMAPS_REQUEST_TIMEOUT", "300"))
+GMAPS_URL = os.environ.get("GMAPS_URL", "").strip().rstrip("/")
+GMAPS_API_KEY = os.environ.get("GMAPS_API_KEY", "").strip()
 ALLOWED_REVIEW = {"pending", "approved", "rejected", "contacted"}
 _discover_lock = threading.Lock()
 
@@ -30,21 +37,43 @@ def now() -> str:
     return datetime.now(timezone.utc).replace(microsecond=0).isoformat()
 
 
+def backend_mode() -> str:
+    configured = os.environ.get("DISCOVERY_BACKEND", "").strip().lower()
+    if configured in {"gmaps", "browser"}:
+        return configured
+    return "gmaps" if os.environ.get("GMAPS_URL", "").strip() else "browser"
+
+
 def db() -> sqlite3.Connection:
     os.makedirs(os.path.dirname(DB_PATH) or ".", exist_ok=True)
-    conn = sqlite3.connect(DB_PATH)
+    conn = sqlite3.connect(DB_PATH, timeout=30)
     conn.row_factory = sqlite3.Row
+    conn.execute("PRAGMA busy_timeout=30000")
     conn.execute("""CREATE TABLE IF NOT EXISTS prospects (
         id INTEGER PRIMARY KEY AUTOINCREMENT,
         name TEXT NOT NULL,
-        website TEXT NOT NULL,
-        domain TEXT NOT NULL,
+        website TEXT NOT NULL DEFAULT '',
+        domain TEXT NOT NULL UNIQUE,
         description TEXT NOT NULL DEFAULT '',
-        query TEXT NOT NULL,
+        query TEXT NOT NULL DEFAULT '',
         discovered_at TEXT NOT NULL,
         review_status TEXT NOT NULL DEFAULT 'pending',
         review_notes TEXT NOT NULL DEFAULT '',
-        UNIQUE(domain)
+        source TEXT NOT NULL DEFAULT 'google_maps',
+        source_id TEXT NOT NULL DEFAULT '',
+        category TEXT NOT NULL DEFAULT '',
+        address TEXT NOT NULL DEFAULT '',
+        phone TEXT NOT NULL DEFAULT '',
+        emails TEXT NOT NULL DEFAULT '',
+        rating REAL,
+        review_count INTEGER NOT NULL DEFAULT 0,
+        maps_url TEXT NOT NULL DEFAULT '',
+        latitude REAL,
+        longitude REAL,
+        listing_status TEXT NOT NULL DEFAULT '',
+        website_status TEXT NOT NULL DEFAULT 'unknown',
+        opportunity_score INTEGER NOT NULL DEFAULT 0,
+        raw_json TEXT NOT NULL DEFAULT '{}'
     )""")
     conn.execute("""CREATE TABLE IF NOT EXISTS runs (
         id INTEGER PRIMARY KEY AUTOINCREMENT,
@@ -55,8 +84,60 @@ def db() -> sqlite3.Connection:
         added INTEGER NOT NULL DEFAULT 0,
         status TEXT NOT NULL,
         error TEXT NOT NULL DEFAULT '',
-        created_at TEXT NOT NULL
+        created_at TEXT NOT NULL,
+        source TEXT NOT NULL DEFAULT 'google_maps',
+        external_job_id TEXT NOT NULL DEFAULT '',
+        stage TEXT NOT NULL DEFAULT 'queued',
+        progress INTEGER NOT NULL DEFAULT 0,
+        completed_at TEXT NOT NULL DEFAULT '',
+        request_json TEXT NOT NULL DEFAULT '{}'
     )""")
+
+    # Migrate databases created by v1-v5 without destroying the user's
+    # catalogue. SQLite permits adding nullable/defaulted columns in place.
+    migrations = {
+        "source": "TEXT NOT NULL DEFAULT 'google_maps'",
+        "source_id": "TEXT NOT NULL DEFAULT ''",
+        "category": "TEXT NOT NULL DEFAULT ''",
+        "address": "TEXT NOT NULL DEFAULT ''",
+        "phone": "TEXT NOT NULL DEFAULT ''",
+        "emails": "TEXT NOT NULL DEFAULT ''",
+        "rating": "REAL",
+        "review_count": "INTEGER NOT NULL DEFAULT 0",
+        "maps_url": "TEXT NOT NULL DEFAULT ''",
+        "latitude": "REAL",
+        "longitude": "REAL",
+        "listing_status": "TEXT NOT NULL DEFAULT ''",
+        "website_status": "TEXT NOT NULL DEFAULT 'unknown'",
+        "opportunity_score": "INTEGER NOT NULL DEFAULT 0",
+        "raw_json": "TEXT NOT NULL DEFAULT '{}'",
+    }
+    existing = {row[1] for row in conn.execute("PRAGMA table_info(prospects)")}
+    for name, ddl in migrations.items():
+        if name not in existing:
+            conn.execute(f"ALTER TABLE prospects ADD COLUMN {name} {ddl}")
+    run_migrations = {
+        "source": "TEXT NOT NULL DEFAULT 'google_maps'",
+        "external_job_id": "TEXT NOT NULL DEFAULT ''",
+        "stage": "TEXT NOT NULL DEFAULT 'queued'",
+        "progress": "INTEGER NOT NULL DEFAULT 0",
+        "completed_at": "TEXT NOT NULL DEFAULT ''",
+        "request_json": "TEXT NOT NULL DEFAULT '{}'",
+    }
+    existing_runs = {row[1] for row in conn.execute("PRAGMA table_info(runs)")}
+    for name, ddl in run_migrations.items():
+        if name not in existing_runs:
+            conn.execute(f"ALTER TABLE runs ADD COLUMN {name} {ddl}")
+    conn.execute("CREATE INDEX IF NOT EXISTS idx_prospects_review ON prospects(review_status)")
+    conn.execute("CREATE INDEX IF NOT EXISTS idx_prospects_score ON prospects(opportunity_score DESC)")
+    conn.execute("CREATE INDEX IF NOT EXISTS idx_runs_created ON runs(id DESC)")
+    # A Maps scrape can legitimately take several minutes. Only recover jobs
+    # that have been abandoned for half an hour, never after 35 seconds.
+    conn.execute("""UPDATE runs SET status='failed', stage='failed',
+        error=CASE WHEN error='' THEN 'stale_run_recovered' ELSE error END,
+        completed_at=CASE WHEN completed_at='' THEN ? ELSE completed_at END
+        WHERE status IN ('queued','running')
+          AND julianday(created_at) < julianday('now','-30 minutes')""", (now(),))
     conn.commit()
     return conn
 
@@ -73,6 +154,208 @@ def clean_domain(url: str) -> str:
     return host
 
 
+def normalize_url(value: Any) -> str:
+    text = str(value or "").strip()
+    if not text:
+        return ""
+    if text.startswith("//"):
+        text = "https:" + text
+    if not text.startswith(("http://", "https://")):
+        return ""
+    return text
+
+
+def as_text(value: Any) -> str:
+    if value is None:
+        return ""
+    if isinstance(value, (list, tuple, set)):
+        return ", ".join(as_text(item) for item in value if as_text(item))
+    if isinstance(value, dict):
+        return ", ".join(as_text(item) for item in value.values() if as_text(item))
+    return " ".join(str(value).split()).strip()
+
+
+def as_float(value: Any) -> float | None:
+    try:
+        if value in (None, ""):
+            return None
+        return float(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def as_int(value: Any) -> int:
+    try:
+        if value in (None, ""):
+            return 0
+        return max(0, int(float(str(value).replace(",", ""))))
+    except (TypeError, ValueError):
+        return 0
+
+
+def score_place(item: dict[str, Any], website: str, phone: str, rating: float | None, review_count: int) -> int:
+    """Small transparent triage score; it is not an AI verdict."""
+    score = 20
+    if not website:
+        score += 35
+    if phone:
+        score += 10
+    if rating is not None and rating >= 4:
+        score += 10
+    if review_count >= 10:
+        score += 10
+    if as_text(item.get("category") or item.get("type")):
+        score += 5
+    return min(100, score)
+
+
+def map_gmaps_result(item: dict[str, Any], query: str) -> dict[str, Any]:
+    """Normalize gosom/google-maps-scraper's place object to our catalogue."""
+    name = as_text(item.get("title") or item.get("name") or item.get("business_name")) or "Unnamed place"
+    website = normalize_url(item.get("website") or item.get("website_url"))
+    maps_url = normalize_url(item.get("link") or item.get("maps_url") or item.get("google_maps_url"))
+    address = as_text(item.get("complete_address") or item.get("address"))
+    phone = as_text(item.get("phone") or item.get("phone_number"))
+    category = as_text(item.get("category") or item.get("type") or item.get("categories"))
+    description = as_text(item.get("descriptions") or item.get("description") or category)
+    emails = as_text(item.get("emails") or item.get("email"))
+    rating = as_float(item.get("review_rating") if item.get("review_rating") is not None else item.get("rating"))
+    review_count = as_int(item.get("review_count") if item.get("review_count") is not None else item.get("reviews"))
+    source_id = as_text(item.get("place_id") or item.get("data_id") or item.get("cid") or maps_url)
+    if not source_id:
+        source_id = hashlib.sha1((name + "|" + address).encode("utf-8")).hexdigest()[:20]
+    domain = clean_domain(website) if website else "maps:" + source_id
+    latitude = as_float(item.get("latitude"))
+    longitude = as_float(item.get("longitude"))
+    return {
+        "name": name[:200],
+        "website": website,
+        "domain": domain[:255],
+        "description": description[:1000],
+        "query": query,
+        "source": "google_maps",
+        "source_id": source_id[:255],
+        "category": category[:255],
+        "address": address[:1000],
+        "phone": phone[:120],
+        "emails": emails[:500],
+        "rating": rating,
+        "review_count": review_count,
+        "maps_url": maps_url,
+        "latitude": latitude,
+        "longitude": longitude,
+        "listing_status": as_text(item.get("status"))[:120],
+        "website_status": "present" if website else "missing",
+        "opportunity_score": score_place(item, website, phone, rating, review_count),
+        "raw_json": json.dumps(item, ensure_ascii=False)[:100_000],
+    }
+
+
+def _looks_like_place(value: dict[str, Any]) -> bool:
+    keys = {str(key).lower() for key in value}
+    return bool(keys.intersection({"title", "name", "business_name"})) and bool(
+        keys.intersection({"address", "complete_address", "website", "phone", "place_id", "link"})
+    )
+
+
+def extract_gmaps_results(raw: Any) -> list[dict[str, Any]]:
+    """Handle both current API payloads and older nested result formats."""
+    output: list[dict[str, Any]] = []
+    seen: set[str] = set()
+
+    def visit(value: Any) -> None:
+        if isinstance(value, list):
+            for item in value:
+                visit(item)
+        elif isinstance(value, dict):
+            if _looks_like_place(value):
+                marker = as_text(value.get("place_id") or value.get("data_id") or value.get("cid") or value.get("link") or value.get("title") or value.get("name"))
+                if marker and marker not in seen:
+                    seen.add(marker)
+                    output.append(value)
+                    return
+            for key in ("results", "data", "items", "places", "result"):
+                if key in value:
+                    visit(value[key])
+            if not any(key in value for key in ("results", "data", "items", "places", "result")):
+                for child in value.values():
+                    if isinstance(child, (list, dict)):
+                        visit(child)
+
+    visit(raw)
+    return output
+
+
+def http_json(method: str, url: str, payload: Any = None, headers: dict[str, str] | None = None, timeout: float = 20) -> Any:
+    data = None if payload is None else json.dumps(payload).encode("utf-8")
+    request_headers = {"Accept": "application/json", **(headers or {})}
+    if data is not None:
+        request_headers["Content-Type"] = "application/json"
+    request = urllib.request.Request(url, data=data, headers=request_headers, method=method.upper())
+    try:
+        with urllib.request.urlopen(request, timeout=timeout) as response:
+            body = response.read()
+    except urllib.error.HTTPError as exc:
+        detail = exc.read().decode("utf-8", "replace")[:300]
+        raise RuntimeError(f"gmaps_http_{exc.code}:{detail}") from exc
+    except urllib.error.URLError as exc:
+        raise RuntimeError(f"gmaps_unreachable:{exc.reason}") from exc
+    try:
+        return json.loads(body or b"{}")
+    except json.JSONDecodeError as exc:
+        raise RuntimeError("gmaps_invalid_json") from exc
+
+
+def gmaps_headers() -> dict[str, str]:
+    if not GMAPS_API_KEY:
+        raise RuntimeError("gmaps_api_key_missing")
+    return {"X-API-Key": GMAPS_API_KEY}
+
+
+def submit_gmaps_job(query: str, include_email: bool = False, extra_reviews: bool = False) -> str:
+    if not GMAPS_URL:
+        raise RuntimeError("gmaps_not_configured")
+    payload = {
+        "keyword": query,
+        "lang": os.environ.get("GMAPS_LANG", "en"),
+        "max_depth": max(1, min(5, int(os.environ.get("GMAPS_MAX_DEPTH", "1")))),
+        "email": bool(include_email),
+        "extra_reviews": bool(extra_reviews),
+        "timeout": max(1, min(300, GMAPS_REQUEST_TIMEOUT)),
+    }
+    response = http_json("POST", GMAPS_URL + "/api/v1/scrape", payload, gmaps_headers(), timeout=30)
+    job_id = response.get("job_id") or response.get("id")
+    if not job_id:
+        raise RuntimeError("gmaps_job_id_missing")
+    return str(job_id)
+
+
+def poll_gmaps_job(job_id: str, progress: Callable[[str, int, int], None] | None = None) -> list[dict[str, Any]]:
+    if not GMAPS_URL:
+        raise RuntimeError("gmaps_not_configured")
+    deadline = time.monotonic() + max(60, GMAPS_REQUEST_TIMEOUT + 60)
+    terminal_ok = {"completed", "succeeded", "success"}
+    terminal_bad = {"failed", "cancelled", "canceled", "discarded"}
+    while time.monotonic() < deadline:
+        response = http_json("GET", GMAPS_URL + "/api/v1/jobs/" + quote_plus(str(job_id)), headers=gmaps_headers(), timeout=30)
+        status = str(response.get("status") or "pending").lower()
+        result_count = as_int(response.get("result_count"))
+        if progress:
+            progress(status, result_count, 0)
+        if status in terminal_ok:
+            return extract_gmaps_results(response.get("results") if "results" in response else response)
+        if status in terminal_bad:
+            raise RuntimeError("gmaps_job_" + status + (":" + as_text(response.get("error")) if response.get("error") else ""))
+        time.sleep(max(0.5, GMAPS_POLL_SECONDS))
+    raise RuntimeError("gmaps_job_timeout")
+
+
+# The browser fallback is bounded and does not bypass challenge pages. It is
+# retained for local testing or when the user deliberately sets backend=browser.
+SEARCH_DEADLINE_SECONDS = 25
+ENGINE_TIMEOUT_MS = 7000
+
+
 def blocked(text: str) -> bool:
     lowered = text.lower()
     return any(marker in lowered for marker in (
@@ -82,12 +365,6 @@ def blocked(text: str) -> bool:
 
 
 def normalize_result_href(raw_href: str) -> str:
-    """Turn a search-result href into the public destination URL.
-
-    Engines commonly use protocol-relative links or redirect wrappers (for
-    example DuckDuckGo's ``uddg`` parameter).  We unwrap those links before
-    applying the catalog/domain filters.
-    """
     href = (raw_href or "").strip()
     if href.startswith("//"):
         href = "https:" + href
@@ -105,9 +382,8 @@ def normalize_result_href(raw_href: str) -> str:
 
 
 def extract_search_results(page: Any, selector: str, title_selector: str | None,
-                           excluded_hosts: set[str], requested: int) -> list[dict[str, str]]:
-    """Extract external result links, with a generic-anchor safety net."""
-    output: list[dict[str, str]] = []
+                           excluded_hosts: set[str], requested: int) -> list[dict[str, Any]]:
+    output: list[dict[str, Any]] = []
     seen_domains: set[str] = set()
 
     def add_anchor(anchor: Any) -> None:
@@ -127,10 +403,8 @@ def extract_search_results(page: Any, selector: str, title_selector: str | None,
         if not domain or domain in excluded_hosts or domain in seen_domains:
             return
         seen_domains.add(domain)
-        output.append({"name": title, "website": href, "domain": domain})
+        output.append({"title": title, "website": href, "name": title})
 
-    # Prefer stable engine-specific result selectors to avoid collecting nav
-    # and footer links.  If markup changed, fall back to generic anchors.
     try:
         anchors = page.locator(selector)
         for i in range(min(anchors.count(), requested * 4)):
@@ -149,13 +423,7 @@ def extract_search_results(page: Any, selector: str, title_selector: str | None,
     return output
 
 
-def google_search(query: str, requested: int) -> list[dict[str, str]]:
-    """Search public engines through Playwright, with safe engine fallback.
-
-    Google frequently challenges datacenter IPs. We do not bypass that
-    challenge; we close the context and try other public engines, still using
-    a real headless browser and the same bounded result extraction.
-    """
+def google_search(query: str, requested: int) -> list[dict[str, Any]]:
     try:
         from playwright.sync_api import sync_playwright, TimeoutError as PlaywrightTimeout
     except ImportError as exc:
@@ -165,64 +433,58 @@ def google_search(query: str, requested: int) -> list[dict[str, str]]:
     if executable:
         options["executable_path"] = executable
     engines = [
-        ("google", "https://www.google.com/search?q=" + quote_plus(query) + "&num=" + str(requested), "a:has(h3)", "h3", {"google.com", "google.co.za"}),
-        ("bing", "https://www.bing.com/search?q=" + quote_plus(query) + "&count=" + str(requested), "li.b_algo h2 a", "h2", {"bing.com", "bingj.com"}),
-        ("duckduckgo", "https://html.duckduckgo.com/html/?q=" + quote_plus(query), "a.result__a,a[data-testid='result-title-a']", None, {"duckduckgo.com"}),
         ("brave", "https://search.brave.com/search?q=" + quote_plus(query), "a.heading-serpresult,a.result-header", None, {"search.brave.com"}),
         ("startpage", "https://www.startpage.com/sp/search?query=" + quote_plus(query), "a.w-gl__result-title", None, {"startpage.com"}),
+        ("duckduckgo", "https://html.duckduckgo.com/html/?q=" + quote_plus(query), "a.result__a,a[data-testid='result-title-a']", None, {"duckduckgo.com"}),
+        ("bing", "https://www.bing.com/search?q=" + quote_plus(query) + "&count=" + str(requested), "li.b_algo h2 a", "h2", {"bing.com", "bingj.com"}),
+        ("google", "https://www.google.com/search?q=" + quote_plus(query) + "&num=" + str(requested), "a:has(h3)", "h3", {"google.com", "google.co.za"}),
     ]
     errors: list[str] = []
+    deadline = time.monotonic() + SEARCH_DEADLINE_SECONDS
     with sync_playwright() as p:
         browser = p.chromium.launch(**options)
         try:
-          for engine, url, selector, title_selector, excluded_hosts in engines:
-            context = None
-            try:
-                context = browser.new_context(
-                    locale="en-ZA",
-                    java_script_enabled=True,
-                    service_workers="block",
-                    viewport={"width": 1280, "height": 900},
-                    user_agent=(
-                        "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
-                        "(KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36"
-                    ),
-                )
-                page = context.new_page()
-                page.set_default_timeout(12000)
-                navigation_timed_out = False
-                # DOMContentLoaded gives us usable markup without waiting for
-                # analytics, ads, or other background requests.  If an engine
-                # keeps navigation open, still inspect the partially loaded
-                # page rather than discarding it as a hard failure.
+            for engine, url, selector, title_selector, excluded_hosts in engines:
+                remaining = deadline - time.monotonic()
+                if remaining <= 0:
+                    errors.append("search_deadline_exceeded")
+                    break
+                context = None
                 try:
-                    page.goto(url, wait_until="domcontentloaded", timeout=15000)
+                    context = browser.new_context(locale="en-ZA", java_script_enabled=True, service_workers="block", viewport={"width": 1280, "height": 900})
+                    page = context.new_page()
+                    page.set_default_timeout(ENGINE_TIMEOUT_MS)
+                    navigation_timed_out = False
+                    try:
+                        page.goto(url, wait_until="domcontentloaded", timeout=min(ENGINE_TIMEOUT_MS, max(1000, int(remaining * 1000))))
+                    except PlaywrightTimeout:
+                        navigation_timed_out = True
+                    page.wait_for_timeout(min(500, max(0, int((deadline - time.monotonic()) * 1000))))
+                    body = page.locator("body").inner_text()
+                    html = page.content()
+                    if blocked(body) or blocked(html):
+                        errors.append(engine + "_blocked")
+                        continue
+                    output = extract_search_results(page, selector, title_selector, excluded_hosts, requested)
+                    if output:
+                        return output
+                    errors.append(engine + ("_timeout" if navigation_timed_out else "_empty"))
                 except PlaywrightTimeout:
-                    navigation_timed_out = True
-                page.wait_for_timeout(1000)
-                body = page.locator("body").inner_text()
-                html = page.content()
-                if blocked(body) or blocked(html):
-                    errors.append(engine + "_blocked")
-                    continue
-                output = extract_search_results(page, selector, title_selector, excluded_hosts, requested)
-                if output:
-                    return output
-                errors.append(engine + ("_timeout" if navigation_timed_out else "_empty"))
-            except PlaywrightTimeout:
-                errors.append(engine + "_timeout")
-            except Exception as exc:
-                errors.append(engine + "_error")
-            finally:
-                if context is not None:
-                    try: context.close()
-                    except Exception: pass
+                    errors.append(engine + "_timeout")
+                except Exception:
+                    errors.append(engine + "_error")
+                finally:
+                    if context is not None:
+                        try:
+                            context.close()
+                        except Exception:
+                            pass
         finally:
             browser.close()
     raise RuntimeError("search_engines_unavailable:" + ",".join(errors))
 
 
-def run_discovery(payload: dict[str, Any]) -> dict[str, Any]:
+def validate_payload(payload: dict[str, Any]) -> tuple[str, str, str, int, bool, bool]:
     category = str(payload.get("category", "")).strip()
     keywords = str(payload.get("keywords", "")).strip()
     location = str(payload.get("location", "South Africa")).strip() or "South Africa"
@@ -233,34 +495,102 @@ def run_discovery(payload: dict[str, Any]) -> dict[str, Any]:
     except (TypeError, ValueError) as exc:
         raise ValueError("invalid_max_results") from exc
     query = f"{keywords} {category} {location}".strip()
-    with _discover_lock:
-        conn = db()
-        cur = conn.execute("INSERT INTO runs(query,location,requested,status,created_at) VALUES(?,?,?,?,?)", (query, location, requested, "running", now()))
-        run_id = cur.lastrowid
+    include_email = bool(payload.get("include_email", False))
+    extra_reviews = bool(payload.get("extra_reviews", False))
+    return category, query, location, requested, include_email, extra_reviews
+
+
+def update_run(run_id: int, **fields: Any) -> None:
+    if not fields:
+        return
+    allowed = {"found", "added", "status", "error", "source", "external_job_id", "stage", "progress", "completed_at"}
+    safe = {key: value for key, value in fields.items() if key in allowed}
+    if not safe:
+        return
+    assignments = ", ".join(f"{key}=?" for key in safe)
+    conn = db()
+    try:
+        conn.execute(f"UPDATE runs SET {assignments} WHERE id=?", (*safe.values(), run_id))
         conn.commit()
+    finally:
+        conn.close()
+
+
+def insert_catalogue(run_id: int, query: str, raw_results: list[dict[str, Any]], requested: int) -> tuple[list[dict[str, Any]], int]:
+    mapped = [map_gmaps_result(item, query) for item in raw_results[:requested]]
+    conn = db()
+    added = 0
+    try:
+        for item in mapped:
+            cur = conn.execute("""INSERT OR IGNORE INTO prospects(
+                name,website,domain,description,query,discovered_at,source,source_id,
+                category,address,phone,emails,rating,review_count,maps_url,latitude,
+                longitude,listing_status,website_status,opportunity_score,raw_json
+            ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", (
+                item["name"], item["website"], item["domain"], item["description"], item["query"], now(),
+                item["source"], item["source_id"], item["category"], item["address"], item["phone"], item["emails"],
+                item["rating"], item["review_count"], item["maps_url"], item["latitude"], item["longitude"],
+                item["listing_status"], item["website_status"], item["opportunity_score"], item["raw_json"],
+            ))
+            if cur.rowcount:
+                added += 1
+        conn.commit()
+    finally:
+        conn.close()
+    return mapped, added
+
+
+def run_discovery(run_id: int, payload: dict[str, Any]) -> None:
+    category, query, location, requested, include_email, extra_reviews = validate_payload(payload)
+    # Serialize local runs so two personal browser jobs do not fight over the
+    # same machine's Chromium/RAM. The Maps worker itself remains async.
+    with _discover_lock:
+        update_run(run_id, status="running", stage="connecting", progress=5, source=backend_mode())
         try:
-            results = google_search(query, requested)
-            added = 0
-            for item in results:
-                try:
-                    conn.execute("INSERT INTO prospects(name,website,domain,description,query,discovered_at) VALUES(?,?,?,?,?,?)", (item["name"], item["website"], item["domain"], "Google search result", query, now()))
-                    added += 1
-                except sqlite3.IntegrityError:
-                    pass
-            conn.execute("UPDATE runs SET found=?,added=?,status='succeeded' WHERE id=?", (len(results), added, run_id))
-            conn.commit()
-            return {"id": run_id, "status": "succeeded", "query": query, "found": len(results), "added": added, "results": results}
+            if backend_mode() == "gmaps":
+                external_job_id = submit_gmaps_job(query, include_email, extra_reviews)
+                update_run(run_id, external_job_id=external_job_id, stage="scraping", progress=10)
+
+                def on_progress(status: str, found: int, _unused: int) -> None:
+                    progress = 15 if status == "pending" else 45 if status == "running" else 75
+                    update_run(run_id, stage="scraping", progress=progress, found=found)
+
+                raw_results = poll_gmaps_job(external_job_id, on_progress)
+            else:
+                update_run(run_id, stage="browser_search", progress=20)
+                raw_results = google_search(query, requested)
+            results, added = insert_catalogue(run_id, query, raw_results, requested)
+            update_run(run_id, found=len(results), added=added, status="succeeded", stage="complete", progress=100, completed_at=now(), error="")
         except Exception as exc:
-            error = str(exc)[:120]
-            conn.execute("UPDATE runs SET status='failed',error=? WHERE id=?", (error, run_id))
-            conn.commit()
-            raise RuntimeError(error)
-        finally:
-            conn.close()
+            message = str(exc).strip()[:500] or "discovery_failed"
+            update_run(run_id, status="failed", stage="failed", progress=100, error=message, completed_at=now())
+
+
+def create_discovery(payload: dict[str, Any]) -> dict[str, Any]:
+    category, query, location, requested, include_email, extra_reviews = validate_payload(payload)
+    request_json = json.dumps({
+        "category": category,
+        "keywords": str(payload.get("keywords", "")).strip(),
+        "location": location,
+        "max_results": requested,
+        "include_email": include_email,
+        "extra_reviews": extra_reviews,
+    }, ensure_ascii=False)
+    conn = db()
+    try:
+        cur = conn.execute("""INSERT INTO runs(query,location,requested,status,created_at,source,stage,request_json)
+            VALUES(?,?,?,?,?,?,?,?)""", (query, location, requested, "queued", now(), backend_mode(), "queued", request_json))
+        run_id = int(cur.lastrowid)
+        conn.commit()
+    finally:
+        conn.close()
+    worker = threading.Thread(target=run_discovery, args=(run_id, payload), name=f"prospect-run-{run_id}", daemon=True)
+    worker.start()
+    return {"id": run_id, "status": "queued", "source": backend_mode(), "query": query, "requested": requested}
 
 
 class Handler(BaseHTTPRequestHandler):
-    server_version = "BrowserProspectOS/1.0"
+    server_version = "PersonalProspectOS/2.0"
 
     def send_json(self, status: int, value: Any) -> None:
         body = json_bytes(value)
@@ -286,22 +616,46 @@ class Handler(BaseHTTPRequestHandler):
         conn = db()
         try:
             if parsed.path == "/api/health":
-                return self.send_json(200, {"status": "ok", "source": "google_playwright"})
+                configured = bool(GMAPS_URL and GMAPS_API_KEY) if backend_mode() == "gmaps" else True
+                return self.send_json(200, {"status": "ok", "source": backend_mode(), "configured": configured, "maps_service": bool(GMAPS_URL), "maps_api_key": bool(GMAPS_API_KEY)})
+            if parsed.path == "/api/config":
+                return self.send_json(200, {
+                    "backend": backend_mode(),
+                    "gmaps_url": bool(GMAPS_URL),
+                    "gmaps_api_key": bool(GMAPS_API_KEY),
+                    "ready": backend_mode() == "browser" or bool(GMAPS_URL and GMAPS_API_KEY),
+                    "email_extraction_default": False,
+                })
             if parsed.path == "/api/prospects":
                 params = parse_qs(parsed.query)
                 q = (params.get("q", [""])[0]).strip().lower()
                 status = params.get("status", [""])[0].strip().lower()
-                rows = conn.execute("SELECT * FROM prospects ORDER BY id DESC").fetchall()
-                items = [dict(row) for row in rows if (not q or q in (row["name"] + " " + row["domain"] + " " + row["description"]).lower()) and (not status or row["review_status"] == status)]
+                rows = conn.execute("SELECT * FROM prospects ORDER BY opportunity_score DESC, id DESC").fetchall()
+                items = []
+                for row in rows:
+                    haystack = " ".join(str(row[key] or "") for key in ("name", "domain", "description", "address", "category", "phone", "emails")).lower()
+                    if q and q not in haystack:
+                        continue
+                    if status and row["review_status"] != status:
+                        continue
+                    items.append(dict(row))
                 return self.send_json(200, {"items": items, "total": len(items)})
             if parsed.path == "/api/runs":
                 rows = [dict(row) for row in conn.execute("SELECT * FROM runs ORDER BY id DESC LIMIT 50")]
                 return self.send_json(200, {"items": rows})
+            match = re.fullmatch(r"/api/runs/(\d+)", parsed.path)
+            if match:
+                row = conn.execute("SELECT * FROM runs WHERE id=?", (int(match.group(1)),)).fetchone()
+                if not row:
+                    return self.send_json(404, {"error": "not_found"})
+                return self.send_json(200, dict(row))
             if parsed.path == "/api/stats":
                 total = conn.execute("SELECT COUNT(*) FROM prospects").fetchone()[0]
                 pending = conn.execute("SELECT COUNT(*) FROM prospects WHERE review_status='pending'").fetchone()[0]
                 approved = conn.execute("SELECT COUNT(*) FROM prospects WHERE review_status='approved'").fetchone()[0]
-                return self.send_json(200, {"total": total, "pending": pending, "approved": approved})
+                contacted = conn.execute("SELECT COUNT(*) FROM prospects WHERE review_status='contacted'").fetchone()[0]
+                running = conn.execute("SELECT COUNT(*) FROM runs WHERE status IN ('queued','running')").fetchone()[0]
+                return self.send_json(200, {"total": total, "pending": pending, "approved": approved, "contacted": contacted, "running": running})
             return self.send_json(404, {"error": "not_found"})
         finally:
             conn.close()
@@ -332,17 +686,18 @@ class Handler(BaseHTTPRequestHandler):
         if self.path != "/api/discover":
             return self.send_json(404, {"error": "not_found"})
         try:
-            result = run_discovery(self.read_json())
-            return self.send_json(201, result)
+            result = create_discovery(self.read_json())
+            return self.send_json(202, result)
         except ValueError as exc:
             return self.send_json(400, {"error": str(exc)})
-        except RuntimeError as exc:
-            return self.send_json(502, {"error": str(exc)})
+        except (TypeError, json.JSONDecodeError) as exc:
+            return self.send_json(400, {"error": str(exc)})
 
     def do_PATCH(self) -> None:
         match = re.fullmatch(r"/api/prospects/(\d+)/review", urlparse(self.path).path)
         if not match:
             return self.send_json(404, {"error": "not_found"})
+        conn: sqlite3.Connection | None = None
         try:
             payload = self.read_json()
             status = str(payload.get("status", "")).strip().lower()
@@ -358,15 +713,15 @@ class Handler(BaseHTTPRequestHandler):
         except (ValueError, TypeError, json.JSONDecodeError) as exc:
             return self.send_json(400, {"error": str(exc)})
         finally:
-            try: conn.close()
-            except Exception: pass
+            if conn is not None:
+                conn.close()
 
 
 def serve() -> None:
     host = os.environ.get("HOST", "0.0.0.0")
     port = int(os.environ.get("PORT", "8000"))
     db()
-    print(f"Browser ProspectOS listening on http://{host}:{port}", flush=True)
+    print(f"Personal ProspectOS listening on http://{host}:{port} (backend={backend_mode()})", flush=True)
     ThreadingHTTPServer((host, port), Handler).serve_forever()