Improve XHS long task downloads

.gitignore

@@ -1,5 +1,14 @@
 .venv/
 __pycache__/
 *.pyc
+.DS_Store
 .xhs-chrome-profile/
+data/
+data_queue_smoke/
+data_search_smoke/
 video/
+video_queue_smoke/
+video_search_smoke/
+video_bad_*/
+video_good_*/
+video_human_test/

HANDOVER.md

@@ -0,0 +1,227 @@
+# xhs_video_crawler 工作移交文档
+
+移交日期：2026-06-08
+
+## 项目概述
+
+本项目用于探索和验证小红书公开视频内容的采集和下载链路。当前实现采用人工登录浏览器 + 本地脚本附着调试端口的方式运行，不包含自动登录、验证码绕过或私有接口调用。
+
+核心入口：
+
+- `login_xhs.py`：启动带远程调试端口的 Chrome，并打开小红书页面供人工登录。
+- `XHS.py`：附着到已登录 Chrome，发现笔记链接，解析视频地址，下载视频并写入元数据。
+- `test_xhs.py`、`test_login_xhs.py`：单元测试。
+- `docs/superpowers/`：需求和实现计划文档，记录了浏览器 feed 下载、长任务队列、搜索来源等设计背景。
+
+## 当前能力
+
+### 单次发现页下载
+
+登录后可直接下载当前发现页或指定起始页的视频：
+
+```bash
+./.venv/bin/python XHS.py --max-videos 10
+```
+
+### 长任务队列下载
+
+长任务使用 JSONL 队列文件保存状态，支持中断后继续：
+
+```bash
+./.venv/bin/python XHS.py \
+  --source video-channel \
+  --target-videos 500 \
+  --queue-file data/xhs_500_queue.jsonl \
+  --metadata-file data/xhs_500_metadata.jsonl \
+  --report-file data/xhs_500_report.json \
+  --output-dir video/xhs_500 \
+  --max-runtime 28800 \
+  --timeout 25 \
+  --retry-limit 2 \
+  --min-wait 2 \
+  --max-wait 6 \
+  --long-break-every 20
+```
+
+队列状态：
+
+- `pending`：待处理。
+- `downloaded`：已下载。
+- `skipped_image`：无视频候选，通常是图文笔记或详情页没有暴露视频地址。
+- `failed`：超过重试次数仍失败，例如响应过小、网络错误等。
+
+### 搜索来源下载
+
+支持关键词搜索来源：
+
+```bash
+./.venv/bin/python XHS.py \
+  --source search \
+  --keyword 猫咪 \
+  --target-videos 100 \
+  --queue-file data/search_cat_queue.jsonl
+```
+
+## 近期重要改动
+
+本次移交提交中包含长任务稳定性补强：
+
+- 下载改为 `requests.get(..., stream=True)` 分块写入，避免大文件一次性读入内存。
+- 下载时先写入 `.part` 临时文件，校验通过后再替换为正式 `.mp4`；失败时清理 `.part`。
+- 增加 `DEFAULT_MIN_VIDEO_BYTES = 200 * 1024`，默认拒绝过小视频响应，避免把异常响应保存为视频。
+- 队列模式支持 `--min-video-bytes` 和 `--report-file`。
+- 队列任务结束后生成运行报告 JSON，包含下载数量、队列状态、元数据行数、评论统计、目录大小、耗时等。
+- 元数据中增加 `file_size_bytes`。
+- 新增对应单元测试，覆盖流式下载、过小响应拒绝、CLI 参数透传、运行报告统计。
+
+## 500 条长任务验证结果
+
+2026-06-01 至 2026-06-02 跑过一次真实 500 条长任务验证。
+
+命令使用 `video-channel` 来源，目标 `500` 条，输出到：
+
+- 视频目录：`video/xhs_500`
+- 队列文件：`data/xhs_500_queue.jsonl`
+- 元数据文件：`data/xhs_500_metadata.jsonl`
+- 运行报告：`data/xhs_500_report.json`
+
+验证结果：
+
+- 下载成功：500
+- 本地 mp4 文件：500
+- 元数据行数：500
+- `.part` 临时文件残留：0
+- 输出目录大小：约 4.5 GB
+- 总耗时：19052.52 秒，约 5 小时 17 分钟
+- 队列最终状态：
+  - `downloaded`: 500
+  - `skipped_image`: 147
+  - `failed`: 2
+  - `pending`: 50
+
+两个 `failed` 都是 `视频响应过小`，被最小字节数保护逻辑拦截。`skipped_image` 主要来自视频频道队列里混入无视频候选的笔记。
+
+结论：当前链路可以稳定跑 500 条级别长任务。主要瓶颈不是进程稳定性，而是视频频道来源筛选精度和页面内容密度。
+
+## 运行环境
+
+推荐环境：
+
+```bash
+python3 -m venv .venv
+source .venv/bin/activate
+pip install requests DrissionPage
+```
+
+启动浏览器：
+
+```bash
+./.venv/bin/python login_xhs.py --browser-port 9223
+```
+
+注意事项：
+
+- Chrome 必须保持运行，脚本通过 `127.0.0.1:9223` 附着。
+- 登录和验证码需要人工处理。
+- 长任务期间不要让电脑睡眠、断网或关闭 Chrome。
+- 如果任务中断，保留 `queue-file` 后重新执行相同命令即可继续。
+
+## 数据和产物
+
+以下目录在 `.gitignore` 中，正常不应提交：
+
+- `.venv/`
+- `.xhs-chrome-profile/`
+- `data/`
+- `video/`
+- `data_queue_smoke/`
+- `data_search_smoke/`
+- `video_queue_smoke/`
+- `video_search_smoke/`
+- `video_human_test/`
+
+原因：
+
+- `.xhs-chrome-profile/` 可能包含本地浏览器登录态。
+- `data/` 和 `video/` 是运行产物，体积大，且可能包含采集数据。
+
+## 测试
+
+提交前建议运行：
+
+```bash
+python3 -m unittest test_xhs.py test_login_xhs.py -v
+```
+
+当前测试覆盖重点：
+
+- 可选运行依赖缺失时模块可导入。
+- Chrome 调试端口检测。
+- feed 和 HTML 中视频候选提取。
+- 元数据、评论、输出文件名构造。
+- 队列加载、保存、去重和状态流转。
+- 搜索和视频频道来源构造。
+- 流式下载和异常响应校验。
+- 队列运行报告统计。
+
+## 已知问题和风险
+
+1. 视频频道来源筛选不够精确
+
+   队列里会混入图文或详情页无视频候选笔记，500 条验证中有 147 条 `skipped_image`。这不影响稳定性，但会降低有效下载效率。
+
+2. 页面依赖小红书前端结构
+
+   当前依赖页面 DOM、内嵌状态和浏览器 feed 响应。如果小红书前端结构变化，解析器可能需要更新。
+
+3. 不适合无人值守处理验证码
+
+   项目设计前提是人工登录和人工处理验证码，不做自动绕过。
+
+4. 单进程串行下载速度有限
+
+   当前设计偏稳健，默认有人类化等待、详情页打开、评论加载和长停留。500 条真实验证耗时约 5 小时 17 分钟。
+
+5. 运行产物没有进入 Git
+
+   真实验证产物保存在本地 `data/`、`video/`，默认不提交。如需归档，应走对象存储、网盘或其他专门的数据归档流程。
+
+## 建议后续优化
+
+1. 提高视频卡片筛选精度
+
+   优先优化 `collect_note_urls_from_page(..., video_only=True)` 的 DOM 判断，降低 `skipped_image` 比例。
+
+2. 增加任务心跳
+
+   可周期性写 `heartbeat` 或增量报告，避免长任务中途只能从队列文件推断状态。
+
+3. 增加中断报告
+
+   当前报告在正常结束时写入。后续可以在 `KeyboardInterrupt` 或异常退出时也写入当前队列摘要。
+
+4. 将长任务参数固化成脚本
+
+   可以新增 `scripts/run_xhs_long_task.sh` 或类似入口，减少手动拼命令出错。
+
+5. 降低评论采集开销
+
+   如果只关注视频下载稳定性，可设置 `--max-comments 0`，能明显提升吞吐。
+
+6. 增加数据入库或对象存储
+
+   目前产物是本地 mp4 + JSONL。若后续要用于生产流程，应考虑上传对象存储并将元数据同步到数据库。
+
+## 接手建议
+
+接手时建议按以下顺序熟悉：
+
+1. 阅读 `README.md`，跑通 1 到 3 条小规模下载。
+2. 阅读 `XHS.py` 中 `run_queue_download`、`download_video`、`extract_video_candidates*`、`collect_note_urls*`。
+3. 运行完整单测。
+4. 用 `--target-videos 20` 跑一次队列模式，查看 queue、metadata、report 三类产物。
+5. 再考虑改动来源筛选或任务报告能力。
+
+## 合规边界
+
+本项目仅用于技术学习、链路验证和授权范围内的数据处理。使用时应遵守平台服务条款、robots 协议和相关法律法规，不应绕过访问控制、验证码或平台风控，不应采集或传播未授权内容。

