From 5e14ddf92cd2c3dfae8304620d47c8e13539b849 Mon Sep 17 00:00:00 2001 From: Marco0300 Date: Sat, 5 Sep 2026 17:06:17 +0200 Subject: [PATCH] deploy Maps v6 update --- .env.example | 6 + README.md | 30 ++-- api/Dockerfile | 2 +- api/intelligence.py | 263 ++++++++++++++++++++++++++++++++ api/server.py | 362 +++++++++++++++++++++++++++++++++++++++++--- docker-compose.yml | 3 + tests/test_app.py | 10 +- web/app.js | 31 +++- web/index.html | 6 +- web/style.css | 1 + 10 files changed, 673 insertions(+), 41 deletions(-) create mode 100644 api/intelligence.py diff --git a/.env.example b/.env.example index cf0bf96..4b8e207 100644 --- a/.env.example +++ b/.env.example @@ -11,3 +11,9 @@ GMAPS_API_KEY= # A single personal run may take several minutes. GMAPS_REQUEST_TIMEOUT=300 GMAPS_POLL_SECONDS=2 + +# Optional OpenAI-compatible provider for AI analysis and tailored drafts. +# Keep these empty to use the rest of the app without AI. +AI_BASE_URL= +AI_API_KEY= +AI_MODEL= diff --git a/README.md b/README.md index 6926874..18eb815 100644 --- a/README.md +++ b/README.md @@ -5,11 +5,11 @@ the bundled [gosom/google-maps-scraper](https://github.com/gosom/google-maps-scr worker, waits for its asynchronous job to finish, filters places already in the local catalogue, and presents the remaining places for human review. -The interface is intentionally focused: discovery, run progress, catalogue -search, and review status. It does not send outreach, create a CRM campaign, or -make an AI decision. The opportunity score is a transparent triage hint based -on fields returned by Maps (for example, a missing website and a listed phone -number); you make the final review decision. +The interface includes discovery, run progress, catalogue search, website +health and email evidence, explainable scoring, domain ideas, suppression, +CSV export, scheduled runs, rescans, and human review. It never sends outreach +automatically. Optional AI can analyse a prospect and draft a subject/email +from supplied evidence and your offer; you edit and approve the draft yourself. ## Start the stack @@ -47,6 +47,9 @@ error rather than remaining stuck forever. | `GMAPS_REQUEST_TIMEOUT` | `300` | Per-job scraper timeout in seconds | | `GMAPS_POLL_SECONDS` | `2` | Poll interval while a job is running | | `DISCOVERY_BACKEND` | `gmaps` when `GMAPS_URL` is set | Set to `browser` only for the bounded public-search fallback | +| `AI_BASE_URL` | empty | Optional OpenAI-compatible `/chat/completions` endpoint | +| `AI_API_KEY` | empty | Optional provider key; keep server-side | +| `AI_MODEL` | empty | Model name used for evidence-grounded drafts | The app stores its catalogue in the `prospect_data` Docker volume. The worker stores its own job data in `gmaps_data`; keep both volumes if you want to retain @@ -73,14 +76,23 @@ and review UI, and does not bypass CAPTCHAs or other access controls. Each saved place includes its name, Maps link, website (when present), root domain, address, phone, category, rating, review count, public emails returned -by the worker when explicitly enabled, source ID, and raw result JSON. A place +by the worker when available, source ID, and raw result JSON. A place with no website is still retained using a stable `maps:` catalogue key, so it can be reviewed as a potential website opportunity. Existing websites are deduplicated by root domain. -Email extraction is off by default in the UI. If you enable it for a run, the -worker may visit the public business website; expect a longer run and review -the resulting provenance before using any contact information. +Email extraction is always enabled. The worker may visit each public business +website found by Maps; expect longer runs and review the resulting provenance +before using any contact information. It does not invent email addresses when +none are published. + +Website scans are bounded and SSRF-safe. A scan records HTTP status, final URL, +latency, HTTPS/TLS, title, technology hints, contact-page emails, and errors. +Use `/api/prospects/{id}/scan` for a manual rescan. The explorer's “Draft AI +email” action calls `/api/prospects/{id}/message`; it requires the optional AI +variables above and accepts your exact offer/instructions (for example, the +Gold Plus R119/month plan and a free rebuild). Suppressions are enforced before +new records are saved and can be managed through `/api/suppressions`. ## Tests diff --git a/api/Dockerfile b/api/Dockerfile index 899dc80..05cd2d6 100644 --- a/api/Dockerfile +++ b/api/Dockerfile @@ -4,7 +4,7 @@ RUN apt-get update && apt-get install -y --no-install-recommends chromium && rm WORKDIR /app COPY api/requirements.txt . RUN pip install --no-cache-dir -r requirements.txt -COPY api/server.py . +COPY api/server.py api/intelligence.py ./ COPY web /app/web RUN useradd --system --create-home app && mkdir -p /data && chown -R app:app /app /data USER app diff --git a/api/intelligence.py b/api/intelligence.py new file mode 100644 index 0000000..b0fd526 --- /dev/null +++ b/api/intelligence.py @@ -0,0 +1,263 @@ +"""Safe, bounded website and domain intelligence for ProspectOS. + +This module deliberately stays dependency-free. It performs small public HTTP +requests, refuses private/loopback destinations, follows only same-domain +redirects, and never treats a failed check as proof that a site does not exist. +""" +from __future__ import annotations + +import html +import ipaddress +import re +import socket +import ssl +import time +from datetime import datetime, timezone +from html.parser import HTMLParser +from typing import Any +from urllib.error import HTTPError, URLError +from urllib.parse import urljoin, urlparse +from urllib.request import HTTPRedirectHandler, Request, build_opener + +EMAIL_RE = re.compile(r"[A-Z0-9._%+\-]+@[A-Z0-9.\-]+\.[A-Z]{2,63}", re.I) +MAX_BODY_BYTES = 700_000 +MAX_REDIRECTS = 2 +MAX_PAGES = 2 +TIMEOUT_SECONDS = 8 + + +class _PageParser(HTMLParser): + def __init__(self) -> None: + super().__init__(convert_charrefs=True) + self.title_parts: list[str] = [] + self.meta: dict[str, str] = {} + self.links: list[tuple[str, str]] = [] + self._in_title = False + + def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: + values = {key.lower(): value or "" for key, value in attrs} + if tag.lower() == "title": + self._in_title = True + if tag.lower() == "meta": + key = (values.get("name") or values.get("property") or "").lower() + if key: + self.meta[key] = values.get("content", "")[:300] + if tag.lower() == "a" and values.get("href"): + self.links.append((values.get("href", "")[:500], "")) + + def handle_endtag(self, tag: str) -> None: + if tag.lower() == "title": + self._in_title = False + + def handle_data(self, data: str) -> None: + if self._in_title: + self.title_parts.append(data) + + +class _NoRedirect(HTTPRedirectHandler): + def redirect_request(self, req: Request, fp: Any, code: int, msg: str, headers: Any, newurl: str) -> None: # type: ignore[override] + return None + + +_opener = build_opener(_NoRedirect) + + +def _host_is_public(host: str) -> tuple[bool, list[str]]: + """Resolve a hostname and reject local, reserved, or unroutable targets.""" + try: + addresses = {item[4][0] for item in socket.getaddrinfo(host, None, type=socket.SOCK_STREAM)} + except (OSError, socket.gaierror): + return False, [] + if not addresses: + return False, [] + for raw in addresses: + try: + ip = ipaddress.ip_address(raw) + except ValueError: + return False, list(addresses) + if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_multicast or ip.is_reserved or ip.is_unspecified: + return False, list(addresses) + return True, list(addresses) + + +def _safe_url(value: str) -> bool: + parsed = urlparse(value) + if parsed.scheme not in {"http", "https"} or not parsed.hostname: + return False + if parsed.username or parsed.password: + return False + allowed, _ = _host_is_public(parsed.hostname) + return allowed + + +def _same_root(left: str, right: str) -> bool: + a = (urlparse(left).hostname or "").lower().rstrip(".") + b = (urlparse(right).hostname or "").lower().rstrip(".") + if a.startswith("www."): + a = a[4:] + if b.startswith("www."): + b = b[4:] + return bool(a and b and (a == b or a.endswith("." + b) or b.endswith("." + a))) + + +def _fetch(url: str) -> dict[str, Any]: + current = url + redirects = 0 + started = time.monotonic() + last_status = 0 + last_headers: Any = {} + body = b"" + while redirects <= MAX_REDIRECTS: + if not _safe_url(current): + return {"error": "unsafe_or_unresolvable_host", "url": current, "latency_ms": int((time.monotonic() - started) * 1000)} + request = Request(current, headers={"User-Agent": "ProspectOS/1.0 (+public-site-check)", "Accept": "text/html,application/xhtml+xml"}) + try: + with _opener.open(request, timeout=TIMEOUT_SECONDS) as response: + last_status = int(getattr(response, "status", 200) or 200) + last_headers = response.headers + body = response.read(MAX_BODY_BYTES) + final_url = response.geturl() or current + return { + "url": final_url, + "initial_url": url, + "status": last_status, + "headers": last_headers, + "body": body, + "latency_ms": int((time.monotonic() - started) * 1000), + "redirects": redirects, + } + except HTTPError as exc: + last_status = int(exc.code or 0) + last_headers = exc.headers + if 300 <= last_status < 400 and exc.headers.get("Location"): + target = urljoin(current, exc.headers.get("Location", "")) + if not _same_root(current, target) or not _safe_url(target): + return {"url": current, "initial_url": url, "status": last_status, "headers": last_headers, "body": b"", "latency_ms": int((time.monotonic() - started) * 1000), "redirects": redirects, "error": "cross_domain_redirect_blocked"} + current = target + redirects += 1 + continue + return {"url": current, "initial_url": url, "status": last_status, "headers": last_headers, "body": b"", "latency_ms": int((time.monotonic() - started) * 1000), "redirects": redirects, "error": f"http_{last_status}"} + except (URLError, TimeoutError, OSError) as exc: + return {"url": current, "initial_url": url, "status": last_status, "headers": last_headers, "body": b"", "latency_ms": int((time.monotonic() - started) * 1000), "redirects": redirects, "error": str(getattr(exc, "reason", exc))[:180]} + return {"url": current, "initial_url": url, "status": last_status, "headers": last_headers, "body": body, "latency_ms": int((time.monotonic() - started) * 1000), "redirects": redirects, "error": "redirect_limit"} + + +def _tls_expiry(url: str) -> str: + parsed = urlparse(url) + if parsed.scheme != "https" or not parsed.hostname: + return "" + try: + context = ssl.create_default_context() + with socket.create_connection((parsed.hostname, parsed.port or 443), timeout=TIMEOUT_SECONDS) as raw: + with context.wrap_socket(raw, server_hostname=parsed.hostname) as secure: + cert = secure.getpeercert() + value = cert.get("notAfter") if cert else "" + if not value: + return "" + stamp = ssl.cert_time_to_seconds(value) + return datetime.fromtimestamp(stamp, timezone.utc).replace(microsecond=0).isoformat() + except (OSError, ssl.SSLError, ValueError): + return "" + + +def _emails(text: str) -> list[str]: + found: list[str] = [] + for value in EMAIL_RE.findall(html.unescape(text or "")): + normalized = value.strip(".,;:()[]{}<>").lower() + if normalized not in found and not normalized.endswith((".png", ".jpg", ".jpeg", ".gif", ".svg")): + found.append(normalized) + return found[:30] + + +def scan_website(url: str) -> dict[str, Any]: + """Return bounded website health, metadata, and public email evidence.""" + if not url: + return {"website_status": "no_detected_website", "website_checked_at": datetime.now(timezone.utc).replace(microsecond=0).isoformat(), "website_emails": [], "email_evidence": []} + first = _fetch(url) + status = int(first.get("status") or 0) + final_url = str(first.get("url") or url) + body = bytes(first.get("body") or b"") + decoded = body.decode("utf-8", "replace") + parser = _PageParser() + try: + parser.feed(decoded) + except Exception: + pass + title = " ".join(" ".join(parser.title_parts).split())[:300] + emails = _emails(decoded) + pages: list[dict[str, Any]] = [first] + if 200 <= status < 400 and len(pages) < MAX_PAGES: + for href, _ in parser.links: + candidate = urljoin(final_url, href) + label = (href or "").lower() + if any(token in label for token in ("contact", "about", "get-in-touch")) and _same_root(final_url, candidate) and _safe_url(candidate): + page = _fetch(candidate) + pages.append(page) + emails.extend(_emails(bytes(page.get("body") or b"").decode("utf-8", "replace"))) + break + emails = list(dict.fromkeys(emails))[:30] + if first.get("error") and not status: + website_status = "unreachable" + elif status >= 500: + website_status = "broken" + elif status >= 400: + website_status = "error" + elif status >= 300 or int(first.get("redirects") or 0) > 0: + website_status = "redirect" + elif int(first.get("latency_ms") or 0) >= 3000: + website_status = "slow" + elif urlparse(final_url).scheme != "https": + website_status = "http_only" + else: + website_status = "healthy" + headers = first.get("headers") or {} + server = str(headers.get("Server", ""))[:120] + generator = parser.meta.get("generator", "")[:120] + technologies = ", ".join(dict.fromkeys(value for value in (server, generator) if value))[:250] + checked = datetime.now(timezone.utc).replace(microsecond=0).isoformat() + evidence = [{"email": value, "source_url": final_url, "found_at": checked, "method": "website_scan", "confidence": "high"} for value in emails] + return { + "website_status": website_status, + "website_http_status": status or None, + "website_final_url": final_url[:1000], + "website_latency_ms": int(first.get("latency_ms") or 0), + "website_https": 1 if urlparse(final_url).scheme == "https" else 0, + "website_title": title, + "website_technology": technologies, + "website_has_viewport": 1 if "viewport" in parser.meta else 0, + "website_tls_expires_at": _tls_expiry(final_url) if urlparse(final_url).scheme == "https" else "", + "website_checked_at": checked, + "website_error": str(first.get("error") or "")[:300], + "website_emails": emails, + "email_evidence": evidence, + "scanned_pages": len(pages), + } + + +def _slug(value: str) -> str: + value = re.sub(r"[^a-z0-9]+", "-", (value or "").lower()).strip("-") + return value[:45] + + +def domain_candidates(name: str, address: str = "", category: str = "") -> list[dict[str, str]]: + """Suggest likely .co.za names; DNS state is evidence, not registrar availability.""" + base = _slug(name) + city = _slug((address or "").split(",")[-1]) + cat = _slug(category) + parts = [base, f"{base}-{cat}" if cat else "", f"{base}-{city}" if city else ""] + seen: set[str] = set() + output: list[dict[str, str]] = [] + for stem in parts: + if not stem: + continue + domain = stem + ".co.za" + if domain in seen: + continue + seen.add(domain) + try: + socket.getaddrinfo(domain, None, type=socket.SOCK_STREAM) + dns_status = "resolves" + except OSError: + dns_status = "no_dns_response" + output.append({"domain": domain, "dns_status": dns_status, "availability_note": "DNS is only a hint; confirm availability with a registrar."}) + return output diff --git a/api/server.py b/api/server.py index add754b..9d86b33 100644 --- a/api/server.py +++ b/api/server.py @@ -25,6 +25,11 @@ 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")) @@ -35,8 +40,13 @@ GMAPS_URL = os.environ.get("GMAPS_URL", "").strip().rstrip("/") # 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: @@ -51,10 +61,17 @@ def backend_mode() -> str: 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, @@ -78,6 +95,25 @@ def db() -> sqlite3.Connection: 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 '{}' )""") @@ -98,6 +134,20 @@ def db() -> sqlite3.Connection: 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 + )""") # Migrate databases created by v1-v5 without destroying the user's # catalogue. SQLite permits adding nullable/defaulted columns in place. @@ -115,6 +165,25 @@ def db() -> sqlite3.Connection: "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 '{}'", } @@ -137,17 +206,26 @@ def db() -> sqlite3.Connection: 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() + _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") @@ -213,20 +291,69 @@ def as_int(value: Any) -> int: 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.""" +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 - if not website: + 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) + 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]: @@ -238,7 +365,8 @@ def map_gmaps_result(item: dict[str, Any], query: str) -> dict[str, Any]: 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")) + 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) @@ -265,7 +393,9 @@ def map_gmaps_result(item: dict[str, Any], query: str) -> dict[str, Any]: "latitude": latitude, "longitude": longitude, "listing_status": as_text(item.get("status"))[:120], - "website_status": "present" if website else "missing", + "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], } @@ -369,6 +499,7 @@ def download_standard_results(job_id: str) -> list[dict[str, Any]]: 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": @@ -575,7 +706,10 @@ def validate_payload(payload: dict[str, Any]) -> tuple[str, str, str, int, bool, 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)) + # 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 @@ -596,21 +730,85 @@ def update_run(run_id: int, **fields: Any) -> None: conn.close() -def insert_catalogue(run_id: int, query: str, raw_results: list[dict[str, Any]], requested: int) -> tuple[list[dict[str, Any]], int]: +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: - for item in mapped: + 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,opportunity_score,raw_json - ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", ( + 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["listing_status"], item["website_status"], item["opportunity_score"], item["raw_json"], + 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 @@ -639,7 +837,13 @@ def run_discovery(run_id: int, payload: dict[str, Any]) -> None: 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, 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" @@ -669,6 +873,50 @@ def create_discovery(payload: dict[str, Any]) -> dict[str, Any]: 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.""" + 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" @@ -710,7 +958,9 @@ class Handler(BaseHTTPRequestHandler): "gmaps_auth_required": auth_required, "gmaps_api_key": bool(GMAPS_API_KEY), "ready": ready, - "email_extraction_default": False, + "email_extraction_default": True, + "email_extraction_always": True, + "ai_configured": bool(AI_BASE_URL and AI_MODEL), }) if parsed.path == "/api/prospects": params = parse_qs(parsed.query) @@ -726,6 +976,27 @@ class Handler(BaseHTTPRequestHandler): 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}) @@ -769,6 +1040,44 @@ class Handler(BaseHTTPRequestHandler): self.wfile.write(content) def do_POST(self) -> None: + 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: @@ -802,11 +1111,22 @@ class Handler(BaseHTTPRequestHandler): 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() diff --git a/docker-compose.yml b/docker-compose.yml index fffcb6c..6ce48fd 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -24,6 +24,9 @@ services: GMAPS_API_KEY: ${GMAPS_API_KEY:-} GMAPS_REQUEST_TIMEOUT: ${GMAPS_REQUEST_TIMEOUT:-300} GMAPS_POLL_SECONDS: ${GMAPS_POLL_SECONDS:-2} + AI_BASE_URL: ${AI_BASE_URL:-} + AI_API_KEY: ${AI_API_KEY:-} + AI_MODEL: ${AI_MODEL:-} volumes: - prospect_data:/data depends_on: diff --git a/tests/test_app.py b/tests/test_app.py index 6dbd7b8..4ed86bd 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -15,12 +15,15 @@ class BrowserOnlyAppTests(unittest.TestCase): self.tmp = tempfile.NamedTemporaryFile(delete=False) self.tmp.close() server.DB_PATH = self.tmp.name + self.scan_patcher = patch("api.server.scan_website", return_value={"website_status": "healthy", "website_http_status": 200, "website_final_url": "", "website_latency_ms": 50, "website_https": 1, "website_title": "", "website_technology": "", "website_has_viewport": 1, "website_tls_expires_at": "", "website_checked_at": server.now(), "website_error": "", "website_emails": [], "email_evidence": []}) + self.scan_patcher.start() self.http = server.ThreadingHTTPServer(("127.0.0.1", 0), server.Handler) self.thread = threading.Thread(target=self.http.serve_forever, daemon=True) self.thread.start() self.conn = HTTPConnection("127.0.0.1", self.http.server_port) def tearDown(self): + self.scan_patcher.stop() self.http.shutdown(); self.http.server_close(); self.thread.join(1) try: os.unlink(self.tmp.name) except FileNotFoundError: pass @@ -32,12 +35,12 @@ class BrowserOnlyAppTests(unittest.TestCase): return response.status, json.loads(response.read() or b"{}") def wait_for_run(self, run_id): - for _ in range(50): + for _ in range(300): status, body = self.request("GET", f"/api/runs/{run_id}") self.assertEqual(status, 200) if body["status"] not in {"queued", "running"}: return body - time.sleep(0.01) + time.sleep(0.02) self.fail("run did not finish") def test_search_redirect_links_are_unwrapped(self): @@ -62,7 +65,7 @@ class BrowserOnlyAppTests(unittest.TestCase): def test_map_without_website_gets_stable_catalogue_key(self): mapped = server.map_gmaps_result({"title": "A Local Plumber", "place_id": "abc123", "address": "Cape Town", "phone": "+27 10", "review_rating": 4.5, "review_count": 12}, "plumbers Cape Town") self.assertEqual(mapped["domain"], "maps:abc123") - self.assertEqual(mapped["website_status"], "missing") + self.assertEqual(mapped["website_status"], "no_detected_website") self.assertGreaterEqual(mapped["opportunity_score"], 70) @patch.dict(os.environ, {"DISCOVERY_BACKEND": "browser"}) @@ -87,6 +90,7 @@ class BrowserOnlyAppTests(unittest.TestCase): first_call = http_json.call_args_list[0] self.assertEqual(first_call.args[0:2], ("POST", "http://gmaps:8080/api/v1/jobs")) self.assertNotIn("X-API-Key", first_call.args[3]) + self.assertTrue(first_call.args[2]["email"]) self.assertEqual(http_bytes.call_args.args[1], "http://gmaps:8080/api/v1/jobs/job-1/download") diff --git a/web/app.js b/web/app.js index 59bb62d..f32d0e8 100644 --- a/web/app.js +++ b/web/app.js @@ -2,6 +2,7 @@ const $ = (id) => document.getElementById(id); let prospects = []; let runs = []; let activePoll = null; +let selectedProspect = null; async function api(path, options = {}) { const response = await fetch(path, { @@ -51,14 +52,14 @@ function renderExplorer() { $("resultCount").textContent = `${plural(rows.length, "place")} ${query || status ? "matching" : "in catalogue"}`; $("explorerList").innerHTML = rows.length ? rows.map((item) => { const score = Number(item.opportunity_score || 0); - const website = item.website ? `${esc(item.domain || item.website)} ↗` : `No website found`; + const website = item.website ? `${esc(item.domain || item.website)} ↗` : `No detected website`; const maps = item.maps_url ? `Open Maps ↗` : ""; const contact = [item.phone && `☎ ${esc(item.phone)}`, item.emails && `✉ ${esc(item.emails)}`].filter(Boolean).join(""); return `
⌖

