MetonaTeam/searxng-use-cli
2
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(v1.8.0): 稳定性修复 + AI Agent 体验增强

Browse Source 
稳定性修复:

- 修复 cache.py SQLite 连接泄漏（contextlib.closing 包装）

- 修复 fetch.py requests stream=True 连接泄漏（try/finally resp.close()）

- RETRYABLE_STATUS 新增 403，激活 UA fallback 切换逻辑

- --cache-stats 移至实例解析前，无需实例即可查询

- classify_error 从错误消息提取 HTTP 状态码，正确分类 E_AUTH/E_RATE_LIMIT

- --stream 与 --queries-file 互斥检查，违规报 E_INPUT

- batch 退出码语义统一（0=有结果 / 1=全部错误 / 2=全部空结果）

AI Agent 体验增强:

- 错误码体系完善：E_CONFIG/E_AUTH/E_NETWORK/E_RATE_LIMIT/E_PARSE/E_EMPTY/E_INPUT/E_INTERNAL

- recovery_hint 恢复提示字段，AI Agent 可程序化决策恢复策略

- stream 模式新增 error 事件类型（含 error_code + recovery_hint）

- 进度事件扩展：instance_try/instance_ok/instance_fail

- batch 模式统一 schema（status 字段区分 success/failed）

- JSON 输出含 schema_version 字段确保版本兼容

测试与文档:

- 测试覆盖：330 -> 352

