#!/usr/bin/env python3 """Compare saved SearXNG/Tavily/Brave JSON responses offline. This utility never calls a search engine, opens a browser or reads credentials. Do NOT supply logs or JSON with confidential user queries unless authorized; output prints only derived aggregate counts. """ from __future__ import annotations import argparse from collections import Counter import json from pathlib import Path import tempfile from urllib.parse import urlsplit MAX_INPUT_BYTES = 2_000_000 MAX_RESULTS_PER_PROVIDER = 200 def _validate_payload(payload: object, provider: str) -> list[dict]: if not isinstance(payload, dict): raise ValueError("Each response must be a JSON object") if provider in ("searxng", "tavily"): raw = payload.get("results") elif provider == "brave": web = payload.get("web") raw = web.get("results") if isinstance(web, dict) else None else: raise ValueError("Unknown provider") if not isinstance(raw, list): raise ValueError(f"Expected a results array for {provider}") if len(raw) > MAX_RESULTS_PER_PROVIDER: raise ValueError("Result count exceeded 200") if any(not isinstance(item, dict) for item in raw): raise ValueError("All results must be objects") return raw def normalize(payload: object, provider: str) -> dict: raw = _validate_payload(payload, provider) cleaned = [] invalid_links = 0 for entry in raw: url = entry.get("url", "") if not isinstance(url, str): invalid_links += 1 continue parsed = urlsplit(url.strip()) if parsed.scheme not in ("http", "https") or not parsed.hostname: invalid_links += 1 continue title = entry.get("title") summary = entry.get("content" if provider != "brave" else "description") title = title if isinstance(title, str) else "" summary = summary if isinstance(summary, str) else "" cleaned.append({ "url_key": ( parsed.hostname.lower().removeprefix("www."), parsed.path.rstrip("/") or "/", parsed.query, ), "title_length": len(title), "snippet_chars": len(summary), }) keys = [x["url_key"] for x in cleaned] return { "provider": provider, "returned_rows": len(raw), "valid_http_urls": len(cleaned), "invalid_urls": invalid_links, "unique_urls": len(set(keys)), "duplicate_url_rows": len(keys) - len(set(keys)), "snippet_characters": sum(x["snippet_chars"] for x in cleaned), "title_characters": sum(x["title_length"] for x in cleaned), "_url_keys": set(keys), } def summarize(inputs: dict[str, object]) -> dict: if set(inputs) != {"searxng", "tavily", "brave"}: raise ValueError("Exactly three provider JSON objects are required") normalized = {name: normalize(payload, name) for name, payload in inputs.items()} overlap = {} for one, two in (("searxng", "tavily"), ("searxng", "brave"), ("tavily", "brave")): overlap[f"{one}_and_{two}"] = len( normalized[one]["_url_keys"] & normalized[two]["_url_keys"] ) metrics = {} for provider, data in normalized.items(): metrics[provider] = {k: v for k, v in data.items() if not k.startswith("_")} return { "status": "OFFLINE_RESPONSE_SHAPE_AND_SIZE_REPORT", "provider_metrics": metrics, "shared_url_counts": overlap, "does_not_measure": ["network_latency", "search_relevance", "accuracy", "price", "LLM_token_cost"], "actual_api_calls": 0, "warning": "Char counts are NOT token counts; URL overlap is NOT retrieval quality.", } def _read_json(path: str) -> dict: p = Path(path).expanduser() if not p.is_file(): raise ValueError("Response file does not exist") if p.stat().st_size > MAX_INPUT_BYTES: raise ValueError("Response exceeds 2 MB") with p.open("r", encoding="utf-8") as f: return json.load(f) def self_test() -> dict: synthetic = { "searxng": {"results": [ {"url": "https://example.org/a", "title": "One", "content": "Hello"}, {"url": "https://example.org/a", "title": "One", "content": "Repeat"}, {"url": "ftp://example.org/no", "title": "bad"}, ]}, "tavily": {"results": [ {"url": "https://example.org/a", "title": "One", "content": "Hello Tavily"}, {"url": "https://example.net/b", "title": "Two", "content": "Hello"}, ]}, "brave": {"web": {"results": [ {"url": "https://example.net/b", "title": "Two", "description": "Hello Brave"}, ]}}, } result = summarize(synthetic) m = result["provider_metrics"] checks = [] assert m["searxng"]["returned_rows"] == 3 checks.append("searxng_result_shape") assert m["searxng"]["invalid_urls"] == 1 checks.append("reject_non_http") assert m["searxng"]["duplicate_url_rows"] == 1 checks.append("duplicate_url_detection") assert m["tavily"]["unique_urls"] == 2 checks.append("tavily_result_shape") assert m["brave"]["snippet_characters"] == len("Hello Brave") checks.append("brave_response_shape") assert result["shared_url_counts"]["searxng_and_tavily"] == 1 checks.append("first_overlap") assert result["shared_url_counts"]["tavily_and_brave"] == 1 checks.append("second_overlap") try: normalize({"unexpected": 1}, "brave") except ValueError: checks.append("reject_missing_web_results") else: raise AssertionError("Malformed payload accepted") with tempfile.TemporaryDirectory(prefix="provider-compare-") as tmp: p = Path(tmp) / "sample.json" p.write_text(json.dumps(synthetic["tavily"]), encoding="utf-8") assert _read_json(str(p)) == synthetic["tavily"] checks.append("local_json_read") assert len(checks) == 9 return {"status": "PASS", "synthetic_checks": checks, "count": len(checks), "provider_api_calls": 0, "private_files_read": 0} def main() -> int: ap = argparse.ArgumentParser(description=__doc__) ap.add_argument("--self-test", action="store_true") ap.add_argument("--searxng", help="Saved SearXNG JSON response") ap.add_argument("--tavily", help="Saved Tavily JSON response") ap.add_argument("--brave", help="Saved Brave Web Search JSON response") args = ap.parse_args() if args.self_test: print(json.dumps(self_test(), indent=2)) return 0 if not (args.searxng and args.tavily and args.brave): ap.error("Supply all three local response files, or --self-test") try: data = {name: _read_json(getattr(args, name)) for name in ("searxng", "tavily", "brave")} print(json.dumps(summarize(data), indent=2)) return 0 except (ValueError, json.JSONDecodeError, OSError): # Avoid echoing input content or paths that may contain private details. print(json.dumps({"status": "INVALID_INPUT_OR_IO_ERROR"})) return 2 if __name__ == "__main__": raise SystemExit(main())