#!/usr/bin/env python3
"""Execute searches via a user-supplied SearXNG instance's JSON API.

Calls GET/POST /search?q=...&format=json on the chosen instance.
Auto-retries on failure with exponential backoff + tries next instance.
Falls back to HTML scraping if the instance blocks JSON output.

Instance URLs are REQUIRED (see --instance / SEARXNG_INSTANCE / config file).
"""

import argparse
import csv
import io
import json
import logging
import os
import random
import re
import sys
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
from concurrent.futures import ThreadPoolExecutor, as_completed
from html.parser import HTMLParser
from pathlib import Path

# Allow running standalone from any working directory
sys.path.insert(0, str(Path(__file__).resolve().parent))
from _config import SCHEMA_VERSION, USER_AGENT, VERSION
from common import (
    RETRYABLE_STATUS,
    RETRY_BACKOFF_BASE,
    MAX_RETRIES,
    RECOVERY_HINTS,
    apply_proxy,
    build_auth_headers,
    build_wayback_url,
    classify_error,
    classify_fetch_error,
    compute_backoff_delay,
    detect_charset,
    emit_progress,
    force_utf8_stdout,
    is_hard_blocked_domain,
    is_similar,
    texts_are_similar,
    parse_retry_after,
    resolve_auth_basic,
    resolve_auth_bearer,
    set_progress_enabled,
    setup_logging,
    should_try_wayback,
    E_CONFIG,
    E_AUTH,
    E_NETWORK,
    E_RATE_LIMIT,
    E_PARSE,
    E_EMPTY,
    E_INPUT,
    E_INTERNAL,
)
from fetch import extract_text, fetch_url
import cache as cache_module

logger = logging.getLogger("searxng.search")


# ----- Auth helpers -----

def _merge_headers(*header_dicts: dict) -> dict:
    """Merge multiple header dicts, with later dicts overriding earlier ones."""
    result = {}
    for d in header_dicts:
        if d:
            result.update(d)
    return result


# ----- HTML fallback: result extractor -----

class SearXNGHTMLParser(HTMLParser):
    """Extract search results + metadata from SearXNG's simple theme HTML.

    Matches:
      <article class="result result-default category-general">
        <a href="..." class="url_header"> ... </a>
        <h3><a href="...">Title</a></h3>
        <p class="content"> Snippet... </p>
        <time datetime="2024-01-15T10:30:00">...</time>
      </article>
      <div id="suggestions"><a>...</a></div>
      <div class="answer">...</div>

    The ``published_date`` field is populated from the ``<time>`` tag's
    ``datetime`` attribute (preferred) or its text content (fallback),
    mirroring the field SearXNG's JSON API exposes.
    """

    def __init__(self):
        super().__init__()
        self.results = []
        self.suggestions = []
        self.answers = []
        self.infoboxes = []

        self._current = None
        self._in_article = False
        self._in_h3 = False
        self._in_content = False
        self._in_time = False
        self._in_suggestions = False
        self._in_answer = False
        self._text_buf = []
        self._skip_depth = 0

    def handle_starttag(self, tag, attrs):
        attrs_dict = dict(attrs)
        classes = attrs_dict.get("class", "").split()
        tag_id = attrs_dict.get("id", "")

        if self._skip_depth > 0:
            self._skip_depth += 1
            return

        # Results
        if tag == "article" and "result" in classes:
            self._in_article = True
            self._current = {"title": "", "url": "", "content": "",
                             "engine": "", "published_date": ""}

        if self._in_article:
            if tag == "a" and "url_header" in classes:
                href = attrs_dict.get("href", "")
                if href:
                    self._current["url"] = href
            elif tag == "h3":
                self._in_h3 = True
                self._text_buf = []
            elif tag == "a" and self._in_h3:
                # h3 内的 <a href> 作为 url 的 fallback：某些 SearXNG 主题
                # 不使用 url_header class，URL 仅出现在 h3 的链接中。
                # url_header 优先（上面已处理），此处仅填充空 url。
                href = attrs_dict.get("href", "")
                if href and not self._current.get("url"):
                    self._current["url"] = href
            elif tag == "p" and "content" in classes:
                self._in_content = True
                self._text_buf = []
            elif tag == "time":
                # SearXNG puts the ISO timestamp in the datetime attribute
                dt = attrs_dict.get("datetime", "").strip()
                if dt and self._current is not None:
                    self._current["published_date"] = dt
                # Still collect text as a fallback for instances that omit
                # the datetime attribute but render a human-readable date.
                self._in_time = True
                self._text_buf = []
            elif tag in ("script", "style"):
                self._skip_depth += 1

        # Suggestions: <div id="suggestions"> or class containing "suggestion"
        if tag_id == "suggestions" or "suggestion" in classes:
            self._in_suggestions = True
            self._text_buf = []
        if self._in_suggestions and tag == "a":
            self._text_buf = []

        # Answer boxes
        if "answer" in classes or tag_id == "answer":
            self._in_answer = True
            self._text_buf = []

    def handle_endtag(self, tag):
        if self._skip_depth > 0:
            self._skip_depth -= 1
            return

        if self._in_article and tag == "article":
            if self._current:
                self._current["title"] = self._current.get("title", "").strip()
                self._current["content"] = self._current.get("content", "").strip()
                if self._current.get("title") or self._current.get("url"):
                    self.results.append(self._current)
            self._in_article = False
            self._current = None
        elif self._in_h3 and tag == "h3":
            if self._current:
                self._current["title"] = " ".join(self._text_buf).strip()
            self._in_h3 = False
        elif self._in_content and tag == "p":
            if self._current:
                self._current["content"] = " ".join(self._text_buf).strip()
            self._in_content = False
        elif self._in_time and tag == "time":
            # Fallback: use text content only if the datetime attribute
            # wasn't already captured at starttag time.
            if self._current and not self._current.get("published_date"):
                text = " ".join(self._text_buf).strip()
                if text:
                    self._current["published_date"] = text
            self._in_time = False

        if self._in_suggestions and tag == "div":
            self._in_suggestions = False
        if self._in_answer and tag == "div":
            if self._text_buf:
                self.answers.append(" ".join(self._text_buf).strip())
            self._in_answer = False

    def handle_data(self, data):
        if self._skip_depth > 0:
            return
        if self._in_h3:
            self._text_buf.append(data)
        elif self._in_content:
            self._text_buf.append(data)
        elif self._in_time:
            self._text_buf.append(data)
        elif self._in_suggestions:
            stripped = data.strip()
            if stripped:
                self.suggestions.append(stripped)
        elif self._in_answer:
            self._text_buf.append(data)


def parse_html_results(html: str, query: str = "") -> dict:
    """Parse SearXNG HTML results page into structured dict."""
    parser = SearXNGHTMLParser()
    parser.feed(html)
    return {
        "query": query,
        "number_of_results": len(parser.results),
        "results": parser.results,
        "answers": parser.answers,
        "corrections": [],  # rarely in simple theme
        "suggestions": parser.suggestions,
        "infoboxes": parser.infoboxes,
        "unresponsive_engines": [],
        "_fallback": "html",
    }


# ----- Instance selection -----

def parse_instances(instance_arg: str) -> list:
    """Parse user-supplied instance URL(s).

    Accepts a single URL or a comma-separated list. Each entry is normalized
    (https:// prefix added if missing, trailing slash stripped). Multiple
    instances enable multi-instance failover.
    """
    urls = []
    for raw in instance_arg.split(","):
        u = raw.strip()
        if not u:
            continue
        if not u.startswith(("http://", "https://")):
            u = "https://" + u
        urls.append(u.rstrip("/"))
    return urls


def _normalize_csv(value: str) -> str:
    """Normalize a comma-separated list: strip each item, drop empties.

    ``"google, bing, brave"`` -> ``"google,bing,brave"`` so downstream params
    match SearXNG's exact engine name requirements.
    """
    return ",".join(p.strip() for p in value.split(",") if p.strip())


def _load_toml(path: Path) -> dict:
    """Load a TOML file with Python-version-aware backend selection.

    * Python 3.11+ → stdlib ``tomllib``
    * Python 3.8–3.10 → optional ``tomli`` backport (``pip install tomli``)
    * Neither available → raise ``RuntimeError`` so the caller can fall back
      to ``instances.txt`` gracefully.

    This is required because the project claims Python 3.8+ support but
    ``tomllib`` only entered the stdlib in 3.11.
    """
    try:
        import tomllib  # Python 3.11+
    except ModuleNotFoundError:
        try:
            import tomli as tomllib  # type: ignore[no-redef]
        except ModuleNotFoundError:
            raise RuntimeError(
                "Reading .toml config requires Python 3.11+ or 'pip install tomli'"
            )
    with open(path, "rb") as f:
        return tomllib.load(f)


def _read_instance_file(path: Path) -> list:
    """Read instance URLs from a config file.

    Supports two formats:
      * ``searxng.toml``  — ``instance = "url"`` or ``instances = ["a", "b"]``
                            under a ``[searxng]`` table (or at the top level).
                            Requires Python 3.11+ or the ``tomli`` backport.
      * ``*.txt``         — one URL per line, comma-separated lists allowed,
                            ``#`` starts a comment.
    Returns an empty list if the file cannot be parsed.
    """
    try:
        if path.suffix == ".toml":
            data = _load_toml(path)
            table = data.get("searxng", data)
            raw = table.get("instance") or table.get("instances")
            if isinstance(raw, str):
                return parse_instances(raw)
            if isinstance(raw, list):
                return [u.rstrip("/") if u.startswith(("http://", "https://"))
                        else "https://" + u.rstrip("/")
                        for u in raw if u and str(u).strip()]
            return []
        # Plain text: one URL per line (comma lists allowed), # comments
        out = []
        for line in path.read_text(encoding="utf-8").splitlines():
            line = line.strip()
            if not line or line.startswith("#"):
                continue
            out.extend(parse_instances(line))
        return out
    except RuntimeError as e:
        # tomllib 缺失（Python 3.8-3.10 未装 tomli）是可恢复的——可改用
        # instances.txt——但必须明确提示用户，而不是静默返回空列表让 main
        # 报 "no instance resolved"，让用户困惑真正的失败原因。
        msg = str(e).lower()
        if "toml" in msg and ("3.11" in msg or "tomli" in msg):
            logger.error(f"Cannot parse '{path}': {e} "
                         f"(consider 'pip install tomli' or use instances.txt)")
        else:
            logger.warning(f"Warning: cannot read instance file '{path}': {e}")
        return []
    except Exception as e:
        logger.warning(f"Warning: cannot read instance file '{path}': {e}")
        return []


def _windows_appdata_config_dir() -> Path:
    """Return the Windows APPDATA config directory, or a sentinel Path if unset.

    On Windows, the conventional per-user app config directory is
    ``%APPDATA%`` (typically ``C:\\Users\\<user>\\AppData\\Roaming``).
    On POSIX, this env var is unset and we return a sentinel
    ``Path("/__no_appdata__")`` which never exists on disk, so the caller
    can unconditionally append it to the candidate list without polluting
    Linux/macOS lookups.

    Note: ``Path("")`` resolves to ``.`` (current directory) on Windows,
    which DOES exist — so we must use an absolute sentinel path instead.
    """
    appdata = os.environ.get("APPDATA", "")
    if appdata:
        return Path(appdata) / "searxng-cli"
    # Sentinel: absolute path that never exists. Using "/" + unlikely name
    # keeps it false on both POSIX and Windows (where "/" is the drive root).
    return Path("/__no_appdata__")


def resolve_instances(cli_arg: str = None) -> list:
    """Resolve instance URLs from (in priority order):

    1. ``--instance`` CLI flag (comma-separated list)
    2. ``SEARXNG_INSTANCE`` environment variable (comma-separated list)
    3. config file search order:
       a. ``./searxng.toml``
       b. ``~/.config/searxng-cli/searxng.toml``
       c. ``%APPDATA%/searxng-cli/searxng.toml``  (Windows only)
       d. ``./instances.txt``
       e. ``~/.config/searxng-cli/instances.txt``
       f. ``%APPDATA%/searxng-cli/instances.txt``  (Windows only)

    Returns an empty list if no instance can be resolved.
    """
    if cli_arg:
        return parse_instances(cli_arg)

    env = os.environ.get("SEARXNG_INSTANCE")
    if env:
        return parse_instances(env)

    win_dir = _windows_appdata_config_dir()
    candidates = [
        Path.cwd() / "searxng.toml",
        Path.home() / ".config" / "searxng-cli" / "searxng.toml",
        win_dir / "searxng.toml",
        Path.cwd() / "instances.txt",
        Path.home() / ".config" / "searxng-cli" / "instances.txt",
        win_dir / "instances.txt",
    ]
    for p in candidates:
        if p.exists():
            urls = _read_instance_file(p)
            if urls:
                return urls
    return []


def load_config(config_path: str = None) -> dict:
    """Load the full config dict from ``searxng.toml`` (if present).

    If ``config_path`` is given (from ``--config``), only that file is
    consulted. Otherwise the default search order applies:
      1. ``./searxng.toml``
      2. ``~/.config/searxng-cli/searxng.toml``

    Returns the ``[searxng]`` table (or top-level table if no section),
    which may contain any of these keys used as CLI defaults:

      * ``instance`` / ``instances`` — instance URL(s)
      * ``timeout``, ``max_retries`` — network tuning
      * ``engines``, ``categories``, ``language`` — search scope
      * ``safesearch``, ``time_range``, ``method`` — search behavior
      * ``format`` — output format (json/brief/urls/csv)
      * ``sort_by`` — result sort key
      * ``proxy`` — proxy URL
      * ``cache_ttl`` — cache TTL in minutes

    Returns an empty dict if no config file exists or it cannot be parsed.
    Only ``.toml`` files are consulted for full config; ``.txt`` files
    only carry instance URLs (handled by :func:`resolve_instances`).
    """
    if config_path:
        p = Path(config_path)
        if not p.exists():
            logger.warning(f"Warning: config file '{p}' not found")
            return {}
        try:
            data = _load_toml(p)
            return data.get("searxng", data)
        except Exception as e:
            logger.warning(f"Warning: cannot read config '{p}': {e}")
            return {}
    win_dir = _windows_appdata_config_dir()
    candidates = [
        Path.cwd() / "searxng.toml",
        Path.home() / ".config" / "searxng-cli" / "searxng.toml",
        win_dir / "searxng.toml",
    ]
    for p in candidates:
        if p.exists():
            try:
                data = _load_toml(p)
                return data.get("searxng", data)
            except Exception as e:
                logger.warning(f"Warning: cannot read config '{p}': {e}")
                return {}
    return {}


def _cfg_int(config: dict, key: str, default: int) -> int:
    """Read an int from config, tolerating str/int forms and bad values.

    argparse's ``type=int`` only converts *command-line* strings; it does NOT
    convert a ``default`` value that came in as a string from TOML. So a
    ``timeout = "15"`` in ``searxng.toml`` would leak through as a str and
    break later numeric comparisons. This helper normalizes that.
    """
    if key not in config:
        return default
    try:
        return int(config[key])
    except (TypeError, ValueError):
        return default


def _cfg_float(config: dict, key: str, default: float) -> float:
    """Read a float from config, tolerating str/int/float forms. See _cfg_int."""
    if key not in config:
        return default
    try:
        return float(config[key])
    except (TypeError, ValueError):
        return default


# ----- Retry logic -----

def _retry_with_backoff(fn, max_retries: int = MAX_RETRIES, base_delay: float = RETRY_BACKOFF_BASE):
    """Call fn with exponential backoff + jitter on transient failures.

    v2.1.0 修复：复用 common.compute_backoff_delay（带 60s 封顶），
    避免高重试次数（如 --retry 10）时 1.5*2^10=1536s 卡死进程。
    同时遵守 Retry-After header（429/503），与 fetch.py 保持一致。

    v2.5.0：403 不再重试。RETRYABLE_STATUS 含 403 是为 fetch.py 的 UA
    轮换设计的（每次重试换新 UA），但本函数仅供 search 使用——search 不
    换 UA，实例级 403 是认证/封禁问题，重试只会空等 ~10.5s 后才 failover。
    现在 403 立即 raise，由 search_multi 直接切换下一实例，classify_error
    仍正确归为 E_AUTH（实例认证失败语义不变）。
    """
    last_error = None
    for attempt in range(max_retries + 1):
        try:
            return fn()
        except urllib.error.HTTPError as e:
            if e.code in RETRYABLE_STATUS and e.code != 403:  # 429 + 5xx
                last_error = e
                if attempt < max_retries:
                    # 429/503：遵守 Retry-After header，避免触发更严厉限流
                    retry_after_sec = 0.0
                    if e.code in (429, 503) and e.headers:
                        retry_after_sec = parse_retry_after(
                            e.headers.get("Retry-After", ""))
                    delay = max(retry_after_sec,
                                compute_backoff_delay(attempt, base=base_delay))
                    logger.info(f"  HTTP {e.code}, retrying in {delay:.1f}s... "
                                f"(attempt {attempt+1}/{max_retries})"
                                f"{f' Retry-After={retry_after_sec:.1f}s' if retry_after_sec > 0 else ''}")
                    time.sleep(delay)
                    continue
            raise
        except (urllib.error.URLError, OSError) as e:
            last_error = e
            if attempt < max_retries:
                delay = compute_backoff_delay(attempt, base=base_delay)
                logger.info(f"  Connection error ({e}), retrying in {delay:.1f}s...")
                time.sleep(delay)
                continue
            raise
    raise last_error


# ----- Search execution -----

# v2.5.0 连接复用：requests 可用时复用模块级 Session（连接池 + TLS 会话
# 恢复），--pages/批量/研究模式的多请求场景显著减少 TLS 握手开销。
# stdlib 路径保持零依赖可用（每次新建连接，功能等价）——与 fetch.py 的
# "requests 路径受益，stdlib 路径功能完整" 设计一致。
_HAS_REQUESTS = False
try:
    import requests as _requests  # type: ignore
    _HAS_REQUESTS = True
except ImportError:
    pass

_session = None


def _get_session():
    """获取（惰性创建）模块级 requests.Session。

    连接池：每主机最多 10 连接，总最多 20 连接。max_retries=0 让 requests
    不做自己的重试——重试统一由 _retry_with_backoff 控制，避免双重退避。
    """
    global _session
    if _session is None and _HAS_REQUESTS:
        _session = _requests.Session()
        adapter = _requests.adapters.HTTPAdapter(
            pool_connections=10, pool_maxsize=10, max_retries=0,
        )
        _session.mount("http://", adapter)
        _session.mount("https://", adapter)
    return _session