- SKILL.md / README.md 同步更新
This commit is contained in:
thzxx
2026-08-01 19:02:44 +08:00
parent dea899143d
commit fb9b2af45f
12 changed files with 871 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+36 -19
View File
@@ -55,36 +55,47 @@ AI 调用后，stdout 输出网页正文（text/html/markdown 三种格式），
| exit 1 | 失败 | 致命错误（所有实例不可用、参数错误等） |
| exit 2 | 空结果 | 搜索成功但无结果 |
**错误处理**：`--format json` 模式下，错误以 JSON 输出到 stdout（非 stderr），格式为 `{"error": "...", "error_code": "E_NETWORK", "exit_code": 1, "query": "..."}`，AI 可程序化捕获。
**错误处理**：`--format json` 模式下，错误以 JSON 输出到 stdout（非 stderr），格式为 `{"error": "...", "error_code": "E_NETWORK", "recovery_hint": "...", "exit_code": 1, "query": "..."}`，AI 可程序化捕获并按 `recovery_hint` 采取恢复行动。
**错误码体系**（`error_code` 字段）：
**错误码体系**（`error_code` + `recovery_hint` 字段）：
| 错误码 | 含义 | AI 恢复策略 |
|--------|------|-------------|
| `E_CONFIG` | 配置错误（无实例） | 提示用户设置 `-i` / `SEARXNG_INSTANCE` |
| `E_AUTH` | 认证失败（401/403） | 检查 token/凭证 |
| `E_NETWORK` | 网络错误（连接失败、5xx） | 重试或切换实例/代理 |
| `E_RATE_LIMIT` | 限流（429） | 等待后重试 |
| `E_PARSE` | 解析错误 | 检查实例 JSON 支持 |
| `E_INPUT` | 输入错误（参数/文件） | 修正参数 |
| `E_INTERNAL` | 内部错误 | 报告 bug |
| 错误码 | 含义 | recovery_hint（节选） |
|--------|------|----------------------|
| `E_CONFIG` | 配置错误（无实例） | 设置 `-i`/`SEARXNG_INSTANCE`/配置文件 |
| `E_AUTH` | 认证失败（401/403） | 检查凭证、token 过期与权限 |
| `E_NETWORK` | 网络错误（连接失败、5xx） | 重试退避、切换实例、检查代理 |
| `E_RATE_LIMIT` | 限流（429） | 等待重试、降低频率、分散负载 |
| `E_PARSE` | 解析错误 | 切换实例、切换 `--method` |
| `E_EMPTY` | 空结果（exit 2） | 调整查询词、扩大 `--time-range`/`--categories` |
| `E_INPUT` | 输入错误（参数/文件） | 检查语法、标志组合、文件路径 |
| `E_INTERNAL` | 内部错误 | 用 `--verbose` 重跑并报告 |
### 流式输出与进度事件（AI 高级用法）
**`--stream`**：JSON Lines 流式输出，每条结果一行 JSON，AI 可增量处理：
```
{"type": "result", "result": {"title": "...", "url": "..."}}
{"type": "done", "count": 10, "query": "..."}
{"type": "done", "schema_version": "1.0", "count": 10, "query": "..."}
```
搜索失败时输出 error 事件（含 `recovery_hint`）：
```
{"type": "error", "error": "...", "error_code": "E_AUTH", "recovery_hint": "Verify credentials...", "query": "..."}
```
注意：`--stream` 仅在单查询 + `--format json` 下有效；与 `--queries-file` 或非 json 格式同用会立即报 `E_INPUT`。
**`--progress`**：进度事件流（JSON Lines 到 stderr），AI 可实时跟踪执行：
```
{"event": "start", "query": "...", "instances": 2}
{"event": "instance_try", "url": "https://inst1.example.com", "attempt": 1}
{"event": "instance_ok", "url": "https://inst1.example.com", "latency": 0.342, "results": 10}
{"event": "instance_fail", "url": "https://inst2.example.com", "error": "HTTP 503", "error_code": "E_NETWORK"}
{"event": "cache_hit", "query": "...", "ttl": 30}
{"event": "fetch_ok", "url": "...", "chars": 12345}
{"event": "done", "results": 10, "query": "..."}
```
**`--dump-schema`**：输出当前版本的 JSON Schema 到 stdout 并退出，AI 可程序化发现字段名与类型，无需解析文档。
```bash
# AI 推荐用法：流式输出 + 进度事件
python scripts/search.py -q "research" -i https://your-instance --stream --progress
@@ -94,7 +105,7 @@ python scripts/search.py -q "research" -i https://your-instance --stream --progr
### 前置条件
- Python 3.8+
- Python 3.8+（3.11+ 开箱即用；3.8-3.10 使用 `searxng.toml` 配置需 `pip install tomli`）
- 一个 SearXNG 实例 URL（自建或受信任的实例）
### 安装
@@ -105,9 +116,13 @@ git clone https://git.metona.cn/MetonaTeam/searxng-use-cli.git
无需安装依赖。`search.py` 仅使用 Python 标准库，开箱即用。
可选安装（提升 `fetch.py` 抓取质量）：
可选安装：
```bash
# 提升 fetch.py 抓取质量（HTTP 连接池 + HTML 解析）
pip install requests beautifulsoup4
# Python 3.8-3.10 使用 searxng.toml 配置文件时需要（3.11+ 内置 tomllib）
pip install tomli
```
### 配置实例
@@ -248,10 +263,12 @@ python scripts/fetch.py -u https://example.com \
**工程**
- 共享 `common.py`（统一重试/字符集/认证/日志）
- 结构化日志（`--verbose` / `--quiet`）
- 结构化错误码（`E_NETWORK` / `E_AUTH` / `E_RATE_LIMIT` 等）
- JSON Lines 流式输出（`--stream`）
- 进度事件（`--progress`，JSON Lines 到 stderr）
- 330 个单元+集成测试
- 结构化错误码 + `recovery_hint` 恢复建议（`E_NETWORK` / `E_AUTH` / `E_RATE_LIMIT` 等）
- JSON 输出含 `schema_version` 字段，`--dump-schema` 输出 JSON Schema 文档
- JSON Lines 流式输出（`--stream`，含 `error` 事件类型）
- 进度事件（`--progress`，含 `instance_try`/`instance_ok`/`instance_fail`，JSON Lines 到 stderr）
- batch 模式统一 schema（`status` 字段区分成功/失败）
- 352 个单元+集成测试
## 跨 Agent 兼容性
@@ -279,7 +296,7 @@ pip install pytest
pytest -q
```
330 个测试覆盖：缓存操作、认证解析、域名过滤、Markdown 转换、搜索逻辑、集成流程、日志配置、HTML 回退、自动抓取、健康检查、输出格式化、实例解析、并行搜索、CLI 端到端、错误码分类、流式输出、进度事件、配置文件认证。
352 个测试覆盖：缓存操作、认证解析、域名过滤、Markdown 转换、搜索逻辑、集成流程、日志配置、HTML 回退、自动抓取、健康检查、输出格式化、实例解析、并行搜索、CLI 端到端、错误码分类、流式输出、进度事件、配置文件认证、schema_version、recovery_hint、batch 统一 schema、--dump-schema。
## 项目结构
SKILL.md
+56 -22
View File
@@ -1,7 +1,7 @@
---
name: searxng-use-cli
description: Use when you need to search the web via your OWN SearXNG instance (no public-instance discovery). 3 CLI scripts + a shared common.py module — execute privacy-respecting searches against a user-supplied instance (with multi-instance failover, 5xx/429 retry, auto-fetch) or via SEARXNG_INSTANCE env / config file, fetch/extract readable text or markdown from web pages. Zero-config replacement for proprietary search APIs.
version: 1.7.0
version: 1.8.0
author: Metona Team
license: MIT
platforms: [linux, macos, windows]
@@ -188,23 +188,24 @@ In `--format json` mode, errors are emitted as structured JSON on stdout:
{
"error": "All 2 instances failed. Last error: connection refused",
"error_code": "E_NETWORK",
"recovery_hint": "Retry with backoff, or try a different SearXNG instance. Check network connectivity, proxy settings, and instance uptime.",
"exit_code": 1,
"query": "search term"
}
```
AI agents can use `error_code` to programmatically decide recovery strategy:
AI agents can use `error_code` to programmatically decide recovery strategy, and `recovery_hint` for a ready-to-use actionable suggestion:
| Code | Meaning | AI Recovery Strategy |
|------|---------|---------------------|
| `E_CONFIG` | Configuration error (no instance resolved) | Prompt user to set `-i` / `SEARXNG_INSTANCE` / config file |
| `E_AUTH` | Authentication failed (401/403) | Check token/credentials, prompt user to re-authenticate |
| `E_NETWORK` | Network error (connection refused, timeout, 5xx, all instances failed) | Retry with backoff, switch instance or proxy |
| `E_RATE_LIMIT` | Rate limited (429) | Wait and retry with reduced frequency |
| `E_PARSE` | Parse error (JSON/HTML parsing failed) | Check if instance supports JSON, try HTML fallback |
| `E_EMPTY` | Empty results (exit code 2) | Adjust query terms or time range |
| `E_INPUT` | Input error (bad parameters, file not found) | Fix CLI arguments or file paths |
| `E_INTERNAL` | Internal error (unexpected exception) | Report bug with full error message |
| Code | Meaning | recovery_hint (abridged) |
|------|---------|--------------------------|
| `E_CONFIG` | Configuration error (no instance resolved) | Provide `-i`/`SEARXNG_INSTANCE`/config file |
| `E_AUTH` | Authentication failed (401/403) | Verify credentials, check token expiry & permissions |
| `E_NETWORK` | Network error (connection refused, timeout, 5xx, all instances failed) | Retry with backoff, switch instance, check proxy |
| `E_RATE_LIMIT` | Rate limited (429) | Wait and retry, reduce frequency, distribute load |
| `E_PARSE` | Parse error (JSON/HTML parsing failed) | Try different instance, switch `--method` |
| `E_EMPTY` | Empty results (exit code 2) | Refine query, broaden `--time-range`, add `--categories` |
| `E_INPUT` | Input error (bad parameters, file not found) | Check query syntax, flag combinations, file paths |
| `E_INTERNAL` | Internal error (unexpected exception) | Re-run with `--verbose`, report bug |
### JSON Lines Streaming (`--stream`)
@@ -221,14 +222,19 @@ Output format (each line is a separate JSON object):
```
{"type": "result", "result": {"title": "...", "url": "...", "content": "..."}}
{"type": "result", "result": {"title": "...", "url": "...", "content": "..."}}
{"type": "done", "count": 2, "query": "large topic"}
{"type": "done", "schema_version": "1.0", "count": 2, "query": "large topic"}
```
- `type: "result"` — one per search result, emitted as soon as available
- `type: "done"` — terminal event with total count, always emitted last
- Exit code 0 on success, 2 on empty results (done event still emitted)
- `type: "done"` — terminal event with total count and `schema_version`, always emitted last
- `type: "error"` — emitted when the search fails, includes `error_code` and `recovery_hint`:
```
{"type": "error", "error": "All 2 instances failed. Last error: HTTP Error 403", "error_code": "E_AUTH", "recovery_hint": "Verify credentials...", "query": "..."}
```
- Exit code 0 on success, 1 on error, 2 on empty results (done event still emitted)
Only valid with `--format json` (the default).
Only valid with `--format json` (single query mode). Using `--stream` with
`--queries-file` or non-json formats raises `E_INPUT` immediately.
### Progress Events (`--progress`)
@@ -243,6 +249,9 @@ Event types (each on its own line, JSON Lines format on stderr):
```jsonl
{"event": "start", "query": "research topic", "instances": 2}
{"event": "instance_try", "url": "https://instance1.example.com", "attempt": 1}
{"event": "instance_ok", "url": "https://instance1.example.com", "latency": 0.342, "results": 10}
{"event": "instance_fail", "url": "https://instance2.example.com", "error": "HTTP 503", "error_code": "E_NETWORK"}
{"event": "cache_hit", "query": "research topic", "ttl": 30}
{"event": "cache_store", "query": "research topic", "ttl": 30}
{"event": "fetch_start", "count": 3}
@@ -264,10 +273,19 @@ logs are human-readable text.
### Output JSON Schema
The default `--format json` output shape (for reference, not enforced):
The default `--format json` output includes a `schema_version` field so AI
agents can detect breaking changes. Run `--dump-schema` to get the full JSON
Schema document programmatically:
```bash
python scripts/search.py --dump-schema
```
Single query output shape:
```json
{
"schema_version": "1.0",
"query": "search term",
"number_of_results": 10,
"results": [
@@ -294,13 +312,28 @@ The default `--format json` output shape (for reference, not enforced):
"text_length": 12345,
"truncated": false,
"final_url": "https://example.com/final",
"user_agent_used": "searxng-cli/1.7.0"
"user_agent_used": "searxng-cli/1.8.0"
}
],
"fetched_source": "json"
}
```
Batch mode (`--queries-file`) wraps results in a unified schema:
```json
{
"schema_version": "1.0",
"queries": [
{"query": "term1", "status": "ok", "results": {...}},
{"query": "term2", "status": "error", "error": "...", "error_code": "E_NETWORK"}
]
}
```
Each batch entry has a `status` field (`"ok"` or `"error"`). Successful entries
contain `results`; failed entries contain `error` and `error_code`.
Fields marked as optional may be absent. The `fetched` and `fetched_source`
fields only appear when `--fetch N` is used.
@@ -431,7 +464,7 @@ usage: search.py [-h] [--query QUERY] [--instance URL]
[--exclude-domain DOMAINS] [--queries-file FILE]
[--cache-ttl MINUTES] [--clear-cache] [--cache-stats]
[--sort-by {score,date,engine,none}] [--no-dedup]
[--config FILE] [--verbose] [--quiet] [--version]
[--config FILE] [--dump-schema] [--verbose] [--quiet] [--version]
```
**What it does:**
@@ -445,14 +478,14 @@ usage: search.py [-h] [--query QUERY] [--instance URL]
8. **Stable auto-fetch:** `--fetch 3` concurrently downloads top 3 result pages with retry, browser-UA fallback, CAPTCHA detection, and **`fetch.py`'s higher-quality text extractor** (the same engine `fetch.py` uses)
9. **Health-check mode:** `--verify` probes each instance (reachability / JSON-API support / latency / POST support / engine list / auth status) and prints a report, then exits without searching — use it to validate your instance list
10. **Result caching:** `--cache-ttl 30` stores results for 30 min; identical queries within the TTL skip the network entirely. Cache lives at `$SEARXNG_CACHE_DIR` or `~/.cache/searxng-cli/cache.db` (SQLite, WAL mode). `--clear-cache` / `--cache-stats` manage it without searching
11. **Batch mode:** `--queries-file FILE` reads one query per line (blank/`#` lines skipped) and runs them in sequence; output is a JSON array (or one block per query in brief/urls). A failed query is recorded but does not abort the batch
11. **Batch mode:** `--queries-file FILE` reads one query per line (blank/`#` lines skipped) and runs them in sequence; JSON output is `{"schema_version": "1.0", "queries": [{"query":..., "status": "ok"|"error", ...}]}` (or one block per query in brief/urls). A failed query is recorded but does not abort the batch. Exit codes: 0 if any query returned results, 1 if all errored, 2 if all empty
12. **Dedup + sort + domain filter:** After search (and cache), duplicate URLs are collapsed (default; `--no-dedup` disables), results are sorted (`--sort-by`; default: score descending), and then `--include-domain`/`--exclude-domain` filter by domain. Matching is case-insensitive and ignores a leading `www.`; when a domain is in both lists, exclude wins
13. **Proxy & auth:** `--proxy URL` routes both search and fetch through a proxy; `--auth-bearer` / `--auth-basic` (plus `*-file` variants, `searxng.toml` `auth_basic`/`auth_bearer` fields, and `SEARXNG_BEARER_TOKEN` / `SEARXNG_BASIC_AUTH` env vars) supply credentials. Priority: CLI flag > file > config file > env var
14. **Config defaults:** `searxng.toml` may pre-set most flags (engines, categories, language, safesearch, time_range, method, format, sort_by, timeout, max_retries, proxy, cache_ttl, fetch, fetch_timeout, fetch_retries, max_size, auth_basic, auth_bearer); explicit CLI flags always win
15. **Structured errors:** in `--format json` mode, failures print a JSON object `{"error": "...", "exit_code": N, "query": "..."}` to stdout so agents can parse them
15. **Structured errors:** in `--format json` mode, failures print a JSON object `{"error": "...", "error_code": "E_*", "recovery_hint": "...", "exit_code": N, "query": "..."}` to stdout so agents can parse them and decide recovery strategy
**Key options:**
- `--query "your search"` — **required unless** `--verify`, `--queries-file`, `--clear-cache`, or `--cache-stats` is used
- `--query "your search"` — **required unless** `--verify`, `--queries-file`, `--clear-cache`, `--cache-stats`, or `--dump-schema` is used
- `--instance https://searx.example.org` — **required unless** `SEARXNG_INSTANCE` env var or a config file supplies it; comma-separated list enables failover
- `--queries-file FILE` — read queries from a file (one per line; blank/`#` skipped) and run them in sequence; overrides `--query`
- `--engines google,duckduckgo` — restrict to specific search engines (whitespace around commas is auto-stripped; see default list above)
@@ -487,6 +520,7 @@ usage: search.py [-h] [--query QUERY] [--instance URL]
- `--cache-ttl MINUTES` — cache results for N minutes (default: `0` = disabled); identical queries within the TTL skip the network
- `--clear-cache` — delete all cached entries and exit (no search)
- `--cache-stats` — print cache statistics (entries, age, size, path) and exit
- `--dump-schema` — print the JSON Schema for `--format json` output and exit; lets AI agents programmatically discover field names and types without parsing prose docs
- `--auth-bearer TOKEN` — `Authorization: Bearer` header for private instances
- `--auth-bearer-file FILE` — read Bearer token from a file (first non-empty, non-`#` line); also honors `SEARXNG_BEARER_TOKEN` env var
- `--auth-basic USER:PASS` — `Authorization: Basic` header (auto base64-encoded)
scripts/_config.py
+3 -2
View File
@@ -1,11 +1,12 @@
"""Package-level constants for searxng-cli scripts.
Import in sibling scripts with:
from _config import VERSION, USER_AGENT
from _config import VERSION, USER_AGENT, SCHEMA_VERSION
Retry settings and shared HTTP utilities now live in ``common.py`` so that
both ``search.py`` and ``fetch.py`` share one consistent implementation.
"""
VERSION = "1.7.0"
VERSION = "1.8.0"
SCHEMA_VERSION = "1.0"
USER_AGENT = f"searxng-cli/{VERSION}"
scripts/cache.py
+11 -3
View File
@@ -22,6 +22,7 @@ Design notes:
cache failure must never break a search.
"""
import contextlib
import hashlib
import json
import os
@@ -40,8 +41,15 @@ def _cache_path() -> Path:
return DEFAULT_CACHE_DIR / "cache.db"
def _connect(path: Path) -> sqlite3.Connection:
"""Open a connection with WAL mode and ensure the schema exists."""
def _connect(path: Path):
"""Open a connection with WAL mode and ensure the schema exists.
Returns a :class:`contextlib.closing` wrapper so ``with _connect(...) as conn:``
closes the connection on exit. ``sqlite3.Connection.__exit__`` only commits /
rolls back the transaction — it does **not** call ``close()``, which leaks
file descriptors across many cache operations (especially under
``--queries-file`` + ``--cache-ttl`` batch runs).
"""
path.parent.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(str(path), timeout=10)
# WAL allows concurrent readers alongside a single writer, which matters
@@ -59,7 +67,7 @@ def _connect(path: Path) -> sqlite3.Connection:
"""
)
conn.commit()
return conn
return contextlib.closing(conn)
def _make_key(params: dict) -> str:
scripts/common.py
+49 -4
View File
@@ -60,8 +60,12 @@ def setup_logging(verbose: bool = False, quiet: bool = False) -> None:
MAX_RETRIES = 3
RETRY_BACKOFF_BASE = 1.5 # seconds; exponential backoff + jitter
# HTTP status codes that are worth retrying (rate limit + gateway errors)
RETRYABLE_STATUS = frozenset({429, 502, 503, 504})
# HTTP status codes worth retrying. 403 is included so fetch_url's UA-fallback
# loop can kick in when a site blocks the default searxng-cli User-Agent
# (a fresh UA is tried on each retry attempt). search.py also retries 403 —
# it doesn't switch UAs, so true auth failures waste ~3 attempts, accepted
# as a trade-off for one shared retry policy across both scripts.
RETRYABLE_STATUS = frozenset({403, 429, 502, 503, 504})
# Browser-like UA strings for fallback when the searxng-cli UA is blocked
FALLBACK_UAS = [
@@ -302,6 +306,29 @@ E_EMPTY = "E_EMPTY"
E_INPUT = "E_INPUT"
E_INTERNAL = "E_INTERNAL"
# 每个 E_* 配套的可操作恢复建议，让 AI Agent 能自决策下一步动作，
# 而不是盲目重试或放弃。在 _emit_error 的 JSON 输出中作为 recovery_hint 字段。
RECOVERY_HINTS = {
E_CONFIG: "Provide -i/--instance, set SEARXNG_INSTANCE env var, or create "
"searxng.toml/instances.txt config file.",
E_AUTH: "Verify --auth-bearer/--auth-basic credentials or "
"SEARXNG_BEARER_TOKEN/SEARXNG_BASIC_AUTH env vars. Check token "
"expiry and instance access permissions.",
E_NETWORK: "Retry with backoff, or try a different SearXNG instance. "
"Check network connectivity, proxy settings, and instance uptime.",
E_RATE_LIMIT: "Wait before retrying (exponential backoff). Reduce query "
"frequency, narrow --time-range, or distribute load across "
"multiple instances.",
E_INPUT: "Check query syntax, --categories values, --time-range format, "
"and flag combinations. Use --help for valid options.",
E_PARSE: "The instance returned malformed data. Try a different instance, "
"switch --method, or check if the instance version is compatible.",
E_EMPTY: "Refine the query (more specific terms), broaden --time-range, "
"add --categories, or increase --pageno to find more results.",
E_INTERNAL: "This is likely a bug. Re-run with --verbose and report the "
"full output for diagnosis.",
}
def classify_error(exc: BaseException) -> str:
"""将异常分类为结构化错误码，供 AI Agent 程序化处理。
@@ -350,18 +377,36 @@ def classify_error(exc: BaseException) -> str:
if isinstance(exc, (ValueError, _json.JSONDecodeError)):
return E_PARSE
# RuntimeError：从消息中推断（search_multi 的 "All N instances failed" 等）
# RuntimeError：从消息中推断（search_multi 的 "All N instances failed.
# Last error: HTTP Error 403: Forbidden" 等）。并行路径现在会把 last_error
# 拼进消息，让这里能提取真实错误类型，而不是一律误判 E_NETWORK。
msg = str(exc).lower()
if isinstance(exc, RuntimeError):
# 先检查 auth/rate-limit 关键字（最常见，来自 last_error 详情）
if "auth" in msg or "403" in msg or "401" in msg:
return E_AUTH
if "rate" in msg or "429" in msg:
return E_RATE_LIMIT
# 从 "http error NNN" / "http NNN" 模式中提取状态码，
# 正确分类 "All instances failed. Last error: HTTP Error 403" 等
import re as _re
status_match = _re.search(r'http(?: error)? (\d{3})', msg)
if status_match:
status = int(status_match.group(1))
if status == 429:
return E_RATE_LIMIT
if status in (401, 403):
return E_AUTH
if 400 <= status < 500:
return E_INPUT
if 500 <= status < 600:
return E_NETWORK
# 所有实例失败的通用模式（无更具体的 HTTP 状态码时才判为网络错误）
if "all" in msg and "instance" in msg and "fail" in msg:
return E_NETWORK
if "parse" in msg or "json" in msg or "html" in msg:
return E_PARSE
if "not found" in msg or "no " in msg and "instance" in msg:
if "not found" in msg or ("no " in msg and "instance" in msg):
return E_CONFIG
return E_INTERNAL
scripts/fetch.py
+37 -30
View File
@@ -513,35 +513,42 @@ def fetch_url(url: str, timeout: int = 15, user_agent: str = None,
if _HAS_REQUESTS:
resp = _requests.get(url, timeout=timeout, headers=headers,
allow_redirects=allow_redirects, stream=True)
resp.raise_for_status()
# stream=True holds the socket open; must close explicitly,
# including on raise_for_status() / max_size break / decode
# errors — otherwise the connection leaks back to the pool
# and long-running agents exhaust ports.
try:
resp.raise_for_status()
# Read: unlimited if max_size is None, chunked with limit otherwise
if max_size is None:
raw = resp.content
truncated = False
else:
chunks = []
total = 0
for chunk in resp.iter_content(chunk_size=65536, decode_unicode=False):
if chunk:
chunks.append(chunk)
total += len(chunk)
if total > max_size:
break
raw = b"".join(chunks)
truncated = total > max_size
# Read: unlimited if max_size is None, chunked with limit otherwise
if max_size is None:
raw = resp.content
truncated = False
else:
chunks = []
total = 0
for chunk in resp.iter_content(chunk_size=65536, decode_unicode=False):
if chunk:
chunks.append(chunk)
total += len(chunk)
if total > max_size:
break
raw = b"".join(chunks)
truncated = total > max_size
if encoding:
content = raw.decode(encoding)
else:
charset = detect_charset(raw, resp.headers.get("Content-Type", ""))
try:
content = raw.decode(charset)
except (UnicodeDecodeError, LookupError):
content = raw.decode("utf-8", errors="replace")
if encoding:
content = raw.decode(encoding)
else:
charset = detect_charset(raw, resp.headers.get("Content-Type", ""))
try:
content = raw.decode(charset)
except (UnicodeDecodeError, LookupError):
content = raw.decode("utf-8", errors="replace")
return FetchResult(content, resp.headers.get("Content-Type", ""),
resp.url, truncated, ua)
return FetchResult(content, resp.headers.get("Content-Type", ""),
resp.url, truncated, ua)
finally:
resp.close()
# stdlib fallback
req = urllib.request.Request(url, headers=headers)
@@ -588,7 +595,7 @@ def fetch_url(url: str, timeout: int = 15, user_agent: str = None,
logger.info(f" Fetch retry {attempt+1}/{max_retries} ({e}) in {delay:.1f}s")
time.sleep(delay)
continue
raise RuntimeError(f"HTTP {e.code} for {url}")
raise RuntimeError(f"HTTP {e.code} for {url}") from e
except (urllib.error.URLError, OSError, TimeoutError) as e:
last_error = e
if attempt < max_retries:
@@ -596,7 +603,7 @@ def fetch_url(url: str, timeout: int = 15, user_agent: str = None,
logger.info(f" Fetch retry {attempt+1}/{max_retries} ({e}) in {delay:.1f}s")
time.sleep(delay)
continue
raise RuntimeError(f"Request failed for {url}: {e}")
raise RuntimeError(f"Request failed for {url}: {e}") from e
except Exception as e:
# requests backend: retry only on connection errors (no response)
# or transient 429/5xx; do NOT retry permanent errors like 404.
@@ -608,9 +615,9 @@ def fetch_url(url: str, timeout: int = 15, user_agent: str = None,
logger.info(f" Fetch retry {attempt+1}/{max_retries} ({e}) in {delay:.1f}s")
time.sleep(delay)
continue
raise RuntimeError(f"Request failed for {url}: {e}")
raise RuntimeError(f"Request failed for {url}: {e}") from e
raise RuntimeError(f"All {max_retries+1} attempts failed for {url}: {last_error}")
raise RuntimeError(f"All {max_retries+1} attempts failed for {url}: {last_error}") from last_error
# ----- Main -----
scripts/search.py
+242 -45
View File
@@ -24,11 +24,12 @@ from pathlib import Path
# Allow running standalone from any working directory
sys.path.insert(0, str(Path(__file__).resolve().parent))
from _config import USER_AGENT, VERSION
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,
classify_error,
@@ -143,7 +144,7 @@ class SearXNGHTMLParser(HTMLParser):
self._in_time = True
self._text_buf = []
elif tag in ("script", "style"):
self._skip_depth = 1
self._skip_depth += 1
# Suggestions: <div id="suggestions"> or class containing "suggestion"
if tag_id == "suggestions" or "suggestion" in classes:
@@ -312,6 +313,17 @@ def _read_instance_file(path: Path) -> list:
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 []
@@ -423,7 +435,7 @@ def _retry_with_backoff(fn, max_retries: int = MAX_RETRIES, base_delay: float =
try:
return fn()
except urllib.error.HTTPError as e:
if e.code in RETRYABLE_STATUS: # 429 rate-limit + 5xx gateway errors
if e.code in RETRYABLE_STATUS: # 403 (UA block) + 429 + 5xx
last_error = e
if attempt < max_retries:
delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
@@ -524,14 +536,22 @@ def search_multi(instance_urls: list, params: dict, method: str = "GET",
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)
return _retry_with_backoff(_do, max_retries=retry_per)
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}")
@@ -540,15 +560,24 @@ def search_multi(instance_urls: list, params: dict, method: str = "GET",
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)
try:
return instance, _retry_with_backoff(_do, max_retries=retry_per)
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):
@@ -558,6 +587,10 @@ def search_multi(instance_urls: list, params: dict, method: str = "GET",
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
@@ -567,7 +600,8 @@ def search_multi(instance_urls: list, params: dict, method: str = "GET",
if u in results_by_url:
return results_by_url[u]
raise RuntimeError(f"All {len(instance_urls)} instances failed (parallel).")
raise RuntimeError(f"All {len(instance_urls)} instances failed (parallel). "
f"Last error: {last_parallel_error}")
# ----- Output formatting -----
@@ -675,7 +709,10 @@ def verify_instances(instance_urls: list, timeout: int = 15,
# 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"))
return {"url": u, "reachable": e.code not in (401, 403, 404),
# 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),
@@ -900,7 +937,9 @@ def deduplicate_results(results: dict) -> dict:
if not results.get("results"):
return results
TRACKING_PREFIXES = ("utm_", "gclid", "fbclid", "mc_", "ref", "ref_")
# 跟踪参数前缀。裸 "ref" 过于宽泛（会误删 reference/refcode 等正常参数），
# 收紧为 "ref_" 只匹配 ref_source/ref_campaign 等跟踪参数。
TRACKING_PREFIXES = ("utm_", "gclid", "fbclid", "mc_", "ref_")
def _normalize_url(url: str) -> str:
try:
@@ -1052,6 +1091,7 @@ def _format_results(results: dict, args) -> str:
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)
@@ -1188,9 +1228,89 @@ def _run_single_query(query: str, args, instance_urls: list,
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.
"""
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' (single query). "
"Batch mode (--queries-file) wraps results in "
'{"schema_version, queries:[]}.',
"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": {
"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"],
},
},
"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.",
},
"fetched": {
"type": "array",
"description": "Present only when --fetch N is used. Page content "
"for the top N results.",
"items": {
"type": "object",
"properties": {
"url": {"type": "string"},
"text": {"type": "string"},
"text_length": {"type": "integer"},
"error": {"type": "string"},
},
},
},
"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.",
},
},
"required": ["query", "results"],
}
def _emit_error(message: str, args, query: str = None, exit_code: int = 1,
error_code: str = None):
"""Emit an error and exit.
@@ -1201,22 +1321,30 @@ def _emit_error(message: str, args, query: str = None, exit_code: int = 1,
The JSON shape is::
{"error": "...", "exit_code": N, "error_code": "E_*", "query": "..."}
{"error": "...", "exit_code": N, "error_code": "E_*",
"recovery_hint": "...", "query": "..."}
``error_code`` 是结构化错误码（E_CONFIG/E_AUTH/E_NETWORK 等），让 AI
Agent 程序化判断错误类型并采取恢复策略。``query`` 仅在提供时包含。
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 ""
logger.error(f"{prefix}{code_prefix}Error: {message}")
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)
@@ -1406,13 +1534,35 @@ Use --config FILE to load a non-default config file (overrides the auto-discover
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()
# --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)
# 启用 --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.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)
# --query is required unless we're doing a non-search operation.
# --queries-file is an alternative to --query for batch mode.
if (not args.verify and not args.query and not args.queries_file
@@ -1427,6 +1577,33 @@ Use --config FILE to load a non-default config file (overrides the auto-discover
apply_proxy(args.proxy)
logger.info(f"Proxy: {args.proxy}")
# --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
@@ -1441,9 +1618,26 @@ Use --config FILE to load a non-default config file (overrides the auto-discover
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.", args, error_code=E_CONFIG)
"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
@@ -1473,29 +1667,6 @@ Use --config FILE to load a non-default config file (overrides the auto-discover
_print_verify_report(report, as_json=(args.format == "json"))
sys.exit(0)
# Cache management modes (no search performed)
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":
print(json.dumps(s, indent=2, ensure_ascii=False))
else:
print(f"Cache path: {s.get('path', '?')}")
print(f"Entries: {s.get('entries', 0)}")
size = s.get("size_bytes", 0)
print(f"Size: {size:,} bytes ({size / 1024:.1f} KB)")
if s.get("oldest_created_at"):
print(f"Oldest: {time.ctime(s['oldest_created_at'])}")
if s.get("newest_created_at"):
print(f"Newest: {time.ctime(s['newest_created_at'])}")
if s.get("error"):
print(f"Error: {s['error']}")
sys.exit(0)
if args.fail_fast:
instance_urls = instance_urls[:1]
@@ -1519,23 +1690,27 @@ Use --config FILE to load a non-default config file (overrides the auto-discover
logger.info(f"Running {len(queries)} queries from {args.queries_file}...")
batch = []
any_ok = False
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, "error": err}
entry = {"query": q, "status": "error", "error": err}
if err_code:
entry["error_code"] = err_code
batch.append(entry)
else:
any_ok = True
batch.append({"query": q, "results": results})
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(batch, indent=2, ensure_ascii=False)
output = json.dumps({"schema_version": SCHEMA_VERSION, "queries": batch},
indent=2, ensure_ascii=False)
elif args.format == "csv":
import csv as csv_mod
import io
@@ -1588,27 +1763,49 @@ Use --config FILE to load a non-default config file (overrides the auto-discover
logger.info(f"Saved results to {args.output}")
else:
print(output)
# Exit 0 if at least one query succeeded; 1 only if all failed.
sys.exit(0 if any_ok else 1)
# Exit code semantics — aligned with single-query mode so AI agents
# can use one consistent rule:
# 1 = all queries errored (fatal)
# 2 = no query returned any results (empty), though at least one
# searched successfully without error
# 0 = at least one query returned results
if error_count == len(queries):
sys.exit(1)
if not any_with_results:
sys.exit(2)
sys.exit(0)
# ----- Single query mode -----
results, err, err_code = _run_single_query(args.query, args, instance_urls,
auth_headers, ttl_seconds)
if err:
_emit_error(err, args, query=args.query, error_code=err_code)
# --stream: JSON Lines 流式输出，每条结果一行，AI 可增量处理
# --stream: JSON Lines 流式输出，每条结果一行，AI 可增量处理。
# stream 模式下所有输出（包括错误）都是单行 JSON，保持 JSON Lines 格式一致性。
# 非 stream 模式的错误走 _emit_error（多行 JSON 或 stderr 文本）。
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:
tests/test_batch.py
+134
View File
@@ -0,0 +1,134 @@
"""Tests for --queries-file batch mode: exit code semantics and JSON schema.
Covers:
* Exit codes: all-empty (no error) -> 2, all-error -> 1, partial -> 0
* Batch JSON shape: {"schema_version": "1.0", "queries": [...]}
* Per-entry "status" field ("ok"/"error") and result/error payload shape
Uses in-process main() with search_multi mocked so no network is touched.
``--retry 0`` is passed to avoid backoff sleeps when search_multi raises.
"""
import json
import sys
import urllib.error
from unittest.mock import patch
import pytest
import search as search_mod
from search import main
def _batch_argv(queries_file_path, fmt="json"):
"""Build the sys.argv for a batch main() invocation.
``--retry 0`` keeps failing tests fast (no backoff sleeps).
"""
return ["search.py", "-i", "https://x.example.com",
"--queries-file", str(queries_file_path),
"--format", fmt, "--retry", "0"]
# ===== batch exit code semantics =====
def test_batch_all_empty_results_exits_2(tmp_path, capsys):
"""All queries return empty results (no error) -> exit 2."""
qf = tmp_path / "queries.txt"
qf.write_text("q1\nq2\n", encoding="utf-8")
empty = {"results": []}
with patch.object(search_mod, "search_multi", return_value=empty):
with pytest.raises(SystemExit) as exc_info:
with patch.object(sys, "argv", _batch_argv(qf)):
with patch.object(search_mod, "setup_logging"):
main()
assert exc_info.value.code == 2
def test_batch_all_error_exits_1(tmp_path, capsys):
"""All queries error -> exit 1."""
qf = tmp_path / "queries.txt"
qf.write_text("q1\nq2\n", encoding="utf-8")
err = urllib.error.URLError("connection refused")
with patch.object(search_mod, "search_multi", side_effect=err):
with pytest.raises(SystemExit) as exc_info:
with patch.object(sys, "argv", _batch_argv(qf)):
with patch.object(search_mod, "setup_logging"):
main()
assert exc_info.value.code == 1
def test_batch_partial_results_exits_0(tmp_path, capsys):
"""At least one query returns results -> exit 0."""
qf = tmp_path / "queries.txt"
qf.write_text("q1\nq2\n", encoding="utf-8")
with_data = {"results": [{"url": "https://a.com/1", "title": "A"}]}
empty = {"results": []}
with patch.object(search_mod, "search_multi",
side_effect=[with_data, empty]):
with pytest.raises(SystemExit) as exc_info:
with patch.object(sys, "argv", _batch_argv(qf)):
with patch.object(search_mod, "setup_logging"):
main()
assert exc_info.value.code == 0
# ===== batch unified JSON schema =====
def test_batch_json_has_schema_version_and_queries(tmp_path, capsys):
"""Batch JSON output wraps entries in {schema_version, queries}."""
qf = tmp_path / "queries.txt"
qf.write_text("q1\n", encoding="utf-8")
mock_results = {"results": [{"url": "https://a.com/1", "title": "A"}]}
with patch.object(search_mod, "search_multi", return_value=mock_results):
with pytest.raises(SystemExit):
with patch.object(sys, "argv", _batch_argv(qf)):
with patch.object(search_mod, "setup_logging"):
main()
out, _ = capsys.readouterr()
data = json.loads(out)
assert data["schema_version"] == "1.0"
assert "queries" in data
assert isinstance(data["queries"], list)
assert len(data["queries"]) == 1
def test_batch_entry_has_status_field(tmp_path, capsys):
"""Each batch entry has a 'status' field of 'ok' or 'error'."""
qf = tmp_path / "queries.txt"
qf.write_text("q1\nq2\n", encoding="utf-8")
ok_results = {"results": [{"url": "https://a.com/1", "title": "A"}]}
err = urllib.error.URLError("down")
with patch.object(search_mod, "search_multi",
side_effect=[ok_results, err]):
with pytest.raises(SystemExit):
with patch.object(sys, "argv", _batch_argv(qf)):
with patch.object(search_mod, "setup_logging"):
main()
out, _ = capsys.readouterr()
data = json.loads(out)
statuses = [e["status"] for e in data["queries"]]
assert "ok" in statuses
assert "error" in statuses
def test_batch_success_entry_has_results_error_entry_has_error_and_code(tmp_path, capsys):
"""ok entry has 'results'; error entry has 'error' and 'error_code'."""
qf = tmp_path / "queries.txt"
qf.write_text("q1\nq2\n", encoding="utf-8")
ok_results = {"results": [{"url": "https://a.com/1", "title": "A"}]}
err = urllib.error.URLError("connection refused")
with patch.object(search_mod, "search_multi",
side_effect=[ok_results, err]):
with pytest.raises(SystemExit):
with patch.object(sys, "argv", _batch_argv(qf)):
with patch.object(search_mod, "setup_logging"):
main()
out, _ = capsys.readouterr()
data = json.loads(out)
ok_entry = next(e for e in data["queries"] if e["status"] == "ok")
assert "results" in ok_entry
err_entry = next(e for e in data["queries"] if e["status"] == "error")
assert "error" in err_entry
assert "error_code" in err_entry
# URLError -> E_NETWORK per classify_error
assert err_entry["error_code"] == "E_NETWORK"
tests/test_cli_e2e.py
+116 -2
View File
@@ -23,7 +23,7 @@ from types import SimpleNamespace
import pytest
from search import _run_single_query, _emit_error, _build_params
from search import _run_single_query, _emit_error, _build_params, _format_results
import cache as cache_module
PROJECT_ROOT = Path(__file__).resolve().parent.parent
@@ -256,7 +256,7 @@ def test_cli_version_prints_version():
"""`--version` exits 0 and prints the version string."""
r = _run_cli("--version")
assert r.returncode == 0
assert "1.7.0" in r.stdout
assert "1.8.0" in r.stdout
assert "searxng-cli" in r.stdout
@@ -294,3 +294,117 @@ def test_cli_cache_stats_outputs_json(tmp_path):
data = json.loads(r.stdout)
assert "entries" in data
assert "path" in data
# ===== --dump-schema (v1.8.0) =====
def test_cli_dump_schema_outputs_valid_json():
"""`--dump-schema` prints valid JSON with schema_version/title/properties.
The schema dump is a non-search operation: it exits 0 and does NOT
require --query or -i (it short-circuits before instance resolution).
"""
r = _run_cli("--dump-schema")
assert r.returncode == 0
data = json.loads(r.stdout)
assert data["schema_version"] == "1.0"
assert data["title"] == "SearXNG CLI Search Result"
assert "properties" in data
assert isinstance(data["properties"], dict)
def test_cli_dump_schema_does_not_require_query():
"""`--dump-schema` exits 0 without --query (non-search operation)."""
r = _run_cli("--dump-schema")
assert r.returncode == 0
# The --query-required check must not fire for --dump-schema.
assert "required" not in r.stderr.lower()
# ===== --stream mutual exclusion (v1.8.0) =====
def test_cli_stream_with_queries_file_rejected(tmp_path):
"""--stream + --queries-file is rejected with E_INPUT on stdout.
Batch mode emits a JSON array, not JSON Lines; the combination is
explicitly rejected so AI agents don't silently get the wrong format.
"""
qf = tmp_path / "queries.txt"
qf.write_text("query\n", encoding="utf-8")
r = _run_cli("--stream", "--queries-file", str(qf),
"--format", "json", "-i", "https://x.example.com")
assert r.returncode == 1
data = json.loads(r.stdout)
assert data["error_code"] == "E_INPUT"
def test_cli_stream_with_csv_format_rejected():
"""--stream + --format csv is rejected with E_INPUT.
csv is a non-json format, so _emit_error routes the error to stderr
(as a ``[E_INPUT]`` prefixed log line) rather than stdout JSON.
"""
r = _run_cli("--stream", "--format", "csv", "-q", "test",
"-i", "https://x.example.com")
assert r.returncode == 1
assert "E_INPUT" in r.stderr
# ===== schema_version in single-query JSON (v1.8.0) =====
def test_format_results_json_includes_schema_version():
"""Single-query JSON output includes schema_version: '1.0'."""
results = {"results": [{"title": "T", "url": "https://example.com"}]}
args = SimpleNamespace(format="json")
out = _format_results(results, args)
parsed = json.loads(out)
assert parsed["schema_version"] == "1.0"
# ===== _fallback field must not leak to JSON output (v1.8.0) =====
def test_run_single_query_html_fallback_pops_underscore_fallback(monkeypatch):
"""HTML-fallback's internal _fallback field is removed before output.
_run_single_query pops _fallback at the end so it never appears in
the JSON payload; fetched_source (the public field) is set separately
on the --fetch path.
"""
args = _make_args(format="json")
fake = {"results": [{"url": "https://a.com/1", "title": "A"}],
"_fallback": "html"}
monkeypatch.setattr("search.search_multi", lambda *a, **kw: fake)
results, err, _ = _run_single_query("test", args,
["https://x.example.com"], {}, 0)
assert err is None
# _fallback must be popped so it doesn't leak into JSON output
assert "_fallback" not in results
# And the serialized JSON also omits it
out = _format_results(results, args)
parsed = json.loads(out)
assert "_fallback" not in parsed
def test_run_single_query_fetch_sets_fetched_source_from_html_fallback(monkeypatch):
"""--fetch path sets fetched_source from _fallback (or 'json').
When the search came via HTML fallback (_fallback='html'), the
fetched_source field reflects that origin, while _fallback itself is
cleaned up and never reaches JSON output.
"""
args = _make_args(fetch=1)
fake = {"results": [{"url": "https://a.com/1"}], "_fallback": "html"}
monkeypatch.setattr("search.search_multi", lambda *a, **kw: fake)
monkeypatch.setattr("search.fetch_top_results", lambda *a, **kw: [
{"url": "https://a.com/1", "status": "ok", "text": "page A",
"text_length": 6, "truncated": False},
])
results, err, _ = _run_single_query("test", args,
["https://x.example.com"], {}, 0)
assert err is None
# fetched_source reflects the HTML fallback origin
assert results["fetched_source"] == "html"
# _fallback still cleaned up
assert "_fallback" not in results
tests/test_common.py
+6 -3
View File
@@ -143,8 +143,10 @@ def test_non_retryable_404():
assert is_retryable_error(_http_error(404)) is False
def test_non_retryable_403():
assert is_retryable_error(_http_error(403)) is False
def test_retryable_403():
# 403 是可重试的：让 fetch_url 的 UA-fallback 循环有机会切换到浏览器 UA。
# 真正的认证错误会在重试耗尽后由 classify_error 归为 E_AUTH。
assert is_retryable_error(_http_error(403)) is True
def test_non_retryable_200():
@@ -160,7 +162,8 @@ def test_retryable_os_error():
def test_retryable_status_set_contents():
assert RETRYABLE_STATUS == frozenset({429, 502, 503, 504})
# 403 加入可重试集合，让 UA-fallback 在 UA 被屏蔽时有机会切换浏览器 UA
assert RETRYABLE_STATUS == frozenset({403, 429, 502, 503, 504})
# ----- apply_proxy -----
tests/test_error_codes.py
+67
View File
@@ -231,3 +231,70 @@ def test_run_single_query_propagates_rate_limit():
_, _, err_code = _run_single_query(
"test", args, ["https://x.example.com"], {}, 0)
assert err_code == E_RATE_LIMIT
# ----- classify_error: HTTP status extraction from RuntimeError messages -----
#
# search_multi wraps the last error into its RuntimeError message, e.g.
# "All 3 instances failed. Last error: HTTP Error 403: Forbidden"
# classify_error must extract the status code from that message rather than
# always falling back to E_NETWORK.
def test_classify_runtime_all_instances_failed_with_http_403_is_auth():
"""'...Last error: HTTP Error 403' -> E_AUTH (not E_NETWORK)."""
err = RuntimeError("All 3 instances failed. Last error: HTTP Error 403: Forbidden")
assert classify_error(err) == E_AUTH
def test_classify_runtime_all_instances_failed_with_http_500_is_network():
"""'...Last error: HTTP Error 500' -> E_NETWORK (5xx server error)."""
err = RuntimeError("All 3 instances failed. Last error: HTTP Error 500: Internal Server Error")
assert classify_error(err) == E_NETWORK
def test_classify_runtime_parallel_failed_with_http_429_is_rate_limit():
"""'...(parallel). Last error: HTTP Error 429' -> E_RATE_LIMIT."""
err = RuntimeError("All 3 instances failed (parallel). Last error: HTTP Error 429: Too Many Requests")
assert classify_error(err) == E_RATE_LIMIT
# ----- _emit_error: recovery_hint in JSON output -----
#
# recovery_hint gives AI agents an actionable suggestion per error_code.
# Only present when error_code is known; omitted otherwise (backwards compat).
def test_emit_error_json_includes_recovery_hint_for_config(capsys):
"""JSON error with E_CONFIG includes the config recovery_hint."""
args = SimpleNamespace(format="json")
with pytest.raises(SystemExit):
_emit_error("no instance resolved", args, error_code=E_CONFIG)
out, _ = capsys.readouterr()
data = json.loads(out)
assert data["error_code"] == E_CONFIG
assert "recovery_hint" in data
hint = data["recovery_hint"].lower()
assert "instance" in hint or "config" in hint
def test_emit_error_json_includes_recovery_hint_for_auth(capsys):
"""JSON error with E_AUTH includes the auth recovery_hint."""
args = SimpleNamespace(format="json")
with pytest.raises(SystemExit):
_emit_error("forbidden", args, error_code=E_AUTH)
out, _ = capsys.readouterr()
data = json.loads(out)
assert data["error_code"] == E_AUTH
assert "recovery_hint" in data
hint = data["recovery_hint"].lower()
assert "credential" in hint or "token" in hint
def test_emit_error_json_no_recovery_hint_without_error_code(capsys):
"""No error_code -> no recovery_hint field (backwards compat)."""
args = SimpleNamespace(format="json")
with pytest.raises(SystemExit):
_emit_error("something broke", args)
out, _ = capsys.readouterr()
data = json.loads(out)
assert "error_code" not in data
assert "recovery_hint" not in data
tests/test_stream_progress.py
+114 -1
View File
@@ -6,8 +6,9 @@ disabled by default, stream only works with --format json.
"""
import json
import sys
import urllib.error
from types import SimpleNamespace
from unittest.mock import patch
from unittest.mock import MagicMock, patch
import pytest
import common
@@ -205,3 +206,115 @@ def test_stream_result_event_shape():
assert parsed[0]["result"]["url"] == "https://x.com"
assert parsed[-1]["type"] == "done"
assert parsed[-1]["count"] == 1
# ----- stream done/error: schema_version & recovery_hint (v1.8.0) -----
def test_stream_done_includes_schema_version(capsys):
"""stream done event includes the schema_version field."""
mock_results = {"results": [{"title": "t", "url": "https://x.com"}]}
with patch.object(search_mod, "search_multi", return_value=mock_results):
with pytest.raises(SystemExit) as exc_info:
with patch.object(sys, "argv", ["search.py", "-q", "test",
"-i", "https://x.example.com",
"--stream", "--format", "json",
"--retry", "0"]):
with patch.object(search_mod, "setup_logging"):
main()
assert exc_info.value.code == 0
out, _ = capsys.readouterr()
lines = [json.loads(l) for l in out.strip().split("\n") if l]
done_events = [e for e in lines if e["type"] == "done"]
assert len(done_events) == 1
assert done_events[0]["schema_version"] == "1.0"
def test_stream_error_includes_recovery_hint(capsys):
"""stream error event includes recovery_hint for classified errors.
HTTP 403 -> E_AUTH, which has a recovery_hint in RECOVERY_HINTS.
--retry 0 avoids backoff sleeps on the (mocked) 403.
"""
err = urllib.error.HTTPError("url", 403, "Forbidden", {}, None)
with patch.object(search_mod, "search_multi", side_effect=err):
with pytest.raises(SystemExit) as exc_info:
with patch.object(sys, "argv", ["search.py", "-q", "test",
"-i", "https://x.example.com",
"--stream", "--format", "json",
"--retry", "0"]):
with patch.object(search_mod, "setup_logging"):
main()
assert exc_info.value.code == 1
out, _ = capsys.readouterr()
lines = [json.loads(l) for l in out.strip().split("\n") if l]
error_events = [e for e in lines if e["type"] == "error"]
assert len(error_events) == 1
assert error_events[0]["error_code"] == "E_AUTH"
assert "recovery_hint" in error_events[0]
# ----- instance_try / instance_ok / instance_fail progress events (v1.8.0) -----
#
# These events are emitted inside search_multi (not _run_single_query), so
# we must let the real search_multi run and only mock urllib.request.urlopen.
def _mock_urlopen_resp(data: bytes, content_type="application/json"):
"""Build a MagicMock that quacks like an urlopen context manager."""
resp = MagicMock()
resp.read.return_value = data
resp.headers = {"Content-Type": content_type}
resp.__enter__.return_value = resp
resp.__exit__.return_value = None
return resp
def _url_router(routing):
"""Build a side_effect that dispatches urlopen by request URL substring."""
def _side_effect(req, *args, **kwargs):
url = getattr(req, "full_url", str(req))
for key, resp in routing.items():
if key in url:
if isinstance(resp, BaseException):
raise resp
return resp
raise AssertionError(f"unexpected urlopen for {url!r}")
return _side_effect
def test_progress_emits_instance_try_ok_fail(capsys):
"""instance_try/instance_ok/instance_fail events are emitted to stderr.
Uses serial mode (parallel=False) so the events come out in a
deterministic order: try a -> fail a -> try b -> ok b.
"""
set_progress_enabled(True)
try:
payload = json.dumps(
{"results": [{"title": "b", "url": "https://b.com"}]}
).encode()
router = _url_router({
"a.example.com": urllib.error.URLError("down"),
"b.example.com": _mock_urlopen_resp(payload),
})
with patch("urllib.request.urlopen", side_effect=router):
search_mod.search_multi(
["https://a.example.com", "https://b.example.com"],
{"q": "test", "format": "json"},
parallel=False, retry_per=0,
)
_, err = capsys.readouterr()
lines = [json.loads(l) for l in err.strip().split("\n") if l]
events = [e["event"] for e in lines]
assert "instance_try" in events
assert "instance_fail" in events
assert "instance_ok" in events
# instance_fail carries error_code
fail_events = [e for e in lines if e["event"] == "instance_fail"]
assert len(fail_events) >= 1
assert "error_code" in fail_events[0]
# instance_ok carries results count
ok_events = [e for e in lines if e["event"] == "instance_ok"]
assert len(ok_events) >= 1
assert "results" in ok_events[0]
finally:
set_progress_enabled(False)