| 1 | """Trustpilot brand-sentiment source for last30days. |
| 2 | |
| 3 | Shells out to ``trustpilot-pp-cli`` to surface a company's TrustScore and |
| 4 | Trustpilot's own AI review summary for brand/company topics. Trustpilot has no |
| 5 | API key, but it sits behind AWS WAF: the CLI harvests an ``aws-waf-token`` via |
| 6 | a one-time headless Chrome launch (~10s), then replays it over plain HTTP until |
| 7 | it expires. |
| 8 | |
| 9 | Activation gate: only available when ``trustpilot-pp-cli`` is on PATH. |
| 10 | ``pipeline.available_sources`` checks ``shutil.which`` before including |
| 11 | ``trustpilot``. |
| 12 | |
| 13 | Default-on safety (three gates): |
| 14 | 1. Brand-shape gate. The CLI is invoked only when the topic resolves to a |
| 15 | company/brand -- a domain-like token, or a short (<=2-word) capitalized |
| 16 | proper noun. Generic phrases ("AI coding agents", "agent memory") and |
| 17 | longer multi-word phrases never call the CLI, so Trustpilot stays quiet -- |
| 18 | and never harvests Chrome -- on non-company topics. An explicit resolved |
| 19 | domain (``--trustpilot-domain`` or an auto-resolve hint) bypasses this |
| 20 | gate: an explicit domain is proof of brand intent. |
| 21 | 2. Browser opt-out. Automated contexts (cron, CI, the eval harness) can set |
| 22 | ``LAST30DAYS_TRUSTPILOT_NO_BROWSER`` to disable the source entirely, so a |
| 23 | headless run never spawns the cookie harvest. |
| 24 | 3. Graceful degradation. Any CLI failure (no Chrome, expired cookie that |
| 25 | cannot re-harvest, timeout) degrades to empty results, never an error. |
| 26 | |
| 27 | Domain resolution: Trustpilot review pages are keyed by domain |
| 28 | (``www.thriftbooks.com``), not company name -- ``info ThriftBooks`` 404s. |
| 29 | Priority chain: user flag (verbatim) > auto-resolve hint (retries via search |
| 30 | on a miss) > domain token in the topic > CLI ``search`` name->domain lookup |
| 31 | (cached per topic) > cleaned topic (legacy behavior). |
| 32 | |
| 33 | Session pre-flight: ``ensure_session_ready`` performs one serialized |
| 34 | ``auth status`` / ``auth login`` before the parallel fan-out so concurrent |
| 35 | streams and vs-mode sub-runs never race their own Chrome harvests. |
| 36 | """ |
| 37 | |
| 38 | from __future__ import annotations |
| 39 | |
| 40 | import json |
| 41 | import os |
| 42 | import re |
| 43 | import shutil |
| 44 | import threading |
| 45 | import time |
| 46 | from typing import Any, Dict, List, Optional |
| 47 | |
| 48 | from . import dates, log, subproc |
| 49 | from .relevance import token_overlap_relevance |
| 50 | |
| 51 | |
| 52 | CLI_BIN = "trustpilot-pp-cli" |
| 53 | |
| 54 | SEARCH_TIMEOUT = 75 # generous: a cold run may harvest a WAF cookie (~10s). |
| 55 | |
| 56 | AUTH_STATUS_TIMEOUT = 20 # auth status is a local SQLite read; fast. |
| 57 | |
| 58 | NO_BROWSER_ENV = "LAST30DAYS_TRUSTPILOT_NO_BROWSER" |
| 59 | |
| 60 | # Among name-matching search hits, the top hit must have this many times the |
| 61 | # runner-up's review volume to win automatically. Lookalike/squatter pages have |
| 62 | # tiny volume (ThriftBooks: 2.8M vs 130); comparable volume means genuine |
| 63 | # ambiguity, where falling back beats silently picking the wrong company. |
| 64 | DOMAIN_DOMINANCE_FACTOR = 50 |
| 65 | |
| 66 | # Domain-like token, e.g. "chownow.com", "nothing.tech". |
| 67 | _DOMAIN_RE = re.compile(r"\b[a-z0-9][a-z0-9-]*\.(com|io|co|net|org|app|ai|dev|gg|tech|shop|store)\b") |
| 68 | |
| 69 | # Generic tokens that disqualify a short capitalized phrase from being a brand. |
| 70 | _GENERIC_TOKENS = { |
| 71 | "ai", "best", "top", "vs", "review", "reviews", "guide", "tutorial", |
| 72 | "how", "what", "why", "agents", "agent", "memory", "tips", "news", |
| 73 | } |
| 74 | |
| 75 | # Single-word programming languages, frameworks, runtimes, OSes, and dev tools. |
| 76 | # A bare capitalized "Python"/"React"/"Docker" query is overwhelmingly about the |
| 77 | # technology, not a company's customer reviews -- letting it through would both |
| 78 | # trigger the Chrome harvest and risk surfacing an unrelated company that shares |
| 79 | # the name. A user who genuinely wants the company can use its domain |
| 80 | # (e.g. "docker.com"), which still passes via the domain branch. |
| 81 | _TECH_TOKENS = { |
| 82 | "python", "javascript", "typescript", "java", "rust", "ruby", "php", |
| 83 | "kotlin", "scala", "golang", "swift", "elixir", "erlang", "haskell", |
| 84 | "react", "vue", "angular", "svelte", "django", "flask", "rails", "spring", |
| 85 | "node", "nodejs", "deno", "bun", "express", "nextjs", "nuxt", |
| 86 | "linux", "ubuntu", "debian", "fedora", "windows", "macos", "android", |
| 87 | "docker", "kubernetes", "k8s", "terraform", "ansible", "nginx", |
| 88 | "redis", "postgres", "postgresql", "mysql", "sqlite", "mongodb", "kafka", |
| 89 | "graphql", "webpack", "vite", "rust", "wasm", |
| 90 | } |
| 91 | |
| 92 | |
| 93 | def _log(msg: str) -> None: |
| 94 | log.source_log("Trustpilot", msg, tty_only=False) |
| 95 | |
| 96 | |
| 97 | def _is_available() -> bool: |
| 98 | """True when the trustpilot-pp-cli binary is on PATH.""" |
| 99 | return shutil.which(CLI_BIN) is not None |
| 100 | |
| 101 | |
| 102 | def _truthy(value: Any) -> bool: |
| 103 | return str(value or "").strip().lower() in ("1", "true", "yes", "on") |
| 104 | |
| 105 | |
| 106 | def _harvest_allowed(config: Optional[Dict[str, Any]]) -> bool: |
| 107 | """False when the browser opt-out is set (automated/headless contexts). |
| 108 | |
| 109 | Reads the opt-out from the merged config AND directly from the process |
| 110 | environment. The env fallback is load-bearing: ``config`` is assembled from |
| 111 | an allowlist in ``env.get_config``, so a fallback here guarantees the |
| 112 | documented kill-switch works even when the key is not propagated into |
| 113 | config (e.g. a bare ``LAST30DAYS_TRUSTPILOT_NO_BROWSER=1`` in cron/CI). |
| 114 | """ |
| 115 | if config and _truthy(config.get(NO_BROWSER_ENV)): |
| 116 | return False |
| 117 | if _truthy(os.environ.get(NO_BROWSER_ENV)): |
| 118 | return False |
| 119 | return True |
| 120 | |
| 121 | |
| 122 | def is_brand_shaped(topic: str) -> bool: |
| 123 | """True when the topic looks like a company/brand Trustpilot can resolve. |
| 124 | |
| 125 | A domain-like token always qualifies. Otherwise the topic must be a short |
| 126 | (<=2-word) capitalized proper noun with no generic tokens -- this lets |
| 127 | "ChowNow", "Nothing Phone", and "OpenAI" through while keeping "AI coding |
| 128 | agents", "agent memory", and "Golden State Warriors" out. |
| 129 | """ |
| 130 | if not topic or not topic.strip(): |
| 131 | return False |
| 132 | text = topic.strip() |
| 133 | if _DOMAIN_RE.search(text.lower()): |
| 134 | return True |
| 135 | words = text.split() |
| 136 | if len(words) > 2: |
| 137 | return False |
| 138 | if any(w.lower() in _GENERIC_TOKENS or w.lower() in _TECH_TOKENS for w in words): |
| 139 | return False |
| 140 | # At least one token must look like a proper noun (leading capital). |
| 141 | return any(w[:1].isupper() for w in words) |
| 142 | |
| 143 | |
| 144 | def _company_identifier(topic: str) -> str: |
| 145 | """Pick the identifier to hand the CLI: a domain token if present, else the |
| 146 | cleaned topic string.""" |
| 147 | m = _DOMAIN_RE.search(topic.lower()) |
| 148 | if m: |
| 149 | return m.group(0) |
| 150 | return topic.strip() |
| 151 | |
| 152 | |
| 153 | def _build_info_args(identifier: str) -> List[str]: |
| 154 | return [CLI_BIN, "info", identifier, "--agent"] |
| 155 | |
| 156 | |
| 157 | def _normalize_name(text: str) -> str: |
| 158 | """Case/whitespace/punctuation-insensitive brand-name key.""" |
| 159 | return re.sub(r"[^a-z0-9]", "", (text or "").lower()) |
| 160 | |
| 161 | |
| 162 | # name->domain results, keyed by normalized topic. Per-topic (NOT a single |
| 163 | # process-wide slot): vs-mode resolves several entities in one process, and a |
| 164 | # single slot would serve entity A's domain to entity B. |
| 165 | _domain_cache: Dict[str, Optional[str]] = {} |
| 166 | _domain_cache_lock = threading.Lock() |
| 167 | |
| 168 | _warmup_lock = threading.Lock() |
| 169 | _warmup_at: Optional[float] = None # time.monotonic() of the last warm-up |
| 170 | |
| 171 | # Warm-up freshness window, matching the CLI's ~4-minute safe replay bound for |
| 172 | # WAF tokens. A boolean-forever flag would leave long-lived host processes |
| 173 | # running stale (and never retrying a failed login); the TTL re-checks cheaply |
| 174 | # via `auth status` once the window lapses. |
| 175 | WARMUP_TTL_SECONDS = 240 |
| 176 | |
| 177 | |
| 178 | def _reset_state_for_tests() -> None: |
| 179 | """Clear module-level caches/flags (tests only).""" |
| 180 | global _warmup_at |
| 181 | with _domain_cache_lock: |
| 182 | _domain_cache.clear() |
| 183 | _warmup_at = None |
| 184 | |
| 185 | |
| 186 | def _warmup_fresh() -> bool: |
| 187 | return _warmup_at is not None and (time.monotonic() - _warmup_at) < WARMUP_TTL_SECONDS |
| 188 | |
| 189 | |
| 190 | def _select_search_hit(topic: str, hits: List[Any]) -> Optional[str]: |
| 191 | """Pick the canonical domain from search hits, or None when ambiguous. |
| 192 | |
| 193 | Name-match is mandatory: review volume must never override a name |
| 194 | mismatch, or the engine attributes another company's reviews to the topic. |
| 195 | Among name-matching hits the winner must dominate on review volume |
| 196 | (DOMAIN_DOMINANCE_FACTOR); comparable volume is genuine ambiguity and |
| 197 | falls back to legacy behavior. |
| 198 | """ |
| 199 | want = _normalize_name(topic) |
| 200 | if not want: |
| 201 | return None |
| 202 | matching: List[tuple[int, str]] = [] |
| 203 | for hit in hits: |
| 204 | if not isinstance(hit, dict): |
| 205 | continue |
| 206 | domain = str(hit.get("domain") or hit.get("identifyingName") or "").strip() |
| 207 | name = str(hit.get("displayName") or hit.get("name") or "").strip() |
| 208 | if not domain or _normalize_name(name) != want: |
| 209 | continue |
| 210 | try: |
| 211 | count = int(hit.get("numberOfReviews") or 0) |
| 212 | except (TypeError, ValueError): |
| 213 | count = 0 |
| 214 | matching.append((count, domain)) |
| 215 | if not matching: |
| 216 | top = next( |
| 217 | (str(h.get("domain") or "").strip() for h in hits |
| 218 | if isinstance(h, dict) and h.get("domain")), |
| 219 | "", |
| 220 | ) |
| 221 | if top: |
| 222 | _log( |
| 223 | f"no name-matching search hit; top candidate was '{top}' - " |
| 224 | f"pass --trustpilot-domain to target it" |
| 225 | ) |
| 226 | return None |
| 227 | matching.sort(reverse=True) |
| 228 | if len(matching) == 1: |
| 229 | return matching[0][1] |
| 230 | top_count, top_domain = matching[0] |
| 231 | runner_count, runner_domain = matching[1] |
| 232 | if top_count >= max(1, runner_count) * DOMAIN_DOMINANCE_FACTOR: |
| 233 | return top_domain |
| 234 | _log( |
| 235 | f"ambiguous search hits ('{top_domain}' vs '{runner_domain}'); " |
| 236 | f"falling back - pass --trustpilot-domain to disambiguate" |
| 237 | ) |
| 238 | return None |
| 239 | |
| 240 | |
| 241 | def _search_domain(topic: str) -> Optional[str]: |
| 242 | """Resolve a company name to its Trustpilot domain via the CLI's search. |
| 243 | |
| 244 | Cached per normalized topic (thread-safe), so repeat lookups cost one |
| 245 | subprocess while vs-mode entities still resolve independently. Only |
| 246 | definitive results are cached: a transient CLI failure (timeout, spawn |
| 247 | error, malformed JSON) returns None WITHOUT caching, so one flaky search |
| 248 | does not suppress resolution for this topic for the rest of the process. |
| 249 | """ |
| 250 | key = _normalize_name(topic) |
| 251 | if not key: |
| 252 | return None |
| 253 | with _domain_cache_lock: |
| 254 | if key in _domain_cache: |
| 255 | return _domain_cache[key] |
| 256 | data = _run_cli( |
| 257 | [CLI_BIN, "search", topic.strip(), "--limit", "5", "--agent"], |
| 258 | timeout=SEARCH_TIMEOUT, |
| 259 | ) |
| 260 | if not isinstance(data, dict) or "error" in data: |
| 261 | return None # transient failure: retry on the next lookup |
| 262 | hits = data.get("hits") |
| 263 | if not isinstance(hits, list): |
| 264 | # Degenerate payload (e.g. empty stdout parses to {}): not a |
| 265 | # definitive no-match -- do not cache, retry on the next lookup. |
| 266 | return None |
| 267 | domain = _select_search_hit(topic, hits) |
| 268 | if domain: |
| 269 | _log(f"resolved '{topic}' -> '{domain}' via search") |
| 270 | with _domain_cache_lock: |
| 271 | _domain_cache[key] = domain |
| 272 | return domain |
| 273 | |
| 274 | |
| 275 | def _is_session_fresh(status: Dict[str, Any]) -> bool: |
| 276 | """Read the freshness signal from an ``auth status --agent`` payload.""" |
| 277 | if not isinstance(status, dict) or "error" in status: |
| 278 | return False |
| 279 | containers: List[Dict[str, Any]] = [status] |
| 280 | session = status.get("session") |
| 281 | if isinstance(session, dict): |
| 282 | containers.append(session) |
| 283 | for container in containers: |
| 284 | for key in ("isFresh", "fresh"): |
| 285 | if key in container: |
| 286 | return bool(container[key]) |
| 287 | return False |
| 288 | |
| 289 | |
| 290 | def ensure_session_ready( |
| 291 | topic: str, |
| 292 | config: Optional[Dict[str, Any]] = None, |
| 293 | has_domain: bool = False, |
| 294 | ) -> None: |
| 295 | """Warm the CLI's WAF session, serialized, at the first Trustpilot fetch. |
| 296 | |
| 297 | Called from ``search_trustpilot`` (never from the pipeline's fan-out |
| 298 | setup), so it only ever delays the one capped Trustpilot stream, never |
| 299 | the other sources -- and never fires for runs whose plan fetches no |
| 300 | Trustpilot at all. The module lock serializes concurrent streams (vs-mode |
| 301 | fans out up to 6 entity sub-runs) so they never race their own Chrome |
| 302 | harvests. Freshness is a monotonic TTL (WARMUP_TTL_SECONDS, matching the |
| 303 | CLI's ~4-minute token bound), not a boolean-forever flag: long-lived host |
| 304 | processes re-check via ``auth status`` after the window lapses, which |
| 305 | also retries a previously failed login. ``auth login`` fires only when |
| 306 | ``auth status`` reports the session missing or stale -- login always |
| 307 | harvests (~10s Chrome), it has no freshness no-op. Logs only structured |
| 308 | status strings; the raw CLI payload carries live WAF-token prefixes and |
| 309 | must never be logged. Never raises. |
| 310 | """ |
| 311 | global _warmup_at |
| 312 | if _warmup_fresh(): |
| 313 | return |
| 314 | if not _is_available(): |
| 315 | return |
| 316 | if not _harvest_allowed(config): |
| 317 | return |
| 318 | if not has_domain and not is_brand_shaped(topic): |
| 319 | return |
| 320 | with _warmup_lock: |
| 321 | if _warmup_fresh(): |
| 322 | return |
| 323 | status = _run_cli( |
| 324 | [CLI_BIN, "auth", "status", "--agent"], timeout=AUTH_STATUS_TIMEOUT |
| 325 | ) |
| 326 | if _is_session_fresh(status): |
| 327 | _log("warm-up: fresh") |
| 328 | _warmup_at = time.monotonic() |
| 329 | return |
| 330 | # Missing session exits non-zero (an error dict here): that is the |
| 331 | # "login needed" signal, not a warm-up failure. |
| 332 | login = _run_cli([CLI_BIN, "auth", "login", "--agent"], timeout=SEARCH_TIMEOUT) |
| 333 | if isinstance(login, dict) and "error" in login: |
| 334 | _log("warm-up failed: auth login did not complete") |
| 335 | else: |
| 336 | _log("warm-up: harvested") |
| 337 | # Stamp even on failure: a broken Chrome will not fix itself within |
| 338 | # the TTL, per-call CLI auto-harvest remains the fallback, and the |
| 339 | # TTL lapse retries the warm-up later. |
| 340 | _warmup_at = time.monotonic() |
| 341 | |
| 342 | |
| 343 | def _run_cli(cmd: List[str], timeout: int) -> Dict[str, Any]: |
| 344 | """Invoke trustpilot-pp-cli and parse the JSON object. Never raises.""" |
| 345 | if not _is_available(): |
| 346 | return {"error": f"{CLI_BIN} not on PATH"} |
| 347 | try: |
| 348 | result = subproc.run_with_timeout(cmd, timeout=timeout) |
| 349 | except subproc.SubprocTimeout as exc: |
| 350 | _log(f"Timeout: {exc}") |
| 351 | return {"error": str(exc)} |
| 352 | except FileNotFoundError as exc: |
| 353 | _log(f"Binary missing: {exc}") |
| 354 | return {"error": str(exc)} |
| 355 | except OSError as exc: |
| 356 | _log(f"Spawn failed: {exc}") |
| 357 | return {"error": str(exc)} |
| 358 | |
| 359 | if result.returncode != 0: |
| 360 | snippet = (result.stderr or "").strip().splitlines()[:1] |
| 361 | first = snippet[0] if snippet else f"exit {result.returncode}" |
| 362 | _log(f"CLI exit {result.returncode}: {first}") |
| 363 | return {"error": first} |
| 364 | |
| 365 | stdout = result.stdout or "" |
| 366 | if not stdout.strip(): |
| 367 | return {} |
| 368 | try: |
| 369 | data = json.loads(stdout) |
| 370 | except json.JSONDecodeError as exc: |
| 371 | _log(f"JSON decode failed: {exc}") |
| 372 | return {"error": f"json decode: {exc}"} |
| 373 | return data if isinstance(data, dict) else {} |
| 374 | |
| 375 | |
| 376 | def search_trustpilot( |
| 377 | topic: str, |
| 378 | from_date: str, |
| 379 | to_date: str, |
| 380 | depth: str = "default", |
| 381 | config: Optional[Dict[str, Any]] = None, |
| 382 | explicit_domain: Optional[str] = None, |
| 383 | domain_is_hint: bool = False, |
| 384 | ) -> Dict[str, Any]: |
| 385 | """Look up a company's Trustpilot sentiment, gated on a brand-shaped topic. |
| 386 | |
| 387 | ``explicit_domain`` is used verbatim as the CLI identifier and bypasses |
| 388 | the brand-shape gate (an explicit domain is proof of brand intent). A |
| 389 | user-set domain is verbatim-final; a resolved hint |
| 390 | (``domain_is_hint=True``) retries via the CLI search when the lookup |
| 391 | misses, since auto-resolve can guess a plausible-but-wrong domain (the |
| 392 | official site is not always Trustpilot's canonical identifyingName). |
| 393 | |
| 394 | Without an explicit domain, a bare company name resolves via the CLI's |
| 395 | ``search`` (cached per topic) before falling back to the cleaned topic. |
| 396 | |
| 397 | Returns ``{"results": [info_dict]}`` for a resolved company, or |
| 398 | ``{"results": []}`` when the topic is not brand-shaped, the browser |
| 399 | opt-out is set, or the CLI fails. |
| 400 | """ |
| 401 | explicit_domain = (explicit_domain or "").strip() or None |
| 402 | user_domain = bool(explicit_domain) and not domain_is_hint |
| 403 | # Only a USER-set domain proves brand intent and bypasses the brand-shape |
| 404 | # gate. An auto-resolved hint must not widen activation beyond |
| 405 | # brand-shaped topics, or a generic topic that happens to yield a hint |
| 406 | # would trigger a Chrome harvest -- violating the module's documented |
| 407 | # "never harvests on non-company topics" contract. |
| 408 | if not user_domain and not is_brand_shaped(topic): |
| 409 | return {"results": []} |
| 410 | if not _is_available(): |
| 411 | return {"results": [], "error": f"{CLI_BIN} not on PATH"} |
| 412 | if not _harvest_allowed(config): |
| 413 | _log("skipped: browser opt-out set") |
| 414 | return {"results": []} |
| 415 | # Serialized session check at first source touch (all gates above have |
| 416 | # passed, so this never fires for topics the source would not fetch). |
| 417 | ensure_session_ready(topic, config=config, has_domain=bool(explicit_domain)) |
| 418 | # Retry-budget timer starts AFTER the warm-up: a slow Chrome harvest must |
| 419 | # not consume the hint-retry budget when the info call itself was fast. |
| 420 | started = time.monotonic() |
| 421 | if explicit_domain: |
| 422 | identifier = explicit_domain |
| 423 | else: |
| 424 | identifier = _company_identifier(topic) |
| 425 | if not _DOMAIN_RE.search(topic.lower()): |
| 426 | # No domain token in the topic: Trustpilot pages are keyed by |
| 427 | # domain, so resolve name -> domain before the info lookup. |
| 428 | identifier = _search_domain(topic) or identifier |
| 429 | _log(f"info '{identifier}'") |
| 430 | data = _run_cli(_build_info_args(identifier), timeout=SEARCH_TIMEOUT) |
| 431 | if ("error" in data or not data) and explicit_domain and domain_is_hint: |
| 432 | # The auto-resolved hint missed. Only user-set flags are |
| 433 | # verbatim-final; a hint falls through to the search resolution. |
| 434 | # Skip the retry chain when the first lookup already consumed a full |
| 435 | # single-call budget (hung CLI) -- one stream must not chain three |
| 436 | # sequential SEARCH_TIMEOUT-bound subprocesses. |
| 437 | if time.monotonic() - started < SEARCH_TIMEOUT: |
| 438 | resolved = _search_domain(topic) |
| 439 | if resolved and resolved != identifier: |
| 440 | _log(f"hint '{identifier}' missed; retrying via search as '{resolved}'") |
| 441 | data = _run_cli(_build_info_args(resolved), timeout=SEARCH_TIMEOUT) |
| 442 | if "error" in data or not data: |
| 443 | return {"results": []} |
| 444 | return {"results": [data]} |
| 445 | |
| 446 | |
| 447 | def _coerce_float(value: Any) -> Optional[float]: |
| 448 | try: |
| 449 | return float(value) |
| 450 | except (TypeError, ValueError): |
| 451 | return None |
| 452 | |
| 453 | |
| 454 | def _coerce_int(value: Any) -> Optional[int]: |
| 455 | try: |
| 456 | return int(value) |
| 457 | except (TypeError, ValueError): |
| 458 | return None |
| 459 | |
| 460 | |
| 461 | def parse_trustpilot_response( |
| 462 | response: Dict[str, Any], |
| 463 | query: str = "", |
| 464 | ) -> List[Dict[str, Any]]: |
| 465 | """Parse a Trustpilot ``info`` envelope into a single normalized item. |
| 466 | |
| 467 | The AI summary is the body (it already balances positive and negative |
| 468 | sentiment). TrustScore and review count feed engagement and metadata. |
| 469 | Returns dicts ready for ``normalize._normalize_trustpilot``. |
| 470 | """ |
| 471 | raw = response.get("results") if isinstance(response, dict) else None |
| 472 | if not isinstance(raw, list) or not raw: |
| 473 | return [] |
| 474 | info = raw[0] |
| 475 | if not isinstance(info, dict): |
| 476 | return [] |
| 477 | |
| 478 | resolved_name = str(info.get("name") or info.get("displayName") or "").strip() |
| 479 | ai_summary = str(info.get("aiSummary") or info.get("summary") or "").strip() |
| 480 | trust_score = _coerce_float(info.get("trustScore") or info.get("score")) |
| 481 | review_count = _coerce_int( |
| 482 | info.get("reviewCount") or info.get("numberOfReviews") or info.get("total") |
| 483 | ) |
| 484 | url = str(info.get("url") or "").strip() |
| 485 | domain = str(info.get("domain") or info.get("identifyingName") or "").strip() |
| 486 | if not url and domain: |
| 487 | url = f"https://www.trustpilot.com/review/{domain}" |
| 488 | |
| 489 | # Require substantive content from the company record itself; do not |
| 490 | # fabricate an item from the query alone when the CLI returned nothing. |
| 491 | if not resolved_name and not ai_summary and trust_score is None and review_count is None: |
| 492 | return [] |
| 493 | |
| 494 | name = resolved_name or query.strip() |
| 495 | |
| 496 | title = f"{name} on Trustpilot" if name else "Trustpilot reviews" |
| 497 | if trust_score is not None: |
| 498 | title = f"{name}: TrustScore {trust_score}" if name else title |
| 499 | |
| 500 | engagement: Dict[str, float | int] = {} |
| 501 | if review_count is not None: |
| 502 | engagement["reviews"] = review_count |
| 503 | if trust_score is not None: |
| 504 | engagement["trustScore"] = trust_score |
| 505 | |
| 506 | relevance = token_overlap_relevance(query, name) if (query and name) else 0.7 |
| 507 | |
| 508 | why = "Trustpilot brand sentiment" |
| 509 | if trust_score is not None and review_count is not None: |
| 510 | why = f"Trustpilot: TrustScore {trust_score} across {review_count} reviews" |
| 511 | elif trust_score is not None: |
| 512 | why = f"Trustpilot: TrustScore {trust_score}" |
| 513 | |
| 514 | return [ |
| 515 | { |
| 516 | "id": domain or name or "trustpilot", |
| 517 | "title": title, |
| 518 | "url": url, |
| 519 | "summary": ai_summary, |
| 520 | "name": name, |
| 521 | "trustScore": trust_score, |
| 522 | "reviewCount": review_count, |
| 523 | "date": dates.get_date_range(1)[0], |
| 524 | "engagement": engagement, |
| 525 | "relevance": round(min(1.0, max(0.4, relevance)), 2), |
| 526 | "why_relevant": why, |
| 527 | } |
| 528 | ] |
| 529 |