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

deploy DOM search v2 update

Browse Source
This commit is contained in:
Marco0300
2026-09-05 23:37:10 +02:00
parent a8182137fc
commit e0764cc716
2 changed files with 56 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/browser_investigator.py
+43 -11
View File
@@ -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]})
tests/test_app.py
+13
View File
@@ -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()))