#!/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, 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 google_search(query: str, requested: int) -> list[dict[str, str]]:
    try:
        from playwright.sync_api import sync_playwright, TimeoutError as PlaywrightTimeout
    except ImportError as exc:
        raise RuntimeError("playwright_not_installed") from exc
    url = "https://www.google.com/search?q=" + quote_plus(query) + "&num=" + str(requested)
    executable = os.environ.get("PLAYWRIGHT_CHROMIUM_EXECUTABLE_PATH", "").strip()
    options: dict[str, Any] = {"headless": True}
    if executable:
        options["executable_path"] = executable
    try:
        with sync_playwright() as p:
            browser = p.chromium.launch(**options)
            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(15000)
                page.goto(url, wait_until="domcontentloaded", timeout=15000)
                page.wait_for_timeout(250)
                body = page.locator("body").inner_text()
                html = page.content()
                if blocked(body) or blocked(html):
                    raise RuntimeError("google_blocked")
                output: list[dict[str, str]] = []
                anchors = page.locator("a:has(h3)")
                for i in range(min(anchors.count(), requested * 3)):
                    anchor = anchors.nth(i)
                    title = " ".join((anchor.locator("h3").inner_text() or "").split())[:200]
                    href = anchor.get_attribute("href") or ""
                    domain = clean_domain(href)
                    if not title or not href or domain in {"google.com", "google.co.za"} or not domain:
                        continue
                    if not href.startswith(("http://", "https://")):
                        continue
                    if domain not in {item["domain"] for item in output}:
                        output.append({"name": title, "website": href, "domain": domain})
                    if len(output) >= requested:
                        break
                return output
            finally:
                try:
                    context.close()
                except Exception:
                    pass
                browser.close()
    except RuntimeError:
        raise
    except PlaywrightTimeout as exc:
        raise RuntimeError("browser_timeout") from exc
    except Exception as exc:
        raise RuntimeError("browser_error") from exc


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()