README.md

@@ -36,11 +36,15 @@ pip install requests DrissionPage
 
 ### 步骤 1：启动 Chrome 并手动登录
 
+如果你已经通过抖音项目启动了调试端口为 `9223` 的 Chrome，可以直接在那个 Chrome 里打开并登录小红书，不需要再运行 `login_xhs.py`。
+
+如果还没有可复用的 Chrome，再运行：
+
 ```bash
 ./.venv/bin/python login_xhs.py
 ```
 
-脚本会打开 `https://www.xiaohongshu.com/explore`。请在打开的浏览器里完成登录；如果出现验证码，也需要手动处理。
+脚本会用默认端口 `9223` 打开 `https://www.xiaohongshu.com/explore`。请在打开的浏览器里完成登录；如果出现验证码，也需要手动处理。
 
 ### 步骤 2：下载发现页视频
 
@@ -59,6 +63,35 @@ pip install requests DrissionPage
 # 指定保存目录
 ./.venv/bin/python XHS.py --max-videos 10 --output-dir video
 
+# 默认启用温和随机浏览节奏；可调整停留时间和阶段长休息
+./.venv/bin/python XHS.py --max-videos 20 --min-wait 2 --max-wait 6 --long-break-every 4
+
+# 测试时可以缩短等待；需要最快速度时可关闭 human mode
+./.venv/bin/python XHS.py --max-videos 3 --min-wait 0.5 --max-wait 1
+./.venv/bin/python XHS.py --max-videos 3 --no-human-mode
+
+# 限制最长运行时间，单位秒
+./.venv/bin/python XHS.py --max-videos 20 --max-runtime 600
+
+# 长任务队列模式：适合下载大量视频，可中断后继续
+# video-channel 对应网页顶部“视频”频道，通常比关键词搜索更适合大量下载
+./.venv/bin/python XHS.py \
+  --source video-channel \
+  --target-videos 1000 \
+  --queue-file data/xhs_queue.jsonl \
+  --metadata-file data/xhs_metadata.jsonl \
+  --max-runtime 7200
+
+# 搜索关键词结果下载：例如猫咪相关视频
+./.venv/bin/python XHS.py \
+  --source search \
+  --keyword 猫咪 \
+  --target-videos 100 \
+  --queue-file data/search_cat_queue.jsonl
+
+# 继续上次未完成的队列任务
+./.venv/bin/python XHS.py --queue-file data/xhs_queue.jsonl --target-videos 1000
+
 # 如果启动 Chrome 时换了端口，下载脚本也要使用同一个端口
 ./.venv/bin/python login_xhs.py --browser-port 9334
 ./.venv/bin/python XHS.py --browser-port 9334 --max-videos 10
@@ -75,7 +108,10 @@ pip install requests DrissionPage
 - 浏览器负责加载小红书页面和保留登录态。
 - 脚本只监听浏览器里已经产生的网络响应。
 - 解析器会递归查找响应 JSON 中的 `master_url`、`backup_urls` 等视频地址字段。
+- 默认会在发现页和详情页之间随机停留、上下滚动，并在阶段下载后长停留。
 - 下载过程会去重，并在单个视频失败时继续处理后续视频。
+- 队列模式会把笔记链接和处理状态保存到 JSONL 文件，支持长任务恢复。
+- 队列模式下载成功后会追加写入元数据 JSONL，包含 note id、标题、描述、封面、作者、点赞/收藏/评论/分享数、视频地址、保存路径，以及页面可见评论（默认最多 20 条，评论不可见时为空数组）。
 
 ## 测试
 

docs/superpowers/plans/2026-05-27-xhs-long-queue-downloader.md

@@ -0,0 +1,45 @@
+# XHS Long Queue Downloader Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Add a resumable JSONL queue mode so long Xiaohongshu video download jobs can target large counts like 1000 videos.
+
+**Architecture:** Keep `XHS.py` as the CLI entry point. Add queue record helpers, source URL helpers, discovery/processing orchestration, and CLI flags while reusing existing parsing, download validation, shared Chrome, and human browsing cadence.
+
+**Tech Stack:** Python 3, unittest, JSONL files, DrissionPage, requests.
+
+---
+
+## File Structure
+
+- Modify `XHS.py`: queue dataclass/helpers, source selection, queue orchestration, CLI flags.
+- Modify `test_xhs.py`: queue unit tests and CLI plumbing tests.
+- Modify `README.md`: long task command examples.
+
+## Task 1: Queue Persistence
+
+- [ ] Write tests for queue load/save, deduping by note_id, counting downloaded records, and status updates.
+- [ ] Run `python3 -m unittest test_xhs.py -v` and verify failures.
+- [ ] Implement `QueueRecord`, `load_queue`, `save_queue`, `merge_note_urls_into_queue`, `count_queue_status`.
+- [ ] Run tests and verify pass.
+
+## Task 2: Source Selection and CLI
+
+- [ ] Write tests for `build_source_url` and parser defaults for `--source`, `--target-videos`, `--queue-file`, `--retry-limit`.
+- [ ] Run tests and verify failures.
+- [ ] Implement source URL selection and CLI argument plumbing.
+- [ ] Run tests and verify pass.
+
+## Task 3: Queue Processing Orchestration
+
+- [ ] Write tests for pure queue status transitions for success, skipped image, failed retry.
+- [ ] Run tests and verify failures.
+- [ ] Implement queue processing helpers and wire queue mode into `main` when `--queue-file` or `--target-videos` is provided.
+- [ ] Run tests and verify pass.
+
+## Task 4: Docs and Verification
+
+- [ ] Update README with 1000-video queue command and resume behavior.
+- [ ] Run `python3 -m unittest test_xhs.py test_login_xhs.py -v`.
+- [ ] Run a small smoke command with low target and short waits if browser is available.
+- [ ] Commit and push.

