1164 lines
58 KiB
Python
1164 lines
58 KiB
Python
#!/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
|
|
from threading import Thread
|
|
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"))
|
|
GMAPS_REQUEST_TIMEOUT = int(os.environ.get("GMAPS_REQUEST_TIMEOUT", "300"))
|
|
GMAPS_URL = os.environ.get("GMAPS_URL", "").strip().rstrip("/")
|
|
# The bundled gosom/google-maps-scraper image exposes a local, unauthenticated
|
|
# REST API. The optional SaaS edition uses a different API and requires an
|
|
# X-API-Key header; keep support for it only when explicitly selected.
|
|
GMAPS_API_MODE = os.environ.get("GMAPS_API_MODE", "standard").strip().lower()
|
|
GMAPS_API_KEY = os.environ.get("GMAPS_API_KEY", "").strip()
|
|
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()
|
|
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 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))
|
|
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):
|
|
item = enrich_place(item)
|
|
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
|
|
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"],
|
|
))
|
|
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 = {}
|
|
try:
|
|
c = db(); runtime = {str(r["key"]): str(r["value"]) for r in c.execute("SELECT key,value FROM settings")}; c.close()
|
|
except Exception:
|
|
pass
|
|
ai_base_url = runtime.get("AI_BASE_URL", AI_BASE_URL).strip().rstrip("/")
|
|
ai_api_key = runtime.get("AI_API_KEY", AI_API_KEY).strip()
|
|
ai_model = runtime.get("AI_MODEL", AI_MODEL).strip()
|
|
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 {}
|
|
result = http_json("POST", ai_base_url + "/chat/completions", {"model": ai_model, "temperature": 0.4, "response_format": {"type": "json_object"}, "messages": [{"role": "system", "content": system}, {"role": "user", "content": prompt}]}, headers=headers, timeout=60)
|
|
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)))
|
|
saved = {str(r["key"]): str(r["value"]) for r in conn.execute("SELECT key,value FROM 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_configured": bool(saved.get("AI_BASE_URL", AI_BASE_URL) and saved.get("AI_MODEL", AI_MODEL)),
|
|
})
|
|
if parsed.path == "/api/settings":
|
|
saved = {str(r["key"]): str(r["value"]) for r in conn.execute("SELECT key,value FROM settings")}
|
|
return self.send_json(200, {"maps": {"backend": backend_mode(), "worker_url": GMAPS_URL, "ready": backend_mode() == "browser" or bool(GMAPS_URL)}, "ai": {"base_url": saved.get("AI_BASE_URL", AI_BASE_URL), "model": saved.get("AI_MODEL", AI_MODEL), "configured": bool(saved.get("AI_BASE_URL", AI_BASE_URL) and saved.get("AI_MODEL", AI_MODEL))}, "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_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()
|
|
return self.send_json(200, {"saved": True, "ai": {"base_url": saved.get("AI_BASE_URL", AI_BASE_URL), "model": saved.get("AI_MODEL", AI_MODEL), "configured": bool(saved.get("AI_BASE_URL", AI_BASE_URL) and saved.get("AI_MODEL", AI_MODEL))}})
|
|
except Exception as exc:
|
|
return self.send_json(400, {"error": str(exc)})
|
|
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()
|
|
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()
|