Web Search Plus

# Web Search Plus v3.1.0
# Copy to .env and fill in any providers you want.
# One credential (or SEARXNG_INSTANCE_URL) is enough to start.
# More providers = better routing/fallback/extraction coverage.

export SERPER_API_KEY=
# https://serper.dev — Google-style general web, shopping, local

export BRAVE_API_KEY=
# https://brave.com/search/api/ — independent web index
# Optional Brave tuning:
# export BRAVE_COUNTRY=US
# export BRAVE_SEARCH_LANG=en
# export BRAVE_SAFESEARCH=moderate

export TAVILY_API_KEY=
# https://tavily.com — research-focused search and extraction

export QUERIT_API_KEY=
# https://querit.ai — multilingual / international AI search

export LINKUP_API_KEY=
# https://linkup.so — source-grounded search and extraction

export EXA_API_KEY=
# https://exa.ai — semantic search, discovery, deep/deep-reasoning, extraction

export FIRECRAWL_API_KEY=
# https://www.firecrawl.dev/app/api-keys — search + extraction

export KILOCODE_API_KEY=
# https://kilo.ai — enables Perplexity Sonar via Kilo gateway
# Optional if you use direct Perplexity auth instead:
# export PERPLEXITY_API_KEY=

export YOU_API_KEY=
# https://api.you.com — current-web and content extraction

export SEARXNG_INSTANCE_URL=
# Self-hosted SearXNG URL, e.g. http://localhost:8080 or https://search.example.com
# Optional override only if you intentionally use a private-network instance:
# export SEARXNG_ALLOW_PRIVATE=1

export SERPBASE_API_KEY=
# https://serpbase.dev — low-cost Google SERP API with prepaid credits.
# Explicit/fallback-only by design (not in default auto-routing). Use:
#   python3 scripts/search.py -p serpbase -q "your query"
# OR add "serpbase" to auto_routing.provider_priority in config.json to opt-in.

.env
.env.*
!.env.example
config.json
!config.example.json
__pycache__/
*.pyc
*.log
.DS_Store
.cache/
.data/
.vercel/

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "$comment": "Web Search Plus v3.1.0 example config — search routing plus extraction-capable provider settings (incl. opt-in SerpBase)",
  "defaults": {
    "provider": "serper",
    "max_results": 5
  },
  "auto_routing": {
    "enabled": true,
    "fallback_provider": "serper",
    "provider_priority": [
      "tavily",
      "linkup",
      "querit",
      "exa",
      "firecrawl",
      "perplexity",
      "brave",
      "serper",
      "you",
      "searxng"
    ],
    "disabled_providers": [],
    "confidence_threshold": 0.3
  },
  "serper": {
    "country": "us",
    "language": "en",
    "type": "search"
  },
  "brave": {
    "country": "US",
    "search_lang": "en",
    "safesearch": "moderate"
  },
  "tavily": {
    "depth": "advanced",
    "topic": "general"
  },
  "querit": {
    "base_url": "https://api.querit.ai",
    "base_path": "/v1/search",
    "timeout": 10
  },
  "linkup": {
    "api_url": "https://api.linkup.so/v1/search",
    "depth": "standard",
    "output_type": "searchResults",
    "timeout": 30
  },
  "exa": {
    "type": "neural",
    "depth": "normal",
    "verbosity": "standard"
  },
  "firecrawl": {
    "api_url": "https://api.firecrawl.dev/v2/search",
    "country": "US",
    "timeout": 30000,
    "sources": ["web"],
    "ignore_invalid_urls": false
  },
  "perplexity": {
    "api_url": "https://api.kilo.ai/api/gateway/chat/completions",
    "model": "perplexity/sonar-pro"
  },
  "you": {
    "country": "us",
    "safesearch": "moderate"
  },
  "searxng": {
    "instance_url": null,
    "safesearch": 0,
    "engines": null,
    "language": "en"
  },
  "serpbase": {
    "$comment": "SerpBase is explicit/fallback-only. To include in auto-routing, append 'serpbase' to auto_routing.provider_priority.",
    "api_url": "https://api.serpbase.com/search",
    "timeout": 30
  }
}

MIT License

Copyright (c) 2026 Robby

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

{
  "name": "web-search-plus",
  "version": "3.1.0",
  "description": "Unified multi-provider web search and URL extraction skill with intelligent auto-routing across Serper, Brave, Tavily, Querit, Linkup, Exa, Firecrawl, Perplexity, You.com, SearXNG, and SerpBase.",
  "keywords": [
    "openclaw",
    "skill",
    "search",
    "web-search",
    "serper",
    "brave",
    "tavily",
    "linkup",
    "exa",
    "firecrawl",
    "you",
    "you.com",
    "serpbase",
    "google-search",
    "research",
    "semantic-search",
    "ai-agent",
    "auto-routing",
    "smart-routing",
    "multi-provider",
    "shopping",
    "product-search",
    "similar-sites",
    "company-discovery",
    "rag",
    "real-time",
    "free-tier",
    "api-aggregator",
    "querit",
    "multilingual-search",
    "web-extract",
    "content-extraction",
    "firecrawl",
    "brave-search",
    "linkup"
  ],
  "author": "robbyczgw-cla",
  "license": "MIT",
  "repository": {
    "type": "git",
    "url": "https://github.com/robbyczgw-cla/web-search-plus.git"
  },
  "homepage": "https://clawhub.ai/robbyczgw-cla/web-search-plus",
  "bugs": {
    "url": "https://github.com/robbyczgw-cla/web-search-plus/issues"
  },
  "files": [
    "SKILL.md",
    "README.md",
    "scripts/",
    ".env.example",
    "config.example.json",
    "CHANGELOG.md"
  ],
  "clawhub": {
    "skill": true,
    "triggers": [
      "search",
      "find",
      "look up",
      "research"
    ],
    "capabilities": [
      "web-search",
      "image-search",
      "semantic-search",
      "multi-provider",
      "web-extract"
    ],
    "requirements": {
      "bins": [
        "python3",
        "bash"
      ],
      "env": {
        "SERPER_API_KEY": "optional",
        "BRAVE_API_KEY": "optional",
        "TAVILY_API_KEY": "optional",
        "QUERIT_API_KEY": "optional",
        "LINKUP_API_KEY": "optional",
        "EXA_API_KEY": "optional",
        "FIRECRAWL_API_KEY": "optional",
        "PERPLEXITY_API_KEY": "optional \u2014 direct Perplexity provider credential",
        "KILOCODE_API_KEY": "optional \u2014 alternative Perplexity provider via Kilo Gateway",
        "YOU_API_KEY": "optional",
        "SEARXNG_INSTANCE_URL": "optional"
      }
    }
  }
}