docs/superpowers/plans/2026-05-27-xhs-search-source.md

@@ -0,0 +1,24 @@
+# XHS Search Source Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Add `--source search --keyword <term>` to the resumable queue downloader.
+
+**Architecture:** Extend `build_source_url`, CLI parser choices, `run_queue_download` arguments, and README examples. Reuse all queue and download code.
+
+**Tech Stack:** Python 3, unittest, DrissionPage, requests.
+
+---
+
+## Task 1: Search URL and CLI
+
+- [x] Write failing tests for encoded search source URL and CLI keyword plumbing.
+- [x] Implement `build_source_url("search", keyword=...)`, parser support, and queue runner forwarding.
+- [x] Run tests.
+
+## Task 2: Docs and Smoke
+
+- [x] Update README with search examples.
+- [x] Run full unit tests.
+- [x] Run a small search smoke test with `--keyword 猫咪 --target-videos 2`.
+- [ ] Commit and push.

docs/superpowers/specs/2026-05-27-xhs-browser-feed-download-design.md

@@ -31,7 +31,7 @@ The tool mirrors the existing Douyin project pattern:
 ```bash
 python3 login_xhs.py
 python3 XHS.py --max-videos 10
-python3 XHS.py --browser-port 9224 --max-videos 20 --output-dir video
+python3 XHS.py --browser-port 9334 --max-videos 20 --output-dir video
 ```
 
 ## Error Handling

docs/superpowers/specs/2026-05-27-xhs-long-queue-downloader-design.md

@@ -0,0 +1,62 @@
+# XHS Long Queue Downloader Design
+
+## Goal
+
+Add a resumable long-task downloader for collecting large numbers of Xiaohongshu videos, such as 1000 videos, without relying on a single recommendation page pass.
+
+## Scope
+
+The feature stays within the existing manually logged-in browser model. It does not automate login, bypass verification, spoof device fingerprints, or call private APIs directly outside what the loaded web pages expose. It improves task durability, source density, and progress tracking.
+
+## Architecture
+
+The downloader becomes two-phase while preserving the current one-command UX:
+
+1. Queue discovery collects note detail URLs from configured sources and writes them to a JSONL queue.
+2. Queue processing opens pending note URLs, extracts video URLs from page state or feed responses, downloads valid videos, and updates each queue item status.
+
+The queue file stores one JSON object per note:
+
+```json
+{"note_id":"...","url":"...","source":"video-channel","status":"pending","attempts":0,"downloaded_path":"","last_error":"","updated_at":"..."}
+```
+
+Statuses are `pending`, `downloaded`, `skipped_image`, and `failed`.
+
+## Sources
+
+The first implementation supports:
+
+- `explore`: current recommendation page.
+- `video-channel`: `https://www.xiaohongshu.com/explore?channel_id=video` as a best-effort source. If Xiaohongshu redirects or changes channel routing, the collector still reads visible `/explore/` cards.
+- `current-page`: process the current browser page.
+
+Future search keyword sources can be added after the queue engine is stable.
+
+## Runtime Behavior
+
+A command such as:
+
+```bash
+./.venv/bin/python XHS.py --source video-channel --target-videos 1000 --queue-file data/xhs_queue.jsonl --max-runtime 7200
+```
+
+will:
+
+1. Load existing queue records.
+2. Count already downloaded items.
+3. Open the selected source page and collect visible note URLs.
+4. Append new pending records, preserving existing statuses.
+5. Process pending records until `target_videos`, `max_runtime`, or queue exhaustion.
+6. If queue is exhausted before target, return to source, scroll, collect more URLs, and continue.
+
+## Error Handling
+
+- Non-video notes become `skipped_image`.
+- Download failures increment attempts and become `failed` after retry limit.
+- The queue is rewritten atomically after status changes.
+- Progress logs include downloaded count, skipped count, failed count, and pending count.
+
+## Testing
+
+Unit tests cover JSONL queue load/save, deduplication, status updates, source URL selection, target counting, and CLI argument plumbing. Existing download and parsing tests remain in place.

docs/superpowers/specs/2026-05-27-xhs-search-source-design.md

@@ -0,0 +1,27 @@
+# XHS Search Source Design
+
+## Goal
+
+Allow the resumable queue downloader to use Xiaohongshu search results as a source, so queries such as `猫咪` or `猫咪 搞笑` can collect and download related video notes.
+
+## Scope
+
+This feature reuses the existing manually logged-in Chrome, queue persistence, page card collection, detail-page video extraction, validation, and human browsing cadence. It does not automate login, bypass verification, or call hidden APIs directly.
+
+## CLI
+
+```bash
+./.venv/bin/python XHS.py --source search --keyword 猫咪 --target-videos 100 --queue-file data/search_cat_queue.jsonl
+```
+
+## Behavior
+
+- `--source search` requires `--keyword`.
+- The source URL is `https://www.xiaohongshu.com/search_result?keyword=<encoded keyword>&source=web_search_result_notes&type=51`, which opens the video-filtered search results page.
+- Search result cards are collected from both `/explore/<note_id>` and tokenized `/search_result/<note_id>` links.
+- Detail links are polled briefly after navigation because Xiaohongshu search result cards are rendered asynchronously.
+- Queue mode handles videos, images, failures, retries, and resume semantics exactly like other sources.
+
+## Testing
+
+Unit tests cover search URL encoding, parser defaults, queue-mode CLI plumbing for keyword, `/search_result/` note ID extraction, tokenized search link normalization, and async result-link polling.

login_xhs.py

@@ -9,7 +9,7 @@ from pathlib import Path
 
 DEFAULT_START_URL = "https://www.xiaohongshu.com/explore"
 DEFAULT_CHROME_PATH = "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome"
-DEFAULT_BROWSER_PORT = 9224
+DEFAULT_BROWSER_PORT = 9223
 DEFAULT_PROFILE_DIR = Path(".xhs-chrome-profile")
 
 
@@ -50,7 +50,7 @@ def build_parser() -> argparse.ArgumentParser:
         "--browser-port",
         type=int,
         default=DEFAULT_BROWSER_PORT,
-        help="Chrome 调试端口，默认 9224",
+        help="Chrome 调试端口，默认 9223",
     )
     parser.add_argument("--start-url", default=DEFAULT_START_URL, help="启动后打开的小红书页面 URL")
     return parser

test_login_xhs.py

