feat: add standalone site adapter collector

This commit is contained in:
jxxghp
2026-07-12 13:51:07 +08:00
parent 30b932e07e
commit 18c1ec4b82
14 changed files with 2352 additions and 2 deletions
+15 -1
View File
@@ -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:
@@ -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"
+1
View File
@@ -37,6 +37,7 @@ coverage.json
htmlcov/
.vscode
venv
moviepilot-site-capture-*.zip
# Pylint
pylint-report.json
+2
View File
@@ -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
+2
View File
@@ -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
+17 -1
View File
@@ -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*
+60
View File
@@ -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、浏览器网络归档或截图中的账号信息。如果采集失败,请只提交终端错误文字,不要用任何原始数据代替脱敏包。
+36
View File
@@ -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 ArtifactRelease 事件会在三个平台全部成功后,用 GitHub CLI 把同一组文件附加到触发本次任务的 Release。
+99
View File
@@ -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 </dev/tty
else
echo "需要可交互终端读取站点地址并等待浏览器采集确认。" >&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 "$@"
File diff suppressed because it is too large Load Diff
+41
View File
@@ -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,
)
@@ -0,0 +1,4 @@
beautifulsoup4~=4.15.0
PyInstaller>=6.14,<7.0
requests>=2.32,<3.0
websocket-client~=1.9.0
+29
View File
@@ -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 "$@"
+469
View File
@@ -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 = """
<!doctype html>
<html>
<head><script>window.token = "embedded-secret-value";</script></head>
<body>
<nav id="account">alice@example.com <span>192.168.1.20</span></nav>
<form><input type="hidden" name="csrf_token" value="embedded-secret-value"></form>
<table id="torrent-table" class="torrents">
<thead>
<tr><th>标题</th><th>大小</th><th>做种</th><th>发布者</th></tr>
</thead>
<tbody>
<tr id="uploader-alice"
class="torrent-row torrent-session-token-cell profile-alice contact-alice@example.com peer-192.168.1.20"
data-id="987" data-private-user="alice" unknown="alice@example.com">
<td>
<a class="torrent-title" title="私密电影标题.2026.1080p" aria-label="私密电影标题"
href="/details.php?id=123&passkey=embedded-secret-value">私密电影标题</a>
<img data-orig="/covers/私密电影标题.jpg" data-original="/covers/private-title.jpg"
data-lazy-src="https://images.example.net/private-title.jpg">
</td>
<td>18.45 GiB</td>
<td>42</td>
<td><a class="username" href="/users/alice">alice</a></td>
<td><span class="date-added" title="2026-07-12 12:34:56">刚刚</span></td>
</tr>
<tr class="torrent-listings-global-freeleech">
<td><a class="torrent-title" href="/details.php?id=124">另一部私密电影</a></td>
<td>8.00 GiB</td><td>12</td><td>bob</td>
</tr>
<tr class="torrent-row">
<td><a class="torrent-title" href="/details.php?id=125">第三部私密电影</a></td>
<td>2.00 GiB</td><td>8</td><td>carol</td>
</tr>
</tbody>
</table>
<footer>unrelated footer</footer>
</body>
</html>
"""
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(
'<a href="/userdetails.php?id=9">user</a>',
"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"""
<tr class="outer-result-row">
<td>
<table class="torrentname">
<tbody><tr><td><a href="/details.php?id={index}">资源 {index}</a></td></tr></tbody>
</table>
</td>
<td>{index}</td>
</tr>
"""
for index in range(1, 31)
)
html = f"""
<html><body>
<table class="torrents">
<thead><tr><th>名称</th><th>做种</th></tr></thead>
<tbody>{result_rows}</tbody>
</table>
</body></html>
"""
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