replace ProspectOS with BrowserOnly build

api/server.py

@@ -0,0 +1,291 @@
+#!/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()