diff --git a/api/browser_investigator.py b/api/browser_investigator.py index b2ae49e..d296f8f 100644 --- a/api/browser_investigator.py +++ b/api/browser_investigator.py @@ -7,6 +7,7 @@ CAPTCHAs, rotate proxies, or continue after an access challenge. from __future__ import annotations import asyncio +import base64 import inspect import json import os @@ -34,6 +35,8 @@ _GENERIC_NAME_WORDS = { "electrician", "electricians", "installation", "installations", } +_ALLOWED_DOMAIN_SUFFIXES = (".co.za", ".com") + def _normalise_url(value: Any) -> str: text = str(value or "").strip() @@ -49,6 +52,8 @@ def _normalise_url(value: Any) -> str: host = parsed.hostname.lower().rstrip(".") if host in {"localhost", "127.0.0.1", "::1"} or host.endswith(".local"): return "" + if not any(host.endswith(suffix) for suffix in _ALLOWED_DOMAIN_SUFFIXES): + return "" if any(host == blocked or host.endswith("." + blocked) for blocked in _BLOCKED_HOSTS): return "" return text[:1000] @@ -82,6 +87,18 @@ def _normalise_search_href(value: Any) -> str: if target.startswith(("http://", "https://")): text = target break + # Bing's click-tracking URL commonly stores the destination + # as ``a1`` followed by URL-safe base64. Decode it before the + # blocked-host/TLD checks, otherwise the real domain is lost. + if key == "u" and target.startswith("a1"): + try: + encoded = target[2:] + decoded = base64.urlsafe_b64decode(encoded + "=" * (-len(encoded) % 4)).decode("utf-8", "ignore") + if decoded.startswith(("http://", "https://")): + text = decoded + break + except (ValueError, UnicodeError): + pass except Exception: return "" return _normalise_url(text) @@ -115,6 +132,7 @@ def _search_query_variants(prospect: dict[str, Any]) -> list[str]: f'"{name}" {location}'.strip(), f'"{name}" {phone}'.strip() if phone else "", f'"{name}" site:co.za'.strip(), + f'"{name}" site:com'.strip(), f'"{name}" {category} {location}'.strip(), f'"{name}"'.strip(), ] @@ -136,7 +154,9 @@ def _candidate_score(prospect: dict[str, Any], url: str, title: str, context: st search page. """ domain = _candidate_domain(url) - haystack = " ".join((title, context, domain)).lower() + title_text = str(title or "").lower() + context_text = str(context or "").lower() + haystack = " ".join((title_text, context_text, domain)).lower() domain_stem = domain if domain.endswith(".co.za"): domain_stem = domain[:-6] @@ -147,15 +167,26 @@ def _candidate_score(prospect: dict[str, Any], url: str, title: str, context: st name_compact = re.sub(r"[^a-z0-9]", "", name.lower()) name_tokens = _tokens(name) evidence: list[str] = [] + domain_matches = [token for token in name_tokens if token in compact_domain] + title_matches = [token for token in name_tokens if token in title_text] + # Require identity evidence from the domain or the result title. A + # generic snippet mention (for example an unrelated directory page that + # happens to contain “electrical”) is not enough to save a candidate. + strong_identity = bool(name_compact and (name_compact in compact_domain or compact_domain in name_compact)) + strong_identity = strong_identity or len(domain_matches) >= 2 + strong_identity = strong_identity or len(title_matches) >= 2 + if not strong_identity: + return 0, [] 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] + matching_tokens = domain_matches if matching_tokens: - score += min(30, 12 * len(matching_tokens)) + score += min(32, 14 * len(matching_tokens)) evidence.append("name token match: " + ", ".join(matching_tokens[:4])) - if any(token in haystack for token in name_tokens): + if title_matches: + score += min(18, 9 * len(title_matches)) score += 8 evidence.append("business name appears in result text") phone = re.sub(r"[^0-9]", "", str(prospect.get("phone") or "")) @@ -287,7 +318,7 @@ def direct_search_candidates( browser = playwright.chromium.launch(**options) try: for engine, template in engines: - for query_index, query in enumerate(queries): + for query in 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] @@ -339,7 +370,7 @@ def direct_search_candidates( # A result must have some visible relationship # to the business; this filters unrelated ads, # maps, and navigation links. - if score < 30: + if score < 48: continue candidate = { "url": href, @@ -364,11 +395,12 @@ def direct_search_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: + # Once a high-confidence domain is visible, stop + # before spending time on broad fallback engines. + # Low-confidence results do not end the search; + # the remaining exact/TLD queries still get a + # chance to expose the official domain. + if found and max(item["confidence"] for item in found.values()) >= 88: 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]}) diff --git a/tests/test_app.py b/tests/test_app.py index 90438c8..09e0012 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -171,6 +171,19 @@ class BrowserOnlyAppTests(unittest.TestCase): self.assertGreaterEqual(confidence, 80) self.assertTrue(any("domain" in item for item in evidence)) + def test_website_candidates_are_limited_to_coza_and_com(self): + raw = '{"candidates":[{"url":"https://bright.example.org","confidence":99},{"url":"https://brightsparkelec.co.za","confidence":80},{"url":"https://brightspark.com","confidence":75}]}' + candidates = browser_investigator._parse_agent_result(raw) + self.assertEqual([item["domain"] for item in candidates], ["brightsparkelec.co.za", "brightspark.com"]) + + def test_direct_search_rejects_generic_result_without_identity_signal(self): + prospect = {"name": "BRIGHTSPARK ELECTRICAL Services", "address": '{"city":"Cape Town"}'} + confidence, evidence = browser_investigator._candidate_score( + prospect, "https://unrelated-electrical.co.za/", "Electrical services in Cape Town", "Directory listing" + ) + self.assertEqual(confidence, 0) + self.assertEqual(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()))