816 lines
36 KiB
Python
816 lines
36 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
|
|
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
|
|
|
|
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()
|
|
ALLOWED_REVIEW = {"pending", "approved", "rejected", "contacted"}
|
|
_discover_lock = threading.Lock()
|
|
|
|
|
|
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:
|
|
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")
|
|
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',
|
|
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 '{}'
|
|
)""")
|
|
|
|
# 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'",
|
|
"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)")
|
|
# A Maps scrape can legitimately take several minutes. Only recover jobs
|
|
# that have been abandoned for half an hour, never after 35 seconds.
|
|
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()
|
|
return conn
|
|
|
|
|
|
def json_bytes(value: Any) -> bytes:
|
|
return json.dumps(value, ensure_ascii=False).encode("utf-8")
|
|
|
|
|
|
def clean_domain(url: str) -> str:
|
|
parsed = urlparse(url if "://" in url else "https://" + url)
|
|
host = (parsed.hostname or "").lower().strip().rstrip(".")
|
|
if host.startswith("www."):
|
|
host = host[4:]
|
|
return host
|
|
|
|
|
|
def 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
|
|
|
|
|
|
def score_place(item: dict[str, Any], website: str, phone: str, rating: float | None, review_count: int) -> int:
|
|
"""Small transparent triage score; it is not an AI verdict."""
|
|
score = 20
|
|
if not website:
|
|
score += 35
|
|
if phone:
|
|
score += 10
|
|
if rating is not None and rating >= 4:
|
|
score += 10
|
|
if review_count >= 10:
|
|
score += 10
|
|
if as_text(item.get("category") or item.get("type")):
|
|
score += 5
|
|
return min(100, score)
|
|
|
|
|
|
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)
|
|
emails = as_text(item.get("emails") or item.get("email"))
|
|
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": "present" if website else "missing",
|
|
"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")
|
|
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()
|
|
include_email = bool(payload.get("include_email", False))
|
|
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 insert_catalogue(run_id: int, query: str, raw_results: list[dict[str, Any]], requested: int) -> tuple[list[dict[str, Any]], int]:
|
|
mapped = [map_gmaps_result(item, query) for item in raw_results[:requested]]
|
|
conn = db()
|
|
added = 0
|
|
try:
|
|
for item in mapped:
|
|
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,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["listing_status"], item["website_status"], 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)
|
|
results, added = insert_catalogue(run_id, query, raw_results, requested)
|
|
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}
|
|
|
|
|
|
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)))
|
|
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": False,
|
|
})
|
|
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/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/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 serve() -> None:
|
|
host = os.environ.get("HOST", "0.0.0.0")
|
|
port = int(os.environ.get("PORT", "8000"))
|
|
db()
|
|
print(f"Personal ProspectOS listening on http://{host}:{port} (backend={backend_mode()})", flush=True)
|
|
ThreadingHTTPServer((host, port), Handler).serve_forever()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
serve()
|