thzxx
10c01a3abd
正确性修复: - 修复 Sec-Ch-Ua 构造 bug: 原实现产出 ""Not_A Brand";v="99"" 双重引号 畸形头(Chrome/Edge 两路径), 严格校验的 WAF 会忽略; 改为品牌数组拼接 - search 403 快速失败: 实例级 403 不再退避重试(~10.5s 空等), 立即 failover - AdaptiveThrottle: --throttle-failure-threshold 0 现为真正禁用语义 - number_of_results 缺失时用 len(results) 兜底(JSON/HTML 路径契约对齐) - 版本对齐: pyproject.toml 与 _config.py 同步 2.5.0 AI 代理体验: - fetch.py --extract json: 结构化骨架(title/meta/headings/links/images) - fetch.py --max-chars N: 提取后语义级截断(区别于 --max-size 字节截断) - search --fetch-total-chars N: --fetch 全局字符预算, 耗尽后 status=skipped - --dedup-fetched-content: 抓取正文 SimHash 去重, status=duplicate - --progress 新增 angle_start/ok/fail + fetch_skip/fetch_duplicate 事件 - fetch.py 补齐 --log-format json + --dump-schema - CSV 媒体列自适应(images/videos 类别自动追加媒体字段列) - --dry-run 批量模式打印实际查询列表 - search 连接复用: requests 可用时走模块级 Session(连接池) 工程治理: - 新增 scripts/release_check.py 发布一致性检查(版本/错误码表漂移) - 新增 tests/test_v250_features.py 46+4 个回归测试(全量 622 通过) - tests/conftest.py: autouse fixture 强制 stdlib 路径(本机有 requests 时 既有 urllib mock 测试不失效)
1090 lines
44 KiB
Python
1090 lines
44 KiB
Python
"""Shared utilities for searxng-cli scripts.
|
||
|
||
This module centralizes code that was previously duplicated across
|
||
``search.py`` and ``fetch.py``:
|
||
|
||
* ``build_auth_headers`` — construct an Authorization header from CLI flags
|
||
* ``resolve_auth_basic`` — resolve basic-auth credentials from file/env/CLI
|
||
(avoids leaving passwords in shell history)
|
||
* ``detect_charset`` — guess a response's text encoding
|
||
* ``is_retryable_error`` — unified transient-error policy (urllib + requests)
|
||
* ``FALLBACK_UAS`` — browser-like User-Agents used when blocked
|
||
* retry constants — ``RETRYABLE_STATUS``, ``RETRY_BACKOFF_BASE``, etc.
|
||
* ``setup_logging`` — shared logging configuration (--verbose/--quiet)
|
||
* ``force_utf8_stdout`` — force stdout to UTF-8 (fix Windows GBK crashes)
|
||
|
||
Centralizing the retry policy guarantees that both scripts treat 429/5xx
|
||
as retryable and connection errors as transient, eliminating the previous
|
||
inconsistency where ``search.py`` ignored 5xx.
|
||
"""
|
||
|
||
import 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()
|
||
# call controls them all.
|
||
_LOG = logging.getLogger("searxng")
|
||
|
||
# Brotli 解压支持检测（v2.2.1）。
|
||
# requests 库自动解压 gzip/deflate，但**不自动解压 Brotli**（除非安装
|
||
# brotli/brotlicffi 包）。若在 Accept-Encoding 中声明 br 而系统未安装
|
||
# 解压器，服务器返回的 br 压缩字节会被当作文本解码 → 全页乱码。
|
||
# 此检测用于 build_browser_headers() 智能声明 Accept-Encoding，避免
|
||
# 声明无法兑现的 br。
|
||
try:
|
||
import brotli as _brotli # type: ignore
|
||
_HAS_BROTLI = True
|
||
except ImportError:
|
||
try:
|
||
import brotlicffi as _brotli # type: ignore
|
||
_HAS_BROTLI = True
|
||
except ImportError:
|
||
_brotli = None
|
||
_HAS_BROTLI = False
|
||
|
||
|
||
def setup_logging(verbose: bool = False, quiet: bool = False,
|
||
log_format: str = "text",
|
||
request_id: str = None) -> None:
|
||
"""Configure the ``searxng`` logger hierarchy.
|
||
|
||
* Default (no flags): ``INFO`` — progress messages, warnings, retry notices.
|
||
Matches the previous ``print(..., file=sys.stderr)`` behavior so existing
|
||
scripts and agents see no change.
|
||
* ``--verbose`` / ``-v``: ``DEBUG`` — also shows HTTP request URLs, response
|
||
status codes, cache keys, and other diagnostic detail.
|
||
* ``--quiet`` / ``-q``: ``WARNING`` — suppresses progress and retry noise;
|
||
only warnings and errors reach stderr.
|
||
* ``--log-format json`` (v2.2.0): 每行一个 JSON 对象，便于 AI Agent
|
||
程序化解析。包含 ts/level/logger/msg/request_id 字段。
|
||
* request_id (v2.2.0): 贯穿所有日志和进度事件的请求标识符。
|
||
|
||
All log output goes to stderr; stdout is reserved for data (JSON/CSV/etc.).
|
||
"""
|
||
if verbose:
|
||
level = logging.DEBUG
|
||
elif quiet:
|
||
level = logging.WARNING
|
||
else:
|
||
level = logging.INFO
|
||
|
||
_LOG.setLevel(level)
|
||
# 存储 request_id 到 logger 全局，供 formatter 和进度事件使用
|
||
_LOG._request_id = request_id
|
||
|
||
# Avoid duplicate handlers if setup_logging() is called twice (e.g. tests).
|
||
if not _LOG.handlers:
|
||
handler = logging.StreamHandler(sys.stderr)
|
||
if log_format == "json":
|
||
handler.setFormatter(_JsonFormatter())
|
||
else:
|
||
handler.setFormatter(logging.Formatter("%(message)s"))
|
||
_LOG.addHandler(handler)
|
||
else:
|
||
# 已有 handler（如测试环境），更新 formatter
|
||
for h in _LOG.handlers:
|
||
if log_format == "json":
|
||
h.setFormatter(_JsonFormatter())
|
||
else:
|
||
h.setFormatter(logging.Formatter("%(message)s"))
|
||
# Don't let root logger add its own handler — we own the searxng namespace.
|
||
_LOG.propagate = False
|
||
|
||
|
||
class _JsonFormatter(logging.Formatter):
|
||
"""JSON 结构化日志 formatter（v2.2.0）。
|
||
|
||
每行输出一个 JSON 对象：{"ts", "level", "logger", "msg", "request_id"}。
|
||
让 AI Agent 可程序化解析日志（统计重试次数、识别慢实例等）。
|
||
"""
|
||
|
||
def format(self, record):
|
||
entry = {
|
||
"ts": _datetime_iso(record),
|
||
"level": record.levelname,
|
||
"logger": record.name,
|
||
"msg": record.getMessage(),
|
||
}
|
||
rid = getattr(_LOG, "_request_id", None)
|
||
if rid:
|
||
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)
|
||
|
||
|
||
def _datetime_iso(record):
|
||
"""格式化日志时间戳为 ISO 8601 字符串。"""
|
||
return datetime.datetime.fromtimestamp(record.created).isoformat(timespec="milliseconds")
|
||
|
||
|
||
def force_utf8_stdout() -> None:
|
||
"""Force stdout/stderr to UTF-8 to prevent Windows GBK encoding crashes.
|
||
|
||
Windows Python defaults ``sys.stdout`` to the OEM codepage (often GBK on
|
||
Chinese Windows). ``print()`` of any character outside that codepage
|
||
(e.g. ``\\xa0`` nbsp, CJK punctuation from foreign pages) raises
|
||
``UnicodeEncodeError`` and kills the process.
|
||
|
||
``PYTHONIOENCODING=utf-8`` is unreliable here because Python 3.7+
|
||
reconfigures stdout after reading that env var in some scenarios (e.g.
|
||
when stdout has already been wrapped). The only reliable fix is to
|
||
reconfigure the stream in-process.
|
||
|
||
Uses ``sys.stdout.reconfigure()`` on Python 3.7+, falling back to
|
||
wrapping ``sys.stdout.buffer`` on older versions. Both paths use
|
||
``errors='replace'`` so an undecodable byte never crashes the script —
|
||
better to emit ``?`` than to lose all output.
|
||
|
||
Safe to call multiple times; subsequent calls are no-ops once the
|
||
encoding is already UTF-8 (or close enough — we check the lowercased
|
||
encoding name to tolerate ``utf-8`` vs ``UTF-8`` vs ``utf8``).
|
||
"""
|
||
for stream_name in ("stdout", "stderr"):
|
||
stream = getattr(sys, stream_name, None)
|
||
if stream is None:
|
||
continue
|
||
# Already UTF-8? Skip (covers Linux/macOS and re-invoked scripts).
|
||
enc = getattr(stream, "encoding", "") or ""
|
||
if enc.lower().replace("-", "") in ("utf8", "utf-8-sig"):
|
||
continue
|
||
# Python 3.7+ has TextIOWrapper.reconfigure()
|
||
reconfigure = getattr(stream, "reconfigure", None)
|
||
if reconfigure is not None:
|
||
try:
|
||
reconfigure(encoding="utf-8", errors="replace")
|
||
continue
|
||
except (ValueError, OSError):
|
||
pass # Fall through to the buffer-wrap path
|
||
# Fallback: wrap the underlying buffer in a new UTF-8 stream.
|
||
buffer = getattr(stream, "buffer", None)
|
||
if buffer is not None:
|
||
try:
|
||
new_stream = io.TextIOWrapper(
|
||
buffer, encoding="utf-8", errors="replace", line_buffering=True,
|
||
)
|
||
setattr(sys, stream_name, new_stream)
|
||
except (ValueError, AttributeError):
|
||
# Last resort: keep the original stream. Better to risk a
|
||
# GBK crash on exotic characters than to break stdout entirely.
|
||
pass
|
||
|
||
# Retry settings (shared by both scripts)
|
||
MAX_RETRIES = 3
|
||
RETRY_BACKOFF_BASE = 1.5 # seconds; exponential backoff + jitter
|
||
RETRY_BACKOFF_CAP = 60.0 # 退避上限：1.5*2^N 无封顶时 N=10 达 1536s，会卡死进程
|
||
|
||
# HTTP status codes worth retrying. 403 is included so fetch_url's UA-fallback
|
||
# loop can kick in when a site blocks the default searxng-cli User-Agent
|
||
# (a fresh UA is tried on each retry attempt). search.py also retries 403 —
|
||
# it doesn't switch UAs, so true auth failures waste ~3 attempts, accepted
|
||
# as a trade-off for one shared retry policy across both scripts.
|
||
RETRYABLE_STATUS = frozenset({403, 429, 502, 503, 504})
|
||
|
||
# Browser-like UA strings for fallback when the searxng-cli UA is blocked.
|
||
#
|
||
# v2.2.2：单一来源（SSOT）。UA 池唯一权威定义在 _config.UA_POOL，
|
||
# 此处直接导入——删除原有的内置副本 _FALLBACK_UAS_BUILTIN。
|
||
# v2.2.0 曾保留一份手工同步副本，两份列表漂移会导致跨脚本 UA 行为不一致
|
||
# （fetch.py 用 FALLBACK_UAS 轮换、search.py 的 --dry-run 报告引用同一池），
|
||
# 且注释要求"保持同步"无任何机制保证。直接引用同一对象后，改一处即全局生效。
|
||
#
|
||
# 维护原则（见 _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:
|
||
"""为域名确定性选择 UA 池索引。
|
||
|
||
用 ``hashlib.sha256`` 而非内置 ``hash()``，因为后者对字符串做了
|
||
随机化（PYTHONHASHSEED），跨进程不可复现。SHA-256 保证同一域名
|
||
永远映射到同一索引，跨进程一致——这对调试和日志分析至关重要。
|
||
"""
|
||
h = hashlib.sha256(domain.encode("utf-8")).digest()
|
||
# 取前 8 字节作为无符号整数，避免负数和短字符串的分布不均
|
||
return int.from_bytes(h[:8], "big") % pool_size
|
||
|
||
|
||
# Per-domain UA 缓存：同一域名 + 同一进程 = 同一 UA，避免会话内 UA 突变
|
||
# 被反爬识别。跨进程通过 SHA-256 哈希复现，见 _ua_index_for_domain。
|
||
_domain_ua_cache: dict = {}
|
||
|
||
# 保护 _domain_ua_cache 的"检查-设置"原子性锁。
|
||
# v2.2.0：并发场景下（如 search_multi 并行请求多实例），多个线程可能同时
|
||
# 检查 domain not in cache 并同时写入，虽不致命但会浪费计算且可能写入不同
|
||
# UA（因 hash 本应稳定，但极端时序下逻辑可读性问题）。用锁串行化 dict 读写。
|
||
# 注意：锁内只做 dict 读写，绝不放网络/重计算，避免阻塞其他线程。
|
||
_domain_ua_lock = threading.Lock()
|
||
|
||
|
||
def get_ua_for_domain(url: str, user_agent: str = None) -> str:
|
||
"""返回适合某域名的 User-Agent。
|
||
|
||
优先级：
|
||
1. ``user_agent`` 显式传入（CLI --user-agent）→ 直接返回
|
||
2. 该域名已缓存 → 返回缓存值
|
||
3. 域名未缓存 → 用 SHA-256 hash 选一个 FALLBACK_UAS，缓存并返回
|
||
|
||
设计理由：真实浏览器访问同一站点时 UA 永远不变。爬虫如果每次请求
|
||
换一个 UA，反而会被反爬系统标记为可疑。确定性轮换保证同一域名
|
||
稳定使用同一 UA，不同域名分散到不同 UA 上降低集体封禁风险。
|
||
|
||
线程安全：缓存的"检查-设置"用 ``_domain_ua_lock`` 保护。本函数无网络
|
||
调用，但遵循"锁内只做 dict 读写"原则——hash 计算放在锁外，写入时做
|
||
双检查（其他线程可能在此期间已写入），兼顾正确性与并发吞吐。
|
||
"""
|
||
if user_agent:
|
||
return user_agent
|
||
|
||
try:
|
||
domain = urllib.parse.urlparse(url).netloc.lower()
|
||
if not domain:
|
||
return FALLBACK_UAS[0]
|
||
except Exception:
|
||
return FALLBACK_UAS[0]
|
||
|
||
# 快速路径：锁内检查缓存命中
|
||
with _domain_ua_lock:
|
||
if domain in _domain_ua_cache:
|
||
return _domain_ua_cache[domain]
|
||
|
||
# 缓存未命中：在锁外计算 UA（SHA-256 hash，无副作用，不阻塞其他线程）
|
||
idx = _ua_index_for_domain(domain, len(FALLBACK_UAS))
|
||
ua = FALLBACK_UAS[idx]
|
||
|
||
# 加锁写入；双检查避免覆盖其他线程并发写入的值
|
||
with _domain_ua_lock:
|
||
if domain in _domain_ua_cache:
|
||
return _domain_ua_cache[domain]
|
||
_domain_ua_cache[domain] = ua
|
||
return ua
|
||
|
||
|
||
def reset_domain_ua_cache() -> None:
|
||
"""清空 per-domain UA 缓存。测试用。"""
|
||
with _domain_ua_lock:
|
||
_domain_ua_cache.clear()
|
||
|
||
|
||
def build_browser_headers(user_agent: str, referer: str = None,
|
||
accept_html: bool = True) -> dict:
|
||
"""构造完整的浏览器请求头，让请求看起来像真浏览器。
|
||
|
||
v2.0.0 核心反爬措施：仅靠 User-Agent 已无法绕过现代 WAF，
|
||
Cloudflare/Akamai/Imperva 都会检查 Sec-* 头和 Accept-Language。
|
||
|
||
Args:
|
||
user_agent: UA 字符串（应来自 get_ua_for_domain）
|
||
referer: Referer URL（可选；从搜索结果抓取时设为实例 URL）
|
||
accept_html: True 时 Accept 包含 text/html（页面抓取）；
|
||
False 时 Accept 为 application/json（API 调用）
|
||
|
||
Returns:
|
||
包含完整浏览器指纹的 headers dict。调用方需自行合并 auth_headers。
|
||
"""
|
||
is_firefox = "Firefox/" in user_agent
|
||
is_edge = "Edg/" in user_agent
|
||
|
||
if accept_html:
|
||
accept = ("text/html,application/xhtml+xml,application/xml;q=0.9,"
|
||
"image/avif,image/webp,*/*;q=0.8")
|
||
else:
|
||
accept = "application/json, text/plain, */*;q=0.8"
|
||
|
||
# v2.2.1 智能声明 Accept-Encoding：仅当本机安装了 brotli 解压器时
|
||
# 才声明 br。否则服务器返回 br 压缩字节而 requests 无法解压 → 乱码。
|
||
# Firefox UA 路径保持只声明 gzip/deflate（与真 Firefox 行为一致，
|
||
# Firefox 虽支持 br 但为减少指纹差异在此工具中不声明）。
|
||
if is_firefox:
|
||
accept_encoding = "gzip, deflate"
|
||
elif _HAS_BROTLI:
|
||
accept_encoding = "gzip, deflate, br"
|
||
else:
|
||
accept_encoding = "gzip, deflate"
|
||
|
||
headers = {
|
||
"User-Agent": user_agent,
|
||
"Accept": accept,
|
||
"Accept-Language": "en-US,en;q=0.9,zh-CN;q=0.8,zh;q=0.7",
|
||
"Accept-Encoding": accept_encoding,
|
||
"Connection": "keep-alive",
|
||
"Upgrade-Insecure-Requests": "1" if accept_html else "0",
|
||
}
|
||
|
||
# Sec-Ch-Ua 系列仅 Chrome/Edge 发送，Firefox 不发
|
||
if not is_firefox:
|
||
# 从 UA 提取主版本号，构造 Sec-Ch-Ua。
|
||
# v2.5.0 修复：原实现把已含引号的 not_a_brand 再包进 f-string 引号，
|
||
# 生成 ""Not_A Brand";v="99"" 的畸形头——Chrome/Edge 两条路径都中招，
|
||
# 严格校验 Sec-CH-UA 的 WAF（Cloudflare 等会校验与 UA 一致性）会直接
|
||
# 忽略或判定不一致。现改为品牌数组统一拼接，产出合法格式：
|
||
# Chrome: "Not_A Brand";v="99", "Chromium";v="138", "Google Chrome";v="138"
|
||
# Edge: 上述 + , "Microsoft Edge";v="138"
|
||
m = re.search(r"Chrome/(\d+)", user_agent)
|
||
ver = m.group(1) if m else "131"
|
||
not_a_brand = "Not_A Brand" if ver != "99" else "Not/A)Brand"
|
||
brands = [
|
||
f'"{not_a_brand}";v="99"',
|
||
f'"Chromium";v="{ver}"',
|
||
f'"Google Chrome";v="{ver}"',
|
||
]
|
||
if is_edge:
|
||
# Edge 额外声明 Microsoft Edge 品牌
|
||
brands.append(f'"Microsoft Edge";v="{ver}"')
|
||
headers["Sec-Ch-Ua"] = ", ".join(brands)
|
||
headers["Sec-Ch-Ua-Mobile"] = '"?1"' if "Mobile" in user_agent else '"?0"'
|
||
# 平台标识
|
||
if "Windows" in user_agent:
|
||
headers["Sec-Ch-Ua-Platform"] = '"Windows"'
|
||
elif "Macintosh" in user_agent:
|
||
headers["Sec-Ch-Ua-Platform"] = '"macOS"'
|
||
elif "Linux" in user_agent:
|
||
headers["Sec-Ch-Ua-Platform"] = '"Linux"'
|
||
# Sec-Fetch 系列（Chrome 76+ 全量发送）
|
||
if accept_html:
|
||
headers["Sec-Fetch-Site"] = "none" if not referer else "cross-site"
|
||
headers["Sec-Fetch-Mode"] = "navigate"
|
||
headers["Sec-Fetch-User"] = "?1"
|
||
headers["Sec-Fetch-Dest"] = "document"
|
||
else:
|
||
headers["Sec-Fetch-Site"] = "same-origin" if referer else "none"
|
||
headers["Sec-Fetch-Mode"] = "cors"
|
||
headers["Sec-Fetch-Dest"] = "empty"
|
||
|
||
if referer:
|
||
headers["Referer"] = referer
|
||
|
||
return headers
|
||
|
||
|
||
def parse_retry_after(header_value: str) -> float:
|
||
"""解析 Retry-After header，返回应等待的秒数。
|
||
|
||
HTTP 规范允许两种格式：
|
||
1. 纯数字：秒数（最常见）
|
||
2. HTTP date：绝对时间（如 ``Wed, 21 Oct 2026 07:28:00 GMT``）
|
||
|
||
返回 0.0 表示无需等待或解析失败。对 HTTP date 格式，若已过期
|
||
也返回 0.0（让调用方立即重试）。
|
||
"""
|
||
if not header_value:
|
||
return 0.0
|
||
header_value = header_value.strip()
|
||
|
||
# 格式 1：纯数字秒数
|
||
try:
|
||
seconds = float(header_value)
|
||
return max(0.0, seconds)
|
||
except ValueError:
|
||
pass
|
||
|
||
# 格式 2：HTTP date
|
||
try:
|
||
from email.utils import parsedate_to_datetime
|
||
from datetime import datetime, timezone
|
||
dt = parsedate_to_datetime(header_value)
|
||
if dt is None:
|
||
return 0.0
|
||
# 确保 timezone-aware
|
||
if dt.tzinfo is None:
|
||
dt = dt.replace(tzinfo=timezone.utc)
|
||
now = datetime.now(timezone.utc)
|
||
delta = (dt - now).total_seconds()
|
||
return max(0.0, delta)
|
||
except (TypeError, ValueError, OverflowError):
|
||
return 0.0
|
||
|
||
|
||
def compute_backoff_delay(attempt: int, base: float = RETRY_BACKOFF_BASE,
|
||
cap: float = RETRY_BACKOFF_CAP) -> float:
|
||
"""计算退避延迟，带封顶和抖动。
|
||
|
||
``base * 2^attempt + jitter``，但不超过 ``cap``。
|
||
v2.0.0 新增封顶：原公式无上限，N=10 时达 1536s 会卡死进程。
|
||
"""
|
||
delay = base * (2 ** attempt) + random.uniform(0, 1)
|
||
return min(delay, cap)
|
||
|
||
|
||
def build_auth_headers(bearer_token: str = None, basic_auth: str = None) -> dict:
|
||
"""Build an Authorization header dict from CLI auth flags.
|
||
|
||
``bearer_token``: raw Bearer token string.
|
||
``basic_auth``: ``"username:password"`` string (base64-encoded).
|
||
|
||
If both are provided, Bearer takes precedence (more common for APIs).
|
||
Returns a dict to merge into request headers, or an empty dict.
|
||
"""
|
||
headers = {}
|
||
if bearer_token:
|
||
headers["Authorization"] = f"Bearer {bearer_token}"
|
||
elif basic_auth:
|
||
encoded = base64.b64encode(basic_auth.encode("utf-8")).decode("ascii")
|
||
headers["Authorization"] = f"Basic {encoded}"
|
||
return headers
|
||
|
||
|
||
def _warn_file_perms(path: str) -> None:
|
||
"""Warn if a credentials file is readable by group/other (POSIX only).
|
||
|
||
On Windows the Unix permission bits in ``st_mode`` do not reflect the
|
||
actual ACL, so the check is skipped to avoid false alarms.
|
||
"""
|
||
if os.name != "posix":
|
||
return
|
||
log = logging.getLogger("searxng.common")
|
||
try:
|
||
mode = os.stat(path).st_mode & 0o777
|
||
if mode & 0o077:
|
||
log.warning(
|
||
f"Warning: credentials file '{path}' has permissions {oct(mode)} "
|
||
f"(accessible by group/other); recommend 'chmod 600' for security."
|
||
)
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
def resolve_auth_basic(cli_value: str = None, file_path: str = None,
|
||
env_var: str = "SEARXNG_BASIC_AUTH",
|
||
config_value: str = None) -> str:
|
||
"""Resolve basic-auth credentials without leaking them via shell history.
|
||
|
||
Priority (highest wins):
|
||
1. ``cli_value`` — explicit ``--auth-basic "user:pass"`` (convenient
|
||
but leaks into shell history; discouraged)
|
||
2. ``file_path`` — ``--auth-basic-file FILE``; first non-empty line
|
||
is read as ``user:pass``. Recommended for shells.
|
||
3. ``config_value`` — ``auth_basic`` field from ``searxng.toml``.
|
||
Convenient for AI agents that read config once.
|
||
4. ``env_var`` — ``SEARXNG_BASIC_AUTH`` environment variable.
|
||
|
||
Returns ``"user:pass"`` or ``None`` if no source provides credentials.
|
||
Raises ``RuntimeError`` if a file is specified but cannot be read.
|
||
"""
|
||
if cli_value:
|
||
return cli_value
|
||
|
||
if file_path:
|
||
try:
|
||
text = Path(file_path).read_text(encoding="utf-8")
|
||
_warn_file_perms(file_path)
|
||
for line in text.splitlines():
|
||
line = line.strip()
|
||
if line and not line.startswith("#"):
|
||
return line
|
||
raise RuntimeError(f"auth file '{file_path}' contains no credentials")
|
||
except OSError as e:
|
||
raise RuntimeError(f"cannot read auth file '{file_path}': {e}") from e
|
||
|
||
if config_value:
|
||
return config_value
|
||
|
||
return os.environ.get(env_var)
|
||
|
||
|
||
def resolve_auth_bearer(cli_value: str = None, file_path: str = None,
|
||
env_var: str = "SEARXNG_BEARER_TOKEN",
|
||
config_value: str = None) -> str:
|
||
"""Resolve a Bearer token from CLI flag, file, config, or environment.
|
||
|
||
Mirrors :func:`resolve_auth_basic` for token-style auth. Useful for
|
||
long-lived API tokens that should not appear in shell history.
|
||
|
||
Priority (highest wins):
|
||
1. ``cli_value`` — explicit ``--auth-bearer "token"``
|
||
2. ``file_path`` — ``--auth-bearer-file FILE``
|
||
3. ``config_value`` — ``auth_bearer`` field from ``searxng.toml``
|
||
4. ``env_var`` — ``SEARXNG_BEARER_TOKEN`` environment variable
|
||
"""
|
||
if cli_value:
|
||
return cli_value
|
||
|
||
if file_path:
|
||
try:
|
||
text = Path(file_path).read_text(encoding="utf-8")
|
||
_warn_file_perms(file_path)
|
||
for line in text.splitlines():
|
||
line = line.strip()
|
||
if line and not line.startswith("#"):
|
||
return line
|
||
raise RuntimeError(f"token file '{file_path}' contains no token")
|
||
except OSError as e:
|
||
raise RuntimeError(f"cannot read token file '{file_path}': {e}") from e
|
||
|
||
if config_value:
|
||
return config_value
|
||
|
||
return os.environ.get(env_var)
|
||
|
||
|
||
def apply_proxy(proxy_url: str) -> None:
|
||
"""Configure proxy via environment variables.
|
||
|
||
Sets ``HTTP_PROXY`` and ``HTTPS_PROXY`` so both urllib (which reads
|
||
them via :func:`urllib.request.getproxies`) and ``requests`` (which
|
||
honors them when ``trust_env=True``, the default) pick up the proxy
|
||
without any changes to call sites.
|
||
|
||
``NO_PROXY`` is set to ``localhost,127.0.0.1,::1`` (if not already set)
|
||
so local traffic stays direct — matters for self-hosted SearXNG on
|
||
localhost behind a corporate proxy.
|
||
|
||
Pass an empty string to clear the proxy env vars (rarely needed; the
|
||
default unset state already means "no proxy").
|
||
"""
|
||
if not proxy_url:
|
||
return
|
||
os.environ["HTTP_PROXY"] = proxy_url
|
||
os.environ["HTTPS_PROXY"] = proxy_url
|
||
# Keep local traffic direct unless the user has explicitly set NO_PROXY
|
||
os.environ.setdefault("NO_PROXY", "localhost,127.0.0.1,::1")
|
||
|
||
|
||
def detect_charset(raw: bytes, content_type: str) -> str:
|
||
"""Detect charset from the Content-Type header, then an HTML meta tag.
|
||
|
||
Falls back to UTF-8 (with replacement) if nothing reliable is found.
|
||
"""
|
||
# 1. HTTP header
|
||
if "charset=" in content_type:
|
||
charset = content_type.split("charset=")[-1].split(";")[0].strip()
|
||
try:
|
||
raw.decode(charset)
|
||
return charset
|
||
except (UnicodeDecodeError, LookupError):
|
||
pass
|
||
|
||
# 2. HTML <meta charset> or <meta http-equiv>
|
||
try:
|
||
head = raw[:4096].decode("ascii", errors="replace")
|
||
m = re.search(r'<meta[^>]+charset=["\']?([a-zA-Z0-9_-]+)', head, re.IGNORECASE)
|
||
if m:
|
||
charset = m.group(1).strip()
|
||
try:
|
||
raw.decode(charset)
|
||
return charset
|
||
except (UnicodeDecodeError, LookupError):
|
||
pass
|
||
except Exception:
|
||
pass
|
||
|
||
# 3. Fallback: UTF-8 with replacement
|
||
return "utf-8"
|
||
|
||
|
||
def is_retryable_error(exc: BaseException) -> bool:
|
||
"""Return True if ``exc`` is a transient error worth retrying.
|
||
|
||
Handles both the stdlib ``urllib`` errors and ``requests`` errors via
|
||
duck-typing (so this module does not need to import ``requests``):
|
||
|
||
* ``urllib.error.HTTPError`` → retry iff status in ``RETRYABLE_STATUS``
|
||
* ``urllib.error.URLError`` / ``OSError`` / ``TimeoutError`` → retry
|
||
(connection refused, DNS failure, timeout — all transient)
|
||
* ``requests.exceptions.HTTPError`` → retry iff ``response.status_code``
|
||
is in ``RETRYABLE_STATUS``
|
||
* ``requests`` connection/timeout errors (no ``.response``) → retry
|
||
"""
|
||
if isinstance(exc, urllib.error.HTTPError):
|
||
return exc.code in RETRYABLE_STATUS
|
||
if isinstance(exc, (urllib.error.URLError, OSError, TimeoutError)):
|
||
return not isinstance(exc, urllib.error.HTTPError)
|
||
|
||
# requests.exceptions.HTTPError / RequestException (duck-typed)
|
||
resp = getattr(exc, "response", None)
|
||
status = getattr(resp, "status_code", None)
|
||
if status is not None:
|
||
return status in RETRYABLE_STATUS
|
||
if resp is None:
|
||
# requests connection/timeout error without a response -> transient
|
||
return True
|
||
return False
|
||
|
||
|
||
# ----- Structured error classification -----
|
||
#
|
||
# 错误码体系：让 AI Agent 程序化地判断错误类型并采取恢复策略。
|
||
# 所有错误码以 E_ 前缀，在 --format json 模式下随 error_code 字段输出。
|
||
#
|
||
# AI 可根据 error_code 决策：
|
||
# E_CONFIG → 检查实例配置/环境变量，提示用户设置
|
||
# E_AUTH → 检查 token/凭证，提示用户重新认证
|
||
# E_NETWORK → 重试或切换实例/代理
|
||
# E_RATE_LIMIT → 等待后重试，降低请求频率
|
||
# E_PARSE → 检查实例是否支持 JSON，尝试 HTML 回退
|
||
# E_EMPTY → 调整查询词或时间范围
|
||
# E_INPUT → 修正参数/文件路径
|
||
# E_INTERNAL → 报告 bug，附带完整错误信息
|
||
|
||
# 错误码常量（供 search.py / fetch.py 引用）
|
||
E_CONFIG = "E_CONFIG"
|
||
E_AUTH = "E_AUTH"
|
||
E_BLOCKED = "E_BLOCKED" # fetch 被站点反爬拦截（403），非凭证问题
|
||
E_NETWORK = "E_NETWORK"
|
||
E_RATE_LIMIT = "E_RATE_LIMIT"
|
||
E_PARSE = "E_PARSE"
|
||
E_EMPTY = "E_EMPTY"
|
||
E_INPUT = "E_INPUT"
|
||
E_INTERNAL = "E_INTERNAL"
|
||
|
||
# 每个 E_* 配套的可操作恢复建议，让 AI Agent 能自决策下一步动作，
|
||
# 而不是盲目重试或放弃。在 _emit_error 的 JSON 输出中作为 recovery_hint 字段。
|
||
RECOVERY_HINTS = {
|
||
E_CONFIG: "Provide -i/--instance, set SEARXNG_INSTANCE env var, or create "
|
||
"searxng.toml/instances.txt config file.",
|
||
E_AUTH: "Verify --auth-bearer/--auth-basic credentials or "
|
||
"SEARXNG_BEARER_TOKEN/SEARXNG_BASIC_AUTH env vars. Check token "
|
||
"expiry and instance access permissions.",
|
||
E_BLOCKED: "The site is blocking automated access (WAF / anti-bot / "
|
||
"geo-block) — this is NOT a credentials problem; no auth "
|
||
"change will help. Use the Wayback Machine fallback (on by "
|
||
"default; --no-fallback disables), fetch a mirror or a "
|
||
"different URL, or drop the domain with --exclude-domain.",
|
||
E_NETWORK: "Retry with backoff, or try a different SearXNG instance. "
|
||
"Check network connectivity, proxy settings, and instance uptime.",
|
||
E_RATE_LIMIT: "Wait before retrying (exponential backoff). Reduce query "
|
||
"frequency, narrow --time-range, or distribute load across "
|
||
"multiple instances.",
|
||
E_INPUT: "Check query syntax, --categories values, --time-range format, "
|
||
"and flag combinations. Use --help for valid options.",
|
||
E_PARSE: "The instance returned malformed data. Try a different instance, "
|
||
"switch --method, or check if the instance version is compatible.",
|
||
E_EMPTY: "Refine the query (more specific terms), broaden --time-range, "
|
||
"add --categories, or increase --pageno to find more results.",
|
||
E_INTERNAL: "This is likely a bug. Re-run with --verbose and report the "
|
||
"full output for diagnosis.",
|
||
}
|
||
|
||
|
||
def _classify_by_type_and_status(exc) -> str:
|
||
"""根据异常类型和 HTTP 状态码分类，返回错误码或 None。
|
||
|
||
检查顺序：HTTP 状态码 → 文件错误 → 连接错误 → 解析错误。
|
||
返回 None 表示该异常无法靠类型/状态码判定，需走字符串 fallback。
|
||
不处理 RuntimeError 消息推断（由 :func:`classify_error` 调用方做 fallback）。
|
||
|
||
抽取为独立函数，便于 :func:`classify_error` 对 ``exc`` 本身和其
|
||
``__cause__`` 复用同一套基于真实类型的判定逻辑。
|
||
"""
|
||
# HTTP 错误（urllib HTTPError 用 .code，requests HTTPError 用 .response.status_code）
|
||
status = None
|
||
if isinstance(exc, urllib.error.HTTPError):
|
||
status = exc.code
|
||
else:
|
||
resp = getattr(exc, "response", None)
|
||
status = getattr(resp, "status_code", None)
|
||
|
||
if status is not None:
|
||
if status == 429:
|
||
return E_RATE_LIMIT
|
||
if status in (401, 403):
|
||
return E_AUTH
|
||
if 400 <= status < 500:
|
||
return E_INPUT
|
||
if 500 <= status < 600:
|
||
return E_NETWORK
|
||
|
||
# 文件/输入错误（FileNotFoundError 是 OSError 子类，必须先于 OSError 检查）
|
||
if isinstance(exc, FileNotFoundError):
|
||
return E_INPUT
|
||
|
||
# 连接级错误
|
||
if isinstance(exc, (urllib.error.URLError, OSError, TimeoutError)):
|
||
return E_NETWORK
|
||
if isinstance(exc, ConnectionError):
|
||
return E_NETWORK
|
||
|
||
# 解析错误
|
||
if isinstance(exc, (ValueError, json.JSONDecodeError)):
|
||
return E_PARSE
|
||
|
||
return None
|
||
|
||
|
||
def classify_error(exc: BaseException) -> str:
|
||
"""将异常分类为结构化错误码，供 AI Agent 程序化处理。
|
||
|
||
分类逻辑（按优先级）：
|
||
1. 优先检查异常链 ``__cause__``：``raise X from Y`` 场景下 Y 才是真实
|
||
错误源（如 ``raise RuntimeError(...) from HTTPError(403)``），
|
||
用 Y 的类型/状态码分类比从 X 的消息字符串推断更可靠、不再脆弱。
|
||
2. HTTP 状态码：429→E_RATE_LIMIT, 401/403→E_AUTH, 4xx→E_INPUT, 5xx→E_NETWORK
|
||
3. 连接类异常（ConnectionError/TimeoutError/URLError/OSError）→ E_NETWORK
|
||
4. 解析错误（ValueError/JSONDecodeError）→ E_PARSE
|
||
5. FileNotFoundError → E_INPUT
|
||
6. RuntimeError：从消息中推断（仅当 ``__cause__`` 缺失时的最后 fallback，
|
||
兼容旧路径——search_multi 把 last_error 拼进消息）
|
||
7. 其他 → E_INTERNAL
|
||
"""
|
||
# 优先检查异常链 __cause__：raise X from Y 时，Y 指向真实底层异常。
|
||
# 旧逻辑只能从外层 RuntimeError 的消息字符串推断（脆弱，依赖 "403"/"auth"
|
||
# 等关键字匹配），新逻辑直接从 __cause__ 的 .code/.status_code 或异常
|
||
# 类型判定。仅检查一层 __cause__，不递归——单层已覆盖 search_multi 的
|
||
# raise-from 模式，深层链罕见且递归有循环风险。
|
||
cause = getattr(exc, "__cause__", None)
|
||
if cause is not None and cause is not exc:
|
||
code = _classify_by_type_and_status(cause)
|
||
if code is not None:
|
||
return code
|
||
|
||
# 检查 exc 本身的类型和状态码
|
||
code = _classify_by_type_and_status(exc)
|
||
if code is not None:
|
||
return code
|
||
|
||
# RuntimeError：从消息中推断（search_multi 的 "All N instances failed.
|
||
# Last error: HTTP Error 403: Forbidden" 等）。并行路径现在会把 last_error
|
||
# 拼进消息，让这里能提取真实错误类型，而不是一律误判 E_NETWORK。
|
||
# 字符串匹配仅作为 __cause__ 缺失时的最后 fallback。
|
||
if isinstance(exc, RuntimeError):
|
||
msg = str(exc).lower()
|
||
# 先检查 auth/rate-limit 关键字（最常见，来自 last_error 详情）
|
||
if "auth" in msg or "403" in msg or "401" in msg:
|
||
return E_AUTH
|
||
if "rate" in msg or "429" in msg:
|
||
return E_RATE_LIMIT
|
||
# 从 "http error NNN" / "http NNN" 模式中提取状态码，
|
||
# 正确分类 "All instances failed. Last error: HTTP Error 403" 等
|
||
status_match = re.search(r'http(?: error)? (\d{3})', msg)
|
||
if status_match:
|
||
status = int(status_match.group(1))
|
||
if status == 429:
|
||
return E_RATE_LIMIT
|
||
if status in (401, 403):
|
||
return E_AUTH
|
||
if 400 <= status < 500:
|
||
return E_INPUT
|
||
if 500 <= status < 600:
|
||
return E_NETWORK
|
||
# 所有实例失败的通用模式（无更具体的 HTTP 状态码时才判为网络错误）
|
||
if "all" in msg and "instance" in msg and "fail" in msg:
|
||
return E_NETWORK
|
||
if "parse" in msg or "json" in msg or "html" in msg:
|
||
return E_PARSE
|
||
if "not found" in msg or ("no " in msg and "instance" in msg):
|
||
return E_CONFIG
|
||
return E_INTERNAL
|
||
|
||
return E_INTERNAL
|
||
|
||
|
||
def _extract_status_code(exc) -> int:
|
||
"""从异常及其 ``__cause__`` 链中提取 HTTP 状态码（无则返回 None）。
|
||
|
||
检查顺序：urllib ``.code`` → requests ``.response.status_code`` →
|
||
消息中的 ``HTTP NNN`` 模式（fetch_url 抛出的 RuntimeError 消息）。
|
||
"""
|
||
target = getattr(exc, "__cause__", None) or exc
|
||
status = getattr(target, "code", None)
|
||
if status is None:
|
||
resp = getattr(target, "response", None)
|
||
status = getattr(resp, "status_code", None)
|
||
if status is None:
|
||
m = re.search(r"HTTP (\d{3})", str(exc))
|
||
if m:
|
||
status = int(m.group(1))
|
||
return status
|
||
|
||
|
||
def classify_fetch_error(exc: BaseException, status_code: int = None) -> str:
|
||
"""fetch 场景的错误分类：403 → E_BLOCKED，401 → E_AUTH，其余委托 classify_error。
|
||
|
||
网页抓取（fetch.py / search.py --fetch）遇到的 403 绝大多数是站点反爬
|
||
拦截（UA/JS 指纹、WAF、区域封锁），**不是凭证错误**。classify_error
|
||
把 401/403 统一归为 E_AUTH 是为 SearXNG 实例认证设计的——若 fetch 也
|
||
用它，AI Agent 会误判为"需要检查凭证"而去做无效的认证重试。本函数在
|
||
classify_error 基础上仅做 fetch 场景的细分。
|
||
|
||
``status_code`` 可选：调用方已从异常链提取时可直接传入，避免重复解析。
|
||
"""
|
||
if status_code is None:
|
||
status_code = _extract_status_code(exc)
|
||
if status_code == 403:
|
||
return E_BLOCKED
|
||
if status_code == 401:
|
||
return E_AUTH
|
||
return classify_error(exc)
|
||
|
||
|
||
# ----- Progress event emitter (for --progress flag) -----
|
||
#
|
||
# 当 --progress 启用时，search.py 会调用 emit_progress() 发射结构化事件到
|
||
# stderr（JSON Lines 格式）。AI Agent 可解析这些事件来跟踪执行进度。
|
||
#
|
||
# 事件类型：
|
||
# {"event": "start", "query": "...", "instances": N}
|
||
# {"event": "instance_try", "url": "...", "attempt": 1}
|
||
# {"event": "instance_ok", "url": "...", "latency": 0.5, "results": 10}
|
||
# {"event": "instance_fail", "url": "...", "error": "...", "error_code": "E_*"}
|
||
# {"event": "cache_hit", "query": "...", "ttl": 30}
|
||
# {"event": "cache_store", "query": "...", "ttl": 30}
|
||
# {"event": "fetch_start", "count": 3}
|
||
# {"event": "fetch_ok", "url": "...", "chars": 1234}
|
||
# {"event": "fetch_fail", "url": "...", "error": "..."}
|
||
# {"event": "done", "results": N, "query": "..."}
|
||
# {"event": "error", "error": "...", "error_code": "E_*", "query": "..."}
|
||
|
||
_progress_enabled = False
|
||
|
||
|
||
def set_progress_enabled(enabled: bool) -> None:
|
||
"""全局开关：是否向 stderr 输出 JSON Lines 格式的进度事件。"""
|
||
global _progress_enabled
|
||
_progress_enabled = enabled
|
||
|
||
|
||
def emit_progress(event: str, **kwargs) -> None:
|
||
"""向 stderr 输出一行 JSON 格式的进度事件。
|
||
|
||
仅当 --progress 启用时才输出。``event`` 是事件类型字符串，
|
||
``kwargs`` 是事件的附加字段。输出格式为 JSON Lines（每行一个 JSON 对象）。
|
||
v2.2.0：自动注入 request_id（如果已设置）。
|
||
"""
|
||
if not _progress_enabled:
|
||
return
|
||
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)
|
||
|
||
|
||
# ----- 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([
|
||
"www.baidu.com", # 百度搜索：强反爬 + Cookie 检测
|
||
"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", # 豆瓣电影
|
||
])
|
||
|
||
# 需要子域匹配的域名（如 *.zhihu.com, *.weibo.com, *.douban.com）
|
||
# 注意：baidu.com 不在此列——其反爬子域（www/baike/zhidao/wenku/tieba）已在
|
||
# 上方精确列表中，而 pan.baidu.com（网盘）/cloud.baidu.com（智能云）等
|
||
# 子域可正常抓取，整体匹配会误伤。zhihu/weibo/douban 的子域基本都被墙。
|
||
_SUBDOMAIN_BLOCKED = frozenset([
|
||
"zhihu.com",
|
||
"weibo.com",
|
||
"douban.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:
|
||
host = urllib.parse.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
|
||
|
||
|
||
# ----- 相似度去重（v2.x：让 AI Agent 在研究模式下获得更干净的 merged_results）-----
|
||
# 同一内容在不同 URL/引擎下常重复出现，仅靠 URL 去重无法合并。
|
||
# SimHash 对标题做局部敏感哈希，汉明距离小的视为近似重复。
|
||
|
||
def _normalize_title(title: str) -> str:
|
||
"""归一化标题：小写、去标点、去多余空格，用于相似度比较。"""
|
||
if not title:
|
||
return ""
|
||
s = title.lower()
|
||
# 保留字母、数字、CJK 和空格，其余替换为空格
|
||
s = re.sub(r'[^\w\s]', ' ', s)
|
||
# \w 包含下划线，单独去掉
|
||
s = s.replace('_', ' ')
|
||
s = re.sub(r'\s+', ' ', s).strip()
|
||
return s
|
||
|
||
|
||
def _simhash(text: str, hash_bits: int = 64) -> int:
|
||
"""计算文本的 SimHash 指纹。
|
||
- 分词（按空格 + CJK 单字符）
|
||
- 每个 token 算普通 hash，按 bit 投票
|
||
- 返回 hash_bits 位的指纹
|
||
"""
|
||
if not text:
|
||
return 0
|
||
# 分词：按空格切分，CJK 字符再逐个拆成单字 token
|
||
tokens = []
|
||
for word in text.split():
|
||
buf = []
|
||
for ch in word:
|
||
if '\u4e00' <= ch <= '\u9fff':
|
||
# 遇到 CJK：先冲出缓冲区里的非 CJK 片段，再加入单字
|
||
if buf:
|
||
tokens.append(''.join(buf))
|
||
buf = []
|
||
tokens.append(ch)
|
||
else:
|
||
buf.append(ch)
|
||
if buf:
|
||
tokens.append(''.join(buf))
|
||
if not tokens:
|
||
return 0
|
||
# 每个 token 算 SHA-256（跨进程可复现，避免 hash() 随机化），按 bit 投票
|
||
v = [0] * hash_bits
|
||
for token in tokens:
|
||
h = hashlib.sha256(token.encode('utf-8')).digest()
|
||
token_hash = int.from_bytes(h[:8], 'big')
|
||
for i in range(hash_bits):
|
||
if (token_hash >> i) & 1:
|
||
v[i] += 1
|
||
else:
|
||
v[i] -= 1
|
||
# 投票为正的位置 1
|
||
fingerprint = 0
|
||
for i in range(hash_bits):
|
||
if v[i] > 0:
|
||
fingerprint |= (1 << i)
|
||
return fingerprint
|
||
|
||
|
||
def _hamming_distance(a: int, b: int) -> int:
|
||
"""两个整数的汉明距离。"""
|
||
return bin(a ^ b).count('1')
|
||
|
||
|
||
def _jaccard_similarity(set_a: set, set_b: set) -> float:
|
||
"""Jaccard 相似度。"""
|
||
if not set_a and not set_b:
|
||
return 0.0
|
||
union = set_a | set_b
|
||
if not union:
|
||
return 0.0
|
||
return len(set_a & set_b) / len(union)
|
||
|
||
|
||
def is_similar(result_a: dict, result_b: dict, threshold: float = 0.85) -> bool:
|
||
"""判断两个搜索结果是否相似。
|
||
- 优先用标题 SimHash（汉明距离 <= 3 视为相似，对应 64 位中约 95% 相似）
|
||
- 标题太短（< 5 字符）时用 URL 域名 + 标题 Jaccard
|
||
- threshold 参数控制严格程度
|
||
"""
|
||
title_a = _normalize_title(result_a.get("title", ""))
|
||
title_b = _normalize_title(result_b.get("title", ""))
|
||
# 标题太短时 SimHash 不稳定，改用 Jaccard
|
||
if len(title_a) < 5 or len(title_b) < 5:
|
||
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
|
||
# 标题足够长：用 SimHash 汉明距离
|
||
hash_a = _simhash(title_a)
|
||
hash_b = _simhash(title_b)
|
||
# threshold → 汉明距离阈值映射：
|
||
# 0.85 → 3（默认，宽松），0.90 → 2，0.95 → 1，1.0 → 0（几乎完全相同）
|
||
max_distance = max(0, min(3, int(round((1.0 - threshold) / 0.05))))
|
||
return _hamming_distance(hash_a, hash_b) <= max_distance
|
||
|
||
|
||
def texts_are_similar(text_a: str, text_b: str, threshold: float = 0.85) -> bool:
|
||
"""判断两段正文是否近似重复（v2.5.0，供抓取正文去重用）。
|
||
|
||
镜像/转载站点的正文与原文高度相似但 URL 不同——URL 级去重无法合并，
|
||
需要按正文指纹判定。复用与 :func:`is_similar` 相同的
|
||
SimHash + 汉明距离逻辑，但针对**正文**而非标题：
|
||
|
||
* 归一化（小写/去标点/CJK 单字分词）后取前 1000 字符做指纹窗口——
|
||
镜像页前 1000 字符通常一致，长文全量计算只会稀释指纹且增加开销
|
||
* 空文本（len<窗口）返回 False，避免短错误文本误判为重复
|
||
* threshold → 汉明距离阈值映射与 is_similar 一致（0.85→3 ... 1.0→0）
|
||
"""
|
||
norm_a = _normalize_title(text_a)[:1000]
|
||
norm_b = _normalize_title(text_b)[:1000]
|
||
if len(norm_a) < 5 or len(norm_b) < 5:
|
||
return False
|
||
hash_a = _simhash(norm_a)
|
||
hash_b = _simhash(norm_b)
|
||
max_distance = max(0, min(3, int(round((1.0 - threshold) / 0.05))))
|
||
return _hamming_distance(hash_a, hash_b) <= max_distance
|
||
|