diff --git a/README.md b/README.md index 8b64ae2..be13603 100644 --- a/README.md +++ b/README.md @@ -206,6 +206,7 @@ python scripts/search.py -q "查询词" -i https://your-instance \ | `--request-delay` | v2.0.0 抓取请求间隔秒数（自适应限流可能增大） | 0.3 | | `--cache-ttl` | 缓存分钟数 | 0（不缓存） | | `--queries-file` | 批量查询文件（每行一个查询） | — | +| `--parallel-queries` | v2.3.0 并发批量查询 worker 数（1-8，输出保序，受自适应限流门控；并发时禁用 --fetch） | 0（串行） | | `--research` | v2.1.0 研究模式：给定主题自动扩展 5 个多角度查询 | — | | `--include-domain` | 域名白名单 | — | | `--exclude-domain` | 域名黑名单 | — | @@ -236,6 +237,7 @@ python scripts/search.py -q "查询词" -i https://your-instance \ ```bash python scripts/fetch.py -u https://example.com \ --extract text|html|markdown \ + --format text|json \ # v2.3.0: json = 结构化 JSON 契约 [--encoding gbk] \ [--max-size 5242880] \ [--timeout 15] \ @@ -249,6 +251,7 @@ python scripts/fetch.py -u https://example.com \ | 参数 | 说明 | 默认值 | |------|------|--------| | `-u / --url` | 目标 URL（必填） | — | +| `-f / --format` | v2.3.0 输出格式：text/json（json = 结构化 JSON 契约，成功 `{status,url,final_url,content_type,extract,truncated,text_length,user_agent}`，失败 `{status,error,error_code,status_code,url}`） | text | | `-e / --extract` | 提取模式：text/html/markdown | text | | `--encoding` | 强制字符编码 | 自动检测 | | `--max-size` | 最大字节数 | 不限 | @@ -263,6 +266,17 @@ python scripts/fetch.py -u https://example.com \ ## 能力清单 +**v2.3.0 新功能与修复** +- `fetch.py --format json`：结构化 JSON 输出契约。成功 `{status, url, final_url, content_type, extract, truncated, text_length, user_agent}`；失败 `{status, error, error_code, status_code, url}`（对齐 search.py 错误码体系，含 `E_UNSUPPORTED_MEDIA`）。AI Agent 可程序化解析抓取结果，无需再解析裸文本 +- `--parallel-queries N`：批量模式并发执行（1-8 workers，输出保持文件顺序），受 AdaptiveThrottle 真实并发门控约束；并发模式下禁用 `--fetch`（嵌套并行抓取不安全）。queries 文件编码自动检测（UTF-8 → GBK 回退） +- 真实并发门控：`AdaptiveThrottle.acquire_slot()/release_slot()`——退避降并发后新请求被快速拒绝（返回 `E_RATE_LIMIT`），实现持久降并发（原实现仅名义降并发，线程池规模固定） +- `--fetch-report json` 字段补齐：`fetch_page` 采集 `title`（页面标题）与 `latency`（耗时秒数），此前恒为 None +- `--dump-schema` 扩展：`fetched.items` 字段补全（final_url/status/error_code/waf_type/fallback_used/title/latency 等），新增 `defs.batch`/`defs.research` 描述批量与研究模式输出 shape +- 修复多页聚合（`--pages N`）的 unresponsive-engine 警告误判：原实现用循环末次 `cached` 变量判断，缓存命中时警告被错误跳过/误触发；现用独立 `performed_live_query` 标记 +- UA 池单一来源：删除 common.py 的手工副本 `_FALLBACK_UAS_BUILTIN`，`FALLBACK_UAS` 直接引用 `_config.UA_POOL`（消除双份漂移风险） +- search_html 解码修复：改用 `detect_charset`（header/meta 自动检测），新增 `--encoding` 强制覆盖（非 UTF-8 SearXNG 实例不再乱码） +- 新增 `pyproject.toml`（可选打包，`searxng-search`/`searxng-fetch` 入口点）；函数内冗余导入收敛 + **v2.2.1 修复** - Brotli 乱码修复：`build_browser_headers()` 智能声明 `Accept-Encoding`——仅当本机安装了 brotli/brotlicffi 包时才声明 `br`，避免服务器返回 Brotli 压缩字节而 requests 无法自动解压导致全页乱码 - `fetch.py` 双路径 br 解压：requests 路径和 stdlib urllib 路径都添加了 Brotli 手动解压逻辑（作为双保险，应对代理/CDN 强制返回 br 的边缘情况） @@ -351,7 +365,7 @@ python scripts/fetch.py -u https://example.com \ - JSON Lines 流式输出（`--stream`，含 `error` 事件类型） - 进度事件（`--progress`，含 `instance_try`/`instance_ok`/`instance_fail`，JSON Lines 到 stderr） - batch 模式统一 schema（`status` 字段区分成功/失败） -- 544 个单元+集成测试 +- 561 个单元+集成测试 ## 跨 Agent 兼容性 @@ -379,7 +393,7 @@ pip install pytest pytest -q ``` -539 个测试覆盖：缓存操作、认证解析、域名过滤、Markdown 转换、搜索逻辑、集成流程、日志配置、HTML 回退、自动抓取、健康检查、输出格式化、实例解析、并行搜索、CLI 端到端、错误码分类、流式输出、进度事件、配置文件认证、schema_version、recovery_hint、batch 统一 schema、--dump-schema、UTF-8 stdout 强制、Windows APPDATA 路径、v2.0.0 浏览器指纹头、WAF 反爬检测（v2.0.1 收窄误判 + title 精准检测）、Wayback Machine 兜底（v2.0.1 修复 HTTP 200 反爬页触发）、自适应限流（v2.1.1 结构化 error_code 检测 429）、v2.1.0 Wayback 共享逻辑（should_try_wayback/build_wayback_url）、被墙站点智能回退（v2.1.1 精确化 baidu 子域，pan/cloud 不再误伤）、--research 研究模式（v2.1.1 跨角度合并去重 + 中英文双语后缀）、--time-range week 支持（v2.1.1）、stdlib gzip/deflate 解压（v2.1.1）、实例引擎挂起检测（v2.1.1 _warn_unresponsive_engines）。 +561 个测试覆盖：缓存操作、认证解析、域名过滤、Markdown 转换、搜索逻辑、集成流程、日志配置、HTML 回退、自动抓取、健康检查、输出格式化、实例解析、并行搜索、CLI 端到端、错误码分类、流式输出、进度事件、配置文件认证、schema_version、recovery_hint、batch 统一 schema、--dump-schema、UTF-8 stdout 强制、Windows APPDATA 路径、v2.0.0 浏览器指纹头、WAF 反爬检测（v2.0.1 收窄误判 + title 精准检测）、Wayback Machine 兜底（v2.0.1 修复 HTTP 200 反爬页触发）、自适应限流（v2.1.1 结构化 error_code 检测 429）、v2.1.0 Wayback 共享逻辑（should_try_wayback/build_wayback_url）、被墙站点智能回退（v2.1.1 精确化 baidu 子域，pan/cloud 不再误伤）、--research 研究模式（v2.1.1 跨角度合并去重 + 中英文双语后缀）、--time-range week 支持（v2.1.1）、stdlib gzip/deflate 解压（v2.1.1）、实例引擎挂起检测（v2.1.1 _warn_unresponsive_engines）、v2.3.0 新功能（多页实时查询跟踪、GBK charset 检测与 --encoding、UA 单源、并发槽位门控、fetch JSON 契约、fetch_page title/latency、GBK queries 文件、--parallel-queries 并发批量）。 ## 项目结构 diff --git a/SKILL.md b/SKILL.md index 14cd93c..3bb0dd8 100644 --- a/SKILL.md +++ b/SKILL.md @@ -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: 2.2.1 +version: 2.3.0 author: Metona Team license: MIT platforms: [linux, macos, windows] @@ -45,8 +45,8 @@ SearXNG is a privacy-respecting metasearch engine that aggregates results from 7 - Similarity deduplication (`--similarity-dedup`; off by default) — title-based SimHash + Jaccard, auto-skipped when results > 500 - Result sorting (`--sort-by {score,date,engine,none}`; default: score descending) — applied after dedup, before `--max-results` - Domain allowlist/blocklist (`--include-domain` / `--exclude-domain`) — case-insensitive, ignores leading `www.`, exclude wins on conflict +- Batch mode (`--queries-file`) — run multiple queries from a file in sequence, combined output; v2.3.0 adds `--parallel-queries N` (1-8 concurrent, output order preserved, gated by AdaptiveThrottle) - Pagination aggregation (`--pages N`) — fetch N pages in one run, merge with cross-page dedup -- Batch mode (`--queries-file`) — run multiple queries from a file in sequence, combined output **Output formats** - JSON (default, rich metadata), brief (title+URL+snippet), urls (plain list), CSV (spreadsheet-ready) @@ -54,6 +54,7 @@ SearXNG is a privacy-respecting metasearch engine that aggregates results from 7 - Structured JSON error output (in `--format json` mode) with `error_code` field for machine-readable failure reporting - JSON Lines streaming (`--stream`) — each result emitted as a separate JSON line to stdout, enabling incremental processing - Progress events (`--progress`) — structured JSON Lines events to stderr for real-time execution tracking, includes `request_id` for correlation +- `fetch.py --format json` (v2.3.0) — structured JSON contract for page fetches: success `{status, url, final_url, content_type, extract, truncated, text_length, user_agent}`, failure `{status, error, error_code, status_code, url}` (aligns with search.py's error-code system) **Caching & config** - SQLite result caching (`--cache-ttl`) with size cap + LRU eviction (`--cache-max-size`, default 100MB) — identical queries within a TTL skip the network entirely; `--clear-cache` / `--cache-stats` manage it @@ -76,14 +77,14 @@ SearXNG is a privacy-respecting metasearch engine that aggregates results from 7 - **WAF fingerprint library** — `_detect_anti_bot()` (in `search.py`, used by `--fetch` flow) identifies Cloudflare / Imperva / PerimeterX / DataDome / Akamai / Anubis / generic challenges via `` tag matching + full-document scan. Returns `waf_type` for AI-agent decisioning - **Wayback Machine fallback** — 404/403/timeout AND anti-bot-blocked pages automatically retry via `https://web.archive.org/web/2/<url>` (latest snapshot). Default ON; `--no-fallback` disables. Wayback timeout = `min(--timeout, 10s)`; Wayback retries = `min(--retries, 2)` - **Hard-blocked domain fallback** — `is_hard_blocked_domain()` (in `common.py`) detects known strong-anti-bot sites (www.baidu.com, baike.baidu.com, zhihu.com, weibo.com, mp.weixin.qq.com, douban.com, etc.) that almost always return 403. These trigger Wayback immediately on failure. List maintained in `HARD_BLOCKED_DOMAINS`. Note: baidu uses exact subdomain matching (www/baike/zhidao/tieba/wenku) to avoid over-blocking pan.baidu.com; zhihu/weibo/douban use subdomain wildcard matching -- **Adaptive throttling** — `AdaptiveThrottle` (in `search.py`) state machine: consecutive failures → double delay + halve concurrency; consecutive successes → gradual recovery; 429 → global pause. Thread-safe. Parameters configurable via `--throttle-failure-threshold` / `--throttle-pause-seconds` / `--throttle-max-delay` +- **Adaptive throttling** — `AdaptiveThrottle` (in `search.py`) state machine: consecutive failures → double delay + halve concurrency; consecutive successes → gradual recovery; 429 → global pause. Thread-safe. Parameters configurable via `--throttle-failure-threshold` / `--throttle-pause-seconds` / `--throttle-max-delay`. v2.3.0: real concurrency gating via `acquire_slot()`/`release_slot()` — after backoff reduces concurrency, new requests are rejected until in-flight drops below the target (persistent throttling, not just nominal) - **readability-lite extraction** — when `<article>`/`<main>`/`role="main"`/content-class `<div>` are all missing and only `<body>` remains, picks the highest text-density `<div>`/`<section>`/`<article>` node (scored by text density + `<p>` count weighting). Text-density threshold is language-aware — CJK content uses 100 chars, other languages use 200 chars - **PDF/document parsing** — `fetch.py` parses PDF (via `pdftotext` subprocess) and `.docx`/`.xlsx` (via stdlib `zipfile`). Unsupported binary types return `E_UNSUPPORTED_MEDIA` - **`--fetch-report [json]`** — structured per-URL report to stderr after `--fetch N`: status, WAF type, fallback used, char count, plus throttle stats. `--fetch-report json` outputs a full JSON report (items array + summary) - **`--referer`** — set Referer header for fetch requests (defaults to the instance URL when fetching result pages) - **`--request-delay`** — configurable delay between fetch requests (default 0.3s; adaptive throttling may increase this on failures) -- **`search.py --fetch` output fields** (in `fetched` array): `anti_bot_detected` (bool), `waf_type` (str|null), `fallback_used` (str|null), `error_code` (str|null — structured error code on fetch failure) -- **`fetch.py` FetchResult fields** (standalone script output): `content` (str), `content_type` (str), `final_url` (str), `truncated` (bool), `user_agent` (str), `error_code` (str|null), `error_message` (str|null) +- **`search.py --fetch` output fields** (in `fetched` array): `anti_bot_detected` (bool), `waf_type` (str|null), `fallback_used` (str|null), `error_code` (str|null — structured error code on fetch failure); v2.3.0 also `title` (str|null) and `latency` (float|null seconds) — consumed by `--fetch-report json` +- **`fetch.py` FetchResult fields** (standalone script output): `content` (str), `content_type` (str), `final_url` (str), `truncated` (bool), `user_agent` (str), `error_code` (str|null), `error_message` (str|null); with `--format json` (v2.3.0) these map to a structured JSON contract **Research mode** - **`--research <topic>`** — given a research topic, auto-expands into 5 multi-angle queries (overview/profile/background/works/review) and runs them in sequence. Results include `research_topic` and `research_queries` metadata so AI agents can structure their final report by angle @@ -432,7 +433,7 @@ usage: search.py [-h] [--query QUERY] [--instance URL] [--auth-basic USER:PASS] [--auth-basic-file FILE] [--proxy URL] [--include-domain DOMAINS] [--exclude-domain DOMAINS] [--queries-file FILE] - [--research TOPIC] [--research-angles ANGLES] + [--parallel-queries N] [--research TOPIC] [--research-angles ANGLES] [--cache-ttl MINUTES] [--cache-max-size MB] [--clear-cache] [--cache-stats] [--sort-by {score,date,engine,none}] [--no-dedup] [--similarity-dedup] [--similarity-threshold FLOAT] @@ -453,7 +454,7 @@ usage: search.py [-h] [--query QUERY] [--instance URL] 8. **Auto-fetch:** `--fetch 3` concurrently downloads top 3 result pages with retry, full browser fingerprint headers, 15-UA deterministic per-domain pool, Retry-After compliance, WAF fingerprint detection, Wayback Machine fallback for 404/403/timeout, adaptive throttling, and `fetch.py`'s readability-lite extractor. `--fetch-report` prints a structured per-URL report to stderr 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, LRU eviction). `--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; 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 +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. v2.3.0: `--parallel-queries N` (1-8) runs queries concurrently with output order preserved; concurrent mode disables `--fetch` (nested parallel fetch is unsafe). Queries-file encoding is auto-detected (UTF-8, GBK fallback — v2.3.0). 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; explicit CLI flags always win @@ -520,6 +521,7 @@ usage: search.py [-h] [--query QUERY] [--instance URL] ``` usage: fetch.py [-h] --url URL [--extract {text,html,markdown}] + [--format {text,json}] [--timeout SEC] [--retries N] [--max-size BYTES] [--user-agent STR] [--encoding CHARSET] [--no-redirect] [--no-fallback] [--referer URL] [--proxy URL] @@ -539,6 +541,7 @@ usage: fetch.py [-h] --url URL [--extract {text,html,markdown}] **Key options:** - `--url https://...` — required +- `--format {text,json}` — output format (default: `text`; `json` = structured contract, v2.3.0). Success: `{status: ok, url, final_url, content_type, extract, truncated, text_length, user_agent}`; failure: `{status: error, url, error, error_code, status_code}`. stdout is pure JSON; logs stay on stderr; exit 0 on success / 1 on failure - `--extract text` — clean readable text (default); PDF/docx/xlsx parsed automatically regardless of extract mode (see "What it does" #7) - `--extract html` — raw HTML - `--extract markdown` — Markdown conversion (tree-based; handles nested tags, GFM tables, fenced code blocks, blockquotes, inline code, ordered/unordered/nested lists, definition lists, images, emphasis) @@ -565,4 +568,4 @@ usage: fetch.py [-h] --url URL [--extract {text,html,markdown}] 4. Collapse whitespace, output clean UTF-8 5. Warn if extracted text < 500 chars (likely JS-heavy or bot-blocked) -**Completion criterion:** Outputs page content. Non-zero exit on HTTP failure. Stderr carries warnings for low-confidence extraction. +**Completion criterion:** Outputs page content (or the structured JSON object with `--format json`). Non-zero exit on HTTP failure. Stderr carries warnings for low-confidence extraction. diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..f61edad --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,42 @@ +[build-system] +requires = ["setuptools>=61.0"] +build-backend = "setuptools.build_meta" + +[project] +name = "searxng-use-cli" +version = "2.3.0" +description = "Privacy-respecting SearXNG search + web-fetch CLI toolkit for AI agents" +readme = "README.md" +license = { text = "MIT" } +requires-python = ">=3.8" +keywords = ["searxng", "search", "web-scraping", "ai-agent", "cli", "privacy"] +classifiers = [ + "Environment :: Console", + "Intended Audience :: Developers", + "License :: OSI Approved :: MIT License", + "Operating System :: OS Independent", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.8", + "Programming Language :: Python :: 3.9", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Topic :: Internet :: WWW/HTTP :: Indexing/Search", + "Topic :: Software Development :: Libraries :: Python Modules", +] +dependencies = [] + +[project.optional-dependencies] +fetch = ["requests", "beautifulsoup4"] +toml = ["tomli; python_version < '3.11'"] +all = ["requests", "beautifulsoup4", "tomli; python_version < '3.11'"] +test = ["pytest", "requests", "beautifulsoup4"] + +[project.scripts] +searxng-search = "scripts.search:main" +searxng-fetch = "scripts.fetch:main" + +[tool.setuptools] +packages = ["scripts"] + +[tool.setuptools.package-data] +scripts = ["*.py"] diff --git a/scripts/_config.py b/scripts/_config.py index 4f2be33..0a25ddf 100644 --- a/scripts/_config.py +++ b/scripts/_config.py @@ -7,7 +7,7 @@ Retry settings and shared HTTP utilities now live in ``common.py`` so that both ``search.py`` and ``fetch.py`` share one consistent implementation. """ -VERSION = "2.2.1" +VERSION = "2.3.0" SCHEMA_VERSION = "1.0" USER_AGENT = f"searxng-cli/{VERSION}" diff --git a/scripts/common.py b/scripts/common.py index f0a2143..8c01eab 100644 --- a/scripts/common.py +++ b/scripts/common.py @@ -18,13 +18,20 @@ as retryable and connection errors as transient, eliminating the previous inconsistency where ``search.py`` ignored 5xx. """ +import base64 +import datetime import hashlib import io +import json import logging +import os +import random import re import sys import threading import urllib.error +import urllib.parse +from pathlib import Path # Root logger for the searxng-cli package. All modules create child loggers # via ``logging.getLogger("searxng.<module>")`` so a single setup_logging() @@ -105,7 +112,6 @@ class _JsonFormatter(logging.Formatter): """ def format(self, record): - import json as _json entry = { "ts": _datetime_iso(record), "level": record.levelname, @@ -117,13 +123,12 @@ class _JsonFormatter(logging.Formatter): entry["request_id"] = rid if record.exc_info and record.exc_info[1]: entry["exception"] = type(record.exc_info[1]).__name__ - return _json.dumps(entry, ensure_ascii=False) + return json.dumps(entry, ensure_ascii=False) def _datetime_iso(record): """格式化日志时间戳为 ISO 8601 字符串。""" - import datetime as _dt - return _dt.datetime.fromtimestamp(record.created).isoformat(timespec="milliseconds") + return datetime.datetime.fromtimestamp(record.created).isoformat(timespec="milliseconds") def force_utf8_stdout() -> None: @@ -191,57 +196,17 @@ RETRYABLE_STATUS = frozenset({403, 429, 502, 503, 504}) # Browser-like UA strings for fallback when the searxng-cli UA is blocked. # -# v2.2.0：UA 池迁移至 _config.py 的 UA_POOL（SSOT），此处通过导入引用。 -# 若 _config.py 不可导入（如 common.py 被单独分发），回退到下方内置副本 -# _FALLBACK_UAS_BUILTIN，保证模块始终可用。两份列表需保持同步， -# _config.UA_POOL 为唯一权威来源。 +# v2.2.2：单一来源（SSOT）。UA 池唯一权威定义在 _config.UA_POOL， +# 此处直接导入——删除原有的内置副本 _FALLBACK_UAS_BUILTIN。 +# v2.2.0 曾保留一份手工同步副本，两份列表漂移会导致跨脚本 UA 行为不一致 +# （fetch.py 用 FALLBACK_UAS 轮换、search.py 的 --dry-run 报告引用同一池）， +# 且注释要求"保持同步"无任何机制保证。直接引用同一对象后，改一处即全局生效。 # -# v2.0.0 起覆盖 Chrome/Edge/Firefox × Windows/macOS/Linux；v2.2.0 升级到 -# 2026 年版本（Chrome 138-140 / Edge 138 / Firefox 140 / Safari 18）。 -# 顺序固定以便 get_ua_for_domain() 的 hash 选择可复现。 -_FALLBACK_UAS_BUILTIN = [ - # Chrome 140 — Windows / macOS / Linux - "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " - "(KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36", - "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 " - "(KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36", - "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 " - "(KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36", - # Chrome 139 — Windows / macOS / Linux - "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " - "(KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36", - "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 " - "(KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36", - "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 " - "(KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36", - # Chrome 138 — Windows / macOS / Linux - "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " - "(KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36", - "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 " - "(KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36", - "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 " - "(KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36", - # Edge 138 — Windows / macOS - "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " - "(KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36 Edg/138.0.0.0", - "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 " - "(KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36 Edg/138.0.0.0", - # Firefox 140 — Windows / macOS / Linux - "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:140.0) " - "Gecko/20100101 Firefox/140.0", - "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:140.0) " - "Gecko/20100101 Firefox/140.0", - "Mozilla/5.0 (X11; Linux x86_64; rv:140.0) Gecko/20100101 Firefox/140.0", - # Safari 18 — macOS（WebKit 指纹，应对 Chromium 针对性拦截） - "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 " - "(KHTML, like Gecko) Version/18.0 Safari/605.1.15", -] - -# 从 _config.py 导入权威 UA_POOL；导入失败时回退到内置副本，保证向后兼容。 -try: - from _config import UA_POOL as FALLBACK_UAS -except ImportError: - FALLBACK_UAS = _FALLBACK_UAS_BUILTIN +# 维护原则（见 _config.UA_POOL 注释）： +# 1. 版本号保持为当前年份的主流浏览器版本 +# 2. 顺序固定——get_ua_for_domain() 用 SHA-256 哈希选索引 +# 3. 至少覆盖 Chrome/Edge/Firefox × Windows/macOS/Linux +from _config import UA_POOL as FALLBACK_UAS def _ua_index_for_domain(domain: str, pool_size: int) -> int: @@ -251,7 +216,6 @@ def _ua_index_for_domain(domain: str, pool_size: int) -> int: 随机化（PYTHONHASHSEED），跨进程不可复现。SHA-256 保证同一域名 永远映射到同一索引，跨进程一致——这对调试和日志分析至关重要。 """ - import hashlib h = hashlib.sha256(domain.encode("utf-8")).digest() # 取前 8 字节作为无符号整数，避免负数和短字符串的分布不均 return int.from_bytes(h[:8], "big") % pool_size @@ -288,9 +252,8 @@ def get_ua_for_domain(url: str, user_agent: str = None) -> str: if user_agent: return user_agent - import urllib.parse as _up try: - domain = _up.urlparse(url).netloc.lower() + domain = urllib.parse.urlparse(url).netloc.lower() if not domain: return FALLBACK_UAS[0] except Exception: @@ -367,7 +330,6 @@ def build_browser_headers(user_agent: str, referer: str = None, # Sec-Ch-Ua 系列仅 Chrome/Edge 发送，Firefox 不发 if not is_firefox: # 从 UA 提取主版本号，构造 Sec-Ch-Ua - import re m = re.search(r"Chrome/(\d+)", user_agent) ver = m.group(1) if m else "131" not_a_brand = '"Not_A Brand";v="99"' if ver != "99" else '"Not/A)Brand";v="99"' @@ -445,7 +407,6 @@ def compute_backoff_delay(attempt: int, base: float = RETRY_BACKOFF_BASE, ``base * 2^attempt + jitter``，但不超过 ``cap``。 v2.0.0 新增封顶：原公式无上限，N=10 时达 1536s 会卡死进程。 """ - import random delay = base * (2 ** attempt) + random.uniform(0, 1) return min(delay, cap) @@ -459,8 +420,6 @@ def build_auth_headers(bearer_token: str = None, basic_auth: str = None) -> dict If both are provided, Bearer takes precedence (more common for APIs). Returns a dict to merge into request headers, or an empty dict. """ - import base64 - headers = {} if bearer_token: headers["Authorization"] = f"Bearer {bearer_token}" @@ -476,7 +435,6 @@ def _warn_file_perms(path: str) -> None: On Windows the Unix permission bits in ``st_mode`` do not reflect the actual ACL, so the check is skipped to avoid false alarms. """ - import os if os.name != "posix": return log = logging.getLogger("searxng.common") @@ -513,7 +471,6 @@ def resolve_auth_basic(cli_value: str = None, file_path: str = None, if file_path: try: - from pathlib import Path text = Path(file_path).read_text(encoding="utf-8") _warn_file_perms(file_path) for line in text.splitlines(): @@ -527,7 +484,6 @@ def resolve_auth_basic(cli_value: str = None, file_path: str = None, if config_value: return config_value - import os return os.environ.get(env_var) @@ -550,7 +506,6 @@ def resolve_auth_bearer(cli_value: str = None, file_path: str = None, if file_path: try: - from pathlib import Path text = Path(file_path).read_text(encoding="utf-8") _warn_file_perms(file_path) for line in text.splitlines(): @@ -564,7 +519,6 @@ def resolve_auth_bearer(cli_value: str = None, file_path: str = None, if config_value: return config_value - import os return os.environ.get(env_var) @@ -583,7 +537,6 @@ def apply_proxy(proxy_url: str) -> None: Pass an empty string to clear the proxy env vars (rarely needed; the default unset state already means "no proxy"). """ - import os if not proxy_url: return os.environ["HTTP_PROXY"] = proxy_url @@ -597,8 +550,6 @@ def detect_charset(raw: bytes, content_type: str) -> str: Falls back to UTF-8 (with replacement) if nothing reliable is found. """ - import re - # 1. HTTP header if "charset=" in content_type: charset = content_type.split("charset=")[-1].split(";")[0].strip() @@ -704,7 +655,7 @@ RECOVERY_HINTS = { } -def _classify_by_type_and_status(exc, _json): +def _classify_by_type_and_status(exc) -> str: """根据异常类型和 HTTP 状态码分类，返回错误码或 None。 检查顺序：HTTP 状态码 → 文件错误 → 连接错误 → 解析错误。 @@ -743,7 +694,7 @@ def _classify_by_type_and_status(exc, _json): return E_NETWORK # 解析错误 - if isinstance(exc, (ValueError, _json.JSONDecodeError)): + if isinstance(exc, (ValueError, json.JSONDecodeError)): return E_PARSE return None @@ -764,8 +715,6 @@ def classify_error(exc: BaseException) -> str: 兼容旧路径——search_multi 把 last_error 拼进消息） 7. 其他 → E_INTERNAL """ - import json as _json - # 优先检查异常链 __cause__：raise X from Y 时，Y 指向真实底层异常。 # 旧逻辑只能从外层 RuntimeError 的消息字符串推断（脆弱，依赖 "403"/"auth" # 等关键字匹配），新逻辑直接从 __cause__ 的 .code/.status_code 或异常 @@ -773,12 +722,12 @@ def classify_error(exc: BaseException) -> str: # raise-from 模式，深层链罕见且递归有循环风险。 cause = getattr(exc, "__cause__", None) if cause is not None and cause is not exc: - code = _classify_by_type_and_status(cause, _json) + code = _classify_by_type_and_status(cause) if code is not None: return code # 检查 exc 本身的类型和状态码 - code = _classify_by_type_and_status(exc, _json) + code = _classify_by_type_and_status(exc) if code is not None: return code @@ -795,8 +744,7 @@ def classify_error(exc: BaseException) -> str: 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) + status_match = re.search(r'http(?: error)? (\d{3})', msg) if status_match: status = int(status_match.group(1)) if status == 429: @@ -855,13 +803,12 @@ def emit_progress(event: str, **kwargs) -> None: """ if not _progress_enabled: return - import json as _json payload = {"event": event} rid = getattr(_LOG, "_request_id", None) if rid: payload["request_id"] = rid payload.update(kwargs) - print(_json.dumps(payload, ensure_ascii=False), file=sys.stderr, flush=True) + print(json.dumps(payload, ensure_ascii=False), file=sys.stderr, flush=True) # ----- Wayback Machine 兜底（v2.1.0 共享逻辑）----- @@ -945,8 +892,7 @@ def is_hard_blocked_domain(url: str) -> bool: return False # 提取域名 try: - from urllib.parse import urlparse - host = urlparse(url).hostname or "" + host = urllib.parse.urlparse(url).hostname or "" except Exception: host = "" if not host: @@ -1051,9 +997,8 @@ def is_similar(result_a: dict, result_b: dict, threshold: float = 0.85) -> bool: title_b = _normalize_title(result_b.get("title", "")) # 标题太短时 SimHash 不稳定，改用 Jaccard if len(title_a) < 5 or len(title_b) < 5: - import urllib.parse as _up - domain_a = _up.urlparse(result_a.get("url", "")).netloc.lower() - domain_b = _up.urlparse(result_b.get("url", "")).netloc.lower() + domain_a = urllib.parse.urlparse(result_a.get("url", "")).netloc.lower() + domain_b = urllib.parse.urlparse(result_b.get("url", "")).netloc.lower() set_a = set(title_a.split()) | {domain_a} set_b = set(title_b.split()) | {domain_b} return _jaccard_similarity(set_a, set_b) >= threshold diff --git a/scripts/fetch.py b/scripts/fetch.py index 57688c2..290feeb 100644 --- a/scripts/fetch.py +++ b/scripts/fetch.py @@ -11,6 +11,7 @@ for improved extraction quality (optional, falls back to stdlib). import argparse import gzip import io +import json import logging import random import re @@ -53,6 +54,7 @@ from common import ( build_auth_headers, build_browser_headers, build_wayback_url, + classify_error, compute_backoff_delay, detect_charset, force_utf8_stdout, @@ -1134,6 +1136,55 @@ def fetch_url(url: str, timeout=15, user_agent: str = None, # ----- Main ----- +def _emit_fetch_result(args, output: str, url: str, final_url: str, + content_type: str, truncated: bool, + user_agent: str = None, + error: str = None, error_code: str = None, + status_code: int = None) -> None: + """输出抓取结果到 stdout / --output 文件。 + + v2.3.0: ``--format json`` 提供结构化 JSON 契约，AI Agent 可程序化 + 解析（成功与失败统一为 {status, url, ...}）。``--format text``（默认） + 保持 v2.2.x 行为：成功输出正文，失败输出空 + stderr 日志。 + + 成功 shape:: + + {"status": "ok", "url", "final_url", "content_type", + "extract", "truncated", "text_length", "user_agent"} + + 失败 shape:: + + {"status": "error", "url", "error", "error_code", "status_code"} + """ + if args.format == "json": + if error: + payload = {"status": "error", "url": url, "error": error} + if error_code: + payload["error_code"] = error_code + if status_code is not None: + payload["status_code"] = status_code + else: + payload = { + "status": "ok", + "url": url, + "final_url": final_url, + "content_type": content_type, + "extract": args.extract, + "truncated": truncated, + "text_length": len(output), + "user_agent": user_agent, + } + text = json.dumps(payload, indent=2, ensure_ascii=False) + else: + text = output + if args.output: + with open(args.output, "w", encoding="utf-8") as f: + f.write(text) + logger.info(f"Saved {len(text)} chars to {args.output}") + else: + print(text) + + def main(): parser = argparse.ArgumentParser( description="Fetch a web page and extract readable content", @@ -1146,11 +1197,18 @@ Examples: %(prog)s -u https://example.com -e markdown markdown conversion %(prog)s -u https://example.com -o page.txt save to file %(prog)s -u https://example.cn -e text --encoding gbk force charset + %(prog)s -u https://example.com --format json structured JSON output """, ) parser.add_argument("--url", "-u", required=True, help="URL to fetch") parser.add_argument("--extract", "-e", choices=["text", "html", "markdown"], default="text", help="Extraction mode (default: text)") + parser.add_argument("--format", "-f", choices=["text", "json"], default="text", + help="Output format (default: text). 'json' emits a structured " + "JSON object {status, url, final_url, content_type, extract, " + "truncated, text_length, user_agent} on success, or " + "{status: error, error, error_code, status_code, url} on " + "failure — machine-readable for agents. v2.3.0.") parser.add_argument("--timeout", "-t", type=int, default=15, help="Request timeout in seconds (default: 15)") parser.add_argument("--retries", type=int, default=3, @@ -1229,6 +1287,18 @@ Examples: logger.info(f"Hard-blocked domain detected — Wayback fallback " f"will be prioritized if main fetch fails") + # v2.3.0: 状态收集变量。所有失败路径设置 fatal_* 后落到统一输出 + # （_emit_fetch_result），json 模式输出结构化错误到 stdout，text 模式 + # 保持 v2.2.x 行为（stdout 空 + stderr 日志 + exit 1）。 + content = None + final_url = args.url + content_type = "" + truncated = False + user_agent = None + fatal_error = None + fatal_error_code = None + fatal_status_code = None + try: result = fetch_url( args.url, timeout=args.timeout, user_agent=args.user_agent, @@ -1237,17 +1307,17 @@ Examples: allow_redirects=not args.no_redirect, referer=args.referer, ) - content, content_type, final_url = ( - result.content, result.content_type, result.final_url, - ) + content = result.content + content_type = result.content_type or "" + final_url = result.final_url + truncated = result.truncated + user_agent = result.user_agent # 文档解析失败（PDF/DOCX/XLSX 等）时 fetch_url 不抛异常， # 而是返回带 error_code 的 FetchResult——必须显式检查， # 否则失败会被静默吞掉（空输出 + exit 0）。 if result.error_code: - logger.error( - f"Error: {result.error_message or result.error_code} " - f"(error_code={result.error_code}, url={args.url})") - sys.exit(1) + fatal_error = result.error_message or result.error_code + fatal_error_code = result.error_code except Exception as e: # 诊断信息增强：从 __cause__ 链中提取 HTTP 状态码、原始异常类型， # 让 AI Agent 能程序化判断失败原因（404 vs 403 vs DNS 失败等）， @@ -1275,7 +1345,10 @@ Examples: # v2.1.0: Wayback Machine 兜底 # 触发条件：兜底启用 + (错误可恢复 OR 命中被墙站点) - error_msg = str(e) if str(e) else e.__class__.__name__ + fatal_error = str(e) if str(e) else e.__class__.__name__ + fatal_error_code = classify_error(e) + fatal_status_code = status_code + error_msg = fatal_error if fallback_enabled and (should_try_wayback(error_msg) or hard_blocked): wayback_url = build_wayback_url(args.url) wb_timeout = min(args.timeout, 10) # Wayback 独立超时，不阻塞 @@ -1289,17 +1362,31 @@ Examples: max_size=args.max_size, allow_redirects=True, ) - content, content_type, final_url = ( - wb_result.content, wb_result.content_type, - wb_result.final_url, - ) + content = wb_result.content + content_type = wb_result.content_type or "" + final_url = wb_result.final_url + truncated = wb_result.truncated + user_agent = wb_result.user_agent + if wb_result.error_code: + fatal_error = (wb_result.error_message or wb_result.error_code) + fatal_error_code = wb_result.error_code + fatal_status_code = None + else: + fatal_error = None + fatal_error_code = None + fatal_status_code = None logger.info(f"[FALLBACK] Wayback recovery successful " f"({len(content)} chars)") except Exception as wb_e: logger.error(f"[FALLBACK] Wayback also failed: {wb_e}") - sys.exit(1) - else: - sys.exit(1) + fatal_error = f"{fatal_error} ; Wayback also failed: {wb_e}" + + if fatal_error: + _emit_fetch_result(args, "", args.url, final_url, content_type, + truncated, user_agent, error=fatal_error, + error_code=fatal_error_code, + status_code=fatal_status_code) + sys.exit(1) if final_url != args.url: logger.info(f"Redirected to: {final_url}") @@ -1320,12 +1407,8 @@ Examples: logger.warning(f"Warning: extracted text is very short ({len(output.strip())} chars). " "The page may be JS-heavy or use anti-bot protection.") - if args.output: - with open(args.output, "w", encoding="utf-8") as f: - f.write(output) - logger.info(f"Saved {len(output)} chars to {args.output}") - else: - print(output) + _emit_fetch_result(args, output, args.url, final_url, content_type, + truncated, user_agent) if __name__ == "__main__": diff --git a/scripts/search.py b/scripts/search.py index 9af0188..ab27b16 100644 --- a/scripts/search.py +++ b/scripts/search.py @@ -9,6 +9,8 @@ Instance URLs are REQUIRED (see --instance / SEARXNG_INSTANCE / config file). """ import argparse +import csv +import io import json import logging import os @@ -37,6 +39,7 @@ from common import ( build_wayback_url, classify_error, compute_backoff_delay, + detect_charset, emit_progress, force_utf8_stdout, is_hard_blocked_domain, @@ -549,8 +552,13 @@ def search_json(instance: str, params: dict, method: str = "GET", def search_html(instance: str, params: dict, timeout: int = 15, - auth_headers: dict = None) -> dict: - """Execute search via HTML scraping fallback.""" + 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``），优先级最高。 + """ 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}" @@ -559,26 +567,41 @@ def search_html(instance: str, params: dict, timeout: int = 15, req = urllib.request.Request(url, headers=headers) try: with urllib.request.urlopen(req, timeout=timeout) as resp: - html = resp.read().decode("utf-8") + raw = resp.read() + content_type = resp.headers.get("Content-Type", "") + 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", "")) except Exception as e: raise RuntimeError(f"HTML search failed for {instance}: {e}") def search_single(instance: str, params: dict, method: str = "GET", - timeout: int = 15, auth_headers: dict = None) -> dict: + 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) + 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) -> dict: + 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 @@ -589,6 +612,9 @@ def search_multi(instance_urls: list, params: dict, method: str = "GET", 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 @@ -604,7 +630,8 @@ def search_multi(instance_urls: list, params: dict, method: str = "GET", try: def _do(): return search_single(instance, params, method=method, - timeout=timeout, auth_headers=auth_headers) + 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), @@ -627,7 +654,8 @@ def search_multi(instance_urls: list, params: dict, method: str = "GET", start = time.time() def _do(): return search_single(instance, params, method=method, - timeout=timeout, auth_headers=auth_headers) + timeout=timeout, auth_headers=auth_headers, + encoding=encoding) try: result = _retry_with_backoff(_do, max_retries=retry_per) emit_progress("instance_ok", url=instance, @@ -864,6 +892,8 @@ def fetch_page(url: str, timeout: int = 10, auth_headers: dict = None, * ``fallback_used`` (str|None): 兜底方式（"wayback"），仅当走兜底时有值 """ # 主抓取 + # v2.3.0: 计时整条链路（主抓取 + Wayback 兜底），填充 latency 字段。 + _start_time = time.monotonic() result = None error_msg = None error_code = None @@ -946,6 +976,9 @@ def fetch_page(url: str, timeout: int = 10, auth_headers: dict = None, "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 也无能为力或兜底被禁用） @@ -957,6 +990,8 @@ def fetch_page(url: str, timeout: int = 10, auth_headers: dict = None, "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 @@ -973,6 +1008,8 @@ def fetch_page(url: str, timeout: int = 10, auth_headers: dict = None, "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), } @@ -1105,6 +1142,19 @@ _TITLE_ANTI_BOT_SIGNATURES = { _TITLE_RE = re.compile(r"<title[^>]*>(.*?)", re.IGNORECASE | re.DOTALL) +def _extract_title(content: str) -> str: + """从 HTML 内容提取 文本（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。 @@ -1173,6 +1223,11 @@ class AdaptiveThrottle: 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: @@ -1228,6 +1283,36 @@ class AdaptiveThrottle: 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: @@ -1292,34 +1377,49 @@ def fetch_top_results(results: dict, count: int, timeout: int = 10, """Fetch one URL with adaptive throttling to avoid rate limiting.""" # 全局暂停检查（429 触发） throttle.wait_if_paused() - # 自适应延迟 - 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() - 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 + # 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() + 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} @@ -1333,7 +1433,8 @@ def fetch_top_results(results: dict, count: int, timeout: int = 10, "error_code": classify_error(e), "text": "", "text_length": 0, "truncated": False, "anti_bot_detected": False, "waf_type": None, - "fallback_used": None}) + "fallback_used": None, + "title": None, "latency": None}) logger.error(f" [ERR] {u[:55]} (thread error: {e})") # Reorder to match original result order @@ -1365,8 +1466,7 @@ def _emit_fetch_report(fetched: list, throttle: "AdaptiveThrottle", def _emit_fetch_report_text(fetched: list, throttle: "AdaptiveThrottle") -> None: """文本表格 + JSON 摘要行（v2.0.0 原始行为，向后兼容）。""" - import sys as _sys - out = _sys.stderr + out = sys.stderr lines = [] lines.append("\n" + "=" * 72) lines.append("FETCH REPORT (v2.0.0)") @@ -1403,13 +1503,12 @@ def _emit_fetch_report_text(fetched: list, throttle: "AdaptiveThrottle") -> None f"consec_ok={s['consecutive_successes']}") # JSON 摘要（一行，便于 Agent 解析） - import json as _json summary = { "total": total, "ok": ok, "error": err, "anti_bot_blocked": anti_bot, "wayback_recovered": wayback, "throttle": s, } - lines.append("JSON: " + _json.dumps(summary, ensure_ascii=False)) + lines.append("JSON: " + json.dumps(summary, ensure_ascii=False)) lines.append("=" * 72 + "\n") print("\n".join(lines), file=out) @@ -1418,11 +1517,9 @@ 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 等字段。fetch_page 未采集 - 的字段（title / latency / fetched_at）为 None，便于 Agent 统一解析。 + error / error_code / latency / fetched_at 等字段。v2.3.0 起 + ``title`` 与 ``latency`` 由 fetch_page 采集填充（此前恒为 None）。 """ - import json as _json - import sys as _sys from datetime import datetime, timezone fetched_at = datetime.now(timezone.utc).isoformat() @@ -1434,12 +1531,12 @@ def _emit_fetch_report_json(fetched: list, throttle: "AdaptiveThrottle") -> None "url": f.get("url", ""), "final_url": f.get("final_url"), "status": f.get("status", ""), - "title": f.get("title"), # fetch_page 未提取，None + "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"), # fetch_page 未计时，None + "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"), @@ -1464,7 +1561,7 @@ def _emit_fetch_report_json(fetched: list, throttle: "AdaptiveThrottle") -> None "throttle": throttle.stats(), "items": items, } - print(_json.dumps(report, ensure_ascii=False), file=_sys.stderr) + print(json.dumps(report, ensure_ascii=False), file=sys.stderr) # ----- Output formatting ----- @@ -1694,10 +1791,8 @@ def _format_results(results: dict, args) -> str: if args.format == "urls": return format_urls(results) if args.format == "csv": - import csv as csv_mod - import io out = io.StringIO() - writer = csv_mod.writer(out, lineterminator="\n") + writer = csv.writer(out, lineterminator="\n") writer.writerow(["title", "url", "engine", "score", "published_date", "content"]) for r in results.get("results", []): @@ -1803,6 +1898,13 @@ def _run_single_query(query: str, args, instance_urls: list, 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: @@ -1821,11 +1923,13 @@ def _run_single_query(query: str, args, instance_urls: list, 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") @@ -1855,6 +1959,7 @@ def _run_single_query(query: str, args, instance_urls: list, retry_per=args.retry, auth_headers=auth_headers, parallel=not args.serial, + encoding=getattr(args, "encoding", None), ) except Exception as e: page_errors += 1 @@ -1862,6 +1967,7 @@ def _run_single_query(query: str, args, instance_urls: list, 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) @@ -1886,7 +1992,7 @@ def _run_single_query(query: str, args, instance_urls: list, # v2.1.1：检测实例侧引擎挂起/限流（仅在实时查询后提示，cache hit 时 # unresponsive_engines 信息可能已过期） - if cached is None: + if performed_live_query: _warn_unresponsive_engines(results, query, result_count=len(results.get("results", []))) @@ -1975,14 +2081,54 @@ def _get_output_schema(): 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 为单独子树。 """ - 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:[]}.', + # 单查询结果条目（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"]}, + "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": { @@ -1999,18 +2145,7 @@ def _get_output_schema(): "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"], - }, + "items": result_item, }, "unresponsive_engines": { "type": "array", @@ -2022,19 +2157,16 @@ def _get_output_schema(): "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": { - "type": "object", - "properties": { - "url": {"type": "string"}, - "text": {"type": "string"}, - "text_length": {"type": "integer"}, - "error": {"type": "string"}, - }, - }, + "items": fetched_item, }, "fetched_source": { "type": "string", @@ -2042,9 +2174,100 @@ def _get_output_schema(): "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, @@ -2087,13 +2310,30 @@ def _emit_error(message: str, args, query: str = None, exit_code: int = 1, 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: - text = Path(path).read_text(encoding="utf-8") + 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() @@ -2260,7 +2500,7 @@ def expand_research_queries(topic: str, custom_angles: list = None) -> list: # ----- Mode handlers (v2.2.0: 从 main() 提取，降低单函数复杂度) ----- # main() 只负责参数解析和分发，四条执行路径各自独立函数，便于维护和测试。 -# 纯提取重构，行为与 v2.1.1 完全一致，539 测试兜底验证。 +# 纯提取重构，行为与 v2.1.1 完全一致，544 测试兜底验证。 def _handle_verify(args, instance_urls: list, auth_headers: dict) -> None: @@ -2340,10 +2580,8 @@ def _handle_research(args, instance_urls: list, auth_headers: dict, "merged_results": merged, }, indent=2, ensure_ascii=False) elif args.format == "csv": - import csv as csv_mod - import io out = io.StringIO() - writer = csv_mod.writer(out, lineterminator="\n") + writer = csv.writer(out, lineterminator="\n") writer.writerow(["angle", "query", "title", "url", "engine", "score", "published_date", "content"]) for br in batch: @@ -2412,11 +2650,33 @@ def _handle_research(args, instance_urls: list, auth_headers: dict, 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) @@ -2426,34 +2686,89 @@ def _handle_batch(args, instance_urls: list, auth_headers: dict, _emit_error(f"no queries found in '{args.queries_file}'", args, error_code=E_INPUT) - 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}) + 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": - import csv as csv_mod - import io out = io.StringIO() - writer = csv_mod.writer(out, lineterminator="\n") + writer = csv.writer(out, lineterminator="\n") writer.writerow(["query", "title", "url", "engine", "score", "published_date", "content"]) for br in batch: @@ -2627,6 +2942,12 @@ Use --config FILE to load a non-default config file (overrides the auto-discover 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, @@ -2769,6 +3090,13 @@ Use --config FILE to load a non-default config file (overrides the auto-discover "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) " @@ -2803,8 +3131,7 @@ Use --config FILE to load a non-default config file (overrides the auto-discover # v2.2.0：生成 request_id 并重新配置 logging（带 log_format + request_id） # request_id 贯穿所有日志、进度事件和错误输出，便于 batch 模式追溯。 - import os as _os - request_id = _os.urandom(4).hex() + 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) diff --git a/tests/test_v230_features.py b/tests/test_v230_features.py new file mode 100644 index 0000000..9d8e1c3 --- /dev/null +++ b/tests/test_v230_features.py @@ -0,0 +1,366 @@ +"""Tests for v2.3.0 changes. + +Covers: + * multi-page (--pages N) live-query tracking — unresponsive-engine + warnings fire when ANY page did a live query (regression for the + ``cached``-variable pollution fix) + * search_html charset auto-detection + --encoding override + * UA pool single-source (common.FALLBACK_UAS is _config.UA_POOL) + * AdaptiveThrottle acquire_slot/release_slot real concurrency gating + (incl. persistent throttling after backoff reduces concurrency) + * fetch_page title/latency fields (consumed by --fetch-report json) + * fetch.py _emit_fetch_result JSON output shapes (success + error) + * _read_queries_file GBK fallback + * --parallel-queries concurrent batch (output order preserved) +""" +import json +import logging +import sys +import time +import urllib.error +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import pytest + +import common as common_mod +import _config +from fetch import FetchResult, _emit_fetch_result +import search as search_mod +from search import ( + AdaptiveThrottle, + _read_queries_file, + _run_single_query, + _build_params, + fetch_page, + fetch_top_results, + search_html, +) +import cache as cache_module + + +def _make_args(**overrides): + """args object matching the argparse.Namespace shape main() produces.""" + base = dict( + query="test", format="json", method="GET", timeout=15, retry=0, + serial=False, no_dedup=False, sort_by="none", max_results=None, + include_domain=None, exclude_domain=None, fetch=0, fetch_timeout=10, + fetch_retries=3, max_size=None, cache_ttl=0, snippet_len=0, + categories=None, language=None, pageno=1, pages=1, time_range="year", + safesearch=0, engines="google,bing", + referer=None, no_fallback=False, fetch_report=False, + request_delay=0.3, throttle_failure_threshold=3, + throttle_pause_seconds=30, throttle_max_delay=10, + similarity_dedup=False, similarity_threshold=0.85, + encoding=None, output=None, parallel_queries=0, + ) + base.update(overrides) + return SimpleNamespace(**base) + + +# ===== Multi-page live-query tracking (cached-var pollution fix) ===== + +def test_multi_page_live_query_warns_unresponsive(isolated_cache, caplog): + """pages=2: page1 live + page2 cached hit -> warning still fires. + + Regression: the old code used the last loop iteration's ``cached`` + variable, so a cache hit on page 2 suppressed the unresponsive-engine + warning even though page 1 hit the network live. + """ + args = _make_args(pages=2, cache_ttl=30) + # Pre-fill page 2 cache so page 2 is a cache hit. + page2_params = dict(_build_params("test", args)) + page2_params["pageno"] = "2" + cache_module.put(page2_params, {"results": []}, 300) + + live = {"results": [{"url": "https://a.com/1", "title": "A"}], + "unresponsive_engines": [["brave", "Suspended: too many requests"]]} + + def fake_multi(urls, params, **kw): + assert params.get("pageno") is None # only page 1 hits the network + return live + + with patch.object(search_mod, "search_multi", side_effect=fake_multi): + with caplog.at_level(logging.WARNING, logger="searxng.search"): + results, err, _ = _run_single_query( + "test", args, ["https://x.example.com"], {}, 300) + assert err is None + assert any("unresponsive" in r.getMessage() + for r in caplog.records), "warning should fire after a live query" + assert results["pages_fetched"] == 2 + + +def test_multi_page_all_cache_hit_no_warning(isolated_cache, caplog): + """pages=2: both pages cached -> no network, no warning.""" + args = _make_args(pages=2, cache_ttl=30) + cached = {"results": [{"url": "https://c.com/1", "title": "C"}]} + for page_no in (1, 2): + p = dict(_build_params("test", args)) + if page_no != 1: + p["pageno"] = str(page_no) + cache_module.put(p, cached, 300) + + with patch.object(search_mod, "search_multi", + side_effect=AssertionError("must not hit network")): + with caplog.at_level(logging.WARNING, logger="searxng.search"): + results, err, _ = _run_single_query( + "test", args, ["https://x.example.com"], {}, 300) + assert err is None + assert not any("unresponsive" in r.getMessage() + for r in caplog.records) + + +# ===== search_html charset detection ===== + +def test_search_html_detects_gbk_charset(): + """GBK HTML detected via <meta charset> (header has no charset).""" + html = ('<html><head><meta charset="gbk"></head><body>' + '<article class="result"><h3><a href="https://x.com">中文标题</a>' + '</h3></article></body></html>') + raw = html.encode("gbk") + resp = MagicMock() + resp.read.return_value = raw + resp.headers = {"Content-Type": "text/html"} # no charset in header + resp.__enter__.return_value = resp + with patch("urllib.request.urlopen", return_value=resp): + r = search_html("https://s.example.com", {"q": "test"}) + assert r["results"][0]["title"] == "中文标题" + + +def test_search_html_encoding_override(): + """--encoding gbk forces the charset even with wrong header.""" + html = '<article class="result"><h3><a href="https://x.com">标题</a></h3></article>' + raw = html.encode("gbk") + resp = MagicMock() + resp.read.return_value = raw + resp.headers = {"Content-Type": "text/html; charset=utf-8"} # lies + resp.__enter__.return_value = resp + with patch("urllib.request.urlopen", return_value=resp): + r = search_html("https://s.example.com", {"q": "test"}, encoding="gbk") + assert r["results"][0]["title"] == "标题" + + +# ===== UA pool single source ===== + +def test_ua_pool_is_single_source(): + """common.FALLBACK_UAS must be the _config.UA_POOL object (no duplicate).""" + assert common_mod.FALLBACK_UAS is _config.UA_POOL + assert len(_config.UA_POOL) >= 12 + + +# ===== AdaptiveThrottle slot gating ===== + +def test_throttle_acquire_release_slot_basic(): + """acquire_slot respects the concurrency cap; release frees it.""" + t = AdaptiveThrottle(0.0, 1) + assert t.acquire_slot() is True + # Second acquisition must fail while one slot is in flight. + assert t.acquire_slot(timeout=0.05) is False + t.release_slot() + assert t.acquire_slot(timeout=0.05) is True + t.release_slot() + + +def test_throttle_slot_gates_on_reduced_concurrency(): + """After backoff halves concurrency, in-flight requests gate new ones. + + Regression: ThreadPoolExecutor size is fixed at creation; the semaphore + + in-flight check must persist the reduced concurrency. With + initial=2, two slots in flight, then backoff -> concurrency=1: a new + acquisition must be rejected even though a semaphore slot is free. + """ + t = AdaptiveThrottle(0.0, 2) + assert t.acquire_slot() is True + assert t.acquire_slot() is True + # 3 consecutive failures -> concurrency 2 -> 1 + for _ in range(3): + t.report_failure() + assert t.concurrency == 1 + # Release one in-flight slot: semaphore has room, but in_flight (1) + # now exceeds the reduced concurrency target (1) -> reject. + t.release_slot() + assert t.acquire_slot(timeout=0.05) is False + t.release_slot() # cleanup the remaining acquired slot + + +def test_fetch_top_results_reports_throttled_skips(): + """fetch_top_results marks requests gated by the concurrency cap. + + Semantics covered deterministically by the acquire_slot unit tests; + this integration test asserts the end-to-end shape: a URL skipped by + the gate appears with status=error + error_code=E_RATE_LIMIT. We + simulate the gate directly by stubbing acquire_slot. + """ + throttle = AdaptiveThrottle(0.0, 5) + real_acquire = throttle.acquire_slot + + def _stub_acquire(timeout=0.05): + # First call wins the slot; every subsequent call is gated. + return False + + results = {"results": [{"url": "https://a.com"}, + {"url": "https://b.com"}]} + + def _fake_fetch(url, **kwargs): + return {"url": url, "status": "ok", "text": "x", "text_length": 1, + "truncated": False} + + with patch.object(search_mod, "fetch_page", side_effect=_fake_fetch): + with patch.object(throttle, "acquire_slot", side_effect=_stub_acquire): + out = fetch_top_results(results, 2, request_delay=0.0, + throttle=throttle) + assert len(out) == 2 + for f in out: + assert f["status"] == "error" + assert f["error_code"] == "E_RATE_LIMIT" + assert "Throttled" in f["error"] + + +# ===== fetch_page title/latency ===== + +def _mock_fetch_result(content, content_type="text/html", + final_url="https://example.com", ua="TestUA"): + return FetchResult(content=content, content_type=content_type, + final_url=final_url, truncated=False, user_agent=ua) + + +def test_fetch_page_collects_title_and_latency(): + html = "<html><head><title>Page Title" \ + "
Hello world
" + with patch.object(search_mod, "fetch_url", + return_value=_mock_fetch_result(html)): + r = fetch_page("https://example.com", fallback_enabled=False) + assert r["status"] == "ok" + assert r["title"] == "Page Title" + assert r["latency"] is not None and r["latency"] >= 0 + + +def test_fetch_page_title_none_for_non_html(): + with patch.object(search_mod, "fetch_url", + return_value=_mock_fetch_result( + "plain", content_type="text/plain")): + r = fetch_page("https://example.com/x.txt", fallback_enabled=False) + assert r["status"] == "ok" + assert r["title"] is None + assert r["latency"] is not None + + +# ===== fetch.py --format json output ===== + +def _json_args(**overrides): + base = dict(url="https://x.example.com", extract="text", format="json", + output=None) + base.update(overrides) + return SimpleNamespace(**base) + + +def test_emit_fetch_result_json_success(capsys): + _emit_fetch_result(_json_args(), "hello world", "https://x.example.com", + "https://x.example.com/", "text/html", False, + user_agent="TestUA") + out, _ = capsys.readouterr() + data = json.loads(out) + assert data["status"] == "ok" + assert data["url"] == "https://x.example.com" + assert data["final_url"] == "https://x.example.com/" + assert data["extract"] == "text" + assert data["text_length"] == 11 + assert data["truncated"] is False + assert data["user_agent"] == "TestUA" + assert "content_type" in data + + +def test_emit_fetch_result_json_error(capsys): + _emit_fetch_result(_json_args(), "", "https://x.example.com", + "https://x.example.com", "", False, + error="HTTP 404", error_code="E_NETWORK", + status_code=404) + out, _ = capsys.readouterr() + data = json.loads(out) + assert data["status"] == "error" + assert data["error_code"] == "E_NETWORK" + assert data["status_code"] == 404 + assert data["url"] == "https://x.example.com" + + +def test_emit_fetch_result_text_mode_unchanged(capsys): + """--format text keeps raw content on stdout (v2.2.x behavior).""" + _emit_fetch_result(_json_args(format="text"), "raw content", + "https://x.example.com", "https://x.example.com", + "text/html", False) + out, _ = capsys.readouterr() + assert out == "raw content\n" + + +# ===== _read_queries_file encoding fallback ===== + +def test_read_queries_file_gbk_fallback(tmp_path): + """GBK-encoded queries file decodes via the utf-8 -> gbk fallback.""" + p = tmp_path / "queries.txt" + p.write_bytes("中文查询一\n中文查询二\n".encode("gbk")) + qs = _read_queries_file(str(p)) + assert qs == ["中文查询一", "中文查询二"] + + +def test_read_queries_file_utf8_preferred(tmp_path): + p = tmp_path / "queries.txt" + p.write_text("q1\n# comment\nq2\n", encoding="utf-8") + assert _read_queries_file(str(p)) == ["q1", "q2"] + + +# ===== --parallel-queries concurrent batch ===== + +def _batch_argv(queries_file_path, parallel): + return ["search.py", "-i", "https://x.example.com", + "--queries-file", str(queries_file_path), + "--parallel-queries", str(parallel), + "--format", "json", "--retry", "0"] + + +def test_parallel_batch_preserves_order(tmp_path, capsys): + """Concurrent batch output keeps the file order (deterministic).""" + qf = tmp_path / "queries.txt" + qf.write_text("q1\nq2\nq3\nq4\n", encoding="utf-8") + + def fake_multi(urls, params, **kw): + q = params["q"] + time.sleep(0.01 * int(q[-1])) # q4 slowest, q1 fastest + return {"results": [{"url": f"https://{q}.com", "title": q}]} + + with patch.object(search_mod, "search_multi", side_effect=fake_multi): + with pytest.raises(SystemExit) as exc_info: + with patch.object(sys, "argv", _batch_argv(qf, 2)): + with patch.object(search_mod, "setup_logging"): + search_mod.main() + assert exc_info.value.code == 0 + out, _ = capsys.readouterr() + data = json.loads(out) + queries = [e["query"] for e in data["queries"]] + assert queries == ["q1", "q2", "q3", "q4"] # order preserved + assert all(e["status"] == "ok" for e in data["queries"]) + + +def test_parallel_batch_records_errors(tmp_path, capsys): + """Concurrent batch keeps per-query error entries without aborting.""" + qf = tmp_path / "queries.txt" + qf.write_text("ok1\nbad1\nok2\n", encoding="utf-8") + err = urllib.error.URLError("connection refused") + + def fake_multi(urls, params, **kw): + if params["q"].startswith("bad"): + raise err + return {"results": [{"url": f"https://{params['q']}.com"}]} + + with patch.object(search_mod, "search_multi", side_effect=fake_multi): + with pytest.raises(SystemExit) as exc_info: + with patch.object(sys, "argv", _batch_argv(qf, 3)): + with patch.object(search_mod, "setup_logging"): + search_mod.main() + assert exc_info.value.code == 0 # partial success + out, _ = capsys.readouterr() + data = json.loads(out) + statuses = {e["query"]: e for e in data["queries"]} + assert statuses["bad1"]["status"] == "error" + assert statuses["bad1"]["error_code"] == "E_NETWORK" + assert statuses["ok1"]["status"] == "ok" + assert statuses["ok2"]["status"] == "ok"