MarketingTool/api/server.py

#!/usr/bin/env python3
"""Personal ProspectOS service backed by Google Maps Scraper.

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

import csv
import hashlib
import io
import json
import os
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, Callable
from urllib.parse import parse_qs, quote_plus, unquote, urlparse

DB_PATH = os.environ.get("PROSPECT_DB", "/data/prospects.db")
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("/")
# The bundled gosom/google-maps-scraper image exposes a local, unauthenticated
# REST API. The optional SaaS edition uses a different API and requires an
# X-API-Key header; keep support for it only when explicitly selected.
GMAPS_API_MODE = os.environ.get("GMAPS_API_MODE", "standard").strip().lower()
GMAPS_API_KEY = os.environ.get("GMAPS_API_KEY", "").strip()
ALLOWED_REVIEW = {"pending", "approved", "rejected", "contacted"}
_discover_lock = threading.Lock()


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, 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 DEFAULT '',
        domain TEXT NOT NULL UNIQUE,
        description TEXT NOT NULL DEFAULT '',
        query TEXT NOT NULL DEFAULT '',
        discovered_at TEXT NOT NULL,
        review_status TEXT NOT NULL DEFAULT 'pending',
        review_notes TEXT NOT NULL DEFAULT '',
        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,
        query TEXT NOT NULL,
        location TEXT NOT NULL,
        requested INTEGER NOT NULL,
        found INTEGER NOT NULL DEFAULT 0,
        added INTEGER NOT NULL DEFAULT 0,
        status TEXT NOT NULL,
        error TEXT NOT NULL DEFAULT '',
        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


def json_bytes(value: Any) -> bytes:
    return json.dumps(value, ensure_ascii=False).encode("utf-8")


def clean_domain(url: str) -> str:
    parsed = urlparse(url if "://" in url else "https://" + url)
    host = (parsed.hostname or "").lower().strip().rstrip(".")
    if host.startswith("www."):
        host = host[4:]
    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 http_bytes(method: str, url: str, payload: Any = None, headers: dict[str, str] | None = None, timeout: float = 20) -> bytes:
    """Make a small HTTP request where the response is not JSON (CSV download)."""
    data = None if payload is None else json.dumps(payload).encode("utf-8")
    request_headers = {"Accept": "*/*", **(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:
            return 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


def gmaps_headers() -> dict[str, str]:
    if GMAPS_API_MODE == "saas":
        if not GMAPS_API_KEY:
            raise RuntimeError("gmaps_api_key_missing_for_saas_mode")
        return {"X-API-Key": GMAPS_API_KEY}
    return {}


def download_standard_results(job_id: str) -> list[dict[str, Any]]:
    """Download and parse the CSV produced by the standard web worker."""
    body = http_bytes(
        "GET",
        GMAPS_URL + "/api/v1/jobs/" + quote_plus(str(job_id)) + "/download",
        headers=gmaps_headers(),
        timeout=30,
    )
    try:
        text = body.decode("utf-8-sig")
        return [dict(row) for row in csv.DictReader(io.StringIO(text))]
    except (UnicodeDecodeError, csv.Error) as exc:
        raise RuntimeError("gmaps_csv_invalid") from exc


def submit_gmaps_job(query: str, include_email: bool = False, extra_reviews: bool = False) -> str:
    if not GMAPS_URL:
        raise RuntimeError("gmaps_not_configured")
    lang = os.environ.get("GMAPS_LANG", "en")
    depth = max(1, min(10, int(os.environ.get("GMAPS_MAX_DEPTH", "1"))))
    if GMAPS_API_MODE == "saas":
        endpoint = "/api/v1/scrape"
        payload = {
            "keyword": query,
            "lang": lang,
            "max_depth": depth,
            "email": bool(include_email),
            "extra_reviews": bool(extra_reviews),
            "timeout": max(1, min(300, GMAPS_REQUEST_TIMEOUT)),
        }
    else:
        # Standard gosom web mode uses the local job API. It has no API-key
        # authentication and returns a UUID in {"id": ...} with HTTP 201.
        endpoint = "/api/v1/jobs"
        payload = {
            "name": query[:180],
            "keywords": [query],
            "lang": lang,
            "zoom": 15,
            "depth": depth,
            "email": bool(include_email),
            "extra_reviews": bool(extra_reviews),
            "max_time": max(60, min(3600, GMAPS_REQUEST_TIMEOUT)),
        }
    response = http_json("POST", GMAPS_URL + endpoint, 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)
    if GMAPS_API_MODE == "saas":
        terminal_ok = {"completed", "succeeded", "success"}
    else:
        terminal_ok = {"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:
            if "results" in response:
                return extract_gmaps_results(response.get("results"))
            if GMAPS_API_MODE == "standard":
                return download_standard_results(job_id)
            return extract_gmaps_results(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 (
        "unusual traffic", "recaptcha", "captcha", "automated queries",
        "access denied", "sorry...", "detected unusual traffic",
    ))


def normalize_result_href(raw_href: str) -> str:
    href = (raw_href or "").strip()
    if href.startswith("//"):
        href = "https:" + href
    if href.startswith("/"):
        return ""
    parsed = urlparse(href)
    if parsed.hostname and parsed.hostname.endswith("duckduckgo.com"):
        params = parse_qs(parsed.query)
        for key in ("uddg", "url", "u"):
            target = params.get(key, [""])[0]
            if target.startswith(("http://", "https://")):
                href = unquote(target)
                break
    return href


def extract_search_results(page: Any, selector: str, title_selector: str | None,
                           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:
        if len(output) >= requested:
            return
        try:
            title_node = anchor.locator(title_selector) if title_selector else anchor
            title = " ".join((title_node.inner_text() or "").split())[:200]
            if not title:
                title = " ".join(((anchor.get_attribute("aria-label") or anchor.get_attribute("title") or "").split()))[:200]
            href = normalize_result_href(anchor.get_attribute("href") or "")
        except Exception:
            return
        domain = clean_domain(href)
        if not title or len(title) < 3 or not href.startswith(("http://", "https://")):
            return
        if not domain or domain in excluded_hosts or domain in seen_domains:
            return
        seen_domains.add(domain)
        output.append({"title": title, "website": href, "name": title})

    try:
        anchors = page.locator(selector)
        for i in range(min(anchors.count(), requested * 4)):
            add_anchor(anchors.nth(i))
    except Exception:
        pass
    if not output:
        try:
            anchors = page.locator("a")
            for i in range(min(anchors.count(), 250)):
                add_anchor(anchors.nth(i))
                if len(output) >= requested:
                    break
        except Exception:
            pass
    return output


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:
        raise RuntimeError("playwright_not_installed") from exc
    executable = os.environ.get("PLAYWRIGHT_CHROMIUM_EXECUTABLE_PATH", "").strip()
    options: dict[str, Any] = {"headless": True}
    if executable:
        options["executable_path"] = executable
    engines = [
        ("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:
                remaining = deadline - time.monotonic()
                if remaining <= 0:
                    errors.append("search_deadline_exceeded")
                    break
                context = None
                try:
                    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:
                    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 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"
    if not category or not keywords:
        raise ValueError("category_and_keywords_required")
    try:
        requested = max(1, min(MAX_RESULTS, int(payload.get("max_results", 10))))
    except (TypeError, ValueError) as exc:
        raise ValueError("invalid_max_results") from exc
    query = f"{keywords} {category} {location}".strip()
    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:
            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:
            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 = "PersonalProspectOS/2.0"

    def send_json(self, status: int, value: Any) -> None:
        body = json_bytes(value)
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Cache-Control", "no-store")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def read_json(self) -> dict[str, Any]:
        length = min(int(self.headers.get("Content-Length", "0") or 0), 100_000)
        value = json.loads(self.rfile.read(length) or b"{}")
        if not isinstance(value, dict):
            raise ValueError("json_object_required")
        return value

    def do_GET(self) -> None:
        parsed = urlparse(self.path)
        if not parsed.path.startswith("/api/"):
            self.serve_static(parsed.path)
            return
        conn = db()
        try:
            if parsed.path == "/api/health":
                configured = bool(GMAPS_URL) if backend_mode() == "gmaps" else True
                if GMAPS_API_MODE == "saas":
                    configured = configured and bool(GMAPS_API_KEY)
                return self.send_json(200, {"status": "ok", "source": backend_mode(), "configured": configured, "maps_service": bool(GMAPS_URL), "maps_api_mode": GMAPS_API_MODE, "maps_api_key": bool(GMAPS_API_KEY)})
            if parsed.path == "/api/config":
                auth_required = GMAPS_API_MODE == "saas"
                ready = backend_mode() == "browser" or (bool(GMAPS_URL) and (not auth_required or bool(GMAPS_API_KEY)))
                return self.send_json(200, {
                    "backend": backend_mode(),
                    "gmaps_url": bool(GMAPS_URL),
                    "gmaps_api_mode": GMAPS_API_MODE,
                    "gmaps_auth_required": auth_required,
                    "gmaps_api_key": bool(GMAPS_API_KEY),
                    "ready": ready,
                    "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 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]
                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()

    def serve_static(self, path: str) -> None:
        relative = "index.html" if path in {"", "/"} else path.lstrip("/")
        root = (Path(__file__).parent / "web").resolve()
        if not root.is_dir():
            root = (Path(__file__).parent.parent / "web").resolve()
        candidate = (root / relative).resolve()
        if root not in candidate.parents and candidate != root:
            return self.send_json(400, {"error": "invalid_path"})
        if not candidate.is_file():
            return self.send_json(404, {"error": "not_found"})
        content = candidate.read_bytes()
        content_type = "text/plain; charset=utf-8"
        if candidate.suffix == ".html": content_type = "text/html; charset=utf-8"
        elif candidate.suffix == ".js": content_type = "text/javascript; charset=utf-8"
        elif candidate.suffix == ".css": content_type = "text/css; charset=utf-8"
        self.send_response(200)
        self.send_header("Content-Type", content_type)
        self.send_header("Cache-Control", "no-cache")
        self.send_header("Content-Length", str(len(content)))
        self.end_headers()
        self.wfile.write(content)

    def do_POST(self) -> None:
        if self.path != "/api/discover":
            return self.send_json(404, {"error": "not_found"})
        try:
            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 (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()
            if status not in ALLOWED_REVIEW:
                return self.send_json(400, {"error": "invalid_review_status"})
            conn = db()
            cur = conn.execute("UPDATE prospects SET review_status=?,review_notes=? WHERE id=?", (status, str(payload.get("notes", ""))[:1000], int(match.group(1))))
            conn.commit()
            if not cur.rowcount:
                return self.send_json(404, {"error": "not_found"})
            row = conn.execute("SELECT * FROM prospects WHERE id=?", (int(match.group(1)),)).fetchone()
            return self.send_json(200, dict(row))
        except (ValueError, TypeError, json.JSONDecodeError) as exc:
            return self.send_json(400, {"error": str(exc)})
        finally:
            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"Personal ProspectOS listening on http://{host}:{port} (backend={backend_mode()})", flush=True)
    ThreadingHTTPServer((host, port), Handler).serve_forever()


if __name__ == "__main__":
    serve()