searxng-use-cli/scripts/common.py

"""Shared utilities for searxng-cli scripts.

This module centralizes code that was previously duplicated across
``search.py`` and ``fetch.py``:

* ``build_auth_headers``  — construct an Authorization header from CLI flags
* ``resolve_auth_basic``  — resolve basic-auth credentials from file/env/CLI
                            (avoids leaving passwords in shell history)
* ``detect_charset``      — guess a response's text encoding
* ``is_retryable_error``  — unified transient-error policy (urllib + requests)
* ``FALLBACK_UAS``        — browser-like User-Agents used when blocked
* retry constants         — ``RETRYABLE_STATUS``, ``RETRY_BACKOFF_BASE``, etc.
* ``setup_logging``       — shared logging configuration (--verbose/--quiet)

Centralizing the retry policy guarantees that both scripts treat 429/5xx
as retryable and connection errors as transient, eliminating the previous
inconsistency where ``search.py`` ignored 5xx.
"""

import logging
import sys
import urllib.error

# Root logger for the searxng-cli package. All modules create child loggers
# via ``logging.getLogger("searxng.<module>")`` so a single setup_logging()
# call controls them all.
_LOG = logging.getLogger("searxng")


def setup_logging(verbose: bool = False, quiet: bool = False) -> None:
    """Configure the ``searxng`` logger hierarchy.

    * Default (no flags): ``INFO`` — progress messages, warnings, retry notices.
      Matches the previous ``print(..., file=sys.stderr)`` behavior so existing
      scripts and agents see no change.
    * ``--verbose`` / ``-v``: ``DEBUG`` — also shows HTTP request URLs, response
      status codes, cache keys, and other diagnostic detail.
    * ``--quiet`` / ``-q``: ``WARNING`` — suppresses progress and retry noise;
      only warnings and errors reach stderr.

    All log output goes to stderr; stdout is reserved for data (JSON/CSV/etc.).
    """
    if verbose:
        level = logging.DEBUG
    elif quiet:
        level = logging.WARNING
    else:
        level = logging.INFO

    _LOG.setLevel(level)
    # Avoid duplicate handlers if setup_logging() is called twice (e.g. tests).
    if not _LOG.handlers:
        handler = logging.StreamHandler(sys.stderr)
        handler.setFormatter(logging.Formatter("%(message)s"))
        _LOG.addHandler(handler)
    # Don't let root logger add its own handler — we own the searxng namespace.
    _LOG.propagate = False

# Retry settings (shared by both scripts)
MAX_RETRIES = 3
RETRY_BACKOFF_BASE = 1.5  # seconds; exponential backoff + jitter

# HTTP status codes that are worth retrying (rate limit + gateway errors)
RETRYABLE_STATUS = frozenset({429, 502, 503, 504})

# Browser-like UA strings for fallback when the searxng-cli UA is blocked
FALLBACK_UAS = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36",
]


def build_auth_headers(bearer_token: str = None, basic_auth: str = None) -> dict:
    """Build an Authorization header dict from CLI auth flags.

    ``bearer_token``: raw Bearer token string.
    ``basic_auth``:   ``"username:password"`` string (base64-encoded).

    If both are provided, Bearer takes precedence (more common for APIs).
    Returns a dict to merge into request headers, or an empty dict.
    """
    import base64

    headers = {}
    if bearer_token:
        headers["Authorization"] = f"Bearer {bearer_token}"
    elif basic_auth:
        encoded = base64.b64encode(basic_auth.encode("utf-8")).decode("ascii")
        headers["Authorization"] = f"Basic {encoded}"
    return headers


def _warn_file_perms(path: str) -> None:
    """Warn if a credentials file is readable by group/other (POSIX only).

    On Windows the Unix permission bits in ``st_mode`` do not reflect the
    actual ACL, so the check is skipped to avoid false alarms.
    """
    import os
    if os.name != "posix":
        return
    log = logging.getLogger("searxng.common")
    try:
        mode = os.stat(path).st_mode & 0o777
        if mode & 0o077:
            log.warning(
                f"Warning: credentials file '{path}' has permissions {oct(mode)} "
                f"(accessible by group/other); recommend 'chmod 600' for security."
            )
    except OSError:
        pass


def resolve_auth_basic(cli_value: str = None, file_path: str = None,
                       env_var: str = "SEARXNG_BASIC_AUTH") -> str:
    """Resolve basic-auth credentials without leaking them via shell history.

    Priority (highest wins):
      1. ``cli_value``  — explicit ``--auth-basic "user:pass"`` (convenient
                          but leaks into shell history; discouraged)
      2. ``file_path``  — ``--auth-basic-file FILE``; first non-empty line
                          is read as ``user:pass``. Recommended for shells.
      3. ``env_var``    — ``SEARXNG_BASIC_AUTH`` environment variable.

    Returns ``"user:pass"`` or ``None`` if no source provides credentials.
    Raises ``RuntimeError`` if a file is specified but cannot be read.
    """
    if cli_value:
        return cli_value

    if file_path:
        try:
            from pathlib import Path
            text = Path(file_path).read_text(encoding="utf-8")
            _warn_file_perms(file_path)
            for line in text.splitlines():
                line = line.strip()
                if line and not line.startswith("#"):
                    return line
            raise RuntimeError(f"auth file '{file_path}' contains no credentials")
        except OSError as e:
            raise RuntimeError(f"cannot read auth file '{file_path}': {e}") from e

    import os
    return os.environ.get(env_var)