import os
import sys
from pathlib import Path


def is_docker() -> bool:
    """Detect if running inside a Docker container.
    
    Checks multiple signals:
    - /.dockerenv file
    - /proc/1/cgroup containing 'docker' or 'kubepods'
    - DOCKER environment variable
    """
    if os.path.exists("/.dockerenv"):
        return True
    
    try:
        with open("/proc/1/cgroup", "r") as f:
            cgroup = f.read().lower()
            if "docker" in cgroup or "kubepods" in cgroup:
                return True
    except (FileNotFoundError, PermissionError):
        pass
    
    if os.environ.get("DOCKER"):
        return True
    
    return False


def get_searxng_url() -> str:
    """Get the appropriate SearXNG URL based on environment.
    
    Returns:
        - http://172.17.0.1:8080 if running in Docker
        - http://127.0.0.1:8080 if running on host
    """
    if is_docker():
        return "http://172.17.0.1:8080"
    return "http://127.0.0.1:8080"


def ensure_searxng_env():
    """Set SEARXNG_INSTANCE_URL environment variable if not already set."""
    if "SEARXNG_INSTANCE_URL" not in os.environ:
        os.environ["SEARXNG_INSTANCE_URL"] = get_searxng_url()


if __name__ == "__main__":
    print(f"In Docker: {is_docker()}")
    print(f"SearXNG URL: {get_searxng_url()}")

#!/usr/bin/env python3
"""
Web Search Plus — URL extraction with automatic provider fallback.
Supports: Firecrawl, Linkup, Tavily, Exa, You.com

Usage:
    python3 scripts/extract.py --url https://example.com
    python3 scripts/extract.py --url https://example.com --provider firecrawl --format markdown
    python3 scripts/extract.py --url https://example.com --url https://example.org --include-images
"""

import argparse
import gzip
import json
import os
import sys
import zlib
from pathlib import Path
from typing import Any, Dict, List, Optional
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen

EXTRACT_PROVIDER_PRIORITY = ["firecrawl", "linkup", "tavily", "exa", "you"]


def _load_env_file() -> None:
    env_paths = [Path(__file__).parent.parent / ".env", Path(__file__).parent / ".env"]
    for env_path in env_paths:
        if not env_path.exists():
            continue
        with open(env_path, encoding="utf-8") as handle:
            for line in handle:
                line = line.strip()
                if line and not line.startswith("#") and "=" in line:
                    if line.startswith("export "):
                        line = line[7:]
                    key, _, value = line.partition("=")
                    key = key.strip()
                    value = value.strip().strip('"').strip("'")
                    if key and key not in os.environ:
                        os.environ[key] = value


_load_env_file()


def _response_header(response, name: str, default: str = "") -> str:
    """Get response header safely (works with HTTPError too)."""
    try:
        val = response.getheader(name) if hasattr(response, "getheader") else response.headers.get(name)
        return val or default
    except Exception:
        return default


def _read_response_body(response) -> bytes:
    """Read response body with gzip/deflate decompression support."""
    raw = response.read()
    encoding = _response_header(response, "Content-Encoding", "").lower().strip()

    if encoding in ("gzip", "x-gzip"):
        try:
            return gzip.decompress(raw)
        except Exception:
            if raw.startswith(b"\x1f\x8b"):
                return gzip.decompress(raw)
            return raw
    elif encoding == "deflate":
        try:
            return zlib.decompress(raw, -zlib.MAX_WBITS)
        except zlib.error:
            try:
                return zlib.decompress(raw)
            except zlib.error:
                return raw
    elif encoding == "br":
        try:
            import brotli
            return brotli.decompress(raw)
        except ImportError:
            raise RuntimeError(
                "Brotli-encoded response received but brotli library not installed. Install with: pip install brotli"
            )
        except Exception:
            raise RuntimeError("Failed to decompress brotli-encoded response")

    return raw


def request_json(url: str, init: Dict[str, Any], timeout: int = 30) -> Any:
    body = init.get("body")
    data = body.encode("utf-8") if isinstance(body, str) else body
    req = Request(url, data=data, method=init.get("method", "GET"))
    for key, value in (init.get("headers") or {}).items():
        req.add_header(key, value)
    try:
        with urlopen(req, timeout=max(1, timeout)) as response:
            text = _read_response_body(response).decode("utf-8")
            return json.loads(text) if text else {}
    except HTTPError as exc:
        payload = exc.read().decode("utf-8", errors="replace") if hasattr(exc, "read") else ""
        try:
            data = json.loads(payload) if payload else {}
        except json.JSONDecodeError:
            data = {}
        message = data.get("error") or data.get("message") or data.get("detail") or data.get("warning") or f"HTTP {exc.code}"
        raise RuntimeError(str(message)) from exc
    except URLError as exc:
        raise RuntimeError(str(exc.reason)) from exc


def title_from_url(url: str) -> str:
    try:
        from urllib.parse import urlparse
        parsed = urlparse(url)
        last_segment = [part for part in parsed.path.split("/") if part]
        return (last_segment[-1] if last_segment else parsed.hostname) or url
    except Exception:
        return url


