root/MarketingTool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

deploy DOM search v1 update

Browse Source
This commit is contained in:
Marco0300
2026-09-05 23:27:20 +02:00
parent 112ab552e9
commit a8182137fc
4 changed files with 507 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+16 -10
View File
@@ -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:
api/browser_investigator.py
+347 -1
View File
@@ -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()
api/server.py
+122 -20
View File
@@ -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)})
tests/test_app.py
+22
View File
@@ -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()))