#!/usr/bin/env python3
"""Small browser-only prospect discovery service.

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.
"""
from __future__ import annotations

import hashlib
import json
import os
import re
import sqlite3
import threading
import time
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Any
from urllib.parse import parse_qs, quote_plus, unquote, urlparse

DB_PATH = os.environ.get("PROSPECT_DB", "/data/prospects.db")
MAX_RESULTS = 25
ALLOWED_REVIEW = {"pending", "approved", "rejected", "contacted"}
_discover_lock = threading.Lock()


def now() -> str:
    return datetime.now(timezone.utc).replace(microsecond=0).isoformat()


def db() -> sqlite3.Connection:
    os.makedirs(os.path.dirname(DB_PATH) or ".", exist_ok=True)
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    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,
        description TEXT NOT NULL DEFAULT '',
        query TEXT NOT NULL,
        discovered_at TEXT NOT NULL,
        review_status TEXT NOT NULL DEFAULT 'pending',
        review_notes TEXT NOT NULL DEFAULT '',
        UNIQUE(domain)
    )""")
    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
    )""")
    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 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:
    """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
    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, str]]:
    """Extract external result links, with a generic-anchor safety net."""
    output: list[dict[str, str]] = []
    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({"name": title, "website": href, "domain": domain})

    # 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)):
            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, 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.
    """
    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 = [
        ("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"}),
    ]
    errors: list[str] = []
    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.
                try:
                    page.goto(url, wait_until="domcontentloaded", timeout=15000)
                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
        finally:
            browser.close()
    raise RuntimeError("search_engines_unavailable:" + ",".join(errors))


def run_discovery(payload: dict[str, Any]) -> dict[str, Any]:
    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()
    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
        conn.commit()
        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}
        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()


class Handler(BaseHTTPRequestHandler):
    server_version = "BrowserProspectOS/1.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":
                return self.send_json(200, {"status": "ok", "source": "google_playwright"})
            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)]
                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})
            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})
            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 = run_discovery(self.read_json())
            return self.send_json(201, 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)})

    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"})
        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:
            try: conn.close()
            except Exception: pass


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)
    ThreadingHTTPServer((host, port), Handler).serve_forever()


if __name__ == "__main__":
    serve()