diff --git a/.github/ISSUE_TEMPLATE/feature_request.yml b/.github/ISSUE_TEMPLATE/feature_request.yml
index 9a43c905e..cd899e53a 100644
--- a/.github/ISSUE_TEMPLATE/feature_request.yml
+++ b/.github/ISSUE_TEMPLATE/feature_request.yml
@@ -8,7 +8,7 @@ body:
value: |
请说明你希望添加的功能。
- 站点适配请求请从 [采集器独立 Release](https://github.com/jxxghp/MoviePilot/releases?q=tag%3Asite-adapter-collector-v1&expanded=true) 下载最新采集器，再按 [站点适配采集说明](https://github.com/jxxghp/MoviePilot/blob/v3/docs/site-adapter-capture.md) 生成脱敏 ZIP，并在下方附加。Issue 及附件是公开内容，提交前必须解压预览四个文件。不要上传 Cookie、Authorization、通行密钥、会话字段或任何原始数据。
+ 站点适配请求请从 [采集器独立 Release](https://github.com/jxxghp/MoviePilot-Site-Collector/releases/latest) 下载最新采集器，再按 [站点适配采集说明](https://github.com/jxxghp/MoviePilot-Site-Collector/blob/main/docs/site-adapter-capture.md) 生成脱敏 ZIP，并在下方附加。Issue 及附件是公开内容，提交前必须解压预览四个文件。不要上传 Cookie、Authorization、通行密钥、会话字段或任何原始数据。
- type: input
id: version
attributes:
diff --git a/.github/workflows/site-adapter-collector.yml b/.github/workflows/site-adapter-collector.yml
deleted file mode 100644
index 97702491d..000000000
--- a/.github/workflows/site-adapter-collector.yml
+++ /dev/null
@@ -1,257 +0,0 @@
-name: Site Adapter Collector
-
-on:
- push:
- branches:
- - v3
- paths:
- - scripts/site_adapter_collector.py
- - scripts/site_adapter_collector.spec
- - scripts/site_adapter_collector_requirements.txt
- - scripts/start-site-adapter-collector.command
- - scripts/collect-site-adapter.sh
- - scripts/collector/**
- - .github/workflows/site-adapter-collector.yml
- workflow_dispatch:
- inputs:
- publish:
- description: Publish an independent release only when collector inputs changed
- required: true
- default: true
- type: boolean
-
-# 串行选择版本和发布，避免多个采集器提交争抢同一个版本。
-concurrency:
- group: site-adapter-collector-release
- cancel-in-progress: false
-
-permissions:
- contents: read
-
-jobs:
- prepare:
- name: Check collector changes and select independent version
- if: github.ref == 'refs/heads/v3'
- runs-on: ubuntu-latest
- outputs:
- changed: ${{ steps.version.outputs.changed }}
- version: ${{ steps.version.outputs.version }}
- tag: ${{ steps.version.outputs.tag }}
- previous_tag: ${{ steps.version.outputs.previous_tag }}
- source_sha: ${{ steps.version.outputs.source_sha }}
- steps:
- - name: Checkout current collector source
- uses: actions/checkout@v7
- with:
- # 排队期间可能已有新提交；统一构建当前 v3，避免旧任务回退采集器。
- ref: v3
- fetch-depth: 0
- fetch-tags: false
-
- - name: Set up Python
- uses: actions/setup-python@v7
- with:
- python-version: '3.14'
-
- - name: Select collector release version
- id: version
- env:
- GH_TOKEN: ${{ github.token }}
- run: |
- gh api "repos/$GITHUB_REPOSITORY/releases?per_page=100" --paginate \
- --jq '.[] | select(.draft == false and .prerelease == false) | .tag_name' \
- > collector-release-tags.txt
- python scripts/collector/release.py plan \
- --tags-file collector-release-tags.txt --output "$GITHUB_OUTPUT"
-
- build:
- name: Build ${{ matrix.platform_name }} collector
- needs: prepare
- if: needs.prepare.outputs.changed == 'true' || (github.event_name == 'workflow_dispatch' && !inputs.publish)
- 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@v7
- with:
- ref: ${{ needs.prepare.outputs.source_sha }}
-
- - name: Set up Python
- uses: actions/setup-python@v7
- with:
- python-version: '3.14'
-
- - name: Set up uv
- uses: astral-sh/setup-uv@v10.0.1
- with:
- enable-cache: true
- cache-dependency-glob: scripts/site_adapter_collector_requirements.txt
-
- - name: Install build dependencies
- run: uv pip install --system --requirement scripts/site_adapter_collector_requirements.txt
-
- - name: Build single-file collector
- shell: bash
- env:
- COLLECTOR_VERSION: ${{ needs.prepare.outputs.version }}
- run: |
- python scripts/collector/release.py stamp --version "$COLLECTOR_VERSION"
- pyinstaller --clean --noconfirm scripts/site_adapter_collector.spec
-
- - name: Smoke-test collector
- env:
- SOURCE_NAME: ${{ matrix.source_name }}
- COLLECTOR_VERSION: ${{ needs.prepare.outputs.version }}
- run: |
- python -c "import os, subprocess; from pathlib import Path; subprocess.run([str((Path('dist') / os.environ['SOURCE_NAME']).resolve()), '--help'], check=True)"
- python -c "import os, subprocess; from pathlib import Path; result = subprocess.run([str((Path('dist') / os.environ['SOURCE_NAME']).resolve()), '--version'], check=True, capture_output=True, text=True); assert result.stdout.strip() == os.environ['COLLECTOR_VERSION'], result.stdout"
-
- - 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', newline='\n')"
-
- - name: Upload collector artifact
- uses: actions/upload-artifact@v7
- with:
- name: ${{ matrix.artifact_name }}
- path: |
- dist/${{ matrix.asset_name }}
- dist/${{ matrix.asset_name }}.sha256
- if-no-files-found: error
- retention-days: 3
-
- publish:
- name: Publish independent collector release
- if: needs.prepare.outputs.changed == 'true' && (github.event_name == 'push' || inputs.publish)
- needs:
- - prepare
- - build
- runs-on: ubuntu-latest
- permissions:
- contents: write
-
- steps:
- - name: Checkout release source
- uses: actions/checkout@v7
- with:
- ref: ${{ needs.prepare.outputs.source_sha }}
- fetch-depth: 0
- fetch-tags: false
-
- - name: Download collector artifacts
- uses: actions/download-artifact@v8
- with:
- pattern: site-adapter-collector-*
- path: release-assets
- merge-multiple: true
-
- - name: Generate collector release notes
- env:
- COLLECTOR_VERSION: ${{ needs.prepare.outputs.version }}
- PREVIOUS_TAG: ${{ needs.prepare.outputs.previous_tag }}
- SOURCE_SHA: ${{ needs.prepare.outputs.source_sha }}
- run: |
- if [[ -n "$PREVIOUS_TAG" ]]; then
- git fetch --no-tags origin "refs/tags/$PREVIOUS_TAG:refs/tags/$PREVIOUS_TAG"
- fi
- python - <<'PY'
- import os
- import subprocess
- from pathlib import Path
- from scripts.collector.release import RELEASE_PATHS
-
- version = os.environ['COLLECTOR_VERSION']
- previous = os.environ['PREVIOUS_TAG']
- revision = f"{previous}..HEAD" if previous else "HEAD"
- records = subprocess.check_output(
- ['git', 'log', revision, '--no-merges', '--format=- %s (%h) — %an', '--', *RELEASE_PATHS],
- text=True,
- ).strip()
- notes = (
- f"MoviePilot 站点适配采集器 {version}，独立于后端版本发布。\n\n"
- "下载与系统对应的程序及同名 `.sha256` 校验文件。\n\n"
- "[使用说明](https://github.com/jxxghp/MoviePilot/blob/v3/docs/site-adapter-collector-release.md)\n\n"
- f"源码提交：`{os.environ['SOURCE_SHA']}`\n\n## 变更记录\n\n{records}\n"
- )
- Path('collector-release-notes.md').write_text(notes, encoding='utf-8')
- PY
-
- - name: Create and publish independent collector release
- env:
- GH_TOKEN: ${{ github.token }}
- RELEASE_TAG: ${{ needs.prepare.outputs.tag }}
- COLLECTOR_VERSION: ${{ needs.prepare.outputs.version }}
- SOURCE_SHA: ${{ needs.prepare.outputs.source_sha }}
- run: |
- if gh release view "$RELEASE_TAG" --repo "$GITHUB_REPOSITORY" \
- --json isDraft,targetCommitish --jq '[.isDraft, .targetCommitish] | @tsv' \
- > collector-draft-state.txt; then
- # 草稿尚未发布时标签可能不存在，应核对创建草稿时绑定的源码提交。
- IFS=$'\t' read -r is_draft draft_source < collector-draft-state.txt
- if [[ "$is_draft" != "true" || "$draft_source" != "$SOURCE_SHA" ]]; then
- echo "Refusing to overwrite a published release or a draft from another source commit" >&2
- exit 1
- fi
- gh release upload "$RELEASE_TAG" release-assets/* \
- --clobber --repo "$GITHUB_REPOSITORY"
- gh release edit "$RELEASE_TAG" --repo "$GITHUB_REPOSITORY" \
- --title "Site Adapter Collector v$COLLECTOR_VERSION" \
- --notes-file collector-release-notes.md
- else
- gh release create "$RELEASE_TAG" release-assets/* \
- --repo "$GITHUB_REPOSITORY" --target "$SOURCE_SHA" --draft \
- --title "Site Adapter Collector v$COLLECTOR_VERSION" \
- --notes-file collector-release-notes.md --latest=false
- fi
- # 全部附件上传完成才转正式发布，上传失败不会消耗正式版本。
- gh release edit "$RELEASE_TAG" --repo "$GITHUB_REPOSITORY" \
- --draft=false --latest=false
diff --git a/README.md b/README.md
index dc233c794..c27405307 100644
--- a/README.md
+++ b/README.md
@@ -61,8 +61,8 @@ curl -fsSL https://raw.githubusercontent.com/jxxghp/MoviePilot/v3/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)
-- 站点采集器下载：[独立 Release](https://github.com/jxxghp/MoviePilot/releases?q=tag%3Asite-adapter-collector-v1&expanded=true)，仅在采集器变化时递增版本并发布。
+- 新站点适配采集与 Feature Request 提交：[MoviePilot-Site-Collector](https://github.com/jxxghp/MoviePilot-Site-Collector/blob/main/docs/site-adapter-capture.md)
+- 站点采集器下载：[独立 Release](https://github.com/jxxghp/MoviePilot-Site-Collector/releases/latest)，仅在采集器变化时递增版本并发布。
- 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 a37bbe8b0..1f896b89b 100644
--- a/README_EN.md
+++ b/README_EN.md
@@ -59,7 +59,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)
+- New site adapter capture and Feature Request submission: [MoviePilot-Site-Collector](https://github.com/jxxghp/MoviePilot-Site-Collector/blob/main/docs/site-adapter-capture.md)
+- Site collector downloads: [independent repository Release](https://github.com/jxxghp/MoviePilot-Site-Collector/releases/latest), published only when collector inputs change.
- REST API documentation: https://api.movie-pilot.org
- Plugin development guide: https://wiki.movie-pilot.org/zh/plugindev
diff --git a/docs/development-setup.md b/docs/development-setup.md
index b7d3a05b0..95568abb3 100644
--- a/docs/development-setup.md
+++ b/docs/development-setup.md
@@ -234,7 +234,7 @@ Docker 镜像直接基于官方基础镜像构建，只安装应用运行所需
报告，不自动更新基线。普通主仓改动不要求额外检出插件仓或运行观察任务。宿主架构门禁和 changed-file
Pylint 由 `v3` PR/push 的 GitHub Actions 执行，`app/` 全量 Pylint 是建议性报告。
-站点采集器使用独立的 `site-adapter-collector-v1.0.x` Release，仅在采集器发布输入变化时递增版本并构建，不跟随后端 Release。下载入口为 [采集器独立 Release](https://github.com/jxxghp/MoviePilot/releases?q=tag%3Asite-adapter-collector-v1&expanded=true)，自动版本和手动测试构建规则见 [采集器发布说明](site-adapter-collector-release.md#维护者发布流程)。
+站点采集器源码与构建工作流位于 [MoviePilot-Site-Collector](https://github.com/jxxghp/MoviePilot-Site-Collector)，使用独立的 `site-adapter-collector-v1.0.x` Release，仅在采集器发布输入变化时递增版本并构建，不跟随后端 Release。下载入口为 [采集器独立 Release](https://github.com/jxxghp/MoviePilot-Site-Collector/releases/latest)，自动版本和手动测试构建规则见 [采集器发布说明](https://github.com/jxxghp/MoviePilot-Site-Collector/blob/main/docs/site-adapter-collector-release.md#维护者发布流程)。
Ruff/Mypy 基线只允许收紧，不接受新增诊断或类型错误增长；受治零错误文件由 `mypy.ini` 的
`files=` 维护。Mypy 完整 ratchet 固定按 Linux/Python 3.14 分析。Coverage job 在 `v3` PR/push
diff --git a/docs/rules/03-commands.md b/docs/rules/03-commands.md
index ae306cd9a..91a3c20fb 100644
--- a/docs/rules/03-commands.md
+++ b/docs/rules/03-commands.md
@@ -286,14 +286,15 @@ moviepilot help scheduler
## Site Adapter Capture — macOS / Linux
```bash
-# Run from a MoviePilot source checkout and reuse its virtual environment
+# Run from the independent MoviePilot-Site-Collector checkout and reuse its virtual environment
+cd /path/to/MoviePilot-Site-Collector
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.
+- Run only the collector from a trusted local [MoviePilot-Site-Collector checkout](https://github.com/jxxghp/MoviePilot-Site-Collector) or its [official Release](https://github.com/jxxghp/MoviePilot-Site-Collector/releases/latest). 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.
diff --git a/docs/site-adapter-capture.md b/docs/site-adapter-capture.md
index 4840a6032..6d90699e0 100644
--- a/docs/site-adapter-capture.md
+++ b/docs/site-adapter-capture.md
@@ -1,60 +1,11 @@
# 站点适配采集
-当开发者没有目标站点账号时，可以由已有账号的用户在本地采集一份经过裁剪和脱敏的搜索页结构，并把采集 ZIP 附加到站点适配 Feature Request。开发者和自动化流程只处理脱敏包，不需要获取用户账号。
+站点采集器已迁入独立仓库 [MoviePilot-Site-Collector](https://github.com/jxxghp/MoviePilot-Site-Collector)。普通用户无需安装 Python、Git 或 MoviePilot 后端：
-## 普通用户一键采集
+1. 从 [采集器最新 Release](https://github.com/jxxghp/MoviePilot-Site-Collector/releases/latest) 下载对应系统的程序。
+2. 按 [采集说明](https://github.com/jxxghp/MoviePilot-Site-Collector/blob/main/docs/site-adapter-capture.md) 在临时浏览器中登录、搜索并生成脱敏 ZIP。
+3. 在本仓库创建“功能改进” Issue，选择“站点适配”，附加生成的 `moviepilot-site-capture-*.zip`。
-普通用户只需准备两样东西：目标站点账号，以及已安装的 Chrome、Edge 或 Chromium。无需安装 Python、Git、MoviePilot、Docker，也不需要查看 HTML、复制 Cookie 或填写 User-Agent。
+Issue 及附件是公开内容。提交前请解压预览 `manifest.json`、`request.json`、`search.html`、`redaction-report.json`，确认没有账号或隐私信息。不要上传 Cookie、Authorization、通行密钥、原始 HTML、HAR 或浏览器网络归档。
-1. 从官方 [站点采集器独立 Release](https://github.com/jxxghp/MoviePilot/releases?q=tag%3Asite-adapter-collector-v1&expanded=true) 选择最新采集器版本，下载与 Windows、macOS 或 Linux 对应的程序；各系统文件名见下方下载说明。
-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、浏览器网络归档或截图中的账号信息。如果采集失败，请只提交终端错误文字，不要用任何原始数据代替脱敏包。
+源码运行、采集限制与脱敏协议均见 [独立仓库完整说明](https://github.com/jxxghp/MoviePilot-Site-Collector/blob/main/docs/site-adapter-capture.md)。本页面保留以兼容旧文档链接。
diff --git a/docs/site-adapter-collector-release.md b/docs/site-adapter-collector-release.md
index 40cd065ac..d169363b8 100644
--- a/docs/site-adapter-collector-release.md
+++ b/docs/site-adapter-collector-release.md
@@ -1,42 +1,12 @@
# 站点适配采集器下载说明
-普通用户优先使用 [站点采集器独立 Release](https://github.com/jxxghp/MoviePilot/releases?q=tag%3Asite-adapter-collector-v1&expanded=true) 提供的单文件采集器，选择列表中最新版本。采集器使用 `site-adapter-collector-v1.0.x` 独立标签，与 MoviePilot 后端版本分开发布。单文件已经包含 Python 和采集器依赖，不需要安装 Python、pip、Git、MoviePilot 后端，也不需要下载源码。电脑只需已安装 Chrome、Edge 或 Chromium 浏览器。
+站点采集器的源码、版本、构建工作流和历史 Release 已迁入独立仓库 [MoviePilot-Site-Collector](https://github.com/jxxghp/MoviePilot-Site-Collector)，与 MoviePilot 主程序发布完全分开。
-## 选择下载文件
+请从 [采集器最新 Release](https://github.com/jxxghp/MoviePilot-Site-Collector/releases/latest) 下载 Windows、macOS 或 Linux 程序及同名 `.sha256` 文件。
-请只从官方 [站点采集器独立 Release](https://github.com/jxxghp/MoviePilot/releases?q=tag%3Asite-adapter-collector-v1&expanded=true) 下载与系统匹配的文件；MoviePilot 主程序 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` 文件，可用于核对下载文件是否完整。程序启动时显示版本，也可执行 `--version` 查看。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 可能因此显示安全提示。仅在文件来自官方 [站点采集器独立 Release](https://github.com/jxxghp/MoviePilot/releases?q=tag%3Asite-adapter-collector-v1&expanded=true)，且校验摘要一致时运行；不要从聊天、网盘或第三方站点接收采集器。
-
-如果系统阻止运行，可改用随 MoviePilot 源码提供的本地采集脚本；该方式需要 Python 3.14+ 及完整后端依赖，不适合作为普通用户的首选路径。
+- [完整下载与运行说明](https://github.com/jxxghp/MoviePilot-Site-Collector/blob/main/docs/site-adapter-collector-release.md)
+- [采集流程和源码入口](https://github.com/jxxghp/MoviePilot-Site-Collector/blob/main/docs/site-adapter-capture.md)
## 维护者发布流程
-`.github/workflows/site-adapter-collector.yml` 在 `v3` 分支的采集器源码、依赖、启动脚本或打包工作流变化时触发，支持手动运行。后端 Release 发布及普通后端代码、文档、测试改动不会触发采集器构建。
-
-构建前会与上一个已正式发布的采集器标签比较输入文件。没有变化就跳过构建和发版；首次发布沿用源码中的 `COLLECTOR_VERSION`（当前为 `1.0.1`），后续有变化时自动递增补丁版本。工作流只在 Runner 构建副本中写入版本，使程序、`--version` 和脱敏包中的 `collector_version` 一致，不向 `v3` 分支回写版本提交。采集包的 `format_version` 协议版本保持独立。
-
-每个平台先执行 `--help` 和 `--version` 启动检查，再上传程序及 SHA-256 摘要。三个平台全部成功后，工作流创建独立版本 Release，自动生成仅包含采集器相关提交的变更记录，并设置 `latest=false`，使主程序的最新 Release 入口继续指向后端版本。
-
-手动触发时 `publish` 默认为 `true`，仍遵守“输入变化才发版”。设为 `false` 时可重复构建测试，只生成保留 3 天的 Artifact，不创建 Release、不消耗版本号。历史主程序 Release 中的采集器附件会迁移移除，下载请始终使用上面的独立 Release 入口。
+采集器仅在自身发布输入变化时自动递增版本并发布，不跟随 MoviePilot 发版。维护者请使用 [独立仓库发布流程](https://github.com/jxxghp/MoviePilot-Site-Collector/blob/main/docs/site-adapter-collector-release.md#维护者发布流程)。本页面保留以兼容旧文档链接。
diff --git a/docs/v2-to-v3-overview.md b/docs/v2-to-v3-overview.md
index 2f6984534..438471bce 100644
--- a/docs/v2-to-v3-overview.md
+++ b/docs/v2-to-v3-overview.md
@@ -399,7 +399,7 @@ MoviePilot/app/application/site
- 后端用户数据解析器可以自动探测 Gazelle、NexusPHP 等站点变种。
- 解析模板支持更完整的标题、文件名、标签和可选字段回退。
- 构建产物覆盖 Linux、Windows、macOS、x86_64、Arm64，以及 V3t 所需的 Python 3.14t ABI。
-- [站点采集器独立 Release](https://github.com/jxxghp/MoviePilot/releases?q=tag%3Asite-adapter-collector-v1&expanded=true) 提供生成脱敏样本的程序，维护者可通过自动化流程匹配模板并创建待审核的适配 PR。
+- [站点采集器独立 Release](https://github.com/jxxghp/MoviePilot-Site-Collector/releases/latest) 提供生成脱敏样本的程序，维护者可通过自动化流程匹配模板并创建待审核的适配 PR。
### 7.4 用户遇到站点问题时应如何判断
diff --git a/scripts/collect-site-adapter.sh b/scripts/collect-site-adapter.sh
deleted file mode 100644
index 798f62f40..000000000
--- a/scripts/collect-site-adapter.sh
+++ /dev/null
@@ -1,99 +0,0 @@
-#!/usr/bin/env bash
-
-set -euo pipefail
-
-ORIGINAL_DIR="$PWD"
-SCRIPT_DIR=""
-PROJECT_ROOT=""
-PYTHON_BIN=""
-COLLECTOR_PATH=""
-
-# 判断候选 Python 是否满足 3.14 最低版本。
-python_version_ok() {
- "$1" - <<'PY' >/dev/null 2>&1
-import sys
-raise SystemExit(0 if sys.version_info >= (3, 14) 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.14 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：https://github.com/jxxghp/MoviePilot/releases?q=tag%3Asite-adapter-collector-v1&expanded=true" >&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/collector/__init__.py b/scripts/collector/__init__.py
deleted file mode 100644
index 1002a46f2..000000000
--- a/scripts/collector/__init__.py
+++ /dev/null
@@ -1 +0,0 @@
-"""站点采集器的独立发布工具。"""
diff --git a/scripts/collector/release.py b/scripts/collector/release.py
deleted file mode 100644
index 80421df3e..000000000
--- a/scripts/collector/release.py
+++ /dev/null
@@ -1,116 +0,0 @@
-"""比较采集器发布输入，按独立 Release 递增版本并写入构建副本。"""
-
-import argparse
-import re
-import subprocess
-from pathlib import Path
-from typing import Optional
-
-TAG_PREFIX = "site-adapter-collector-v"
-RELEASE_PATHS = (
- "scripts/site_adapter_collector.py",
- "scripts/site_adapter_collector.spec",
- "scripts/site_adapter_collector_requirements.txt",
- "scripts/start-site-adapter-collector.command",
- "scripts/collect-site-adapter.sh",
- "scripts/collector",
- ".github/workflows/site-adapter-collector.yml",
-)
-VERSION_PATTERN = re.compile(r"(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)")
-SOURCE_VERSION_PATTERN = re.compile(r'^COLLECTOR_VERSION = "([\d.]+)"$', re.MULTILINE)
-
-
-def _version_parts(version: str) -> tuple[int, int, int]:
- """校验独立版本，避免外部标签进入 Git 参数或构建源码。"""
- match = VERSION_PATTERN.fullmatch(version)
- if not match:
- raise ValueError(f"无效采集器版本：{version}")
- return tuple(int(part) for part in match.groups())
-
-
-def latest_release_tag(tags: list[str]) -> Optional[str]:
- """从已正式发布的标签中选取采集器最高版本，忽略后端版本。"""
- collector_tags = [
- tag for tag in tags
- if tag.startswith(TAG_PREFIX) and VERSION_PATTERN.fullmatch(tag[len(TAG_PREFIX):])
- ]
- return max(collector_tags, key=lambda tag: _version_parts(tag[len(TAG_PREFIX):]), default=None)
-
-
-def release_plan(repository: Path, previous_tag: Optional[str]) -> dict[str, str]:
- """仅比较采集器输入；未变化时跳过发布，首次沿用源码版本。"""
- source = repository / "scripts/site_adapter_collector.py"
- source_match = SOURCE_VERSION_PATTERN.search(source.read_text(encoding="utf-8"))
- if not source_match:
- raise ValueError("采集器源码缺少唯一版本声明")
- source_version = source_match.group(1)
- source_parts = _version_parts(source_version)
- source_sha = subprocess.check_output(
- ["git", "rev-parse", "HEAD"], cwd=repository, text=True,
- ).strip()
- changed = True
- version = source_version
- if previous_tag:
- previous_parts = _version_parts(previous_tag.removeprefix(TAG_PREFIX))
- result = subprocess.run(
- ["git", "diff", "--quiet", previous_tag, "HEAD", "--", *RELEASE_PATHS],
- cwd=repository, check=False,
- )
- if result.returncode not in (0, 1):
- raise RuntimeError(f"无法比较采集器 Release：{previous_tag}")
- changed = result.returncode == 1
- if changed:
- next_parts = max(source_parts, (*previous_parts[:2], previous_parts[2] + 1))
- version = ".".join(str(part) for part in next_parts)
- else:
- version = previous_tag[len(TAG_PREFIX):]
- return {
- "changed": str(changed).lower(),
- "version": version,
- "tag": f"{TAG_PREFIX}{version}",
- "previous_tag": previous_tag or "",
- "source_sha": source_sha,
- }
-
-
-def stamp_version(source: Path, version: str) -> None:
- """只修改 Runner 构建副本，使程序及脱敏包版本与独立 Release 一致。"""
- _version_parts(version)
- content, count = SOURCE_VERSION_PATTERN.subn(
- f'COLLECTOR_VERSION = "{version}"', source.read_text(encoding="utf-8"),
- )
- if count != 1:
- raise ValueError("采集器源码缺少唯一版本声明")
- source.write_text(content, encoding="utf-8")
-
-
-def main() -> None:
- """供工作流选择独立发布版本或给构建副本写入版本。"""
- parser = argparse.ArgumentParser(description=__doc__)
- commands = parser.add_subparsers(dest="command", required=True)
- plan_parser = commands.add_parser("plan")
- plan_parser.add_argument("--tags-file", type=Path, required=True)
- plan_parser.add_argument("--output", type=Path, required=True)
- stamp_parser = commands.add_parser("stamp")
- stamp_parser.add_argument("--version", required=True)
- args = parser.parse_args()
- repository = Path(__file__).resolve().parents[2]
- if args.command == "stamp":
- stamp_version(repository / "scripts/site_adapter_collector.py", args.version)
- return
- tags = args.tags_file.read_text(encoding="utf-8").splitlines()
- previous_tag = latest_release_tag(tags)
- if previous_tag:
- subprocess.run(
- ["git", "fetch", "--no-tags", "origin", f"refs/tags/{previous_tag}:refs/tags/{previous_tag}"],
- cwd=repository, check=True,
- )
- plan = release_plan(repository, previous_tag)
- with args.output.open("a", encoding="utf-8") as output:
- for name, value in plan.items():
- output.write(f"{name}={value}\n")
- print(f"采集器版本：{plan['version']}；输入变化：{plan['changed']}")
-
-
-if __name__ == "__main__":
- main()
diff --git a/scripts/site_adapter_collector.py b/scripts/site_adapter_collector.py
deleted file mode 100644
index b4f7e6a4f..000000000
--- a/scripts/site_adapter_collector.py
+++ /dev/null
@@ -1,1462 +0,0 @@
-#!/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("--version", action="version", version=COLLECTOR_VERSION)
- 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 _configure_standard_streams() -> None:
- """将可重配置的标准输出流统一为 UTF-8，避免 Windows 本地代码页无法输出中文。"""
- for stream in (sys.stdout, sys.stderr):
- reconfigure = getattr(stream, "reconfigure", None)
- if not callable(reconfigure):
- continue
- try:
- reconfigure(encoding="utf-8", errors="replace")
- except (LookupError, OSError):
- continue
-
-
-def main() -> int:
- """交互式读取采集参数并生成站点适配 ZIP。"""
- _configure_standard_streams()
- args = _parse_args()
- print(f"MoviePilot 站点适配采集器 {COLLECTOR_VERSION}")
- 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
deleted file mode 100644
index 087802c7c..000000000
--- a/scripts/site_adapter_collector.spec
+++ /dev/null
@@ -1,41 +0,0 @@
-# -*- 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
deleted file mode 100644
index c9693f345..000000000
--- a/scripts/site_adapter_collector_requirements.txt
+++ /dev/null
@@ -1,4 +0,0 @@
-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
deleted file mode 100755
index 534c95623..000000000
--- a/scripts/start-site-adapter-collector.command
+++ /dev/null
@@ -1,29 +0,0 @@
-#!/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_architecture_ci.py b/tests/test_architecture_ci.py
index c41c3f6d5..5ef84a6be 100644
--- a/tests/test_architecture_ci.py
+++ b/tests/test_architecture_ci.py
@@ -241,7 +241,7 @@ def test_pylint_workflow_runs_for_v3_pull_requests_and_pushes():
def test_upload_artifact_actions_share_node24_major():
- """所有工件上传入口必须共用 Node 24 的 v7 主版本，避免 CI 告警与版本分叉。"""
+ """主仓工件上传统一为 v7；采集器工作流已迁出，不得重新混入主仓发布。"""
actions_by_workflow: dict[str, list[str]] = {}
for workflow_path in WORKFLOW_ROOT.glob("*.yml"):
workflow = _load_workflow(workflow_path.name)
@@ -258,7 +258,6 @@ def test_upload_artifact_actions_share_node24_major():
"architecture-observe.yml": ["actions/upload-artifact@v7"],
"native-dependency-update.yml": ["actions/upload-artifact@v7"],
"pylint.yml": ["actions/upload-artifact@v7"],
- "site-adapter-collector.yml": ["actions/upload-artifact@v7"],
"test.yml": [
"actions/upload-artifact@v7",
"actions/upload-artifact@v7",
diff --git a/tests/test_site_adapter_collector.py b/tests/test_site_adapter_collector.py
deleted file mode 100644
index 95807cf6a..000000000
--- a/tests/test_site_adapter_collector.py
+++ /dev/null
@@ -1,488 +0,0 @@
-import hashlib
-import json
-import os
-import subprocess
-import sys
-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
-
-
-def test_help_uses_utf8_when_parent_forces_legacy_code_page() -> None:
- """父进程强制 cp1252 时，中文 argparse 帮助仍应以 UTF-8 正常输出。"""
- environment = os.environ.copy()
- environment["PYTHONIOENCODING"] = "cp1252"
-
- result = subprocess.run(
- [sys.executable, str(Path(collector.__file__)), "--help"],
- check=False,
- capture_output=True,
- env=environment,
- timeout=30,
- )
-
- assert result.returncode == 0, result.stderr.decode("utf-8", errors="replace")
- assert "站点适配" in result.stdout.decode("utf-8")
diff --git a/tests/test_site_adapter_collector_release.py b/tests/test_site_adapter_collector_release.py
deleted file mode 100644
index 6a0fa18ed..000000000
--- a/tests/test_site_adapter_collector_release.py
+++ /dev/null
@@ -1,193 +0,0 @@
-"""验证采集器独立版本及发布边界，所有 Git 操作仅使用临时仓库。"""
-
-import os
-import subprocess
-from pathlib import Path
-
-import pytest
-from ruamel.yaml import YAML
-
-from scripts.collector import release
-
-PROJECT_ROOT = Path(__file__).parents[1]
-
-
-def _git(repository: Path, *arguments: str) -> str:
- """在临时仓库运行 Git，隔离真实仓库和用户提交身份。"""
- return subprocess.check_output(
- ["git", "-c", "user.name=Collector Test", "-c", "user.email=collector@example.invalid",
- *arguments], cwd=repository, text=True,
- ).strip()
-
-
-@pytest.fixture
-def collector_repository(tmp_path: Path) -> Path:
- """创建已发布 1.0.1 的临时采集器仓库。"""
- _git(tmp_path, "init", "-b", "v3")
- for relative in release.RELEASE_PATHS:
- path = tmp_path / relative
- if relative == "scripts/collector":
- path = path / "release.py"
- path.parent.mkdir(parents=True, exist_ok=True)
- path.write_text("initial\n", encoding="utf-8")
- (tmp_path / "scripts/site_adapter_collector.py").write_text(
- 'COLLECTOR_VERSION = "1.0.1"\n', encoding="utf-8",
- )
- _git(tmp_path, "add", ".")
- _git(tmp_path, "commit", "-m", "initial collector")
- _git(tmp_path, "tag", "site-adapter-collector-v1.0.1")
- return tmp_path
-
-
-def test_latest_release_ignores_backend_and_orders_versions_numerically():
- """后端版本和非法标签不得参与采集器版本排序。"""
- assert release.latest_release_tag([
- "v3.1.4", "site-adapter-collector-v1.0.9", "site-adapter-collector-v1.0.10",
- "site-adapter-collector-v1.0.11-rc1", "site-adapter-collector-v1.01.1",
- ]) == "site-adapter-collector-v1.0.10"
- assert release.latest_release_tag(["v3.1.4"]) is None
-
-
-def test_first_release_uses_existing_collector_version(collector_repository: Path):
- """首次独立发布沿用采集器版本，不能使用后端版本号。"""
- plan = release.release_plan(collector_repository, None)
- assert plan["tag"] == "site-adapter-collector-v1.0.1"
- assert plan["changed"] == "true"
- assert plan["source_sha"] == _git(collector_repository, "rev-parse", "HEAD")
-
-
-def test_backend_and_document_changes_do_not_release(collector_repository: Path):
- """后端和文档提交、手动重复执行都不能消耗采集器版本。"""
- for name in ("version.py", "README.md", "tests/test_collector.py"):
- path = collector_repository / name
- path.parent.mkdir(parents=True, exist_ok=True)
- path.write_text("unrelated\n", encoding="utf-8")
- _git(collector_repository, "add", ".")
- _git(collector_repository, "commit", "-m", "backend and docs")
- for _ in range(2):
- plan = release.release_plan(collector_repository, "site-adapter-collector-v1.0.1")
- assert plan["changed"] == "false"
- assert plan["version"] == "1.0.1"
-
-
-@pytest.mark.parametrize("relative", release.RELEASE_PATHS)
-def test_collector_input_change_increments_patch(collector_repository: Path, relative: str):
- """采集器任意发布输入改变都必须生成下一个补丁版本。"""
- path = collector_repository / relative
- if relative == "scripts/collector":
- path = path / "release.py"
- with path.open("a", encoding="utf-8") as output:
- output.write("# collector change\n")
- _git(collector_repository, "add", ".")
- _git(collector_repository, "commit", "-m", "change collector input")
- plan = release.release_plan(collector_repository, "site-adapter-collector-v1.0.1")
- assert plan["changed"] == "true"
- assert plan["version"] == "1.0.2"
-
-
-def test_subsequent_release_uses_published_version(collector_repository: Path):
- """源码版本无需回写，下次补丁必须以正式 Release 版本为准。"""
- _git(collector_repository, "tag", "site-adapter-collector-v1.0.9")
- (collector_repository / "scripts/site_adapter_collector.spec").unlink()
- _git(collector_repository, "add", ".")
- _git(collector_repository, "commit", "-m", "remove collector build input")
- assert release.release_plan(
- collector_repository, "site-adapter-collector-v1.0.9",
- )["version"] == "1.0.10"
-
-
-def test_missing_previous_tag_fails_instead_of_publishing(collector_repository: Path):
- """比较失败不能误判为采集器改变并发版。"""
- with pytest.raises(RuntimeError, match="无法比较"):
- release.release_plan(collector_repository, "site-adapter-collector-v1.0.8")
-
-
-def test_stamp_sets_runtime_and_manifest_version(tmp_path: Path):
- """构建版本写入现有声明，其他源码和采集包协议版本保持一致。"""
- source = tmp_path / "collector.py"
- source.write_text('COLLECTOR_VERSION = "1.0.1"\nFORMAT_VERSION = 1\n', encoding="utf-8")
- release.stamp_version(source, "1.0.10")
- assert source.read_text(encoding="utf-8") == 'COLLECTOR_VERSION = "1.0.10"\nFORMAT_VERSION = 1\n'
- with pytest.raises(ValueError, match="无效采集器版本"):
- release.stamp_version(source, "1.0.10\ninvalid")
-
-
-def test_workflow_separates_releases_and_pins_all_build_sources():
- """工作流只跟随采集器输入，三平台成功才发布且保留主程序 latest。"""
- workflow = YAML(typ="safe").load(
- (PROJECT_ROOT / ".github/workflows/site-adapter-collector.yml").read_text(encoding="utf-8"),
- )
- assert "release" not in workflow["on"]
- assert workflow["on"]["push"]["branches"] == ["v3"]
- paths = workflow["on"]["push"]["paths"]
- assert set(paths) == set(release.RELEASE_PATHS) - {"scripts/collector"} | {"scripts/collector/**"}
- assert workflow["concurrency"]["cancel-in-progress"] is False
- jobs = workflow["jobs"]
- assert "needs.prepare.outputs.changed == 'true'" in jobs["build"]["if"]
- assert "!inputs.publish" in jobs["build"]["if"]
- assert jobs["publish"]["needs"] == ["prepare", "build"]
- for job in ("build", "publish"):
- checkout = jobs[job]["steps"][0]
- assert checkout["with"]["ref"] == "${{ needs.prepare.outputs.source_sha }}"
- build_step = next(step for step in jobs["build"]["steps"] if step["name"] == "Build single-file collector")
- assert build_step["shell"] == "bash"
- assert "release.py stamp" in build_step["run"]
- publish = jobs["publish"]["steps"][-1]["run"]
- assert "--target \"$SOURCE_SHA\" --draft" in publish
- assert "--draft=false --latest=false" in publish
- assert "--json isDraft,targetCommitish" in publish
- assert '"$draft_source" != "$SOURCE_SHA"' in publish
- assert '"$RELEASE_TAG^{commit}"' not in publish
-
-
-@pytest.mark.parametrize(
- ("draft_flag", "draft_source", "accepted"),
- [("true", "source-sha", True), ("true", "other-sha", False), ("false", "source-sha", False)],
-)
-def test_draft_retry_checks_source_without_requiring_git_tag(
- tmp_path: Path, draft_flag: str, draft_source: str, accepted: bool,
-):
- """实际运行发布 Shell，确认无标签草稿可重试且拒绝错误源码或正式版本。"""
- workflow = YAML(typ="safe").load(
- (PROJECT_ROOT / ".github/workflows/site-adapter-collector.yml").read_text(encoding="utf-8"),
- )
- commands = workflow["jobs"]["publish"]["steps"][-1]["run"]
- binaries = tmp_path / "bin"
- binaries.mkdir()
- gh_stub = binaries / "gh"
- gh_stub.write_text(
- '#!/bin/bash\n'
- 'if [[ "$1 $2" == "release view" ]]; then\n'
- ' printf "%s\\t%s\\n" "$DRAFT_FLAG" "$DRAFT_SOURCE"\n'
- 'else\n'
- ' printf "%s\\n" "$*" >> "$GH_LOG"\n'
- 'fi\n', encoding="utf-8",
- )
- git_stub = binaries / "git"
- git_stub.write_text("#!/bin/bash\nexit 79\n", encoding="utf-8")
- for path in (gh_stub, git_stub):
- path.chmod(0o755)
- log = tmp_path / "gh.log"
- environment = {
- **os.environ,
- "PATH": f"{binaries}{os.pathsep}{os.environ['PATH']}",
- "DRAFT_FLAG": draft_flag,
- "DRAFT_SOURCE": draft_source,
- "GH_LOG": str(log),
- "SOURCE_SHA": "source-sha",
- "RELEASE_TAG": "site-adapter-collector-v1.0.2",
- "COLLECTOR_VERSION": "1.0.2",
- "GITHUB_REPOSITORY": "test/collector",
- }
- result = subprocess.run(
- ["bash", "-e", "-o", "pipefail", "-c", commands],
- cwd=tmp_path, env=environment, capture_output=True, text=True, check=False,
- )
- if accepted:
- assert result.returncode == 0, result.stderr
- calls = log.read_text(encoding="utf-8")
- assert "release upload site-adapter-collector-v1.0.2" in calls
- assert "--draft=false --latest=false" in calls
- else:
- assert result.returncode != 0
- assert not log.exists()
-