From a8182137fca58da8d01f779003cd95318d5e832a Mon Sep 17 00:00:00 2001 From: Marco0300 Date: Sat, 5 Sep 2026 23:27:20 +0200 Subject: [PATCH] deploy DOM search v1 update --- README.md | 26 +-- api/browser_investigator.py | 348 +++++++++++++++++++++++++++++++++++- api/server.py | 142 ++++++++++++--- tests/test_app.py | 22 +++ 4 files changed, 507 insertions(+), 31 deletions(-) diff --git a/README.md b/README.md index ea5af14..7203224 100644 --- a/README.md +++ b/README.md @@ -24,8 +24,13 @@ passwords are redacted. Logs are retained up to `LOG_RETENTION` entries (20,000 by default). From a prospect's detail drawer, **Assess with AI** stores a contact probability, recommendation, and short evidence-grounded rationale for that specific company. -The same drawer can optionally run a bounded Browser Use investigation to look -for a likely official website when Maps did not list one. Candidate domains are +The same drawer can optionally run a bounded website investigation to look for +a likely official website when Maps did not list one. The investigator first +opens public Bing/alternative result pages with headless Playwright and reads +the visible result links and snippets directly from the DOM; it does not rely +on unstable Browser Use element indexes. Search links are persisted immediately +and then checked for reachability/title identity evidence. Browser Use remains +as a fallback when no parseable result links are found. Candidate domains are shown with evidence and must be accepted by you before they replace the catalogue record. @@ -107,14 +112,15 @@ is never returned by the Settings API or sent to the browser after saving. ### Browser Use website investigator -Browser Use is an optional local browser agent. It uses the AI provider already -configured in Settings to navigate a small number of public search/result pages, -then returns candidate URLs as structured data. It is intentionally limited to -unauthenticated research: no proxy rotation, stealth fingerprints, CAPTCHA -solving, login, or challenge bypass is configured. If a site blocks the browser, -the action returns an explicit error and the rest of the prospect workflow is -left intact. Browser Use itself is open source, but each investigation still -uses your configured model/provider and may consume its quota. +Browser Use is an optional local browser agent. The website finder first uses +deterministic Playwright DOM extraction so a search result cannot be lost just +because an AI click target became stale. It uses the AI provider only as a +bounded fallback when no usable result link was extracted. Both paths are +limited to unauthenticated public research: no proxy rotation, stealth +fingerprints, CAPTCHA solving, login, or challenge bypass is configured. If a +site blocks the browser, the action records the block and leaves any partial +candidate links intact. Browser Use itself is open source, but fallback +investigations still use your configured model/provider and may consume quota. To use it: diff --git a/api/browser_investigator.py b/api/browser_investigator.py index f4cba9b..b2ae49e 100644 --- a/api/browser_investigator.py +++ b/api/browser_investigator.py @@ -11,14 +11,27 @@ import inspect import json import os import re +import time from typing import Any, Callable -from urllib.parse import urlparse +from urllib.parse import parse_qs, quote_plus, unquote, urlparse _BLOCKED_HOSTS = { "facebook.com", "instagram.com", "linkedin.com", "youtube.com", "x.com", "twitter.com", "google.com", "google.co.za", "bing.com", "brave.com", "duckduckgo.com", "startpage.com", "yelp.com", "yellowpages.com", + "yellosa.co.za", "snupit.co.za", "zaubee.com", "mapquest.com", +} + +_SEARCH_BLOCKED_MARKERS = ( + "unusual traffic", "recaptcha", "captcha", "automated queries", + "access denied", "detected unusual traffic", "verify you are human", +) + +_GENERIC_NAME_WORDS = { + "and", "the", "services", "service", "solutions", "company", "pty", + "ltd", "inc", "llc", "south", "africa", "sa", "electrical", + "electrician", "electricians", "installation", "installations", } @@ -41,6 +54,339 @@ def _normalise_url(value: Any) -> str: return text[:1000] +def _candidate_domain(url: str) -> str: + host = (urlparse(url).hostname or "").lower().rstrip(".") + return host.removeprefix("www.") + + +def _normalise_search_href(value: Any) -> str: + """Turn a search-result href into the real public URL. + + Search engines sometimes wrap result links in redirect URLs. We unwrap + the common query-string forms before applying the blocked-host check; the + old implementation rejected those wrappers before it could see the real + business domain. + """ + text = str(value or "").strip() + if text.startswith("//"): + text = "https:" + text + if text.startswith("/"): + return "" + try: + parsed = urlparse(text) + query = parse_qs(parsed.query) + if parsed.hostname and parsed.hostname.lower().endswith(("duckduckgo.com", "bing.com", "bingj.com")): + for key in ("uddg", "url", "u", "r"): + target = query.get(key, [""])[0] + target = unquote(target) + if target.startswith(("http://", "https://")): + text = target + break + except Exception: + return "" + return _normalise_url(text) + + +def _tokens(value: str, *, include_generic: bool = False) -> list[str]: + values = re.findall(r"[a-z0-9]+", str(value or "").lower()) + return [value for value in values if len(value) >= 3 and (include_generic or value not in _GENERIC_NAME_WORDS)] + + +def _location_from_address(address: str) -> str: + text = str(address or "").strip() + try: + parsed = json.loads(text) + if isinstance(parsed, dict): + return str(parsed.get("city") or parsed.get("town") or parsed.get("municipality") or parsed.get("state") or "").strip() + except (TypeError, ValueError, json.JSONDecodeError): + pass + parts = [part.strip() for part in re.split(r"[,|]", text) if part.strip()] + return parts[-1] if parts else "" + + +def _search_query_variants(prospect: dict[str, Any]) -> list[str]: + name = str(prospect.get("name") or "").strip() + if not name: + return [] + location = _location_from_address(str(prospect.get("address") or "")) + category = str(prospect.get("category") or "").strip() + phone = re.sub(r"[^0-9]+", "", str(prospect.get("phone") or "")) + queries = [ + f'"{name}" {location}'.strip(), + f'"{name}" {phone}'.strip() if phone else "", + f'"{name}" site:co.za'.strip(), + f'"{name}" {category} {location}'.strip(), + f'"{name}"'.strip(), + ] + output: list[str] = [] + seen: set[str] = set() + for query in queries: + query = " ".join(query.split()) + if query and query not in seen: + seen.add(query) + output.append(query) + return output[:5] + + +def _candidate_score(prospect: dict[str, Any], url: str, title: str, context: str) -> tuple[int, list[str]]: + """Score a result using only visible search-result evidence. + + This is intentionally deterministic. It prevents an LLM's final + ``done`` message from discarding a domain that was visibly present on the + search page. + """ + domain = _candidate_domain(url) + haystack = " ".join((title, context, domain)).lower() + domain_stem = domain + if domain.endswith(".co.za"): + domain_stem = domain[:-6] + elif domain.count(".") >= 1: + domain_stem = domain.rsplit(".", 1)[0] + compact_domain = re.sub(r"[^a-z0-9]", "", domain_stem) + name = str(prospect.get("name") or "") + name_compact = re.sub(r"[^a-z0-9]", "", name.lower()) + name_tokens = _tokens(name) + evidence: list[str] = [] + score = 18 + if name_compact and (name_compact in compact_domain or compact_domain in name_compact): + score += 52 + evidence.append("domain closely matches business name") + matching_tokens = [token for token in name_tokens if token in compact_domain] + if matching_tokens: + score += min(30, 12 * len(matching_tokens)) + evidence.append("name token match: " + ", ".join(matching_tokens[:4])) + if any(token in haystack for token in name_tokens): + score += 8 + evidence.append("business name appears in result text") + phone = re.sub(r"[^0-9]", "", str(prospect.get("phone") or "")) + context_digits = re.sub(r"[^0-9]", "", context) + if phone and len(phone) >= 7 and phone[-7:] in context_digits: + score += 14 + evidence.append("phone appears in result text") + location_tokens = _tokens(_location_from_address(str(prospect.get("address") or "")), include_generic=True) + if location_tokens and any(token in haystack for token in location_tokens): + score += 6 + evidence.append("location appears in result text") + if domain.endswith(".co.za"): + score += 5 + evidence.append("South African .co.za domain") + return max(0, min(100, score)), evidence + + +def _extract_visible_result_links(page: Any) -> list[dict[str, str]]: + """Read result anchors/snippets from the loaded page DOM. + + No indexed click targets are used. This is the critical reliability path + for Bing pages whose element indexes can change after navigation. + """ + script = """ + () => { + const cardSelectors = ["li.b_algo", "article", ".result", ".result-item", ".w-gl__result"]; + const cards = cardSelectors.flatMap(selector => Array.from(document.querySelectorAll(selector))); + const records = []; + const seen = new Set(); + const add = (anchor, card) => { + if (!anchor || seen.has(anchor)) return; + const href = anchor.href || anchor.getAttribute("href") || ""; + const title = (anchor.innerText || anchor.textContent || anchor.getAttribute("aria-label") || "").trim(); + const context = ((card && card.innerText) || title || "").trim(); + if (!href || !title || title.length < 3) return; + seen.add(anchor); + records.push({href, title, context: context.slice(0, 1600)}); + }; + for (const card of cards) { + const anchor = card.querySelector("h2 a, h3 a, a.result__a, a.heading-serpresult, a.w-gl__result-title, a[href]"); + add(anchor, card); + } + if (!records.length) { + for (const anchor of Array.from(document.querySelectorAll("a[href]"))) { + const card = anchor.closest("li, article, .result, .result-item, .w-gl__result"); + add(anchor, card); + if (records.length >= 80) break; + } + } + return records.slice(0, 80); + } + """ + try: + raw = page.evaluate(script) + except Exception: + return [] + if not isinstance(raw, list): + return [] + output: list[dict[str, str]] = [] + for item in raw: + if not isinstance(item, dict): + continue + output.append({ + "href": str(item.get("href") or "")[:1200], + "title": " ".join(str(item.get("title") or "").split())[:300], + "context": " ".join(str(item.get("context") or "").split())[:1800], + }) + return output + + +def _safe_log(log_callback: Callable[[str, str, dict[str, Any]], Any] | None, + kind: str, event: str, details: dict[str, Any]) -> None: + if not log_callback: + return + try: + result = log_callback(kind, event, details) + # The route callback is synchronous. Do not attempt to run an + # arbitrary coroutine here; Browser Use's async path emits its own + # events through ``emit`` below. + if inspect.isawaitable(result): + try: + result.close() # type: ignore[attr-defined] + except Exception: + pass + except Exception: + return + + +def direct_search_candidates( + prospect: dict[str, Any], + *, + user_agent: str = "ProspectOS/2.0 (+personal-use)", + log_callback: Callable[[str, str, dict[str, Any]], Any] | None = None, + requested: int = 5, + timeout_seconds: int = 90, +) -> list[dict[str, Any]]: + """Search public result pages and extract links directly from their DOM. + + Bing is tried first because it is usually available when Google has + challenged the browser. Other public engines are fallbacks. This path + is deterministic and can return candidates even when Browser Use runs out + of steps or produces malformed JSON. + """ + try: + from playwright.sync_api import sync_playwright, TimeoutError as PlaywrightTimeout + except ImportError: + _safe_log(log_callback, "search", "website_search_unavailable", {"error": "playwright_not_installed"}) + return [] + queries = _search_query_variants(prospect) + if not queries: + return [] + try: + hard_timeout = max(15, min(180, int(timeout_seconds))) + except (TypeError, ValueError): + hard_timeout = 90 + deadline = time.monotonic() + hard_timeout + executable = os.environ.get("PLAYWRIGHT_CHROMIUM_EXECUTABLE_PATH", "").strip() + options: dict[str, Any] = {"headless": True} + if executable: + options["executable_path"] = executable + engines = ( + ("bing", "https://www.bing.com/search?q={query}&count=10&cc=ZA&setlang=en-ZA"), + ("duckduckgo", "https://html.duckduckgo.com/html/?q={query}"), + ("brave", "https://search.brave.com/search?q={query}"), + ) + found: dict[str, dict[str, Any]] = {} + try: + with sync_playwright() as playwright: + browser = playwright.chromium.launch(**options) + try: + for engine, template in engines: + for query_index, query in enumerate(queries): + if time.monotonic() >= deadline: + _safe_log(log_callback, "search", "website_search_deadline", {"engine": engine, "query": query}) + return sorted(found.values(), key=lambda item: item["confidence"], reverse=True)[:requested] + url = template.format(query=quote_plus(query)) + _safe_log(log_callback, "search", "website_search_started", {"engine": engine, "query": query, "url": url}) + context = None + try: + context = browser.new_context( + locale="en-ZA", java_script_enabled=True, service_workers="block", + viewport={"width": 1365, "height": 900}, user_agent=user_agent, + ) + page = context.new_page() + page.set_default_timeout(8_000) + response = None + navigation_timed_out = False + try: + response = page.goto(url, wait_until="domcontentloaded", timeout=10_000) + except PlaywrightTimeout: + navigation_timed_out = True + try: + page.wait_for_timeout(800) + except Exception: + pass + body = "" + try: + body = page.locator("body").inner_text()[:20_000] + except Exception: + pass + page_title = "" + try: + page_title = str(page.title() or "")[:300] + except Exception: + pass + blocked_page = any(marker in (body + " " + page_title).lower() for marker in _SEARCH_BLOCKED_MARKERS) + visible = [] if blocked_page else _extract_visible_result_links(page) + page_candidates = 0 + for item in visible: + href = _normalise_search_href(item.get("href")) + domain = _candidate_domain(href) if href else "" + if not href or not domain or domain in found: + if href and domain in found: + score, evidence = _candidate_score(prospect, href, item.get("title", ""), item.get("context", "")) + existing = found[domain] + existing["confidence"] = max(existing["confidence"], score) + existing["evidence"] = list(dict.fromkeys(existing["evidence"] + evidence))[:8] + existing["queries"] = list(dict.fromkeys(existing.get("queries", []) + [query]))[:8] + continue + score, evidence = _candidate_score(prospect, href, item.get("title", ""), item.get("context", "")) + # A result must have some visible relationship + # to the business; this filters unrelated ads, + # maps, and navigation links. + if score < 30: + continue + candidate = { + "url": href, + "domain": domain, + "confidence": score, + "evidence": evidence + [f"visible on {engine} search results"], + "status": "candidate", + "source": "browser_dom", + "search_engine": engine, + "title": item.get("title", "")[:300], + "snippet": item.get("context", "")[:1200], + "queries": [query], + } + found[domain] = candidate + page_candidates += 1 + _safe_log(log_callback, "search", "website_search_candidate", {"engine": engine, "query": query, **candidate}) + _safe_log(log_callback, "search", "website_search_page", { + "engine": engine, "query": query, "url": url, "title": page_title, + "http_status": int(getattr(response, "status", 0) or 0) if response else None, + "blocked": blocked_page, "navigation_timed_out": navigation_timed_out, + "visible_result_count": len(visible), "candidate_count": page_candidates, + }) + if len(found) >= requested: + return sorted(found.values(), key=lambda item: item["confidence"], reverse=True)[:requested] + # One or two targeted Bing queries are normally + # enough. Do not burn the full browser deadline + # just to fill a five-domain quota when the + # official-looking result has already been found. + if found and query_index >= 1: + return sorted(found.values(), key=lambda item: item["confidence"], reverse=True)[:requested] + except Exception as exc: + _safe_log(log_callback, "search", "website_search_engine_error", {"engine": engine, "query": query, "url": url, "error": str(exc)[:500]}) + finally: + if context is not None: + try: + context.close() + except Exception: + pass + finally: + browser.close() + except Exception as exc: + _safe_log(log_callback, "search", "website_search_failed", {"error": str(exc)[:500]}) + results = sorted(found.values(), key=lambda item: item["confidence"], reverse=True)[:requested] + _safe_log(log_callback, "search", "website_search_completed", {"candidate_count": len(results), "candidates": results}) + return results + + def _parse_agent_result(raw: Any) -> list[dict[str, Any]]: """Parse and normalise the agent's final JSON candidate list.""" text = str(raw or "").strip() diff --git a/api/server.py b/api/server.py index 84249bf..8bb0498 100644 --- a/api/server.py +++ b/api/server.py @@ -27,10 +27,10 @@ from urllib.parse import parse_qs, quote_plus, unquote, urlparse try: from intelligence import domain_candidates, scan_website - from browser_investigator import investigate_websites + from browser_investigator import direct_search_candidates, investigate_websites except ImportError: # tests import the module as api.server from api.intelligence import domain_candidates, scan_website - from api.browser_investigator import investigate_websites + from api.browser_investigator import direct_search_candidates, investigate_websites DB_PATH = os.environ.get("PROSPECT_DB", "/data/prospects.db") MAX_RESULTS = 50 @@ -417,6 +417,95 @@ def recover_stale_runs() -> None: conn.close() +def persist_website_candidates(prospect_id: int, candidates: list[dict[str, Any]]) -> list[dict[str, Any]]: + """Upsert candidate domains immediately, preserving accepted decisions. + + Search-page extraction and Browser Use are both allowed to return partial + results. Persisting in one small transaction before any optional AI work + means a timeout, model error, or disconnected UI cannot discard links + that were already visible to the browser. + """ + conn = db() + try: + for candidate in candidates: + url = str(candidate.get("url") or "").strip() + domain = str(candidate.get("domain") or clean_domain(url)).lower().strip() + if not url or not domain: + continue + evidence = candidate.get("evidence") + if not isinstance(evidence, list): + evidence = [] + raw = dict(candidate) + source = str(candidate.get("source") or "browser_use").strip()[:80] or "browser_use" + conn.execute( + """INSERT INTO website_candidates(prospect_id,url,domain,confidence,evidence,source,status,discovered_at,raw_json) + VALUES(?,?,?,?,?,?,?,?,?) + ON CONFLICT(prospect_id,domain) DO UPDATE SET + url=excluded.url, confidence=MAX(website_candidates.confidence, excluded.confidence), + evidence=excluded.evidence, source=excluded.source, discovered_at=excluded.discovered_at, + raw_json=excluded.raw_json, + status=CASE WHEN website_candidates.status='accepted' THEN 'accepted' ELSE 'candidate' END""", + (prospect_id, url[:1000], domain[:255], max(0, min(100, as_int(candidate.get("confidence")))), + json.dumps(evidence[:12], ensure_ascii=False), source, "candidate", now(), + json.dumps(raw, ensure_ascii=False, default=str)[:100_000]), + ) + conn.commit() + rows = conn.execute("SELECT * FROM website_candidates WHERE prospect_id=? ORDER BY confidence DESC, id DESC", (prospect_id,)).fetchall() + return [dict(item) for item in rows] + finally: + conn.close() + + +def verify_website_candidates(prospect: dict[str, Any], candidates: list[dict[str, Any]]) -> list[dict[str, Any]]: + """Attach a small, evidence-grounded reachability/identity check. + + Search results are useful leads, not proof of ownership. A bounded + website scan confirms whether the domain responds and whether its title or + final URL contains business-name tokens. Failed checks leave the result + as a reviewable candidate instead of deleting it. + """ + name_tokens = [token for token in re.findall(r"[a-z0-9]+", str(prospect.get("name") or "").lower()) + if len(token) >= 3 and token not in {"and", "the", "services", "service", "company", "pty", "ltd", "electrical"}] + output: list[dict[str, Any]] = [] + for original in candidates: + candidate = dict(original) + url = str(candidate.get("url") or "").strip() + if not url: + output.append(candidate) + continue + try: + checked = scan_website(url, timeout_seconds=website_scan_timeout_seconds()) + status = str(checked.get("website_status") or "unknown") + title = str(checked.get("website_title") or "").strip() + final_url = str(checked.get("website_final_url") or url).strip() + haystack = (title + " " + final_url).lower() + evidence = [str(item).strip() for item in (candidate.get("evidence") or []) if str(item).strip()] + matched = [token for token in name_tokens if token in haystack] + if matched: + evidence.append("website title or URL matches: " + ", ".join(matched[:4])) + candidate["confidence"] = min(100, max(int(candidate.get("confidence") or 0), int(candidate.get("confidence") or 0) + 8)) + if checked.get("website_http_status"): + evidence.append(f"website responded HTTP {int(checked.get('website_http_status'))}") + if status in {"healthy", "redirect", "http_only", "slow"}: + evidence.append("domain is reachable for review") + elif checked.get("website_error"): + evidence.append("website check: " + str(checked.get("website_error"))[:160]) + candidate["evidence"] = list(dict.fromkeys(evidence))[:12] + candidate["verification"] = { + "status": status, + "http_status": checked.get("website_http_status"), + "final_url": final_url[:1000], + "title": title[:300], + "identity_match": bool(matched), + "emails_found": len(checked.get("website_emails") or []), + } + except Exception as exc: + candidate.setdefault("evidence", []).append("website verification failed: " + str(exc)[:160]) + candidate["verification"] = {"status": "unknown", "identity_match": False, "error": str(exc)[:200]} + output.append(candidate) + return output + + def json_bytes(value: Any) -> bytes: return json.dumps(value, ensure_ascii=False).encode("utf-8") @@ -1711,6 +1800,34 @@ class Handler(BaseHTTPRequestHandler): def browser_log(kind: str, event: str, details: dict[str, Any]) -> None: record_log(kind, event, prospect_id=prospect_id, details=details) + # First use deterministic DOM extraction. This does not need + # an AI key and is deliberately run before Browser Use so a + # model timeout or malformed final response cannot erase a + # link that was already visible on Bing. + direct_candidates = direct_search_candidates( + dict(row), user_agent=OUTBOUND_USER_AGENT, log_callback=browser_log, + requested=5, timeout_seconds=min(180, int(browser_runtime["timeout_seconds"])), + ) + if direct_candidates: + # Save the raw search links first. Verification is a + # separate bounded step and must never erase a useful + # result if a public site is slow or unavailable. + persist_website_candidates(prospect_id, direct_candidates) + verified_candidates = verify_website_candidates(dict(row), direct_candidates) + saved_rows = persist_website_candidates(prospect_id, verified_candidates) + for candidate in verified_candidates: + record_log("search", "website_candidate_verified", prospect_id=prospect_id, + details={"domain": candidate.get("domain"), "confidence": candidate.get("confidence"), + "verification": candidate.get("verification", {})}) + record_log("browser", "website_candidates_persisted", prospect_id=prospect_id, + details={"source": "browser_dom", "candidate_count": len(direct_candidates), + "domains": [item.get("domain") for item in direct_candidates]}) + return self.send_json(200, {"items": saved_rows, "count": len(saved_rows), "source": "browser_dom"}) + + # Keep Browser Use as a bounded fallback for cases where the + # public result pages yielded no parseable links. + record_log("browser", "browser_use_fallback_started", prospect_id=prospect_id, + details={"reason": "direct_search_no_candidates", "model": runtime["model"]}) candidates = investigate_websites( dict(row), base_url=runtime["base_url"], api_key=runtime["api_key"], model=runtime["model"], user_agent=OUTBOUND_USER_AGENT, log_callback=browser_log, @@ -1718,24 +1835,9 @@ class Handler(BaseHTTPRequestHandler): timeout_seconds=browser_runtime["timeout_seconds"], llm_timeout_seconds=browser_runtime["llm_timeout_seconds"], ) record_log("browser", "website_investigation_completed", prospect_id=prospect_id, - details={"candidate_count": len(candidates), "model": runtime["model"]}) - conn = db() - for candidate in candidates: - encoded_evidence = json.dumps(candidate.get("evidence") or [], ensure_ascii=False) - encoded_raw = json.dumps(candidate, ensure_ascii=False) - conn.execute( - """INSERT INTO website_candidates(prospect_id,url,domain,confidence,evidence,source,status,discovered_at,raw_json) - VALUES(?,?,?,?,?,?,?,?,?) - ON CONFLICT(prospect_id,domain) DO UPDATE SET - url=excluded.url, confidence=excluded.confidence, evidence=excluded.evidence, - source=excluded.source, discovered_at=excluded.discovered_at, raw_json=excluded.raw_json, - status=CASE WHEN website_candidates.status='accepted' THEN 'accepted' ELSE 'candidate' END""", - (prospect_id, candidate["url"], candidate["domain"], candidate["confidence"], encoded_evidence, - "browser_use", "candidate", now(), encoded_raw), - ) - conn.commit() - rows = conn.execute("SELECT * FROM website_candidates WHERE prospect_id=? ORDER BY confidence DESC, id DESC", (prospect_id,)).fetchall() - return self.send_json(200, {"items": [dict(item) for item in rows], "count": len(rows)}) + details={"candidate_count": len(candidates), "model": runtime["model"], "source": "browser_use"}) + saved_rows = persist_website_candidates(prospect_id, candidates) + return self.send_json(200, {"items": saved_rows, "count": len(saved_rows), "source": "browser_use"}) except RuntimeError as exc: record_log("browser", "website_investigation_failed", level="error", prospect_id=prospect_id, error=str(exc)) return self.send_json(503, {"error": str(exc)}) diff --git a/tests/test_app.py b/tests/test_app.py index 8f85c9c..90438c8 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -163,6 +163,28 @@ class BrowserOnlyAppTests(unittest.TestCase): self.assertEqual(candidates[0]["confidence"], 88) self.assertEqual(candidates[1]["evidence"], []) + def test_direct_search_scoring_recognises_compact_business_domain(self): + prospect = {"name": "BRIGHTSPARK ELECTRICAL Services", "address": '{"city":"Cape Town"}', "phone": "067 732 2090"} + confidence, evidence = browser_investigator._candidate_score( + prospect, "https://brightsparkelec.co.za/", "Brightspark Electrical Services", "Cape Town 067 732 2090" + ) + self.assertGreaterEqual(confidence, 80) + self.assertTrue(any("domain" in item for item in evidence)) + + def test_dom_search_candidates_are_saved_before_browser_agent(self): + conn = server.db() + conn.execute("INSERT INTO prospects(name,domain,category,address,discovered_at) VALUES(?,?,?,?,?)", ("Brightspark Electrical Services", "maps:bright", "Electrician", "Cape Town", server.now())) + conn.commit(); conn.close() + found = [{"url": "https://brightsparkelec.co.za/", "domain": "brightsparkelec.co.za", "confidence": 88, "evidence": ["visible on bing search results"], "source": "browser_dom"}] + with patch("api.server.direct_search_candidates", return_value=found) as direct, patch("api.server.investigate_websites") as agent: + status, body = self.request("POST", "/api/prospects/1/website-candidates", {}) + self.assertEqual(status, 200) + self.assertEqual(body["source"], "browser_dom") + self.assertEqual(body["items"][0]["domain"], "brightsparkelec.co.za") + self.assertEqual(body["items"][0]["source"], "browser_dom") + direct.assert_called_once() + agent.assert_not_called() + def test_browser_website_candidates_can_be_found_and_accepted(self): conn = server.db() conn.execute("INSERT INTO prospects(name,domain,category,address,discovered_at) VALUES(?,?,?,?,?)", ("Acme Plumbing", "maps:abc", "plumber", "Cape Town", server.now()))