@@ -13,7 +13,7 @@ class LoginXhsModuleTests(unittest.TestCase):
         command = module.build_login_command(
             chrome_path="/Applications/Google Chrome.app/Contents/MacOS/Google Chrome",
             profile_dir=Path("/tmp/xhs-profile"),
-            browser_port=9224,
+            browser_port=9223,
             start_url="https://www.xiaohongshu.com/explore",
         )
         self.assertEqual(
@@ -24,7 +24,7 @@ class LoginXhsModuleTests(unittest.TestCase):
                 "/Applications/Google Chrome.app",
                 "--args",
                 "--user-data-dir=/tmp/xhs-profile",
-                "--remote-debugging-port=9224",
+                "--remote-debugging-port=9223",
                 "https://www.xiaohongshu.com/explore",
             ],
         )
@@ -32,7 +32,7 @@ class LoginXhsModuleTests(unittest.TestCase):
     def test_build_parser_uses_expected_defaults(self) -> None:
         module = importlib.import_module("login_xhs")
         args = module.build_parser().parse_args([])
-        self.assertEqual(args.browser_port, 9224)
+        self.assertEqual(args.browser_port, 9223)
         self.assertEqual(args.chrome_path, module.DEFAULT_CHROME_PATH)
         self.assertEqual(args.start_url, module.DEFAULT_START_URL)
 
@@ -79,7 +79,7 @@ class LoginXhsModuleTests(unittest.TestCase):
                         )
         self.assertEqual(exit_code, 0)
         self.assertIn("./.venv/bin/python XHS.py", stdout.getvalue())
-        self.assertNotIn("--browser-port 9224", stdout.getvalue())
+        self.assertNotIn("--browser-port 9223", stdout.getvalue())
 
     def test_main_returns_error_when_chrome_path_missing(self) -> None:
         module = importlib.import_module("login_xhs")

test_xhs.py

@@ -1,5 +1,7 @@
 import importlib
+import tempfile
 import unittest
+from pathlib import Path
 from unittest import mock
 
 
@@ -9,6 +11,159 @@ class FakeResponse:
         self.raw_body = raw_body
 
 
+class FakeDownloadResponse:
+    def __init__(self, content: bytes, content_type: str = "video/mp4", status_code: int = 200):
+        self.content = content
+        self.headers = {"content-type": content_type}
+        self.status_code = status_code
+
+    def raise_for_status(self) -> None:
+        if self.status_code >= 400:
+            raise RuntimeError(f"HTTP {self.status_code}")
+
+
+class FakeRequests:
+    def __init__(self, response: FakeDownloadResponse):
+        self.response = response
+        self.calls = []
+
+    def get(self, video_url, headers, timeout, **kwargs):
+        self.calls.append(
+            {
+                "video_url": video_url,
+                "headers": headers,
+                "timeout": timeout,
+                **kwargs,
+            }
+        )
+        return self.response
+
+
+class FakeStreamingResponse:
+    def __init__(self, chunks: list[bytes], content_type: str = "video/mp4", status_code: int = 200):
+        self.chunks = chunks
+        self.headers = {"content-type": content_type}
+        self.status_code = status_code
+
+    @property
+    def content(self) -> bytes:
+        raise AssertionError("streaming download should not read full response.content")
+
+    def raise_for_status(self) -> None:
+        if self.status_code >= 400:
+            raise RuntimeError(f"HTTP {self.status_code}")
+
+    def iter_content(self, chunk_size: int):
+        for chunk in self.chunks:
+            yield chunk
+
+
+class FakeScrollPage:
+    def __init__(self):
+        self.scripts = []
+
+    def run_js(self, script):
+        self.scripts.append(script)
+
+
+class FakeDelayedCommentPage:
+    def __init__(self):
+        self.comment_checks = 0
+        self.scroll_scripts = []
+
+    def run_js(self, script):
+        if "xhsVisibleCommentCount" in script:
+            self.comment_checks += 1
+            return 1 if self.comment_checks >= 2 else 0
+        if "xhsScrollCommentContainer" in script:
+            self.scroll_scripts.append(script)
+            return True
+        return None
+
+
+class FakeLinkPage:
+    def __init__(self, links):
+        self.links = links
+        self.scripts = []
+
+    def run_js(self, script):
+        self.scripts.append(script)
+        return self.links
+
+
+class FakeDelayedLinkPage:
+    def __init__(self):
+        self.calls = 0
+
+    def run_js(self, script):
+        self.calls += 1
+        if self.calls == 1:
+            return []
+        return ["https://www.xiaohongshu.com/search_result/abc?xsec_token=token"]
+
+
+class FakeGrowingLinkPage:
+    def __init__(self):
+        self.collect_calls = 0
+
+    def run_js(self, script):
+        if "querySelectorAll" not in script:
+            return None
+        self.collect_calls += 1
+        if self.collect_calls == 1:
+            return ["https://www.xiaohongshu.com/search_result/one?xsec_token=token1"]
+        return [
+            "https://www.xiaohongshu.com/search_result/one?xsec_token=token1",
+            "https://www.xiaohongshu.com/search_result/two?xsec_token=token2",
+        ]
+
+
+class FakeVideoOnlyLinkPage:
+    def __init__(self):
+        self.scripts = []
+
+    def run_js(self, script):
+        self.scripts.append(script)
+        if "play-icon" in script:
+            return ["https://www.xiaohongshu.com/search_result/video?xsec_token=video-token"]
+        return [
+            "https://www.xiaohongshu.com/search_result/image?xsec_token=image-token",
+            "https://www.xiaohongshu.com/search_result/video?xsec_token=video-token",
+        ]
+
+
+class FakeMetadataPage:
+    def run_js(self, script):
+        if "detail-title" not in script:
+            return None
+        return {
+            "note_id": "",
+            "title": "这一碗能把面食脑袋香得七荤八素",
+            "description": "#豆角焖面 #面条",
+            "cover_url": "https://sns-img.xhscdn.com/cover.jpg",
+            "author": {
+                "id": "author123",
+                "nickname": "日食记",
+                "avatar_url": "https://sns-avatar.xhscdn.com/a.jpg",
+                "profile_url": "https://www.xiaohongshu.com/user/profile/author123",
+            },
+            "stats": {
+                "liked_count": "3.5万",
+                "collected_count": "2.5万",
+                "comment_count": "1220",
+                "share_count": "",
+            },
+            "comments": [
+                {
+                    "author": "莫多西卡多西",
+                    "content": "不相信面能熟",
+                    "liked_count": "290",
+                    "time": "5天前重庆",
+                }
+            ],
+        }
+
+
 class XhsModuleTests(unittest.TestCase):
     def test_module_can_import_without_optional_runtime_dependencies(self) -> None:
         module = importlib.import_module("XHS")
@@ -94,6 +249,152 @@ class XhsModuleTests(unittest.TestCase):
         self.assertEqual(candidates[0].author_name, "摄影师")
         self.assertEqual(candidates[0].source_key, "master_url")
 