def normalize_images(images: Any) -> Optional[List[Dict[str, str]]]:
    if not isinstance(images, list):
        return None
    normalized = []
    for image in images:
        if not image:
            continue
        if isinstance(image, str):
            normalized.append({"url": image})
        elif isinstance(image, dict) and isinstance(image.get("url"), str) and image.get("url"):
            item = {"url": image["url"]}
            if isinstance(image.get("alt"), str) and image.get("alt"):
                item["alt"] = image["alt"]
            normalized.append(item)
    return normalized or None


def normalize_result(provider: str, url: str, title: str = "", content: str = "", raw_content: Optional[str] = None, **extra: Any) -> Dict[str, Any]:
    result = {
        "url": url,
        "title": title or title_from_url(url),
        "content": content or "",
        "raw_content": raw_content if raw_content is not None else content or "",
        "provider": provider,
    }
    for key, value in extra.items():
        if value is not None:
            result[key] = value
    return result


def get_extract_api_key(provider: str) -> Optional[str]:
    env_map = {
        "firecrawl": "FIRECRAWL_API_KEY",
        "linkup": "LINKUP_API_KEY",
        "tavily": "TAVILY_API_KEY",
        "exa": "EXA_API_KEY",
        "you": "YOU_API_KEY",
    }
    return os.environ.get(env_map[provider])