${esc(item.name)}

${esc(item.category || "Business")} ${item.address ? ` · ${esc(item.address)}` : ""}
-
${score}/100 opportunity${item.website_status === "missing" ? 'Website missing' : ""}${item.rating != null ? `★ ${esc(Number(item.rating).toFixed(1))} (${esc(item.review_count || 0)})` : ""}
+
${score}/100 opportunity${esc((item.website_status || "unknown").replaceAll("_", " "))}${item.rating != null ? `★ ${esc(Number(item.rating).toFixed(1))} (${esc(item.review_count || 0)})` : ""}
${website}${contact ? `${contact}` : ""}
-
Found ${esc(formatDate(item.discovered_at))}·${esc(item.source || "google_maps")}${maps ? `${maps}` : ""}
+
Found ${esc(formatDate(item.discovered_at))}·${esc(item.source || "google_maps")}${maps ? `${maps}` : ""}
`; @@ -95,6 +96,7 @@ function renderRuns() { async function loadProspects() { prospects = (await api("/api/prospects")).items || []; renderExplorer(); } async function loadRuns() { runs = (await api("/api/runs")).items || []; renderRuns(); } async function loadStats() { renderStats(await api("/api/stats")); } +async function loadSchedules() { const data=await api("/api/schedules"); $("scheduleList").innerHTML=(data.items||[]).map(s=>`
${esc(s.keywords)} · ${esc(s.category)} · ${esc(s.location)}every ${s.interval_hours}h · next ${esc(formatDate(s.next_run_at))}
`).join("") || "No schedules configured."; } async function loadConfig() { const config = await api("/api/config"); @@ -110,7 +112,7 @@ async function loadConfig() { async function refreshAll() { $("updated").textContent = "Refreshing…"; - try { await Promise.all([loadConfig(), loadProspects(), loadRuns(), loadStats()]); $("updated").textContent = `Updated ${new Date().toLocaleTimeString([], { hour: "2-digit", minute: "2-digit" })}`; } + try { await Promise.all([loadConfig(), loadProspects(), loadRuns(), loadStats(), loadSchedules()]); $("updated").textContent = `Updated ${new Date().toLocaleTimeString([], { hour: "2-digit", minute: "2-digit" })}`; } catch (error) { $("updated").textContent = "Refresh failed"; $("discoverMessage").textContent = error.message; $("discoverMessage").className = "form-message error"; } } @@ -131,7 +133,7 @@ $("discoverForm").addEventListener("submit", async (event) => { const button = $("discoverBtn"); const message = $("discoverMessage"); button.disabled = true; message.className = "form-message"; message.textContent = "Queueing Maps search…"; - const payload = { category: $("category").value.trim(), keywords: $("keywords").value.trim(), location: $("location").value.trim(), max_results: Number($("max-results").value), include_email: $("include-email").checked, extra_reviews: $("extra-reviews").checked }; + const payload = { category: $("category").value.trim(), keywords: $("keywords").value.trim(), location: $("location").value.trim(), max_results: Number($("max-results").value), include_email: true, extra_reviews: $("extra-reviews").checked }; try { const run = await api("/api/discover", { method: "POST", body: JSON.stringify(payload) }); runs.unshift(run); renderRuns(); message.textContent = "Search queued · this can take a few minutes."; await pollRun(run.id); } catch (error) { button.disabled = false; message.textContent = `Could not start run · ${error.message}`; message.className = "form-message error"; } }); @@ -142,12 +144,31 @@ $("explorerList").addEventListener("change", async (event) => { try { const updated = await api(`/api/prospects/${id}/review`, { method: "PATCH", body: JSON.stringify({ status: event.target.value }) }); const index = prospects.findIndex((item) => item.id === updated.id); if (index >= 0) prospects[index] = updated; renderExplorer(); await loadStats(); } catch (error) { event.target.disabled = false; alert(error.message); } }); +$("explorerList").addEventListener("click", (event) => { + const id = event.target.dataset.message; if (!id) return; + selectedProspect = prospects.find((item) => String(item.id) === String(id)); + $("messageTitle").textContent = `Draft for ${selectedProspect?.name || "prospect"}`; + $("offerInput").value = ""; $("subjectOutput").value = selectedProspect?.ai_subject || ""; $("bodyOutput").value = selectedProspect?.ai_message || ""; $("messageStatus").textContent = ""; $("messageDialog").showModal(); +}); +$("explorerList").addEventListener("click", async (event) => { + const d=event.target.dataset.domains, r=event.target.dataset.rescan; + if (d) { try { const x=await api(`/api/prospects/${d}/domains`); alert((x.items||[]).map(i=>`${i.domain} · ${i.status}`).join("\n") || "No domain ideas found."); } catch(e){ alert(e.message); } } + if (r) { event.target.disabled=true; try { const updated=await api(`/api/prospects/${r}/scan`,{method:"POST",body:"{}"}); const i=prospects.findIndex(p=>p.id===updated.id); if(i>=0) prospects[i]=updated; renderExplorer(); } catch(e){ alert(e.message); } } +}); +$("generateMessage").addEventListener("click", async () => { + if (!selectedProspect) return; const status = $("messageStatus"); status.textContent = "Generating evidence-grounded draft…"; + try { const draft = await api(`/api/prospects/${selectedProspect.id}/message`, {method:"POST", body:JSON.stringify({offer:$("offerInput").value})}); $("subjectOutput").value=draft.subject; $("bodyOutput").value=draft.message; status.textContent="Draft generated. Edit it before sending."; status.className="form-message success"; selectedProspect.ai_subject=draft.subject; selectedProspect.ai_message=draft.message; } + catch (error) { status.textContent = error.message; status.className="form-message error"; } +}); +$("copyMessage").addEventListener("click", async () => { const text = `Subject: ${$("subjectOutput").value}\n\n${$("bodyOutput").value}`; await navigator.clipboard.writeText(text); $("messageStatus").textContent="Copied to clipboard."; }); $("search").addEventListener("input", renderExplorer); $("status").addEventListener("change", renderExplorer); $("refresh").addEventListener("click", loadProspects); +document.getElementById("exportCsv")?.addEventListener("click", () => { window.location.href = "/api/export.csv"; }); $("runsRefresh").addEventListener("click", loadRuns); $("refreshAll").addEventListener("click", refreshAll); $("showRuns").addEventListener("click", (event) => { event.preventDefault(); $("runs").scrollIntoView({ behavior: "smooth", block: "center" }); }); +$("scheduleForm").addEventListener("submit", async (event) => { event.preventDefault(); const m=$("scheduleMessage"); try { await api("/api/schedules", {method:"POST",body:JSON.stringify({category:$("schedule-category").value,keywords:$("schedule-keywords").value,location:$("schedule-location").value,interval_hours:Number($("schedule-interval").value)})}); m.textContent="Schedule saved."; m.className="form-message success"; await loadSchedules(); } catch(e){m.textContent=e.message;m.className="form-message error";} }); refreshAll().catch((error) => { $("explorerList").innerHTML = `
Unable to load workspace