+    def test_extract_video_candidates_ignores_plain_image_url_fields(self) -> None:
+        module = importlib.import_module("XHS")
+        payload = {
+            "id": "note-image",
+            "display_title": "图片笔记",
+            "user": {"nickname": "作者"},
+            "image_list": [
+                {"url": "https://sns-img.xhscdn.com/example.webp"},
+                {"url": "https://sns-img.xhscdn.com/example.jpg"},
+            ],
+        }
+        self.assertEqual(module.extract_video_candidates(payload), [])
+
+    def test_extract_metadata_from_nested_note_payload(self) -> None:
+        module = importlib.import_module("XHS")
+        payload = {
+            "data": {
+                "items": [
+                    {
+                        "id": "note123",
+                        "note_card": {
+                            "display_title": "海边日落",
+                            "desc": "一段描述",
+                            "cover": {"url": "https://sns-img.xhscdn.com/cover.jpg"},
+                            "user": {
+                                "user_id": "user123",
+                                "nickname": "摄影师",
+                                "avatar": "https://sns-avatar.xhscdn.com/a.jpg",
+                            },
+                            "interact_info": {
+                                "liked_count": "12",
+                                "collected_count": "3",
+                                "comment_count": "4",
+                                "share_count": "5",
+                            },
+                        },
+                    }
+                ]
+            }
+        }
+        metadata = module.extract_note_metadata(payload, note_id="note123")
+        self.assertEqual(metadata["note_id"], "note123")
+        self.assertEqual(metadata["title"], "海边日落")
+        self.assertEqual(metadata["description"], "一段描述")
+        self.assertEqual(metadata["cover_url"], "https://sns-img.xhscdn.com/cover.jpg")
+        self.assertEqual(metadata["author"]["id"], "user123")
+        self.assertEqual(metadata["author"]["nickname"], "摄影师")
+        self.assertEqual(metadata["stats"]["liked_count"], "12")
+        self.assertEqual(metadata["stats"]["collected_count"], "3")
+        self.assertEqual(metadata["stats"]["comment_count"], "4")
+        self.assertEqual(metadata["stats"]["share_count"], "5")
+
+    def test_build_download_metadata_record_includes_download_context(self) -> None:
+        module = importlib.import_module("XHS")
+        candidate = module.VideoCandidate(
+            video_id="note123",
+            title="视频标题",
+            video_url="https://sns-video.xhscdn.com/a.mp4",
+            author_name="作者",
+            source_key="master_url",
+        )
+        base_metadata = {"title": "真实标题", "author": {"nickname": "真实作者"}}
+        with mock.patch.object(module, "current_timestamp", return_value="2026-05-27T17:00:00+0800"):
+            record = module.build_download_metadata_record(
+                base_metadata=base_metadata,
+                candidate=candidate,
+                queue_record=module.QueueRecord("note123", "https://www.xiaohongshu.com/explore/note123", "video-channel"),
+                output_path=Path("video/a.mp4"),
+            )
+        self.assertEqual(record["note_id"], "note123")
+        self.assertEqual(record["title"], "真实标题")
+        self.assertEqual(record["author"]["nickname"], "真实作者")
+        self.assertEqual(record["source"], "video-channel")
+        self.assertEqual(record["note_url"], "https://www.xiaohongshu.com/explore/note123")
+        self.assertEqual(record["video_url"], "https://sns-video.xhscdn.com/a.mp4")
+        self.assertEqual(record["downloaded_path"], "video/a.mp4")
+        self.assertEqual(record["downloaded_at"], "2026-05-27T17:00:00+0800")
+        self.assertEqual(record["comments"], [])
+
+    def test_build_download_metadata_record_preserves_metadata_comments(self) -> None:
+        module = importlib.import_module("XHS")
+        candidate = module.VideoCandidate(
+            video_id="note123",
+            title="视频标题",
+            video_url="https://sns-video.xhscdn.com/a.mp4",
+            author_name="作者",
+            source_key="master_url",
+        )
+        base_metadata = {
+            "comments": [
+                {
+                    "author": "评论用户",
+                    "content": "评论内容",
+                    "liked_count": "9",
+                    "time": "1小时前",
+                }
+            ]
+        }
+
+        record = module.build_download_metadata_record(
+            base_metadata=base_metadata,
+            candidate=candidate,
+            queue_record=module.QueueRecord("note123", "https://www.xiaohongshu.com/explore/note123", "video-channel"),
+            output_path=Path("video/a.mp4"),
+        )
+
+        self.assertEqual(record["comments"], base_metadata["comments"])
+
+    def test_extract_note_metadata_from_page_uses_visible_dom(self) -> None:
+        module = importlib.import_module("XHS")
+        metadata = module.extract_note_metadata_from_page(FakeMetadataPage(), note_id="note123", max_comments=20)
+        self.assertEqual(metadata["note_id"], "note123")
+        self.assertEqual(metadata["title"], "这一碗能把面食脑袋香得七荤八素")
+        self.assertEqual(metadata["description"], "#豆角焖面 #面条")
+        self.assertEqual(metadata["cover_url"], "https://sns-img.xhscdn.com/cover.jpg")
+        self.assertEqual(metadata["author"]["id"], "author123")
+        self.assertEqual(metadata["author"]["nickname"], "日食记")
+        self.assertEqual(metadata["stats"]["liked_count"], "3.5万")
+        self.assertEqual(metadata["stats"]["collected_count"], "2.5万")
+        self.assertEqual(metadata["stats"]["comment_count"], "1220")
+        self.assertEqual(metadata["comments"][0]["content"], "不相信面能熟")
+
+    def test_append_jsonl_record_writes_utf8_json_line(self) -> None:
+        module = importlib.import_module("XHS")
+        with tempfile.TemporaryDirectory() as temp_dir:
+            path = Path(temp_dir) / "meta" / "records.jsonl"
+            module.append_jsonl_record(path, {"title": "海边日落", "count": 1})
+            module.append_jsonl_record(path, {"title": "猫咪", "count": 2})
+            lines = path.read_text(encoding="utf-8").splitlines()
+        self.assertEqual(len(lines), 2)
+        self.assertIn("海边日落", lines[0])
+        self.assertIn("猫咪", lines[1])
+
+    def test_extract_video_candidates_from_escaped_html_state(self) -> None:
+        module = importlib.import_module("XHS")
+        html = (
+            '<script>{"display_title":"视频标题","nickname":"作者",'
+            r'\"master_url\":\"http:\/\/sns-video-qc.xhscdn.com\/stream\/a.mp4?sign=1&t=2\"'
+            '}</script>'
+        )
+        candidates = module.extract_video_candidates_from_html(html, video_id="note123")
+        self.assertEqual(len(candidates), 1)
+        self.assertEqual(candidates[0].video_id, "note123")
+        self.assertEqual(candidates[0].video_url, "http://sns-video-qc.xhscdn.com/stream/a.mp4?sign=1&t=2")
+        self.assertEqual(candidates[0].source_key, "html_master_url")
+
     def test_build_output_path_uses_author_title_and_video_id(self) -> None:
         module = importlib.import_module("XHS")
         candidate = module.VideoCandidate(
@@ -108,7 +409,7 @@ class XhsModuleTests(unittest.TestCase):
 
     def test_build_browser_address_from_port(self) -> None:
         module = importlib.import_module("XHS")
-        self.assertEqual(module.build_browser_address(9224), "127.0.0.1:9224")
+        self.assertEqual(module.build_browser_address(9223), "127.0.0.1:9223")
         self.assertIsNone(module.build_browser_address(None))
 
     def test_ensure_browser_debug_port_ready_accepts_open_port(self) -> None:
@@ -117,14 +418,14 @@ class XhsModuleTests(unittest.TestCase):
         connection.__enter__.return_value = connection
         connection.__exit__.return_value = False
         with mock.patch.object(module.socket, "create_connection", return_value=connection) as mocked_connect:
-            module.ensure_browser_debug_port_ready(9224)
+            module.ensure_browser_debug_port_ready(9223)
         mocked_connect.assert_called_once()
 
     def test_ensure_browser_debug_port_ready_rejects_closed_port(self) -> None:
         module = importlib.import_module("XHS")
         with mock.patch.object(module.socket, "create_connection", side_effect=OSError("boom")):
             with self.assertRaisesRegex(RuntimeError, "login_xhs.py"):
-                module.ensure_browser_debug_port_ready(9224)
+                module.ensure_browser_debug_port_ready(9223)
 
     def test_extract_feed_payload_uses_dict_body(self) -> None:
         module = importlib.import_module("XHS")
@@ -144,9 +445,21 @@ class XhsModuleTests(unittest.TestCase):
         args = module.build_parser().parse_args([])
         self.assertEqual(args.max_videos, 10)
         self.assertEqual(args.output_dir, "video")
-        self.assertEqual(args.browser_port, 9224)
+        self.assertEqual(args.browser_port, 9223)
         self.assertEqual(args.timeout, 20)
         self.assertEqual(args.start_url, module.DEFAULT_EXPLORE_URL)
+        self.assertFalse(args.use_current_page)
+        self.assertTrue(args.human_mode)
+        self.assertEqual(args.min_wait, 2.0)
+        self.assertEqual(args.max_wait, 6.0)
+        self.assertEqual(args.long_break_every, 4)
+        self.assertEqual(args.max_runtime, 0.0)
+        self.assertEqual(args.source, "explore")
+        self.assertIsNone(args.queue_file)
+        self.assertEqual(args.target_videos, 0)
+        self.assertEqual(args.retry_limit, 1)
+        self.assertEqual(args.min_video_bytes, 200 * 1024)
+        self.assertIsNone(args.report_file)
 
     def test_main_invokes_collect_videos_with_cli_values(self) -> None:
         module = importlib.import_module("XHS")
@@ -172,6 +485,403 @@ class XhsModuleTests(unittest.TestCase):
         self.assertEqual(kwargs["output_dir"].as_posix(), "downloads")
         self.assertEqual(kwargs["browser_port"], 9334)
         self.assertEqual(kwargs["timeout"], 7)
+        self.assertFalse(kwargs["use_current_page"])
+        self.assertTrue(kwargs["human_mode"])
+
+    def test_build_source_url_supports_video_channel_and_explore(self) -> None:
+        module = importlib.import_module("XHS")
+        self.assertEqual(module.build_source_url("explore"), module.DEFAULT_EXPLORE_URL)
+        self.assertEqual(
+            module.build_source_url("video-channel"),
+            "https://www.xiaohongshu.com/explore?channel_id=homefeed.video_v3",
+        )
+
+    def test_build_source_url_supports_encoded_search_keyword(self) -> None:
+        module = importlib.import_module("XHS")
+        self.assertEqual(
+            module.build_source_url("search", keyword="猫咪 搞笑"),
+            "https://www.xiaohongshu.com/search_result?keyword=%E7%8C%AB%E5%92%AA%20%E6%90%9E%E7%AC%91&source=web_search_result_notes&type=51",
+        )
+
+    def test_main_invokes_queue_mode_when_queue_file_is_provided(self) -> None:
+        module = importlib.import_module("XHS")
+        with mock.patch.object(module, "run_queue_download", return_value=5) as mocked_run:
+            exit_code = module.main(
+                [
+                    "--source",
+                    "video-channel",
+                    "--target-videos",
+                    "1000",
+                    "--queue-file",
+                    "data/q.jsonl",
+                    "--retry-limit",
+                    "2",
+                    "--keyword",
+                    "猫咪",
+                ]
+            )
+        self.assertEqual(exit_code, 0)
+        mocked_run.assert_called_once()
+        _, kwargs = mocked_run.call_args
+        self.assertEqual(kwargs["source"], "video-channel")
+        self.assertEqual(kwargs["target_videos"], 1000)
+        self.assertEqual(kwargs["queue_file"].as_posix(), "data/q.jsonl")
+        self.assertEqual(kwargs["retry_limit"], 2)
+        self.assertEqual(kwargs["keyword"], "猫咪")
+
+    def test_main_passes_queue_report_and_min_video_bytes_options(self) -> None:
+        module = importlib.import_module("XHS")
+        with mock.patch.object(module, "run_queue_download", return_value=5) as mocked_run:
+            exit_code = module.main(
+                [
+                    "--target-videos",
+                    "5",
+                    "--queue-file",
+                    "data/q.jsonl",
+                    "--min-video-bytes",
+                    "4096",
+                    "--report-file",
+                    "data/report.json",
+                ]
+            )
+        self.assertEqual(exit_code, 0)
+        _, kwargs = mocked_run.call_args
+        self.assertEqual(kwargs["min_video_bytes"], 4096)
+        self.assertEqual(kwargs["report_file"].as_posix(), "data/report.json")
+
+    def test_download_video_rejects_webp_response_before_writing_file(self) -> None:
+        module = importlib.import_module("XHS")
+        response = FakeDownloadResponse(b"RIFF....WEBP", content_type="image/webp")
+        with self.assertRaisesRegex(ValueError, "非视频响应"):
+            module.download_video(
+                requests_module=FakeRequests(response),
+                headers={},
+                video_url="https://sns-img.xhscdn.com/example.webp",
+                output_path=mock.MagicMock(),
+            )
+
+    def test_download_video_accepts_mp4_signature(self) -> None:
+        module = importlib.import_module("XHS")
+        response = FakeDownloadResponse(b"\x00\x00\x00\x18ftypmp42payload", content_type="application/octet-stream")
+        with tempfile.TemporaryDirectory() as temp_dir:
+            output_path = Path(temp_dir) / "example.mp4"
+            module.download_video(
+                requests_module=FakeRequests(response),
+                headers={},
+                video_url="https://sns-video.xhscdn.com/example.mp4",
+                output_path=output_path,
+                min_video_bytes=0,
+            )
+            self.assertEqual(output_path.read_bytes(), b"\x00\x00\x00\x18ftypmp42payload")
+
+    def test_download_video_streams_chunks_without_loading_full_content(self) -> None:
+        module = importlib.import_module("XHS")
+        chunks = [b"\x00\x00\x00\x18ftypmp42", b"payload", b"more"]
+        requests = FakeRequests(FakeStreamingResponse(chunks, content_type="application/octet-stream"))
+        with tempfile.TemporaryDirectory() as temp_dir:
+            output_path = Path(temp_dir) / "streamed.mp4"
+
+            module.download_video(
+                requests_module=requests,
+                headers={"referer": "note"},
+                video_url="https://sns-video.xhscdn.com/example.mp4",
+                output_path=output_path,
+                min_video_bytes=0,
+            )
+
+            self.assertEqual(output_path.read_bytes(), b"".join(chunks))
+        self.assertTrue(requests.calls[0]["stream"])
+
+    def test_validate_video_response_rejects_tiny_video_payload(self) -> None:
+        module = importlib.import_module("XHS")
+        response = FakeDownloadResponse(b"\x00\x00\x00\x18ftypmp42payload", content_type="application/octet-stream")
+
+        with self.assertRaisesRegex(ValueError, "视频响应过小"):
+            module.validate_video_response(
+                response,
+                "https://sns-video.xhscdn.com/example.mp4",
+                min_video_bytes=1024,
+            )
+
+    def test_normalize_note_urls_deduplicates_explore_links(self) -> None:
+        module = importlib.import_module("XHS")
+        urls = module.normalize_note_urls(
+            [
+                "https://www.xiaohongshu.com/explore/abc",
+                "https://www.xiaohongshu.com/explore/abc?xsec_token=token",
+                "/explore/def?xsec_token=token",
+                "https://www.xiaohongshu.com/user/profile/123",
+            ]
+        )
+        self.assertEqual(
+            urls,
+            [
+                "https://www.xiaohongshu.com/explore/abc?xsec_token=token",
+                "https://www.xiaohongshu.com/explore/def?xsec_token=token",
+            ],
+        )
+
+    def test_normalize_note_urls_prefers_xsec_token_url_for_same_note(self) -> None:
+        module = importlib.import_module("XHS")
+        urls = module.normalize_note_urls(
+            [
+                "https://www.xiaohongshu.com/explore/abc",
+                "https://www.xiaohongshu.com/explore/abc?xsec_token=token&xsec_source=",
+            ],
+        )
+        self.assertEqual(urls, ["https://www.xiaohongshu.com/explore/abc?xsec_token=token&xsec_source="])
+
+    def test_extract_note_id_from_url_supports_search_result_detail(self) -> None:
+        module = importlib.import_module("XHS")
+        self.assertEqual(
+            module.extract_note_id_from_url("https://www.xiaohongshu.com/search_result/abc?xsec_token=token"),
+            "abc",
+        )
+
+    def test_normalize_note_urls_preserves_tokenized_search_result_url(self) -> None:
+        module = importlib.import_module("XHS")
+        urls = module.normalize_note_urls(
+            [
+                "https://www.xiaohongshu.com/explore/abc",
+                "https://www.xiaohongshu.com/search_result/abc?xsec_token=token&xsec_source=",
+            ],
+        )
+        self.assertEqual(urls, ["https://www.xiaohongshu.com/search_result/abc?xsec_token=token&xsec_source="])
+
+    def test_collect_note_urls_from_page_includes_search_result_links(self) -> None:
+        module = importlib.import_module("XHS")
+        page = FakeLinkPage(
+            [
+                "https://www.xiaohongshu.com/search_result/abc?xsec_token=token",
+                "https://www.xiaohongshu.com/explore/def?xsec_token=token2",
+            ]
+        )
+        urls = module.collect_note_urls_from_page(page, limit=10)
+        self.assertEqual(
+            urls,
+            [
+                "https://www.xiaohongshu.com/search_result/abc?xsec_token=token",
+                "https://www.xiaohongshu.com/explore/def?xsec_token=token2",
+            ],
+        )
+        self.assertIn('/search_result/', page.scripts[0])
+
+    def test_collect_note_urls_from_page_can_filter_video_cards(self) -> None:
+        module = importlib.import_module("XHS")
+        page = FakeVideoOnlyLinkPage()
+        urls = module.collect_note_urls_from_page(page, limit=10, video_only=True)
+        self.assertEqual(urls, ["https://www.xiaohongshu.com/search_result/video?xsec_token=video-token"])
+        self.assertIn("play-icon", page.scripts[0])
+
+    def test_wait_for_note_urls_from_page_polls_until_links_are_rendered(self) -> None:
+        module = importlib.import_module("XHS")
+        page = FakeDelayedLinkPage()
+        with mock.patch.object(module.time, "sleep") as mocked_sleep:
+            urls = module.wait_for_note_urls_from_page(page, limit=10, timeout=2, interval=0.1)
+        self.assertEqual(urls, ["https://www.xiaohongshu.com/search_result/abc?xsec_token=token"])
+        mocked_sleep.assert_called_once_with(0.1)
+
+    def test_collect_note_urls_with_browse_accumulates_after_scroll(self) -> None:
+        module = importlib.import_module("XHS")
+        page = FakeGrowingLinkPage()
+        settings = module.HumanBrowseSettings(enabled=False)
+        with mock.patch.object(module, "run_human_browse_sequence") as mocked_browse:
+            urls = module.collect_note_urls_with_browse(page, limit=10, human_settings=settings, rounds=2)
+        self.assertEqual(
+            urls,
+            [
+                "https://www.xiaohongshu.com/search_result/one?xsec_token=token1",
+                "https://www.xiaohongshu.com/search_result/two?xsec_token=token2",
+            ],
+        )
+        mocked_browse.assert_called_once()
+
+    def test_filter_unvisited_note_urls_skips_seen_note_ids(self) -> None:
+        module = importlib.import_module("XHS")
+        urls = [
+            "https://www.xiaohongshu.com/explore/abc?xsec_token=token",
+            "https://www.xiaohongshu.com/explore/def?xsec_token=token",
+        ]
+        self.assertEqual(
+            module.filter_unvisited_note_urls(urls, {"abc"}),
+            ["https://www.xiaohongshu.com/explore/def?xsec_token=token"],
+        )
+
+    def test_create_human_browse_plan_uses_wait_and_scroll_ranges(self) -> None:
+        module = importlib.import_module("XHS")
+        settings = module.HumanBrowseSettings(
+            min_wait=2.0,
+            max_wait=6.0,
+            reverse_scroll_probability=1.0,
+            min_scroll=500,
+            max_scroll=1200,
+        )
+        plan = module.create_human_browse_plan(settings, random_module=module.random.Random(7))
+        self.assertGreaterEqual(plan.primary_wait, 2.0)
+        self.assertLessEqual(plan.primary_wait, 6.0)
+        self.assertGreaterEqual(plan.down_distance, 500)
+        self.assertLessEqual(plan.down_distance, 1200)
+        self.assertGreater(plan.reverse_distance, 0)
+
+    def test_run_human_browse_sequence_scrolls_and_waits(self) -> None:
+        module = importlib.import_module("XHS")
+        page = FakeScrollPage()
+        plan = module.HumanBrowsePlan(
+            down_distance=800,
+            primary_wait=2.5,
+            reverse_distance=200,
+            reverse_wait=1.5,
+            settle_wait=3.0,
+        )
+        with mock.patch.object(module.time, "sleep") as mocked_sleep:
+            module.run_human_browse_sequence(page, plan)
+        self.assertIn("const distance = 800;", page.scripts[0])
+        self.assertIn("const distance = -200;", page.scripts[1])
+        self.assertIn("const distance = 400;", page.scripts[2])
+        self.assertIn("scrollBy(0, distance)", page.scripts[0])
+        mocked_sleep.assert_has_calls([mock.call(2.5), mock.call(1.5), mock.call(3.0)])
+
+    def test_load_visible_comments_scrolls_until_comment_dom_exists(self) -> None:
+        module = importlib.import_module("XHS")
+        page = FakeDelayedCommentPage()
+        settings = module.HumanBrowseSettings(enabled=True, min_wait=0.1, max_wait=0.1)
+
+        with mock.patch.object(module.time, "sleep") as mocked_sleep:
+            loaded = module.load_visible_comments(page, human_settings=settings, max_comments=20, timeout=1.0)
+
+        self.assertTrue(loaded)
+        self.assertGreaterEqual(page.comment_checks, 2)
+        self.assertEqual(len(page.scroll_scripts), 1)
+        mocked_sleep.assert_called_once_with(0.1)
+
+    def test_should_take_long_break_uses_every_n_downloads(self) -> None:
+        module = importlib.import_module("XHS")
+        settings = module.HumanBrowseSettings(long_break_every=4)
+        self.assertFalse(module.should_take_long_break(0, settings))
+        self.assertFalse(module.should_take_long_break(3, settings))
+        self.assertTrue(module.should_take_long_break(4, settings))
+        self.assertTrue(module.should_take_long_break(8, settings))
+
+    def test_queue_round_trip_jsonl(self) -> None:
+        module = importlib.import_module("XHS")
+        with tempfile.TemporaryDirectory() as temp_dir:
+            queue_path = Path(temp_dir) / "queue.jsonl"
+            records = [
+                module.QueueRecord(
+                    note_id="note1",
+                    url="https://www.xiaohongshu.com/explore/note1?xsec_token=a",
+                    source="video-channel",
+                )
+            ]
+            module.save_queue(queue_path, records)
+            loaded = module.load_queue(queue_path)
+        self.assertEqual(loaded, records)
+
+    def test_merge_note_urls_into_queue_deduplicates_existing_notes(self) -> None:
+        module = importlib.import_module("XHS")
+        records = [
+            module.QueueRecord(
+                note_id="note1",
+                url="https://www.xiaohongshu.com/explore/note1?xsec_token=a",
+                source="explore",
+                status="downloaded",
+            )
+        ]
+        merged = module.merge_note_urls_into_queue(
+            records,
+            [
+                "https://www.xiaohongshu.com/explore/note1?xsec_token=a",
+                "https://www.xiaohongshu.com/explore/note2?xsec_token=b",
+            ],
+            source="video-channel",
+        )
+        self.assertEqual([record.note_id for record in merged], ["note1", "note2"])
+        self.assertEqual(merged[0].status, "downloaded")
+        self.assertEqual(merged[1].status, "pending")
+
+    def test_count_queue_status_counts_records_by_status(self) -> None:
+        module = importlib.import_module("XHS")
+        records = [
+            module.QueueRecord("one", "url1", "source", status="downloaded"),
+            module.QueueRecord("two", "url2", "source", status="failed"),
+            module.QueueRecord("three", "url3", "source", status="downloaded"),
+        ]
+        self.assertEqual(
+            module.count_queue_status(records),
+            {"downloaded": 2, "failed": 1},
+        )
+
+    def test_mark_queue_record_downloaded_updates_status_and_path(self) -> None:
+        module = importlib.import_module("XHS")
+        record = module.QueueRecord("note1", "url", "source")
+        updated = module.mark_queue_record_downloaded(record, Path("video/a.mp4"))
+        self.assertEqual(updated.status, "downloaded")
+        self.assertEqual(updated.downloaded_path, "video/a.mp4")
+        self.assertEqual(updated.last_error, "")
+
+    def test_mark_queue_record_skipped_records_reason(self) -> None:
+        module = importlib.import_module("XHS")
+        record = module.QueueRecord("note1", "url", "source")
+        updated = module.mark_queue_record_skipped(record, "no video")
+        self.assertEqual(updated.status, "skipped_image")
+        self.assertEqual(updated.last_error, "no video")
+
+    def test_mark_queue_record_failed_respects_retry_limit(self) -> None:
+        module = importlib.import_module("XHS")
+        record = module.QueueRecord("note1", "url", "source", attempts=0)
+        retry = module.mark_queue_record_failed(record, "timeout", retry_limit=2)
+        self.assertEqual(retry.status, "pending")
+        self.assertEqual(retry.attempts, 1)
+        failed = module.mark_queue_record_failed(retry, "timeout", retry_limit=2)
+        self.assertEqual(failed.status, "failed")
+        self.assertEqual(failed.attempts, 2)
+
+    def test_build_run_report_summarizes_queue_metadata_and_files(self) -> None:
+        module = importlib.import_module("XHS")
+        with tempfile.TemporaryDirectory() as temp_dir:
+            output_dir = Path(temp_dir) / "video"
+            output_dir.mkdir()
+            video_path = output_dir / "a.mp4"
+            video_path.write_bytes(b"\x00\x00\x00\x18ftypmp42payload")
+            metadata_file = output_dir / "metadata.jsonl"
+            metadata_file.write_text(
+                "\n".join(
+                    [
+                        '{"note_id":"one","comments":[{"content":"a"},{"content":"b"}],"file_size_bytes":20}',
+                        '{"note_id":"two","comments":[],"file_size_bytes":30}',
+                    ]
+                )
+                + "\n",
+                encoding="utf-8",
+            )
+            records = [
+                module.QueueRecord("one", "url1", "source", status="downloaded", downloaded_path=video_path.as_posix()),
+                module.QueueRecord("two", "url2", "source", status="failed", last_error="download_timeout"),
+            ]
+
+            report = module.build_run_report(
+                source="video-channel",
+                target_videos=10,
+                queue_file=Path(temp_dir) / "queue.jsonl",
+                output_dir=output_dir,
+                metadata_file=metadata_file,
+                records=records,
+                downloaded_this_run=1,
+                started_at=100.0,
+                finished_at=130.0,
+            )
+
+        self.assertEqual(report["source"], "video-channel")
+        self.assertEqual(report["target_videos"], 10)
+        self.assertEqual(report["downloaded_this_run"], 1)
+        self.assertEqual(report["queue_status"]["downloaded"], 1)
+        self.assertEqual(report["queue_status"]["failed"], 1)
+        self.assertEqual(report["metadata_rows"], 2)
+        self.assertEqual(report["metadata_with_comments"], 1)
+        self.assertEqual(report["total_comments"], 2)
+        self.assertEqual(report["video_files"], 1)
+        self.assertGreater(report["output_dir_bytes"], 0)
+        self.assertEqual(report["elapsed_seconds"], 30.0)
 
 
 if __name__ == "__main__":