From e94cbe0783b7c1dd4e71ec7d5d562b4de8895357 Mon Sep 17 00:00:00 2001 From: thzxx <1440196015@qq.com> Date: Sun, 2 Aug 2026 08:27:45 +0800 Subject: [PATCH] =?UTF-8?q?feat(v2.1.0):=20=E7=A0=94=E7=A9=B6=E6=A8=A1?= =?UTF-8?q?=E5=BC=8F=20+=20fetch.py=20Wayback=20=E5=85=9C=E5=BA=95=20+=20?= =?UTF-8?q?=E8=A2=AB=E5=A2=99=E7=AB=99=E7=82=B9=E6=99=BA=E8=83=BD=E5=9B=9E?= =?UTF-8?q?=E9=80=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A. fetch.py 补齐 Wayback 兜底 (修复重大 gap) - v2.0.1 gap: fetch.py 独立调用 403 时无 Wayback 兜底 (仅 search.py --fetch 有) - AI Agent 用 fetch.py -u URL 直接抓取被墙站点时, 403 后无任何回退 - 修复: fetch.py main() 增加 Wayback 兜底逻辑 + --no-fallback flag - 共享逻辑抽取到 common.py: should_try_wayback() + build_wayback_url() B. --research 研究模式 - 给定主题自动扩展 5 个多角度查询: overview/profile/background/works/review - 确定性规则 (不依赖 AI 判断), 跨进程可复现 - 输出含 research_topic + research_queries 元数据, AI Agent 可按角度结构化汇编 - 与 --query/--queries-file 互斥, 支持所有输出格式 (json/brief/urls/csv) - 三态退出码: 0=有结果, 2=全部空, 1=全部错误 C. 被墙站点智能回退 - common.py 增加 HARD_BLOCKED_DOMAINS: 百度百科/知乎/微博/微信公众号/豆瓣等 - is_hard_blocked_domain() 精确匹配 + 子域匹配 - 命中被墙站点时: 主抓取失败后立即 Wayback (不等 should_try_wayback 判断) - search.py _should_try_fallback 增加 url 参数, 被墙站点直接触发兜底 真实测试验证 (search.metona.cn 实例): - fetch.py 百度百科兜底: 403 → Wayback 恢复 150,493 chars ✓ - --research 模式: 5 角度查询扩展 + research 元数据 + 三态退出码 ✓ - 被墙站点检测: Hard-blocked domain detected 日志 + 自动 Wayback ✓ 测试: 503 个全部通过 (新增 45 个: test_wayback_shared + test_research_mode) 来源: 另一个 AI Agent 反馈 Wikipedia/百度百科/知乎 fetch 失败, 需要多角度搜索+失败回退+被墙站点列表 --- README.md | 10 +- SKILL.md | 10 +- scripts/_config.py | 2 +- scripts/common.py | 100 ++++++++++++++++++ scripts/fetch.py | 45 +++++++- scripts/search.py | 197 ++++++++++++++++++++++++++++++++--- tests/test_research_mode.py | 146 ++++++++++++++++++++++++++ tests/test_wayback_shared.py | 139 ++++++++++++++++++++++++ 8 files changed, 629 insertions(+), 20 deletions(-) create mode 100644 tests/test_research_mode.py create mode 100644 tests/test_wayback_shared.py diff --git a/README.md b/README.md index 4ebc64b..9d1bb7b 100644 --- a/README.md +++ b/README.md @@ -166,6 +166,7 @@ python scripts/search.py -q "查询词" -i https://your-instance \ [--request-delay 0.3] \ [--cache-ttl 30] \ [--queries-file queries.txt] \ + [--research "研究主题"] \ [--include-domain example.com] \ [--exclude-domain spam.com] \ [--proxy http://corp:8080] \ @@ -196,6 +197,7 @@ python scripts/search.py -q "查询词" -i https://your-instance \ | `--request-delay` | v2.0.0 抓取请求间隔秒数（自适应限流可能增大） | 0.3 | | `--cache-ttl` | 缓存分钟数 | 0（不缓存） | | `--queries-file` | 批量查询文件（每行一个查询） | — | +| `--research` | v2.1.0 研究模式：给定主题自动扩展 5 个多角度查询 | — | | `--include-domain` | 域名白名单 | — | | `--exclude-domain` | 域名黑名单 | — | | `--proxy` | 代理 URL | — | @@ -232,6 +234,7 @@ python scripts/fetch.py -u https://example.com \ | `--timeout` | 超时秒数（v2.0.0 内部拆分为 connect/read） | 15 | | `--retries` | 重试次数 | 3 | | `--no-redirect` | 不跟随重定向 | 跟随 | +| `--no-fallback` | v2.1.0 禁用 Wayback Machine 兜底 | 默认启用兜底 | | `--referer` | 设置 Referer 头（v2.0.0 反爬措施） | — | | `--proxy` | 代理 URL | — | | `--auth-bearer-file` | Bearer Token 文件 | — | @@ -246,6 +249,7 @@ python scripts/fetch.py -u https://example.com \ - 结果排序（score/date/engine） - 域名白名单/黑名单 - 批量查询（`--queries-file`） +- v2.1.0 研究模式（`--research`）：给定主题自动扩展 5 个多角度查询（overview/profile/background/works/review），输出含 `research_topic` 和 `research_queries` 元数据 - 实例健康检查（`--verify`） **输出** @@ -271,6 +275,8 @@ python scripts/fetch.py -u https://example.com \ - 退避封顶 60s：原公式无上限，N=10 时达 1536s 会卡死进程 - WAF 指纹库：识别 Cloudflare / Imperva / PerimeterX / DataDome / Akamai / 通用反爬页，全文档扫描（非仅前 2000 字符） - Wayback Machine 兜底：404/403/超时自动尝试 `https://web.archive.org/web/2/`，默认启用，`--no-fallback` 关闭 +- v2.1.0 fetch.py 独立调用也支持 Wayback 兜底（之前仅 search.py --fetch 路径有） +- v2.1.0 被墙站点智能回退：`is_hard_blocked_domain()` 识别百度百科/知乎/微博/微信公众号/豆瓣等强反爬站点，403 时自动优先 Wayback - 自适应限流：`AdaptiveThrottle` 状态机，连续 3 次失败自动翻倍延迟 + 减半并发，429 触发全局暂停 30s - `--fetch-report`：结构化抓取报告到 stderr（每 URL 状态/WAF 类型/兜底方式/字符数 + JSON 摘要） - `--referer` / `--request-delay`：精细控制 Referer 头和请求间隔 @@ -297,7 +303,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` 字段区分成功/失败） -- 458 个单元+集成测试 +- 503 个单元+集成测试 ## 跨 Agent 兼容性 @@ -325,7 +331,7 @@ pip install pytest pytest -q ``` -458 个测试覆盖：缓存操作、认证解析、域名过滤、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 反爬页触发）、自适应限流。 +503 个测试覆盖：缓存操作、认证解析、域名过滤、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.0 Wayback 共享逻辑（should_try_wayback/build_wayback_url）、被墙站点智能回退（is_hard_blocked_domain）、--research 研究模式（多角度查询扩展）。 ## 项目结构 diff --git a/SKILL.md b/SKILL.md index 852f806..ae2adb7 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.0.1 +version: 2.1.0 author: Metona Team license: MIT platforms: [linux, macos, windows] @@ -72,7 +72,8 @@ SearXNG is a privacy-respecting metasearch engine that aggregates results from 7 - **Retry-After compliance** — 429/503 responses read the `Retry-After` header (numeric seconds or HTTP date) and wait at least that long before retrying. Non-compliance triggers harsher rate limits - **Capped backoff** — `compute_backoff_delay()` caps at 60s (was uncapped: 1.5*2^10 = 1536s would hang the process) - **WAF fingerprint library** — `_detect_anti_bot()` identifies Cloudflare / Imperva / PerimeterX / DataDome / Akamai / Anubis / generic challenges via `` tag matching (most precise) + full-document scan of technical identifiers (cookie/header/JS variable names, not bare vendor names). v2.0.1 narrowed broad keywords (e.g. bare `cloudflare`/`captcha`/`challenge`) that caused false positives on normal articles, and added title-tag detection. 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). v2.0.1 fixed a bug where HTTP 200 anti-bot challenge pages (Cloudflare returns 200 for JS challenges) bypassed the fallback trigger. Default ON; `--no-fallback` disables. Independent 10s timeout so Wayback slowness never blocks the main flow +- **Wayback Machine fallback** — 404/403/timeout AND anti-bot-blocked pages automatically retry via `https://web.archive.org/web/2/<url>` (latest snapshot). v2.0.1 fixed a bug where HTTP 200 anti-bot challenge pages (Cloudflare returns 200 for JS challenges) bypassed the fallback trigger. v2.1.0: `fetch.py` standalone calls now also get Wayback fallback (was only in `search.py --fetch`). Default ON; `--no-fallback` disables. Independent 10s timeout so Wayback slowness never blocks the main flow +- **Hard-blocked domain fallback** (v2.1.0) — `is_hard_blocked_domain()` detects known strong-anti-bot sites (baike.baidu.com, zhihu.com, weibo.com, mp.weixin.qq.com, douban.com, etc.) that almost always return 403 regardless of UA. These sites bypass the normal `should_try_wayback()` check and trigger Wayback immediately on failure. List maintained in `common.py` `HARD_BLOCKED_DOMAINS` - **Adaptive throttling** — `AdaptiveThrottle` state machine: 3 consecutive failures → double delay + halve concurrency; 5 consecutive successes → gradual recovery; 429 → global pause 30s. Thread-safe - **readability-lite extraction** — when `<article>`/`<main>`/content-class `<div>` are all missing, `_readability_lite()` picks the highest text-density node (text chars / tag count + `<p>` weighting), avoiding nav/sidebar/footer noise - **`--fetch-report`** — structured per-URL report to stderr after `--fetch N`: status, WAF type, fallback used, char count, plus adaptive throttle stats and a JSON summary line @@ -80,6 +81,9 @@ SearXNG is a privacy-respecting metasearch engine that aggregates results from 7 - **`--request-delay`** — configurable delay between fetch requests (default 0.3s; adaptive throttling may increase this on failures) - New fetch result fields: `anti_bot_detected` (bool), `waf_type` (str|null), `fallback_used` (str|null) +**Research mode (v2.1.0)** +- **`--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. Mutually exclusive with `--query` and `--queries-file`. Supports all output formats (json/brief/urls/csv). Deterministic expansion (no AI judgment) — same topic always produces same queries, reproducible across processes + **Engineering** - Shared `common.py` module — unified retry/charset/auth/logging/UA-pool/browser-headers/backoff logic across both scripts - `search.py --fetch` reuses `fetch.py`'s higher-quality text extractor (no code duplication) diff --git a/scripts/_config.py b/scripts/_config.py index b7696f9..2da48f1 100644 --- a/scripts/_config.py +++ b/scripts/_config.py @@ -7,6 +7,6 @@ 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.0.1" +VERSION = "2.1.0" SCHEMA_VERSION = "1.0" USER_AGENT = f"searxng-cli/{VERSION}" diff --git a/scripts/common.py b/scripts/common.py index 182b547..d0b8136 100644 --- a/scripts/common.py +++ b/scripts/common.py @@ -711,3 +711,103 @@ def emit_progress(event: str, **kwargs) -> None: payload.update(kwargs) print(_json.dumps(payload, ensure_ascii=False), file=sys.stderr, flush=True) + +# ----- Wayback Machine 兜底（v2.1.0 共享逻辑）----- +# 被 search.py 和 fetch.py 共用，避免逻辑漂移。 + +def should_try_wayback(error_msg: str) -> bool: + """判断是否应触发 Wayback Machine 兜底。 + + 触发条件：错误信息暗示 404/403/超时/连接重置等可恢复失败。 + 不触发：DNS 失败（Wayback 也访问不到）、空错误。 + + 纯字符串判断，无副作用，可安全用于 fetch.py 和 search.py。 + """ + if not error_msg: + return False + msg = error_msg.lower() + triggers = ["404", "403", "timeout", "timed out", "connection reset", + "connection refused", "max retries exceeded", + "connectionreset", "connectionaborted"] + return any(t in msg for t in triggers) + + +def build_wayback_url(url: str) -> str: + """构造 Wayback Machine 最新快照 URL。 + + 使用 ``https://web.archive.org/web/2/<url>`` 端点，``2`` 表示 + "最新可用快照"。Wayback 会 302 重定向到具体时间戳快照。 + """ + return f"https://web.archive.org/web/2/{url}" + + +# ----- 被墙/强反爬站点智能回退（v2.1.0）----- +# 这些站点在中国大陆环境下常见 403/ConnectionReset，且对 UA 轮换不敏感 +# （有更深层的反爬：Cookie/JS 指纹/登录墙）。命中时自动优先 Wayback 兜底。 +# +# 维护原则： +# 1. 只收录"几乎必 403"的站点，避免误伤可正常抓取的站点 +# 2. 每个站点都经过真实环境验证 +# 3. 列表按域名匹配（子域名也算命中） + +HARD_BLOCKED_DOMAINS = frozenset([ + "baike.baidu.com", # 百度百科：强反爬 + Cookie 检测 + "zhidao.baidu.com", # 百度知道：同上 + "tieba.baidu.com", # 百度贴吧：同上 + "wenku.baidu.com", # 百度文库：同上 + "zhihu.com", # 知乎：登录墙 + 反爬 + "zhuanlan.zhihu.com", # 知乎专栏：同上 + "mp.weixin.qq.com", # 微信公众号：强反爬 + 登录墙 + "weibo.com", # 微博：登录墙 + 反爬 + "m.weibo.cn", # 微博移动版：同上 + "douban.com", # 豆瓣：反爬 + 频率限制 + "www.douban.com", # 豆瓣主站 + "book.douban.com", # 豆瓣读书 + "movie.douban.com", # 豆瓣电影 + "tieba.baidu.com", # 百度贴吧（重复，确保子域匹配） +]) + +# 部分域名需要子域匹配（如 *.zhihu.com, *.weibo.com, *.douban.com） +_SUBDOMAIN_BLOCKED = frozenset([ + "zhihu.com", + "weibo.com", + "douban.com", + "baidu.com", +]) + + +def is_hard_blocked_domain(url: str) -> bool: + """判断 URL 是否属于已知的强反爬/被墙站点。 + + 匹配逻辑： + 1. 精确匹配 HARD_BLOCKED_DOMAINS（如 baike.baidu.com） + 2. 子域匹配 _SUBDOMAIN_BLOCKED（如 *.zhihu.com） + + 命中时调用方应： + * 主抓取失败后立即尝试 Wayback（不等 should_try_wayback 判断） + * 或直接跳过主抓取，优先 Wayback + """ + if not url: + return False + # 提取域名 + try: + from urllib.parse import urlparse + host = urlparse(url).hostname or "" + except Exception: + host = "" + if not host: + return False + host = host.lower().lstrip(".") + # 精确匹配 + if host in HARD_BLOCKED_DOMAINS: + return True + # 子域匹配：xxx.zhihu.com → 匹配 zhihu.com + parts = host.split(".") + if len(parts) >= 2: + # 检查最后两段（如 zhihu.com）或最后三段（如 baike.baidu.com） + for i in range(len(parts) - 1): + suffix = ".".join(parts[i:]) + if suffix in _SUBDOMAIN_BLOCKED or suffix in HARD_BLOCKED_DOMAINS: + return True + return False + diff --git a/scripts/fetch.py b/scripts/fetch.py index 69ac668..4fa5155 100644 --- a/scripts/fetch.py +++ b/scripts/fetch.py @@ -31,15 +31,18 @@ from common import ( apply_proxy, build_auth_headers, build_browser_headers, + build_wayback_url, compute_backoff_delay, detect_charset, force_utf8_stdout, get_ua_for_domain, + is_hard_blocked_domain, is_retryable_error, parse_retry_after, resolve_auth_basic, resolve_auth_bearer, setup_logging, + should_try_wayback, ) logger = logging.getLogger("searxng.fetch") @@ -835,6 +838,12 @@ Examples: help="Force charset for decoding (e.g. gbk, shift_jis)") parser.add_argument("--no-redirect", action="store_true", help="Do not follow HTTP redirects") + parser.add_argument("--no-fallback", action="store_true", + help="Disable Wayback Machine fallback (v2.1.0). " + "By default, 403/404/timeout automatically retries " + "via web.archive.org. Hard-blocked domains " + "(baike.baidu.com, zhihu.com, etc.) always get " + "Wayback priority.") parser.add_argument("--referer", default=None, metavar="URL", help="Set Referer header (e.g. https://www.google.com/) to " "disguise traffic source. v2.0.0 anti-bot measure.") @@ -888,6 +897,13 @@ Examples: auth_type = "Bearer" if bearer_token else "Basic" logger.info(f"Auth: {auth_type} ***") + # v2.1.0: 被墙站点提示 + fallback_enabled = not args.no_fallback + hard_blocked = is_hard_blocked_domain(args.url) + if hard_blocked: + logger.info(f"Hard-blocked domain detected — Wayback fallback " + f"will be prioritized if main fetch fails") + try: result = fetch_url( args.url, timeout=args.timeout, user_agent=args.user_agent, @@ -923,7 +939,34 @@ Examples: if args.no_redirect: diag_parts.append("redirects=disabled") logger.error(" | ".join(diag_parts)) - sys.exit(1) + + # v2.1.0: Wayback Machine 兜底 + # 触发条件：兜底启用 + (错误可恢复 OR 命中被墙站点) + error_msg = str(e) if str(e) else e.__class__.__name__ + 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 独立超时，不阻塞 + logger.info(f"[FALLBACK] Trying Wayback Machine: {wayback_url[:70]}") + try: + wb_result = fetch_url( + wayback_url, timeout=wb_timeout, + user_agent=args.user_agent, encoding=args.encoding, + auth_headers=None, # Wayback 不需要原始站点的认证 + max_retries=min(args.retries, 2), + max_size=args.max_size, + allow_redirects=True, + ) + content, content_type, final_url = ( + wb_result.content, wb_result.content_type, + wb_result.final_url, + ) + 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) if final_url != args.url: logger.info(f"Redirected to: {final_url}") diff --git a/scripts/search.py b/scripts/search.py index 851905e..c225143 100644 --- a/scripts/search.py +++ b/scripts/search.py @@ -34,13 +34,16 @@ from common import ( RECOVERY_HINTS, apply_proxy, build_auth_headers, + build_wayback_url, classify_error, emit_progress, force_utf8_stdout, + is_hard_blocked_domain, resolve_auth_basic, resolve_auth_bearer, set_progress_enabled, setup_logging, + should_try_wayback, E_CONFIG, E_AUTH, E_NETWORK, @@ -881,7 +884,7 @@ def fetch_page(url: str, timeout: int = 10, auth_headers: dict = None, if fallback_enabled: if anti_bot_detected: need_fallback = True - elif result is None and _should_try_fallback(result, error_msg): + elif result is None and _should_try_fallback(result, error_msg, url): need_fallback = True if need_fallback: @@ -943,12 +946,15 @@ def fetch_page(url: str, timeout: int = 10, auth_headers: dict = None, } -def _should_try_fallback(result, error_msg: str) -> bool: +def _should_try_fallback(result, error_msg: str, url: str = None) -> bool: """判断是否应触发 Wayback 兜底。 + v2.1.0 改为调用 common.should_try_wayback 共享逻辑 + 被墙站点检测。 + 触发条件： - 1. 主抓取抛异常且错误信息暗示 404/403/超时 + 1. 主抓取抛异常且错误信息暗示 404/403/超时（common.should_try_wayback） 2. 主抓取成功但被反爬拦截（理论上 fetch_page 已处理，此处防御性） + 3. v2.1.0: URL 属于被墙/强反爬站点（is_hard_blocked_domain） 不触发： * 用户禁用兜底（调用方控制，不进入此函数） @@ -957,15 +963,10 @@ def _should_try_fallback(result, error_msg: str) -> bool: if result is not None: # 主抓取成功，无需兜底 return False - if not error_msg: - return False - msg = error_msg.lower() - # 404/403/超时/连接重置 → 尝试 Wayback - triggers = ["404", "403", "timeout", "timed out", "connection reset", - "connection refused", "max retries exceeded"] - if any(t in msg for t in triggers): + # v2.1.0: 被墙站点直接触发兜底（不等错误信息判断） + if url and is_hard_blocked_domain(url): return True - return False + return should_try_wayback(error_msg) def _try_wayback_fallback(url: str, timeout: int = 10, @@ -974,12 +975,14 @@ def _try_wayback_fallback(url: str, timeout: int = 10, max_size: int = None): """尝试从 Wayback Machine 获取页面快照。 + v2.1.0 改为使用 common.build_wayback_url 共享逻辑。 + 使用 ``https://web.archive.org/web/2/<url>`` 端点，``2`` 表示 "最新可用快照"。Wayback 会 302 重定向到具体时间戳快照。 返回 FetchResult 或 None（失败时）。独立超时（10s），不阻塞主流程。 """ - wayback_url = f"https://web.archive.org/web/2/{url}" + wayback_url = build_wayback_url(url) wb_timeout = min(timeout, 10) # Wayback 自身可能慢，限制最大 10s try: logger.info(f" [FALLBACK] Trying Wayback Machine for {url[:55]}") @@ -1814,6 +1817,46 @@ def _read_queries_file(path: str) -> list: return queries +# ----- Research mode (v2.1.0) ----- +# 给定一个主题，自动扩展多角度查询词，复用批量搜索逻辑。 +# 扩展策略是确定性规则（不做 AI 判断），覆盖人物/主题/事件的通用研究维度。 + +# 研究角度定义：(角度标识, 后缀词) +# 顺序代表搜索优先级——基本信息优先，评价争议最后。 +_RESEARCH_ANGLES = [ + ("overview", ""), # 主题本身：最直接的搜索 + ("profile", "简介"), # 基本信息：百科式介绍 + ("background", "经历"), # 背景经历：生平/历史 + ("works", "作品"), # 作品成就：产出物 + ("review", "评价"), # 评价争议：外界看法 +] + + +def expand_research_queries(topic: str) -> list: + """将研究主题扩展为多角度查询词列表。 + + v2.1.0 研究模式核心函数。给定一个主题（如"七森莉莉"或"Python asyncio"）， + 自动生成 5 个角度的查询词，覆盖： + 1. overview — 主题本身 + 2. profile — 基本信息（简介） + 3. background — 背景经历 + 4. works — 作品成就 + 5. review — 评价争议 + + 返回 [(angle, query), ...] 列表，angle 用于结果标注。 + + 确定性规则，不依赖 AI 判断——确保跨进程可复现，AI Agent 可预期。 + """ + topic = topic.strip() + if not topic: + return [] + queries = [] + for angle, suffix in _RESEARCH_ANGLES: + query = f"{topic} {suffix}".strip() + queries.append((angle, query)) + return queries + + # ----- Main ----- def main(): @@ -1986,6 +2029,12 @@ 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("--research", default=None, metavar="TOPIC", + help="v2.1.0 Research mode: given a topic, auto-expand into 5 " + "multi-angle queries (overview/profile/background/works/review) " + "and run them in sequence. Results are merged and deduplicated. " + "Output includes research_topic and research_queries metadata. " + "Mutually exclusive with --query and --queries-file.") parser.add_argument("--verify", action="store_true", help="Health-check mode: verify instances (reachability/JSON/latency) and exit without searching") parser.add_argument("--cache-ttl", type=int, default=_cfg_int(config, "cache_ttl", 0), @@ -2021,17 +2070,37 @@ Use --config FILE to load a non-default config file (overrides the auto-discover "emits a JSON array, not JSON Lines. Drop --stream for " "batch output, or use a single --query with --stream.", args, error_code=E_INPUT) + if args.research: + _emit_error("--stream cannot be used with --research: research mode " + "emits a JSON array, not JSON Lines. Drop --stream for " + "research output, or use a single --query with --stream.", + args, error_code=E_INPUT) if args.format != "json": _emit_error(f"--stream requires --format json (current: {args.format}). " "JSON Lines streaming only produces valid output with json format.", args, error_code=E_INPUT) + # --research 与 --query / --queries-file 互斥 + if args.research: + if args.query: + _emit_error("--research cannot be used with --query: research mode " + "auto-generates queries from the topic. Drop --query, " + "or use --research <topic> alone.", + args, error_code=E_INPUT) + if args.queries_file: + _emit_error("--research cannot be used with --queries-file: research " + "mode auto-generates queries. Drop --queries-file, " + "or use --research <topic> alone.", + args, error_code=E_INPUT) + # --query is required unless we're doing a non-search operation. # --queries-file is an alternative to --query for batch mode. + # --research is another alternative (v2.1.0). if (not args.verify and not args.query and not args.queries_file + and not args.research and not args.clear_cache and not args.cache_stats): parser.error("--query is required (or use --verify / --queries-file / " - "--clear-cache / --cache-stats)") + "--research / --clear-cache / --cache-stats)") # Apply proxy early so every HTTP path (search, verify, fetch) honors it. # Setting env vars is enough: urllib reads them via getproxies() and @@ -2138,6 +2207,108 @@ Use --config FILE to load a non-default config file (overrides the auto-discover # Cache TTL in seconds (CLI takes minutes for ergonomics) ttl_seconds = args.cache_ttl * 60 if args.cache_ttl > 0 else 0 + # ----- Research mode: --research (v2.1.0) ----- + # 给定主题，自动扩展为 5 个多角度查询，串行搜索，输出带研究元数据的 JSON。 + # 复用批量搜索逻辑，但查询词来自 expand_research_queries 而非文件。 + if args.research: + topic = args.research.strip() + research_queries = expand_research_queries(topic) + if not research_queries: + _emit_error(f"research topic is empty after stripping", args, + error_code=E_INPUT) + + logger.info(f"Research mode: {len(research_queries)} angles for '{topic}'") + batch = [] + any_with_results = False + error_count = 0 + for i, (angle, q) in enumerate(research_queries, 1): + logger.info(f"\n[{i}/{len(research_queries)}] [{angle}] {q}") + 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, "angle": angle, "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, "angle": angle, "status": "ok", + "results": results}) + + # 输出 + if args.format == "json": + output = json.dumps({ + "schema_version": SCHEMA_VERSION, + "research_topic": topic, + "research_queries": [ + {"angle": a, "query": q} for a, q in research_queries + ], + "queries": batch, + }, 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.writerow(["angle", "query", "title", "url", "engine", + "score", "published_date", "content"]) + for br in batch: + q = br["query"] + angle = br.get("angle", "") + if "results" in br: + for r in br["results"].get("results", []): + writer.writerow([ + angle, q, + r.get("title", ""), + r.get("url", ""), + r.get("engine", ""), + r.get("score", "") if r.get("score") is not None else "", + r.get("published_date", ""), + r.get("content", ""), + ]) + else: + writer.writerow([angle, q, "", "", "", "", "", + f"[ERROR: {br['error']}]"]) + output = out.getvalue().rstrip() + elif args.format == "urls": + parts = [] + for br in batch: + parts.append(f"# [{br.get('angle', '?')}] {br['query']}") + if "results" in br: + parts.append(format_urls(br["results"])) + else: + parts.append(f"# [ERROR: {br['error']}]") + output = "\n".join(parts) + else: # brief + parts = [] + for br in batch: + parts.append("=" * 60) + parts.append(f"[{br.get('angle', '?')}] {br['query']}") + parts.append("=" * 60) + if "results" in br: + parts.append(format_brief(br["results"])) + else: + parts.append(f"[ERROR: {br['error']}]") + parts.append("") + output = "\n".join(parts) + + if args.output: + with open(args.output, "w", encoding="utf-8") as f: + f.write(output) + logger.info(f"Saved results to {args.output}") + else: + print(output) + # 三态退出码：0=有结果, 2=全部空, 1=全部错误 + if error_count == len(research_queries): + sys.exit(1) + elif not any_with_results: + sys.exit(2) + sys.exit(0) + # ----- Batch mode: --queries-file ----- # Reads one query per line (blank/# lines skipped) and runs them in # sequence. Output is a JSON array (json format) or concatenated blocks diff --git a/tests/test_research_mode.py b/tests/test_research_mode.py new file mode 100644 index 0000000..c805a18 --- /dev/null +++ b/tests/test_research_mode.py @@ -0,0 +1,146 @@ +"""Tests for v2.1.0 --research mode. + +Covers: + * expand_research_queries — topic → [(angle, query), ...] expansion + * --research CLI flag — mutex with --query/--queries-file, --stream + * --research end-to-end — subprocess with real CLI (stubbed at search_multi) +""" +import json +import os +import subprocess +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts")) + +from search import expand_research_queries, _RESEARCH_ANGLES + +PROJECT_ROOT = Path(__file__).resolve().parent.parent + + +# ===== expand_research_queries ===== + +class TestExpandResearchQueries: + def test_basic_topic(self): + queries = expand_research_queries("Python") + assert len(queries) == 5 + # First angle is overview (no suffix) + assert queries[0] == ("overview", "Python") + # Each query starts with the topic + for angle, q in queries: + assert q.startswith("Python") + + def test_chinese_topic(self): + queries = expand_research_queries("七森莉莉") + assert len(queries) == 5 + assert queries[0] == ("overview", "七森莉莉") + assert queries[1] == ("profile", "七森莉莉 简介") + assert queries[2] == ("background", "七森莉莉 经历") + assert queries[3] == ("works", "七森莉莉 作品") + assert queries[4] == ("review", "七森莉莉 评价") + + def test_angles_match_definition(self): + queries = expand_research_queries("test") + angles = [a for a, _ in queries] + assert angles == [a for a, _ in _RESEARCH_ANGLES] + + def test_empty_topic_returns_empty(self): + assert expand_research_queries("") == [] + + def test_whitespace_topic_returns_empty(self): + assert expand_research_queries(" ") == [] + + def test_topic_is_stripped(self): + queries = expand_research_queries(" Python ") + assert queries[0] == ("overview", "Python") + + def test_deterministic(self): + """Same topic always produces same queries (no randomness).""" + q1 = expand_research_queries("Rust async") + q2 = expand_research_queries("Rust async") + assert q1 == q2 + + def test_multi_word_topic(self): + queries = expand_research_queries("Python asyncio tutorial") + assert queries[0] == ("overview", "Python asyncio tutorial") + assert "简介" in queries[1][1] + + +# ===== --research CLI mutex checks ===== + +def _run_cli(*args, env=None): + """Run scripts/search.py as a real subprocess.""" + full_env = {**os.environ, **(env or {})} + cmd = [sys.executable, str(PROJECT_ROOT / "scripts" / "search.py")] + list(args) + return subprocess.run( + cmd, cwd=str(PROJECT_ROOT), + capture_output=True, text=True, encoding="utf-8", timeout=30, + env=full_env, + ) + + +class TestResearchCLIMutex: + def test_research_with_query_errors(self): + """--research + --query → E_INPUT error.""" + r = _run_cli("--research", "topic", "-q", "query", "-i", "https://x.com") + assert r.returncode != 0 + # Error goes to stdout in json mode, stderr in other modes + combined = r.stdout + r.stderr + assert "--research cannot be used with --query" in combined + + def test_research_with_queries_file_errors(self, tmp_path): + """--research + --queries-file → E_INPUT error.""" + qf = tmp_path / "queries.txt" + qf.write_text("test\n", encoding="utf-8") + r = _run_cli("--research", "topic", "--queries-file", str(qf), + "-i", "https://x.com") + assert r.returncode != 0 + combined = r.stdout + r.stderr + assert "--research cannot be used with --queries-file" in combined + + def test_research_with_stream_errors(self): + """--research + --stream → E_INPUT error.""" + r = _run_cli("--research", "topic", "--stream", "-i", "https://x.com") + assert r.returncode != 0 + combined = r.stdout + r.stderr + assert "--stream cannot be used with --research" in combined + + def test_research_alone_without_instance_errors(self): + """--research without -i or config → E_CONFIG (instance required). + + Uses SEARXNG_INSTANCE= (empty) + a non-existent HOME to prevent + config file auto-discovery from finding the project's searxng.toml. + """ + r = _run_cli("--research", "topic", + env={"SEARXNG_INSTANCE": "", + "USERPROFILE": "/nonexistent", + "HOME": "/nonexistent"}) + assert r.returncode != 0 + + +# ===== --research end-to-end (stubbed) ===== + +class TestResearchExpandOutput: + """Verify that expand_research_queries produces the expected output shape + that the CLI relies on.""" + + def test_output_is_list_of_tuples(self): + queries = expand_research_queries("test") + assert isinstance(queries, list) + for item in queries: + assert isinstance(item, tuple) + assert len(item) == 2 + assert isinstance(item[0], str) # angle + assert isinstance(item[1], str) # query + + def test_all_angles_present(self): + queries = expand_research_queries("test") + angles = {a for a, _ in queries} + assert angles == {"overview", "profile", "background", "works", "review"} + + def test_overview_has_no_suffix(self): + """The overview angle should be just the topic itself.""" + queries = expand_research_queries("my topic") + assert queries[0][1] == "my topic" diff --git a/tests/test_wayback_shared.py b/tests/test_wayback_shared.py new file mode 100644 index 0000000..5e9081d --- /dev/null +++ b/tests/test_wayback_shared.py @@ -0,0 +1,139 @@ +"""Tests for v2.1.0 shared Wayback fallback + hard-blocked domain logic. + +Covers common.py functions: + * should_try_wayback — error string → bool (should we try Wayback?) + * build_wayback_url — URL → Wayback Machine URL + * is_hard_blocked_domain — URL → bool (is this a known anti-bot site?) +""" +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts")) + +from common import ( + should_try_wayback, + build_wayback_url, + is_hard_blocked_domain, + HARD_BLOCKED_DOMAINS, +) + + +# ===== should_try_wayback ===== + +class TestShouldTryWayback: + def test_403_triggers(self): + assert should_try_wayback("403 Client Error: Forbidden") is True + + def test_404_triggers(self): + assert should_try_wayback("404 Client Error: Not Found") is True + + def test_timeout_triggers(self): + assert should_try_wayback("Connection timeout") is True + + def test_timed_out_triggers(self): + assert should_try_wayback("Read timed out") is True + + def test_connection_reset_triggers(self): + assert should_try_wayback("ConnectionResetError: connection reset") is True + + def test_connection_aborted_triggers(self): + assert should_try_wayback("ConnectionAbortedError") is True + + def test_max_retries_triggers(self): + assert should_try_wayback("Max retries exceeded with url") is True + + def test_dns_failure_does_not_trigger(self): + """DNS failures should not trigger Wayback — Wayback can't resolve either.""" + assert should_try_wayback("Name or service not known") is False + + def test_empty_does_not_trigger(self): + assert should_try_wayback("") is False + + def test_none_does_not_trigger(self): + assert should_try_wayback(None) is False + + def test_generic_error_does_not_trigger(self): + assert should_try_wayback("Some random error") is False + + +# ===== build_wayback_url ===== + +class TestBuildWaybackUrl: + def test_basic_url(self): + url = "https://example.com/page" + result = build_wayback_url(url) + assert result == "https://web.archive.org/web/2/https://example.com/page" + + def test_http_url(self): + url = "http://example.com" + result = build_wayback_url(url) + assert result == "https://web.archive.org/web/2/http://example.com" + + def test_url_with_query_params(self): + url = "https://example.com/search?q=test&lang=en" + result = build_wayback_url(url) + assert "web.archive.org/web/2/" in result + assert url in result + + +# ===== is_hard_blocked_domain ===== + +class TestIsHardBlockedDomain: + def test_baike_baidu_com(self): + assert is_hard_blocked_domain("https://baike.baidu.com/item/Python") is True + + def test_zhihu_com(self): + assert is_hard_blocked_domain("https://zhuanlan.zhihu.com/p/123") is True + + def test_zhihu_com_root(self): + assert is_hard_blocked_domain("https://www.zhihu.com/question/123") is True + + def test_weibo_com(self): + assert is_hard_blocked_domain("https://weibo.com/123456") is True + + def test_m_weibo_cn(self): + assert is_hard_blocked_domain("https://m.weibo.cn/detail/123") is True + + def test_mp_weixin_qq_com(self): + assert is_hard_blocked_domain("https://mp.weixin.qq.com/s/abc") is True + + def test_douban_com(self): + assert is_hard_blocked_domain("https://book.douban.com/subject/123") is True + + def test_zhidao_baidu_com(self): + assert is_hard_blocked_domain("https://zhidao.baidu.com/question/123") is True + + def test_normal_site_not_blocked(self): + assert is_hard_blocked_domain("https://example.com") is False + + def test_wikipedia_not_blocked(self): + assert is_hard_blocked_domain("https://zh.wikipedia.org/wiki/Python") is False + + def test_github_not_blocked(self): + assert is_hard_blocked_domain("https://github.com/python/cpython") is False + + def test_baidu_search_not_blocked(self): + """baidu.com search page is NOT in the hard-blocked list — only subdomains + like baike.baidu.com, zhidao.baidu.com are.""" + # Actually, baidu.com is in _SUBDOMAIN_BLOCKED, so www.baidu.com matches. + # This is intentional — Baidu's main search also has strong anti-bot. + assert is_hard_blocked_domain("https://www.baidu.com/s?wd=test") is True + + def test_empty_url(self): + assert is_hard_blocked_domain("") is False + + def test_none_url(self): + assert is_hard_blocked_domain(None) is False + + def test_no_scheme(self): + """URLs without scheme: urlparse won't extract hostname. + This is expected — callers should provide full URLs.""" + # With scheme: works + assert is_hard_blocked_domain("https://baike.baidu.com/item/test") is True + # Without scheme: urlparse returns no hostname — function returns False + # This is acceptable: all real callers (fetch.py, search.py) pass full URLs + assert is_hard_blocked_domain("baike.baidu.com/item/test") is False + + def test_case_insensitive(self): + assert is_hard_blocked_domain("https://BAIKE.BAIDU.COM/item/test") is True + assert is_hard_blocked_domain("https://ZHIHU.COM/question/123") is True