#!/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 try: from intelligence import domain_candidates, scan_website except ImportError: # tests import the module as api.server from api.intelligence import domain_candidates, scan_website DB_PATH = os.environ.get("PROSPECT_DB", "/data/prospects.db") MAX_RESULTS = 50 GMAPS_POLL_SECONDS = float(os.environ.get("GMAPS_POLL_SECONDS", "2")) # Maps searches can legitimately take several minutes, especially when email # extraction is enabled. Fifteen minutes is the safe default; it remains # configurable for smaller or larger personal runs. GMAPS_REQUEST_TIMEOUT = int(os.environ.get("GMAPS_REQUEST_TIMEOUT", "900")) 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() AI_BASE_URL = os.environ.get("AI_BASE_URL", "").strip().rstrip("/") AI_API_KEY = os.environ.get("AI_API_KEY", "").strip() AI_MODEL = os.environ.get("AI_MODEL", "").strip() AI_PROVIDER = os.environ.get("AI_PROVIDER", "custom").strip().lower() NOUS_BASE_URL = "https://inference-api.nousresearch.com/v1" NOUS_FREE_MODELS = [ {"id": "meituan/longcat-2.0:free", "label": "Meituan · LongCat 2.0", "free": True}, {"id": "upstage/solar-pro-4:free", "label": "Upstage · Solar Pro 4", "free": True}, {"id": "stepfun/step-3.7-flash:free", "label": "StepFun · Step 3.7 Flash", "free": True}, {"id": "poolside/laguna-s-2.1:free", "label": "Poolside · Laguna S 2.1", "free": True}, {"id": "poolside/laguna-xs-2.1:free", "label": "Poolside · Laguna XS 2.1", "free": True}, {"id": "inclusionai/ling-3.0-flash-fin:free", "label": "inclusionAI · Ling 3.0 Flash Fin", "free": True}, {"id": "inclusionai/ling-3.0-flash-sante:free", "label": "inclusionAI · Ling 3.0 Flash Sante", "free": True}, ] _nous_models_cache: list[dict[str, Any]] = NOUS_FREE_MODELS.copy() _nous_models_cached_at = 0.0 ALLOWED_REVIEW = {"pending", "approved", "rejected", "contacted"} _discover_lock = threading.Lock() _schema_lock = threading.Lock() _schema_path: str | None = None 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: global _schema_path 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") if _schema_path == DB_PATH: return conn _schema_lock.acquire() if _schema_path == DB_PATH: _schema_lock.release() return conn 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', website_http_status INTEGER, website_final_url TEXT NOT NULL DEFAULT '', website_latency_ms INTEGER NOT NULL DEFAULT 0, website_https INTEGER NOT NULL DEFAULT 0, website_title TEXT NOT NULL DEFAULT '', website_technology TEXT NOT NULL DEFAULT '', website_has_viewport INTEGER NOT NULL DEFAULT 0, website_tls_expires_at TEXT NOT NULL DEFAULT '', website_checked_at TEXT NOT NULL DEFAULT '', website_error TEXT NOT NULL DEFAULT '', email_evidence TEXT NOT NULL DEFAULT '[]', score_reasons TEXT NOT NULL DEFAULT '[]', ai_probability REAL, ai_recommendation TEXT NOT NULL DEFAULT '', ai_reasoning TEXT NOT NULL DEFAULT '', ai_checked_at TEXT NOT NULL DEFAULT '', ai_subject TEXT NOT NULL DEFAULT '', ai_message TEXT NOT NULL DEFAULT '', ai_offer TEXT NOT NULL DEFAULT '', 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 '{}' )""") conn.execute("""CREATE TABLE IF NOT EXISTS suppressions ( id INTEGER PRIMARY KEY AUTOINCREMENT, kind TEXT NOT NULL, value TEXT NOT NULL, reason TEXT NOT NULL DEFAULT '', created_at TEXT NOT NULL, UNIQUE(kind, value) )""") conn.execute("""CREATE TABLE IF NOT EXISTS schedules ( id INTEGER PRIMARY KEY AUTOINCREMENT, category TEXT NOT NULL, keywords TEXT NOT NULL, location TEXT NOT NULL DEFAULT 'South Africa', max_results INTEGER NOT NULL DEFAULT 25, interval_hours INTEGER NOT NULL DEFAULT 24, enabled INTEGER NOT NULL DEFAULT 1, next_run_at TEXT NOT NULL, last_run_id INTEGER, created_at TEXT NOT NULL )""") conn.execute("""CREATE TABLE IF NOT EXISTS settings ( key TEXT PRIMARY KEY, value TEXT NOT NULL DEFAULT '', updated_at TEXT NOT NULL )""") # 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'", "website_http_status": "INTEGER", "website_final_url": "TEXT NOT NULL DEFAULT ''", "website_latency_ms": "INTEGER NOT NULL DEFAULT 0", "website_https": "INTEGER NOT NULL DEFAULT 0", "website_title": "TEXT NOT NULL DEFAULT ''", "website_technology": "TEXT NOT NULL DEFAULT ''", "website_has_viewport": "INTEGER NOT NULL DEFAULT 0", "website_tls_expires_at": "TEXT NOT NULL DEFAULT ''", "website_checked_at": "TEXT NOT NULL DEFAULT ''", "website_error": "TEXT NOT NULL DEFAULT ''", "email_evidence": "TEXT NOT NULL DEFAULT '[]'", "score_reasons": "TEXT NOT NULL DEFAULT '[]'", "ai_probability": "REAL", "ai_recommendation": "TEXT NOT NULL DEFAULT ''", "ai_reasoning": "TEXT NOT NULL DEFAULT ''", "ai_checked_at": "TEXT NOT NULL DEFAULT ''", "ai_subject": "TEXT NOT NULL DEFAULT ''", "ai_message": "TEXT NOT NULL DEFAULT ''", "ai_offer": "TEXT NOT NULL DEFAULT ''", "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)") conn.commit() _schema_path = DB_PATH _schema_lock.release() return conn def recover_stale_runs() -> None: """Recover abandoned runs once at process startup, not on every request.""" conn = db() try: 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() finally: conn.close() 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 field_value(value: Any, *names: str) -> Any: """Read JSON fields from both tagged and default Go JSON casing.""" if not isinstance(value, dict): return None for name in names: if name in value: return value[name] wanted = {name.lower() for name in names} for key, item in value.items(): if str(key).lower() in wanted: return item return None 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 EMAIL_RE = re.compile(r"[A-Z0-9._%+\-]+@[A-Z0-9.\-]+\.[A-Z]{2,63}", re.I) FREE_EMAIL_DOMAINS = {"gmail.com", "googlemail.com", "outlook.com", "hotmail.com", "yahoo.com", "icloud.com"} def normalize_public_emails(value: Any) -> list[str]: found: list[str] = [] text = as_text(value) for address in EMAIL_RE.findall(text): address = address.strip(".,;:()[]{}<>").lower() if address not in found: found.append(address) return found[:30] def score_opportunity(item: dict[str, Any]) -> tuple[int, list[str]]: """Transparent, explainable triage score; it is not an AI verdict.""" score = 20 reasons: list[str] = [] website = str(item.get("website") or "") website_status = str(item.get("website_status") or "unknown") emails = normalize_public_emails(item.get("emails")) phone = str(item.get("phone") or "") rating = as_float(item.get("rating")) review_count = as_int(item.get("review_count")) if not website or website_status in {"no_detected_website", "missing"}: score += 35 reasons.append("No detected website") elif website_status in {"broken", "unreachable", "error"}: score += 30 reasons.append("Website appears unavailable or broken") elif website_status == "slow": score += 15 reasons.append("Website response is slow") elif website_status == "http_only": score += 10 reasons.append("Website has no HTTPS") elif website_status == "redirect": score += 5 reasons.append("Website relies on redirects") if emails: score += 12 reasons.append("Public email found") if any(address.split("@", 1)[-1] in FREE_EMAIL_DOMAINS for address in emails): score += 8 reasons.append("Free-mail address used") if phone: score += 10 reasons.append("Phone listed") if rating is not None and rating >= 4: score += 10 reasons.append("Strong public rating") if review_count >= 10: score += 10 reasons.append("Established review activity") if as_text(item.get("category") or item.get("type")): score += 5 return min(100, score), reasons def score_place(item: dict[str, Any], website: str, phone: str, rating: float | None, review_count: int) -> int: item = {**item, "website": website, "phone": phone, "rating": rating, "review_count": review_count, "website_status": "no_detected_website" if not website else "unknown"} return score_opportunity(item)[0] 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) email_values = normalize_public_emails(item.get("emails") or item.get("email")) emails = ", ".join(email_values) 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": "unknown" if website else "no_detected_website", "email_evidence": json.dumps([{"email": value, "source_url": website or maps_url, "found_at": now(), "method": "maps_worker", "confidence": "medium"} for value in email_values], ensure_ascii=False), "score_reasons": "[]", "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 ai_runtime_settings() -> dict[str, str]: saved: dict[str, str] = {} try: conn = db() saved = {str(r["key"]): str(r["value"]) for r in conn.execute("SELECT key,value FROM settings")} conn.close() except Exception: pass provider = saved.get("AI_PROVIDER", AI_PROVIDER).strip().lower() or "custom" base = saved.get("AI_BASE_URL", AI_BASE_URL).strip().rstrip("/") if provider == "nous": base = NOUS_BASE_URL return {"provider": provider, "base_url": base, "api_key": saved.get("AI_API_KEY", AI_API_KEY).strip(), "model": saved.get("AI_MODEL", AI_MODEL).strip()} def nous_models() -> list[dict[str, Any]]: """Return the live Nous catalog when reachable, with stable free fallbacks.""" global _nous_models_cache, _nous_models_cached_at # Avoid making every Settings refresh wait on an external catalog request. # A successful catalog is reused for ten minutes; fallback data is also # cached after a failed request and can be replaced on the next refresh. if time.monotonic() - _nous_models_cached_at < 600: return _nous_models_cache try: response = http_json("GET", NOUS_BASE_URL + "/models", timeout=10) raw = response.get("data", response) if isinstance(response, dict) else response models: list[dict[str, Any]] = [] if isinstance(raw, list): for item in raw: if not isinstance(item, dict): continue model_id = str(item.get("id") or "").strip() if not model_id: continue pricing = item.get("pricing") if isinstance(item.get("pricing"), dict) else {} prices = [str(pricing[k]).strip() for k in ("prompt", "completion", "input", "output") if k in pricing] free = bool(prices) and all(price in {"0", "0.0", "0.00"} for price in prices) models.append({"id": model_id, "label": model_id, "free": free, "context_length": item.get("context_length")}) free = [item for item in models if item.get("free")] if free: _nous_models_cache = free _nous_models_cached_at = time.monotonic() return free except Exception: pass _nous_models_cached_at = time.monotonic() return _nous_models_cache 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") include_email = True 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 = field_value(response, "job_id", "id", "JobID", "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(field_value(response, "status", "Status") or "pending").lower() result_count = as_int(field_value(response, "result_count", "ResultCount")) 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: error = field_value(response, "error", "Error") raise RuntimeError("gmaps_job_" + status + (":" + as_text(error) if error else "")) time.sleep(max(0.5, GMAPS_POLL_SECONDS)) # The worker can finish writing its CSV just after its status endpoint # exceeds our polling window. Recover a completed/partial result when it # is available rather than discarding the run as a false failure. if GMAPS_API_MODE == "standard": try: recovered = download_standard_results(job_id) if recovered: return recovered except Exception: pass 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() # Email discovery is part of this app's core use case. The server forces # it on so a client/UI change cannot accidentally create a run without # attempting public email extraction. include_email = True 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 _json_list(value: Any) -> list[Any]: try: parsed = json.loads(str(value or "[]")) return parsed if isinstance(parsed, list) else [] except (TypeError, ValueError, json.JSONDecodeError): return [] def enrich_place(item: dict[str, Any]) -> dict[str, Any]: """Add website health, public email evidence, and explainable scoring.""" prior_evidence = _json_list(item.get("email_evidence")) if item.get("website"): try: item.update(scan_website(str(item["website"]))) except Exception as exc: # a single hostile site must not abort a run item.update({"website_status": "unknown", "website_checked_at": now(), "website_error": str(exc)[:300], "website_emails": [], "email_evidence": []}) else: item.setdefault("website_status", "no_detected_website") item.setdefault("website_emails", []) item.setdefault("email_evidence", []) existing_emails = normalize_public_emails(item.get("emails")) new_emails = normalize_public_emails(item.get("website_emails")) all_emails = list(dict.fromkeys(existing_emails + new_emails))[:30] item["emails"] = ", ".join(all_emails) evidence = prior_evidence scanned_evidence = item.get("email_evidence") if isinstance(scanned_evidence, list): evidence.extend(scanned_evidence) # The Maps worker evidence was serialized before the website scan. Keep it # only once, then append the scanner's page-level evidence. seen_evidence: set[tuple[str, str]] = set() compact: list[dict[str, Any]] = [] for record in evidence: if not isinstance(record, dict): continue marker = (str(record.get("email", "")), str(record.get("source_url", ""))) if marker[0] and marker not in seen_evidence: seen_evidence.add(marker) compact.append(record) item["email_evidence"] = json.dumps(compact[:50], ensure_ascii=False) score, reasons = score_opportunity(item) item["opportunity_score"] = score item["score_reasons"] = json.dumps(reasons, ensure_ascii=False) return item def insert_catalogue(run_id: int, query: str, raw_results: list[dict[str, Any]], requested: int, progress: Callable[[int, int], None] | None = None) -> tuple[list[dict[str, Any]], int]: mapped = [map_gmaps_result(item, query) for item in raw_results[:requested]] conn = db() added = 0 try: suppressed_domains = {str(row[0]).lower() for row in conn.execute("SELECT value FROM suppressions WHERE kind='domain'")} suppressed_emails = {str(row[0]).lower() for row in conn.execute("SELECT value FROM suppressions WHERE kind='email'")} total = len(mapped) for index, item in enumerate(mapped, 1): # A hostile or slow public site must not invalidate an otherwise # successful Maps run. Keep the raw listing and mark enrichment as # unknown; it can be rescanned later from the explorer. try: item = enrich_place(item) except Exception as exc: item.update({"website_status": "unknown", "website_checked_at": now(), "website_error": "analysis_failed:" + str(exc)[:240], "email_evidence": item.get("email_evidence", "[]"), "score_reasons": "[]"}) item["opportunity_score"] = score_opportunity(item)[0] # Release SQLite's read/write transaction before the progress # callback opens its own connection to update the run row. Without # this commit, a multi-result enrichment can deadlock forever in # the website_analysis stage while waiting on the catalogue read. conn.commit() if progress: progress(index, total) emails = set(normalize_public_emails(item.get("emails"))) if item["domain"].lower() in suppressed_domains or emails.intersection(suppressed_emails): continue try: 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,website_http_status,website_final_url, website_latency_ms,website_https,website_title,website_technology,website_has_viewport, website_tls_expires_at,website_checked_at,website_error,email_evidence,score_reasons, ai_probability,ai_recommendation,ai_reasoning,ai_checked_at,ai_subject,ai_message,ai_offer,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.get("listing_status", ""), item.get("website_status", "unknown"), item.get("website_http_status"), item.get("website_final_url", ""), item.get("website_latency_ms", 0), item.get("website_https", 0), item.get("website_title", ""), item.get("website_technology", ""), item.get("website_has_viewport", 0), item.get("website_tls_expires_at", ""), item.get("website_checked_at", ""), item.get("website_error", ""), item.get("email_evidence", "[]"), item.get("score_reasons", "[]"), None, "", "", "", "", "", "", item["opportunity_score"], item["raw_json"], )) except (sqlite3.Error, TypeError, ValueError): # Keep processing the remaining listings if one malformed # provider row cannot be written. continue 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) update_run(run_id, stage="website_analysis", progress=76, found=len(raw_results)) def enrichment_progress(index: int, total: int) -> None: if total: update_run(run_id, stage="website_analysis", progress=76 + int(index * 19 / total), found=index) results, added = insert_catalogue(run_id, query, raw_results, requested, enrichment_progress) 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} def scheduler_loop() -> None: while True: try: conn = db(); due = conn.execute("SELECT * FROM schedules WHERE enabled=1 AND next_run_at<=?", (now(),)).fetchall() for row in due: payload = {"category": row["category"], "keywords": row["keywords"], "location": row["location"], "max_results": row["max_results"], "extra_reviews": False} run = create_discovery(payload) conn.execute("UPDATE schedules SET next_run_at=?,last_run_id=? WHERE id=?", (datetime.now(timezone.utc).replace(microsecond=0).timestamp() and datetime.fromtimestamp(datetime.now(timezone.utc).timestamp()+int(row["interval_hours"])*3600,timezone.utc).isoformat(), run["id"], row["id"])) conn.commit(); conn.close() except Exception: try: conn.close() except Exception: pass time.sleep(30) def ai_generate_message(prospect: dict[str, Any], offer: str) -> dict[str, Any]: """Generate an evidence-grounded draft through an OpenAI-compatible endpoint.""" runtime = ai_runtime_settings() ai_base_url = runtime["base_url"] ai_api_key = runtime["api_key"] ai_model = runtime["model"] if runtime["provider"] == "nous" and not ai_api_key: raise RuntimeError("nous_api_key_required_for_standalone_app") if not ai_base_url or not ai_model: raise RuntimeError("ai_not_configured_set_AI_BASE_URL_and_AI_MODEL") evidence = { "business": prospect.get("name"), "category": prospect.get("category"), "location": prospect.get("address"), "website": prospect.get("website"), "website_status": prospect.get("website_status"), "website_title": prospect.get("website_title"), "technology": prospect.get("website_technology"), "emails": prospect.get("emails"), "phone": prospect.get("phone"), "score": prospect.get("opportunity_score"), "score_reasons": _json_list(prospect.get("score_reasons")), } system = ("You write concise, respectful B2B prospecting drafts for a South African web hosting and design business. " "Use only the supplied evidence; never invent facts, prices, names, services, or claims. " "Do not mention scraping or hidden analysis. Return strict JSON with keys subject, message, rationale. " "The message must be plain text, personalised, transparent, and include a polite opt-out line.") prompt = json.dumps({"evidence": evidence, "user_offer_or_instruction": offer[:2000]}, ensure_ascii=False) headers = {"Authorization": "Bearer " + ai_api_key} if ai_api_key else {} request_payload: dict[str, Any] = { "model": ai_model, "temperature": 0.4, "max_tokens": 700, "messages": [{"role": "system", "content": system}, {"role": "user", "content": prompt}], } # Nous exposes an OpenAI-compatible chat endpoint, but not every free model # accepts OpenAI's optional response_format parameter. The prompt still # requires strict JSON and the parser below validates the response. if runtime["provider"] != "nous": request_payload["response_format"] = {"type": "json_object"} result = http_json("POST", ai_base_url + "/chat/completions", request_payload, headers=headers, timeout=90) content = (((result.get("choices") or [{}])[0].get("message") or {}).get("content") or "") if isinstance(result, dict) else "" try: parsed = json.loads(content) except (TypeError, ValueError, json.JSONDecodeError) as exc: raise RuntimeError("ai_invalid_json_response") from exc if not isinstance(parsed, dict) or not str(parsed.get("subject") or "").strip() or not str(parsed.get("message") or "").strip(): raise RuntimeError("ai_incomplete_message") return {"subject": str(parsed["subject"])[:200], "message": str(parsed["message"])[:8000], "rationale": str(parsed.get("rationale") or "")[:2000]} 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))) ai_runtime = ai_runtime_settings() 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": True, "email_extraction_always": True, "ai_provider": ai_runtime["provider"], "ai_configured": bool(ai_runtime["base_url"] and ai_runtime["model"] and (ai_runtime["provider"] != "nous" or ai_runtime["api_key"])), }) if parsed.path == "/api/ai/models": provider = (parse_qs(parsed.query).get("provider", [AI_PROVIDER])[0] or AI_PROVIDER).strip().lower() if provider == "nous": return self.send_json(200, {"provider": "nous", "base_url": NOUS_BASE_URL, "models": nous_models()}) return self.send_json(200, {"provider": provider or "custom", "models": []}) if parsed.path == "/api/settings": ai_runtime = ai_runtime_settings() models = nous_models() if ai_runtime["provider"] == "nous" else [] return self.send_json(200, {"maps": {"backend": backend_mode(), "worker_url": GMAPS_URL, "ready": backend_mode() == "browser" or bool(GMAPS_URL)}, "ai": {"provider": ai_runtime["provider"], "base_url": ai_runtime["base_url"], "model": ai_runtime["model"], "configured": bool(ai_runtime["base_url"] and ai_runtime["model"] and (ai_runtime["provider"] != "nous" or ai_runtime["api_key"])), "models": models}, "email_collection": {"always_on": True, "public_only": True}, "limits": {"max_results": MAX_RESULTS, "website_scan_timeout_seconds": 8}}) 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/export.csv": params = parse_qs(parsed.query); q = params.get("q", [""])[0].strip().lower() rows = [dict(r) for r in conn.execute("SELECT * FROM prospects ORDER BY opportunity_score DESC, id DESC")] if q: rows = [r for r in rows if q in " ".join(str(r.get(k) or "") for k in ("name","domain","address","category","emails")).lower()] out = io.StringIO(); fields = ["id","name","category","address","phone","emails","domain","website","website_status","opportunity_score","score_reasons","review_status","last_seen"] writer = csv.DictWriter(out, fieldnames=fields); writer.writeheader(); writer.writerows({k:r.get(k) for k in fields} for r in rows) body = out.getvalue().encode(); self.send_response(200); self.send_header("Content-Type","text/csv; charset=utf-8"); self.send_header("Content-Disposition","attachment; filename=prospects.csv"); self.send_header("Content-Length",str(len(body))); self.end_headers(); self.wfile.write(body); return if parsed.path == "/api/suppressions": return self.send_json(200, {"items": [dict(r) for r in conn.execute("SELECT * FROM suppressions ORDER BY id DESC")]}) if parsed.path == "/api/schedules": return self.send_json(200, {"items": [dict(r) for r in conn.execute("SELECT * FROM schedules ORDER BY id DESC")]}) match = re.fullmatch(r"/api/prospects/(\d+)/domains", parsed.path) if match: row = conn.execute("SELECT * FROM prospects WHERE id=?", (int(match.group(1)),)).fetchone() if not row: return self.send_json(404, {"error":"not_found"}) return self.send_json(200, {"items": domain_candidates(row["name"], row["address"], row["category"])}) match = re.fullmatch(r"/api/prospects/(\d+)", parsed.path) if match: row = conn.execute("SELECT * FROM prospects 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/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/settings": try: payload = self.read_json(); ai = payload.get("ai") if isinstance(payload.get("ai"), dict) else {} conn = db() for key in ("AI_PROVIDER", "AI_BASE_URL", "AI_API_KEY", "AI_MODEL"): if key in ai: value = str(ai.get(key) or "").strip()[:500] conn.execute("INSERT INTO settings(key,value,updated_at) VALUES(?,?,?) ON CONFLICT(key) DO UPDATE SET value=excluded.value,updated_at=excluded.updated_at", (key, value, now())) conn.commit(); saved={str(r["key"]):str(r["value"]) for r in conn.execute("SELECT key,value FROM settings")}; conn.close() ai_runtime = ai_runtime_settings() return self.send_json(200, {"saved": True, "ai": {"provider": ai_runtime["provider"], "base_url": ai_runtime["base_url"], "model": ai_runtime["model"], "configured": bool(ai_runtime["base_url"] and ai_runtime["model"] and (ai_runtime["provider"] != "nous" or ai_runtime["api_key"]))}}) except Exception as exc: return self.send_json(400, {"error": str(exc)}) if urlparse(self.path).path == "/api/ai/test": try: runtime = ai_runtime_settings() if not runtime["base_url"] or not runtime["model"]: return self.send_json(400, {"error": "ai_not_configured_set_provider_and_model"}) if runtime["provider"] == "nous" and not runtime["api_key"]: return self.send_json(400, {"error": "nous_api_key_required_for_standalone_app"}) headers = {"Authorization": "Bearer " + runtime["api_key"]} if runtime["api_key"] else {} result = http_json("POST", runtime["base_url"] + "/chat/completions", {"model": runtime["model"], "temperature": 0, "max_tokens": 8, "messages": [{"role": "user", "content": "Reply with exactly OK."}]}, headers=headers, timeout=30) content = (((result.get("choices") or [{}])[0].get("message") or {}).get("content") or "") if isinstance(result, dict) else "" if not str(content).strip(): return self.send_json(502, {"error": "ai_empty_response"}) return self.send_json(200, {"ok": True, "provider": runtime["provider"], "model": runtime["model"], "reply": str(content).strip()[:200]}) except RuntimeError as exc: return self.send_json(502, {"error": str(exc)}) except Exception as exc: return self.send_json(502, {"error": "ai_test_failed:" + str(exc)[:200]}) message_match = re.fullmatch(r"/api/prospects/(\d+)/message", urlparse(self.path).path) if message_match: conn = None try: conn = db(); row = conn.execute("SELECT * FROM prospects WHERE id=?", (int(message_match.group(1)),)).fetchone() if not row: return self.send_json(404, {"error":"not_found"}) payload = self.read_json(); offer = str(payload.get("offer") or "").strip() if not offer: return self.send_json(400, {"error":"offer_required"}) draft = ai_generate_message(dict(row), offer) stamp = now(); conn.execute("UPDATE prospects SET ai_subject=?,ai_message=?,ai_offer=?,ai_reasoning=?,ai_checked_at=? WHERE id=?", (draft["subject"],draft["message"],offer,draft["rationale"],stamp,int(message_match.group(1)))); conn.commit() return self.send_json(200, {**draft,"ai_checked_at":stamp,"prospect_id":int(message_match.group(1))}) except RuntimeError as exc: return self.send_json(503, {"error":str(exc)}) except (ValueError, TypeError, json.JSONDecodeError) as exc: return self.send_json(400, {"error":str(exc)}) finally: if conn is not None: conn.close() scan_match = re.fullmatch(r"/api/prospects/(\d+)/scan", urlparse(self.path).path) if scan_match: conn = None try: conn=db(); row=conn.execute("SELECT * FROM prospects WHERE id=?",(int(scan_match.group(1)),)).fetchone() if not row: return self.send_json(404,{"error":"not_found"}) updated=enrich_place(dict(row)); cols=["website_status","website_http_status","website_final_url","website_latency_ms","website_https","website_title","website_technology","website_has_viewport","website_tls_expires_at","website_checked_at","website_error","emails","email_evidence","score_reasons","opportunity_score"] conn.execute("UPDATE prospects SET "+",".join(c+"=?" for c in cols)+" WHERE id=?",tuple(updated.get(c) for c in cols)+(int(scan_match.group(1)),)); conn.commit(); return self.send_json(200,dict(conn.execute("SELECT * FROM prospects WHERE id=?",(int(scan_match.group(1)),)).fetchone())) except Exception as exc: return self.send_json(500,{"error":str(exc)}) finally: if conn is not None: conn.close() if self.path == "/api/suppressions": try: p=self.read_json(); kind=str(p.get("kind","domain")).lower(); value=str(p.get("value","")).strip().lower() if kind not in {"domain","email"} or not value: return self.send_json(400,{"error":"kind_and_value_required"}) conn=db(); conn.execute("INSERT OR IGNORE INTO suppressions(kind,value,reason,created_at) VALUES(?,?,?,?)",(kind,value,str(p.get("reason","manual"))[:200],now())); conn.commit(); row=conn.execute("SELECT * FROM suppressions WHERE kind=? AND value=?",(kind,value)).fetchone(); conn.close(); return self.send_json(201,dict(row)) except Exception as exc: return self.send_json(400,{"error":str(exc)}) if self.path == "/api/schedules": try: p=self.read_json(); category=str(p.get("category") or "").strip(); keywords=str(p.get("keywords") or "").strip(); location=str(p.get("location") or "South Africa").strip(); interval=max(1,min(168,int(p.get("interval_hours",24)))); max_results=max(1,min(MAX_RESULTS,int(p.get("max_results",25)))) if not category or not keywords: return self.send_json(400,{"error":"category_and_keywords_required"}) next_at=datetime.fromtimestamp(datetime.now(timezone.utc).timestamp()+interval*3600,timezone.utc).isoformat(); conn=db(); cur=conn.execute("INSERT INTO schedules(category,keywords,location,max_results,interval_hours,enabled,next_run_at,created_at) VALUES(?,?,?,?,?,?,?,?)",(category,keywords,location,max_results,interval,1,next_at,now())); conn.commit(); row=conn.execute("SELECT * FROM schedules WHERE id=?",(cur.lastrowid,)).fetchone(); conn.close(); return self.send_json(201,dict(row)) except Exception as exc: return self.send_json(400,{"error":str(exc)}) 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 do_DELETE(self) -> None: match = re.fullmatch(r"/api/suppressions/(\d+)", urlparse(self.path).path) if not match: return self.send_json(404, {"error":"not_found"}) conn=db() try: cur=conn.execute("DELETE FROM suppressions WHERE id=?",(int(match.group(1)),)); conn.commit() return self.send_json(200,{"deleted":bool(cur.rowcount)}) finally: conn.close() def serve() -> None: host = os.environ.get("HOST", "0.0.0.0") port = int(os.environ.get("PORT", "8000")) db() recover_stale_runs() threading.Thread(target=scheduler_loop, name="prospect-scheduler", daemon=True).start() print(f"Personal ProspectOS listening on http://{host}:{port} (backend={backend_mode()})", flush=True) ThreadingHTTPServer((host, port), Handler).serve_forever() if __name__ == "__main__": serve()