def _reset_session() -> None:
    """关闭并重置模块级 Session。测试用。"""
    global _session
    if _session is not None:
        try:
            _session.close()
        except Exception:
            pass
        _session = None


def _finalize_json_result(data):
    """兜底填充 SearXNG JSON 响应的可选契约字段。

    部分实例/版本的 JSON 输出缺失 ``number_of_results``（服务端
    ``search.result_number()`` 的估算值，见 --dump-schema 描述）。缺失时
    用实际结果数近似填充，保证 JSON 路径与 HTML fallback 路径
    （parse_html_results 必然填充）的输出契约一致——AI Agent 依赖该字段
    判断结果规模与是否需要翻页。
    """
    if isinstance(data, dict) and "number_of_results" not in data:
        data["number_of_results"] = len(data.get("results", []))
    return data


def search_json(instance: str, params: dict, method: str = "GET",
                timeout: int = 15, auth_headers: dict = None) -> dict:
    """Execute search via JSON API. Returns None if JSON unsupported.

    v2.5.0: requests 可用时走模块级 Session（连接池复用）；否则 stdlib。
    两个后端发出完全相同的 URL（query_string 拼好后原样使用），解码逻辑
    一致（resp.content 手动 decode，不用 resp.text 的自动解码），保证
    行为可预测、跨后端可复现。
    """
    query_string = urllib.parse.urlencode(params)
    headers = _merge_headers({"User-Agent": USER_AGENT}, auth_headers)

    if _HAS_REQUESTS:
        session = _get_session()
        try:
            if method.upper() == "POST":
                resp = session.post(f"{instance}/search", data=query_string,
                                    headers=headers, timeout=timeout)
            else:
                resp = session.get(f"{instance}/search?{query_string}",
                                   headers=headers, timeout=timeout)
            if resp.status_code == 404:
                # 404 = JSON endpoint truly absent → fall back to HTML scraping
                return None
            if resp.status_code in (401, 403):
                # auth / IP issue → raise so it isn't silently masked by an
                # HTML fallback that would just 403 again.
                resp.raise_for_status()
            raw = resp.content.decode("utf-8")
            if raw.strip().startswith("{") or raw.strip().startswith("["):
                return _finalize_json_result(json.loads(raw))
            # Got HTML — JSON unsupported
            return None
        except _requests.exceptions.HTTPError as e:
            if getattr(e, "response", None) is not None and \
                    e.response.status_code == 404:
                return None
            raise

    # stdlib path（零依赖可用）
    if method.upper() == "POST":
        data = query_string.encode("utf-8")
        req = urllib.request.Request(f"{instance}/search", data=data,
                                     headers=headers, method="POST")
    else:
        req = urllib.request.Request(f"{instance}/search?{query_string}",
                                     headers=headers)

    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            raw = resp.read().decode("utf-8")
            if raw.strip().startswith("{") or raw.strip().startswith("["):
                return _finalize_json_result(json.loads(raw))
            # Got HTML — JSON unsupported
            return None
    except urllib.error.HTTPError as e:
        # 404 = JSON endpoint truly absent → fall back to HTML scraping
        if e.code == 404:
            return None
        # 403 / 401 = auth or IP issue → raise so it isn't silently masked
        # by an HTML fallback that would just 403 again. Also surfaces
        # instances that disable the JSON format via a 403 (rare but seen).
        raise


def search_html(instance: str, params: dict, timeout: int = 15,
                auth_headers: dict = None, encoding: str = None) -> dict:
    """Execute search via HTML scraping fallback.

    v2.2.2：解码改用 common.detect_charset（此前硬编码 utf-8，GBK/Shift-JIS
    等非 UTF-8 实例的页面会整体乱码）。``encoding`` 为显式覆盖（来自
    ``--language`` 无关的 CLI ``--encoding``），优先级最高。
    v2.5.0：requests 可用时走模块级 Session（与 search_json 一致）。
    """
    html_params = {k: v for k, v in params.items() if k != "format"}
    query_string = urllib.parse.urlencode(html_params)
    url = f"{instance}/search?{query_string}"

    headers = _merge_headers({"User-Agent": USER_AGENT}, auth_headers)

    if _HAS_REQUESTS:
        try:
            resp = _get_session().get(url, headers=headers, timeout=timeout)
            raw = resp.content
            content_type = resp.headers.get("Content-Type", "")
        except Exception as e:
            raise RuntimeError(f"HTML search failed for {instance}: {e}")
    else:
        req = urllib.request.Request(url, headers=headers)
        try:
            with urllib.request.urlopen(req, timeout=timeout) as resp:
                raw = resp.read()
                content_type = resp.headers.get("Content-Type", "")
        except Exception as e:
            raise RuntimeError(f"HTML search failed for {instance}: {e}")

    if encoding:
        try:
            html = raw.decode(encoding)
        except (UnicodeDecodeError, LookupError):
            html = raw.decode("utf-8", errors="replace")
    else:
        charset = detect_charset(raw, content_type)
        try:
            html = raw.decode(charset)
        except (UnicodeDecodeError, LookupError):
            html = raw.decode("utf-8", errors="replace")
    return parse_html_results(html, query=params.get("q", ""))


def search_single(instance: str, params: dict, method: str = "GET",
                  timeout: int = 15, auth_headers: dict = None,
                  encoding: str = None) -> dict:
    """Execute one search attempt, preferring JSON with HTML fallback."""
    result = search_json(instance, params, method=method, timeout=timeout,
                         auth_headers=auth_headers)
    if result is not None:
        return result
    logger.warning(f"Warning: {instance} does not support format=json, falling back to HTML parsing")
    return search_html(instance, params, timeout=timeout, auth_headers=auth_headers,
                       encoding=encoding)


def search_multi(instance_urls: list, params: dict, method: str = "GET",
                 timeout: int = 15, retry_per: int = None,
                 auth_headers: dict = None, parallel: bool = True,
                 encoding: str = None) -> dict:
    """Search across multiple instances, failing over on error.

    With parallel=True (default, multi-instance only): every instance is
    queried concurrently. The first *successful* result in the user's
    original instance order is returned — this keeps output deterministic
    (same input list always yields the same source) while drastically
    speeding up failover when an early instance is down or slow.

    With parallel=False (or a single instance): strictly sequential, one
    request at a time, trying the next instance only after the current fails.

    ``encoding`` (v2.2.2) is forwarded to the HTML-fallback path for
    non-UTF-8 instances; JSON responses are always UTF-8.
    """
    if retry_per is None:
        retry_per = MAX_RETRIES
    instance_urls = [u.rstrip("/") for u in instance_urls]

    # Single instance, or explicit serial mode: straightforward sequential path
    if len(instance_urls) <= 1 or not parallel:
        last_error = None
        for instance in instance_urls:
            logger.info(f"Trying {instance}...")
            emit_progress("instance_try", url=instance, attempt=1)
            start = time.time()
            try:
                def _do():
                    return search_single(instance, params, method=method,
                                         timeout=timeout, auth_headers=auth_headers,
                                         encoding=encoding)
                result = _retry_with_backoff(_do, max_retries=retry_per)
                emit_progress("instance_ok", url=instance,
                              latency=round(time.time() - start, 3),
                              results=len(result.get("results", [])) if result else 0)
                return result
            except Exception as e:
                last_error = e
                logger.info(f"  Failed: {e}")
                emit_progress("instance_fail", url=instance,
                              error=str(e), error_code=classify_error(e))
                continue
        raise RuntimeError(f"All {len(instance_urls)} instances failed. Last error: {last_error}")

    # Parallel path: one search per instance concurrently (each with its own
    # backoff retries), then return the first success in original order.
    logger.info(f"Trying {len(instance_urls)} instances in parallel...")

    def _task(instance: str):
        emit_progress("instance_try", url=instance, attempt=1)
        start = time.time()
        def _do():
            return search_single(instance, params, method=method,
                                 timeout=timeout, auth_headers=auth_headers,
                                 encoding=encoding)
        try:
            result = _retry_with_backoff(_do, max_retries=retry_per)
            emit_progress("instance_ok", url=instance,
                          latency=round(time.time() - start, 3),
                          results=len(result.get("results", [])) if result else 0)
            return instance, result
        except Exception as e:
            emit_progress("instance_fail", url=instance,
                          error=str(e), error_code=classify_error(e))
            return instance, e

    results_by_url = {}
    last_parallel_error = None
    with ThreadPoolExecutor(max_workers=min(len(instance_urls), 8)) as ex:
        futures = {ex.submit(_task, u): u for u in instance_urls}
        for fut in as_completed(futures):
            u = futures[fut]
            try:
                inst, res = fut.result()
            except Exception:
                continue
            if isinstance(res, Exception):
                # 保留最后一个失败详情，让 classify_error 能从消息中提取
                # 真实错误类型（401/403→E_AUTH，429→E_RATE_LIMIT 等），
                # 而不是一律误判为 E_NETWORK。
                last_parallel_error = res
                logger.info(f"  Failed {u}: {res}")
            else:
                results_by_url[u] = res

    # Deterministic: return first successful in the user-supplied order
    for u in instance_urls:
        if u in results_by_url:
            return results_by_url[u]

    raise RuntimeError(f"All {len(instance_urls)} instances failed (parallel). "
                       f"Last error: {last_parallel_error}")


# ----- Output formatting -----

# ----- Instance health verification (E5) -----

def _probe_config_endpoint(instance: str, timeout: int,
                           auth_headers: dict = None) -> dict:
    """Probe ``/config``; return engines list and categories if reachable.

    SearXNG's ``/config`` endpoint exposes the instance's engine list,
    categories, and other settings as JSON. Some instances disable it; in
    that case we return ``reachable=False`` with an empty engine list.
    """
    headers = _merge_headers({"User-Agent": USER_AGENT}, auth_headers)
    req = urllib.request.Request(f"{instance}/config", headers=headers)
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            raw = resp.read().decode("utf-8")
            data = json.loads(raw)
            engines = [e.get("name") for e in data.get("engines", [])
                       if isinstance(e, dict) and e.get("name")]
            cats = data.get("categories", {})
            if isinstance(cats, dict):
                categories = list(cats.keys())
            elif isinstance(cats, list):
                categories = [c for c in cats if isinstance(c, str)]
            else:
                categories = []
            return {"reachable": True, "engines": engines,
                    "categories": categories, "error": None}
    except urllib.error.HTTPError as e:
        return {"reachable": False, "engines": [], "categories": [],
                "error": f"HTTP {e.code}"}
    except Exception as e:
        return {"reachable": False, "engines": [], "categories": [],
                "error": str(e)[:80]}


def verify_instances(instance_urls: list, timeout: int = 15,
                     auth_headers: dict = None, concurrency: int = 5) -> list:
    """Health-check user-supplied instances: reachability, JSON support, latency,
    POST support, /config endpoint, engine list, and auth status.

    Replaces the removed public-instance discovery: you bring your own
    instances, and ``--verify`` tells you which are alive, which support the
    JSON API, which accept POST, what engines they expose, and how fast
    each responds.

    Returns a list of dicts (one per instance, in input order), each with:
      url, reachable (bool), json_supported (bool|None), post_supported (bool|None),
      config_endpoint (bool|None), engines (list[str]), latency (float|None),
      result_count (int|None), auth_status (str), error (str|None)
    """
    instance_urls = [u.rstrip("/") for u in instance_urls]
    has_auth = bool(auth_headers)

    def _check(u: str) -> dict:
        start = time.time()
        # 1. GET /search?q=test&format=json — the primary reachability probe
        try:
            data = search_json(u, {"q": "test", "format": "json"}, "GET",
                               timeout=timeout, auth_headers=auth_headers)
            latency = round(time.time() - start, 3)
            auth_status = "ok" if has_auth else "n/a"

            if data is None:
                # Instance returned HTML — JSON format disabled but reachable
                base = {"url": u, "reachable": True, "json_supported": False,
                        "latency": latency, "result_count": None,
                        "auth_status": auth_status,
                        "error": "returned HTML (format=json unsupported)"}
            else:
                base = {"url": u, "reachable": True, "json_supported": True,
                        "latency": latency,
                        "result_count": len(data.get("results", [])),
                        "auth_status": auth_status, "error": None}

            # 2. POST probe — only meaningful when JSON works
            if data is not None:
                try:
                    post_data = search_json(u, {"q": "test", "format": "json"}, "POST",
                                            timeout=timeout, auth_headers=auth_headers)
                    base["post_supported"] = post_data is not None
                except urllib.error.HTTPError as e:
                    base["post_supported"] = False
                    base["error"] = (base.get("error") or "") + f" POST: HTTP {e.code}"
                except Exception as e:
                    base["post_supported"] = False
                    base["error"] = (base.get("error") or "") + f" POST: {str(e)[:60]}"
            else:
                base["post_supported"] = None

            # 3. /config probe — engine list (independent of JSON search)
            cfg = _probe_config_endpoint(u, timeout, auth_headers)
            base["config_endpoint"] = cfg["reachable"]
            base["engines"] = cfg["engines"]
            if cfg["error"] and not cfg["reachable"]:
                # /config down is non-fatal; note it but don't overwrite search error
                base["error"] = (base.get("error") or "") + f" /config: {cfg['error']}"

            return base
        except urllib.error.HTTPError as e:
            # Distinguish auth rejection (401/403) from other HTTP errors so
            # users know whether their token is wrong or the instance is down.
            auth_status = ("rejected" if has_auth and e.code in (401, 403)
                           else ("ok" if has_auth else "n/a"))
            # 5xx 是服务器错误，实例虽然响应了但不可用，应视为不可达。
            # 400 可能只是请求格式问题，实例本身在线，仍算可达。
            is_5xx = 500 <= e.code < 600
            return {"url": u, "reachable": e.code not in (401, 403, 404) and not is_5xx,
                    "json_supported": False, "post_supported": None,
                    "config_endpoint": None, "engines": [],
                    "latency": round(time.time() - start, 3),
                    "result_count": None, "auth_status": auth_status,
                    "error": f"HTTP {e.code}"}
        except Exception as e:
            return {"url": u, "reachable": False, "json_supported": False,
                    "post_supported": None, "config_endpoint": None,
                    "engines": [], "latency": None, "result_count": None,
                    "auth_status": "unknown" if has_auth else "n/a",
                    "error": str(e)[:120]}

    out = {}
    with ThreadPoolExecutor(max_workers=min(concurrency, len(instance_urls))) as ex:
        futures = {ex.submit(_check, u): u for u in instance_urls}
        for fut in as_completed(futures):
            u = futures[fut]
            try:
                out[u] = fut.result()
            except Exception as e:
                out[u] = {"url": u, "reachable": False, "json_supported": False,
                          "post_supported": None, "config_endpoint": None,
                          "engines": [], "latency": None, "result_count": None,
                          "auth_status": "unknown" if has_auth else "n/a",
                          "error": str(e)[:120]}
    # Preserve input order
    return [out[u] for u in instance_urls]


def _print_verify_report(report: list, as_json: bool):
    """Print a health-check report (human table or JSON).

    The table shows one line per instance with the most important signals
    (reach / JSON / POST / latency / result count / engine count / auth).
    The full engine list per instance is only included in the JSON output
    to keep the table readable.
    """
    if as_json:
        print(json.dumps(report, indent=2, ensure_ascii=False))
        return
    header = (f"{'URL':<38} {'REACH':<6} {'JSON':<5} {'POST':<5} "
              f"{'LAT':<8} {'RES':<5} {'ENGS':<5} {'AUTH':<9} ERROR")
    lines = [header, "-" * len(header)]
    for r in report:
        reach = "yes" if r["reachable"] else "NO"
        js = "-" if r["json_supported"] is None else ("yes" if r["json_supported"] else "no")
        post = "-" if r.get("post_supported") is None else ("yes" if r["post_supported"] else "no")
        lat = f"{r['latency']}s" if r["latency"] is not None else "-"
        rc = "-" if r["result_count"] is None else str(r["result_count"])
        engs = str(len(r.get("engines", []))) if r.get("engines") is not None else "-"
        auth = r.get("auth_status", "-")
        err = (r["error"] or "").replace("\n", " ")[:40]
        lines.append(f"{r['url'][:37]:<38} {reach:<6} {js:<5} {post:<5} "
                     f"{lat:<8} {rc:<5} {engs:<5} {auth:<9} {err}")
    ok = sum(1 for r in report if r["reachable"])
    lines.append("-" * len(header))
    lines.append(f"{ok}/{len(report)} instances reachable")
    print("\n".join(lines))


# ----- Auto-fetch result pages -----
# NOTE: HTTP transport (retry, charset, UA fallback, size limit) is delegated
# to fetch.fetch_url — this module only post-processes its result. This
# eliminates ~70 lines of duplicated HTTP code that previously drifted
# between search.fetch_page and fetch.fetch_url.