def resolve_auth_bearer(cli_value: str = None, file_path: str = None,
                        env_var: str = "SEARXNG_BEARER_TOKEN") -> str:
    """Resolve a Bearer token from CLI flag, file, or environment variable.

    Mirrors :func:`resolve_auth_basic` for token-style auth. Useful for
    long-lived API tokens that should not appear in shell history.
    """
    if cli_value:
        return cli_value

    if file_path:
        try:
            from pathlib import Path
            text = Path(file_path).read_text(encoding="utf-8")
            _warn_file_perms(file_path)
            for line in text.splitlines():
                line = line.strip()
                if line and not line.startswith("#"):
                    return line
            raise RuntimeError(f"token file '{file_path}' contains no token")
        except OSError as e:
            raise RuntimeError(f"cannot read token file '{file_path}': {e}") from e

    import os
    return os.environ.get(env_var)


def apply_proxy(proxy_url: str) -> None:
    """Configure proxy via environment variables.

    Sets ``HTTP_PROXY`` and ``HTTPS_PROXY`` so both urllib (which reads
    them via :func:`urllib.request.getproxies`) and ``requests`` (which
    honors them when ``trust_env=True``, the default) pick up the proxy
    without any changes to call sites.

    ``NO_PROXY`` is set to ``localhost,127.0.0.1,::1`` (if not already set)
    so local traffic stays direct — matters for self-hosted SearXNG on
    localhost behind a corporate proxy.

    Pass an empty string to clear the proxy env vars (rarely needed; the
    default unset state already means "no proxy").
    """
    import os
    if not proxy_url:
        return
    os.environ["HTTP_PROXY"] = proxy_url
    os.environ["HTTPS_PROXY"] = proxy_url
    # Keep local traffic direct unless the user has explicitly set NO_PROXY
    os.environ.setdefault("NO_PROXY", "localhost,127.0.0.1,::1")


def detect_charset(raw: bytes, content_type: str) -> str:
    """Detect charset from the Content-Type header, then an HTML meta tag.

    Falls back to UTF-8 (with replacement) if nothing reliable is found.
    """
    import re

    # 1. HTTP header
    if "charset=" in content_type:
        charset = content_type.split("charset=")[-1].split(";")[0].strip()
        try:
            raw.decode(charset)
            return charset
        except (UnicodeDecodeError, LookupError):
            pass

    # 2. HTML <meta charset> or <meta http-equiv>
    try:
        head = raw[:4096].decode("ascii", errors="replace")
        m = re.search(r'<meta[^>]+charset=["\']?([a-zA-Z0-9_-]+)', head, re.IGNORECASE)
        if m:
            charset = m.group(1).strip()
            try:
                raw.decode(charset)
                return charset
            except (UnicodeDecodeError, LookupError):
                pass
    except Exception:
        pass

    # 3. Fallback: UTF-8 with replacement
    return "utf-8"


def is_retryable_error(exc: BaseException) -> bool:
    """Return True if ``exc`` is a transient error worth retrying.

    Handles both the stdlib ``urllib`` errors and ``requests`` errors via
    duck-typing (so this module does not need to import ``requests``):

    * ``urllib.error.HTTPError`` → retry iff status in ``RETRYABLE_STATUS``
    * ``urllib.error.URLError`` / ``OSError`` / ``TimeoutError`` → retry
      (connection refused, DNS failure, timeout — all transient)
    * ``requests.exceptions.HTTPError`` → retry iff ``response.status_code``
      is in ``RETRYABLE_STATUS``
    * ``requests`` connection/timeout errors (no ``.response``) → retry
    """
    if isinstance(exc, urllib.error.HTTPError):
        return exc.code in RETRYABLE_STATUS
    if isinstance(exc, (urllib.error.URLError, OSError, TimeoutError)):
        return not isinstance(exc, urllib.error.HTTPError)

    # requests.exceptions.HTTPError / RequestException (duck-typed)
    resp = getattr(exc, "response", None)
    status = getattr(resp, "status_code", None)
    if status is not None:
        return status in RETRYABLE_STATUS
    if resp is None:
        # requests connection/timeout error without a response -> transient
        return True
    return False