def extract_firecrawl(urls: List[str], api_key: str, output_format: str = "markdown", include_images: bool = False, include_raw_html: bool = False, render_js: bool = False, api_url: str = "https://api.firecrawl.dev/v2/scrape", timeout: int = 60) -> Dict[str, Any]:
    formats = ["html"] if output_format == "html" else ["markdown"]
    if include_raw_html and "html" not in formats:
        formats.append("html")
    results = []
    for url in urls:
        try:
            body: Dict[str, Any] = {"url": url, "formats": formats}
            if render_js:
                body["waitFor"] = 1000
            data = request_json(api_url, {
                "method": "POST",
                "headers": {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
                "body": json.dumps(body),
            }, timeout)
            if data.get("success") is False:
                results.append(normalize_result("firecrawl", url, error=str(data.get("error") or data.get("warning") or "Firecrawl scrape failed")))
                continue
            payload = data.get("data") if isinstance(data.get("data"), dict) else data
            metadata = payload.get("metadata") if isinstance(payload.get("metadata"), dict) else {}
            final_url = metadata.get("sourceURL") or metadata.get("url") or url
            title = metadata.get("title") or ""
            markdown = str(payload.get("markdown") or "")
            html = str(payload.get("html") or payload.get("rawHtml") or "")
            content = html if output_format == "html" else (markdown or html)
            images = None
            if include_images:
                seen = set()
                parsed = []
                og_image = metadata.get("ogImage") or metadata.get("og:image")
                if isinstance(og_image, str) and og_image and og_image not in seen:
                    parsed.append({"alt": "og:image", "url": og_image})
                    seen.add(og_image)
                import re
                for match in re.finditer(r'!\[([^\]]*)\]\(([^)]+)\)', markdown):
                    image_url = match.group(2)
                    if image_url and image_url not in seen:
                        item = {"url": image_url}
                        if match.group(1):
                            item["alt"] = match.group(1)
                        parsed.append(item)
                        seen.add(image_url)
                images = parsed or None
            results.append(normalize_result("firecrawl", final_url, title, content, content, raw_html=html or None, images=images, metadata=metadata or None))
        except Exception as exc:
            results.append(normalize_result("firecrawl", url, error=str(exc)))
    return {"provider": "firecrawl", "results": results}


def extract_linkup(urls: List[str], api_key: str, output_format: str = "markdown", include_images: bool = False, include_raw_html: bool = False, render_js: bool = False, api_url: str = "https://api.linkup.so/v1/fetch", timeout: int = 30) -> Dict[str, Any]:
    results = []
    for url in urls:
        try:
            data = request_json(api_url, {
                "method": "POST",
                "headers": {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
                "body": json.dumps({
                    "url": url,
                    "extractImages": include_images,
                    "includeRawHtml": include_raw_html or output_format == "html",
                    "renderJs": render_js,
                }),
            }, timeout)
            if data.get("error"):
                results.append(normalize_result("linkup", url, error=str(data["error"])))
                continue
            markdown = str(data.get("markdown") or "")
            raw_html = str(data.get("rawHtml") or data.get("raw_html") or "")
            content = raw_html if output_format == "html" else (markdown or raw_html)
            results.append(normalize_result("linkup", url, content=content, raw_content=content, raw_html=raw_html or None, images=normalize_images(data.get("images")) if include_images else None, metadata=data.get("metadata") if isinstance(data.get("metadata"), dict) else None))
        except Exception as exc:
            results.append(normalize_result("linkup", url, error=str(exc)))
    return {"provider": "linkup", "results": results}


def extract_tavily(urls: List[str], api_key: str, output_format: str = "markdown", include_images: bool = False, include_raw_html: bool = False, render_js: bool = False, api_url: str = "https://api.tavily.com/extract", timeout: int = 30) -> Dict[str, Any]:
    del output_format, include_raw_html, render_js
    data = request_json(api_url, {
        "method": "POST",
        "headers": {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
        "body": json.dumps({"urls": urls, "include_images": include_images}),
    }, timeout)
    results = []
    for item in data.get("results") or []:
        content = str(item.get("raw_content") or item.get("content") or "")
        results.append(normalize_result("tavily", str(item.get("url") or ""), str(item.get("title") or ""), content, content, images=normalize_images(item.get("images")) if include_images else None, metadata=item.get("metadata") if isinstance(item.get("metadata"), dict) else None))
    for failed in data.get("failed_results") or []:
        results.append(normalize_result("tavily", str(failed.get("url") or ""), error=str(failed.get("error") or "Tavily extract failed")))
    return {"provider": "tavily", "results": results}


def extract_exa(urls: List[str], api_key: str, output_format: str = "markdown", include_images: bool = False, include_raw_html: bool = False, render_js: bool = False, api_url: str = "https://api.exa.ai/contents", timeout: int = 30) -> Dict[str, Any]:
    del output_format, include_raw_html, render_js
    data = request_json(api_url, {
        "method": "POST",
        "headers": {"x-api-key": api_key, "Content-Type": "application/json"},
        "body": json.dumps({"urls": urls, "text": True}),
    }, timeout)
    results = []
    for item in data.get("results") or []:
        url = str(item.get("url") or item.get("id") or "")
        content = str(item.get("text") or item.get("summary") or "")
        metadata = {}
        for src, dest in [("summary", "summary"), ("highlights", "highlights"), ("publishedDate", "published_date"), ("author", "author"), ("favicon", "favicon")]:
            if item.get(src) is not None:
                metadata[dest] = item.get(src)
        images = [{"alt": "image", "url": str(item.get("image"))}] if include_images and item.get("image") else None
        results.append(normalize_result("exa", url, str(item.get("title") or ""), content, content, images=images, metadata=metadata or None))
    return {"provider": "exa", "results": results}


def extract_you(urls: List[str], api_key: str, output_format: str = "markdown", include_images: bool = False, include_raw_html: bool = False, render_js: bool = False, api_url: str = "https://ydc-index.io/v1/contents", timeout: int = 30) -> Dict[str, Any]:
    del include_images, render_js
    formats = ["html"] if output_format == "html" else ["markdown"]
    if include_raw_html and "html" not in formats:
        formats.append("html")
    if "metadata" not in formats:
        formats.append("metadata")
    data = request_json(api_url, {
        "method": "POST",
        "headers": {"X-API-Key": api_key, "Content-Type": "application/json"},
        "body": json.dumps({"urls": urls, "formats": formats, "crawl_timeout": max(1, min(timeout, 60))}),
    }, timeout)
    raw_items = data if isinstance(data, list) else data.get("results") or data.get("data") or []
    results = []
    for item in raw_items:
        url = str(item.get("url") or "")
        markdown = str(item.get("markdown") or "")
        html = str(item.get("html") or "")
        content = html if output_format == "html" else (markdown or html)
        results.append(normalize_result("you", url, str(item.get("title") or ""), content, content, raw_html=html or None, metadata=item.get("metadata") if isinstance(item.get("metadata"), dict) else None))
    return {"provider": "you", "results": results}


def extract_plus(urls: List[str], provider: str = "auto", output_format: str = "markdown", include_images: bool = False, include_raw_html: bool = False, render_js: bool = False) -> Dict[str, Any]:
    requested_provider = provider or "auto"
    if not urls:
        return {"provider": requested_provider, "results": [], "error": "No URLs provided", "routing": {"requested_provider": requested_provider}}
    cleaned_urls = [url.strip() for url in urls if isinstance(url, str)]
    invalid_urls = [url for url in cleaned_urls if not url.startswith(("http://", "https://"))]
    if invalid_urls:
        return {"provider": requested_provider, "results": [], "error": f"Invalid URL(s) — must start with http:// or https://: {json.dumps(invalid_urls)}", "routing": {"requested_provider": requested_provider}}
    providers = EXTRACT_PROVIDER_PRIORITY if requested_provider == "auto" else [requested_provider] + [p for p in EXTRACT_PROVIDER_PRIORITY if p != requested_provider]
    errors = []
    for current_provider in providers:
        if current_provider not in EXTRACT_PROVIDER_PRIORITY:
            errors.append({"provider": current_provider, "error": f"Provider {current_provider} does not support extraction"})
            continue
        credential = get_extract_api_key(current_provider)
        if not credential:
            errors.append({"provider": current_provider, "error": "missing_api_key"})
            continue
        try:
            if current_provider == "firecrawl":
                result = extract_firecrawl(cleaned_urls, credential, output_format, include_images, include_raw_html, render_js)
            elif current_provider == "linkup":
                result = extract_linkup(cleaned_urls, credential, output_format, include_images, include_raw_html, render_js)
            elif current_provider == "tavily":
                result = extract_tavily(cleaned_urls, credential, output_format, include_images, include_raw_html, render_js)
            elif current_provider == "exa":
                result = extract_exa(cleaned_urls, credential, output_format, include_images, include_raw_html, render_js)
            else:
                result = extract_you(cleaned_urls, credential, output_format, include_images, include_raw_html, render_js)
            result_list = result.get("results") or []
            if result_list and all(item.get("error") for item in result_list):
                errors.append({"provider": current_provider, "error": "all_urls_failed", "details": [item.get("error") for item in result_list]})
                continue
            result["routing"] = {
                "provider": current_provider,
                "requested_provider": requested_provider,
                "fallback_used": bool(errors),
                "fallback_errors": errors,
            }
            return result
        except Exception as exc:
            errors.append({"provider": current_provider, "error": str(exc)})
    return {
        "provider": requested_provider,
        "results": [],
        "error": "All extraction providers failed",
        "fallback_errors": errors,
        "routing": {"requested_provider": requested_provider, "fallback_used": bool(errors), "fallback_errors": errors},
    }


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Extract URL content with automatic provider fallback")
    parser.add_argument("--url", dest="urls", action="append", help="URL to extract (repeatable)")
    parser.add_argument("--provider", default="auto", choices=["auto"] + EXTRACT_PROVIDER_PRIORITY)
    parser.add_argument("--format", default="markdown", choices=["markdown", "html"])
    parser.add_argument("--include-images", action="store_true")
    parser.add_argument("--include-raw-html", action="store_true")
    parser.add_argument("--render-js", action="store_true")
    parser.add_argument("--compact", action="store_true", help="Print compact JSON")
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    result = extract_plus(args.urls or [], args.provider, args.format, args.include_images, args.include_raw_html, args.render_js)
    if args.compact:
        print(json.dumps(result, ensure_ascii=False))
    else:
        print(json.dumps(result, ensure_ascii=False, indent=2))
    if result.get("error"):
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Web Search Plus - Interactive Setup Wizard
==========================================

Runs on first use (when no config.json exists) to configure providers and API keys.
Creates config.json with your settings. API keys are stored locally only.

Usage:
    python3 scripts/setup.py          # Interactive setup
    python3 scripts/setup.py --reset  # Reset and reconfigure
"""

import json
import os
import sys
from pathlib import Path

# ANSI colors for terminal output
class Colors:
    HEADER = '\033[95m'
    BLUE = '\033[94m'
    CYAN = '\033[96m'
    GREEN = '\033[92m'
    YELLOW = '\033[93m'
    RED = '\033[91m'
    BOLD = '\033[1m'
    DIM = '\033[2m'
    RESET = '\033[0m'

def color(text: str, c: str) -> str:
    """Wrap text in color codes."""
    return f"{c}{text}{Colors.RESET}"

def print_header():
    """Print the setup wizard header."""
    print()
    print(color("╔════════════════════════════════════════════════════════════╗", Colors.CYAN))
    print(color("║          🔍 Web Search Plus - Setup Wizard                 ║", Colors.CYAN))
    print(color("╚════════════════════════════════════════════════════════════╝", Colors.CYAN))
    print()
    print(color("This wizard will help you configure your search providers.", Colors.DIM))
    print(color("API keys are stored locally in config.json (gitignored).", Colors.DIM))
    print()

def print_provider_info():
    """Print information about each provider."""
    print(color("📚 Available Providers:", Colors.BOLD))
    print()

    providers = [
        {
            "name": "Serper",
            "emoji": "🔎",
            "best_for": "Google-style general web, shopping, local, news",
            "free_tier": "2,500 queries/month",
            "signup": "https://serper.dev",
            "strengths": ["Shopping and local intent", "Knowledge Graph", "Broad fallback coverage"]
        },
        {
            "name": "Brave",
            "emoji": "🦁",
            "best_for": "Independent general web index, broad factual queries",
            "free_tier": "$5/mo free credits",
            "signup": "https://brave.com/search/api/",
            "strengths": ["Independent index", "Good generic-web fallback", "Pairs well with Serper tie-breaking"]
        },
        {
            "name": "Tavily",
            "emoji": "📖",
            "best_for": "Research, explanations, in-depth analysis",
            "free_tier": "1,000 queries/month",
            "signup": "https://tavily.com",
            "strengths": ["Research intent", "Full page content", "Academic-friendly"]
        },
        {
            "name": "Querit",
            "emoji": "🌍",
            "best_for": "Multilingual, international, recency-aware AI search",
            "free_tier": "Varies",
            "signup": "https://querit.ai",
            "strengths": ["Cross-language queries", "International updates", "AI-ready metadata"]
        },
        {
            "name": "Linkup",
            "emoji": "🔗",
            "best_for": "Source-grounded search, citations, evidence-first retrieval",
            "free_tier": "€5 free monthly credits",
            "signup": "https://linkup.so",
            "strengths": ["Citation-heavy queries", "Grounding", "Good companion to extraction"]
        },
        {
            "name": "Exa",
            "emoji": "🧠",
            "best_for": "Semantic search, discovery, deep and deep-reasoning modes",
            "free_tier": "1,000 queries/month",
            "signup": "https://exa.ai",
            "strengths": ["Semantic discovery", "Similar pages", "Deep synthesis"]
        },
        {
            "name": "Firecrawl",
            "emoji": "🔥",
            "best_for": "Search with scrape-ready content and extraction-friendly metadata",
            "free_tier": "Free plan credits",
            "signup": "https://www.firecrawl.dev/app/api-keys",
            "strengths": ["Search + scrape metadata", "Recency filters", "Great extraction fallback"]
        },
        {
            "name": "Perplexity via Kilo",
            "emoji": "⚡",
            "best_for": "Direct answers with citations",
            "free_tier": "Depends on Kilo plan",
            "signup": "https://kilo.ai",
            "strengths": ["Answer-first output", "Current status questions", "Citation support"]
        },
        {
            "name": "You.com",
            "emoji": "🤖",
            "best_for": "RAG applications, real-time info, LLM-ready snippets",
            "free_tier": "Limited free tier",
            "signup": "https://api.you.com",
            "strengths": ["LLM-ready snippets", "Combined web + news", "Live content API"]
        },
        {
            "name": "SearXNG",
            "emoji": "🔒",
            "best_for": "Privacy-first search, multi-source aggregation, $0 API cost",
            "free_tier": "FREE (self-hosted)",
            "signup": "https://docs.searxng.org/admin/installation.html",
            "strengths": ["Privacy-preserving", "70+ search engines", "Self-hosted control"]
        }
    ]

    for p in providers:
        print(f"  {p['emoji']} {color(p['name'], Colors.BOLD)}")
        print(f"     Best for: {color(p['best_for'], Colors.GREEN)}")
        print(f"     Free tier: {p['free_tier']}")
        print(f"     Sign up: {color(p['signup'], Colors.BLUE)}")
        print()

def ask_yes_no(prompt: str, default: bool = True) -> bool:
    """Ask a yes/no question."""
    suffix = "[Y/n]" if default else "[y/N]"
    while True:
        response = input(f"{prompt} {color(suffix, Colors.DIM)}: ").strip().lower()
        if response == "":
            return default
        if response in ("y", "yes"):
            return True
        if response in ("n", "no"):
            return False
        print(color("  Please enter 'y' or 'n'", Colors.YELLOW))

def ask_choice(prompt: str, options: list, default: str = None) -> str:
    """Ask user to choose from a list of options."""
    print(f"\n{prompt}")
    for i, opt in enumerate(options, 1):
        marker = color("→", Colors.GREEN) if opt == default else " "
        print(f"  {marker} {i}. {opt}")

    while True:
        hint = f" [default: {default}]" if default else ""
        response = input(f"Enter number (1-{len(options)}){color(hint, Colors.DIM)}: ").strip()

        if response == "" and default:
            return default

        try:
            idx = int(response)
            if 1 <= idx <= len(options):
                return options[idx - 1]
        except ValueError:
            pass

        print(color(f"  Please enter a number between 1 and {len(options)}", Colors.YELLOW))

def ask_api_key(provider: str, signup_url: str) -> str:
    """Ask for an API key with validation."""
    print()
    print(f"  {color(f'Get your {provider} API key:', Colors.DIM)} {color(signup_url, Colors.BLUE)}")

    while True:
        key = input(f"  Enter your {provider} API key: ").strip()

        if not key:
            print(color("    ⚠️  No key entered. This provider will be disabled.", Colors.YELLOW))
            return None

        # Basic validation
        if len(key) < 10:
            print(color("    ⚠️  Key seems too short. Please check and try again.", Colors.YELLOW))
            continue

        # Mask key for confirmation
        masked = key[:4] + "..." + key[-4:] if len(key) > 12 else key[:2] + "..."
        print(color(f"    ✓ Key saved: {masked}", Colors.GREEN))
        return key


def ask_searxng_instance(docs_url: str) -> str:
    """Ask for SearXNG instance URL with connection test."""
    print()
    print(f"  {color('SearXNG is self-hosted. You need your own instance.', Colors.DIM)}")
    print(f"  {color('Setup guide:', Colors.DIM)} {color(docs_url, Colors.BLUE)}")
    print()
    print(f"  {color('Example URLs:', Colors.DIM)}")
    print(f"    • http://localhost:8080 (local Docker)")
    print(f"    • https://searx.your-domain.com (self-hosted)")
    print()

    while True:
        url = input(f"  Enter your SearXNG instance URL: ").strip()

        if not url:
            print(color("    ⚠️  No URL entered. SearXNG will be disabled.", Colors.YELLOW))
            return None

        # Basic URL validation
        if not url.startswith(("http://", "https://")):
            print(color("    ⚠️  URL must start with http:// or https://", Colors.YELLOW))
            continue

        # SSRF protection: validate URL before connecting
        try:
            import ipaddress
            import socket
            from urllib.parse import urlparse as _urlparse
            _parsed = _urlparse(url)
            _hostname = _parsed.hostname or ""
            _blocked = {"169.254.169.254", "metadata.google.internal", "metadata.internal"}
            if _hostname in _blocked:
                print(color(f"    ❌ Blocked: {_hostname} is a cloud metadata endpoint.", Colors.RED))
                continue
            if not os.environ.get("SEARXNG_ALLOW_PRIVATE", "").strip() == "1":
                _resolved = socket.getaddrinfo(_hostname, _parsed.port or 80, proto=socket.IPPROTO_TCP)
                for _fam, _t, _p, _cn, _sa in _resolved:
                    _ip = ipaddress.ip_address(_sa[0])
                    if _ip.is_loopback or _ip.is_private or _ip.is_link_local or _ip.is_reserved:
                        print(color(f"    ❌ Blocked: {_hostname} resolves to private IP {_ip}.", Colors.RED))
                        print(color(f"       Set SEARXNG_ALLOW_PRIVATE=1 if intentional.", Colors.DIM))
                        raise ValueError("private_ip")
        except ValueError as _ve:
            if str(_ve) == "private_ip":
                continue
            raise
        except socket.gaierror:
            print(color(f"    ❌ Cannot resolve hostname: {_hostname}", Colors.RED))
            continue

        # Test connection
        print(color(f"    Testing connection to {url}...", Colors.DIM))
        try:
            import urllib.request
            import urllib.error

            test_url = f"{url.rstrip('/')}/search?q=test&format=json"
            req = urllib.request.Request(
                test_url,
                headers={"User-Agent": "ClawdBot-WebSearchPlus/2.5", "Accept": "application/json"}
            )

            with urllib.request.urlopen(req, timeout=10) as response:
                data = response.read().decode("utf-8")
                import json
                result = json.loads(data)

                # Check if it looks like SearXNG JSON response
                if "results" in result or "query" in result:
                    print(color(f"    ✓ Connection successful! SearXNG instance is working.", Colors.GREEN))
                    return url.rstrip("/")
                else:
                    print(color(f"    ⚠️  Connected but response doesn't look like SearXNG JSON.", Colors.YELLOW))
                    if ask_yes_no("    Use this URL anyway?", default=False):
                        return url.rstrip("/")

        except urllib.error.HTTPError as e:
            if e.code == 403:
                print(color(f"    ⚠️  JSON API is disabled (403 Forbidden).", Colors.YELLOW))
                print(color(f"       Enable JSON in settings.yml: search.formats: [html, json]", Colors.DIM))
            else:
                print(color(f"    ⚠️  HTTP error: {e.code} {e.reason}", Colors.YELLOW))

            if ask_yes_no("    Try a different URL?", default=True):
                continue
            return None

        except urllib.error.URLError as e:
            print(color(f"    ⚠️  Cannot reach instance: {e.reason}", Colors.YELLOW))
            if ask_yes_no("    Try a different URL?", default=True):
                continue
            return None

        except Exception as e:
            print(color(f"    ⚠️  Error: {e}", Colors.YELLOW))
            if ask_yes_no("    Try a different URL?", default=True):
                continue
            return None

def ask_result_count() -> int:
    """Ask for default result count."""
    options = ["3 (fast, minimal)", "5 (balanced - recommended)", "10 (comprehensive)"]
    choice = ask_choice("Default number of results per search?", options, "5 (balanced - recommended)")

    if "3" in choice:
        return 3
    elif "10" in choice:
        return 10
    return 5

def run_setup(skill_dir: Path, force_reset: bool = False):
    """Run the interactive setup wizard."""
    config_path = skill_dir / "config.json"
    example_path = skill_dir / "config.example.json"

    # Check if config already exists
    if config_path.exists() and not force_reset:
        print(color("✓ config.json already exists!", Colors.GREEN))
        print()
        if not ask_yes_no("Do you want to reconfigure?", default=False):
            print(color("Setup cancelled. Your existing config is unchanged.", Colors.DIM))
            return False
        print()

    print_header()
    print_provider_info()

    # Load example config as base
    if example_path.exists():
        with open(example_path) as f:
            config = json.load(f)
    else:
        config = {
            "defaults": {"provider": "serper", "max_results": 5},
            "auto_routing": {"enabled": True, "fallback_provider": "serper"},
            "serper": {},
            "brave": {},
            "tavily": {},
            "querit": {},
            "linkup": {},
            "exa": {},
            "firecrawl": {},
            "perplexity": {},
            "you": {},
            "searxng": {},
            "serpbase": {}
        }

    # Remove any existing API keys from example
    for provider in ["serper", "brave", "tavily", "querit", "linkup", "exa", "firecrawl", "perplexity", "you", "serpbase"]:
        if provider in config:
            config[provider].pop("api_key", None)

    enabled_providers = []

    # ===== Question 1: Which providers to enable =====
    print(color("─" * 60, Colors.DIM))
    print(color("\n📋 Step 1: Choose Your Providers\n", Colors.BOLD))
    print("Select which search providers you want to enable.")
    print(color("(You need at least one API key to use this skill)", Colors.DIM))
    print()

    providers_info = {
        "serper": ("Serper", "https://serper.dev", "Google-style results, shopping, local"),
        "brave": ("Brave", "https://brave.com/search/api/", "Independent general web search"),
        "tavily": ("Tavily", "https://tavily.com", "Research, explanations, analysis"),
        "querit": ("Querit", "https://querit.ai", "Multilingual and international AI search"),
        "linkup": ("Linkup", "https://linkup.so", "Citation and source-grounded search"),
        "exa": ("Exa", "https://exa.ai", "Semantic search, discovery, deep synthesis"),
        "firecrawl": ("Firecrawl", "https://www.firecrawl.dev/app/api-keys", "Search with scrape-ready metadata"),
        "perplexity": ("Perplexity via Kilo", "https://kilo.ai", "Direct answers with citations"),
        "you": ("You.com", "https://api.you.com", "RAG applications, real-time info"),
        "searxng": ("SearXNG", "https://docs.searxng.org/admin/installation.html", "Privacy-first, self-hosted, $0 cost"),
        "serpbase": ("SerpBase", "https://serpbase.dev", "Low-cost Google SERP, prepaid credits (explicit/fallback-only)")
    }

    for provider, (name, url, desc) in providers_info.items():
        print(f"  {color(name, Colors.BOLD)}: {desc}")

        # Special handling for SearXNG
        if provider == "searxng":
            print(color("    Note: SearXNG requires a self-hosted instance (no API key needed)", Colors.DIM))
            if ask_yes_no(f"    Do you have a SearXNG instance?", default=False):
                instance_url = ask_searxng_instance(url)
                if instance_url:
                    if "searxng" not in config:
                        config["searxng"] = {}
                    config["searxng"]["instance_url"] = instance_url
                    enabled_providers.append(provider)
                else:
                    print(color(f"    → {name} disabled (no instance URL)", Colors.DIM))
            else:
                print(color(f"    → {name} skipped (no instance)", Colors.DIM))
        else:
            if ask_yes_no(f"    Enable {name}?", default=True):
                # ===== Question 2: API key for each enabled provider =====
                api_key = ask_api_key(name, url)
                if api_key:
                    config[provider]["api_key"] = api_key
                    enabled_providers.append(provider)
                else:
                    print(color(f"    → {name} disabled (no API key)", Colors.DIM))
            else:
                print(color(f"    → {name} disabled", Colors.DIM))
        print()

    if not enabled_providers:
        print()
        print(color("⚠️  No providers enabled!", Colors.RED))
        print("You need at least one API key to use web-search-plus.")
        print("Run this setup again when you have an API key.")
        return False

    # ===== Question 3: Default provider =====
    print(color("─" * 60, Colors.DIM))
    print(color("\n⚙️  Step 2: Default Settings\n", Colors.BOLD))

    if len(enabled_providers) > 1:
        default_provider = ask_choice(
            "Which provider should be the default for general queries?",
            enabled_providers,
            enabled_providers[0]
        )
    else:
        default_provider = enabled_providers[0]
        print(f"Default provider: {color(default_provider, Colors.GREEN)} (only one enabled)")

    config["defaults"]["provider"] = default_provider
    config["auto_routing"]["fallback_provider"] = default_provider

    # ===== Question 4: Auto-routing =====
    print()
    print(color("Auto-routing", Colors.BOLD) + " automatically picks the best provider for each query:")
    print(color("  • 'weather in Vienna today' → Brave or Serper (generic current web)", Colors.DIM))
    print(color("  • 'find credible sources for X' → Linkup (source-grounded intent)", Colors.DIM))
    print(color("  • 'deep research on X' → Exa deep/deep-reasoning or Tavily", Colors.DIM))
    print()

    auto_routing = ask_yes_no("Enable auto-routing?", default=True)
    config["auto_routing"]["enabled"] = auto_routing

    if not auto_routing:
        print(color(f"  → All queries will use {default_provider}", Colors.DIM))

    # ===== Question 5: Result count =====
    print()
    max_results = ask_result_count()
    config["defaults"]["max_results"] = max_results

    # Set disabled providers (serpbase is excluded from default auto-routing — opt-in only)
    all_providers = ["serper", "brave", "tavily", "querit", "linkup", "exa", "firecrawl", "perplexity", "you", "searxng", "serpbase"]
    disabled = [p for p in all_providers if p not in enabled_providers]
    config["auto_routing"]["disabled_providers"] = disabled

    # ===== Save config =====
    print()
    print(color("─" * 60, Colors.DIM))
    print(color("\n💾 Saving Configuration\n", Colors.BOLD))

    with open(config_path, 'w') as f:
        json.dump(config, f, indent=2)

    print(color(f"✓ Configuration saved to: {config_path}", Colors.GREEN))
    print()

    # ===== Summary =====
    print(color("📋 Configuration Summary:", Colors.BOLD))
    print(f"   Enabled providers: {', '.join(enabled_providers)}")
    print(f"   Default provider: {default_provider}")
    print(f"   Auto-routing: {'enabled' if auto_routing else 'disabled'}")
    print(f"   Results per search: {max_results}")
    print()

    # ===== Test suggestion =====
    print(color("🚀 Ready to search! Try:", Colors.BOLD))
    print(color(f"   python3 scripts/search.py -q \"your query here\"", Colors.CYAN))
    print()

    return True

def check_first_run(skill_dir: Path) -> bool:
    """Check if this is the first run (no config.json)."""
    config_path = skill_dir / "config.json"
    return not config_path.exists()

def main():
    # Determine skill directory
    script_path = Path(__file__).resolve()
    skill_dir = script_path.parent.parent

    # Check for --reset flag
    force_reset = "--reset" in sys.argv

    # Check for --check flag (just check if setup needed)
    if "--check" in sys.argv:
        if check_first_run(skill_dir):
            print("Setup required: config.json not found")
            sys.exit(1)
        else:
            print("Setup complete: config.json exists")
            sys.exit(0)

    # Run setup
    success = run_setup(skill_dir, force_reset)
    sys.exit(0 if success else 1)

if __name__ == "__main__":
    main()

#!/bin/bash

# Test Auto-Routing Feature
# Tests various query types to verify routing works correctly

# Load from environment or .env file
if [ -f .env ]; then
  source .env
fi

# Check required keys
if [ -z "$SERPER_API_KEY" ]; then
  echo "Error: SERPER_API_KEY not set. Copy .env.example to .env and add your keys."
  exit 1
fi

echo "Testing auto-routing..."
python3 scripts/search.py -q "buy iPhone 15 price" --auto
python3 scripts/search.py -q "how does quantum computing work" --auto  
python3 scripts/search.py -q "companies like Stripe" --auto

import os
import sys
import unittest
from pathlib import Path
from unittest import mock

SKILL_DIR = Path(__file__).resolve().parents[1]
if str(SKILL_DIR) not in sys.path:
    sys.path.insert(0, str(SKILL_DIR))

from scripts import extract, search


class RoutingParityTests(unittest.TestCase):
    def make_analyzer(self):
        config = {
            "auto_routing": {
                "fallback_provider": "serper",
                "provider_priority": [
                    "tavily",
                    "linkup",
                    "querit",
                    "exa",
                    "firecrawl",
                    "perplexity",
                    "brave",
                    "serper",
                    "you",
                    "searxng",
                ],
                "disabled_providers": ["searxng"],
            },
            "serper": {"api_key": "x"},
            "brave": {"api_key": "x"},
            "tavily": {"api_key": "x"},
            "querit": {"api_key": "x"},
            "linkup": {"api_key": "x"},
            "exa": {"api_key": "x"},
            "firecrawl": {"api_key": "x"},
            "perplexity": {"api_key": "x"},
            "you": {"api_key": "x"},
            "searxng": {},
        }
        return search.QueryAnalyzer(config)

    def test_linkup_routing_for_source_grounded_query(self):
        routed = self.make_analyzer().route("find credible sources and citations for AI tutoring outcomes")
        self.assertEqual(routed["provider"], "linkup")

    def test_multilingual_recency_query_scores_querit_signal(self):
        routed = self.make_analyzer().route("latest AI policy updates in Germany")
        self.assertGreater(routed["scores"]["querit"], 0)
        self.assertIn(routed["provider"], {"serper", "brave", "perplexity", "querit", "tavily"})

    def test_generic_current_web_query_uses_brave_or_serper(self):
        routed = self.make_analyzer().route("weather in Vienna today")
        self.assertIn(routed["provider"], {"brave", "serper", "perplexity"})

    def test_tie_breaker_is_deterministic(self):
        winners = ["brave", "serper"]
        first = search._choose_tie_winner("weather in Vienna today", winners, winners)
        second = search._choose_tie_winner("weather in Vienna today", winners, winners)
        self.assertEqual(first, second)
        self.assertIn(first, winners)


class ExtractTests(unittest.TestCase):
    def test_invalid_url_rejected(self):
        result = extract.extract_plus(["example.com"], provider="auto")
        self.assertIn("Invalid URL", result["error"])

    def test_missing_keys_reported(self):
        with mock.patch.dict(os.environ, {}, clear=True):
            result = extract.extract_plus(["https://example.com"], provider="auto")
        self.assertEqual(result["error"], "All extraction providers failed")
        self.assertEqual(result["fallback_errors"][0]["provider"], "firecrawl")
        self.assertEqual(result["fallback_errors"][0]["error"], "missing_api_key")

    def test_auto_fallback_uses_next_provider_after_failure(self):
        with mock.patch.dict(os.environ, {"FIRECRAWL_API_KEY": "fire", "LINKUP_API_KEY": "link"}, clear=True):
            with mock.patch.object(extract, "extract_firecrawl", side_effect=RuntimeError("boom")):
                with mock.patch.object(extract, "extract_linkup", return_value={
                    "provider": "linkup",
                    "results": [{"url": "https://example.com", "title": "Example", "content": "ok", "raw_content": "ok", "provider": "linkup"}],
                }):
                    result = extract.extract_plus(["https://example.com"], provider="auto")
        self.assertEqual(result["provider"], "linkup")
        self.assertTrue(result["routing"]["fallback_used"])
        self.assertEqual(result["routing"]["fallback_errors"][0]["provider"], "firecrawl")


if __name__ == "__main__":
    unittest.main()