diff --git a/.github/ISSUE_TEMPLATE/feature_request.yml b/.github/ISSUE_TEMPLATE/feature_request.yml
index 67be3c027..13e88cfac 100644
--- a/.github/ISSUE_TEMPLATE/feature_request.yml
+++ b/.github/ISSUE_TEMPLATE/feature_request.yml
@@ -7,11 +7,13 @@ body:
attributes:
value: |
请说明你希望添加的功能。
+
+ 站点适配请求请先按 [站点适配采集说明](https://github.com/jxxghp/MoviePilot/blob/v2/docs/site-adapter-capture.md) 生成脱敏 ZIP，并在下方附加。Issue 及附件是公开内容，提交前必须解压预览四个文件。不要上传 Cookie、Authorization、通行密钥、会话字段或任何原始数据。
- type: input
id: version
attributes:
label: 当前程序版本
- description: 目前使用的程序版本
+ description: 目前使用的程序版本；仅提供站点采集文件且未安装 MoviePilot 时填写“不适用”
validations:
required: true
- type: dropdown
@@ -22,6 +24,9 @@ body:
options:
- Docker
- Windows
+ - macOS
+ - Linux
+ - 仅提供站点采集文件
validations:
required: true
- type: dropdown
@@ -32,6 +37,7 @@ body:
options:
- 主程序
- 插件
+ - 站点适配
- 其他
validations:
required: true
@@ -43,6 +49,14 @@ body:
placeholder: "功能改进"
validations:
required: true
+ - type: textarea
+ id: site-adapter-capture
+ attributes:
+ label: 站点适配采集文件
+ description: 站点适配请求必须把采集器生成并人工预览确认过的脱敏 ZIP 拖到这里；Issue 附件公开，严禁附加 Cookie、原始 HTML、HAR 或浏览器网络归档。其他类型请填写“不适用”。
+ placeholder: "将 moviepilot-site-capture-*.zip 拖到这里；非站点适配填写：不适用"
+ validations:
+ required: true
- type: textarea
id: references
attributes:
diff --git a/.github/workflows/site-adapter-collector.yml b/.github/workflows/site-adapter-collector.yml
new file mode 100644
index 000000000..30403f8dd
--- /dev/null
+++ b/.github/workflows/site-adapter-collector.yml
@@ -0,0 +1,128 @@
+name: Site Adapter Collector
+
+on:
+ workflow_dispatch:
+ release:
+ types:
+ - published
+
+permissions:
+ contents: read
+
+jobs:
+ build:
+ name: Build ${{ matrix.platform_name }} collector
+ runs-on: ${{ matrix.runner }}
+ timeout-minutes: 30
+ strategy:
+ fail-fast: false
+ matrix:
+ include:
+ - platform_name: Windows
+ platform_id: windows
+ runner: windows-latest
+ source_name: moviepilot-site-collector.exe
+ asset_name: moviepilot-site-collector-windows.exe
+ artifact_name: site-adapter-collector-windows
+ - platform_name: macOS
+ platform_id: macos
+ runner: macos-latest
+ source_name: moviepilot-site-collector
+ asset_name: MoviePilot-Site-Collector-macOS.zip
+ artifact_name: site-adapter-collector-macos
+ - platform_name: Linux
+ platform_id: linux
+ runner: ubuntu-latest
+ source_name: moviepilot-site-collector
+ asset_name: moviepilot-site-collector-linux
+ artifact_name: site-adapter-collector-linux
+
+ steps:
+ - name: Checkout code
+ uses: actions/checkout@v4
+
+ - name: Set up Python
+ uses: actions/setup-python@v5
+ with:
+ python-version: '3.12'
+ cache: pip
+ cache-dependency-path: scripts/site_adapter_collector_requirements.txt
+
+ - name: Install build dependencies
+ run: |
+ python -m pip install --upgrade pip setuptools wheel
+ pip install -r scripts/site_adapter_collector_requirements.txt
+
+ - name: Build single-file collector
+ run: |
+ pyinstaller --clean --noconfirm scripts/site_adapter_collector.spec
+
+ - name: Smoke-test collector
+ env:
+ SOURCE_NAME: ${{ matrix.source_name }}
+ run: |
+ python -c "import os, subprocess; from pathlib import Path; subprocess.run([str((Path('dist') / os.environ['SOURCE_NAME']).resolve()), '--help'], check=True)"
+
+ - name: Package macOS double-click archive
+ if: matrix.platform_id == 'macos'
+ shell: bash
+ env:
+ ASSET_NAME: ${{ matrix.asset_name }}
+ SOURCE_NAME: ${{ matrix.source_name }}
+ run: |
+ package_dir="dist/MoviePilot-Collector"
+ mkdir -p "$package_dir"
+ cp "dist/$SOURCE_NAME" "$package_dir/moviepilot-site-collector-macos"
+ cp scripts/start-site-adapter-collector.command "$package_dir/start-site-adapter-collector.command"
+ chmod +x "$package_dir/moviepilot-site-collector-macos"
+ chmod +x "$package_dir/start-site-adapter-collector.command"
+ cd dist
+ COPYFILE_DISABLE=1 zip -q -r -X "$ASSET_NAME" MoviePilot-Collector
+
+ - name: Rename Windows and Linux collector
+ if: matrix.platform_id != 'macos'
+ env:
+ ASSET_NAME: ${{ matrix.asset_name }}
+ SOURCE_NAME: ${{ matrix.source_name }}
+ run: |
+ python -c "import os; from pathlib import Path; (Path('dist') / os.environ['SOURCE_NAME']).replace(Path('dist') / os.environ['ASSET_NAME'])"
+
+ - name: Generate SHA-256 checksum
+ env:
+ ASSET_NAME: ${{ matrix.asset_name }}
+ run: |
+ python -c "import hashlib, os; from pathlib import Path; path = Path('dist') / os.environ['ASSET_NAME']; path.with_name(path.name + '.sha256').write_text(f'{hashlib.sha256(path.read_bytes()).hexdigest()} {path.name}\n', encoding='utf-8')"
+
+ - name: Upload collector artifact
+ uses: actions/upload-artifact@v4
+ with:
+ name: ${{ matrix.artifact_name }}
+ path: |
+ dist/${{ matrix.asset_name }}
+ dist/${{ matrix.asset_name }}.sha256
+ if-no-files-found: error
+ retention-days: 14
+
+ publish:
+ name: Upload collectors to release
+ if: github.event_name == 'release' && github.event.action == 'published'
+ needs:
+ - build
+ runs-on: ubuntu-latest
+ permissions:
+ contents: write
+
+ steps:
+ - name: Download collector artifacts
+ uses: actions/download-artifact@v4
+ with:
+ pattern: site-adapter-collector-*
+ path: release-assets
+ merge-multiple: true
+
+ - name: Upload assets to published release
+ env:
+ GH_TOKEN: ${{ github.token }}
+ RELEASE_TAG: ${{ github.event.release.tag_name }}
+ run: |
+ gh release upload "$RELEASE_TAG" release-assets/* --clobber --repo "$GITHUB_REPOSITORY"
diff --git a/.gitignore b/.gitignore
index c58eadcc4..9a8b78cf3 100644
--- a/.gitignore
+++ b/.gitignore
@@ -37,6 +37,7 @@ coverage.json
htmlcov/
.vscode
venv
+moviepilot-site-capture-*.zip
# Pylint
pylint-report.json
diff --git a/README.md b/README.md
index 2a9bd7876..68d89bc9c 100644
--- a/README.md
+++ b/README.md
@@ -59,6 +59,8 @@ curl -fsSL https://raw.githubusercontent.com/jxxghp/MoviePilot/v2/scripts/bootst
- 文档规则入口：[docs/rules/README.md](docs/rules/README.md)
- 开发环境与本地源码运行：[docs/development-setup.md](docs/development-setup.md)
- 测试说明：[docs/testing.md](docs/testing.md)
+- 新站点适配采集与 Feature Request 提交：[docs/site-adapter-capture.md](docs/site-adapter-capture.md)
+- 普通用户独立采集器下载与运行：[docs/site-adapter-collector-release.md](docs/site-adapter-collector-release.md)
- REST API 文档：https://api.movie-pilot.org
- 插件开发说明：https://wiki.movie-pilot.org/zh/plugindev
diff --git a/README_EN.md b/README_EN.md
index 85b6060fa..f3b69d262 100644
--- a/README_EN.md
+++ b/README_EN.md
@@ -58,6 +58,8 @@ Before contributing, read the repository rules and local environment guide, keep
- Rule index: [docs/rules/README.md](docs/rules/README.md)
- Development setup and local source run: [docs/development-setup.md](docs/development-setup.md)
- Testing guide: [docs/testing.md](docs/testing.md)
+- New site adapter capture and Feature Request submission: [docs/site-adapter-capture.md](docs/site-adapter-capture.md)
+- Standalone site collector download and runtime: [docs/site-adapter-collector-release.md](docs/site-adapter-collector-release.md)
- REST API documentation: https://api.movie-pilot.org
- Plugin development guide: https://wiki.movie-pilot.org/zh/plugindev
diff --git a/docs/rules/03-commands.md b/docs/rules/03-commands.md
index be87bb63d..8f7ae7097 100644
--- a/docs/rules/03-commands.md
+++ b/docs/rules/03-commands.md
@@ -269,4 +269,20 @@ moviepilot help tool
moviepilot help scheduler
```
-*Last Updated: 2026-05-25*
+---
+
+## Site Adapter Capture — macOS / Linux
+
+```bash
+# Run from a MoviePilot source checkout and reuse its virtual environment
+bash scripts/collect-site-adapter.sh
+```
+
+**Rules:**
+- The default collector asks only for the site HTTPS address, opens an isolated local Chrome/Edge profile, and reads the completed search page after the user confirms.
+- Users must not be asked to inspect HTML or copy Cookie/User-Agent values in the default flow. `--manual-cookie` is an advanced fallback only.
+- Run only the collector shipped with a trusted local MoviePilot source checkout or installation package. Do not pipe a remote branch script into a shell.
+- Never put a Cookie or other credential in command arguments or shell history.
+- Feature Request attachments are public. Review all four files in the generated ZIP before attaching it, and never attach raw HTML, HAR, or browser network archives.
+
+*Last Updated: 2026-07-12*
diff --git a/docs/site-adapter-capture.md b/docs/site-adapter-capture.md
new file mode 100644
index 000000000..6c1bc5d8f
--- /dev/null
+++ b/docs/site-adapter-capture.md
@@ -0,0 +1,60 @@
+# 站点适配采集
+
+当开发者没有目标站点账号时，可以由已有账号的用户在本地采集一份经过裁剪和脱敏的搜索页结构，并把采集 ZIP 附加到站点适配 Feature Request。开发者和自动化流程只处理脱敏包，不需要获取用户账号。
+
+## 普通用户一键采集
+
+普通用户只需准备两样东西：目标站点账号，以及已安装的 Chrome、Edge 或 Chromium。无需安装 Python、Git、MoviePilot、Docker，也不需要查看 HTML、复制 Cookie 或填写 User-Agent。
+
+1. 从 MoviePilot 官方 Release 下载与 Windows、macOS 或 Linux 对应的 `moviepilot-site-collector-*` 单文件采集器。
+2. 运行采集器，只输入站点首页地址，例如 `https://tracker.example.com`。
+3. 程序会打开一个临时浏览器窗口。在这个窗口里正常登录站点，搜索一个能返回至少 3 条结果的常见关键词，并保持搜索结果页打开。
+4. 回到采集器按回车。程序会自动识别搜索地址、关键词、Cookie 和 User-Agent，完成本地裁剪与脱敏后生成 ZIP。
+
+临时浏览器使用独立的一次性用户目录，不会读取日常浏览器的历史登录状态。采集完成后程序会关闭临时浏览器并清理这次登录数据。原始页面和 Cookie 只在内存中处理，不会写入采集包。
+
+各系统下载文件和首次运行方式见 [站点适配采集器下载说明](site-adapter-collector-release.md)。
+
+## 开发者源码入口
+
+已经有 MoviePilot 源码和 Python 环境的开发者，也可以在项目目录执行：
+
+```bash
+bash scripts/collect-site-adapter.sh
+```
+
+源码入口和独立程序使用同一套浏览器采集流程。只有排查兼容问题时才使用 `--manual-cookie` 高级模式；普通用户不需要接触 Cookie。
+
+## 第一版限制
+
+当前采集器会读取浏览器渲染后的页面，因此可由用户在临时窗口中完成验证码、Cloudflare 检查和普通登录。但自动适配协议仍要求搜索结果地址能够表示为 HTTPS GET URL。以下场景第一版不做自动适配：
+
+- 必须 POST 表单才能搜索的站点。
+- 搜索完成后地址栏完全没有关键词或可复用搜索参数的站点。
+- 需要专用 API、复杂签名或无法从一次搜索结果页观察出 FREE/HR 规则的站点。
+
+遇到这些场景时，请在 Feature Request 中说明失败步骤和终端错误文字，等待人工确认采集方案。不要用原始 HAR、原始 HTML 或包含账号信息的截图替代脱敏 ZIP。
+
+## 脱敏范围
+
+采集器只接受 HTTPS 地址。默认模式从本机临时浏览器只读当前搜索结果页；高级手动模式设置 30 秒超时和 5 MiB 响应上限，并禁止携带 Cookie 跨 origin 重定向。读取页面后会在本机完成以下处理：
+
+- 只保留种子列表、表头和最多 25 条结果相关 DOM，丢弃账号导航、页脚、脚本、样式、隐藏表单和其他页面内容。
+- 替换种子标题、用户名、邮箱、IP、时间、大小、统计值和长随机标识，只保留适配所需的标签、class、字段与链接结构。
+- URL 转为同源相对路径，查询值统一替换为占位符，凭据语义的字段直接移除。
+- Cookie 和浏览器 UA 仅用于本次请求，不写入采集包；写入前还会使用 Cookie 原值执行二次泄露检查。
+
+输出 ZIP 根目录固定包含：
+
+- `manifest.json`：包版本、站点标识、采集时间、结果行数、HTML 摘要和隐私声明。
+- `request.json`：仅包含 GET、origin、相对路径和脱敏后的查询参数，搜索值固定为 `{keyword}`。
+- `search.html`：本地裁剪并脱敏后的种子列表结构。
+- `redaction-report.json`：脱敏状态和各类处理计数。
+
+## 提交 Feature Request
+
+在 GitHub 创建“功能改进” Issue，类型选择“站点适配”，然后把生成的 `moviepilot-site-capture-*.zip` 直接拖入“站点适配采集文件”输入框。
+
+Feature Request 及其附件是公开内容。提交前请先在本地解压 ZIP，确认根目录只有上述四个文件，并逐一预览确认没有站点账号、搜索隐私或其他不希望公开的信息。
+
+站点适配请求必须附加采集器生成并人工复核过的 ZIP。严禁手工上传 Cookie、Authorization、通行密钥、会话字段、原始 HTML、原始 HAR、浏览器网络归档或截图中的账号信息。如果采集失败，请只提交终端错误文字，不要用任何原始数据代替脱敏包。
diff --git a/docs/site-adapter-collector-release.md b/docs/site-adapter-collector-release.md
new file mode 100644
index 000000000..bc436bc34
--- /dev/null
+++ b/docs/site-adapter-collector-release.md
@@ -0,0 +1,36 @@
+# 站点适配采集器下载说明
+
+普通用户优先使用 MoviePilot 正式 Release 提供的单文件采集器。单文件已经包含 Python 和采集器依赖，不需要安装 Python、pip、Git、MoviePilot 后端，也不需要下载源码。电脑只需已安装 Chrome、Edge 或 Chromium 浏览器。
+
+## 选择下载文件
+
+请只从 MoviePilot 官方 GitHub Release 下载与系统匹配的文件：
+
+| 系统 | 下载文件 | 用户侧运行环境 |
+|---|---|---|
+| Windows | `moviepilot-site-collector-windows.exe` | Chrome、Edge 或 Chromium |
+| macOS | `MoviePilot-Site-Collector-macOS.zip` | Chrome、Edge 或 Chromium |
+| Linux | `moviepilot-site-collector-linux` | Chrome、Edge 或 Chromium |
+
+每个程序旁边还有同名的 `.sha256` 文件，可用于核对下载文件是否完整。GitHub Actions 的手动构建产物主要用于维护者测试；普通用户应使用正式 Release 资产。
+
+## 运行采集器
+
+Windows 用户下载后双击 `.exe`，按窗口提示操作即可。macOS 用户解压 ZIP 后双击 `start-site-adapter-collector.command`，不要打开构建目录中的 `.pkg` 文件。Linux 用户在下载目录打开终端，只需首次赋予执行权限后运行：
+
+```bash
+chmod +x moviepilot-site-collector-linux
+./moviepilot-site-collector-linux
+```
+
+运行后只需输入站点首页地址，随后在弹出的临时浏览器中登录并搜索，最后回到采集器按回车。采集器会在当前目录生成 `moviepilot-site-capture-*.zip`，用户只需把这个 ZIP 附加到站点适配 Feature Request，不需要提交任何源码、Cookie 或 HTML。
+
+## 系统安全提示
+
+当前自动构建产物尚未接入 Windows 或 Apple 代码签名。Windows SmartScreen 或 macOS Gatekeeper 可能因此显示安全提示。仅在文件来自 MoviePilot 官方 GitHub Release，且校验摘要一致时运行；不要从聊天、网盘或第三方站点接收采集器。
+
+如果系统阻止运行，可改用随 MoviePilot 源码提供的本地采集脚本；该方式需要 Python 3.11 及完整后端依赖，不适合作为普通用户的首选路径。
+
+## 维护者发布流程
+
+`.github/workflows/site-adapter-collector.yml` 支持手动触发，也会在 Release 发布后自动构建 Windows、macOS 和 Linux 单文件程序。每个平台先执行 `--help` 启动检查，再上传程序及 SHA-256 摘要为 Workflow Artifact；Release 事件会在三个平台全部成功后，用 GitHub CLI 把同一组文件附加到触发本次任务的 Release。
diff --git a/scripts/collect-site-adapter.sh b/scripts/collect-site-adapter.sh
new file mode 100644
index 000000000..5256fcd96
--- /dev/null
+++ b/scripts/collect-site-adapter.sh
@@ -0,0 +1,99 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+ORIGINAL_DIR="$PWD"
+SCRIPT_DIR=""
+PROJECT_ROOT=""
+PYTHON_BIN=""
+COLLECTOR_PATH=""
+
+# 判断候选 Python 是否满足 3.11 最低版本。
+python_version_ok() {
+ "$1" - <<'PY' >/dev/null 2>&1
+import sys
+raise SystemExit(0 if sys.version_info >= (3, 11) else 1)
+PY
+}
+
+# 从当前源码目录、环境变量或脚本位置查找 MoviePilot 根目录。
+find_project_root() {
+ local candidate=""
+ local source_path="${BASH_SOURCE[0]:-}"
+ if [[ -n "$source_path" && -f "$source_path" ]]; then
+ SCRIPT_DIR="$(cd "$(dirname "$source_path")" && pwd)"
+ fi
+ for candidate in "${MOVIEPILOT_ROOT:-}" "$ORIGINAL_DIR" "${SCRIPT_DIR:+$SCRIPT_DIR/..}"; do
+ if [[ -n "$candidate" && -f "$candidate/scripts/site_adapter_collector.py" ]]; then
+ PROJECT_ROOT="$(cd "$candidate" && pwd)"
+ return 0
+ fi
+ done
+ return 1
+}
+
+# 查找可用的项目虚拟环境或系统 Python。
+find_python() {
+ local candidate=""
+ local resolved=""
+ for candidate in \
+ "${PROJECT_ROOT:+$PROJECT_ROOT/venv/bin/python}" \
+ "${PROJECT_ROOT:+$PROJECT_ROOT/.venv/bin/python}" \
+ "${VIRTUAL_ENV:+$VIRTUAL_ENV/bin/python}" \
+ python3.13 python3.12 python3.11 python3; do
+ [[ -n "$candidate" ]] || continue
+ if [[ "$candidate" == */* ]]; then
+ resolved="$candidate"
+ else
+ resolved="$(command -v "$candidate" 2>/dev/null || true)"
+ fi
+ if [[ -n "$resolved" && -x "$resolved" ]] && python_version_ok "$resolved"; then
+ PYTHON_BIN="$resolved"
+ return 0
+ fi
+ done
+ return 1
+}
+
+# 判断 Python 是否已具备独立采集器所需的最小运行依赖。
+project_runtime_ready() {
+ PYTHONPATH="$PROJECT_ROOT${PYTHONPATH:+:$PYTHONPATH}" "$PYTHON_BIN" - <<'PY' >/dev/null 2>&1
+import requests
+import websocket
+from bs4 import BeautifulSoup
+PY
+}
+
+# 让脚本始终从终端安全读取交互输入。
+restore_terminal_input() {
+ if [[ -r /dev/tty ]]; then
+ exec &2
+ exit 1
+ fi
+}
+
+# 定位本地项目运行环境后启动随发行包提供的采集器。
+main() {
+ if ! find_project_root; then
+ echo "未找到本地 MoviePilot 源码或安装目录，请在 MoviePilot 目录中运行此脚本。" >&2
+ exit 1
+ fi
+ if ! find_python; then
+ echo "未找到 Python 3.11 或更高版本，请先安装 Python。" >&2
+ exit 1
+ fi
+
+ if ! project_runtime_ready; then
+ echo "本地 Python 环境缺少采集器依赖，请优先下载官方 Release 的独立采集器。" >&2
+ exit 1
+ fi
+ COLLECTOR_PATH="$PROJECT_ROOT/scripts/site_adapter_collector.py"
+
+ restore_terminal_input
+ cd "$ORIGINAL_DIR"
+ PYTHONPATH="$PROJECT_ROOT${PYTHONPATH:+:$PYTHONPATH}" "$PYTHON_BIN" "$COLLECTOR_PATH" "$@"
+}
+
+main "$@"
diff --git a/scripts/site_adapter_collector.py b/scripts/site_adapter_collector.py
new file mode 100644
index 000000000..aee03a0cc
--- /dev/null
+++ b/scripts/site_adapter_collector.py
@@ -0,0 +1,1449 @@
+#!/usr/bin/env python3
+"""采集并脱敏站点搜索页结构，生成站点适配数据包。"""
+
+import argparse
+import copy
+import getpass
+import hashlib
+import json
+import os
+import re
+import shutil
+import socket
+import subprocess
+import sys
+import tempfile
+import time
+import zipfile
+from contextlib import contextmanager
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Generator, Optional, Union
+from urllib.parse import parse_qsl, quote, unquote, urlencode, urljoin, urlsplit, urlunsplit
+
+import requests
+import websocket
+from bs4 import BeautifulSoup, Comment, NavigableString, Tag
+
+
+COLLECTOR_VERSION = "1.0.1"
+FORMAT_VERSION = 1
+MAX_RESPONSE_BYTES = 5 * 1024 * 1024
+MAX_REDIRECTS = 3
+MAX_RESULT_ROWS = 25
+MIN_RESULT_ROWS = 3
+REQUEST_TIMEOUT_SECONDS = 30
+CDP_HTTP_TIMEOUT_SECONDS = 5
+BROWSER_START_TIMEOUT_SECONDS = 20
+CDP_COMMAND_TIMEOUT_SECONDS = 15
+DEFAULT_USER_AGENT = (
+ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+ "AppleWebKit/537.36 (KHTML, like Gecko) "
+ "Chrome/136.0.0.0 Safari/537.36"
+)
+ARCHIVE_FILE_NAMES = (
+ "manifest.json",
+ "request.json",
+ "search.html",
+ "redaction-report.json",
+)
+REDIRECT_STATUS_CODES = {301, 302, 303, 307, 308}
+SEARCH_PARAMETER_NAMES = {
+ "keyword",
+ "q",
+ "query",
+ "search",
+ "searchstr",
+ "searchtext",
+ "search_text",
+}
+SENSITIVE_NAME_PATTERN = re.compile(
+ r"(?:auth(?:orization)?|cookie|csrf|pass(?:key|word)?|secret|session(?:id)?|token)",
+ re.IGNORECASE,
+)
+TORRENT_PATH_PATTERN = re.compile(r"(?:^|/)torrents?/[^/?#]+", re.IGNORECASE)
+RESULT_DESCRIPTOR_PATTERN = re.compile(
+ r"(?:browse|result|search|torrent)",
+ re.IGNORECASE,
+)
+IDENTITY_DESCRIPTOR_PATTERN = re.compile(
+ r"(?:member|profile|uploaded[_-]?by|uploader|user(?:details|name)?|author)",
+ re.IGNORECASE,
+)
+EMAIL_PATTERN = re.compile(r"\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}\b", re.IGNORECASE)
+IP_ADDRESS_PATTERN = re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}\b")
+HIGH_ENTROPY_VALUE_PATTERN = re.compile(
+ r"\b[a-fA-F0-9]{24,}\b|"
+ r"\b(?=[A-Za-z0-9_-]{32,}\b)(?=[A-Za-z0-9_-]*[a-z])"
+ r"(?=[A-Za-z0-9_-]*[A-Z])(?=[A-Za-z0-9_-]*[0-9])[A-Za-z0-9_-]+\b"
+)
+DATE_TIME_PATTERN = re.compile(
+ r"\b\d{4}[-/.]\d{1,2}[-/.]\d{1,2}(?:[ T]\d{1,2}:\d{2}(?::\d{2})?)?\b"
+)
+SIZE_PATTERN = re.compile(r"^\s*\d+(?:\.\d+)?\s*(?:[KMGTPE]i?B|bytes?)\s*$", re.IGNORECASE)
+NUMBER_PATTERN = re.compile(r"^\s*[+-]?\d+(?:\.\d+)?\s*$")
+URL_ATTRIBUTE_NAMES = {
+ "action",
+ "background",
+ "cite",
+ "data-href",
+ "data-lazy-src",
+ "data-orig",
+ "data-original",
+ "data-src",
+ "data-url",
+ "formaction",
+ "href",
+ "poster",
+ "src",
+}
+URL_SET_ATTRIBUTE_NAMES = {"data-srcset", "srcset"}
+TEXT_ATTRIBUTE_NAMES = {
+ "alt",
+ "aria-label",
+ "data-date",
+ "data-original-title",
+ "data-time",
+ "data-title",
+ "datetime",
+ "title",
+}
+ALLOWED_ATTRIBUTE_NAMES = {
+ "aria-expanded",
+ "aria-hidden",
+ "aria-sort",
+ "class",
+ "colspan",
+ "data-capture-kind",
+ "data-id",
+ "data-leechers",
+ "data-seeders",
+ "data-size",
+ "data-timestamp",
+ "decoding",
+ "headers",
+ "height",
+ "id",
+ "loading",
+ "rel",
+ "role",
+ "rowspan",
+ "scope",
+ "target",
+ "type",
+ "width",
+}
+STRUCTURAL_PAGE_EXTENSIONS = {".asp", ".aspx", ".htm", ".html", ".php"}
+NON_SEARCH_PARAMETER_NAMES = {
+ "action",
+ "area",
+ "category",
+ "cat",
+ "filter",
+ "incldead",
+ "page",
+ "sort",
+ "type",
+}
+
+
+class RequestUtils:
+ """为独立采集器提供与原调用方式兼容的最小 HTTP 客户端。"""
+
+ def __init__(
+ self,
+ headers: Optional[dict[str, str]] = None,
+ cookies: Optional[str] = None,
+ timeout: int = REQUEST_TIMEOUT_SECONDS,
+ ) -> None:
+ """
+ 初始化仅用于当前采集任务的 requests 会话。
+
+ :param headers: 请求头
+ :param cookies: 原始 Cookie 请求头，仅驻留内存
+ :param timeout: 默认请求超时秒数
+ """
+ self._session = requests.Session()
+ self._session.headers.update(headers or {})
+ if cookies:
+ self._session.headers["Cookie"] = cookies
+ self._timeout = timeout
+
+ def get_res(self, url: str, **kwargs) -> requests.Response:
+ """
+ 发起 GET 请求并返回 requests 响应对象。
+
+ :param url: 请求地址
+ :param kwargs: 传递给 requests 的请求参数
+ :return: HTTP 响应对象
+ :raises RuntimeError: 网络请求失败时抛出
+ """
+ kwargs.setdefault("timeout", self._timeout)
+ try:
+ return self._session.get(url, **kwargs)
+ except requests.RequestException as error:
+ raise RuntimeError(f"搜索页请求失败：{str(error)}") from error
+
+ def close(self) -> None:
+ """关闭当前采集任务使用的 HTTP 会话。"""
+ self._session.close()
+
+
+@dataclass(frozen=True)
+class _BrowserCapture:
+ """保存从当前浏览器标签页读取的临时采集数据。"""
+
+ url: str
+ html: str
+ cookie: str
+ user_agent: str
+ search_inputs: list[dict[str, str]]
+
+
+@dataclass
+class _BrowserSession:
+ """管理临时浏览器进程、CDP 地址和一次性用户目录。"""
+
+ process: subprocess.Popen
+ profile_guard: tempfile.TemporaryDirectory
+ port: int
+ browser_websocket_url: str
+
+ def close(self) -> None:
+ """关闭浏览器进程并删除包含登录状态的一次性用户目录。"""
+ try:
+ with _CdpClient(self.browser_websocket_url) as client:
+ client.call("Browser.close")
+ except (RuntimeError, websocket.WebSocketException):
+ pass
+
+ if self.process.poll() is None:
+ try:
+ self.process.wait(timeout=5)
+ except subprocess.TimeoutExpired:
+ self.process.terminate()
+ try:
+ self.process.wait(timeout=5)
+ except subprocess.TimeoutExpired:
+ self.process.kill()
+ self.process.wait(timeout=5)
+ self.profile_guard.cleanup()
+
+
+class _CdpClient:
+ """通过本机 WebSocket 执行最小 Chrome DevTools Protocol 命令。"""
+
+ def __init__(self, websocket_url: str) -> None:
+ """
+ 连接已校验为本机回环地址的 CDP WebSocket。
+
+ :param websocket_url: CDP WebSocket 地址
+ """
+ parsed = urlsplit(websocket_url)
+ if parsed.scheme not in {"ws", "wss"} or parsed.hostname not in {
+ "127.0.0.1",
+ "localhost",
+ "::1",
+ }:
+ raise RuntimeError("浏览器调试地址不是本机回环地址，已停止采集")
+ try:
+ self._socket = websocket.create_connection(
+ websocket_url,
+ timeout=CDP_COMMAND_TIMEOUT_SECONDS,
+ origin="http://127.0.0.1",
+ http_proxy_host=None,
+ )
+ except websocket.WebSocketException as error:
+ raise RuntimeError(f"无法连接本机浏览器调试接口：{str(error)}") from error
+ self._message_id = 0
+
+ def __enter__(self) -> "_CdpClient":
+ """返回已连接的 CDP 客户端。"""
+ return self
+
+ def __exit__(self, exc_type, exc_value, traceback) -> None:
+ """离开上下文时关闭 WebSocket。"""
+ self.close()
+
+ def close(self) -> None:
+ """关闭 CDP WebSocket。"""
+ self._socket.close()
+
+ def call(self, method: str, params: Optional[dict] = None) -> dict:
+ """
+ 发送一条 CDP 命令并等待对应响应。
+
+ :param method: CDP 方法名
+ :param params: 可选命令参数
+ :return: CDP 结果字典
+ :raises RuntimeError: CDP 返回错误、超时或连接中断时抛出
+ """
+ self._message_id += 1
+ message_id = self._message_id
+ command = {"id": message_id, "method": method}
+ if params:
+ command["params"] = params
+ try:
+ self._socket.send(json.dumps(command))
+ deadline = time.monotonic() + CDP_COMMAND_TIMEOUT_SECONDS
+ while time.monotonic() < deadline:
+ message = json.loads(self._socket.recv())
+ if message.get("id") != message_id:
+ continue
+ if message.get("error"):
+ error_message = message["error"].get("message", "未知错误")
+ raise RuntimeError(f"浏览器调试命令失败：{error_message}")
+ result = message.get("result")
+ return result if isinstance(result, dict) else {}
+ except (json.JSONDecodeError, OSError, websocket.WebSocketException) as error:
+ raise RuntimeError(f"读取浏览器页面失败：{str(error)}") from error
+ raise RuntimeError("读取浏览器页面超时，请保持浏览器窗口打开后重试")
+
+ def evaluate(self, expression: str):
+ """
+ 在当前页面执行只读 JavaScript 并返回可序列化值。
+
+ :param expression: JavaScript 表达式
+ :return: 表达式结果
+ :raises RuntimeError: 页面脚本执行失败时抛出
+ """
+ response = self.call(
+ "Runtime.evaluate",
+ {
+ "expression": expression,
+ "returnByValue": True,
+ "awaitPromise": True,
+ },
+ )
+ if response.get("exceptionDetails"):
+ raise RuntimeError("浏览器无法读取当前页面，请刷新页面后重试")
+ remote_result = response.get("result", {})
+ return remote_result.get("value")
+
+
+@dataclass(frozen=True)
+class _CaptureRequest:
+ """保存规范化后的搜索请求及可公开元数据。"""
+
+ url: str
+ origin: str
+ path: str
+ params: dict[str, str]
+ public_params: dict[str, str]
+ domain: str
+ site_id: str
+ site_name: str
+
+
+@dataclass
+class _RedactionStats:
+ """记录本地裁剪和脱敏动作的数量。"""
+
+ discarded_nodes: int = 0
+ removed_attributes: int = 0
+ redacted_urls: int = 0
+ redacted_identity_values: int = 0
+ redacted_result_values: int = 0
+ redacted_sensitive_values: int = 0
+
+
+def _is_sensitive_name(value: str) -> bool:
+ """判断字段名或路径是否带有凭据语义。"""
+ return bool(SENSITIVE_NAME_PATTERN.search(value))
+
+
+def _normalize_origin(url: str) -> tuple[str, str, str]:
+ """
+ 校验 HTTPS 地址并返回 origin、主机名和规范路径。
+
+ :param url: 用户输入的搜索页地址
+ :return: origin、域名和以斜杠开头的相对路径
+ :raises ValueError: 地址不是安全的 HTTPS 地址时抛出
+ """
+ parsed = urlsplit(url.strip())
+ if parsed.scheme.lower() != "https":
+ raise ValueError("搜索页 URL 仅支持 https")
+ if not parsed.hostname or parsed.username or parsed.password:
+ raise ValueError("搜索页 URL 必须包含有效域名，且不能内嵌账号或密码")
+
+ host = parsed.hostname.encode("idna").decode("ascii").lower()
+ default_port = 80 if parsed.scheme.lower() == "http" else 443
+ port = parsed.port
+ authority = host if not port or port == default_port else f"{host}:{port}"
+ origin = f"{parsed.scheme.lower()}://{authority}"
+ path = parsed.path or "/"
+ if not path.startswith("/"):
+ path = f"/{path}"
+ if _is_sensitive_name(path):
+ raise ValueError("搜索页 URL 路径疑似包含凭据，请改用不带凭据的搜索地址")
+ return origin, host, path
+
+
+def _normalize_user_start_url(value: str) -> str:
+ """允许普通用户只输入域名，并统一补成可访问的 HTTPS 地址。"""
+ normalized = value.strip()
+ if "://" not in normalized:
+ normalized = f"https://{normalized}"
+ _normalize_origin(normalized)
+ return normalized
+
+
+def _make_site_id(domain: str) -> str:
+ """根据域名生成仅含小写字母、数字和连字符的安全站点标识。"""
+ normalized = domain.lower()
+ if normalized.startswith("www."):
+ normalized = normalized[4:]
+ slug = re.sub(r"[^a-z0-9]+", "-", normalized).strip("-")
+ slug = re.sub(r"-+", "-", slug)[:63].rstrip("-")
+ if not slug:
+ raise ValueError("无法根据域名生成安全的站点标识")
+ if not slug[0].isalnum():
+ slug = f"site-{slug}"
+ return slug
+
+
+def _prepare_capture_request(
+ url: str,
+ keyword: str,
+ site_name: Optional[str] = None,
+) -> _CaptureRequest:
+ """
+ 规范化搜索请求，并将公开参数中的关键词替换为占位符。
+
+ :param url: 包含搜索参数或 ``{keyword}`` 占位符的搜索页 URL
+ :param keyword: 用于本次采集的搜索关键词
+ :param site_name: 可选站点显示名称，留空时使用域名
+ :return: 可用于请求与写入采集包的结构化参数
+ :raises ValueError: URL 不含可识别的搜索参数或包含敏感参数时抛出
+ """
+ if not keyword.strip():
+ raise ValueError("搜索关键词不能为空")
+
+ origin, domain, path = _normalize_origin(url)
+ parsed = urlsplit(url.strip())
+ pairs = parse_qsl(parsed.query, keep_blank_values=True)
+ request_params: dict[str, str] = {}
+ public_params: dict[str, str] = {}
+ keyword_assigned = False
+
+ for name, value in pairs:
+ if _is_sensitive_name(name):
+ raise ValueError(f"搜索页 URL 参数 {name!r} 疑似包含凭据，请先移除")
+ normalized_name = name.lower()
+ has_keyword_value = "{keyword}" in value or value == keyword
+ is_keyword_parameter = has_keyword_value or (
+ normalized_name in SEARCH_PARAMETER_NAMES and not keyword_assigned
+ )
+ if is_keyword_parameter:
+ if keyword_assigned:
+ raise ValueError("URL 中存在多个关键词参数，请只保留一个搜索参数")
+ request_params[name] = keyword
+ public_params[name] = "{keyword}"
+ keyword_assigned = True
+ else:
+ request_params[name] = value
+ public_params[name] = value
+
+ if not keyword_assigned:
+ raise ValueError(
+ "URL 中未找到搜索参数，请粘贴搜索后的 URL，或将关键词值写成 {keyword}"
+ )
+
+ site_id = _make_site_id(domain)
+ request_url = urlunsplit((urlsplit(origin).scheme, urlsplit(origin).netloc, path, "", ""))
+ return _CaptureRequest(
+ url=request_url,
+ origin=origin,
+ path=path,
+ params=request_params,
+ public_params=public_params,
+ domain=domain,
+ site_id=site_id,
+ site_name=site_name.strip() if site_name and site_name.strip() else domain,
+ )
+
+
+def _same_origin(url: str, expected_origin: str) -> bool:
+ """判断地址是否与预期 origin 完全同源。"""
+ try:
+ origin, _, _ = _normalize_origin(url)
+ except (TypeError, ValueError):
+ return False
+ return origin == expected_origin
+
+
+def _find_browser_executable() -> Path:
+ """查找本机已安装的 Chrome、Edge 或 Chromium 可执行文件。"""
+ candidates: list[Path] = []
+ if sys.platform == "darwin":
+ candidates.extend(
+ Path(path)
+ for path in (
+ "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome",
+ "/Applications/Microsoft Edge.app/Contents/MacOS/Microsoft Edge",
+ "/Applications/Chromium.app/Contents/MacOS/Chromium",
+ str(Path.home() / "Applications/Google Chrome.app/Contents/MacOS/Google Chrome"),
+ str(Path.home() / "Applications/Microsoft Edge.app/Contents/MacOS/Microsoft Edge"),
+ )
+ )
+ elif os.name == "nt":
+ roots = [
+ os.environ.get("PROGRAMFILES"),
+ os.environ.get("PROGRAMFILES(X86)"),
+ os.environ.get("LOCALAPPDATA"),
+ ]
+ relative_paths = (
+ "Google/Chrome/Application/chrome.exe",
+ "Microsoft/Edge/Application/msedge.exe",
+ "Chromium/Application/chrome.exe",
+ )
+ candidates.extend(
+ Path(root) / relative_path
+ for root in roots
+ if root
+ for relative_path in relative_paths
+ )
+ else:
+ for command in (
+ "google-chrome",
+ "google-chrome-stable",
+ "microsoft-edge",
+ "microsoft-edge-stable",
+ "chromium",
+ "chromium-browser",
+ ):
+ executable = shutil.which(command)
+ if executable:
+ candidates.append(Path(executable))
+
+ for candidate in candidates:
+ if candidate.is_file():
+ return candidate
+ raise RuntimeError("未找到 Chrome、Edge 或 Chromium，请先安装其中一个浏览器")
+
+
+def _reserve_loopback_port() -> int:
+ """让操作系统分配一个仅供本机临时浏览器使用的空闲端口。"""
+ with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server:
+ server.bind(("127.0.0.1", 0))
+ return int(server.getsockname()[1])
+
+
+def _load_cdp_json(port: int, path: str) -> object:
+ """从本机 Chrome DevTools HTTP 接口读取 JSON。"""
+ session = requests.Session()
+ session.trust_env = False
+ try:
+ response = session.get(
+ f"http://127.0.0.1:{port}{path}",
+ timeout=CDP_HTTP_TIMEOUT_SECONDS,
+ )
+ response.raise_for_status()
+ return response.json()
+ except (requests.RequestException, ValueError) as error:
+ raise RuntimeError(f"无法读取本机浏览器状态：{str(error)}") from error
+ finally:
+ session.close()
+
+
+def _wait_for_browser_session(process: subprocess.Popen, port: int) -> str:
+ """等待临时浏览器启动并返回浏览器级 CDP WebSocket 地址。"""
+ deadline = time.monotonic() + BROWSER_START_TIMEOUT_SECONDS
+ last_error: Optional[Exception] = None
+ while time.monotonic() < deadline:
+ if process.poll() is not None:
+ raise RuntimeError("浏览器启动后立即退出，请关闭已有异常进程后重试")
+ try:
+ version = _load_cdp_json(port, "/json/version")
+ if isinstance(version, dict):
+ websocket_url = version.get("webSocketDebuggerUrl")
+ if isinstance(websocket_url, str) and websocket_url:
+ return websocket_url
+ except RuntimeError as error:
+ last_error = error
+ time.sleep(0.2)
+ detail = f"：{str(last_error)}" if last_error else ""
+ raise RuntimeError(f"等待浏览器启动超时{detail}")
+
+
+@contextmanager
+def _launch_browser_session(start_url: str) -> Generator[_BrowserSession, None, None]:
+ """使用一次性用户目录启动可视浏览器，并在结束时清理全部登录状态。"""
+ _normalize_origin(start_url)
+ browser_path = _find_browser_executable()
+ port = _reserve_loopback_port()
+ profile_guard = tempfile.TemporaryDirectory(
+ prefix="moviepilot-site-collector-",
+ ignore_cleanup_errors=True,
+ )
+ command = [
+ str(browser_path),
+ f"--remote-debugging-port={port}",
+ "--remote-debugging-address=127.0.0.1",
+ "--remote-allow-origins=http://127.0.0.1",
+ f"--user-data-dir={profile_guard.name}",
+ "--no-first-run",
+ "--no-default-browser-check",
+ "--disable-sync",
+ "--new-window",
+ start_url,
+ ]
+ process: Optional[subprocess.Popen] = None
+ try:
+ process = subprocess.Popen(
+ command,
+ stdin=subprocess.DEVNULL,
+ stdout=subprocess.DEVNULL,
+ stderr=subprocess.DEVNULL,
+ )
+ browser_websocket_url = _wait_for_browser_session(process, port)
+ session = _BrowserSession(
+ process=process,
+ profile_guard=profile_guard,
+ port=port,
+ browser_websocket_url=browser_websocket_url,
+ )
+ except Exception:
+ if process and process.poll() is None:
+ process.terminate()
+ try:
+ process.wait(timeout=5)
+ except subprocess.TimeoutExpired:
+ process.kill()
+ process.wait(timeout=5)
+ profile_guard.cleanup()
+ raise
+
+ try:
+ yield session
+ finally:
+ session.close()
+
+
+def _select_search_page_target(port: int) -> dict:
+ """从临时浏览器标签页中选择当前 HTTPS 搜索结果页。"""
+ targets = _load_cdp_json(port, "/json/list")
+ if not isinstance(targets, list):
+ raise RuntimeError("浏览器标签页列表格式异常")
+ pages = [
+ target
+ for target in targets
+ if isinstance(target, dict)
+ and target.get("type") == "page"
+ and isinstance(target.get("url"), str)
+ and target["url"].startswith("https://")
+ and isinstance(target.get("webSocketDebuggerUrl"), str)
+ ]
+ if not pages:
+ raise RuntimeError("未找到 HTTPS 页面，请在弹出的浏览器中打开站点搜索结果页")
+
+ def target_priority(target: dict) -> tuple[int, int]:
+ """优先选择 URL 中带明显搜索参数的页面。"""
+ query_names = {
+ name.lower()
+ for name, value in parse_qsl(urlsplit(target["url"]).query, keep_blank_values=True)
+ if value
+ }
+ return (len(query_names & SEARCH_PARAMETER_NAMES), len(urlsplit(target["url"]).query))
+
+ return max(pages, key=target_priority)
+
+
+def _read_browser_capture(session: _BrowserSession) -> _BrowserCapture:
+ """通过本机 CDP 读取当前页面 URL、渲染后 HTML、Cookie 和 User-Agent。"""
+ target = _select_search_page_target(session.port)
+ with _CdpClient(target["webSocketDebuggerUrl"]) as client:
+ current_url = client.evaluate("window.location.href")
+ html_text = client.evaluate("document.documentElement.outerHTML")
+ user_agent = client.evaluate("window.navigator.userAgent")
+ search_inputs = client.evaluate(
+ """
+ Array.from(document.querySelectorAll('input, textarea'))
+ .filter((element) => element.value && !['hidden', 'password'].includes(element.type))
+ .map((element) => ({
+ name: String(element.name || ''),
+ id: String(element.id || ''),
+ value: String(element.value || '')
+ }))
+ """
+ )
+ try:
+ cookie_result = client.call("Network.getAllCookies")
+ except RuntimeError:
+ cookie_result = client.call("Storage.getCookies")
+
+ if not isinstance(current_url, str) or not current_url.startswith("https://"):
+ raise RuntimeError("当前标签页不是 HTTPS 搜索结果页")
+ if not isinstance(html_text, str) or not html_text.strip():
+ raise RuntimeError("无法读取当前搜索结果页内容")
+ if len(html_text.encode("utf-8")) > MAX_RESPONSE_BYTES:
+ raise RuntimeError("当前搜索结果页超过 5 MiB，已停止采集")
+ if not isinstance(user_agent, str):
+ user_agent = ""
+ if not isinstance(search_inputs, list):
+ search_inputs = []
+
+ hostname = urlsplit(current_url).hostname or ""
+ cookie_pairs: list[str] = []
+ cookies = cookie_result.get("cookies", [])
+ if isinstance(cookies, list):
+ for cookie in cookies:
+ if not isinstance(cookie, dict):
+ continue
+ domain = str(cookie.get("domain", "")).lstrip(".").lower()
+ name = cookie.get("name")
+ value = cookie.get("value")
+ if (
+ isinstance(name, str)
+ and isinstance(value, str)
+ and domain
+ and (hostname == domain or hostname.endswith(f".{domain}"))
+ ):
+ cookie_pairs.append(f"{name}={value}")
+ return _BrowserCapture(
+ url=current_url,
+ html=html_text,
+ cookie="; ".join(cookie_pairs),
+ user_agent=user_agent,
+ search_inputs=[item for item in search_inputs if isinstance(item, dict)],
+ )
+
+
+def _infer_search_keyword(capture: _BrowserCapture) -> str:
+ """从当前 URL 与搜索输入框中自动识别本次搜索关键词。"""
+ pairs = parse_qsl(urlsplit(capture.url).query, keep_blank_values=True)
+ named_inputs = {
+ str(item.get("name") or item.get("id") or "").lower(): str(item.get("value") or "")
+ for item in capture.search_inputs
+ if item.get("name") or item.get("id")
+ }
+ candidates: list[tuple[int, str]] = []
+ for name, value in pairs:
+ normalized_name = name.lower()
+ if not value or _is_sensitive_name(name) or normalized_name in NON_SEARCH_PARAMETER_NAMES:
+ continue
+ priority = 0
+ if normalized_name in SEARCH_PARAMETER_NAMES:
+ priority = 3
+ elif named_inputs.get(normalized_name) == value:
+ priority = 2
+ elif any(input_value == value for input_value in named_inputs.values()):
+ priority = 1
+ if priority:
+ candidates.append((priority, value))
+ if not candidates:
+ raise ValueError(
+ "无法从当前地址识别搜索关键词；请确认搜索后地址栏包含关键词，再重新采集"
+ )
+ best_priority = max(priority for priority, _ in candidates)
+ values = {value for priority, value in candidates if priority == best_priority}
+ if len(values) != 1:
+ raise ValueError("当前地址包含多个搜索关键词，请只保留一次普通搜索后重试")
+ return values.pop()
+
+
+def _read_limited_response(response) -> bytes:
+ """
+ 读取流式响应并限制最大字节数。
+
+ :param response: ``RequestUtils`` 返回的响应对象
+ :return: 不超过上限的响应字节
+ :raises ValueError: 响应超过上限时抛出
+ """
+ content_length = response.headers.get("Content-Length")
+ if content_length:
+ try:
+ declared_size = int(content_length)
+ except ValueError:
+ declared_size = 0
+ if declared_size > MAX_RESPONSE_BYTES:
+ raise ValueError("搜索页响应超过 5 MiB，已停止采集")
+
+ chunks: list[bytes] = []
+ size = 0
+ for chunk in response.iter_content(chunk_size=64 * 1024):
+ if not chunk:
+ continue
+ size += len(chunk)
+ if size > MAX_RESPONSE_BYTES:
+ raise ValueError("搜索页响应超过 5 MiB，已停止采集")
+ chunks.append(chunk)
+ return b"".join(chunks)
+
+
+def _decode_html(content: bytes, response) -> str:
+ """依据响应编码解码 HTML，无法确定时回退 UTF-8。"""
+ encoding = response.encoding or getattr(response, "apparent_encoding", None) or "utf-8"
+ try:
+ return content.decode(encoding, errors="replace")
+ except LookupError:
+ return content.decode("utf-8", errors="replace")
+
+
+def _fetch_search_page(request: _CaptureRequest, cookie: str, user_agent: str) -> str:
+ """
+ 通过 ``RequestUtils`` 获取搜索页，并仅跟随同源重定向。
+
+ :param request: 已规范化的搜索请求
+ :param cookie: 仅驻留内存的登录 Cookie
+ :param user_agent: 仅用于本次请求且不写入采集包的浏览器 User-Agent
+ :return: 搜索页 HTML 文本
+ :raises RuntimeError: 请求失败、响应异常或发生跨域重定向时抛出
+ """
+ client = RequestUtils(
+ headers={
+ "Accept": "text/html,application/xhtml+xml",
+ "Accept-Encoding": "identity",
+ "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
+ "Referer": f"{request.origin}/",
+ "User-Agent": user_agent,
+ },
+ cookies=cookie,
+ timeout=REQUEST_TIMEOUT_SECONDS,
+ )
+ current_url = request.url
+ params: Optional[dict[str, str]] = request.params
+
+ for redirect_count in range(MAX_REDIRECTS + 1):
+ response = client.get_res(
+ url=current_url,
+ params=params,
+ allow_redirects=False,
+ stream=True,
+ verify=True,
+ )
+ params = None
+ if response is None:
+ raise RuntimeError("搜索页请求失败，请检查 URL、网络和 Cookie")
+ try:
+ status_code = response.status_code
+ if status_code in REDIRECT_STATUS_CODES:
+ location = response.headers.get("Location")
+ if not location:
+ raise RuntimeError("站点返回了没有 Location 的重定向")
+ next_url = urljoin(str(response.url or current_url), location)
+ if not _same_origin(next_url, request.origin):
+ raise RuntimeError("站点尝试跨域重定向，为避免泄露 Cookie 已停止采集")
+ if redirect_count >= MAX_REDIRECTS:
+ raise RuntimeError("站点重定向次数过多，已停止采集")
+ current_url = next_url
+ continue
+
+ if status_code != 200:
+ raise RuntimeError(f"搜索页返回 HTTP {status_code}")
+ final_url = str(response.url or current_url)
+ if not _same_origin(final_url, request.origin):
+ raise RuntimeError("最终响应与输入 URL 不同源，已停止采集")
+ content_type = response.headers.get("Content-Type", "").lower()
+ if content_type and "html" not in content_type:
+ raise RuntimeError(f"搜索页返回了非 HTML 内容：{content_type}")
+ return _decode_html(_read_limited_response(response), response)
+ finally:
+ response.close()
+
+ raise RuntimeError("站点重定向次数过多，已停止采集")
+
+
+def _is_torrent_link(tag: Tag) -> bool:
+ """判断链接是否指向种子详情或下载资源。"""
+ if tag.name != "a":
+ return False
+ href = str(tag.get("href", ""))
+ parsed = urlsplit(href)
+ path = parsed.path.lower()
+ basename = path.rsplit("/", 1)[-1]
+ if basename in {"details.php", "download.php", "torrent.php", "viewtorrent.php"}:
+ return True
+ query_names = {name.lower() for name, _ in parse_qsl(parsed.query, keep_blank_values=True)}
+ if basename == "torrents.php" and ({"torrentid", "id"} & query_names):
+ return True
+ return bool(TORRENT_PATH_PATTERN.search(path))
+
+
+def _descriptor(tag: Tag) -> str:
+ """合并标签的 id、class 和角色属性，供结构启发式判断使用。"""
+ classes = tag.get("class", [])
+ if isinstance(classes, str):
+ classes = [classes]
+ return " ".join(
+ [str(tag.get("id", "")), *[str(item) for item in classes], str(tag.get("role", ""))]
+ )
+
+
+def _deduplicate_nodes(nodes: list[Tag]) -> list[Tag]:
+ """移除重复或已被父级候选覆盖的 DOM 节点。"""
+ unique: list[Tag] = []
+ seen: set[int] = set()
+ for node in nodes:
+ if id(node) in seen:
+ continue
+ seen.add(id(node))
+ if any(parent is existing for parent in node.parents for existing in unique):
+ continue
+ unique = [existing for existing in unique if node not in existing.parents]
+ unique.append(node)
+ return unique
+
+
+def _find_candidate_nodes(soup: BeautifulSoup) -> list[Tag]:
+ """
+ 从完整页面中定位最小的种子列表相关 DOM 节点。
+
+ :param soup: 完整搜索页 DOM
+ :return: 表格或独立结果项节点，最多保留约定数量
+ """
+ links = [tag for tag in soup.find_all("a") if _is_torrent_link(tag)]
+ tables = _deduplicate_nodes(
+ [table for link in links if (table := link.find_parent("table")) is not None]
+ )
+ descriptor_tables = [
+ table
+ for table in soup.find_all("table")
+ if RESULT_DESCRIPTOR_PATTERN.search(_descriptor(table))
+ and len(table.find_all("tr")) > 1
+ ]
+ tables = _deduplicate_nodes([*tables, *descriptor_tables])
+ if tables:
+ return tables
+
+ items: list[Tag] = []
+ for link in links:
+ item = link.find_parent(["article", "li"])
+ if item is None:
+ for parent in link.parents:
+ if not isinstance(parent, Tag) or parent.name in {"body", "html"}:
+ break
+ if RESULT_DESCRIPTOR_PATTERN.search(_descriptor(parent)):
+ item = parent
+ break
+ if item is None:
+ item = link.find_parent("div")
+ if item is not None:
+ items.append(item)
+ return _deduplicate_nodes(items)[:MAX_RESULT_ROWS]
+
+
+def _trim_table(table: Tag, max_rows: int) -> tuple[Tag, int, int]:
+ """
+ 裁剪表格中的非结果内容并限制种子结果行数。
+
+ :param table: 待裁剪的种子表格
+ :param max_rows: 本表格最多保留的结果行数
+ :return: 裁剪后的表格、保留行数和丢弃行数
+ """
+ cloned = copy.deepcopy(table)
+ rows = [
+ row
+ for row in cloned.find_all("tr")
+ if row.find_parent("table") is cloned
+ ]
+ result_rows = [row for row in rows if any(_is_torrent_link(link) for link in row.find_all("a"))]
+ if not result_rows:
+ result_rows = [row for row in rows if not row.find("th")]
+
+ discarded = 0
+ kept_result_ids = {id(row) for row in result_rows[:max_rows]}
+ for row in rows:
+ is_header = row.find("th") is not None
+ if not is_header and id(row) not in kept_result_ids:
+ row.decompose()
+ discarded += 1
+ return cloned, min(len(result_rows), max_rows), discarded
+
+
+def _sanitize_url(value: str, origin: str, stats: _RedactionStats) -> str:
+ """把 DOM URL 转为不含站点凭据和资源标识的相对结构地址。"""
+ stripped = value.strip()
+ if not stripped or stripped.startswith(("#", "data:")):
+ return stripped
+ if stripped.lower().startswith(("javascript:", "mailto:", "tel:")):
+ stats.redacted_urls += 1
+ return "#"
+
+ absolute_url = urljoin(f"{origin}/", stripped)
+ parsed = urlsplit(absolute_url)
+ if not _same_origin(absolute_url, origin):
+ stats.redacted_urls += 1
+ return "#external-resource"
+
+ path_segments = []
+ redact_next_segment = False
+ for segment in parsed.path.split("/"):
+ if redact_next_segment and segment:
+ path_segments.append("{identity}")
+ redact_next_segment = False
+ continue
+ path_segments.append(_sanitize_path_segment(segment))
+ if segment.lower() in {"author", "member", "members", "profile", "profiles", "user", "users"}:
+ redact_next_segment = True
+ safe_pairs = []
+ for name, _ in parse_qsl(parsed.query, keep_blank_values=True):
+ if _is_sensitive_name(name):
+ continue
+ safe_pairs.append((name, _query_value_placeholder(name)))
+ stats.redacted_urls += 1
+ query = urlencode(safe_pairs)
+ return urlunsplit(("", "", "/".join(path_segments) or "/", query, ""))
+
+
+def _query_value_placeholder(name: str) -> str:
+ """为 URL 查询参数生成不含原值但保留解析语义的合成值。"""
+ normalized_name = name.lower()
+ if (
+ normalized_name in {"cat", "category", "id", "page", "tid", "torrentid", "type"}
+ or normalized_name.endswith("id")
+ ):
+ return "1"
+ return "{value}"
+
+
+def _sanitize_path_segment(segment: str) -> str:
+ """保留路由语义，并替换 URL 路径中的资源名和对象标识。"""
+ if not segment:
+ return segment
+ if _is_sensitive_name(segment):
+ return "{redacted}"
+ if segment.isdigit():
+ return "{id}"
+ if HIGH_ENTROPY_VALUE_PATTERN.fullmatch(segment) or not segment.isascii():
+ return "{value}"
+
+ dot_index = segment.rfind(".")
+ suffix = segment[dot_index:].lower() if dot_index > 0 else ""
+ if suffix and suffix not in STRUCTURAL_PAGE_EXTENSIONS:
+ safe_suffix = suffix if re.fullmatch(r"\.[a-z0-9]{1,6}", suffix) else ""
+ return f"{{asset}}{safe_suffix}"
+ if len(segment) >= 16 and re.search(r"[-_\d]", segment):
+ return "{value}"
+ return segment
+
+
+def _placeholder_for_text(value: str) -> str:
+ """根据结果文本形态生成不携带原始内容的占位值。"""
+ if not value.strip():
+ return value
+ if SIZE_PATTERN.fullmatch(value):
+ return "1.00 GiB"
+ if DATE_TIME_PATTERN.search(value):
+ return "2000-01-01 00:00"
+ if NUMBER_PATTERN.fullmatch(value):
+ return "0"
+ return "[REDACTED]"
+
+
+def _sanitize_selector_token(value: str) -> str:
+ """脱敏 class/id 中的身份后缀，同时保留可用于选择器的结构前缀。"""
+ redacted = EMAIL_PATTERN.sub("redacted", value)
+ redacted = IP_ADDRESS_PATTERN.sub("redacted", redacted)
+ if HIGH_ENTROPY_VALUE_PATTERN.search(redacted):
+ return "redacted"
+ identity_match = re.match(
+ r"^(profile|uploader|uploaded[-_]by|user|username|member|author)([-_]).+$",
+ redacted,
+ flags=re.IGNORECASE,
+ )
+ if identity_match:
+ return f"{identity_match.group(1)}{identity_match.group(2)}redacted"
+ return redacted
+
+
+def _sanitize_selector_attribute(value: object) -> Union[list[str], str]:
+ """逐项脱敏 BeautifulSoup 的 class 列表或普通 id 字符串。"""
+ if isinstance(value, list):
+ return [_sanitize_selector_token(str(item)) for item in value]
+ return _sanitize_selector_token(str(value))
+
+
+def _redact_result_text(node: Tag, stats: _RedactionStats) -> None:
+ """替换种子结果项中的标题、用户、时间和统计值。"""
+ for text_node in list(node.find_all(string=True)):
+ if not isinstance(text_node, NavigableString) or not text_node.strip():
+ continue
+ parent = text_node.parent
+ if parent and parent.name in {"th", "thead"}:
+ continue
+ replacement = _placeholder_for_text(str(text_node))
+ if replacement != str(text_node):
+ text_node.replace_with(replacement)
+ stats.redacted_result_values += 1
+
+
+def _sanitize_dom(root: Tag, origin: str, stats: _RedactionStats) -> None:
+ """移除活动内容、隐藏字段、身份信息和可疑属性。"""
+ for comment in list(root.find_all(string=lambda value: isinstance(value, Comment))):
+ comment.extract()
+ stats.discarded_nodes += 1
+
+ for node in list(root.find_all(["script", "style", "iframe", "object", "embed", "noscript", "template"])):
+ node.decompose()
+ stats.discarded_nodes += 1
+
+ for hidden_input in list(root.find_all("input", attrs={"type": re.compile("hidden", re.IGNORECASE)})):
+ hidden_input.decompose()
+ stats.discarded_nodes += 1
+
+ for tag in root.find_all(True):
+ descriptor = " ".join(
+ [
+ _descriptor(tag),
+ str(tag.get("href", "")),
+ str(tag.get("name", "")),
+ ]
+ )
+ is_identity_element = (
+ tag.name in {"a", "span", "div"}
+ and bool(IDENTITY_DESCRIPTOR_PATTERN.search(descriptor))
+ )
+ if is_identity_element:
+ tag.clear()
+ tag.append("[REDACTED_USER]")
+ stats.redacted_identity_values += 1
+
+ for attribute_name in list(tag.attrs):
+ lowered_name = attribute_name.lower()
+ is_allowed = (
+ lowered_name in ALLOWED_ATTRIBUTE_NAMES
+ or lowered_name in URL_ATTRIBUTE_NAMES
+ or lowered_name in URL_SET_ATTRIBUTE_NAMES
+ or lowered_name in TEXT_ATTRIBUTE_NAMES
+ )
+ if not is_allowed:
+ del tag.attrs[attribute_name]
+ stats.removed_attributes += 1
+ continue
+ attribute_value = tag.attrs[attribute_name]
+ if lowered_name in {"class", "id"}:
+ tag.attrs[attribute_name] = _sanitize_selector_attribute(attribute_value)
+ continue
+ if lowered_name in URL_ATTRIBUTE_NAMES:
+ if is_identity_element:
+ tag.attrs[attribute_name] = "#redacted-identity"
+ stats.redacted_urls += 1
+ else:
+ tag.attrs[attribute_name] = _sanitize_url(str(attribute_value), origin, stats)
+ continue
+ if lowered_name in URL_SET_ATTRIBUTE_NAMES:
+ tag.attrs[attribute_name] = "[REDACTED_URL_SET]"
+ stats.redacted_urls += 1
+ continue
+ if lowered_name in TEXT_ATTRIBUTE_NAMES:
+ tag.attrs[attribute_name] = _placeholder_for_text(str(attribute_value))
+ stats.redacted_result_values += 1
+ continue
+ if lowered_name.startswith("data-") and lowered_name != "data-capture-kind":
+ tag.attrs[attribute_name] = _placeholder_for_text(str(attribute_value))
+ stats.redacted_sensitive_values += 1
+ continue
+ serialized_value = " ".join(attribute_value) if isinstance(attribute_value, list) else str(attribute_value)
+ if lowered_name in {"content", "value"}:
+ tag.attrs[attribute_name] = "[REDACTED]"
+ stats.redacted_sensitive_values += 1
+ elif HIGH_ENTROPY_VALUE_PATTERN.search(serialized_value):
+ tag.attrs[attribute_name] = "redacted"
+ stats.redacted_sensitive_values += 1
+
+ for text_node in list(root.find_all(string=True)):
+ value = str(text_node)
+ redacted = EMAIL_PATTERN.sub("[REDACTED_EMAIL]", value)
+ redacted = IP_ADDRESS_PATTERN.sub("[REDACTED_IP]", redacted)
+ redacted = HIGH_ENTROPY_VALUE_PATTERN.sub("[REDACTED_VALUE]", redacted)
+ if redacted != value:
+ text_node.replace_with(redacted)
+ stats.redacted_sensitive_values += 1
+
+
+def _sanitize_search_html(
+ html: str,
+ origin: str,
+ keyword: str,
+) -> tuple[str, int, dict]:
+ """
+ 本地裁剪并脱敏种子列表 DOM，不保留原始页面。
+
+ :param html: 原始搜索页 HTML，仅在内存中存在
+ :param origin: 搜索站点 origin
+ :param keyword: 本次搜索关键词
+ :return: 脱敏 HTML、采集行数和脱敏报告
+ :raises ValueError: 页面中无法定位种子结果结构时抛出
+ """
+ soup = BeautifulSoup(html, "html.parser")
+ candidates = _find_candidate_nodes(soup)
+ if not candidates:
+ raise ValueError("页面中未找到种子列表，请确认 Cookie 有效且 URL 指向搜索结果页")
+
+ output = BeautifulSoup(
+ 'Site search structure',
+ "html.parser",
+ )
+ container = output.new_tag("main", attrs={"data-capture-kind": "torrent-search"})
+ output.body.append(container)
+ stats = _RedactionStats()
+ row_count = 0
+
+ for candidate in candidates:
+ remaining_rows = MAX_RESULT_ROWS - row_count
+ if remaining_rows <= 0:
+ break
+ if candidate.name == "table":
+ cloned, captured_rows, discarded = _trim_table(candidate, remaining_rows)
+ row_count += captured_rows
+ stats.discarded_nodes += discarded
+ else:
+ cloned = copy.deepcopy(candidate)
+ row_count += 1
+ _redact_result_text(cloned, stats)
+ container.append(cloned)
+
+ _sanitize_dom(output, origin, stats)
+ if row_count < MIN_RESULT_ROWS:
+ raise ValueError(
+ f"仅定位到 {row_count} 条种子结果，请换用更宽泛的关键词，至少采集 {MIN_RESULT_ROWS} 条"
+ )
+ sanitized_html = str(output)
+ if keyword:
+ sanitized_html, count = re.subn(re.escape(keyword), "{keyword}", sanitized_html, flags=re.IGNORECASE)
+ stats.redacted_sensitive_values += count
+ report = {
+ "redacted": True,
+ "contains_credentials": False,
+ "strategy": "local-dom-cropping-and-allowlist",
+ "captured_rows": row_count,
+ "changes": {
+ "discarded_nodes": stats.discarded_nodes,
+ "removed_attributes": stats.removed_attributes,
+ "redacted_urls": stats.redacted_urls,
+ "redacted_identity_values": stats.redacted_identity_values,
+ "redacted_result_values": stats.redacted_result_values,
+ "redacted_sensitive_values": stats.redacted_sensitive_values,
+ },
+ }
+ return sanitized_html, row_count, report
+
+
+def _sha256(value: bytes) -> str:
+ """计算字节内容的 SHA-256 摘要。"""
+ return hashlib.sha256(value).hexdigest()
+
+
+def _json_bytes(value: dict) -> bytes:
+ """把字典序列化为稳定、可读的 UTF-8 JSON。"""
+ return (json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True) + "\n").encode("utf-8")
+
+
+def _credential_values(cookie: str) -> list[str]:
+ """提取需要在最终采集包中执行二次检查的较长凭据值。"""
+ raw_values = [cookie.strip()]
+ for part in cookie.split(";"):
+ _, separator, value = part.partition("=")
+ if separator:
+ raw_values.append(value.strip())
+ values = {cookie.strip()} if cookie.strip() else set()
+ values.update({
+ variant
+ for raw_value in raw_values
+ for variant in (raw_value, unquote(raw_value), quote(raw_value, safe=""))
+ if len(variant) >= 8
+ })
+ return sorted(values)
+
+
+def _verify_payload(payload: dict[str, bytes], cookie: str) -> None:
+ """
+ 验证采集包文件集合及内容中不包含已知凭据。
+
+ :param payload: 待写入 ZIP 的根目录文件
+ :param cookie: 用户本次输入的 Cookie，仅用于内存比对
+ :raises ValueError: 协议不完整或发现凭据值时抛出
+ """
+ if tuple(payload) != ARCHIVE_FILE_NAMES:
+ raise ValueError("采集包文件协议不完整")
+ combined = b"\n".join(payload.values())
+ for value in _credential_values(cookie):
+ if value.encode("utf-8") in combined:
+ raise ValueError("脱敏检查发现采集包仍包含凭据值，已停止写入")
+
+ combined_text = combined.decode("utf-8", errors="replace")
+ if EMAIL_PATTERN.search(combined_text) or IP_ADDRESS_PATTERN.search(combined_text):
+ raise ValueError("脱敏检查发现采集包仍包含身份信息，已停止写入")
+
+ for name, content in payload.items():
+ scan_text = content.decode("utf-8", errors="replace")
+ if name == "manifest.json":
+ manifest = json.loads(scan_text)
+ manifest["files"] = {}
+ scan_text = json.dumps(manifest, ensure_ascii=False)
+ if HIGH_ENTROPY_VALUE_PATTERN.search(scan_text):
+ raise ValueError("脱敏检查发现采集包仍包含疑似凭据，已停止写入")
+
+
+def _build_payload(request: _CaptureRequest, html: str, cookie: str) -> dict[str, bytes]:
+ """构造符合站点适配协议的四个根目录文件。"""
+ sanitized_html, row_count, report = _sanitize_search_html(
+ html=html,
+ origin=request.origin,
+ keyword=next(
+ request.params[name]
+ for name, value in request.public_params.items()
+ if value == "{keyword}"
+ ),
+ )
+ html_bytes = sanitized_html.encode("utf-8")
+ captured_at = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
+ manifest = {
+ "format_version": FORMAT_VERSION,
+ "collector_version": COLLECTOR_VERSION,
+ "site": {
+ "id": request.site_id,
+ "name": request.site_name,
+ "domain": request.origin,
+ "public": False,
+ },
+ "capture": {
+ "kind": "search",
+ "captured_at": captured_at,
+ "row_count": row_count,
+ },
+ "files": {"search.html": _sha256(html_bytes)},
+ "privacy": {
+ "redacted": True,
+ "contains_credentials": False,
+ },
+ }
+ public_request = {
+ "method": "get",
+ "path": request.path,
+ "params": request.public_params,
+ "origin": request.origin,
+ }
+ payload = {
+ "manifest.json": _json_bytes(manifest),
+ "request.json": _json_bytes(public_request),
+ "search.html": html_bytes,
+ "redaction-report.json": _json_bytes(report),
+ }
+ _verify_payload(payload, cookie)
+ return payload
+
+
+def _write_archive(payload: dict[str, bytes], output_dir: Path, site_id: str) -> Path:
+ """把固定文件集合写入 ZIP 根目录并返回文件路径。"""
+ output_dir.mkdir(parents=True, exist_ok=True)
+ timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
+ archive_path = output_dir / f"moviepilot-site-capture-{site_id}-{timestamp}.zip"
+ with zipfile.ZipFile(archive_path, "w", compression=zipfile.ZIP_DEFLATED) as archive:
+ for name, content in payload.items():
+ archive.writestr(name, content)
+ return archive_path
+
+
+def collect_site_capture(
+ url: str,
+ keyword: str,
+ cookie: str,
+ output_dir: Path,
+ user_agent: str = DEFAULT_USER_AGENT,
+ site_name: Optional[str] = None,
+) -> Path:
+ """
+ 获取、脱敏并打包一个站点搜索页。
+
+ :param url: 搜索页 URL，需含搜索参数或 ``{keyword}`` 占位符
+ :param keyword: 本次请求使用的搜索关键词
+ :param cookie: 登录 Cookie，仅在采集期间驻留内存
+ :param output_dir: ZIP 输出目录
+ :param user_agent: 可选浏览器 User-Agent，仅用于请求且不会写入采集包
+ :param site_name: 可选站点显示名称，留空时使用域名
+ :return: 生成的 ZIP 文件路径
+ """
+ request = _prepare_capture_request(url=url, keyword=keyword, site_name=site_name)
+ html = _fetch_search_page(request=request, cookie=cookie, user_agent=user_agent)
+ payload = _build_payload(request=request, html=html, cookie=cookie)
+ return _write_archive(payload=payload, output_dir=output_dir, site_id=request.site_id)
+
+
+def collect_site_capture_with_browser(start_url: str, output_dir: Path) -> Path:
+ """
+ 启动临时浏览器，让用户完成登录和搜索后生成脱敏采集包。
+
+ :param start_url: 用户要访问的站点 HTTPS 地址
+ :param output_dir: ZIP 输出目录
+ :return: 生成的 ZIP 文件路径
+ """
+ normalized_start_url = _normalize_user_start_url(start_url)
+ origin, _, _ = _normalize_origin(normalized_start_url)
+ with _launch_browser_session(normalized_start_url) as browser_session:
+ print("\n浏览器已打开，请在临时窗口中完成以下操作：")
+ print("1. 登录站点。")
+ print("2. 搜索一个能返回至少 3 条结果的常见关键词。")
+ print("3. 保持搜索结果页打开，然后回到这里按回车。")
+ input("\n完成后按回车开始脱敏采集：")
+ capture = _read_browser_capture(browser_session)
+
+ capture_origin, _, _ = _normalize_origin(capture.url)
+ if capture_origin != origin:
+ print(f"提示：浏览器最终进入了 {capture_origin}，将按该地址生成站点配置。")
+ keyword = _infer_search_keyword(capture)
+ request = _prepare_capture_request(
+ url=capture.url,
+ keyword=keyword,
+ )
+ payload = _build_payload(request=request, html=capture.html, cookie=capture.cookie)
+ return _write_archive(payload=payload, output_dir=output_dir, site_id=request.site_id)
+
+
+def _parse_args() -> argparse.Namespace:
+ """解析普通浏览器模式和高级手动 Cookie 模式的命令行参数。"""
+ parser = argparse.ArgumentParser(description="采集并脱敏 MoviePilot 站点适配所需的搜索页结构")
+ parser.add_argument(
+ "--output-dir",
+ type=Path,
+ default=Path.cwd(),
+ help="采集 ZIP 输出目录，默认当前目录",
+ )
+ parser.add_argument(
+ "--manual-cookie",
+ action="store_true",
+ help="高级模式：手工输入搜索 URL、关键词和 Cookie，不启动临时浏览器",
+ )
+ return parser.parse_args()
+
+
+def main() -> int:
+ """交互式读取采集参数并生成站点适配 ZIP。"""
+ args = _parse_args()
+ print("MoviePilot 站点适配采集器")
+ print("原始页面不会落盘；临时浏览器关闭后会清理登录状态，输出前会裁剪并脱敏。")
+ try:
+ if args.manual_cookie:
+ url = input("搜索结果页 URL：").strip()
+ keyword = input("刚才使用的搜索关键词：").strip()
+ cookie = getpass.getpass("Cookie（输入不会回显）：").strip()
+ if not cookie:
+ raise ValueError("Cookie 不能为空")
+ archive_path = collect_site_capture(
+ url=url,
+ keyword=keyword,
+ cookie=cookie,
+ output_dir=args.output_dir,
+ )
+ else:
+ start_url = input("请输入站点地址（例如 https://tracker.example.com）：").strip()
+ archive_path = collect_site_capture_with_browser(
+ start_url=start_url,
+ output_dir=args.output_dir,
+ )
+ except (EOFError, KeyboardInterrupt):
+ print("\n已取消采集。", file=sys.stderr)
+ return 130
+ except (RuntimeError, ValueError) as error:
+ print(f"采集失败：{str(error)}", file=sys.stderr)
+ return 1
+
+ print(f"采集完成：{archive_path}")
+ print("请在站点适配 Feature Request 中附加此 ZIP；不要上传 Cookie 或原始 HTML。")
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/scripts/site_adapter_collector.spec b/scripts/site_adapter_collector.spec
new file mode 100644
index 000000000..087802c7c
--- /dev/null
+++ b/scripts/site_adapter_collector.spec
@@ -0,0 +1,41 @@
+# -*- mode: python ; coding: utf-8 -*-
+"""将站点适配采集器构建为不依赖本地 Python 的单文件程序。"""
+
+from pathlib import Path
+
+
+PROJECT_ROOT = Path(SPECPATH).resolve().parent
+ENTRYPOINT = PROJECT_ROOT / "scripts" / "site_adapter_collector.py"
+analysis = Analysis(
+ [str(ENTRYPOINT)],
+ pathex=[str(PROJECT_ROOT)],
+ binaries=[],
+ datas=[],
+ hiddenimports=[],
+ hookspath=[],
+ hooksconfig={},
+ runtime_hooks=[],
+ excludes=[],
+ noarchive=False,
+ optimize=0,
+)
+python_archive = PYZ(analysis.pure)
+
+executable = EXE(
+ python_archive,
+ analysis.scripts,
+ analysis.binaries,
+ analysis.datas,
+ [],
+ name="moviepilot-site-collector",
+ debug=False,
+ bootloader_ignore_signals=False,
+ strip=False,
+ upx=False,
+ console=True,
+ disable_windowed_traceback=False,
+ argv_emulation=False,
+ target_arch=None,
+ codesign_identity=None,
+ entitlements_file=None,
+)
diff --git a/scripts/site_adapter_collector_requirements.txt b/scripts/site_adapter_collector_requirements.txt
new file mode 100644
index 000000000..c9693f345
--- /dev/null
+++ b/scripts/site_adapter_collector_requirements.txt
@@ -0,0 +1,4 @@
+beautifulsoup4~=4.15.0
+PyInstaller>=6.14,<7.0
+requests>=2.32,<3.0
+websocket-client~=1.9.0
diff --git a/scripts/start-site-adapter-collector.command b/scripts/start-site-adapter-collector.command
new file mode 100755
index 000000000..534c95623
--- /dev/null
+++ b/scripts/start-site-adapter-collector.command
@@ -0,0 +1,29 @@
+#!/usr/bin/env bash
+
+set -u
+
+# 从解压目录启动 macOS 单文件采集器，并在结束后保留终端窗口供用户查看结果。
+main() {
+ local script_dir=""
+ local collector_path=""
+ local status=0
+
+ script_dir="$(CDPATH= cd -- "$(dirname -- "$0")" && pwd)"
+ collector_path="$script_dir/moviepilot-site-collector-macos"
+ if [[ ! -f "$collector_path" ]]; then
+ echo "未找到 moviepilot-site-collector-macos，请完整解压 ZIP 后再双击启动。" >&2
+ status=1
+ else
+ chmod +x "$collector_path"
+ cd "$script_dir" || status=1
+ if [[ "$status" -eq 0 ]]; then
+ "$collector_path" || status=$?
+ fi
+ fi
+
+ echo
+ read -r -p "按回车关闭窗口..." _
+ exit "$status"
+}
+
+main "$@"
diff --git a/tests/test_site_adapter_collector.py b/tests/test_site_adapter_collector.py
new file mode 100644
index 000000000..3144bfad6
--- /dev/null
+++ b/tests/test_site_adapter_collector.py
@@ -0,0 +1,469 @@
+import hashlib
+import json
+import zipfile
+from contextlib import contextmanager
+from pathlib import Path
+
+import pytest
+
+from scripts import site_adapter_collector as collector
+
+
+SEARCH_HTML = """
+
+
+
+
+
+
+
+
+ | 标题 | 大小 | 做种 | 发布者 |
|---|
+
+
+
+
+ 私密电影标题
+ ![]()
+ |
+ 18.45 GiB |
+ 42 |
+ alice |
+ 刚刚 |
+
+
+ | 另一部私密电影 |
+ 8.00 GiB | 12 | bob |
+
+
+ | 第三部私密电影 |
+ 2.00 GiB | 8 | carol |
+
+
+
+
+
+
+"""
+
+
+class _FakeResponse:
+ """提供采集器测试所需的最小响应接口。"""
+
+ def __init__(
+ self,
+ status_code: int,
+ url: str,
+ body: bytes = b"",
+ headers: dict = None,
+ ):
+ """初始化状态、地址、响应体和响应头。"""
+ self.status_code = status_code
+ self.url = url
+ self._body = body
+ self.headers = headers or {}
+ self.encoding = "utf-8"
+ self.apparent_encoding = "utf-8"
+ self.closed = False
+
+ def iter_content(self, chunk_size: int):
+ """按给定块大小返回内存响应体。"""
+ for start in range(0, len(self._body), chunk_size):
+ yield self._body[start:start + chunk_size]
+
+ def close(self) -> None:
+ """记录响应已关闭。"""
+ self.closed = True
+
+
+class _FakeClient:
+ """按顺序返回预设响应，避免测试产生真实外网请求。"""
+
+ def __init__(self, responses: list[_FakeResponse]):
+ """保存待返回响应和请求记录。"""
+ self.responses = responses
+ self.calls: list[dict] = []
+
+ def get_res(self, **kwargs) -> _FakeResponse:
+ """记录请求参数并返回下一条响应。"""
+ self.calls.append(kwargs)
+ return self.responses.pop(0)
+
+
+def test_prepare_capture_request_uses_safe_slug_and_keyword_placeholder():
+ """搜索请求应生成安全站点标识和严格关键词占位符。"""
+ request = collector._prepare_capture_request(
+ url="https://Tracker.Example.com/torrents.php?search={keyword}&category=1",
+ keyword="Movie 2026",
+ )
+
+ assert request.site_id == "tracker-example-com"
+ assert request.origin == "https://tracker.example.com"
+ assert request.path == "/torrents.php"
+ assert request.params == {"search": "Movie 2026", "category": "1"}
+ assert request.public_params == {"search": "{keyword}", "category": "1"}
+
+
+@pytest.mark.parametrize(
+ "url",
+ [
+ "ftp://tracker.example.com/torrents.php?search={keyword}",
+ "http://tracker.example.com/torrents.php?search={keyword}",
+ "https://tracker.example.com/passkey/value/torrents.php?search={keyword}",
+ "https://tracker.example.com/torrents.php?token=value&search={keyword}",
+ "https://tracker.example.com/torrents.php?search={keyword}&q={keyword}",
+ ],
+)
+def test_prepare_capture_request_rejects_unsafe_urls(url: str):
+ """采集器应拒绝非 HTTPS 地址及疑似内嵌凭据的 URL。"""
+ with pytest.raises(ValueError):
+ collector._prepare_capture_request(url=url, keyword="Movie")
+
+
+def test_user_details_link_is_not_a_torrent_result():
+ """用户详情链接不得被误判为种子详情链接。"""
+ soup = collector.BeautifulSoup(
+ 'user',
+ "html.parser",
+ )
+
+ assert collector._is_torrent_link(soup.a) is False
+
+
+def test_sanitize_search_html_crops_and_redacts_result_structure():
+ """脱敏结果应仅保留种子列表结构，不包含身份、标题和凭据。"""
+ sanitized_html, row_count, report = collector._sanitize_search_html(
+ html=SEARCH_HTML,
+ origin="https://tracker.example.com",
+ keyword="Movie",
+ )
+ lowered = sanitized_html.lower()
+
+ assert row_count == 3
+ assert "torrent-table" in sanitized_html
+ assert "torrent-row" in sanitized_html
+ assert "torrent-listings-global-freeleech" in sanitized_html
+ assert "torrent-session-token-cell" in sanitized_html
+ assert "unrelated footer" not in sanitized_html
+ assert "alice" not in lowered
+ assert "私密电影标题" not in sanitized_html
+ assert "private-title" not in sanitized_html
+ assert "embedded-secret-value" not in sanitized_html
+ assert "passkey" not in lowered
+ assert "csrf" not in lowered
+ assert "window.token" not in lowered
+ sanitized_soup = collector.BeautifulSoup(sanitized_html, "html.parser")
+ result_row = sanitized_soup.select_one("tr.torrent-row")
+ assert result_row["id"] == "uploader-redacted"
+ assert "profile-redacted" in result_row["class"]
+ assert "redacted" in result_row["class"]
+ assert "peer-redacted" in result_row["class"]
+ assert result_row["data-id"] == "0"
+ assert not result_row.has_attr("data-private-user")
+ assert not result_row.has_attr("unknown")
+ assert sanitized_soup.select_one("a.torrent-title")["href"] == "/details.php?id=1"
+ assert sanitized_soup.select_one("a.username")["href"] == "#redacted-identity"
+ assert sanitized_soup.select_one("a.torrent-title")["title"] == "[REDACTED]"
+ assert sanitized_soup.select_one("span.date-added")["title"] == "2000-01-01 00:00"
+ assert report["redacted"] is True
+ assert report["contains_credentials"] is False
+ assert report["captured_rows"] == 3
+
+
+def test_nested_nexus_table_counts_only_outer_result_rows():
+ """NexusPHP 资源名内嵌表格不得重复计数或被当成外层结果裁剪。"""
+ result_rows = "".join(
+ f"""
+
+ |
+
+ |
+ {index} |
+
+ """
+ for index in range(1, 31)
+ )
+ html = f"""
+
+
+
+ """
+
+ sanitized_html, row_count, report = collector._sanitize_search_html(
+ html=html,
+ origin="https://tracker.example.com",
+ keyword="Movie",
+ )
+ soup = collector.BeautifulSoup(sanitized_html, "html.parser")
+
+ assert row_count == collector.MAX_RESULT_ROWS
+ assert len(soup.select("tr.outer-result-row")) == collector.MAX_RESULT_ROWS
+ assert len(soup.select("table.torrentname")) == collector.MAX_RESULT_ROWS
+ assert report["captured_rows"] == collector.MAX_RESULT_ROWS
+
+
+def test_fetch_search_page_blocks_cross_origin_redirect(monkeypatch):
+ """携带 Cookie 的采集请求遇到跨 origin 重定向时必须立即停止。"""
+ response = _FakeResponse(
+ status_code=302,
+ url="https://tracker.example.com/torrents.php?search=Movie",
+ headers={"Location": "https://login.example.net/sign-in"},
+ )
+ client = _FakeClient([response])
+ monkeypatch.setattr(collector, "RequestUtils", lambda **_: client)
+ request = collector._prepare_capture_request(
+ url="https://tracker.example.com/torrents.php?search={keyword}",
+ keyword="Movie",
+ )
+
+ with pytest.raises(RuntimeError, match="跨域重定向"):
+ collector._fetch_search_page(request, "session=very-secret-cookie", "Browser UA")
+
+ assert len(client.calls) == 1
+ assert client.calls[0]["allow_redirects"] is False
+ assert client.calls[0]["verify"] is True
+ assert response.closed is True
+
+
+def test_read_limited_response_rejects_oversized_content_length():
+ """响应头声明超过容量上限时不应读取响应体。"""
+ response = _FakeResponse(
+ status_code=200,
+ url="https://tracker.example.com/torrents.php",
+ headers={"Content-Length": str(collector.MAX_RESPONSE_BYTES + 1)},
+ )
+
+ with pytest.raises(ValueError, match="超过 5 MiB"):
+ collector._read_limited_response(response)
+
+
+def test_collect_site_capture_writes_fixed_protocol(monkeypatch, tmp_path: Path):
+ """采集结果应使用固定四文件协议，并通过摘要和脱敏声明自校验。"""
+ cookie = "session=very-secret-cookie-value"
+ monkeypatch.setattr(
+ collector,
+ "_fetch_search_page",
+ lambda request, cookie, user_agent: SEARCH_HTML,
+ )
+
+ archive_path = collector.collect_site_capture(
+ url="https://tracker.example.com/torrents.php?search={keyword}&category=1",
+ keyword="Movie",
+ cookie=cookie,
+ output_dir=tmp_path,
+ user_agent="Browser UA",
+ site_name="示例站点",
+ )
+
+ with zipfile.ZipFile(archive_path) as archive:
+ assert tuple(archive.namelist()) == collector.ARCHIVE_FILE_NAMES
+ contents = {name: archive.read(name) for name in archive.namelist()}
+
+ manifest = json.loads(contents["manifest.json"])
+ request = json.loads(contents["request.json"])
+ report = json.loads(contents["redaction-report.json"])
+ assert manifest["format_version"] == 1
+ assert manifest["site"] == {
+ "id": "tracker-example-com",
+ "name": "示例站点",
+ "domain": "https://tracker.example.com",
+ "public": False,
+ }
+ assert manifest["capture"]["kind"] == "search"
+ assert manifest["capture"]["row_count"] == 3
+ assert manifest["privacy"] == {
+ "redacted": True,
+ "contains_credentials": False,
+ }
+ assert manifest["files"]["search.html"] == hashlib.sha256(
+ contents["search.html"]
+ ).hexdigest()
+ assert request == {
+ "method": "get",
+ "path": "/torrents.php",
+ "params": {"category": "1", "search": "{keyword}"},
+ "origin": "https://tracker.example.com",
+ }
+ assert report["redacted"] is True
+ assert report["contains_credentials"] is False
+ assert b"very-secret-cookie-value" not in b"\n".join(contents.values())
+ assert b"Browser UA" not in b"\n".join(contents.values())
+
+
+@pytest.mark.parametrize(
+ "unsafe_value",
+ [
+ "alice@example.com",
+ "192.168.1.20",
+ "AbCdEfGhIjKlMnOpQrStUvWxYz0123456789",
+ ],
+)
+def test_verify_payload_rejects_residual_private_values(unsafe_value: str):
+ """最终序列化检查应阻止残留身份信息和疑似高熵凭据写入。"""
+ payload = {
+ "manifest.json": b'{"files": {}}',
+ "request.json": b"{}",
+ "search.html": unsafe_value.encode("utf-8"),
+ "redaction-report.json": b"{}",
+ }
+
+ with pytest.raises(ValueError):
+ collector._verify_payload(payload, "session=safe-cookie-value")
+
+
+def test_verify_payload_checks_complete_short_cookie():
+ """即使 Cookie 很短，最终序列化检查也不得忽略其完整原值。"""
+ payload = {
+ "manifest.json": b'{"files": {}}',
+ "request.json": b"{}",
+ "search.html": b"id=1",
+ "redaction-report.json": b"{}",
+ }
+
+ with pytest.raises(ValueError, match="凭据值"):
+ collector._verify_payload(payload, "id=1")
+
+
+def test_collect_does_not_write_archive_when_final_scan_fails(monkeypatch, tmp_path: Path):
+ """最终序列化检查失败时不得在输出目录留下 ZIP。"""
+ monkeypatch.setattr(
+ collector,
+ "_fetch_search_page",
+ lambda request, cookie, user_agent: SEARCH_HTML,
+ )
+
+ with pytest.raises(ValueError, match="身份信息"):
+ collector.collect_site_capture(
+ url="https://tracker.example.com/torrents.php?search={keyword}",
+ keyword="Movie",
+ cookie="session=safe-cookie-value",
+ output_dir=tmp_path,
+ site_name="alice@example.com",
+ )
+
+ assert list(tmp_path.iterdir()) == []
+
+
+def test_infer_search_keyword_from_url_and_visible_input():
+ """普通浏览器模式应自动识别 URL 中与搜索框一致的关键词。"""
+ capture = collector._BrowserCapture(
+ url="https://tracker.example.com/browse.php?term=Movie%202026&category=1",
+ html=SEARCH_HTML,
+ cookie="session=safe-cookie-value",
+ user_agent="Browser UA",
+ search_inputs=[{"name": "term", "id": "search", "value": "Movie 2026"}],
+ )
+
+ assert collector._infer_search_keyword(capture) == "Movie 2026"
+
+
+def test_infer_search_keyword_rejects_post_only_page():
+ """地址栏不包含搜索参数时应友好拒绝，避免生成不可复用的 GET 配置。"""
+ capture = collector._BrowserCapture(
+ url="https://tracker.example.com/browse.php",
+ html=SEARCH_HTML,
+ cookie="session=safe-cookie-value",
+ user_agent="Browser UA",
+ search_inputs=[{"name": "search", "id": "search", "value": "Movie"}],
+ )
+
+ with pytest.raises(ValueError, match="地址栏包含关键词"):
+ collector._infer_search_keyword(capture)
+
+
+def test_read_browser_capture_filters_cookies_to_current_site(monkeypatch):
+ """浏览器采集只应保留当前站点域名的 Cookie 用于本地泄露检查。"""
+ class _FakeCdpClient:
+ """模拟只读 CDP 页面与 Cookie 返回。"""
+
+ def __init__(self, websocket_url: str):
+ """保存测试传入的 WebSocket 地址。"""
+ self.websocket_url = websocket_url
+
+ def __enter__(self):
+ """返回模拟 CDP 客户端。"""
+ return self
+
+ def __exit__(self, exc_type, exc_value, traceback):
+ """结束模拟 CDP 上下文。"""
+
+ def evaluate(self, expression: str):
+ """按表达式返回当前页面模拟数据。"""
+ if "location.href" in expression:
+ return "https://tracker.example.com/browse.php?search=Movie"
+ if "outerHTML" in expression:
+ return SEARCH_HTML
+ if "userAgent" in expression:
+ return "Browser UA"
+ return [{"name": "search", "id": "search", "value": "Movie"}]
+
+ def call(self, method: str):
+ """返回当前站点与外部站点的模拟 Cookie。"""
+ assert method == "Network.getAllCookies"
+ return {
+ "cookies": [
+ {"domain": ".example.com", "name": "session", "value": "site-secret"},
+ {"domain": ".external.test", "name": "other", "value": "external-secret"},
+ ]
+ }
+
+ monkeypatch.setattr(
+ collector,
+ "_select_search_page_target",
+ lambda port: {"webSocketDebuggerUrl": "ws://127.0.0.1/devtools/page/1"},
+ )
+ monkeypatch.setattr(collector, "_CdpClient", _FakeCdpClient)
+ session = collector._BrowserSession(
+ process=None,
+ profile_guard=None,
+ port=9222,
+ browser_websocket_url="ws://127.0.0.1/devtools/browser/1",
+ )
+
+ capture = collector._read_browser_capture(session)
+
+ assert capture.cookie == "session=site-secret"
+ assert "external-secret" not in capture.cookie
+ assert capture.user_agent == "Browser UA"
+
+
+def test_browser_capture_builds_archive_without_manual_cookie_input(monkeypatch, tmp_path: Path):
+ """普通模式应从临时浏览器数据直接生成四文件 ZIP。"""
+ @contextmanager
+ def fake_browser_session(start_url: str):
+ """提供无需启动真实浏览器的模拟会话。"""
+ assert start_url == "https://tracker.example.com"
+ yield object()
+
+ browser_capture = collector._BrowserCapture(
+ url="https://tracker.example.com/torrents.php?search=Movie&category=1",
+ html=SEARCH_HTML,
+ cookie="session=very-secret-cookie-value",
+ user_agent="Browser UA",
+ search_inputs=[{"name": "search", "id": "search", "value": "Movie"}],
+ )
+ monkeypatch.setattr(collector, "_launch_browser_session", fake_browser_session)
+ monkeypatch.setattr(collector, "_read_browser_capture", lambda session: browser_capture)
+ monkeypatch.setattr("builtins.input", lambda prompt="": "")
+
+ archive_path = collector.collect_site_capture_with_browser(
+ start_url="https://tracker.example.com",
+ output_dir=tmp_path,
+ )
+
+ with zipfile.ZipFile(archive_path) as archive:
+ manifest = json.loads(archive.read("manifest.json"))
+ request = json.loads(archive.read("request.json"))
+ combined = b"\n".join(archive.read(name) for name in archive.namelist())
+ assert manifest["collector_version"] == "1.0.1"
+ assert request["params"]["search"] == "{keyword}"
+ assert b"very-secret-cookie-value" not in combined
+