def fetch_page(url: str, timeout: int = 10, auth_headers: dict = None,
               max_retries: int = 3, max_size: int = None,
               referer: str = None,
               fallback_enabled: bool = True) -> dict:
    """Fetch a single page; returns metadata dict with 'status'='ok' or 'error'.

    Thin wrapper around :func:`fetch.fetch_url` that adds:
      * CAPTCHA / bot-block detection with WAF fingerprinting (v2.0.0)
      * Wayback Machine fallback on 404/403/timeout (v2.0.0, default on)
      * automatic text extraction via :func:`fetch.extract_text`
      * dict-shaped return suitable for the auto-fetch feature

    All HTTP transport concerns (retry, charset, UA fallback, size limit,
    browser headers, Retry-After compliance) are handled by ``fetch_url``.

    v2.0.0 新字段：
      * ``anti_bot_detected`` (bool): 是否检测到反爬页面
      * ``waf_type`` (str|None): WAF 类型（cloudflare/imperva/perimeterx/
        datadome/akamai/generic），仅当 anti_bot_detected=True 时有值
      * ``fallback_used`` (str|None): 兜底方式（"wayback"），仅当走兜底时有值
    """
    # 主抓取
    # v2.3.0: 计时整条链路（主抓取 + Wayback 兜底），填充 latency 字段。
    _start_time = time.monotonic()
    result = None
    error_msg = None
    error_code = None
    try:
        result = fetch_url(
            url, timeout=timeout, auth_headers=auth_headers,
            max_retries=max_retries, max_size=max_size,
            allow_redirects=True, referer=referer,
        )
    except Exception as e:
        error_msg = str(e) if str(e) else e.__class__.__name__
        # v2.1.0：结构化错误码，让 AdaptiveThrottle 能用 error_code 检测 429
        # 而非字符串匹配（"Too Many Requests" 不含 "429" 会漏判）。
        # v2.4.0：抓取场景用 classify_fetch_error——403 反爬拦截细分为
        # E_BLOCKED，避免误判为 E_AUTH（凭证问题）。
        error_code = classify_fetch_error(e)

    # fetch_url 对 PDF/DOCX/XLSX 解析失败不抛异常，而是返回带 error_code
    # 的 FetchResult——此处必须显式检查，否则失败会被当作成功处理
    # （空文本 + status="ok"）。
    if result is not None and result.error_code:
        error_msg = result.error_message or result.error_code
        error_code = result.error_code
        result = None

    # 反爬检测（v2.0.0 增强：全文档扫描 + WAF 指纹库）
    # 必须在 Wayback 兜底判断之前执行：Cloudflare 质询页常返回 HTTP 200，
    # 此时 result 非 None 但内容是反爬页，必须识别出来才能触发兜底。
    anti_bot_detected = False
    waf_type = None
    if result is not None:
        content = result.content
        content_type = result.content_type or ""
        is_html = ("html" in content_type.lower() or
                   content.strip().startswith("<!") or
                   content.strip().startswith("<htm"))
        if is_html:
            waf_type = _detect_anti_bot(content)
            if waf_type:
                anti_bot_detected = True

    # Wayback 兜底触发条件（v2.0.1 修复）：
    #   1. 主抓取抛异常且错误暗示 404/403/超时（_should_try_fallback）
    #   2. 主抓取"成功"但被反爬拦截（anti_bot_detected=True）
    # 原 v2.0.0 bug：仅条件 1 触发兜底，条件 2 漏网（Cloudflare 200 质询页）。
    fallback_used = None
    need_fallback = False
    if fallback_enabled:
        if anti_bot_detected:
            need_fallback = True
        elif result is None and _should_try_fallback(result, error_msg, url):
            need_fallback = True

    if need_fallback:
        wb_result = _try_wayback_fallback(url, timeout=timeout,
                                          auth_headers=auth_headers,
                                          max_retries=max_retries,
                                          max_size=max_size)
        if wb_result is not None:
            result = wb_result
            error_msg = None
            fallback_used = "wayback"
            # Wayback 结果重新做反爬检测（防御性：Wayback 快照极少是反爬页）
            anti_bot_detected = False
            waf_type = None
            content = result.content
            content_type = result.content_type or ""
            is_html = ("html" in content_type.lower() or
                       content.strip().startswith("<!") or
                       content.strip().startswith("<htm"))
            if is_html:
                waf_type = _detect_anti_bot(content)
                if waf_type:
                    anti_bot_detected = True

    # 仍然失败
    if result is None:
        return {
            "url": url, "status": "error",
            "error": error_msg or "unknown error",
            "error_code": error_code,
            "text": "", "text_length": 0, "truncated": False,
            "anti_bot_detected": False, "waf_type": None,
            "fallback_used": None,
            # v2.3.0: title/latency 字段（--fetch-report json 消费）
            "title": None,
            "latency": round(time.monotonic() - _start_time, 3),
        }

    # 反爬仍被检测到（Wayback 也无能为力或兜底被禁用）
    if anti_bot_detected:
        return {
            "url": url, "final_url": result.final_url, "status": "error",
            "error": f"Bot protection detected ({waf_type})",
            "error_code": E_PARSE,
            "text": "", "text_length": 0, "truncated": False,
            "anti_bot_detected": True, "waf_type": waf_type,
            "fallback_used": fallback_used,
            "title": _extract_title(content) if is_html else None,
            "latency": round(time.monotonic() - _start_time, 3),
        }

    text = extract_text(content) if is_html else content
    return {
        "url": url,
        "final_url": result.final_url,
        "status": "ok",
        "content_type": content_type,
        "text": text,
        "text_length": len(text),
        "truncated": result.truncated,
        "truncated_at": max_size if result.truncated else None,
        "user_agent_used": result.user_agent,
        "anti_bot_detected": False,
        "waf_type": None,
        "fallback_used": fallback_used,
        "title": _extract_title(content) if is_html else None,
        "latency": round(time.monotonic() - _start_time, 3),
    }


def _should_try_fallback(result, error_msg: str, url: str = None) -> bool:
    """判断是否应触发 Wayback 兜底。

    v2.1.0 改为调用 common.should_try_wayback 共享逻辑 + 被墙站点检测。

    触发条件：
      1. 主抓取抛异常且错误信息暗示 404/403/超时（common.should_try_wayback）
      2. 主抓取成功但被反爬拦截（理论上 fetch_page 已处理，此处防御性）
      3. v2.1.0: URL 属于被墙/强反爬站点（is_hard_blocked_domain）

    不触发：
      * 用户禁用兜底（调用方控制，不进入此函数）
      * 错误是 DNS 失败（Wayback 也访问不到）
    """
    if result is not None:
        # 主抓取成功，无需兜底
        return False
    # v2.1.0: 被墙站点直接触发兜底（不等错误信息判断）
    if url and is_hard_blocked_domain(url):
        return True
    return should_try_wayback(error_msg)


def _try_wayback_fallback(url: str, timeout: int = 10,
                          auth_headers: dict = None,
                          max_retries: int = 2,
                          max_size: int = None):
    """尝试从 Wayback Machine 获取页面快照。

    v2.1.0 改为使用 common.build_wayback_url 共享逻辑。

    使用 ``https://web.archive.org/web/2/<url>`` 端点，``2`` 表示
    "最新可用快照"。Wayback 会 302 重定向到具体时间戳快照。

    返回 FetchResult 或 None（失败时）。独立超时（10s），不阻塞主流程。
    """
    wayback_url = build_wayback_url(url)
    wb_timeout = min(timeout, 10)  # Wayback 自身可能慢，限制最大 10s
    try:
        logger.info(f"  [FALLBACK] Trying Wayback Machine for {url[:55]}")
        result = fetch_url(
            wayback_url, timeout=wb_timeout, auth_headers=None,
            max_retries=max_retries, max_size=max_size,
            allow_redirects=True,
        )
        # Wayback 包装页也算成功——它返回的是原始页面内容
        return result
    except Exception as e:
        logger.info(f"  [FALLBACK] Wayback failed for {url[:55]}: {e}")
        return None


# ----- 反爬检测（v2.0.1 收窄误判 + title 精准检测）-----
# WAF 指纹库：每项 = (waf_type, [指示词])
#
# v2.0.1 修复：v2.0.0 用裸公司名（cloudflare/akamai）和宽泛词（captcha/
# challenge/dd-）做全文匹配，导致正常文章（如引用 Cloudflare 文档、讨论
# "coding challenges" 的文章、Wayback 归档正文）被误判为反爬页，Wayback
# 兜底返回的有效内容也被丢弃。收窄原则：
#   1. 专用指纹只用 WAF 厂商的技术标识符（cookie 名/HTTP header 名/JS 变量名）
#      ——这些不会出现在文章正文里
#   2. 通用文案用完整短语而非单词（如 "please complete the captcha" 而非
#      "captcha"），避免正常内容误判
#   3. 增加 <title> 标签检测——反爬页 title 是特征文案，最精准
WAF_FINGERPRINTS = [
    ("cloudflare", [
        # Cloudflare 技术标识符（cookie/header/JS 变量名，不会出现在正文）
        "cf-ray", "cf-chl-bypass", "cf-mitigated",
        "cf-browser-verification", "cf-error-details", "cf-error-code",
        # 质询页特征文案（足够具体，正常内容不会完整出现）
        "just a moment", "checking your browser before accessing",
        "attention required! | cloudflare",
        "enable javascript and cookies to continue",
    ]),
    ("imperva", [
        # Incapsula cookie/技术标识符
        "incap_ses", "visid_incap", "incap_ses_",
        "incapsula incident id", "request unsuccessful. incapsula",
        "visit denied by incapsula",
    ]),
    ("perimeterx", [
        # PerimeterX 专有标识符（_px 太短会匹配 CSS 类名，已移除）
        "px-captcha", "pxhd", "pxcts", "pxcookie",
        "_pxff", "_pxhd",
        "press & hold to confirm you are a human",
    ]),
    ("datadome", [
        # DataDome 专有标识符（dd- 太宽泛会匹配 dd-class 等，已移除）
        "datadome", "data-dome",
        "protected by datadome", "datadome-bot-protect",
    ]),
    ("akamai", [
        # Akamai Bot Manager cookie/标识符（akamai 裸名会匹配正文引用，已移除）
        "bm_sz", "_abck", "akamaighost", "akamai-bot-manager",
        "ak_bmsc",
    ]),
    # 通用反爬指示词（v2.0.1 收窄：完整短语而非单词，避免正文误判）
    ("generic", [
        "verify you are human", "verify that you are human",
        "making sure you're not a bot", "are you a robot",
        "robot or human", "human verification",
        "please complete the captcha", "complete the security check",
        "please enable javascript to continue",
        "enable javascript to continue",
        "ddos protection by", "access denied - sucuri",
        "you have been blocked", "unusual traffic from your computer",
        "pardon our interruption", "we'll be right back",
        "bot protection", "anti-bot protection",
        "anubis_challenge", "miserere",  # Anubis 反爬系统（拦截 AI 爬虫）
    ]),
]

# <title> 标签检测：反爬页 title 通常是特征文案，比全文扫描更精准。
# key = title 中的特征子串（小写），value = 对应 WAF 类型
_TITLE_ANTI_BOT_SIGNATURES = {
    "just a moment": "cloudflare",
    "attention required": "cloudflare",
    "access denied": "generic",
    "are you a robot": "generic",
    "robot check": "generic",
    "human verification": "generic",
    "please verify you are human": "generic",
    "security check": "generic",
    "verify you are human": "generic",
}

_TITLE_RE = re.compile(r"<title[^>]*>(.*?)</title>", re.IGNORECASE | re.DOTALL)


def _extract_title(content: str) -> str:
    """从 HTML 内容提取 <title> 文本（v2.3.0，供 --fetch-report json 使用）。

    截断到 200 字符，防止异常页面标题撑爆报告。非 HTML 内容返回空串。
    """
    if not content:
        return ""
    m = _TITLE_RE.search(content)
    if m:
        return m.group(1).strip()[:200]
    return ""


def _detect_anti_bot(content: str) -> str:
    """检测反爬页面，返回 WAF 类型或 None。

    v2.0.1 改进：
      * 增加 <title> 标签精准检测（反爬页 title 是特征文案，误判率极低）
      * 收窄 WAF 指纹库关键词（移除裸公司名和宽泛词，改用技术标识符 + 完整短语）
      * 保留全文档扫描（v2.0.0 改进，大页面反爬页可能在前 2000 字之外）

    检测顺序：title 标签（最精准）→ 全文档指纹扫描（技术标识符 + 文案短语）。

    性能：全文档 lower() 一次 + title regex 一次，对 5MB 页面约 5ms，可接受。
    """
    if not content:
        return None
    # 1. <title> 标签检测（最精准，误判率极低）
    title_match = _TITLE_RE.search(content)
    if title_match:
        title_lower = title_match.group(1).strip().lower()
        for sig, waf_type in _TITLE_ANTI_BOT_SIGNATURES.items():
            if sig in title_lower:
                return waf_type
    # 2. 全文档指纹扫描（收窄后的技术标识符 + 完整文案）
    lower = content.lower()
    for waf_type, indicators in WAF_FINGERPRINTS:
        for ind in indicators:
            if ind in lower:
                return waf_type
    return None


def _is_blocked_page(content: str) -> bool:
    """[已废弃] 快速检测反爬页面。保留向后兼容，内部调用 _detect_anti_bot。

    v2.0.0 起请使用 _detect_anti_bot() 获取具体 WAF 类型。
    """
    return _detect_anti_bot(content) is not None