# ----- Structured error classification -----
#
# 错误码体系：让 AI Agent 程序化地判断错误类型并采取恢复策略。
# 所有错误码以 E_ 前缀，在 --format json 模式下随 error_code 字段输出。
#
# AI 可根据 error_code 决策：
#   E_CONFIG     → 检查实例配置/环境变量，提示用户设置
#   E_AUTH       → 检查 token/凭证，提示用户重新认证
#   E_NETWORK    → 重试或切换实例/代理
#   E_RATE_LIMIT → 等待后重试，降低请求频率
#   E_PARSE      → 检查实例是否支持 JSON，尝试 HTML 回退
#   E_EMPTY      → 调整查询词或时间范围
#   E_INPUT      → 修正参数/文件路径
#   E_INTERNAL   → 报告 bug，附带完整错误信息

# 错误码常量（供 search.py / fetch.py 引用）
E_CONFIG = "E_CONFIG"
E_AUTH = "E_AUTH"
E_NETWORK = "E_NETWORK"
E_RATE_LIMIT = "E_RATE_LIMIT"
E_PARSE = "E_PARSE"
E_EMPTY = "E_EMPTY"
E_INPUT = "E_INPUT"
E_INTERNAL = "E_INTERNAL"


def classify_error(exc: BaseException) -> str:
    """将异常分类为结构化错误码，供 AI Agent 程序化处理。

    分类逻辑（按优先级）：
      1. 429 → E_RATE_LIMIT
      2. 401/403 → E_AUTH
      3. 4xx（非上述）→ E_INPUT（请求参数问题）
      4. 5xx / URLError / OSError / TimeoutError → E_NETWORK
      5. json.JSONDecodeError / ValueError → E_PARSE
      6. FileNotFoundError → E_INPUT
      7. RuntimeError → 尝试从消息中提取线索，否则 E_INTERNAL
      8. 其他 → E_INTERNAL
    """
    import json as _json

    # HTTP 错误（urllib 和 requests 都有 .code 或 .status_code）
    status = None
    if isinstance(exc, urllib.error.HTTPError):
        status = exc.code
    else:
        resp = getattr(exc, "response", None)
        status = getattr(resp, "status_code", None)

    if status is not None:
        if status == 429:
            return E_RATE_LIMIT
        if status in (401, 403):
            return E_AUTH
        if 400 <= status < 500:
            return E_INPUT
        if 500 <= status < 600:
            return E_NETWORK

    # 文件/输入错误（FileNotFoundError 是 OSError 子类，必须先于 OSError 检查）
    if isinstance(exc, FileNotFoundError):
        return E_INPUT

    # 连接级错误
    if isinstance(exc, (urllib.error.URLError, OSError, TimeoutError)):
        return E_NETWORK
    if isinstance(exc, ConnectionError):
        return E_NETWORK

    # 解析错误
    if isinstance(exc, (ValueError, _json.JSONDecodeError)):
        return E_PARSE

    # RuntimeError：从消息中推断（search_multi 的 "All N instances failed" 等）
    msg = str(exc).lower()
    if isinstance(exc, RuntimeError):
        if "auth" in msg or "403" in msg or "401" in msg:
            return E_AUTH
        if "rate" in msg or "429" in msg:
            return E_RATE_LIMIT
        if "all" in msg and "instance" in msg and "fail" in msg:
            return E_NETWORK
        if "parse" in msg or "json" in msg or "html" in msg:
            return E_PARSE
        if "not found" in msg or "no " in msg and "instance" in msg:
            return E_CONFIG
        return E_INTERNAL

    return E_INTERNAL


# ----- Progress event emitter (for --progress flag) -----
#
# 当 --progress 启用时，search.py 会调用 emit_progress() 发射结构化事件到
# stderr（JSON Lines 格式）。AI Agent 可解析这些事件来跟踪执行进度。
#
# 事件类型：
#   {"event": "start", "query": "...", "instances": N}
#   {"event": "instance_try", "url": "...", "attempt": 1}
#   {"event": "instance_ok", "url": "...", "latency": 0.5, "results": 10}
#   {"event": "instance_fail", "url": "...", "error": "...", "error_code": "E_*"}
#   {"event": "cache_hit", "query": "...", "ttl": 30}
#   {"event": "cache_store", "query": "...", "ttl": 30}
#   {"event": "fetch_start", "count": 3}
#   {"event": "fetch_ok", "url": "...", "chars": 1234}
#   {"event": "fetch_fail", "url": "...", "error": "..."}
#   {"event": "done", "results": N, "query": "..."}
#   {"event": "error", "error": "...", "error_code": "E_*", "query": "..."}

_progress_enabled = False


def set_progress_enabled(enabled: bool) -> None:
    """全局开关：是否向 stderr 输出 JSON Lines 格式的进度事件。"""
    global _progress_enabled
    _progress_enabled = enabled


def emit_progress(event: str, **kwargs) -> None:
    """向 stderr 输出一行 JSON 格式的进度事件。

    仅当 --progress 启用时才输出。``event`` 是事件类型字符串，
    ``kwargs`` 是事件的附加字段。输出格式为 JSON Lines（每行一个 JSON 对象）。
    """
    if not _progress_enabled:
        return
    import json as _json
    payload = {"event": event}
    payload.update(kwargs)
    print(_json.dumps(payload, ensure_ascii=False), file=sys.stderr, flush=True)