数量也是正文信号
p_count = len(node.find_all("p"))
score = density + (p_count * 10)
if score > best_score:
best_score = score
best_node = node
return best_node
def extract_text(html_content: str) -> str:
"""Extract readable text from HTML, preferring bs4 if available."""
if _HAS_BS4:
try:
return extract_with_bs4(html_content)
except Exception:
pass
return extract_with_stdlib(html_content)
# ----- Tree-based HTML-to-Markdown converter (robust) -----
class MarkdownConverter(HTMLParser):
"""Convert HTML to Markdown using a tag-stack approach.
Handles nested tags in elements correctly unlike regex. Supports:
* GFM tables (``
`` → ``| a | b |`` with separator row)
* fenced code blocks (```` → triple-backtick fences)
* inline code (```` → backticks)
* blockquotes (```` → ``> `` prefix per line)
* headings, lists, images, emphasis, links
"""
def __init__(self):
super().__init__()
self._output = []
self._skip_depth = 0
self._list_stack = [] # list of [type, counter] for ol numbering
self._pending_indent = "" # preserved leading indent for li/dd
self._block_buffer = ""
self._link_href = None
self._link_text = []
self._in_link = False
self._in_pre = False
self._pre_content = []
self._heading_level = 0
self._block_empty = True
# Blockquote support
self._in_blockquote = False
# Table support (GFM)
self._in_table = False
self._table_rows = [] # list of (cells_list, is_header)
self._current_row = None
self._current_cell = None
self._in_cell = False
def _append_inline(self, text: str):
"""Append inline markup, routing to cell/link/block buffer.
In cell context, fragments go to ``_current_cell`` so table cell text
accumulates correctly. In link context, they go to ``_link_text``
and are collapsed at ```` time. Otherwise they append to the
block buffer; whitespace is normalized at ``_flush_block`` time.
"""
if self._in_cell:
self._current_cell.append(text)
elif self._in_link:
self._link_text.append(text)
else:
self._block_buffer += text
self._block_empty = False
def _flush_block(self):
t = self._block_buffer
# Normalize whitespace: collapse runs of spaces/tabs/newlines
t = re.sub(r'\s+', ' ', t).strip()
if t:
if self._in_blockquote:
# Prefix each line with "> " for markdown blockquote syntax
t = "\n".join(
("> " + line) if line.strip() else ">"
for line in t.split("\n")
)
if self._pending_indent:
self._output.append(self._pending_indent + t)
else:
self._output.append(t)
self._block_buffer = ""
self._block_empty = True
self._pending_indent = ""
def _emit_table(self):
"""Emit a GFM table from accumulated rows.
The first row becomes the header; a ``| --- | --- |`` separator row
follows; remaining rows become the body. Cells are padded to the
header width so the table renders correctly in strict GFM parsers.
"""
if not self._table_rows:
return
header_row, _ = self._table_rows[0]
body_rows = self._table_rows[1:]
if not header_row:
return
ncols = len(header_row)
self._output.append("| " + " | ".join(header_row) + " |")
self._output.append("| " + " | ".join("---" for _ in range(ncols)) + " |")
for row, _ in body_rows:
# Pad short rows; truncate long rows to header width
while len(row) < ncols:
row.append("")
self._output.append("| " + " | ".join(row[:ncols]) + " |")
self._output.append("")
def handle_starttag(self, tag, attrs):
tag_lower = tag.lower()
attrs_dict = dict(attrs)
if tag_lower in ("script", "style", "nav", "footer", "header",
"noscript", "iframe", "svg", "canvas", "template"):
self._skip_depth += 1
return
if self._skip_depth > 0:
self._skip_depth += 1
return
# Table structural elements are handled up-front so cell content
# routing (via _in_cell) takes effect before any other tag handler.
if tag_lower == "table":
self._flush_block()
self._in_table = True
self._table_rows = []
return
if self._in_table:
if tag_lower == "tr":
self._current_row = []
return
elif tag_lower in ("th", "td"):
self._in_cell = True
self._current_cell = []
return
elif tag_lower in ("thead", "tbody", "tfoot"):
return # container only — rows/cells drive the output
# Other tags inside cells (a/strong/em/code/br) fall through
# to normal handling; _append_inline routes them to _current_cell.
if tag_lower in ("p", "div", "section"):
self._flush_block()
elif tag_lower == "blockquote":
self._flush_block()
self._in_blockquote = True
elif tag_lower == "br":
self._append_inline("\n")
elif tag_lower in ("h1", "h2", "h3", "h4", "h5", "h6"):
self._flush_block()
self._heading_level = int(tag_lower[1])
elif tag_lower == "pre":
self._flush_block()
self._in_pre = True
self._pre_content = []
elif tag_lower == "code":
# Inline code. is handled by the pre path, which
# captures raw text verbatim — so only emit backticks when
# we're NOT inside a pre block.
if not self._in_pre:
self._append_inline("`")
elif tag_lower in ("ul", "ol"):
self._list_stack.append([tag_lower, 0])
elif tag_lower == "li":
self._flush_block()
depth = max(0, len(self._list_stack) - 1)
self._pending_indent = " " * depth
if self._list_stack and self._list_stack[-1][0] == "ol":
self._list_stack[-1][1] += 1
marker = f"{self._list_stack[-1][1]}. "
else:
marker = "- "
self._block_buffer = marker
self._block_empty = False
elif tag_lower == "dt":
self._flush_block()
self._block_buffer = "**"
self._block_empty = False
elif tag_lower == "dd":
self._flush_block()
self._pending_indent = " "
self._block_buffer = ""
self._block_empty = True
elif tag_lower == "a":
self._in_link = True
self._link_href = attrs_dict.get("href", "")
self._link_text = []
elif tag_lower in ("strong", "b"):
self._append_inline("**")
elif tag_lower in ("em", "i"):
self._append_inline("*")
elif tag_lower == "img":
alt = attrs_dict.get("alt", "")
src = attrs_dict.get("src", "")
if alt or src:
self._append_inline(f"")
def handle_endtag(self, tag):
tag_lower = tag.lower()
if self._skip_depth > 0:
self._skip_depth -= 1
return
if tag_lower == "pre":
self._in_pre = False
code = "\n".join(self._pre_content)
self._output.append(f"```\n{code}\n```")
self._pre_content = []
self._block_empty = True
elif tag_lower in ("h1", "h2", "h3", "h4", "h5", "h6"):
prefix = "#" * self._heading_level
self._output.append(f"\n{prefix} {self._block_buffer.strip()}\n")
self._block_buffer = ""
self._heading_level = 0
self._block_empty = True
elif tag_lower in ("th", "td"):
# Collapse cell content to a single line (newlines from
# would break the GFM table row). Pipe chars are escaped to
# avoid prematurely terminating cells.
cell_text = " ".join(self._current_cell).strip()
cell_text = re.sub(r'\s+', ' ', cell_text)
cell_text = cell_text.replace("|", "\\|")
if self._current_row is not None:
self._current_row.append(cell_text)
self._in_cell = False
self._current_cell = None
elif tag_lower == "tr":
if self._current_row is not None:
is_header = False # GFM doesn't distinguish; first row is header
self._table_rows.append((self._current_row, is_header))
self._current_row = None
elif tag_lower == "table":
self._emit_table()
self._in_table = False
self._table_rows = []
self._current_row = None
self._current_cell = None
self._in_cell = False
elif tag_lower == "blockquote":
self._flush_block()
self._in_blockquote = False
self._output.append("")
elif tag_lower in ("p", "div", "section"):
self._flush_block()
self._output.append("")
elif tag_lower in ("ul", "ol"):
if self._list_stack:
self._list_stack.pop()
self._output.append("")
elif tag_lower == "li":
self._flush_block()
elif tag_lower == "dt":
self._block_buffer = self._block_buffer.rstrip() + "**"
self._flush_block()
elif tag_lower == "dd":
self._flush_block()
elif tag_lower == "a":
if self._in_link:
link_text = " ".join("".join(self._link_text).split())
if link_text and self._link_href:
rendered = f"[{link_text}]({self._link_href})"
elif self._link_href:
rendered = f"<{self._link_href}>"
else:
rendered = ""
if self._in_cell:
self._current_cell.append(rendered)
else:
self._block_buffer += rendered
self._block_empty = False
self._in_link = False
self._link_href = None
self._link_text = []
elif tag_lower == "code":
if not self._in_pre:
self._append_inline("`")
elif tag_lower in ("strong", "b"):
self._append_inline("**")
elif tag_lower in ("em", "i"):
self._append_inline("*")
def handle_data(self, data):
if self._skip_depth > 0:
return
if self._in_pre:
self._pre_content.append(data)
elif self._in_cell:
self._current_cell.append(data)
elif self._in_link:
self._link_text.append(data)
else:
# Don't strip — let _flush_block normalize whitespace
if data.strip(): # only skip purely whitespace nodes
self._block_buffer += data
self._block_empty = False
def get_markdown(self) -> str:
self._flush_block()
# Emit a table if one was left open (malformed HTML)
if self._in_table and self._table_rows:
self._emit_table()
text = "\n".join(self._output)
text = re.sub(r"\n{3,}", "\n\n", text)
return text.strip()
def html_to_markdown(html_content: str) -> str:
"""Convert HTML to Markdown using tree-based parser."""
# Quick strip of scripts/styles first
html_content = re.sub(r']*>.*?', '', html_content,
flags=re.DOTALL | re.IGNORECASE)
html_content = re.sub(r']*>.*?', '', html_content,
flags=re.DOTALL | re.IGNORECASE)
converter = MarkdownConverter()
converter.feed(html_content)
return converter.get_markdown()
# ----- HTTP Fetch -----
# RETRY_BACKOFF_BASE, FALLBACK_UAS, detect_charset and is_retryable_error are
# imported from common.py (shared with search.py for a consistent retry policy).
#
# v2.0.0 改进：
# * 模块级 requests.Session 复用连接池 + cookie，减少 TLS 握手开销
# * 超时分离 (connect, read) 元组，避免大页面下载中途超时浪费已建连接
# * 集成 build_browser_headers() 发送完整浏览器指纹
# * 遵守 Retry-After header，避免盲目重试触发更严厉限流
# * compute_backoff_delay() 带封顶，避免高重试次数卡死进程
# 模块级 Session：复用 TCP 连接池、TLS 会话、cookie。
# 仅在 requests 可用时启用；stdlib 路径不受益但功能完整。
_session = None
def _get_session():
"""获取（惰性创建）模块级 requests.Session。
Session 复用带来：
* HTTP Keep-Alive 连接池（同站点多页面只握手一次）
* Cookie 持久化（某些站点登录态/反爬 cookie 自动携带）
* TLS 会话恢复（session resumption，节省 1-RTT）
单元测试可通过 ``_reset_session()`` 重置后用 ``_HAS_REQUESTS=False``
强制走 stdlib 路径。
"""
global _session
if _session is None and _HAS_REQUESTS:
_session = _requests.Session()
# 配置连接池：每主机最多 10 连接，总最多 20 连接
adapter = _requests.adapters.HTTPAdapter(
pool_connections=10, pool_maxsize=10, max_retries=0,
)
_session.mount("http://", adapter)
_session.mount("https://", adapter)
return _session
def _reset_session() -> None:
"""关闭并重置模块级 Session。测试用。"""
global _session
if _session is not None:
try:
_session.close()
except Exception:
pass
_session = None
class _NoRedirectHandler(urllib.request.HTTPRedirectHandler):
"""urllib handler that returns the 3xx response instead of following it."""
def http_error_302(self, req, fp, code, msg, headers):
return fp
http_error_301 = http_error_303 = http_error_307 = http_error_308 = http_error_302
FetchResult = namedtuple(
"FetchResult",
["content", "content_type", "final_url", "truncated", "user_agent"],
)
def fetch_url(url: str, timeout=15, user_agent: str = None,
encoding: str = None, auth_headers: dict = None,
max_retries: int = 3, max_size: int = None,
allow_redirects: bool = True,
referer: str = None) -> "FetchResult":
"""Fetch a URL with retry, encoding detection, UA fallback, and optional size limit.
Returns a :class:`FetchResult` namedtuple with fields:
``content`` (str), ``content_type`` (str), ``final_url`` (str),
``truncated`` (bool — True if ``max_size`` cut the response short),
``user_agent`` (str — the UA string that succeeded; useful for logging
whether a fallback UA was needed).
max_size=None means unlimited (full page). Set to e.g. 5242880 for a 5MB cap.
allow_redirects=False stops the client from following HTTP 3xx redirects.
v2.0.0 改进：
* ``timeout`` 支持标量（向后兼容）或 ``(connect, read)`` 元组；
标量会被转换为 ``(timeout, timeout*2)`` 分离建连和读超时
* ``referer`` 参数：设置 Referer 头，伪装来自搜索引擎的流量
* 浏览器指纹头：通过 build_browser_headers() 发送完整 Sec-* 头
* 确定性 UA：通过 get_ua_for_domain() 为同域名固定 UA
* Retry-After：429/503 响应读取 Retry-After header 作为最小重试延迟
* 退避封顶：compute_backoff_delay() 上限 60s
* Session 复用：requests 路径复用模块级 Session
"""
# 超时归一化：标量 → (connect, read) 元组
if isinstance(timeout, (int, float)):
connect_timeout = min(float(timeout), 10.0) # 建连不超过 10s
read_timeout = float(timeout)
timeout_tuple = (connect_timeout, read_timeout)
else:
timeout_tuple = timeout # 已是元组，原样使用
# 确定性 UA 选择：同域名固定 UA
effective_ua = get_ua_for_domain(url, user_agent)
last_error = None
# UA 轮换池：首次用 effective_ua，后续重试轮换其他 UA
user_agents = [effective_ua] + [ua for ua in FALLBACK_UAS if ua != effective_ua]
for attempt in range(max_retries + 1):
# 同一域名内 UA 固定；仅在重试失败后才换（避免会话内突变）
# 但当退避原因可能是 UA 被屏蔽（403）时，必须换 UA
ua = user_agents[min(attempt, len(user_agents) - 1)]
# 构造完整浏览器头（v2.0.0 核心）
headers = build_browser_headers(ua, referer=referer, accept_html=True)
if auth_headers:
headers.update(auth_headers)
try:
if _HAS_REQUESTS:
session = _get_session()
resp = session.get(url, timeout=timeout_tuple, headers=headers,
allow_redirects=allow_redirects, stream=True)
# stream=True holds the socket open; must close explicitly,
# including on raise_for_status() / max_size break / decode
# errors — otherwise the connection leaks back to the pool
# and long-running agents exhaust ports.
try:
# 429/503：读取 Retry-After，作为最小重试延迟
if resp.status_code in (429, 503) and attempt < max_retries:
retry_after_raw = resp.headers.get("Retry-After", "")
retry_after_sec = parse_retry_after(retry_after_raw)
resp.close()
delay = max(retry_after_sec,
compute_backoff_delay(attempt))
logger.info(f" Fetch retry {attempt+1}/{max_retries} "
f"(HTTP {resp.status_code}, Retry-After={retry_after_sec:.1f}s) "
f"in {delay:.1f}s")
time.sleep(delay)
continue
resp.raise_for_status()
# Read: unlimited if max_size is None, chunked with limit otherwise
if max_size is None:
raw = resp.content
truncated = False
else:
chunks = []
total = 0
for chunk in resp.iter_content(chunk_size=65536, decode_unicode=False):
if chunk:
chunks.append(chunk)
total += len(chunk)
if total > max_size:
break
raw = b"".join(chunks)
truncated = total > max_size
if encoding:
content = raw.decode(encoding)
else:
charset = detect_charset(raw, resp.headers.get("Content-Type", ""))
try:
content = raw.decode(charset)
except (UnicodeDecodeError, LookupError):
content = raw.decode("utf-8", errors="replace")
return FetchResult(content, resp.headers.get("Content-Type", ""),
resp.url, truncated, ua)
finally:
resp.close()
# stdlib fallback
req = urllib.request.Request(url, headers=headers)
if allow_redirects:
_opener = urllib.request.urlopen(req, timeout=timeout_tuple[1])
else:
_opener = urllib.request.build_opener(_NoRedirectHandler).open(
req, timeout=timeout_tuple[1])
with _opener as resp:
if max_size is None:
raw = resp.read()
truncated = False
else:
chunks = []
total = 0
while True:
chunk = resp.read(65536)
if not chunk:
break
chunks.append(chunk)
total += len(chunk)
if total > max_size:
break
raw = b"".join(chunks)
truncated = total > max_size
content_type = resp.headers.get("Content-Type", "")
final_url = resp.geturl()
# v2.1.1 修复：stdlib urllib 不自动解压 gzip/deflate
# 服务器返回压缩字节流时 raw.decode() 会失败 →
# errors="replace" → 全页 U+FFFD 乱码。
# requests 库会自动处理 Content-Encoding，但 stdlib 不会。
# 此前该 bug 被沙箱伪响应掩盖（两者都产生 U+FFFD），
# 实际在无 requests 的真实环境中会复现。
content_encoding = (resp.headers.get("Content-Encoding", "")
.lower().strip())
if content_encoding and raw:
try:
if "gzip" in content_encoding:
raw = gzip.decompress(raw)
elif "deflate" in content_encoding:
# deflate 可能是 zlib 包装或裸 deflate
try:
raw = zlib.decompress(raw)
except zlib.error:
raw = zlib.decompress(raw, -zlib.MAX_WBITS)
except (OSError, zlib.error) as e:
logger.debug(f" decompress failed ({content_encoding}): {e}")
# 解压失败保留原 raw，让下游 decode 兜底
if encoding:
charset = encoding
else:
charset = detect_charset(raw, content_type)
try:
content = raw.decode(charset)
except (UnicodeDecodeError, LookupError):
content = raw.decode("utf-8", errors="replace")
return FetchResult(content, content_type, final_url, truncated, ua)
except urllib.error.HTTPError as e:
last_error = e
# 429/503：读取 Retry-After（stdlib 路径）
if e.code in (429, 503) and attempt < max_retries:
retry_after_raw = e.headers.get("Retry-After", "") if e.headers else ""
retry_after_sec = parse_retry_after(retry_after_raw)
# HTTPError 本身是可读的响应对象（fp 已被 urllib 消费），
# 无需显式 close；直接进入退避。
delay = max(retry_after_sec, compute_backoff_delay(attempt))
logger.info(f" Fetch retry {attempt+1}/{max_retries} "
f"(HTTP {e.code}, Retry-After={retry_after_sec:.1f}s) "
f"in {delay:.1f}s")
time.sleep(delay)
continue
if is_retryable_error(e) and attempt < max_retries:
delay = compute_backoff_delay(attempt)
logger.info(f" Fetch retry {attempt+1}/{max_retries} ({e}) in {delay:.1f}s")
time.sleep(delay)
continue
raise RuntimeError(f"HTTP {e.code} for {url}") from e
except (urllib.error.URLError, OSError, TimeoutError) as e:
last_error = e
if attempt < max_retries:
delay = compute_backoff_delay(attempt)
logger.info(f" Fetch retry {attempt+1}/{max_retries} ({e}) in {delay:.1f}s")
time.sleep(delay)
continue
raise RuntimeError(f"Request failed for {url}: {e}") from e
except Exception as e:
# requests backend: retry only on connection errors (no response)
# or transient 429/5xx; do NOT retry permanent errors like 404.
if _HAS_REQUESTS and isinstance(e, _requests.exceptions.RequestException):
last_error = e
status = getattr(getattr(e, "response", None), "status_code", None)
# 429/503 with Retry-After
if status in (429, 503) and attempt < max_retries:
resp_obj = getattr(e, "response", None)
retry_after_raw = ""
if resp_obj is not None:
retry_after_raw = resp_obj.headers.get("Retry-After", "")
retry_after_sec = parse_retry_after(retry_after_raw)
if resp_obj is not None:
resp_obj.close()
delay = max(retry_after_sec, compute_backoff_delay(attempt))
logger.info(f" Fetch retry {attempt+1}/{max_retries} "
f"(HTTP {status}, Retry-After={retry_after_sec:.1f}s) "
f"in {delay:.1f}s")
time.sleep(delay)
continue
if (status is None or status in RETRYABLE_STATUS) and attempt < max_retries:
delay = compute_backoff_delay(attempt)
logger.info(f" Fetch retry {attempt+1}/{max_retries} ({e}) in {delay:.1f}s")
time.sleep(delay)
continue
raise RuntimeError(f"Request failed for {url}: {e}") from e
raise RuntimeError(f"All {max_retries+1} attempts failed for {url}: {last_error}") from last_error
# ----- Main -----
def main():
parser = argparse.ArgumentParser(
description="Fetch a web page and extract readable content",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=f"""searxng-cli v{VERSION}
Examples:
%(prog)s -u https://example.com extract clean text
%(prog)s -u https://example.com -e html raw HTML
%(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
""",
)
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("--timeout", "-t", type=int, default=15,
help="Request timeout in seconds (default: 15)")
parser.add_argument("--retries", type=int, default=3,
help="Max retries on transient errors (default: 3)")
parser.add_argument("--max-size", type=int, default=None, metavar="BYTES",
help="Max page size in bytes (default: unlimited). Set to 5242880 for 5MB cap.")
parser.add_argument("--user-agent", default=None,
help="Custom User-Agent header")
parser.add_argument("--encoding", default=None,
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.")
parser.add_argument("--proxy", default=None, metavar="URL",
help="HTTP/HTTPS proxy URL (e.g. http://corp-proxy:8080). "
"Respects existing HTTP_PROXY/HTTPS_PROXY env vars when omitted.")
parser.add_argument("--output", "-o", default=None,
help="Save to file instead of stdout")
parser.add_argument("--auth-bearer", default=None, metavar="TOKEN",
help="Authorization: Bearer for authenticated endpoints")
parser.add_argument("--auth-bearer-file", default=None, metavar="FILE",
help="Read Bearer token from a file (first non-empty, non-# line). "
"Avoids leaving tokens in shell history.")
parser.add_argument("--auth-basic", default=None, metavar="USER:PASS",
help="Authorization: Basic base64(user:pass) for authenticated endpoints")
parser.add_argument("--auth-basic-file", default=None, metavar="FILE",
help="Read basic auth 'user:pass' from a file (first non-empty, non-# line). "
"Avoids leaving passwords in shell history. "
"Env var SEARXNG_BASIC_AUTH is also honored.")
parser.add_argument("--verbose", "-v", action="store_true", default=False,
help="Verbose output: show debug-level diagnostics on stderr")
parser.add_argument("--quiet", action="store_true", default=False,
help="Quiet output: suppress progress messages on stderr; "
"only warnings and errors are shown")
parser.add_argument("--version", action="version", version=f"searxng-cli v{VERSION}")
args = parser.parse_args()
setup_logging(verbose=args.verbose, quiet=args.quiet)
force_utf8_stdout() # Windows: prevent GBK crash on non-ASCII chars
if not args.url.startswith(("http://", "https://")):
logger.error("Error: URL must start with http:// or https://")
sys.exit(1)
# Apply proxy via env vars so both urllib and requests honor it.
if args.proxy:
apply_proxy(args.proxy)
logger.info(f"Proxy: {args.proxy}")
try:
bearer_token = resolve_auth_bearer(args.auth_bearer, args.auth_bearer_file)
basic_auth = resolve_auth_basic(args.auth_basic, args.auth_basic_file)
except RuntimeError as e:
logger.error(f"Error: {e}")
sys.exit(1)
auth_headers = build_auth_headers(
bearer_token=bearer_token,
basic_auth=basic_auth,
)
if auth_headers:
auth_type = "Bearer" if bearer_token else "Basic"
logger.info(f"Auth: {auth_type} ***")
# v2.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,
encoding=args.encoding, auth_headers=auth_headers,
max_retries=args.retries, max_size=args.max_size,
allow_redirects=not args.no_redirect,
referer=args.referer,
)
content, content_type, final_url = (
result.content, result.content_type, result.final_url,
)
except Exception as e:
# 诊断信息增强：从 __cause__ 链中提取 HTTP 状态码、原始异常类型，
# 让 AI Agent 能程序化判断失败原因（404 vs 403 vs DNS 失败等），
# 而不是只看到一句 "HTTP 404 for ..."。
cause = e.__cause__
status_code = None
cause_type = type(cause).__name__ if cause else type(e).__name__
# urllib HTTPError 有 .code 属性；requests HTTPError 有 .response.status_code
if cause is not None:
status_code = (getattr(cause, "code", None) or
getattr(getattr(cause, "response", None), "status_code", None))
if status_code is None:
# 最后一道兜底：从异常消息里提取 "HTTP NNN" 模式
m = re.search(r'HTTP (\d{3})', str(e))
if m:
status_code = int(m.group(1))
diag_parts = [f"Error: {e}"]
if status_code is not None:
diag_parts.append(f"status_code={status_code}")
diag_parts.append(f"cause={cause_type}")
diag_parts.append(f"url={args.url}")
if args.no_redirect:
diag_parts.append("redirects=disabled")
logger.error(" | ".join(diag_parts))
# 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}")
is_html = ("html" in content_type.lower() or
content.strip().startswith("