class AdaptiveThrottle:
    """自适应限流状态机（v2.0.0）。

    在 fetch_top_results 的并发抓取过程中，根据成功/失败反馈动态调整：
      * 连续 >=3 次失败 → request_delay 翻倍，concurrency 减半
      * 连续 >=5 次成功 → 逐步恢复原参数
      * 收到 429 → 标记全局暂停 N 秒（N 来自 Retry-After 或默认 30s），
        所有线程在下次请求前等待

    线程安全：所有方法加锁。状态由 fetch_top_results 的 _fetch_one 回调驱动。
    """

    def __init__(self, initial_delay: float, initial_concurrency: int,
                 failure_threshold: int = 3, pause_seconds: float = 30.0,
                 max_delay: float = 10.0):
        """v2.2.0：参数化 failure_threshold/pause_seconds/max_delay。

        默认值与 v2.1.1 硬编码一致，保持向后兼容。
        """
        self._lock = threading.Lock()
        self._delay = initial_delay
        self._initial_delay = initial_delay
        self._concurrency = initial_concurrency
        self._initial_concurrency = initial_concurrency
        self._consecutive_failures = 0
        self._consecutive_successes = 0
        self._global_pause_until = 0.0  # time.monotonic() 时间戳
        self._failure_threshold = failure_threshold
        self._pause_seconds = pause_seconds
        self._max_delay = max_delay
        # v2.2.2：真实并发门控。计数信号量约束"瞬时在飞请求峰值"，
        # _in_flight 结合当前 concurrency 判断是否应放行新请求——退避降
        # 并发后，新请求会被快速拒绝（限流语义），而不是名义降并发。
        self._semaphore = threading.BoundedSemaphore(initial_concurrency)
        self._in_flight = 0

    @property
    def delay(self) -> float:
        with self._lock:
            return self._delay

    @property
    def concurrency(self) -> int:
        with self._lock:
            return self._concurrency

    def report_success(self) -> None:
        with self._lock:
            self._consecutive_failures = 0
            self._consecutive_successes += 1
            # 连续 5 次成功 → 逐步恢复
            if self._consecutive_successes >= 5:
                self._consecutive_successes = 0
                self._delay = max(self._initial_delay, self._delay / 2)
                if self._concurrency < self._initial_concurrency:
                    self._concurrency = min(self._initial_concurrency,
                                            self._concurrency * 2)

    def report_failure(self, error_msg: str = "",
                       error_code: str = None) -> None:
        """报告一次失败，触发自适应退避。

        v2.1.0：优先用结构化 error_code 检测 429/限流（E_RATE_LIMIT），
        回退到字符串匹配兼容旧调用方。原代码仅检查 "429" 字面量，
        "Too Many Requests" 会漏判。
        """
        with self._lock:
            self._consecutive_successes = 0
            self._consecutive_failures += 1
            # 429/限流 → 全局暂停 30s
            # 优先用 error_code，回退到字符串匹配（兼容无 error_code 的旧调用）
            is_rate_limit = (error_code == E_RATE_LIMIT or
                             "429" in error_msg.lower() or
                             "rate limit" in error_msg.lower())
            if is_rate_limit:
                self._global_pause_until = time.monotonic() + self._pause_seconds
            # 连续失败达阈值 → 退避 + 降并发。
            # v2.5.0：threshold <= 0 表示"禁用自适应退避"——只统计失败次数
            # （供 stats()/报告展示），不再触发翻倍延迟与降并发。原实现
            # threshold=0 时 ``0 >= 0`` 恒真，首次失败即退避，与 CLI 帮助
            # 声称的 "0 disables throttling" 相反。
            if self._failure_threshold > 0 and \
                    self._consecutive_failures >= self._failure_threshold:
                self._consecutive_failures = 0
                self._delay = min(self._delay * 2, self._max_delay)
                self._concurrency = max(1, self._concurrency // 2)

    def wait_if_paused(self) -> None:
        """如果处于全局暂停期，阻塞等待直到解除。请求前调用。"""
        with self._lock:
            remaining = self._global_pause_until - time.monotonic()
        if remaining > 0:
            logger.info(f"  [THROTTLE] Global pause: waiting {remaining:.1f}s (429)")
            time.sleep(remaining)

    def acquire_slot(self, timeout: float = 0.05) -> bool:
        """获取一个并发执行槽位（v2.2.2 真实并发门控）。

        线程池规模（ThreadPoolExecutor max_workers）创建时一次性固定，
        无法随退避动态缩小。槽位机制在"请求真正发出前"做门控：
        信号量约束瞬时峰值不超过初始并发；``_in_flight`` 结合当前
        ``concurrency`` 判断——退避降并发后，即使信号量有空位，只要在飞
        请求数已 >= 当前并发目标，新请求也会被快速拒绝（返回 False），
        由调用方跳过本次抓取，实现持久降并发。

        返回 True 表示拿到槽位，调用方必须在 finally 中 release_slot()。
        """
        if not self._semaphore.acquire(timeout=timeout):
            return False
        with self._lock:
            if self._in_flight >= self._concurrency:
                self._semaphore.release()
                return False
            self._in_flight += 1
            return True

    def release_slot(self) -> None:
        """释放并发槽位。必须与 acquire_slot 成对使用。"""
        with self._lock:
            self._in_flight = max(0, self._in_flight - 1)
        try:
            self._semaphore.release()
        except ValueError:
            pass

    def stats(self) -> dict:
        """返回当前状态快照，供 --fetch-report 使用。"""
        with self._lock:
            return {
                "current_delay": round(self._delay, 3),
                "current_concurrency": self._concurrency,
                "consecutive_failures": self._consecutive_failures,
                "consecutive_successes": self._consecutive_successes,
                "global_paused": time.monotonic() < self._global_pause_until,
            }


def fetch_top_results(results: dict, count: int, timeout: int = 10,
                       concurrency: int = 5, auth_headers: dict = None,
                       max_retries: int = 3, max_size: int = None,
                       request_delay: float = 0.3,
                       referer: str = None,
                       fallback_enabled: bool = True,
                       throttle: "AdaptiveThrottle" = None,
                       total_chars: int = 0) -> list:
    """Fetch full text of top N result pages concurrently.

    Features:
    - Retries transient errors with exponential backoff (in fetch_url)
    - v2.0.0 自适应限流：连续失败自动降并发+加延迟，429 全局暂停
    - v2.0.0 Wayback 兜底：404/403/超时自动尝试 Wayback Machine
    - v2.0.0 反爬检测：WAF 指纹库识别 Cloudflare/Imperva/PerimeterX 等
    - Falls back to browser User-Agent if blocked (in fetch_url)
    - Small delay between requests to avoid rate limits

    v2.5.0 全局字符预算（``total_chars``）：
    按原始结果顺序从顶部开始分配字符额度——前一页实际消费后剩余的预算
    留给下一页（顺序填满，保证最重要的结果拿到完整正文）。预算耗尽后
    剩余 URL 返回 ``status="skipped"``（不发请求），让 AI Agent 明确知道
    是预算跳过而非抓取失败。默认 0 = 不限制。

    Args:
        referer: Referer URL（v2.0.0，通常设为 SearXNG 实例 URL）
        fallback_enabled: 是否启用 Wayback 兜底（默认 True）
        throttle: 外部传入的 AdaptiveThrottle 实例（可选；不传则内部创建）
        total_chars: 全局字符预算（v2.5.0；0 = 不限）
    """
    urls = []
    seen = set()
    for r in results.get("results", []):
        url = r.get("url", "")
        if url and url not in seen:
            urls.append(url)
            seen.add(url)
        if len(urls) >= count:
            break

    if not urls:
        return []

    # 自适应限流器（外部未传入则创建）
    if throttle is None:
        throttle = AdaptiveThrottle(request_delay, concurrency)

    logger.info(f"\nFetching {len(urls)} result pages "
                f"(timeout={timeout}s, retries={max_retries}, "
                f"delay={throttle.delay}s, concurrency={throttle.concurrency})...")
    if total_chars and total_chars > 0:
        logger.info(f"Char budget: {total_chars:,} total (--fetch-total-chars)")
    fetched = []
    ok_count = [0]
    err_count = [0]
    skipped_count = [0]
    anti_bot_count = [0]
    fallback_count = [0]

    # v2.5.0 全局预算的线程安全计数器：_budget_remaining[0] 是剩余可用字符。
    # _budget_enabled[0] 区分"未启用预算"（total_chars<=0，无限）与"已耗尽"
    # （remaining=0）——前者放行所有 URL，后者跳过。_consume_budget 只在
    # 抓取成功后按实际 text_length 扣减，未用掉的额度自动留给下一页。
    _budget_enabled = [total_chars is not None and total_chars > 0]
    _budget_remaining = [total_chars if _budget_enabled[0] else 0]
    _budget_lock = threading.Lock()

    def _take_budget() -> "tuple":
        """分配本 URL 的字符上限。返回 (cap, allowed)。

        cap 为 None 表示预算未启用（无限）。allowed=False 表示预算已耗尽，
        调用方应跳过本 URL（status="skipped"，不发请求）。
        """
        if not _budget_enabled[0]:
            return None, True
        with _budget_lock:
            remaining = _budget_remaining[0]
            if remaining <= 0:
                return 0, False
            return remaining, True

    def _consume_budget(used: int) -> None:
        """抓取成功后按实际消费的字符数扣减预算。"""
        with _budget_lock:
            _budget_remaining[0] = max(0, _budget_remaining[0] - used)

    def _fetch_one(u: str) -> dict:
        """Fetch one URL with adaptive throttling to avoid rate limiting."""
        # 全局暂停检查（429 触发）
        throttle.wait_if_paused()
        # v2.5.0 预算检查放在取槽位之前：预算耗尽时连并发槽位都不占用
        cap, allowed = _take_budget()
        if not allowed:
            skipped_count[0] += 1
            logger.info(f"  [BUDGET] char budget exhausted, skipping {u[:55]}")
            return {"url": u, "status": "skipped",
                    "error": "char budget exhausted (--fetch-total-chars)",
                    "text": "", "text_length": 0, "truncated": False,
                    "anti_bot_detected": False, "waf_type": None,
                    "fallback_used": None,
                    "title": None, "latency": None}
        # v2.2.2：真实并发门控。退避降并发后，超出当前并发目标的新请求
        # 在这里被拒绝（不占用 fetch_page），实现持久降并发而非名义降并发。
        if not throttle.acquire_slot():
            logger.info(f"  [THROTTLE] concurrency cap ({throttle.concurrency}) "
                        f"reached, skipping {u[:55]}")
            return {"url": u, "status": "error",
                    "error": "Throttled: concurrency limit reached",
                    "error_code": E_RATE_LIMIT,
                    "text": "", "text_length": 0, "truncated": False,
                    "anti_bot_detected": False, "waf_type": None,
                    "fallback_used": None,
                    "title": None, "latency": None}
        try:
            # 自适应延迟
            d = throttle.delay
            if d > 0:
                time.sleep(d * random.uniform(0.5, 1.5))
            result = fetch_page(u, timeout=timeout, auth_headers=auth_headers,
                                max_retries=max_retries, max_size=max_size,
                                referer=referer, fallback_enabled=fallback_enabled)
            if result["status"] == "ok":
                ok_count[0] += 1
                throttle.report_success()
                # v2.5.0 字符预算：按分配到的上限截断正文（提取后语义截断），
                # 再按实际消费扣减预算。cap 为 None 表示未启用预算。
                if cap:
                    text = result.get("text", "")
                    if len(text) > cap:
                        result["text"] = text[:cap]
                        result["text_length"] = cap
                        result["truncated"] = True
                    _consume_budget(result["text_length"])
                trunc = ", TRUNCATED" if result.get("truncated") else ""
                ua_note = ""
                if result.get("user_agent_used") != USER_AGENT:
                    ua_note = " [fallback UA]"
                fb_note = " [wayback]" if result.get("fallback_used") else ""
                if fb_note:
                    fallback_count[0] += 1
                logger.info(f"  [OK] {u[:55]} ({result['text_length']:,} chars"
                            f"{trunc}{ua_note}{fb_note})")
            else:
                err_count[0] += 1
                throttle.report_failure(result.get("error", ""),
                                        error_code=result.get("error_code"))
                # 统计反爬拦截
                if result.get("anti_bot_detected"):
                    anti_bot_count[0] += 1
                logger.error(f"  [ERR] {u[:55]} ({result.get('error', 'unknown')})")
            return result
        finally:
            throttle.release_slot()

    with ThreadPoolExecutor(max_workers=min(throttle.concurrency, len(urls))) as ex:
        future_map = {ex.submit(_fetch_one, u): u for u in urls}
        for future in as_completed(future_map):
            try:
                result = future.result()
                fetched.append(result)
            except Exception as e:
                u = future_map[future]
                fetched.append({"url": u, "status": "error", "error": str(e),
                                "error_code": classify_fetch_error(e),
                                "text": "", "text_length": 0, "truncated": False,
                                "anti_bot_detected": False, "waf_type": None,
                                "fallback_used": None,
                                "title": None, "latency": None})
                logger.error(f"  [ERR] {u[:55]} (thread error: {e})")

    # Reorder to match original result order
    url_order = {u: i for i, u in enumerate(urls)}
    fetched.sort(key=lambda f: url_order.get(f["url"], 999))

    logger.info(f"Fetched: {ok_count[0]} ok, {err_count[0]} errors"
                f" (anti-bot: {anti_bot_count[0]}, wayback: {fallback_count[0]}, "
                f"skipped: {skipped_count[0]})")
    return fetched


def deduplicate_fetched_content(fetched: list, threshold: float = 0.85) -> int:
    """按正文相似度去重已抓取页面（v2.5.0 --dedup-fetched-content）。

    镜像/转载站点的正文与原文高度相似但 URL 不同——URL 级去重
    （deduplicate_results）无法合并。这里对每个成功页面（status="ok"）
    的正文计算 SimHash 指纹，与已保留的页面两两比较，后出现的近似重复
    条目标记为 ``status="duplicate"``（text 清空，保留 url/title/error 供
    AI 查看原因），避免 AI 阅读同一内容的多个副本浪费 token。

    只对 ok 条目参与去重；error/skipped/duplicate 条目原样保留。
    ``fetched`` 保持原始结果顺序（首个出现的保留，通常 score 最高）。
    返回标记为 duplicate 的条目数。
    """
    kept = []
    dup_count = 0
    for f in fetched:
        if f.get("status") != "ok":
            kept.append(f)
            continue
        text = f.get("text", "")
        is_dup = False
        for k in kept:
            if k.get("status") != "ok":
                continue
            if texts_are_similar(text, k.get("text", ""), threshold=threshold):
                is_dup = True
                break
        if is_dup:
            dup_count += 1
            f["status"] = "duplicate"
            f["text"] = ""
            f["text_length"] = 0
            f["error"] = f"duplicate content (similar to {k.get('url', '')})"
            kept.append(f)
        else:
            kept.append(f)
    return dup_count


def _emit_fetch_report(fetched: list, throttle: "AdaptiveThrottle",
                       fmt: str = "text") -> None:
    """输出结构化抓取报告到 stderr。

    让 AI Agent 可程序化分析抓取过程：哪些 URL 被反爬拦截、用了什么兜底、
    自适应限流如何调整。

    ``fmt``:
      * ``"text"``（默认）：人类可读表格 + JSON 摘要行（v2.0.0 行为，向后兼容）
      * ``"json"``：完整 JSON 对象（items 数组 + summary），便于 Agent 解析
      * 其他值按 ``"text"`` 处理
    """
    if fmt == "json":
        _emit_fetch_report_json(fetched, throttle)
    else:
        _emit_fetch_report_text(fetched, throttle)


def _emit_fetch_report_text(fetched: list, throttle: "AdaptiveThrottle") -> None:
    """文本表格 + JSON 摘要行（v2.0.0 原始行为，向后兼容）。"""
    out = sys.stderr
    lines = []
    lines.append("\n" + "=" * 72)
    lines.append("FETCH REPORT (v2.0.0)")
    lines.append("=" * 72)

    # Per-URL 表
    header = f"{'URL':<45} {'Status':<8} {'WAF':<12} {'Fallback':<10} {'Chars':>10}"
    lines.append(header)
    lines.append("-" * len(header))
    for f in fetched:
        url = f.get("url", "")[:44]
        status = "OK" if f.get("status") == "ok" else "ERR"
        waf = f.get("waf_type") or "-"
        fb = f.get("fallback_used") or "-"
        chars = f.get("text_length", 0)
        lines.append(f"{url:<45} {status:<8} {waf:<12} {fb:<10} {chars:>10,}")

    # 统计摘要
    total = len(fetched)
    ok = sum(1 for f in fetched if f.get("status") == "ok")
    err = total - ok
    skipped = sum(1 for f in fetched if f.get("status") == "skipped")
    dup = sum(1 for f in fetched if f.get("status") == "duplicate")
    err = err - skipped - dup  # skipped/duplicate 不是错误，单独统计
    anti_bot = sum(1 for f in fetched if f.get("anti_bot_detected"))
    wayback = sum(1 for f in fetched if f.get("fallback_used") == "wayback")
    lines.append("-" * len(header))
    lines.append(f"Total: {total} | OK: {ok} | Error: {err} | "
                 f"Skipped: {skipped} | Duplicate: {dup} | "
                 f"Anti-bot blocked: {anti_bot} | Wayback recovered: {wayback}")

    # 自适应限流状态
    s = throttle.stats()
    lines.append(f"Throttle: delay={s['current_delay']}s "
                 f"concurrency={s['current_concurrency']} "
                 f"paused={s['global_paused']} "
                 f"consec_fail={s['consecutive_failures']} "
                 f"consec_ok={s['consecutive_successes']}")

    # JSON 摘要（一行，便于 Agent 解析）
    summary = {
        "total": total, "ok": ok, "error": err, "skipped": skipped,
        "duplicate": dup,
        "anti_bot_blocked": anti_bot, "wayback_recovered": wayback,
        "throttle": s,
    }
    lines.append("JSON: " + json.dumps(summary, ensure_ascii=False))
    lines.append("=" * 72 + "\n")
    print("\n".join(lines), file=out)


def _emit_fetch_report_json(fetched: list, throttle: "AdaptiveThrottle") -> None:
    """完整 JSON 报告（items 数组 + summary），输出到 stderr。

    每个 URL 一个对象，包含 url / status / title / content_length /
    error / error_code / latency / fetched_at 等字段。v2.3.0 起
    ``title`` 与 ``latency`` 由 fetch_page 采集填充（此前恒为 None）。
    """
    from datetime import datetime, timezone

    fetched_at = datetime.now(timezone.utc).isoformat()

    # 每个 URL 一个对象
    items = []
    for f in fetched:
        items.append({
            "url": f.get("url", ""),
            "final_url": f.get("final_url"),
            "status": f.get("status", ""),
            "title": f.get("title"),            # v2.3.0: fetch_page 已采集
            "content_length": f.get("text_length", 0),
            "content_type": f.get("content_type"),
            "error": f.get("error"),
            "error_code": f.get("error_code"),
            "latency": f.get("latency"),         # v2.3.0: fetch_page 已计时
            "truncated": f.get("truncated", False),
            "fetched_at": f.get("fetched_at") or fetched_at,
            "waf_type": f.get("waf_type"),
            "fallback_used": f.get("fallback_used"),
            "anti_bot_detected": f.get("anti_bot_detected", False),
        })

    # 统计摘要
    total = len(fetched)
    ok = sum(1 for f in fetched if f.get("status") == "ok")
    err = total - ok
    skipped = sum(1 for f in fetched if f.get("status") == "skipped")
    dup = sum(1 for f in fetched if f.get("status") == "duplicate")
    err = err - skipped - dup  # skipped/duplicate 不是错误，单独统计
    anti_bot = sum(1 for f in fetched if f.get("anti_bot_detected"))
    wayback = sum(1 for f in fetched if f.get("fallback_used") == "wayback")

    report = {
        "format": "json",
        "total": total,
        "ok": ok,
        "error": err,
        "skipped": skipped,
        "duplicate": dup,
        "anti_bot_blocked": anti_bot,
        "wayback_recovered": wayback,
        "throttle": throttle.stats(),
        "items": items,
    }
    print(json.dumps(report, ensure_ascii=False), file=sys.stderr)


# ----- Output formatting -----

def deduplicate_results(results: dict) -> dict:
    """Remove duplicate results by normalized URL, keeping first occurrence.

    Different engines often return the same URL. The first occurrence keeps
    its engine/score metadata; subsequent duplicates are dropped.

    URL normalization:
      * lowercase scheme + host
      * strip fragment (``#...``)
      * drop common tracking query params (``utm_*``, ``gclid``, ``fbclid``,
        ``mc_*``, ``ref``, ``ref_*``)
      * sort remaining params so ``?b=2&a=1`` matches ``?a=1&b=2``

    Results without a URL are kept as-is (never deduped). Mutates
    ``results['results']`` in place and returns ``results`` for chaining.
    """
    if not results.get("results"):
        return results

    # 跟踪参数前缀。裸 "ref" 过于宽泛（会误删 reference/refcode 等正常参数），
    # 收紧为 "ref_" 只匹配 ref_source/ref_campaign 等跟踪参数。
    TRACKING_PREFIXES = ("utm_", "gclid", "fbclid", "mc_", "ref_")

    def _normalize_url(url: str) -> str:
        try:
            p = urllib.parse.urlparse(url)
            scheme = p.scheme.lower()
            netloc = p.netloc.lower()
            qs = urllib.parse.parse_qsl(p.query, keep_blank_values=False)
            qs = [(k, v) for k, v in qs
                  if not any(k.lower().startswith(pref) for pref in TRACKING_PREFIXES)]
            qs.sort()
            query = urllib.parse.urlencode(qs)
            return urllib.parse.urlunparse((scheme, netloc, p.path, p.params,
                                            query, ""))
        except Exception:
            return url

    seen = set()
    deduped = []
    for r in results["results"]:
        url = r.get("url", "")
        key = _normalize_url(url) if url else ""
        if key and key in seen:
            continue
        if key:
            seen.add(key)
        deduped.append(r)

    before = len(results["results"])
    results["results"] = deduped
    if before != len(deduped):
        logger.info(f"Dedup: {before} -> {len(deduped)} results (by normalized URL)")
    return results


def deduplicate_similar(results: dict, threshold: float = 0.85) -> int:
    """对搜索结果进行相似度去重（基于标题 SimHash）。

    遍历结果列表，对每对结果用 :func:`is_similar` 判断，保留先出现的
    （通常 score 更高），删除相似的后续结果。

    Args:
        results: 含 ``"results"`` 列表的 dict，会原地修改。
        threshold: 相似度阈值，越高越严格（默认 0.85）。

    Returns:
        删除的条目数。

    Note:
        O(n²) 复杂度，结果列表通常 < 100，可接受。若 > 500 则跳过并
        log warning，避免在大结果集上卡顿。
    """
    items = results.get("results", [])
    if not items:
        return 0
    n = len(items)
    if n > 500:
        logger.warning(
            f"Similarity dedup skipped: {n} results > 500 "
            f"(O(n²) would be too slow)"
        )
        return 0
    to_remove = set()
    for i in range(n):
        if i in to_remove:
            continue
        for j in range(i + 1, n):
            if j in to_remove:
                continue
            if is_similar(items[i], items[j], threshold=threshold):
                to_remove.add(j)
    if not to_remove:
        return 0
    removed = len(to_remove)
    results["results"] = [items[i] for i in range(n) if i not in to_remove]
    logger.info(
        f"Similarity dedup: removed {removed} similar results "
        f"(threshold={threshold}, {n} -> {n - removed})"
    )
    return removed


def sort_results(results: dict, sort_by: str) -> dict:
    """Sort results by the given key.

    * ``score``  — descending (highest first); entries without score keep
                   their relative order at the end.
    * ``date``   — descending (newest first) by ``published_date`` (ISO
                   strings sort lexicographically); entries without a date
                   keep their relative order at the end.
    * ``engine`` — ascending (alphabetical by engine name).
    * ``none``   — no-op (preserve original order).

    Mutates ``results['results']`` in place and returns ``results``.
    """
    if not results.get("results") or sort_by == "none":
        return results

    rs = results["results"]
    if sort_by == "score":
        with_score = [r for r in rs if r.get("score") is not None]
        without = [r for r in rs if r.get("score") is None]
        with_score.sort(key=lambda r: r["score"], reverse=True)
        results["results"] = with_score + without
    elif sort_by == "date":
        with_date = [r for r in rs if r.get("published_date")]
        without = [r for r in rs if not r.get("published_date")]
        with_date.sort(key=lambda r: r["published_date"], reverse=True)
        results["results"] = with_date + without
    elif sort_by == "engine":
        rs.sort(key=lambda r: r.get("engine") or "")
        results["results"] = rs
    return results


def filter_results_by_domain(results: dict, include_domains: list = None,
                              exclude_domains: list = None) -> dict:
    """Filter search results by domain (allowlist or blocklist).

    Modifies ``results['results']`` in place and returns ``results`` for
    chaining. Matching is case-insensitive and ignores a leading ``www.``
    so ``example.com`` matches both ``example.com`` and ``www.example.com``.

    ``include_domains`` (allowlist) takes precedence: if non-empty, only
    results whose domain matches are kept. ``exclude_domains`` (blocklist)
    then drops any remaining matches. Empty inputs are no-ops.
    """
    if not results.get("results"):
        return results
    if not include_domains and not exclude_domains:
        return results

    def _get_domain(url: str) -> str:
        try:
            return urllib.parse.urlparse(url).netloc.lower()
        except Exception:
            return ""

    def _normalize(d: str) -> str:
        return d.lower().lstrip(".").lstrip("www.")

    include_set = {_normalize(d) for d in include_domains} if include_domains else None
    exclude_set = {_normalize(d) for d in exclude_domains} if exclude_domains else None

    filtered = []
    for r in results["results"]:
        domain = _get_domain(r.get("url", ""))
        # Strip leading www. for matching
        match_domain = domain[4:] if domain.startswith("www.") else domain

        if include_set is not None and match_domain not in include_set:
            continue
        if exclude_set is not None and match_domain in exclude_set:
            continue
        filtered.append(r)

    results["results"] = filtered
    return results


def format_brief(results: dict, snippet_len: int = 0) -> str:
    """Format results as brief text. snippet_len=0 means no truncation."""
    lines = []
    for i, r in enumerate(results.get("results", []), 1):
        title = r.get("title", "No title")
        url = r.get("url", "")
        content = r.get("content", "")
        lines.append(f"{i}. {title}")
        lines.append(f"   {url}")
        if content:
            if snippet_len > 0:
                content = content[:snippet_len]
            lines.append(f"   {content}")
        lines.append("")

    # Include suggestions if present
    sugs = results.get("suggestions", [])
    if sugs:
        lines.append(f"Suggestions: {', '.join(sugs)}")

    # Include answers if present
    for ans in results.get("answers", []):
        lines.append(f"Answer: {ans}")

    return "\n".join(lines)


def format_urls(results: dict) -> str:
    """Format results as plain URL list."""
    return "\n".join(r.get("url", "") for r in results.get("results", []) if r.get("url"))


# ----- Media-category CSV columns (v2.5.0) -----
# SearXNG 的 images 类别结果（template="images.html"）带 img_src/thumbnail_src/
# resolution/source，videos 类别（template="videos.html"）带 iframe_src/
# thumbnail_src。--format json 已透传这些字段（原始 dict 原样序列化），但
# CSV 固定列会丢弃。以下辅助让 CSV 在结果含媒体字段时自动追加对应列，
# 表格分析（图片缩略图链接、视频内嵌 URL）直接可用。
_MEDIA_COLUMNS = ["img_src", "thumbnail_src", "resolution", "iframe_src", "source"]


def _detect_media_columns(results_iter) -> list:
    """返回结果集中实际存在的媒体列子集（保持 _MEDIA_COLUMNS 顺序）。

    仅当至少一个结果包含非空值时才列出该列，避免 CSV 出现整列空白。
    """
    present = set()
    for r in results_iter:
        for col in _MEDIA_COLUMNS:
            if r.get(col):
                present.add(col)
    return [c for c in _MEDIA_COLUMNS if c in present]


def _format_results(results: dict, args) -> str:
    """Format a results dict into the output string selected by args.format.

    Pulled out of :func:`main` so the batch runner (--queries-file) can reuse
    the exact same formatting for each per-query block.
    """
    if args.format == "json":
        results["schema_version"] = SCHEMA_VERSION
        return json.dumps(results, indent=2, ensure_ascii=False)
    if args.format == "urls":
        return format_urls(results)
    if args.format == "csv":
        out = io.StringIO()
        writer = csv.writer(out, lineterminator="\n")
        rs = results.get("results", [])
        media_cols = _detect_media_columns(rs)
        writer.writerow(["title", "url", "engine", "score",
                         "published_date", "content"] + media_cols)
        for r in rs:
            writer.writerow([
                r.get("title", ""),
                r.get("url", ""),
                r.get("engine", ""),
                r.get("score", "") if r.get("score") is not None else "",
                r.get("published_date", ""),
                r.get("content", ""),
            ] + [r.get(c, "") for c in media_cols])
        return out.getvalue().rstrip()
    # brief (also the safe fallthrough)
    output = format_brief(results, snippet_len=args.snippet_len)
    if args.fetch > 0 and results.get("fetched"):
        output += "\n\n" + "=" * 60 + "\n"
        output += f"FETCHED PAGES ({len(results['fetched'])} pages)\n"
        output += "=" * 60 + "\n"
        for f in results["fetched"]:
            output += f"\n--- {f['url']} ---\n"
            if f["status"] == "ok":
                output += f"{f['text']}\n"
            elif f["status"] == "skipped":
                output += f"[SKIPPED: {f.get('error', 'char budget exhausted')}]\n"
            elif f["status"] == "duplicate":
                output += f"[DUPLICATE: {f.get('error', 'similar content')}]\n"
            else:
                output += f"[ERROR: {f.get('error', 'unknown')}]\n"
    return output


def _build_params(query: str, args) -> dict:
    """Build the SearXNG API params dict for one query.

    Centralized so the single-query path and the --queries-file batch path
    construct identical params (and thus share cache entries).
    """
    params = {"q": query, "format": "json"}
    if args.categories:
        params["categories"] = _normalize_csv(args.categories)
    if args.language:
        params["language"] = args.language
    if args.pageno != 1:
        params["pageno"] = str(args.pageno)
    if args.time_range and args.time_range != "none":
        params["time_range"] = args.time_range
    if args.safesearch is not None:
        params["safesearch"] = str(args.safesearch)
    if args.engines:
        params["engines"] = _normalize_csv(args.engines)
    return params


def _warn_unresponsive_engines(results: dict, query: str,
                               result_count: int = None) -> None:
    """检测并提示实例侧引擎挂起/限流（v2.1.1）。

    SearXNG JSON API 返回的 ``unresponsive_engines`` 字段格式为::

        [["brave", "Suspended: too many requests"],
         ["duckduckgo", "CAPTCHA"]]

    当该字段非空时，说明实例内多个引擎被上游限流挂起。此时：
      1. 用 logger.warning 输出挂起的引擎列表及原因（到 stderr，
         不污染 stdout 数据流）
      2. 如果结果数较少，建议用 --engines 限定未挂起引擎

    这是真实运营问题（见执行问题记录 #3）：连续查询后 brave/duckduckgo/
    startpage 等引擎会被上游限流挂起，导致结果骤减或全空。让用户及时
    感知引擎状态，避免误判为"无结果"而反复重试触发更严厉限流。
    """
    unresponsive = results.get("unresponsive_engines", [])
    if not unresponsive:
        return
    # 格式化引擎列表：兼容 [engine, reason] 和 [engine] 两种格式
    parts = []
    for entry in unresponsive:
        if isinstance(entry, (list, tuple)) and len(entry) >= 2:
            parts.append(f"{entry[0]} ({entry[1]})")
        elif isinstance(entry, (list, tuple)) and len(entry) == 1:
            parts.append(str(entry[0]))
        else:
            parts.append(str(entry))
    engine_list = ", ".join(parts)
    logger.warning(f"Instance engines unresponsive: {engine_list}")
    # 结果数少 + 引擎挂起 → 建议规避
    if result_count is not None and result_count < 3 and len(unresponsive) >= 2:
        # 找出可能未挂起的常见引擎提示
        logger.warning("Hint: multiple engines suspended — consider using "
                       "--engines to target responsive ones, or wait before retrying")


def _run_single_query(query: str, args, instance_urls: list,
                      auth_headers: dict, ttl_seconds: int):
    """Run one query end-to-end: search → limit → domain-filter → fetch.

    Returns ``(results_dict, error_str, error_code)``。成功时后两者为 None。
    ``error_code`` 是结构化错误码（E_NETWORK/E_AUTH 等），让 AI Agent
    程序化判断错误类型。Cache hits 跳过网络。后处理（limit/filter/fetch）
    总是执行，保证 batch 调用方看到与单查询一致的形状。
    """
    params = _build_params(query, args)

    pages_to_fetch = getattr(args, "pages", 1) or 1
    if pages_to_fetch < 1:
        pages_to_fetch = 1

    emit_progress("start", query=query, instances=len(instance_urls))

    # v2.2.2：区分"实时查询"与"缓存命中"。原实现用循环末次赋值的
    # ``cached`` 变量判断是否实时查询——多页路径下该变量保存的是最后一页
    # 的状态，导致：最后一页命中缓存但前页实时查询时，unresponsive_engines
    # 警告被错误跳过；反之仅最后一页未命中时误触发。用独立布尔标记精确
    # 跟踪"本次运行是否发起了至少一次实时查询"。
    performed_live_query = False

    # v2.2.0：--pages N 多页聚合。循环 pageno=1..N，每页独立缓存
    # （cache key 含 pageno），合并后统一 dedup/sort/max-results。
    if pages_to_fetch == 1:
        # 单页路径（保持原行为，最常见场景无额外开销）
        cached = cache_module.get(params, ttl_seconds) if ttl_seconds > 0 else None
        if cached is not None:
            logger.info(f"[cache hit] q={query!r} TTL={args.cache_ttl}min, skipping network")
            emit_progress("cache_hit", query=query, ttl=args.cache_ttl)
            results = cached
        else:
            try:
                results = search_multi(
                    instance_urls, params,
                    method=args.method,
                    timeout=args.timeout,
                    retry_per=args.retry,
                    auth_headers=auth_headers,
                    parallel=not args.serial,
                    encoding=getattr(args, "encoding", None),
                )
            except Exception as e:
                err_code = classify_error(e)
                emit_progress("error", error=str(e), error_code=err_code, query=query)
                return None, str(e), err_code
            performed_live_query = True
            if ttl_seconds > 0:
                cache_module.put(params, results, ttl_seconds)
                logger.info(f"[cache stored] q={query!r} TTL={args.cache_ttl}min")
                emit_progress("cache_store", query=query, ttl=args.cache_ttl)
    else:
        # 多页聚合路径
        logger.info(f"Multi-page: fetching {pages_to_fetch} pages for q={query!r}")
        merged_results_list = []
        base_results = None
        page_errors = 0
        for page_no in range(1, pages_to_fetch + 1):
            page_params = dict(params)
            if page_no != 1:
                page_params["pageno"] = str(page_no)

            cached = cache_module.get(page_params, ttl_seconds) if ttl_seconds > 0 else None
            if cached is not None:
                logger.info(f"[cache hit] q={query!r} page={page_no}")
                emit_progress("cache_hit", query=query, ttl=args.cache_ttl, page=page_no)
                page_results = cached
            else:
                try:
                    page_results = search_multi(
                        instance_urls, page_params,
                        method=args.method,
                        timeout=args.timeout,
                        retry_per=args.retry,
                        auth_headers=auth_headers,
                        parallel=not args.serial,
                        encoding=getattr(args, "encoding", None),
                    )
                except Exception as e:
                    page_errors += 1
                    logger.warning(f"Page {page_no} failed: {e}")
                    emit_progress("page_fail", query=query, page=page_no,
                                  error=str(e), error_code=classify_error(e))
                    continue
                performed_live_query = True
                if ttl_seconds > 0:
                    cache_module.put(page_params, page_results, ttl_seconds)
                    emit_progress("cache_store", query=query, ttl=args.cache_ttl, page=page_no)

            if base_results is None:
                base_results = page_results
            merged_results_list.extend(page_results.get("results", []))
            emit_progress("page_ok", query=query, page=page_no,
                          results=len(page_results.get("results", [])))

        if base_results is None:
            # 所有页都失败了
            return None, f"All {pages_to_fetch} pages failed", E_NETWORK

        # 合并所有页的结果，后续统一去重排序
        results = base_results
        results["results"] = merged_results_list
        results["pages_fetched"] = pages_to_fetch - page_errors
        results["page_errors"] = page_errors
        logger.info(f"Multi-page merged: {len(merged_results_list)} results "
                    f"from {pages_to_fetch - page_errors}/{pages_to_fetch} pages")

    # v2.1.1：检测实例侧引擎挂起/限流（仅在实时查询后提示，cache hit 时
    # unresponsive_engines 信息可能已过期）
    if performed_live_query:
        _warn_unresponsive_engines(results, query,
                                   result_count=len(results.get("results", [])))

    # Dedup (default on; --no-dedup disables) then sort, both BEFORE limit
    # so --max-results keeps the highest-scoring / newest items.
    if not args.no_dedup:
        deduplicate_results(results)
    # 相似度去重（默认关闭，需 --similarity-dedup 启用）
    if getattr(args, "similarity_dedup", False):
        deduplicate_similar(results, threshold=args.similarity_threshold)
    sort_results(results, args.sort_by)

    if args.max_results and "results" in results:
        results["results"] = results["results"][:args.max_results]

    if args.include_domain or args.exclude_domain:
        include_list = ([d.strip() for d in args.include_domain.split(",") if d.strip()]
                        if args.include_domain else None)
        exclude_list = ([d.strip() for d in args.exclude_domain.split(",") if d.strip()]
                        if args.exclude_domain else None)
        before = len(results.get("results", []))
        filter_results_by_domain(results, include_domains=include_list,
                                 exclude_domains=exclude_list)
        after = len(results.get("results", []))
        logger.info(f"Domain filter: {before} -> {after} results")

    if args.fetch > 0 and results.get("results"):
        emit_progress("fetch_start", count=args.fetch)
        # v2.0.0: Referer 默认设为首个实例 URL，伪装流量来自搜索引擎
        referer = getattr(args, "referer", None)
        if referer is None and instance_urls:
            referer = instance_urls[0]
        # v2.0.0: 创建共享 throttle 实例，用于 --fetch-report 输出
        request_delay = getattr(args, "request_delay", 0.3)
        fetch_throttle = AdaptiveThrottle(
            request_delay, min(5, args.fetch),
            failure_threshold=getattr(args, "throttle_failure_threshold", 3),
            pause_seconds=getattr(args, "throttle_pause_seconds", 30),
            max_delay=getattr(args, "throttle_max_delay", 10))
        fetched = fetch_top_results(
            results, args.fetch,
            timeout=args.fetch_timeout,
            auth_headers=auth_headers,
            max_retries=args.fetch_retries,
            max_size=args.max_size,
            request_delay=request_delay,
            referer=referer,
            fallback_enabled=not getattr(args, "no_fallback", False),
            throttle=fetch_throttle,
            total_chars=getattr(args, "fetch_total_chars", 0),
        )
        # v2.5.0: 正文相似度去重（--dedup-fetched-content），复用
        # --similarity-threshold 阈值。镜像/转载页标记为 status="duplicate"。
        if getattr(args, "dedup_fetched_content", False):
            dup_n = deduplicate_fetched_content(fetched,
                                                threshold=args.similarity_threshold)
            if dup_n:
                logger.info(f"Fetched-content dedup: {dup_n} duplicate page(s) "
                            f"dropped (threshold={args.similarity_threshold})")
        # Emit fetch_ok / fetch_fail / fetch_duplicate / fetch_skip events
        for f in fetched:
            if f.get("status") == "ok":
                emit_progress("fetch_ok", url=f.get("url", ""),
                              chars=f.get("text_length", 0),
                              fallback=f.get("fallback_used"))
            elif f.get("status") == "duplicate":
                emit_progress("fetch_duplicate", url=f.get("url", ""),
                              error=f.get("error", "duplicate content"))
            elif f.get("status") == "skipped":
                emit_progress("fetch_skip", url=f.get("url", ""),
                              error=f.get("error", "budget exhausted"))
            else:
                emit_progress("fetch_fail", url=f.get("url", ""),
                              error=f.get("error", "unknown"),
                              waf_type=f.get("waf_type"))
        results["fetched"] = fetched
        results["fetched_source"] = results.get("_fallback", "json")

        # v2.0.0: --fetch-report 输出到 stderr
        # 兼容旧的 bool（True→"text"，False→不输出）和新的 str（"text"/"json"）
        fetch_report_fmt = getattr(args, "fetch_report", False)
        if fetch_report_fmt is True:
            fetch_report_fmt = "text"
        elif not fetch_report_fmt:
            fetch_report_fmt = None
        elif fetch_report_fmt not in ("text", "json"):
            fetch_report_fmt = "text"
        if fetch_report_fmt:
            _emit_fetch_report(fetched, fetch_throttle, fmt=fetch_report_fmt)

    result_count = len(results.get("results", []))
    emit_progress("done", results=result_count, query=query)
    # 清理内部 _fallback 字段，避免泄漏到 JSON 输出。
    # fetched_source（对 AI 有用的公开字段）已在 --fetch 路径中设置。
    results.pop("_fallback", None)
    return results, None, None


def _get_output_schema():
    """Return the JSON Schema describing --format json output.

    Used by ``--dump-schema`` so AI agents can programmatically discover the
    output structure without parsing prose documentation.

    v2.3.0: fetched.items 字段补全（与 fetch_page 实际输出对齐——
    final_url/status/error_code/waf_type/fallback_used/title/latency 等），
    并新增 ``batch`` 与 ``research`` 两个属性描述对应模式的输出 shape。
    顶层 ``properties`` 仍以单查询为主，batch/research 为单独子树。
    """
    # 单查询结果条目（results[] 的元素）——batch/research 复用
    result_item = {
        "type": "object",
        "properties": {
            "title": {"type": "string"},
            "url": {"type": "string", "format": "uri"},
            "engine": {"type": "string", "description": "Source engine name."},
            "score": {"type": ["number", "null"]},
            "published_date": {"type": ["string", "null"]},
            "content": {"type": "string", "description": "Snippet/summary text."},
        },
        "required": ["title", "url"],
    }
    # --fetch N 时的 fetched[] 元素（与 search.fetch_page 返回对齐）
    fetched_item = {
        "type": "object",
        "properties": {
            "url": {"type": "string"},
            "final_url": {"type": ["string", "null"],
                          "description": "URL after redirects / Wayback."},
            "status": {"type": "string",
                        "enum": ["ok", "error", "skipped", "duplicate"],
                        "description": "'skipped' (v2.5.0) means the page was "
                                       "not fetched because the "
                                       "--fetch-total-chars budget was "
                                       "exhausted. 'duplicate' (v2.5.0) means "
                                       "the body was near-identical to an "
                                       "earlier page and was dropped by "
                                       "--dedup-fetched-content."},
            "content_type": {"type": ["string", "null"]},
            "text": {"type": "string", "description": "Extracted page text."},
            "text_length": {"type": "integer"},
            "truncated": {"type": "boolean"},
            "title": {"type": ["string", "null"],
                      "description": "Page <title>, when available."},
            "latency": {"type": ["number", "null"],
                        "description": "Fetch latency in seconds."},
            "user_agent_used": {"type": ["string", "null"]},
            "error": {"type": ["string", "null"]},
            "error_code": {"type": ["string", "null"]},
            "waf_type": {"type": ["string", "null"]},
            "fallback_used": {"type": ["string", "null"],
                              "description": "'wayback' when the Wayback "
                                             "Machine recovered the page."},
            "anti_bot_detected": {"type": "boolean"},
        },
        "required": ["url", "status"],
    }
    # 单查询输出
    single_query = {
        "type": "object",
        "properties": {
            "schema_version": {
                "type": "string",
                "const": SCHEMA_VERSION,
                "description": "Output schema version. Bump on breaking field changes.",
            },
            "query": {"type": "string", "description": "The search query string."},
            "number_of_results": {
                "type": "integer",
                "description": "Total matches reported by SearXNG (JSON path) or "
                               "count of parsed results (HTML fallback path).",
            },
            "results": {
                "type": "array",
                "description": "Search result items, ordered by relevance (score desc).",
                "items": result_item,
            },
            "unresponsive_engines": {
                "type": "array",
                "items": {"type": "string"},
                "description": "Engines that failed to respond.",
            },
            "suggestions": {
                "type": "array",
                "items": {"type": "string"},
                "description": "Related query suggestions from the instance.",
            },
            "answers": {
                "type": "array",
                "items": {"type": "string"},
                "description": "Direct answers from the instance.",
            },
            "fetched": {
                "type": "array",
                "description": "Present only when --fetch N is used. Page content "
                               "for the top N results.",
                "items": fetched_item,
            },
            "fetched_source": {
                "type": "string",
                "enum": ["json", "html"],
                "description": "Present only when --fetch is used. Indicates whether "
                               "search results came from JSON API or HTML fallback.",
            },
            "pages_fetched": {
                "type": ["integer", "null"],
                "description": "Present only when --pages N > 1. Pages that "
                               "succeeded (before cross-page merge).",
            },
        },
        "required": ["query", "results"],
    }
    # --queries-file 批量输出
    batch_schema = {
        "type": "object",
        "description": "Batch mode (--queries-file) output shape.",
        "properties": {
            "schema_version": {"type": "string", "const": SCHEMA_VERSION},
            "queries": {
                "type": "array",
                "description": "One entry per query, in file order. "
                               "Entry: {query, status: ok|error, results} on "
                               "success or {query, status: error, error, "
                               "error_code} on failure.",
                "items": {"type": "object",
                          "required": ["query", "status"],
                          "properties": {
                              "query": {"type": "string"},
                              "status": {"type": "string", "enum": ["ok", "error"]},
                              "results": single_query,
                              "error": {"type": "string"},
                              "error_code": {"type": "string"},
                          }},
            },
        },
        "required": ["schema_version", "queries"],
    }
    # --research 研究模式输出
    research_schema = {
        "type": "object",
        "description": "Research mode (--research TOPIC) output shape.",
        "properties": {
            "schema_version": {"type": "string", "const": SCHEMA_VERSION},
            "research_topic": {"type": "string"},
            "research_queries": {
                "type": "array",
                "description": "Expanded per-angle queries (deterministic rules).",
                "items": {"type": "object",
                          "properties": {
                              "angle": {"type": "string"},
                              "query": {"type": "string"},
                          }},
            },
            "queries": {"type": "array",
                        "description": "Per-angle results. Entry: {query, angle, "
                                       "status: ok|error, results|error, error_code}.",
                        "items": {"type": "object",
                                  "properties": {
                                      "query": {"type": "string"},
                                      "angle": {"type": "string"},
                                      "status": {"type": "string",
                                                 "enum": ["ok", "error"]},
                                      "results": single_query,
                                      "error": {"type": "string"},
                                      "error_code": {"type": "string"},
                                  }}},
            "merged_results": {
                "type": "object",
                "description": "Cross-angle merged + deduplicated result set "
                               "(same shape as single-query 'results').",
                "properties": {
                    "query": {"type": "string"},
                    "results": {"type": "array", "items": result_item},
                },
            },
        },
        "required": ["schema_version", "research_topic", "queries",
                     "merged_results"],
    }
    return {
        "$schema": "https://json-schema.org/draft/2020-12/schema",
        "title": "SearXNG CLI Search Result",
        "schema_version": SCHEMA_VERSION,
        "description": "Output schema for 'python search.py --format json'. "
                       "Top-level 'properties' describe single-query output; "
                       "the 'batch' and 'research' subtrees describe "
                       "--queries-file and --research output respectively. "
                       "v2.3.0: fetched.items fields now match fetch_page "
                       "output (title/latency/waf_type/error_code etc.).",
        "type": "object",
        "properties": single_query["properties"],
        "required": single_query["required"],
        "defs": {
            "single_query": single_query,
            "batch": batch_schema,
            "research": research_schema,
        },
    }


def _emit_error(message: str, args, query: str = None, exit_code: int = 1,
                error_code: str = None):
    """Emit an error and exit.

    In ``--format json`` mode the error is printed to **stdout** as a
    structured JSON object so agents piping stdout can parse it. All other
    formats print to stderr (keeping stdout clean for data) and exit.

    The JSON shape is::

        {"error": "...", "exit_code": N, "error_code": "E_*",
         "recovery_hint": "...", "query": "..."}

    ``error_code`` 是结构化错误码（E_CONFIG/E_AUTH/E_NETWORK 等），让 AI
    Agent 程序化判断错误类型并采取恢复策略。``recovery_hint`` 给出可操作的
    恢复建议，让 AI 能自决策下一步动作。``query`` 仅在提供时包含。
    """
    if getattr(args, "format", None) == "json":
        payload = {"error": message, "exit_code": exit_code}
        if error_code:
            payload["error_code"] = error_code
            hint = RECOVERY_HINTS.get(error_code)
            if hint:
                payload["recovery_hint"] = hint
        if query:
            payload["query"] = query
        print(json.dumps(payload, indent=2, ensure_ascii=False))
    else:
        prefix = f"[query: {query}] " if query else ""
        code_prefix = f"[{error_code}] " if error_code else ""
        hint_suffix = ""
        if error_code and error_code in RECOVERY_HINTS:
            hint_suffix = f"\n  Hint: {RECOVERY_HINTS[error_code]}"
        logger.error(f"{prefix}{code_prefix}Error: {message}{hint_suffix}")
    sys.exit(exit_code)


def _read_queries_file(path: str) -> list:
    """Read queries from a file: one per line, skip blanks and ``#`` comments.

    v2.3.0: 编码自动检测。先按 UTF-8 读取；若解码失败（Windows 下 GBK 等
    非 UTF-8 文件常见），回退 GBK，再失败回退 utf-8 errors=replace——
    绝不因编码问题让批量任务整体失败。

    Raises :class:`RuntimeError` if the file cannot be read, so the caller
    can route it through :func:`_emit_error`.
    """
    raw = None
    try:
        raw = Path(path).read_bytes()
    except OSError as e:
        raise RuntimeError(f"cannot read queries file '{path}': {e}")

    text = None
    for enc in ("utf-8", "gbk"):
        try:
            text = raw.decode(enc)
            break
        except (UnicodeDecodeError, LookupError):
            continue
    if text is None:
        # 最后兜底：UTF-8 + 替换符，保证任务可继续
        text = raw.decode("utf-8", errors="replace")

    queries = []
    for line in text.splitlines():
        line = line.strip()
        if not line or line.startswith("#"):
            continue
        queries.append(line)
    return queries


def _save_config(args, path: str) -> None:
    """v2.2.0：将当前 CLI 参数保存为 searxng.toml 配置文件。

    只保存有实际值的参数（非 None / 非默认），让生成的配置文件干净可用。
    """
    lines = ["# searxng-cli config (auto-generated by --save-config)", ""]
    _cfg_map = {
        "instances": "instances",
        "categories": "categories",
        "language": "language",
        "pageno": "pageno",
        "pages": "pages",
        "time_range": "time_range",
        "safesearch": "safesearch",
        "method": "method",
        "format": "format",
        "sort_by": "sort_by",
        "timeout": "timeout",
        "max_retries": "retry",
        "max_results": "max_results",
        "cache_ttl": "cache_ttl",
        "cache_max_size": "cache_max_size",
        "fetch": "fetch",
        "fetch_timeout": "fetch_timeout",
        "fetch_retries": "fetch_retries",
        "fetch_total_chars": "fetch_total_chars",
        "throttle_failure_threshold": "throttle_failure_threshold",
        "throttle_pause_seconds": "throttle_pause_seconds",
        "throttle_max_delay": "throttle_max_delay",
        "similarity_threshold": "similarity_threshold",
        "log_format": "log_format",
    }
    for toml_key, attr in _cfg_map.items():
        val = getattr(args, attr, None)
        if val is not None and val != "" and val != 0:
            if isinstance(val, bool):
                lines.append(f'{toml_key} = {str(val).lower()}')
            elif isinstance(val, (int, float)):
                lines.append(f'{toml_key} = {val}')
            else:
                lines.append(f'{toml_key} = "{val}"')
    lines.append("")
    try:
        Path(path).write_text("\n".join(lines), encoding="utf-8")
        print(f"Config saved to {path}")
        logger.info(f"Config saved to {path}")
    except OSError as e:
        _emit_error(f"cannot write config file '{path}': {e}", args,
                    error_code=E_INPUT)


def _dry_run_preview(args, instance_urls: list, auth_headers: dict) -> None:
    """v2.2.0：--dry-run 预览模式，不发 HTTP 请求，打印将执行的操作。

    输出 JSON 到 stdout，包含 action/url/params/headers_count 等字段。
    """
    preview = {
        "dry_run": True,
        "instances": instance_urls,
        "headers_count": len(auth_headers) if auth_headers else 0,
    }
    if args.query:
        preview["action"] = "search"
        preview["query"] = args.query
        preview["params"] = _build_params(args.query, args)
    elif args.research:
        preview["action"] = "research"
        preview["topic"] = args.research
        custom_angles = None
        if getattr(args, "research_angles", None):
            custom_angles = [a.strip() for a in args.research_angles.split(",") if a.strip()]
        preview["angles"] = [a for a, _ in expand_research_queries(
            args.research, custom_angles=custom_angles)]
    elif args.queries_file:
        preview["action"] = "batch"
        preview["queries_file"] = args.queries_file
        # v2.5.0: dry-run 批量模式打印实际查询列表（读本地文件，不发 HTTP）。
        # 读取失败时降级为仅文件名，不阻塞预览。
        try:
            qs = _read_queries_file(args.queries_file)
            preview["queries"] = qs
        except RuntimeError as e:
            preview["queries_error"] = str(e)
    elif args.verify:
        preview["action"] = "verify"
    else:
        preview["action"] = "noop"
    if getattr(args, "pages", 1) > 1:
        preview["pages"] = args.pages
    if args.fetch:
        preview["fetch"] = args.fetch
    print(json.dumps(preview, indent=2, ensure_ascii=False))
    logger.info("Dry run complete — no HTTP requests sent")
    sys.exit(0)


# ----- Research mode (v2.1.0) -----
# 给定一个主题，自动扩展多角度查询词，复用批量搜索逻辑。
# 扩展策略是确定性规则（不做 AI 判断），覆盖人物/主题/事件的通用研究维度。

# 研究角度定义：(角度标识, 中文后缀, 英文后缀)
# 顺序代表搜索优先级——基本信息优先，评价争议最后。
# v2.1.0：支持中英文双语后缀，根据主题语言自动选择。
# 中文主题用中文后缀（"简介"/"经历"等），英文主题用英文后缀
# （"profile"/"background"等），避免 "Python asyncio 经历" 这类
# 跨语言组合在英文引擎上匹配度低的问题。angle 标识符保持英文，
# 便于 AI Agent 程序化处理。
_RESEARCH_ANGLES = [
    ("overview",   "",       ""),            # 主题本身：最直接的搜索
    ("profile",    "简介",   "profile"),     # 基本信息：百科式介绍
    ("background", "经历",   "background"),  # 背景经历：生平/历史
    ("works",      "作品",   "works"),       # 作品成就：产出物
    ("review",     "评价",   "reviews"),     # 评价争议：外界看法
]


def _is_chinese_topic(topic: str) -> bool:
    """检测主题是否包含中文字符（CJK 统一表意文字范围）。

    用于 expand_research_queries 选择中文还是英文后缀。
    纯英文主题（如 "Python asyncio"）返回 False，用英文后缀。
    """
    return bool(re.search(r'[\u4e00-\u9fff]', topic))


def expand_research_queries(topic: str, custom_angles: list = None) -> list:
    """将研究主题扩展为多角度查询词列表。

    v2.1.0 研究模式核心函数。给定一个主题（如"七森莉莉"或"Python asyncio"），
    自动生成 5 个角度的查询词，覆盖：
      1. overview   — 主题本身
      2. profile    — 基本信息（简介）
      3. background — 背景经历
      4. works      — 作品成就
      5. review     — 评价争议

    v2.1.0 修复：根据主题语言自动切换后缀。含中文字符的主题用中文后缀
    （"七森莉莉 简介"），纯英文主题用英文后缀（"Python asyncio profile"），
    避免跨语言组合在搜索引擎上匹配度低。

    v2.2.0 新增：custom_angles 参数。传入自定义角度列表时（如
    ["overview","profile","timeline","controversy"]），每个角度直接作为
    查询后缀（中英文通用），覆盖默认 5 角度。

    返回 [(angle, query), ...] 列表，angle 用于结果标注。

    确定性规则，不依赖 AI 判断——确保跨进程可复现，AI Agent 可预期。
    """
    topic = topic.strip()
    if not topic:
        return []
    if custom_angles:
        # v2.2.0：自定义角度模式，每个角度作为后缀直接拼接到主题
        return [(angle, f"{topic} {angle}".strip()) for angle in custom_angles]
    use_chinese = _is_chinese_topic(topic)
    queries = []
    for angle, cn_suffix, en_suffix in _RESEARCH_ANGLES:
        suffix = cn_suffix if use_chinese else en_suffix
        query = f"{topic} {suffix}".strip()
        queries.append((angle, query))
    return queries


# ----- Mode handlers (v2.2.0: 从 main() 提取，降低单函数复杂度) -----
# main() 只负责参数解析和分发，四条执行路径各自独立函数，便于维护和测试。
# 纯提取重构，行为与 v2.1.1 完全一致，544 测试兜底验证。


def _handle_verify(args, instance_urls: list, auth_headers: dict) -> None:
    """健康检查模式：探测实例状态并退出（不执行搜索）。"""
    logger.info(f"Verifying {len(instance_urls)} instance(s)...")
    report = verify_instances(instance_urls, timeout=args.timeout,
                              auth_headers=auth_headers)
    _print_verify_report(report, as_json=(args.format == "json"))
    sys.exit(0)


def _handle_research(args, instance_urls: list, auth_headers: dict,
                     ttl_seconds: int) -> None:
    """研究模式：给定主题自动扩展多角度查询，串行搜索，输出带研究元数据的结果。

    v2.1.0 引入，v2.1.1 增加跨角度合并去重 + 中英文双语后缀，
    v2.2.0 增加 --research-angles 自定义角度 + --similarity-dedup 相似度去重，
    v2.2.0 从 main() 提取为独立函数（纯重构，行为不变）。
    """
    topic = args.research.strip()
    # v2.2.0：支持 --research-angles 自定义角度
    custom_angles = None
    if getattr(args, "research_angles", None):
        custom_angles = [a.strip() for a in args.research_angles.split(",") if a.strip()]
    research_queries = expand_research_queries(topic, custom_angles=custom_angles)
    if not research_queries:
        _emit_error(f"research topic is empty after stripping", args,
                    error_code=E_INPUT)

    logger.info(f"Research mode: {len(research_queries)} angles for '{topic}'")
    batch = []
    any_with_results = False
    error_count = 0
    for i, (angle, q) in enumerate(research_queries, 1):
        logger.info(f"\n[{i}/{len(research_queries)}] [{angle}] {q}")
        # v2.5.0: 角度级进度事件——_run_single_query 内部的事件无法区分
        # "当前第几个角度"，加 angle_* 事件让 AI Agent 能跟踪多角度进度
        emit_progress("angle_start", angle=angle, query=q,
                      index=i, total=len(research_queries))
        results, err, err_code = _run_single_query(q, args, instance_urls,
                                                    auth_headers, ttl_seconds)
        if err:
            error_count += 1
            logger.error(f"  [ERROR] {err}")
            emit_progress("angle_fail", angle=angle, query=q,
                          error=err, error_code=err_code)
            entry = {"query": q, "angle": angle, "status": "error",
                     "error": err}
            if err_code:
                entry["error_code"] = err_code
            batch.append(entry)
        else:
            angle_count = len(results.get("results", []))
            if angle_count > 0:
                any_with_results = True
            emit_progress("angle_ok", angle=angle, query=q, results=angle_count)
            batch.append({"query": q, "angle": angle, "status": "ok",
                          "results": results})

    # 跨角度合并去重（v2.1.0 修复）
    merged = {"query": topic, "results": []}
    for br in batch:
        if br.get("status") == "ok" and "results" in br:
            merged["results"].extend(br["results"].get("results", []))
    if merged["results"]:
        deduplicate_results(merged)
        # 相似度去重（默认关闭，需 --similarity-dedup 启用）
        if getattr(args, "similarity_dedup", False):
            deduplicate_similar(merged, threshold=args.similarity_threshold)
        sort_results(merged, args.sort_by)
        if args.max_results:
            merged["results"] = merged["results"][:args.max_results]
    merged_count = len(merged["results"])
    logger.info(f"Research merged: {merged_count} unique results after dedup")

    # 输出
    if args.format == "json":
        output = json.dumps({
            "schema_version": SCHEMA_VERSION,
            "research_topic": topic,
            "research_queries": [
                {"angle": a, "query": q} for a, q in research_queries
            ],
            "queries": batch,
            "merged_results": merged,
        }, indent=2, ensure_ascii=False)
    elif args.format == "csv":
        out = io.StringIO()
        writer = csv.writer(out, lineterminator="\n")
        # v2.5.0: 媒体列检测——研究模式内任意角度含图片/视频字段则追加
        media_cols = _detect_media_columns(
            br["results"].get("results", [])
            for br in batch if "results" in br)
        writer.writerow(["angle", "query", "title", "url", "engine",
                         "score", "published_date", "content"] + media_cols)
        for br in batch:
            q = br["query"]
            angle = br.get("angle", "")
            if "results" in br:
                for r in br["results"].get("results", []):
                    writer.writerow([
                        angle, q,
                        r.get("title", ""),
                        r.get("url", ""),
                        r.get("engine", ""),
                        r.get("score", "") if r.get("score") is not None else "",
                        r.get("published_date", ""),
                        r.get("content", ""),
                    ] + [r.get(c, "") for c in media_cols])
            else:
                writer.writerow([angle, q, "", "", "", "", "",
                                 f"[ERROR: {br['error']}]"])
        output = out.getvalue().rstrip()
    elif args.format == "urls":
        parts = []
        # 先输出 per-angle 结果
        for br in batch:
            parts.append(f"# [{br.get('angle', '?')}] {br['query']}")
            if "results" in br:
                parts.append(format_urls(br["results"]))
            else:
                parts.append(f"# [ERROR: {br['error']}]")
        # 再输出合并去重后的总览
        if merged_count > 0:
            parts.append("")
            parts.append(f"# [MERGED] {topic} ({merged_count} unique results)")
            parts.append(format_urls(merged))
        output = "\n".join(parts)
    else:  # brief
        parts = []
        for br in batch:
            parts.append("=" * 60)
            parts.append(f"[{br.get('angle', '?')}] {br['query']}")
            parts.append("=" * 60)
            if "results" in br:
                parts.append(format_brief(br["results"]))
            else:
                parts.append(f"[ERROR: {br['error']}]")
            parts.append("")
        # 合并去重后的总览
        if merged_count > 0:
            parts.append("=" * 60)
            parts.append(f"[MERGED] {topic} ({merged_count} unique results)")
            parts.append("=" * 60)
            parts.append(format_brief(merged))
        output = "\n".join(parts)

    if args.output:
        with open(args.output, "w", encoding="utf-8") as f:
            f.write(output)
        logger.info(f"Saved results to {args.output}")
    else:
        print(output)
    # 三态退出码：0=有结果, 2=全部空, 1=全部错误
    if error_count == len(research_queries):
        sys.exit(1)
    elif not any_with_results:
        sys.exit(2)
    sys.exit(0)


def _run_single_query_wrapper(query: str, args, instance_urls: list,
                              auth_headers: dict, ttl_seconds: int):
    """并发批量用的 _run_single_query 包装（v2.3.0）。

    移除不可跨线程共享的参数：
      * ``args.fetch`` 置 0 —— --fetch 的 ThreadPoolExecutor 在查询线程内
        创建，worker 再经 ThreadPoolExecutor 二次并发会超过线程安全上限；
        并发批量模式用 ``--queries-file`` 不适合内嵌抓取。
      * ``args.output`` 置 None —— 输出写入统一交给 _handle_batch。
    其余参数原样透传（行为与串行路径一致）。
    """
    import copy as _copy
    qargs = _copy.copy(args)
    qargs.fetch = 0
    qargs.output = None
    return _run_single_query(query, qargs, instance_urls, auth_headers,
                             ttl_seconds)


def _handle_batch(args, instance_urls: list, auth_headers: dict,
                  ttl_seconds: int) -> None:
    """批量模式：从文件读取多个查询，串行或并发执行，输出合并结果。

    v2.2.0 从 main() 提取为独立函数（纯重构，行为不变）。
    v2.3.0 新增 --parallel-queries N：并发执行（受 AdaptiveThrottle
    约束），输出保持文件顺序；并发模式下 --fetch 被禁用（见
    _run_single_query_wrapper），日志压缩为每查询一行。
    """
    try:
        queries = _read_queries_file(args.queries_file)
    except RuntimeError as e:
        _emit_error(str(e), args, error_code=E_INPUT)
    if not queries:
        _emit_error(f"no queries found in '{args.queries_file}'", args,
                    error_code=E_INPUT)

    parallel = getattr(args, "parallel_queries", 0) or 0
    if parallel < 1:
        parallel = 1
    if parallel > 8:
        parallel = 8

    if parallel > 1:
        logger.info(f"Running {len(queries)} queries from {args.queries_file} "
                    f"in parallel ({parallel} workers)...")
        batch = [None] * len(queries)
        any_with_results = [False]
        error_count = [0]

        def _run(idx_q):
            idx, q = idx_q
            logger.info(f"[{idx+1}/{len(queries)}] {q}")
            results, err, err_code = _run_single_query_wrapper(
                q, args, instance_urls, auth_headers, ttl_seconds)
            return idx, q, results, err, err_code

        throttle = AdaptiveThrottle(0.0, parallel)

        def _job(idx_q):
            # 全局暂停（429）+ 并发槽位门控（退避降并发时快速拒绝）
            throttle.wait_if_paused()
            if not throttle.acquire_slot():
                logger.info(f"  [THROTTLE] concurrency cap ({throttle.concurrency}) "
                            f"reached, skipping query {idx_q[0]+1}")
                return idx_q[0], idx_q[1], None, "Throttled: concurrency limit", E_RATE_LIMIT
            try:
                return _run(idx_q)
            finally:
                throttle.release_slot()

        with ThreadPoolExecutor(max_workers=parallel) as ex:
            futures = [ex.submit(_job, item) for item in enumerate(queries)]
            for fut in as_completed(futures):
                idx, q, results, err, err_code = fut.result()
                if err:
                    error_count[0] += 1
                    entry = {"query": q, "status": "error", "error": err}
                    if err_code:
                        entry["error_code"] = err_code
                    batch[idx] = entry
                else:
                    if len(results.get("results", [])) > 0:
                        any_with_results[0] = True
                    batch[idx] = {"query": q, "status": "ok", "results": results}

        # 日志：并发模式下错误信息以单行输出（串行模式按序打印多行）
        if error_count[0]:
            logger.warning(f"Parallel batch: {error_count[0]} errors, "
                           f"{len(queries) - error_count[0]} ok")

        error_count = error_count[0]
        any_with_results = any_with_results[0]
    else:
        logger.info(f"Running {len(queries)} queries from {args.queries_file}...")
        batch = []
        any_with_results = False
        error_count = 0
        for i, q in enumerate(queries, 1):
            logger.info(f"\n[{i}/{len(queries)}] {q}")
            results, err, err_code = _run_single_query(q, args, instance_urls,
                                                        auth_headers, ttl_seconds)
            if err:
                error_count += 1
                logger.error(f"  [ERROR] {err}")
                entry = {"query": q, "status": "error", "error": err}
                if err_code:
                    entry["error_code"] = err_code
                batch.append(entry)
            else:
                if len(results.get("results", [])) > 0:
                    any_with_results = True
                batch.append({"query": q, "status": "ok", "results": results})

    if args.format == "json":
        output = json.dumps({"schema_version": SCHEMA_VERSION, "queries": batch},
                            indent=2, ensure_ascii=False)
    elif args.format == "csv":
        out = io.StringIO()
        writer = csv.writer(out, lineterminator="\n")
        # v2.5.0: 媒体列检测——批量内任意结果含图片/视频字段则追加对应列
        media_cols = _detect_media_columns(
            br["results"].get("results", [])
            for br in batch if "results" in br)
        writer.writerow(["query", "title", "url", "engine", "score",
                         "published_date", "content"] + media_cols)
        for br in batch:
            q = br["query"]
            if "results" in br:
                for r in br["results"].get("results", []):
                    writer.writerow([
                        q,
                        r.get("title", ""),
                        r.get("url", ""),
                        r.get("engine", ""),
                        r.get("score", "") if r.get("score") is not None else "",
                        r.get("published_date", ""),
                        r.get("content", ""),
                    ] + [r.get(c, "") for c in media_cols])
            else:
                writer.writerow([q, "", "", "", "", "",
                                 f"[ERROR: {br['error']}]"])
        output = out.getvalue().rstrip()
    elif args.format == "urls":
        parts = []
        for br in batch:
            parts.append(f"# {br['query']}")
            if "results" in br:
                parts.append(format_urls(br["results"]))
            else:
                parts.append(f"# [ERROR: {br['error']}]")
        output = "\n".join(parts)
    else:  # brief
        parts = []
        for br in batch:
            parts.append("=" * 60)
            parts.append(f"QUERY: {br['query']}")
            parts.append("=" * 60)
            if "results" in br:
                parts.append(_format_results(br["results"], args))
            else:
                parts.append(f"[ERROR: {br['error']}]")
            parts.append("")
        output = "\n".join(parts)

    if args.output:
        with open(args.output, "w", encoding="utf-8") as f:
            f.write(output)
        logger.info(f"Saved results to {args.output}")
    else:
        print(output)
    # Exit code: 1=全错, 2=全空, 0=有结果
    if error_count == len(queries):
        sys.exit(1)
    if not any_with_results:
        sys.exit(2)
    sys.exit(0)


def _handle_single(args, instance_urls: list, auth_headers: dict,
                   ttl_seconds: int) -> None:
    """单查询模式：执行一次搜索并输出结果。

    支持 --stream 流式输出（JSON Lines）。
    v2.2.0 从 main() 提取为独立函数（纯重构，行为不变）。
    """
    results, err, err_code = _run_single_query(args.query, args, instance_urls,
                                                auth_headers, ttl_seconds)

    # --stream: JSON Lines 流式输出
    if getattr(args, "stream", False) and args.format == "json":
        if err:
            error_event = {"type": "error", "error": err, "query": args.query}
            if err_code:
                error_event["error_code"] = err_code
                hint = RECOVERY_HINTS.get(err_code)
                if hint:
                    error_event["recovery_hint"] = hint
            print(json.dumps(error_event, ensure_ascii=False), flush=True)
            sys.exit(1)
        for r in results.get("results", []):
            print(json.dumps({"type": "result", "result": r},
                             ensure_ascii=False), flush=True)
        print(json.dumps({"type": "done",
                          "schema_version": SCHEMA_VERSION,
                          "count": len(results.get("results", [])),
                          "query": args.query}, ensure_ascii=False), flush=True)
        if not results.get("results"):
            sys.exit(2)
        sys.exit(0)

    if err:
        _emit_error(err, args, query=args.query, error_code=err_code)

    output = _format_results(results, args)
    if args.output:
        with open(args.output, "w", encoding="utf-8") as f:
            f.write(output)
        logger.info(f"Saved results to {args.output}")
    else:
        print(output)

    if not results.get("results"):
        logger.warning("Warning: no results returned")
        sys.exit(2)


# ----- Main -----

def main():
    # Phase 1: pre-scan for --config / --verbose / --quiet so logging is
    # configured BEFORE config loading (which may emit warnings).
    pre = argparse.ArgumentParser(add_help=False)
    pre.add_argument("--config", default=None)
    pre.add_argument("--verbose", "-v", action="store_true", default=False)
    pre.add_argument("--quiet", action="store_true", default=False)
    pre.add_argument("--log-format", default="text", choices=["text", "json"])
    pre_args, _ = pre.parse_known_args()

    setup_logging(verbose=pre_args.verbose, quiet=pre_args.quiet,
                  log_format=pre_args.log_format)
    force_utf8_stdout()  # Windows: prevent GBK crash on non-ASCII chars

    # Load config defaults from --config file, else ./searxng.toml or
    # ~/.config/searxng-cli/searxng.toml. Every CLI flag below can be
    # pre-set here; explicit flags still win.
    config = load_config(pre_args.config)

    parser = argparse.ArgumentParser(
        description="Search via a user-supplied SearXNG instance",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=f"""searxng-cli v{VERSION}

Examples:
  %(prog)s -q "python asyncio" -i https://my-searxng.example.com
  %(prog)s -q "machine learning" -i https://a.example.com,https://b.example.com --format brief
  %(prog)s -q "climate change" -i https://s.example.com --time-range month --language en
  %(prog)s -q "open source" -i https://s.example.com --format urls > result_urls.txt
  %(prog)s -q "ai news" -i https://s.example.com --engines google,duckduckgo --method POST
  %(prog)s --verify -i https://a.example.com,https://b.example.com   # health-check, no search
  %(prog)s -q "latest news" -i https://s.example.com --time-range none   # disable time filter
  %(prog)s -q "test" --config ./my-config.toml -i https://s.example.com  # explicit config file

Instance URLs are resolved from (in priority order):
  1. -i/--instance (comma-separated for parallel failover)
  2. SEARXNG_INSTANCE environment variable
  3. ./searxng.toml or ~/.config/searxng-cli/searxng.toml  (instance = "..." or instances = [...])
     ./instances.txt or ~/.config/searxng-cli/instances.txt  (one URL per line)
At least one instance is REQUIRED (public-instance discovery has been removed).
Use --verify to health-check your instances (reachability / JSON support / latency).

searxng.toml may ALSO set defaults for most flags below (engines, categories, language,
safesearch, time_range, method, format, sort_by, timeout, max_retries, proxy, cache_ttl,
fetch, fetch_timeout, fetch_retries, max_size). Explicit CLI flags always override config values.
Use --config FILE to load a non-default config file (overrides the auto-discovered one).
        """,
    )
    parser.add_argument("--config", default=None, metavar="FILE",
                        help="Path to a searxng.toml config file. Overrides the default "
                             "auto-discovery (./searxng.toml -> ~/.config/searxng-cli/searxng.toml). "
                             "Must be the first flag if you want its values to set defaults "
                             "for other flags.")
    parser.add_argument("--verbose", "-v", action="store_true", default=False,
                        help="Verbose output: show debug-level diagnostics (HTTP request URLs, "
                             "response codes, cache keys, etc.) on stderr")
    parser.add_argument("--quiet", action="store_true", default=False,
                        help="Quiet output: suppress progress messages and retry notices on stderr; "
                             "only warnings and errors are shown. (No short flag: -q is --query)")
    parser.add_argument("--query", "-q", required=False, default=None,
                        help="Search query (required unless --verify / --queries-file is used)")
    parser.add_argument("--instance", "-i", required=False, default=None,
                        help="SearXNG instance URL(s), comma-separated for failover. "
                             "Optional if SEARXNG_INSTANCE env var or a config file is set.")
    parser.add_argument("--categories", "-c", default=config.get("categories"),
                        help="Comma-separated categories (e.g. general,images,news)")
    parser.add_argument("--language", "-l", default=config.get("language"),
                        help="Language code (e.g. en, zh-CN, de)")
    parser.add_argument("--encoding", default=config.get("encoding"),
                        help="Force charset for HTML-fallback decoding "
                             "(e.g. gbk, shift_jis). v2.2.2: auto-detected "
                             "from the HTTP header / HTML meta when omitted; "
                             "this flag overrides auto-detection for "
                             "misconfigured instances.")
    parser.add_argument("--pageno", "-p", type=int, default=1,
                        help="Page number (default: 1)")
    parser.add_argument("--pages", type=int, default=1,
                        help="Fetch N pages and merge with cross-page dedup "
                             "(default: 1). When > 1, overrides --pageno "
                             "starting from page 1..N.")
    parser.add_argument("--time-range", "-t",
                        choices=["day", "week", "month", "year", "none"],
                        default=config.get("time_range", "year"),
                        help="Time range filter (default: year; 'none' disables filtering). "
                             "SearXNG API standard four tiers: day/week/month/year")
    parser.add_argument("--safesearch", "-s", type=int, choices=[0, 1, 2],
                        default=_cfg_int(config, "safesearch", 0),
                        help="Safe search: 0=off, 1=moderate, 2=strict (default: 0=off)")
    parser.add_argument("--engines",
                        default=config.get("engines", "google,bing,brave,duckduckgo,startpage,wikipedia,wikidata"),
                        help="Comma-separated engine names "
                             "(default: google,bing,brave,duckduckgo,startpage,wikipedia,wikidata)")
    parser.add_argument("--method", choices=["GET", "POST"],
                        default=config.get("method", "GET"),
                        help="HTTP method (default: GET)")
    parser.add_argument("--max-results", type=int, default=None,
                        help="Limit number of results (applied AFTER dedup+sort, "
                             "so the highest-scoring/newest items are kept)")
    parser.add_argument("--sort-by", choices=["score", "date", "engine", "none"],
                        default=config.get("sort_by", "score"),
                        help="Sort results (default: score descending; 'none' preserves "
                             "instance order). Applied after dedup, before --max-results. "
                             "HTML-fallback results have no score and keep their order.")
    parser.add_argument("--no-dedup", action="store_true",
                        help="Disable cross-engine deduplication (by default, duplicate "
                             "URLs — same page ignoring tracking params/fragment — are "
                             "collapsed, keeping the first occurrence's engine/score)")
    parser.add_argument("--similarity-dedup", action="store_true",
                        help="启用相似度去重（基于标题 SimHash），默认关闭。"
                             "合并跨引擎/跨 URL 的近似重复标题，保留首个（通常 score 最高）。"
                             "O(n²) 复杂度，结果数 > 500 时自动跳过。")
    parser.add_argument("--similarity-threshold", type=float, default=0.85,
                        metavar="FLOAT",
                        help="相似度去重阈值（默认: 0.85）。越高越严格，1.0 要求标题几乎完全相同。"
                             "仅当 --similarity-dedup 启用时生效。")
    parser.add_argument("--dedup-fetched-content", action="store_true",
                        help="v2.5.0: deduplicate fetched page content by "
                             "similarity (SimHash). Mirror/republished pages "
                             "with near-identical bodies are marked "
                             "status='duplicate' (text cleared) — keeps AI "
                             "from reading the same content multiple times. "
                             "Uses --similarity-threshold (default 0.85). "
                             "Only applies with --fetch.")
    parser.add_argument("--format", "-f", choices=["json", "brief", "urls", "csv"],
                        default=config.get("format", "json"),
                        help="Output format (default: json). 'csv' exports "
                             "title,url,engine,score,published_date,content.")
    parser.add_argument("--snippet-len", type=int, default=0,
                        help="Snippet length in brief format (default: 0 = full, no truncation)")
    parser.add_argument("--fetch", type=int, default=_cfg_int(config, "fetch", 0), metavar="N",
                        help="After search, auto-fetch full text of top N result pages")
    parser.add_argument("--fetch-timeout", type=int, default=_cfg_int(config, "fetch_timeout", 10),
                        help="Timeout per page fetch in seconds (default: 10)")
    parser.add_argument("--fetch-retries", type=int, default=_cfg_int(config, "fetch_retries", 3),
                        help="Max retries per page fetch (default: 3)")
    parser.add_argument("--fetch-total-chars", type=int,
                        default=_cfg_int(config, "fetch_total_chars", 0),
                        metavar="N",
                        help="v2.5.0: global character budget for --fetch. "
                             "Chars are allocated top-down in result order — "
                             "unused budget rolls over to the next page. Once "
                             "exhausted, remaining URLs are marked status="
                             "'skipped' (no request sent). Lets token-limited "
                             "agents cap total fetched content. 0 = unlimited "
                             "(default).")
    parser.add_argument("--fetch-report", dest="fetch_report",
                        nargs="?", const="text", default=False, metavar="FORMAT",
                        help="When used with --fetch, emit a structured fetch report to stderr "
                             "after completion. FORMAT accepts 'text' (default: human-readable "
                             "table + JSON summary line, v2.0.0 behavior) or 'json' (full JSON "
                             "object with per-URL items and summary, easier for agents to parse). "
                             "Without a value, defaults to 'text'. v2.0.0; json output v2.x.")
    parser.add_argument("--no-fallback", action="store_true",
                        help="Disable Wayback Machine fallback for failed fetches (404/403/timeout). "
                             "By default Wayback fallback is ENABLED to maximize success rate. v2.0.0.")
    parser.add_argument("--referer", default=None, metavar="URL",
                        help="Set Referer header for fetch requests (e.g. the SearXNG instance URL). "
                             "Defaults to the instance URL when fetching result pages. v2.0.0.")
    parser.add_argument("--request-delay", type=float,
                        default=_cfg_float(config, "request_delay", 0.3),
                        metavar="SECONDS",
                        help="Delay between fetch requests to avoid rate limiting (default: 0.3s). "
                             "v2.0.0: adaptive throttling may increase this on consecutive failures.")
    parser.add_argument("--max-size", type=int, default=_cfg_int(config, "max_size", None),
                        metavar="BYTES",
                        help="Max page size in bytes (default: unlimited). Set to 5242880 for 5MB cap.")
    parser.add_argument("--auth-bearer", default=None, metavar="TOKEN",
                        help="Authorization: Bearer <TOKEN> for private instances")
    parser.add_argument("--auth-bearer-file", default=None, metavar="FILE",
                        help="Read Bearer token from a file (first non-empty, non-# line). "
                             "Avoids leaving tokens in shell history.")
    parser.add_argument("--auth-basic", default=None, metavar="USER:PASS",
                        help="Authorization: Basic base64(user:pass) for private instances")
    parser.add_argument("--auth-basic-file", default=None, metavar="FILE",
                        help="Read basic auth 'user:pass' from a file (first non-empty, non-# line). "
                             "Avoids leaving passwords in shell history. "
                             "Env var SEARXNG_BASIC_AUTH is also honored.")
    parser.add_argument("--output", "-o", default=None,
                        help="Save to file instead of stdout")
    parser.add_argument("--stream", action="store_true",
                        help="Stream results as JSON Lines (one JSON object per line) to stdout. "
                             "Each line is a {\"type\": \"result\", \"result\": {...}} event. "
                             "Ends with {\"type\": \"done\", \"count\": N}. "
                             "AI Agent can process results incrementally without waiting for full output. "
                             "Only valid with --format json.")
    parser.add_argument("--progress", action="store_true",
                        help="Emit structured progress events as JSON Lines to stderr. "
                             "Events: start, instance_try, instance_ok, instance_fail, "
                             "cache_hit, cache_store, fetch_start, fetch_ok, fetch_fail, done. "
                             "AI Agent can track execution progress programmatically.")
    parser.add_argument("--log-format", choices=["text", "json"], default="text",
                        help="Log output format (default: text). 'json' emits one JSON "
                             "object per line for programmatic parsing by AI Agents.")
    parser.add_argument("--dry-run", action="store_true",
                        help="Preview actions without sending HTTP requests. Prints "
                             "{action, url, params, headers_count} as JSON to stdout.")
    parser.add_argument("--throttle-failure-threshold", type=int, default=3, metavar="N",
                        help="AdaptiveThrottle: consecutive failures before global pause "
                             "(default: 3). 0 disables throttling.")
    parser.add_argument("--throttle-pause-seconds", type=int, default=30, metavar="SEC",
                        help="AdaptiveThrottle: pause duration when threshold reached "
                             "(default: 30).")
    parser.add_argument("--throttle-max-delay", type=int, default=10, metavar="SEC",
                        help="AdaptiveThrottle: max per-request delay cap "
                             "(default: 10).")
    parser.add_argument("--research-angles", default=None, metavar="LIST",
                        help="Custom research angles (comma-separated). Overrides default "
                             "5 angles. Example: 'overview,profile,timeline,controversy'")
    parser.add_argument("--save-config", default=None, metavar="FILE",
                        help="Save current CLI arguments to a searxng.toml config file "
                             "and exit. Useful for persisting commonly used settings.")
    parser.add_argument("--timeout", type=int, default=_cfg_int(config, "timeout", 15),
                        help="Request timeout in seconds (default: 15)")
    parser.add_argument("--retry", type=int, default=_cfg_int(config, "max_retries", None),
                        help="Max retries per instance (default: 3)")
    parser.add_argument("--fail-fast", action="store_true",
                        help="Exit on first instance failure (no multi-instance retry)")
    parser.add_argument("--serial", action="store_true",
                        help="Disable parallel multi-instance probing; search instances one at a time")
    parser.add_argument("--proxy", default=config.get("proxy"), metavar="URL",
                        help="HTTP/HTTPS proxy URL (e.g. http://corp-proxy:8080). "
                             "Applies to both search and fetch requests. "
                             "Respects existing HTTP_PROXY/HTTPS_PROXY env vars when omitted.")
    parser.add_argument("--include-domain", default=None, metavar="DOMAINS",
                        help="Comma-separated allowlist; only results from these domains are kept "
                             "(e.g. 'example.com,wikipedia.org'). Applied after search.")
    parser.add_argument("--exclude-domain", default=None, metavar="DOMAINS",
                        help="Comma-separated blocklist; results from these domains are dropped "
                             "(e.g. 'pinterest.com,quora.com'). Applied after search.")
    parser.add_argument("--queries-file", default=None, metavar="FILE",
                        help="Read queries from a file (one per line; blank lines and lines "
                             "starting with '#' are skipped) and run them in sequence. "
                             "Results are emitted as a JSON array (or one brief block per query). "
                             "Overrides --query when set.")
    parser.add_argument("--parallel-queries", type=int, default=0, metavar="N",
                        help="v2.3.0: run batch queries concurrently with N workers "
                             "(1-8, capped at 8; default 0 = sequential). Output order "
                             "is preserved. Concurrency is gated by AdaptiveThrottle — "
                             "on repeated failures workers are throttled, not spammed. "
                             "When enabled, --fetch is disabled (nested parallel fetch "
                             "is unsafe) and per-query logs are compressed.")
    parser.add_argument("--research", default=None, metavar="TOPIC",
                        help="v2.1.0 Research mode: given a topic, auto-expand into 5 "
                             "multi-angle queries (overview/profile/background/works/review) "
                             "and run them in sequence. Results are merged and deduplicated. "
                             "Output includes research_topic and research_queries metadata. "
                             "Mutually exclusive with --query and --queries-file.")
    parser.add_argument("--verify", action="store_true",
                        help="Health-check mode: verify instances (reachability/JSON/latency) and exit without searching")
    parser.add_argument("--cache-ttl", type=int, default=_cfg_int(config, "cache_ttl", 0),
                        metavar="MINUTES",
                        help="Cache search results for N minutes (default: 0 = disabled). "
                             "Identical queries within the TTL skip the network entirely. "
                             "Cache lives at $SEARXNG_CACHE_DIR or ~/.cache/searxng-cli/cache.db")
    parser.add_argument("--cache-max-size", type=int,
                        default=_cfg_int(config, "cache_max_size", 0),
                        metavar="MB",
                        help="Cache size cap in MB with LRU eviction (default: 0 = use "
                             "$SEARXNG_CACHE_MAX_SIZE_BYTES env var or 100MB built-in default). "
                             "When the cap is exceeded, least-recently-accessed entries are "
                             "evicted. 0 here does NOT mean unlimited — it means 'defer to "
                             "env/default'. Use a very large value for effectively unlimited.")
    parser.add_argument("--clear-cache", action="store_true",
                        help="Delete all cached entries and exit (no search performed)")
    parser.add_argument("--cache-stats", action="store_true",
                        help="Print cache statistics (entry count, age, size, path) and exit")
    parser.add_argument("--dump-schema", action="store_true",
                        help="Print the JSON Schema for --format json output and exit. "
                             "Lets AI agents programmatically discover field names and types.")
    parser.add_argument("--version", action="version", version=f"searxng-cli v{VERSION}")

    args = parser.parse_args()

    # v2.2.0：生成 request_id 并重新配置 logging（带 log_format + request_id）
    # request_id 贯穿所有日志、进度事件和错误输出，便于 batch 模式追溯。
    request_id = os.urandom(4).hex()
    setup_logging(verbose=args.verbose, quiet=args.quiet,
                  log_format=getattr(args, "log_format", "text"),
                  request_id=request_id)

    # --dump-schema：输出 JSON Schema 到 stdout 并退出，AI Agent 可程序化发现字段
    if getattr(args, "dump_schema", False):
        print(json.dumps(_get_output_schema(), indent=2, ensure_ascii=False))
        sys.exit(0)

    # --save-config：保存当前参数到 searxng.toml 并退出
    if getattr(args, "save_config", None):
        _save_config(args, args.save_config)
        sys.exit(0)

    # 启用 --progress 进度事件（JSON Lines 到 stderr）
    set_progress_enabled(getattr(args, "progress", False))

    # --stream 只在单查询 + --format json 下有效。batch 模式输出 JSON 数组，
    # 非 json 格式无 JSON Lines 语义；两种组合都显式报错 E_INPUT，避免静默失效
    # 让 AI Agent 误以为流式输出已生效。
    if getattr(args, "stream", False):
        if args.queries_file:
            _emit_error("--stream cannot be used with --queries-file: batch mode "
                        "emits a JSON array, not JSON Lines. Drop --stream for "
                        "batch output, or use a single --query with --stream.",
                        args, error_code=E_INPUT)
        if args.research:
            _emit_error("--stream cannot be used with --research: research mode "
                        "emits a JSON array, not JSON Lines. Drop --stream for "
                        "research output, or use a single --query with --stream.",
                        args, error_code=E_INPUT)
        if args.format != "json":
            _emit_error(f"--stream requires --format json (current: {args.format}). "
                        "JSON Lines streaming only produces valid output with json format.",
                        args, error_code=E_INPUT)

    # --research 与 --query / --queries-file 互斥
    if args.research:
        if args.query:
            _emit_error("--research cannot be used with --query: research mode "
                        "auto-generates queries from the topic. Drop --query, "
                        "or use --research <topic> alone.",
                        args, error_code=E_INPUT)
        if args.queries_file:
            _emit_error("--research cannot be used with --queries-file: research "
                        "mode auto-generates queries. Drop --queries-file, "
                        "or use --research <topic> alone.",
                        args, error_code=E_INPUT)

    # --query is required unless we're doing a non-search operation.
    # --queries-file is an alternative to --query for batch mode.
    # --research is another alternative (v2.1.0).
    if (not args.verify and not args.query and not args.queries_file
            and not args.research
            and not args.clear_cache and not args.cache_stats):
        parser.error("--query is required (or use --verify / --queries-file / "
                     "--research / --clear-cache / --cache-stats)")

    # Apply proxy early so every HTTP path (search, verify, fetch) honors it.
    # Setting env vars is enough: urllib reads them via getproxies() and
    # requests honors them via trust_env=True (its default).
    if args.proxy:
        apply_proxy(args.proxy)
        logger.info(f"Proxy: {args.proxy}")

    # --cache-max-size: CLI 参数注入全局缓存单例（优先级高于环境变量）。
    # 必须在 --clear-cache / --cache-stats / 任何 search 操作之前执行，
    # 保证 stats() 的 max_size_bytes 字段和 LRU 淘汰都用正确上限。
    # args.cache_max_size <= 0 表示"defer to env/default"，不调用注入。
    if args.cache_max_size > 0:
        cache_module.set_max_size_bytes(args.cache_max_size * 1024 * 1024)
        logger.debug(f"Cache max size set to {args.cache_max_size}MB via CLI")

    # --clear-cache / --cache-stats 不需要实例，在实例解析之前处理并退出。
    # 避免无 -i 时报 E_CONFIG "no instance resolved" 让 AI 困惑。
    if args.clear_cache:
        removed = cache_module.clear()
        logger.info(f"Cleared {removed} cache entr{'y' if removed == 1 else 'ies'}.")
        sys.exit(0)

    if args.cache_stats:
        s = cache_module.stats()
        if args.format == "json":
            # JSON 模式：结构化数据走 stdout，AI Agent 可管道解析
            print(json.dumps(s, indent=2, ensure_ascii=False))
        else:
            # 非 JSON 模式：人类可读的状态信息走 stderr，保持 stdout 纯净，
            # 避免 AI Agent 用 --format json 解析 stdout 时被非 JSON 污染。
            print(f"Cache path:  {s.get('path', '?')}", file=sys.stderr)
            print(f"Entries:     {s.get('entries', 0)}", file=sys.stderr)
            size = s.get("size_bytes", 0)
            print(f"Size:        {size:,} bytes ({size / 1024:.1f} KB)", file=sys.stderr)
            if s.get("oldest_created_at"):
                print(f"Oldest:      {time.ctime(s['oldest_created_at'])}", file=sys.stderr)
            if s.get("newest_created_at"):
                print(f"Newest:      {time.ctime(s['newest_created_at'])}", file=sys.stderr)
            if s.get("error"):
                print(f"Error:       {s['error']}", file=sys.stderr)
        sys.exit(0)

    # Resolve instance(s): -i > SEARXNG_INSTANCE env > config file
    instance_urls = resolve_instances(args.instance)
    # Fallback: if --config was used, instance may be in the config dict
    # but not in the default search paths that resolve_instances checks.
    if not instance_urls and config.get("instance"):
        instance_urls = parse_instances(config["instance"])
    elif not instance_urls and config.get("instances"):
        raw = config["instances"]
        if isinstance(raw, str):
            instance_urls = parse_instances(raw)
        elif isinstance(raw, list):
            instance_urls = [u if u.startswith(("http://", "https://"))
                             else "https://" + u for u in raw if u and str(u).strip()]
    if not instance_urls:
        # Python 3.8-3.10 无 tomllib 时，.toml 配置文件无法读取。检查这种
        # 情况并在错误信息中附加提示，让 AI Agent 能给出可操作的恢复建议。
        toml_hint = ""
        try:
            import tomllib  # Python 3.11+
        except ModuleNotFoundError:
            try:
                import tomli  # type: ignore[import-not-found]
            except ModuleNotFoundError:
                toml_candidates = [
                    Path.cwd() / "searxng.toml",
                    Path.home() / ".config" / "searxng-cli" / "searxng.toml",
                ]
                if any(p.exists() for p in toml_candidates):
                    toml_hint = (" (hint: a searxng.toml file exists but cannot be "
                                 "read on Python < 3.11 without the 'tomli' package. "
                                 "Run 'pip install tomli' or use instances.txt instead.)")
        _emit_error("no SearXNG instance resolved. Provide -i/--instance, set the "
                    "SEARXNG_INSTANCE environment variable, or create a searxng.toml / "
                    "instances.txt config file." + toml_hint, args, error_code=E_CONFIG)

    # Build auth headers if provided (needed by both verify and search).
    # Credentials may come from CLI flag, file, config file, or env var
    # (in priority order) to avoid leaking secrets via shell history or `ps`.
    try:
        bearer_token = resolve_auth_bearer(
            args.auth_bearer, args.auth_bearer_file,
            config_value=config.get("auth_bearer"))
        basic_auth = resolve_auth_basic(
            args.auth_basic, args.auth_basic_file,
            config_value=config.get("auth_basic"))
    except RuntimeError as e:
        _emit_error(str(e), args, error_code=E_AUTH)
    auth_headers = build_auth_headers(
        bearer_token=bearer_token,
        basic_auth=basic_auth,
    )
    if auth_headers:
        auth_type = "Bearer" if bearer_token else "Basic"
        logger.info(f"Auth: {auth_type} ***")

    # v2.2.0：--dry-run 预览模式，不发 HTTP 请求
    if getattr(args, "dry_run", False):
        _dry_run_preview(args, instance_urls, auth_headers)

    # Health-check mode: report instance status and exit (no search performed)
    if args.verify:
        _handle_verify(args, instance_urls, auth_headers)

    if args.fail_fast:
        instance_urls = instance_urls[:1]

    logger.info(f"Instances to try: {len(instance_urls)}")

    # Cache TTL in seconds (CLI takes minutes for ergonomics)
    ttl_seconds = args.cache_ttl * 60 if args.cache_ttl > 0 else 0

    # ----- Research mode: --research (v2.1.0) -----
    # 给定主题，自动扩展为 5 个多角度查询，串行搜索，输出带研究元数据的 JSON。
    # 复用批量搜索逻辑，但查询词来自 expand_research_queries 而非文件。
    if args.research:
        _handle_research(args, instance_urls, auth_headers, ttl_seconds)

    # ----- Batch mode: --queries-file -----
    # Reads one query per line (blank/# lines skipped) and runs them in
    # sequence. Output is a JSON array (json format) or concatenated blocks
    # (brief/urls). A failed query is recorded but does not abort the batch.
    if args.queries_file:
        _handle_batch(args, instance_urls, auth_headers, ttl_seconds)

    # ----- Single query mode -----
    _handle_single(args, instance_urls, auth_headers, ttl_seconds)


if __name__ == "__main__":
    main()