${esc(error.message)}

`; }); diff --git a/web/index.html b/web/index.html index 2946e14..3ec7026 100644 --- a/web/index.html +++ b/web/index.html @@ -52,7 +52,7 @@
-
+
Results are deduplicated before saving.
@@ -65,7 +65,7 @@
-
03 · CATALOGUE

Prospect explorer

Sort by opportunity score, inspect the evidence, and mark your next action.

⌕
+
03 · CATALOGUE

Prospect explorer

Sort by opportunity score, inspect the evidence, and mark your next action.

⌕
0 places Higher opportunity first
Loading catalogue…
@@ -73,10 +73,12 @@
SETUP

One-time worker connection

The bundled standard Maps worker runs locally and does not need a Google or scraper API key.

  1. Run docker compose up -d --build.
  2. Keep GMAPS_API_MODE=standard in .env.
  3. Start a discovery run. Open port 8081 only if you want to inspect the worker's native UI.
+
AUTOMATION

Scheduled discovery

Repeat a search on a bounded interval and review the new results.

ProspectOS · personal research workspacePublic data only · no automated outreach
+
AI ASSISTED DRAFT

Tailor a message

Review the evidence, add your offer, then edit before sending.

 diff --git a/web/style.css b/web/style.css index f560c55..4b58f11 100644 --- a/web/style.css +++ b/web/style.css @@ -14,6 +14,7 @@ background: var(--soft); line-height: 1.45; } +.message-dialog{width:min(680px,calc(100% - 32px));border:0;border-radius:18px;padding:0;box-shadow:0 24px 80px #11182733}.message-dialog::backdrop{background:#11182766}.message-dialog form{padding:26px;display:grid;gap:10px}.dialog-head{display:flex;justify-content:space-between;gap:16px;margin-bottom:8px}.message-dialog textarea,.message-dialog input{width:100%;box-sizing:border-box;border:1px solid #d7dbea;border-radius:10px;padding:11px;font:inherit}.dialog-actions{display:flex;justify-content:flex-end;gap:8px;margin-top:8px} * { box-sizing: border-box; } html { scroll-behavior: smooth; }