feat: add search result video crawling

Douyin.py

@@ -12,6 +12,7 @@ from __future__ import annotations
 
 import argparse
 import json
+import random
 import re
 import socket
 import sys
@@ -19,6 +20,7 @@ import time
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
+from urllib.parse import quote
 
 DEFAULT_USER_URL = (
     "https://www.douyin.com/user/"
@@ -27,8 +29,12 @@ DEFAULT_USER_URL = (
 )
 DEFAULT_BROWSER_PORT = 9223
 LISTEN_TARGET = "web/aweme/post/"
+RECOMMENDATION_LISTEN_TARGET = "aweme/v2/web/module/feed/"
 SINGLE_VIDEO_LISTEN_TARGET = "web/aweme/detail/"
+SEARCH_LISTEN_TARGET = "aweme/v1/web/general/search/single"
+MAX_FILENAME_BYTES = 240
 INVALID_FILENAME_CHARS = re.compile(r'[\\/:*?"<>|\r\n\t]')
+RECOMMENDATION_URL_PATTERN = re.compile(r"^https?://www\.douyin\.com/?(?:jingxuan)?(?:\?.*)?$")
 CREATOR_URL_PATTERN = re.compile(r"^https?://www\.douyin\.com/user/[^/?#]+(?:\?.*)?$")
 VIDEO_URL_PATTERN = re.compile(r"^https?://www\.douyin\.com/video/(?P<aweme_id>\d+)(?:[/?#].*)?$")
 AWEME_ID_PATTERN = re.compile(r"^\d{5,}$")
@@ -42,11 +48,52 @@ class ResolvedTarget:
     aweme_id: str | None = None
 
 
+@dataclass(frozen=True)
+class ScrollSettings:
+    mode: str = "human"
+    min_wait: float = 2.0
+    max_wait: float = 8.0
+    reverse_scroll_probability: float = 0.2
+    max_runtime: float = 600.0
+    min_scroll: int = 300
+    max_scroll: int = 900
+    min_reverse_scroll: int = 80
+    max_reverse_scroll: int = 250
+
+
+@dataclass(frozen=True)
+class HumanScrollPlan:
+    down_distance: int
+    down_wait: float
+    reverse_distance: int = 0
+    reverse_wait: float = 0.0
+    settle_wait: float = 0.0
+
+
 def sanitize_filename(value: str, fallback: str = "untitled") -> str:
     cleaned = INVALID_FILENAME_CHARS.sub("_", value).strip(" ._")
     return cleaned or fallback
 
 
+def truncate_utf8_bytes(value: str, max_bytes: int) -> str:
+    if len(value.encode("utf-8")) <= max_bytes:
+        return value
+
+    result = ""
+    used = 0
+    for character in value:
+        character_bytes = len(character.encode("utf-8"))
+        if used + character_bytes > max_bytes:
+            break
+        result += character
+        used += character_bytes
+    return result.rstrip(" ._")
+
+
+def is_recommendation_url(value: str) -> bool:
+    return bool(RECOMMENDATION_URL_PATTERN.match(value.strip()))
+
+
 def is_creator_url(value: str) -> bool:
     return bool(CREATOR_URL_PATTERN.match(value.strip()))
 
@@ -70,8 +117,14 @@ def build_video_page_url(aweme_id: str) -> str:
     return f"https://www.douyin.com/video/{aweme_id}"
 
 
+def build_search_page_url(keyword: str) -> str:
+    return f"https://www.douyin.com/search/{quote(keyword)}?type=general"
+
+
 def parse_target_input(value: str, source: str) -> ResolvedTarget:
     normalized = value.strip()
+    if is_recommendation_url(normalized):
+        return ResolvedTarget(kind="recommendation", value=normalized, source=source)
     if is_creator_url(normalized):
         return ResolvedTarget(kind="creator", value=normalized, source=source)
     if is_video_url(normalized):
@@ -131,9 +184,62 @@ def choose_video_url(url_list: list[str]) -> str:
     raise ValueError("url_list 为空，无法选择视频地址。")
 
 
-def build_output_path(title: str, video_id: str, output_dir: Path = Path("video")) -> Path:
+def extract_url_list_from_play_addr(play_addr: Any) -> list[str]:
+    if not isinstance(play_addr, dict):
+        return []
+
+    url_list = play_addr.get("url_list") or []
+    if not isinstance(url_list, list):
+        return []
+
+    return [str(url) for url in url_list if str(url).strip()]
+
+
+def extract_video_url_list(video: Any) -> list[str]:
+    if not isinstance(video, dict):
+        return []
+
+    for address_key in ("play_addr", "play_addr_h264", "play_addr_lowbr"):
+        url_list = extract_url_list_from_play_addr(video.get(address_key))
+        if url_list:
+            return url_list
+
+    bit_rate_list = video.get("bit_rate") or []
+    if not isinstance(bit_rate_list, list):
+        return []
+
+    for bit_rate in bit_rate_list:
+        if not isinstance(bit_rate, dict):
+            continue
+        url_list = extract_url_list_from_play_addr(bit_rate.get("play_addr"))
+        if url_list:
+            return url_list
+
+    return []
+
+
+def build_output_path(
+    title: str, 
+    video_id: str, 
+    output_dir: Path = Path("video"),
+    author_name: str | None = None,
+) -> Path:
     safe_title = sanitize_filename(title, fallback="untitled")
-    return output_dir / f"{safe_title}-{video_id}.mp4"
+    suffix = f"-{video_id}.mp4"
+    if author_name:
+        safe_author = sanitize_filename(author_name, fallback="unknown")
+        prefix = f"[{safe_author}]"
+    else:
+        prefix = ""
+
+    title_budget = MAX_FILENAME_BYTES - len(prefix.encode("utf-8")) - len(suffix.encode("utf-8"))
+    if title_budget < 1:
+        prefix_budget = MAX_FILENAME_BYTES - len(suffix.encode("utf-8")) - 1
+        prefix = truncate_utf8_bytes(prefix, max(1, prefix_budget))
+        title_budget = MAX_FILENAME_BYTES - len(prefix.encode("utf-8")) - len(suffix.encode("utf-8"))
+
+    filename = f"{prefix}{truncate_utf8_bytes(safe_title, max(1, title_budget))}{suffix}"
+    return output_dir / filename
 
 
 def build_browser_address(browser_port: int | None) -> str | None:
@@ -182,8 +288,7 @@ def parse_aweme_items(body: Any) -> list[dict[str, str]]:
             continue
 
         video = aweme.get("video") or {}
-        play_addr = video.get("play_addr") or {}
-        url_list = play_addr.get("url_list") or []
+        url_list = extract_video_url_list(video)
         if not url_list:
             continue
 
@@ -192,11 +297,18 @@ def parse_aweme_items(body: Any) -> list[dict[str, str]]:
             continue
 
         title = str(aweme.get("desc") or "").strip() or "untitled"
+        
+        author = aweme.get("author") or {}
+        author_name = str(author.get("nickname") or "").strip() or "unknown"
+        author_id = str(author.get("uid") or "").strip() or "unknown"
+        
         items.append(
             {
                 "title": title,
                 "video_id": video_id,
-                "video_url": choose_video_url([str(url) for url in url_list]),
+                "video_url": choose_video_url(url_list),
+                "author_name": author_name,
+                "author_id": author_id,
             }
         )
 
@@ -219,6 +331,25 @@ def parse_single_aweme_item(body: Any) -> dict[str, str]:
     raise ValueError("接口响应中缺少可下载的单视频数据。")
 
 
+def parse_search_items(body: Any) -> list[dict[str, str]]:
+    if not isinstance(body, dict):
+        raise ValueError("接口响应不是字典，无法解析。")
+
+    data = body.get("data")
+    if not isinstance(data, list):
+        raise ValueError("搜索接口响应中缺少 data。")
+
+    aweme_list = []
+    for entry in data:
+        if not isinstance(entry, dict):
+            continue
+        aweme_info = entry.get("aweme_info")
+        if isinstance(aweme_info, dict):
+            aweme_list.append(aweme_info)
+
+    return parse_aweme_items({"aweme_list": aweme_list})
+
+
 def build_headers(referer: str) -> dict[str, str]:
     return {
         "referer": referer,
@@ -260,7 +391,8 @@ def create_page(chromium_page_cls: Any, chromium_options_cls: Any, browser_port:
 
 def wait_for_aweme_packet(page: Any, timeout: int) -> Any | None:
     try:
-        return page.listen.wait(timeout=timeout)
+        packet = page.listen.wait(timeout=timeout)
+        return packet if packet else None
     except Exception as exc:
         print(f"[WARN] 等待接口数据超时或失败: {exc}")
         return None
@@ -271,6 +403,95 @@ def scroll_to_next_page(page: Any) -> None:
     time.sleep(2)
 
 
+def create_human_scroll_plan(
+    settings: ScrollSettings,
+    random_module: Any = random,
+) -> HumanScrollPlan:
+    down_distance = random_module.randint(settings.min_scroll, settings.max_scroll)
+    down_wait = random_module.uniform(settings.min_wait, settings.max_wait)
+    settle_wait = random_module.uniform(settings.min_wait, settings.max_wait)
+
+    reverse_distance = 0
+    reverse_wait = 0.0
+    if random_module.random() < settings.reverse_scroll_probability:
+        reverse_distance = random_module.randint(
+            settings.min_reverse_scroll,
+            settings.max_reverse_scroll,
+        )
+        reverse_wait = random_module.uniform(1.0, min(3.0, settings.max_wait))
+
+    return HumanScrollPlan(
+        down_distance=down_distance,
+        down_wait=down_wait,
+        reverse_distance=reverse_distance,
+        reverse_wait=reverse_wait,
+        settle_wait=settle_wait,
+    )
+
+
+def run_scroll_step(page: Any, distance: int) -> bool:
+    script = f"""
+const distance = {distance};
+function findMainScrollContainer() {{
+  const preferredSelectors = ['.tKqwmYAX', '.route-scroll-container', '.semi-tabs-content'];
+  for (const selector of preferredSelectors) {{
+    const el = document.querySelector(selector);
+    if (el && el.scrollHeight > el.clientHeight + 20) {{
+      return el;
+    }}
+  }}
+
+  const candidates = Array.from(document.querySelectorAll('*'))
+    .filter((el) => {{
+      const rect = el.getBoundingClientRect();
+      return rect.width > 300
+        && rect.height > 200
+        && el.scrollHeight > el.clientHeight + 20;
+    }})
+    .sort((a, b) => {{
+      const areaA = a.getBoundingClientRect().width * a.getBoundingClientRect().height;
+      const areaB = b.getBoundingClientRect().width * b.getBoundingClientRect().height;
+      return areaB - areaA;
+    }});
+
+  return candidates[0] || null;
+}}
+
+const scrollTarget = findMainScrollContainer();
+if (scrollTarget) {{
+  scrollTarget.scrollBy(0, distance);
+  return true;
+}}
+return false;
+"""
+    scrolled_container = bool(page.run_js(script))
+    if not scrolled_container:
+        page.run_js(f"window.scrollBy(0, {distance});")
+    return scrolled_container
+
+
+def run_human_scroll_sequence(page: Any, plan: HumanScrollPlan) -> None:
+    run_scroll_step(page, plan.down_distance)
+    print(f"[INFO] 向下滚动 {plan.down_distance}px，停留 {plan.down_wait:.1f}s")
+    time.sleep(plan.down_wait)
+
+    if plan.reverse_distance > 0:
+        run_scroll_step(page, -plan.reverse_distance)
+        print(f"[INFO] 小幅回滚 {plan.reverse_distance}px，停留 {plan.reverse_wait:.1f}s")
+        time.sleep(plan.reverse_wait)
+        forward_distance = plan.reverse_distance * 2
+        run_scroll_step(page, forward_distance)
+
+    if plan.settle_wait > 0:
+        print(f"[INFO] 继续停留 {plan.settle_wait:.1f}s")
+        time.sleep(plan.settle_wait)
+
+
+def human_like_scroll(page: Any, settings: ScrollSettings | None = None) -> None:
+    scroll_settings = settings or ScrollSettings()
+    run_human_scroll_sequence(page, create_human_scroll_plan(scroll_settings))
+
+
 def download_video(
     requests_module: Any,
     headers: dict[str, str],
@@ -363,6 +584,208 @@ def collect_videos(
     return downloaded
 
 
+def collect_recommendations(
+    max_videos: int,
+    timeout: int,
+    output_dir: Path,
+    browser_port: int | None,
+    scroll_settings: ScrollSettings | None = None,
+) -> int:
+    requests_module, chromium_page_cls, chromium_options_cls = import_runtime_dependencies()
+    headers = build_headers("https://www.douyin.com/")
+    if browser_port is not None:
+        ensure_browser_debug_port_ready(browser_port)
+    page = create_page(chromium_page_cls, chromium_options_cls, browser_port)
+    page.listen.start(RECOMMENDATION_LISTEN_TARGET)
+
+    print("[INFO] 正在打开抖音推荐流。若出现登录或验证码，请先在浏览器窗口里完成。")
+    page.get("https://www.douyin.com/")
+    time.sleep(3)
+
+    downloaded = 0
+    seen_ids: set[str] = set()
+    consecutive_empty = 0
+    max_consecutive_empty = 6
+    settings = scroll_settings or ScrollSettings()
+    started_at = time.monotonic()
+
+    while downloaded < max_videos:
+        if settings.max_runtime > 0 and time.monotonic() - started_at >= settings.max_runtime:
+            print("[INFO] 已达到最大运行时间，结束抓取。")
+            break
+
+        packet = wait_for_aweme_packet(page, timeout=timeout)
+        if packet is None:
+            consecutive_empty += 1
+            if consecutive_empty >= max_consecutive_empty:
+                print("[INFO] 连续多次未获取到新数据，结束抓取。")
+                break
+            human_like_scroll(page, settings=settings)
+            continue
+
+        try:
+            payload = extract_aweme_payload(packet.response)
+            items = parse_aweme_items(payload)
+        except Exception as exc:
+            print(f"[WARN] 解析接口数据失败: {exc}")
+            consecutive_empty += 1
+            if consecutive_empty >= max_consecutive_empty:
+                break
+            human_like_scroll(page, settings=settings)
+            continue
+
+        if not items:
+            consecutive_empty += 1
+            if consecutive_empty >= max_consecutive_empty:
+                break
+            human_like_scroll(page, settings=settings)
+            continue
+
+        consecutive_empty = 0
+        new_items_in_batch = 0
+        
+        for item in items:
+            if item["video_id"] in seen_ids:
+                continue
+            
+            if downloaded >= max_videos:
+                break
+
+            seen_ids.add(item["video_id"])
+            output_path = build_output_path(
+                title=item["title"],
+                video_id=item["video_id"],
+                output_dir=output_dir,
+                author_name=item.get("author_name"),
+            )
+
+            try:
+                download_video(
+                    requests_module=requests_module,
+                    headers=headers,
+                    video_url=item["video_url"],
+                    output_path=output_path,
+                )
+            except Exception as exc:
+                print(f"[WARN] 下载失败 {item['video_id']}: {exc}")
+                continue
+
+            downloaded += 1
+            new_items_in_batch += 1
+            print(f"[OK] 已保存: {output_path}")
+
+        if new_items_in_batch == 0:
+            consecutive_empty += 1
+            if consecutive_empty >= max_consecutive_empty:
+                break
+
+        human_like_scroll(page, settings=settings)
+
+    return downloaded
+
+
+def collect_search_results(
+    keyword: str,
+    max_videos: int,
+    timeout: int,
+    output_dir: Path,
+    browser_port: int | None,
+    scroll_settings: ScrollSettings | None = None,
+) -> int:
+    requests_module, chromium_page_cls, chromium_options_cls = import_runtime_dependencies()
+    search_url = build_search_page_url(keyword)
+    headers = build_headers(search_url)
+    if browser_port is not None:
+        ensure_browser_debug_port_ready(browser_port)
+    page = create_page(chromium_page_cls, chromium_options_cls, browser_port)
+    page.listen.start(SEARCH_LISTEN_TARGET)
+
+    print(f"[INFO] 正在打开抖音搜索页：{keyword}。若出现登录或验证码，请先在浏览器窗口里完成。")
+    page.get(search_url)
+    time.sleep(3)
+
+    downloaded = 0
+    seen_ids: set[str] = set()
+    consecutive_empty = 0
+    max_consecutive_empty = 6
+    settings = scroll_settings or ScrollSettings()
+    started_at = time.monotonic()
+
+    while downloaded < max_videos:
+        if settings.max_runtime > 0 and time.monotonic() - started_at >= settings.max_runtime:
+            print("[INFO] 已达到最大运行时间，结束抓取。")
+            break
+
+        packet = wait_for_aweme_packet(page, timeout=timeout)
+        if packet is None:
+            consecutive_empty += 1
+            if consecutive_empty >= max_consecutive_empty:
+                print("[INFO] 连续多次未获取到新搜索数据，结束抓取。")
+                break
+            human_like_scroll(page, settings=settings)
+            continue
+
+        try:
+            payload = extract_aweme_payload(packet.response)
+            items = parse_search_items(payload)
+        except Exception as exc:
+            print(f"[WARN] 解析搜索接口数据失败: {exc}")
+            consecutive_empty += 1
+            if consecutive_empty >= max_consecutive_empty:
+                break
+            human_like_scroll(page, settings=settings)
+            continue
+
+        if not items:
+            consecutive_empty += 1
+            if consecutive_empty >= max_consecutive_empty:
+                break
+            human_like_scroll(page, settings=settings)
+            continue
+
+        consecutive_empty = 0
+        new_items_in_batch = 0
+
+        for item in items:
+            if item["video_id"] in seen_ids:
+                continue
+
+            if downloaded >= max_videos:
+                break
+
+            seen_ids.add(item["video_id"])
+            output_path = build_output_path(
+                title=item["title"],
+                video_id=item["video_id"],
+                output_dir=output_dir,
+                author_name=item.get("author_name"),
+            )
+
+            try:
+                download_video(
+                    requests_module=requests_module,
+                    headers=headers,
+                    video_url=item["video_url"],
+                    output_path=output_path,
+                )
+            except Exception as exc:
+                print(f"[WARN] 下载失败 {item['video_id']}: {exc}")
+                continue
+
+            downloaded += 1
+            new_items_in_batch += 1
+            print(f"[OK] 已保存: {output_path}")
+
+        if new_items_in_batch == 0:
+            consecutive_empty += 1
+            if consecutive_empty >= max_consecutive_empty:
+                break
+
+        human_like_scroll(page, settings=settings)
+
+    return downloaded
+
+
 def collect_single_video(
     target: ResolvedTarget,
     timeout: int,
@@ -430,6 +853,47 @@ def build_parser() -> argparse.ArgumentParser:
         default=DEFAULT_BROWSER_PORT,
         help="附着到已启动 Chrome 的调试端口，默认 9223",
     )
+    parser.add_argument(
+        "--max-videos",
+        type=int,
+        default=50,
+        help="推荐流最大抓取数量，默认 50",
+    )
+    parser.add_argument(
+        "--search-keyword",
+        default=None,
+        help="搜索关键词；提供后抓取搜索结果页视频",
+    )
+    parser.add_argument(
+        "--scroll-mode",
+        choices=["human"],
+        default="human",
+        help="推荐流滚动模式，默认 human",
+    )
+    parser.add_argument(
+        "--min-wait",
+        type=float,
+        default=2.0,
+        help="推荐流每次滚动后的最短等待秒数，默认 2",
+    )
+    parser.add_argument(
+        "--max-wait",
+        type=float,
+        default=8.0,
+        help="推荐流每次滚动后的最长等待秒数，默认 8",
+    )
+    parser.add_argument(
+        "--reverse-scroll-probability",
+        type=float,
+        default=0.2,
+        help="推荐流小幅回滚概率，取值 0 到 1，默认 0.2",
+    )
+    parser.add_argument(
+        "--max-runtime",
+        type=float,
+        default=600.0,
+        help="推荐流最大运行秒数，默认 600；设置为 0 表示不限制",
+    )
     return parser
 
 
@@ -443,8 +907,36 @@ def main(argv: list[str] | None = None) -> int:
         parser.error("--timeout 必须大于 0")
     if args.browser_port is not None and args.browser_port <= 0:
         parser.error("--browser-port 必须大于 0")
+    if args.max_videos <= 0:
+        parser.error("--max-videos 必须大于 0")
+    if args.min_wait < 0:
+        parser.error("--min-wait 不能小于 0")
+    if args.max_wait < args.min_wait:
+        parser.error("--max-wait 必须大于或等于 --min-wait")
+    if not 0 <= args.reverse_scroll_probability <= 1:
+        parser.error("--reverse-scroll-probability 必须在 0 到 1 之间")
+    if args.max_runtime < 0:
+        parser.error("--max-runtime 不能小于 0")
+
+    scroll_settings = ScrollSettings(
+        mode=args.scroll_mode,
+        min_wait=args.min_wait,
+        max_wait=args.max_wait,
+        reverse_scroll_probability=args.reverse_scroll_probability,
+        max_runtime=args.max_runtime,
+    )
 
     try:
+        if args.search_keyword:
+            total = collect_search_results(
+                keyword=args.search_keyword,
+                max_videos=args.max_videos,
+                timeout=args.timeout,
+                output_dir=Path(args.output_dir),
+                browser_port=args.browser_port,
+                scroll_settings=scroll_settings,
+            )
+        else:
             target = resolve_cli_target(args.target, browser_port=args.browser_port)
             if target.kind == "creator":
                 total = collect_videos(
@@ -455,6 +947,14 @@ def main(argv: list[str] | None = None) -> int:
                     browser_port=args.browser_port,
                     auto_scroll=args.pages > 1,
                 )
+            elif target.kind == "recommendation":
+                total = collect_recommendations(
+                    max_videos=args.max_videos,
+                    timeout=args.timeout,
+                    output_dir=Path(args.output_dir),
+                    browser_port=args.browser_port,
+                    scroll_settings=scroll_settings,
+                )
             elif target.kind == "single-video":
                 total = collect_single_video(
                     target=target,

README.md

@@ -1,97 +1,236 @@
-# 抖音视频爬取工具
+# 抖音视频下载工具（中学生也能看懂版）
 
-这是一个面向 macOS 的抖音视频下载项目。
+## 这个工具能做什么？
 
-它当前采用“两步式”方式工作：
+帮你下载抖音上的视频！支持两种方式：
 
-1. 先启动一个可见的 Chrome 浏览器，让你手动登录抖音并完成验证码
-2. 再让脚本附着到这个浏览器，抓取博主主页当前已加载的作品视频并下载到本地
+1. **下载推荐页视频** - 就像你打开抖音看到的首页视频流
+2. **下载某个博主的主页视频** - 下载你喜欢的博主发布的视频
 
-这个项目已经完成过真实验证：在本机登录成功后，可以正常下载视频到 `video/` 目录。
+## 你需要准备什么？
 
-## 适合谁使用
+- 一台 Mac 电脑
+- 已经下载了这个项目到本地
+- 一点点耐心（需要手动登录抖音）
 
-适合以下用户：
+## 重要提醒
 
-- 使用 Mac
-- 项目已经在本地
-- 想快速下载某个抖音博主主页当前可见的作品视频
+- 不能自动登录抖音（需要你手动扫码或输入密码）
+- 不能自动过验证码（遇到验证码需要你自己点）
+- 只能下载当前页面上已经显示的视频（不会自动翻页下载全部历史视频）
 
-## 当前能做什么
+---
 
-- 启动一个带调试端口的 Chrome 浏览器
-- 手动登录抖音后附着到浏览器
-- 自动识别当前浏览器页面是博主主页还是单视频页
-- 抓取某个博主主页当前已加载的作品
-- 下载当前单视频页对应的那一条视频
-- 下载视频到本地 `video/` 目录
-- 支持传入指定博主主页 URL、单视频 URL 或 `aweme_id`
+## 第一次使用（安装环境）
 
-## 当前不能做什么
-
-- 不能自动帮你登录抖音
-- 不能自动替你过验证码
-- 不能默认抓完整个博主的全部历史作品
-- 不能抓任意网页
-- 不能自动筛选你想要的视频
-
-## 快速开始
-
-如果你已经把项目下载到本地，最快的使用方式是：
+打开终端（Terminal），依次输入以下命令：
 
 ```bash
+# 1. 进入项目文件夹
 cd /你的项目目录/douyin-crawler-poc
+
+# 2. 创建虚拟环境（隔离项目依赖）
 python3 -m venv .venv
+
+# 3. 激活虚拟环境
 source .venv/bin/activate
+
+# 4. 安装需要的库
 pip install requests DrissionPage
-./.venv/bin/python login_douyin.py
-./.venv/bin/python Douyin.py
 ```
 
-说明：
+**什么是虚拟环境？** 就像给这个项目建了一个独立的房间，里面放的工具不会影响电脑其他地方。
 
-- 第一个命令用于创建虚拟环境
-- 第二个命令用于进入虚拟环境
-- 第三个命令用于安装依赖
-- 第四个命令会打开 Chrome，让你登录抖音
-- 第五个命令会读取你当前浏览器页面并自动开始抓取或下载
+---
 
-如果自动判断失败，也可以手动传入一个目标：
+## 使用方法一：下载推荐页视频（首页视频流）
+
+### 步骤 1：启动浏览器并登录抖音
 
 ```bash
-./.venv/bin/python Douyin.py "https://www.douyin.com/user/你的博主主页"
-./.venv/bin/python Douyin.py "https://www.douyin.com/video/某个视频ID"
+./.venv/bin/python login_douyin.py
+```
+
+运行后会发生什么：
+- 会自动打开 Chrome 浏览器
+- 浏览器会显示抖音登录页面
+- **你需要手动登录**（扫码或输入账号密码）
+- 如果遇到验证码，也需要手动完成
+
+### 步骤 2：下载视频
+
+登录成功后，在终端输入：
+
+```bash
+# 下载默认数量（50个视频）
+./.venv/bin/python Douyin.py
+
+# 或者只下载10个视频
+./.venv/bin/python Douyin.py --max-videos 10
+
+# 或者只下载3个视频（测试用）
+./.venv/bin/python Douyin.py --max-videos 3
+```
+
+**注意：** 下载前请确保浏览器显示的是抖音推荐页（就是打开抖音看到的第一个页面，有很多视频往下滚动的那种）。
+
+---
+
+## 使用方法二：下载某个博主的主页视频
+
+### 步骤 1：同样先启动浏览器并登录
+
+```bash
+./.venv/bin/python login_douyin.py
+```
+
+### 步骤 2：进入博主主页
+
+在浏览器中：
+1. 搜索你想下载的博主（比如某个美食博主）
+2. 点击进入他的主页
+3. 等待页面加载完成（看到博主的视频列表）
+
+### 步骤 3：下载视频
+
+在终端输入：
+
+```bash
+# 下载当前页面显示的视频（默认只下载当前页）
+./.venv/bin/python Douyin.py
+
+# 或者下载多页（自动向下滚动加载）
+./.venv/bin/python Douyin.py --pages 3
+```
+
+**注意：** `--pages 3` 表示滚动加载3页，但抖音可能会限制，不一定能下载到那么多。
+
+---
+
+## 方法三：直接下载单个视频
+
+如果你只想下载某一个具体的视频：
+
+```bash
+# 方法 A：通过视频链接下载
+./.venv/bin/python Douyin.py "https://www.douyin.com/video/视频ID"
+
+# 方法 B：直接通过视频ID下载
 ./.venv/bin/python Douyin.py "7619989983668240802"
 ```
 
-## 下载结果在哪里
+---
 
-抓取成功后，视频会保存到项目根目录下的 `video/` 文件夹。
+## 下载的视频在哪里？
 
-文件名格式一般是：
+所有下载的视频都会保存在项目文件夹里的 `video/` 文件夹中。
 
-```text
-视频标题-aweme_id.mp4
+文件名格式：
+
+```
+[博主昵称]视频标题-视频ID.mp4
 ```
 
-## 详细图文说明
+例如：
+```
+[小张一人食]花2XXX在汕头喝白粥...-7633717884061725297.mp4
+[相声老司机]春晚不好看？...-7606185972144901412.mp4
+```
 
-详细操作步骤请看这份手册：
+---
+
+## 常见问题
+
+### Q1：运行时报错 "缺少 requests" 或 "缺少 DrissionPage"
+
+**解决：** 没有安装依赖，运行：
+```bash
+pip install requests DrissionPage
+```
+
+### Q2：提示 "当前页面不是受支持的抖音页面"
+
+**解决：** 
+- 如果要用推荐页：确保浏览器显示的是抖音首页（有视频流往下滚动的页面）
+- 如果要用博主页：确保你在某个博主的主页
+
+### Q3：下载了0个视频
+
+**解决：**
+- 检查是否已登录抖音
+- 检查当前页面是否有视频显示
+- 等待页面完全加载后再运行下载命令
+
+### Q4：Chrome 浏览器没打开
+
+**解决：**
+- 确保你用的是 Mac
+- 确保已安装 Google Chrome
+- 检查是否有其他 Chrome 正在运行，先关闭再试
+
+### Q5：我想下载更多视频怎么办？
+
+**推荐页：**
+```bash
+./.venv/bin/python Douyin.py --max-videos 100
+```
+
+**博主页：**
+```bash
+./.venv/bin/python Douyin.py --pages 5
+```
+
+---
+
+## 所有可用命令汇总
+
+```bash
+# 查看所有参数说明
+./.venv/bin/python Douyin.py --help
+
+# 推荐页下载（默认50个）
+./.venv/bin/python Douyin.py
+
+# 推荐页下载（指定数量）
+./.venv/bin/python Douyin.py --max-videos 20
+
+# 博主页下载（默认当前页）
+./.venv/bin/python Douyin.py
+
+# 博主页下载（多页）
+./.venv/bin/python Douyin.py --pages 3
+
+# 单视频下载
+./.venv/bin/python Douyin.py "https://www.douyin.com/video/xxx"
+
+# 修改等待时间（如果网络慢）
+./.venv/bin/python Douyin.py --timeout 20
+
+# 修改保存位置
+./.venv/bin/python Douyin.py --output-dir 我的视频
+```
+
+---
+
+## 详细图文教程
+
+如果你还是不太明白，可以查看这份更详细的图文教程：
 
 [小白图文操作手册](/Users/wangshaoqing/Desktop/MiaoSi/Study/douyin-crawler-poc/externaldocs/beginner-guide.md)
 
-如果你完全不会代码，建议直接从这份手册开始照着做。
+---
 
-## 相关文档
+## 技术说明（给感兴趣的同学）
 
-- [当前抓取能力需求说明](/Users/wangshaoqing/Desktop/MiaoSi/Study/douyin-crawler-poc/externaldocs/2026-04-17-readme-and-beginner-guide-requirements.md)
-- [后续定向抓取需求说明](/Users/wangshaoqing/Desktop/MiaoSi/Study/douyin-crawler-poc/externaldocs/2026-04-17-douyin-targeted-crawling-requirements.md)
+这个工具使用了两步式工作流：
 
-## 当前验证状态
+1. **login_douyin.py** - 启动带调试端口的 Chrome，让你手动登录
+2. **Douyin.py** - 附着到已登录的浏览器，监听抖音的 API 请求，提取视频地址并下载
 
-当前项目已验证：
+为什么需要手动登录？因为抖音有反爬虫机制，自动登录容易被封。
 
-- 单元测试通过
-- 登录浏览器入口可用
-- 抖音抓取脚本可附着到浏览器
-- 成功下载出 mp4 文件
+---
+
+## 免责声明
+
+本工具仅供学习交流使用，请勿用于商业用途或侵犯他人权益。下载的视频版权归原博主所有。

docs/superpowers/plans/2026-05-06-douyin-recommendation-crawling.md

@@ -0,0 +1,584 @@
+# 抖音推荐流视频抓取实现计划
+
+> **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:** 扩展 Douyin.py 支持抓取抖音推荐流（For You页面）视频，记录博主信息，支持滚动加载最多50条
+
+**Architecture:** 新建 `collect_recommendations()` 函数处理推荐流，复用现有的下载和工具函数。通过 `parse_target_input()` 扩展识别推荐流URL。
+
+**Tech Stack:** Python 3, DrissionPage, requests, unittest
+
+---
+
+## 文件结构
+
+| 文件 | 操作 | 说明 |
+|------|------|------|
+| `Douyin.py` | 修改 | 添加推荐流识别、解析、抓取逻辑 |
+| `test_douyin.py` | 修改 | 添加推荐流相关测试 |
+
+---
+
+## Task 1: 推荐流URL识别
+
+**Files:**
+- Modify: `Douyin.py:17-19`（添加正则表达式）
+- Test: `test_douyin.py`（添加测试）
+
+- [ ] **Step 1: 编写失败测试**
+
+```python
+def test_is_recommendation_url_accepts_douyin_homepage(self) -> None:
+    module = importlib.import_module("Douyin")
+    self.assertTrue(module.is_recommendation_url("https://www.douyin.com/"))
+    self.assertTrue(module.is_recommendation_url("https://www.douyin.com"))
+    self.assertTrue(module.is_recommendation_url("https://www.douyin.com/?from=web"))
+    self.assertFalse(module.is_recommendation_url("https://www.douyin.com/user/xxx"))
+    self.assertFalse(module.is_recommendation_url("https://www.douyin.com/video/123"))
+```
+
+- [ ] **Step 2: 运行测试确认失败**
+
+Run: `python3 -m pytest test_douyin.py::DouyinModuleTests::test_is_recommendation_url_accepts_douyin_homepage -v`
+Expected: FAIL with "module has no attribute 'is_recommendation_url'"
+
+- [ ] **Step 3: 实现最小代码**
+
+在 `Douyin.py` 中添加：
+```python
+RECOMMENDATION_URL_PATTERN = re.compile(r"^https?://www\.douyin\.com/?(?:\?.*)?$")
+
+def is_recommendation_url(value: str) -> bool:
+    return bool(RECOMMENDATION_URL_PATTERN.match(value.strip()))
+```
+
+- [ ] **Step 4: 运行测试确认通过**
+
+Run: `python3 -m pytest test_douyin.py::DouyinModuleTests::test_is_recommendation_url_accepts_douyin_homepage -v`
+Expected: PASS
+
+- [ ] **Step 5: 提交**
+
+```bash
+git add Douyin.py test_douyin.py
+git commit -m "feat: add recommendation URL recognition"
+```
+
+---
+
+## Task 2: 扩展目标解析支持推荐流
+
+**Files:**
+- Modify: `Douyin.py:52-68`（修改 `parse_target_input`）
+- Test: `test_douyin.py`（添加测试）
+
+- [ ] **Step 1: 编写失败测试**
+
+```python
+def test_parse_target_input_classifies_recommendation_url(self) -> None:
+    module = importlib.import_module("Douyin")
+    target = module.parse_target_input("https://www.douyin.com/", source="manual")
+    self.assertEqual(target.kind, "recommendation")
+    self.assertEqual(target.value, "https://www.douyin.com/")
+    self.assertEqual(target.source, "manual")
+```
+
+- [ ] **Step 2: 运行测试确认失败**
+
+Run: `python3 -m pytest test_douyin.py::DouyinModuleTests::test_parse_target_input_classifies_recommendation_url -v`
+Expected: FAIL with "不支持的目标"
+
+- [ ] **Step 3: 修改 `parse_target_input`**
+
+```python
+def parse_target_input(value: str, source: str) -> ResolvedTarget:
+    normalized = value.strip()
+    if is_recommendation_url(normalized):
+        return ResolvedTarget(kind="recommendation", value=normalized, source=source)
+    if is_creator_url(normalized):
+        return ResolvedTarget(kind="creator", value=normalized, source=source)
+    # ... 其余保持不变
+```
+
+- [ ] **Step 4: 运行测试确认通过**
+
+Run: `python3 -m pytest test_douyin.py::DouyinModuleTests::test_parse_target_input_classifies_recommendation_url -v`
+Expected: PASS
+
+- [ ] **Step 5: 提交**
+
+```bash
+git add Douyin.py test_douyin.py
+git commit -m "feat: extend target parsing to support recommendation URLs"
+```
+
+---
+
+## Task 3: 增强数据解析提取博主信息
+
+**Files:**
+- Modify: `Douyin.py:140-170`（修改 `parse_aweme_items`）
+- Test: `test_douyin.py`（添加测试）
+
+- [ ] **Step 1: 编写失败测试**
+
+```python
+def test_parse_aweme_items_extracts_author_info(self) -> None:
+    module = importlib.import_module("Douyin")
+    payload = {
+        "aweme_list": [
+            {
+                "aweme_id": "7619989983668240802",
+                "desc": "测试视频",
+                "author": {
+                    "nickname": "测试博主",
+                    "uid": "123456789"
+                },
+                "video": {
+                    "play_addr": {
+                        "url_list": ["https://v26-web.douyinvod.com/example/video.mp4"]
+                    }
+                },
+            }
+        ]
+    }
+    items = module.parse_aweme_items(payload)
+    self.assertEqual(len(items), 1)
+    self.assertEqual(items[0]["author_name"], "测试博主")
+    self.assertEqual(items[0]["author_id"], "123456789")
+```
+
+- [ ] **Step 2: 运行测试确认失败**
+
+Run: `python3 -m pytest test_douyin.py::DouyinModuleTests::test_parse_aweme_items_extracts_author_info -v`
+Expected: FAIL with KeyError or missing author_name
+
+- [ ] **Step 3: 修改 `parse_aweme_items`**
+
+```python
+def parse_aweme_items(body: Any) -> list[dict[str, str]]:
+    # ... 现有代码 ...
+    
+    for aweme in aweme_list:
+        # ... 现有视频提取代码 ...
+        
+        author = aweme.get("author") or {}
+        author_name = str(author.get("nickname") or "").strip() or "unknown"
+        author_id = str(author.get("uid") or "").strip() or "unknown"
+        
+        items.append(
+            {
+                "title": title,
+                "video_id": video_id,
+                "video_url": choose_video_url([str(url) for url in url_list]),
+                "author_name": author_name,
+                "author_id": author_id,
+            }
+        )
+    
+    return items
+```
+
+- [ ] **Step 4: 运行测试确认通过**
+
+Run: `python3 -m pytest test_douyin.py::DouyinModuleTests::test_parse_aweme_items_extracts_author_info -v`
+Expected: PASS
+
+- [ ] **Step 5: 提交**
+
+```bash
+git add Douyin.py test_douyin.py
+git commit -m "feat: extract author info from aweme items"
+```
+
+---
+
+## Task 4: 支持带博主信息的文件名构建
+
+**Files:**
+- Modify: `Douyin.py:102-104`（修改 `build_output_path`）
+- Test: `test_douyin.py`（添加测试）
+
+- [ ] **Step 1: 编写失败测试**
+
+```python
+def test_build_output_path_with_author_uses_bracket_format(self) -> None:
+    module = importlib.import_module("Douyin")
+    output_path = module.build_output_path(
+        title="测试标题", 
+        video_id="123456",
+        author_name="测试博主"
+    )
+    self.assertEqual(output_path.as_posix(), "video/[测试博主]测试标题-123456.mp4")
+
+def test_build_output_path_without_author_uses_original_format(self) -> None:
+    module = importlib.import_module("Douyin")
+    output_path = module.build_output_path("测试标题", "123456")
+    self.assertEqual(output_path.as_posix(), "video/测试标题-123456.mp4")
+```
+
+- [ ] **Step 2: 运行测试确认失败**
+
+Run: `python3 -m pytest test_douyin.py::DouyinModuleTests::test_build_output_path_with_author_uses_bracket_format -v`
+Expected: FAIL with unexpected keyword argument 'author_name'
+
+- [ ] **Step 3: 修改 `build_output_path`**
+
+```python
+def build_output_path(
+    title: str, 
+    video_id: str, 
+    output_dir: Path = Path("video"),
+    author_name: str | None = None,
+) -> Path:
+    safe_title = sanitize_filename(title, fallback="untitled")
+    if author_name:
+        safe_author = sanitize_filename(author_name, fallback="unknown")
+        filename = f"[{safe_author}]{safe_title}-{video_id}.mp4"
+    else:
+        filename = f"{safe_title}-{video_id}.mp4"
+    return output_dir / filename
+```
+
+- [ ] **Step 4: 运行测试确认通过**
+
+Run: `python3 -m pytest test_douyin.py::DouyinModuleTests::test_build_output_path_with_author_uses_bracket_format test_douyin.py::DouyinModuleTests::test_build_output_path_without_author_uses_original_format -v`
+Expected: PASS
+
+- [ ] **Step 5: 提交**
+
+```bash
+git add Douyin.py test_douyin.py
+git commit -m "feat: support author prefix in output filename"
+```
+
+---
+
+## Task 5: 实现 `collect_recommendations()` 函数
+
+**Files:**
+- Modify: `Douyin.py`（添加新函数）
+- Test: `test_douyin.py`（添加测试）
+
+- [ ] **Step 1: 编写失败测试**
+
+```python
+def test_collect_recommendations_downloads_videos_with_author_prefix(self) -> None:
+    module = importlib.import_module("Douyin")
+    packet = FakePacket(
+        {
+            "aweme_list": [
+                {
+                    "aweme_id": "7619989983668240802",
+                    "desc": "推荐视频1",
+                    "author": {"nickname": "博主A", "uid": "111"},
+                    "video": {
+                        "play_addr": {
+                            "url_list": ["https://v26-web.douyinvod.com/example/video1.mp4"]
+                        }
+                    },
+                }
+            ]
+        }
+    )
+    page = FakeRuntimePage("https://www.douyin.com/", packet)
+    
+    with mock.patch.object(module, "import_runtime_dependencies", return_value=(object(), object(), object())):
+        with mock.patch.object(module, "create_page", return_value=page):
+            with mock.patch.object(module, "download_video") as mocked_download:
+                downloaded = module.collect_recommendations(
+                    max_videos=50,
+                    timeout=10,
+                    output_dir=module.Path("video"),
+                    browser_port=None,
+                )
+    
+    self.assertEqual(downloaded, 1)
+    # 验证文件名包含博主前缀
+    call_kwargs = mocked_download.call_args[1]
+    self.assertIn("[博主A]", str(call_kwargs["output_path"]))
+```
+
+- [ ] **Step 2: 运行测试确认失败**
+
+Run: `python3 -m pytest test_douyin.py::DouyinModuleTests::test_collect_recommendations_downloads_videos_with_author_prefix -v`
+Expected: FAIL with "module has no attribute 'collect_recommendations'"
+
+- [ ] **Step 3: 实现 `collect_recommendations`**
+
+```python
+def collect_recommendations(
+    max_videos: int,
+    timeout: int,
+    output_dir: Path,
+    browser_port: int | None,
+) -> int:
+    requests_module, chromium_page_cls, chromium_options_cls = import_runtime_dependencies()
+    headers = build_headers("https://www.douyin.com/")
+    if browser_port is not None:
+        ensure_browser_debug_port_ready(browser_port)
+    page = create_page(chromium_page_cls, chromium_options_cls, browser_port)
+    page.listen.start(LISTEN_TARGET)
+
+    print("[INFO] 正在打开抖音推荐流。若出现登录或验证码，请先在浏览器窗口里完成。")
+    page.get("https://www.douyin.com/")
+    time.sleep(3)
+
+    downloaded = 0
+    seen_ids: set[str] = set()
+    consecutive_empty = 0
+    max_consecutive_empty = 3
+
+    while downloaded < max_videos:
+        packet = wait_for_aweme_packet(page, timeout=timeout)
+        if packet is None:
+            consecutive_empty += 1
+            if consecutive_empty >= max_consecutive_empty:
+                print("[INFO] 连续多次未获取到新数据，结束抓取。")
+                break
+            scroll_to_next_page(page)
+            continue
+
+        try:
+            payload = extract_aweme_payload(packet.response)
+            items = parse_aweme_items(payload)
+        except Exception as exc:
+            print(f"[WARN] 解析接口数据失败: {exc}")
+            consecutive_empty += 1
+            if consecutive_empty >= max_consecutive_empty:
+                break
+            scroll_to_next_page(page)
+            continue
+
+        if not items:
+            consecutive_empty += 1
+            if consecutive_empty >= max_consecutive_empty:
+                break
+            scroll_to_next_page(page)
+            continue
+
+        consecutive_empty = 0
+        new_items_in_batch = 0
+        
+        for item in items:
+            if item["video_id"] in seen_ids:
+                continue
+            
+            if downloaded >= max_videos:
+                break
+
+            seen_ids.add(item["video_id"])
+            output_path = build_output_path(
+                title=item["title"],
+                video_id=item["video_id"],
+                output_dir=output_dir,
+                author_name=item.get("author_name"),
+            )
+
+            try:
+                download_video(
+                    requests_module=requests_module,
+                    headers=headers,
+                    video_url=item["video_url"],
+                    output_path=output_path,
+                )
+            except Exception as exc:
+                print(f"[WARN] 下载失败 {item['video_id']}: {exc}")
+                continue
+
+            downloaded += 1
+            new_items_in_batch += 1
+            print(f"[OK] 已保存: {output_path}")
+
+        if new_items_in_batch == 0:
+            consecutive_empty += 1
+            if consecutive_empty >= max_consecutive_empty:
+                break
+
+        scroll_to_next_page(page)
+
+    return downloaded
+```
+
+- [ ] **Step 4: 运行测试确认通过**
+
+Run: `python3 -m pytest test_douyin.py::DouyinModuleTests::test_collect_recommendations_downloads_videos_with_author_prefix -v`
+Expected: PASS
+
+- [ ] **Step 5: 提交**
+
+```bash
+git add Douyin.py test_douyin.py
+git commit -m "feat: implement collect_recommendations() for For You page"
+```
+
+---
+
+## Task 6: 添加 `--max-videos` 命令行参数
+
+**Files:**
+- Modify: `Douyin.py:295-305`（修改 `build_parser`）
+- Modify: `Douyin.py:310-350`（修改 `main`）
+- Test: `test_douyin.py`（添加测试）
+
+- [ ] **Step 1: 编写失败测试**
+
+```python
+def test_build_parser_has_max_videos_argument(self) -> None:
+    module = importlib.import_module("Douyin")
+    args = module.build_parser().parse_args(["--max-videos", "30"])
+    self.assertEqual(args.max_videos, 30)
+
+def test_main_dispatches_recommendation_flow_for_recommendation_url(self) -> None:
+    module = importlib.import_module("Douyin")
+    stdout = io.StringIO()
+    recommendation_target = module.ResolvedTarget(
+        kind="recommendation",
+        value="https://www.douyin.com/",
+        source="current-page",
+    )
+    with redirect_stdout(stdout):
+        with mock.patch.object(module, "resolve_cli_target", return_value=recommendation_target):
+            with mock.patch.object(module, "collect_recommendations", return_value=5) as mocked_collect:
+                exit_code = module.main([])
+    self.assertEqual(exit_code, 0)
+    mocked_collect.assert_called_once_with(
+        max_videos=50,
+        timeout=10,
+        output_dir=module.Path("video"),
+        browser_port=9223,
+    )
+```
+
+- [ ] **Step 2: 运行测试确认失败**
+
+Run: `python3 -m pytest test_douyin.py::DouyinModuleTests::test_build_parser_has_max_videos_argument -v`
+Expected: FAIL with "unrecognized arguments: --max-videos"
+
+- [ ] **Step 3: 修改 `build_parser` 和 `main`**
+
+```python
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description="附着抖音登录浏览器并下载当前页面或指定目标的视频")
+    parser.add_argument(
+        "target",
+        nargs="?",
+        default=None,
+        help="可选：博主主页 URL、单视频 URL 或 aweme_id；不传则读取当前浏览器页面",
+    )
+    parser.add_argument("--pages", type=int, default=1, help="创作者抓取最多处理多少页；默认 1")
+    parser.add_argument("--timeout", type=int, default=10, help="单次等待接口响应秒数，默认 10")
+    parser.add_argument(
+        "--output-dir",
+        default="video",
+        help="视频输出目录，默认 video",
+    )
+    parser.add_argument(
+        "--browser-port",
+        type=int,
+        default=DEFAULT_BROWSER_PORT,
+        help="附着到已启动 Chrome 的调试端口，默认 9223",
+    )
+    parser.add_argument(
+        "--max-videos",
+        type=int,
+        default=50,
+        help="推荐流最大抓取数量，默认 50",
+    )
+    return parser
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = build_parser()
+    args = parser.parse_args(argv)
+
+    if args.pages <= 0:
+        parser.error("--pages 必须大于 0")
+    if args.timeout <= 0:
+        parser.error("--timeout 必须大于 0")
+    if args.browser_port is not None and args.browser_port <= 0:
+        parser.error("--browser-port 必须大于 0")
+    if args.max_videos <= 0:
+        parser.error("--max-videos 必须大于 0")
+
+    try:
+        target = resolve_cli_target(args.target, browser_port=args.browser_port)
+        if target.kind == "creator":
+            total = collect_videos(
+                user_url=target.value,
+                max_pages=args.pages,
+                timeout=args.timeout,
+                output_dir=Path(args.output_dir),
+                browser_port=args.browser_port,
+                auto_scroll=args.pages > 1,
+            )
+        elif target.kind == "recommendation":
+            total = collect_recommendations(
+                max_videos=args.max_videos,
+                timeout=args.timeout,
+                output_dir=Path(args.output_dir),
+                browser_port=args.browser_port,
+            )
+        elif target.kind == "single-video":
+            total = collect_single_video(
+                target=target,
+                timeout=args.timeout,
+                output_dir=Path(args.output_dir),
+                browser_port=args.browser_port,
+            )
+        else:
+            raise RuntimeError(f"不支持的目标类型: {target.kind}")
+    except RuntimeError as exc:
+        print(f"[ERROR] {exc}")
+        return 1
+    except KeyboardInterrupt:
+        print("\n[INFO] 用户中断。")
+        return 130
+
+    print(f"[INFO] 处理结束，共下载 {total} 个视频。")
+    return 0
+```
+
+- [ ] **Step 4: 运行测试确认通过**
+
+Run: `python3 -m pytest test_douyin.py::DouyinModuleTests::test_build_parser_has_max_videos_argument test_douyin.py::DouyinModuleTests::test_main_dispatches_recommendation_flow_for_recommendation_url -v`
+Expected: PASS
+
+- [ ] **Step 5: 提交**
+
+```bash
+git add Douyin.py test_douyin.py
+git commit -m "feat: add --max-videos argument and wire recommendation flow in main"
+```
+
+---
+
+## Task 7: 运行全部测试并验证
+
+- [ ] **Step 1: 运行全部测试**
+
+Run: `python3 -m pytest test_douyin.py -v`
+Expected: 所有测试通过
+
+- [ ] **Step 2: 运行主脚本帮助确认**
+
+Run: `python3 Douyin.py --help`
+Expected: 显示包含 `--max-videos` 的帮助信息
+
+- [ ] **Step 3: 提交**
+
+```bash
+git add -A
+git commit -m "test: verify all tests pass for recommendation crawling feature"
+```
+
+---
+
+## 完成标准
+
+1. ✅ `Douyin.py` 支持识别 `https://www.douyin.com/` 为推荐流目标
+2. ✅ `collect_recommendations()` 函数实现滚动加载、最多50条、去重
+3. ✅ 视频文件名包含博主昵称：`[博主名]标题-aweme_id.mp4`
+4. ✅ `--max-videos` 命令行参数可用
+5. ✅ 所有现有测试继续通过
+6. ✅ 新增测试覆盖推荐流功能

docs/superpowers/specs/2026-05-06-douyin-recommendation-crawling-design.md

@@ -0,0 +1,165 @@
+# 抖音推荐流视频抓取设计文档
+
+## 背景
+
+当前系统支持抓取指定博主主页或单视频。现需扩展支持抓取抖音推荐流（For You页面）当前可见的视频。
+
+## 目标
+
+- 支持抓取抖音推荐流（`https://www.douyin.com/`）的视频
+- 记录每个视频对应的博主信息
+- 支持滚动加载，最多抓取50条
+- 视频统一保存到 `video/` 目录
+- 保持现有两步式工作流不变
+
+## 方案选择
+
+采用**方案B：新建推荐流专用抓取函数**
+
+理由：
+- 逻辑清晰，推荐流和博主页完全解耦
+- 便于后续分别维护
+- 工作量适中，可快速实现验证
+
+## 详细设计
+
+### 1. 目标识别扩展
+
+新增推荐流URL识别模式：
+
+```python
+RECOMMENDATION_URL_PATTERN = re.compile(
+    r"^https?://www\.douyin\.com/?(?:\?.*)?$"
+)
+```
+
+目标类型扩展为三种：
+- `creator` - 博主主页（现有）
+- `single-video` - 单视频（现有）
+- `recommendation` - 推荐流（新增）
+
+### 2. 核心流程
+
+```
+用户执行 ./.venv/bin/python Douyin.py
+    ↓
+读取当前浏览器页面URL
+    ↓
+判断页面类型：
+    - 推荐流 → 执行 collect_recommendations()
+    - 博主页 → 执行 collect_videos()（现有）
+    - 单视频 → 执行 collect_single_video()（现有）
+    - 其他 → 报错提示
+```
+
+### 3. collect_recommendations() 函数
+
+**参数：**
+- `max_videos`: 最大抓取数量（默认50）
+- `timeout`: 单次等待接口响应秒数（默认10）
+- `output_dir`: 输出目录（默认 `video/`）
+- `browser_port`: 浏览器调试端口（默认9223）
+
+**行为：**
+1. 通过 `page.get("https://www.douyin.com/")` 打开推荐流页面（复用现有页面打开逻辑，不切换标签页）
+2. 启动监听，目标接口：`web/aweme/post/`（推荐流与博主页共用此接口）
+3. 循环直到收集够 `max_videos` 条或无法继续加载：
+   - 等待接口响应
+   - 解析视频列表，提取：标题、视频ID、视频URL、博主信息
+   - 过滤已下载（按 `video_id` 去重，使用 `seen_ids: set[str]` 集合）
+   - 下载视频
+   - 向下滚动加载更多
+4. 返回实际下载数量
+
+### 4. 数据解析增强
+
+新增博主信息提取字段：
+
+```python
+{
+    "title": "视频标题",
+    "video_id": "aweme_id",
+    "video_url": "下载链接",
+    "author_name": "博主昵称",
+    "author_id": "博主ID",
+}
+```
+
+### 5. 文件名格式
+
+**推荐流视频：**
+```
+[博主昵称]视频标题-aweme_id.mp4
+```
+示例：`[张三]搞笑视频-7619989983668240802.mp4`
+
+**博主页视频（保持现有）：**
+```
+视频标题-aweme_id.mp4
+```
+
+### 6. 命令行参数
+
+**新增参数：**
+```bash
+--max-videos 50    # 推荐流最大抓取数量（默认50）
+```
+
+**使用示例：**
+```bash
+# 零参数，自动识别当前页面
+./.venv/bin/python Douyin.py
+
+# 自定义抓取数量（仅对推荐流有效）
+./.venv/bin/python Douyin.py --max-videos 30
+
+# 显式传入URL时，--max-videos 不适用（博主页和单视频页忽略此参数）
+./.venv/bin/python Douyin.py "https://www.douyin.com/user/xxx"
+```
+
+### 7. 错误处理
+
+- **推荐流页面未加载数据**：提示用户先滚动加载内容
+- **滚动后无新数据**：正常结束，返回已下载数量
+- **达到最大数量**：正常结束
+- **其他错误**：复用现有错误处理机制
+
+### 8. 测试覆盖（TDD）
+
+必须覆盖以下测试场景：
+
+- [ ] 推荐流URL识别测试
+- [ ] 推荐流页面解析测试（模拟API响应含博主信息）
+- [ ] 滚动加载逻辑测试
+- [ ] 最大数量限制测试
+- [ ] 文件名构建测试（含博主名）
+- [ ] 博主信息提取测试
+- [ ] 去重逻辑测试
+
+## 非目标
+
+- 自动登录抖音
+- 自动过验证码
+- 抓取非推荐流页面（如话题页、搜索页）
+- 自动筛选视频内容
+- 抓取超过50条视频（如需更多，需用户手动调整参数）
+
+## 验收标准
+
+1. 用户可以在推荐流页面执行 `./.venv/bin/python Douyin.py` 抓取视频
+2. 系统能自动识别当前页面是推荐流
+3. 支持滚动加载，最多抓取50条
+4. 文件名包含博主昵称
+5. 所有测试通过
+6. 关键失败场景有明确报错
+
+## 实现步骤
+
+1. 编写测试（TDD）
+2. 实现推荐流URL识别
+3. 实现 `collect_recommendations()` 函数
+4. 增强数据解析（提取博主信息）
+5. 修改文件名构建逻辑
+6. 更新命令行参数
+7. 运行全部测试
+8. 提交代码

login_douyin.py

@@ -7,7 +7,8 @@ import sys
 import time
 from pathlib import Path
 
-from Douyin import DEFAULT_USER_URL
+DEFAULT_RECOMMENDATION_URL = "https://www.douyin.com/"
+DEFAULT_USER_URL = DEFAULT_RECOMMENDATION_URL
 
 DEFAULT_CHROME_PATH = "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome"
 DEFAULT_BROWSER_PORT = 9223
@@ -54,7 +55,7 @@ def build_parser() -> argparse.ArgumentParser:
         default=DEFAULT_BROWSER_PORT,
         help="Chrome 调试端口，默认 9223",
     )
-    parser.add_argument("--user-url", default=DEFAULT_USER_URL, help="启动后打开的抖音主页 URL")
+    parser.add_argument("--user-url", default=DEFAULT_RECOMMENDATION_URL, help="启动后打开的抖音页面 URL，默认推荐流首页")
     return parser
 
 

test_douyin.py

@@ -31,12 +31,16 @@ class FakeListener:
     def __init__(self, packet):
         self.packet = packet
         self.started_targets = []
+        self.call_count = 0
 
     def start(self, target):
         self.started_targets.append(target)
 
     def wait(self, timeout):
+        self.call_count += 1
+        if self.call_count == 1:
             return self.packet
+        return None
 
 
 class FakeRuntimePage:
@@ -50,9 +54,32 @@ class FakeRuntimePage:
         self.url = url
 
     def run_js(self, script):
+        # Allow both old scroll_to_next_page and new human_like_scroll
+        if "window.scrollTo" in script or "window.scrollBy" in script:
+            return
         raise AssertionError(f"unexpected scroll script: {script}")
 
 
+class FakeScrollPage:
+    def __init__(self):
+        self.scripts = []
+
+    def run_js(self, script):
+        self.scripts.append(script)
+
+
+class FakeContainerScrollPage:
+    def __init__(self, container_found=True):
+        self.container_found = container_found
+        self.scripts = []
+
+    def run_js(self, script):
+        self.scripts.append(script)
+        if "findMainScrollContainer" in script:
+            return self.container_found
+        return None
+
+
 class DouyinModuleTests(unittest.TestCase):
     def test_module_can_import_without_optional_runtime_dependencies(self) -> None:
         module = importlib.import_module("Douyin")
@@ -82,6 +109,25 @@ class DouyinModuleTests(unittest.TestCase):
         output_path = module.build_output_path("测试标题", "123456")
         self.assertEqual(output_path.as_posix(), "video/测试标题-123456.mp4")
 
+    def test_build_output_path_with_author_uses_bracket_format(self) -> None:
+        module = importlib.import_module("Douyin")
+        output_path = module.build_output_path(
+            title="测试标题", 
+            video_id="123456",
+            author_name="测试博主"
+        )
+        self.assertEqual(output_path.as_posix(), "video/[测试博主]测试标题-123456.mp4")
+
+    def test_build_output_path_limits_long_filename(self) -> None:
+        module = importlib.import_module("Douyin")
+        output_path = module.build_output_path(
+            title="超长标题" * 100,
+            video_id="7619989983668240802",
+            author_name="超长博主名" * 20,
+        )
+        self.assertLessEqual(len(output_path.name.encode("utf-8")), 240)
+        self.assertTrue(output_path.name.endswith("-7619989983668240802.mp4"))
+
     def test_extract_aweme_payload_uses_dict_body(self) -> None:
         module = importlib.import_module("Douyin")
         response = FakeResponse({"aweme_list": []}, "")
@@ -95,11 +141,79 @@ class DouyinModuleTests(unittest.TestCase):
             {"aweme_list": [{"aweme_id": "1"}]},
         )
 
+    def test_wait_for_aweme_packet_treats_false_listener_result_as_missing(self) -> None:
+        module = importlib.import_module("Douyin")
+        page = mock.MagicMock()
+        page.listen.wait.return_value = False
+        self.assertIsNone(module.wait_for_aweme_packet(page, timeout=10))
+
     def test_build_browser_address_from_port(self) -> None:
         module = importlib.import_module("Douyin")
         self.assertEqual(module.build_browser_address(9223), "127.0.0.1:9223")
         self.assertIsNone(module.build_browser_address(None))
 
+    def test_default_scroll_settings_uses_human_mode(self) -> None:
+        module = importlib.import_module("Douyin")
+        settings = module.ScrollSettings()
+        self.assertEqual(settings.mode, "human")
+        self.assertEqual(settings.min_wait, 2.0)
+        self.assertEqual(settings.max_wait, 8.0)
+        self.assertEqual(settings.reverse_scroll_probability, 0.2)
+
+    def test_create_human_scroll_plan_uses_configured_ranges(self) -> None:
+        module = importlib.import_module("Douyin")
+        settings = module.ScrollSettings(
+            min_wait=2.0,
+            max_wait=4.0,
+            min_scroll=300,
+            max_scroll=900,
+            reverse_scroll_probability=0.0,
+        )
+        plan = module.create_human_scroll_plan(settings, random_module=module.random.Random(7))
+        self.assertGreaterEqual(plan.down_distance, 300)
+        self.assertLessEqual(plan.down_distance, 900)
+        self.assertGreaterEqual(plan.down_wait, 2.0)
+        self.assertLessEqual(plan.down_wait, 4.0)
+        self.assertEqual(plan.reverse_distance, 0)
+
+    def test_create_human_scroll_plan_can_include_reverse_scroll(self) -> None:
+        module = importlib.import_module("Douyin")
+        settings = module.ScrollSettings(reverse_scroll_probability=1.0)
+        plan = module.create_human_scroll_plan(settings, random_module=module.random.Random(3))
+        self.assertGreaterEqual(plan.reverse_distance, 80)
+        self.assertLessEqual(plan.reverse_distance, 250)
+        self.assertGreater(plan.reverse_wait, 0)
+
+    def test_run_human_scroll_sequence_scrolls_down_and_optionally_back_up(self) -> None:
+        module = importlib.import_module("Douyin")
+        page = FakeScrollPage()
+        plan = module.HumanScrollPlan(
+            down_distance=500,
+            down_wait=2.5,
+            reverse_distance=120,
+            reverse_wait=1.0,
+            settle_wait=3.0,
+        )
+        with mock.patch.object(module.time, "sleep") as mocked_sleep:
+            module.run_human_scroll_sequence(page, plan)
+        self.assertIn("window.scrollBy(0, 500);", page.scripts)
+        self.assertIn("window.scrollBy(0, -120);", page.scripts)
+        self.assertIn("window.scrollBy(0, 240);", page.scripts)
+        mocked_sleep.assert_has_calls([mock.call(2.5), mock.call(1.0), mock.call(3.0)])
+
+    def test_run_scroll_step_prefers_main_scroll_container(self) -> None:
+        module = importlib.import_module("Douyin")
+        page = FakeContainerScrollPage(container_found=True)
+        self.assertTrue(module.run_scroll_step(page, 500))
+        self.assertIn("const distance = 500;", page.scripts[-1])
+        self.assertIn("scrollTarget.scrollBy(0, distance);", page.scripts[-1])
+
+    def test_run_scroll_step_falls_back_to_window_when_container_is_missing(self) -> None:
+        module = importlib.import_module("Douyin")
+        page = FakeContainerScrollPage(container_found=False)
+        self.assertFalse(module.run_scroll_step(page, 500))
+        self.assertEqual(page.scripts[-1], "window.scrollBy(0, 500);")
+
     def test_ensure_browser_debug_port_ready_accepts_open_port(self) -> None:
         module = importlib.import_module("Douyin")
         connection = mock.MagicMock()
@@ -115,6 +229,14 @@ class DouyinModuleTests(unittest.TestCase):
             with self.assertRaisesRegex(RuntimeError, "login_douyin.py"):
                 module.ensure_browser_debug_port_ready(9223)
 
+    def test_is_recommendation_url_accepts_douyin_homepage(self) -> None:
+        module = importlib.import_module("Douyin")
+        self.assertTrue(module.is_recommendation_url("https://www.douyin.com/"))
+        self.assertTrue(module.is_recommendation_url("https://www.douyin.com"))
+        self.assertTrue(module.is_recommendation_url("https://www.douyin.com/?from=web"))
+        self.assertFalse(module.is_recommendation_url("https://www.douyin.com/user/xxx"))
+        self.assertFalse(module.is_recommendation_url("https://www.douyin.com/video/123"))
+
     def test_is_creator_url_accepts_supported_douyin_creator_url(self) -> None:
         module = importlib.import_module("Douyin")
         self.assertTrue(
@@ -136,6 +258,13 @@ class DouyinModuleTests(unittest.TestCase):
         self.assertTrue(module.is_aweme_id("7619989983668240802"))
         self.assertFalse(module.is_aweme_id("not-an-aweme-id"))
 
+    def test_parse_target_input_classifies_recommendation_url(self) -> None:
+        module = importlib.import_module("Douyin")
+        target = module.parse_target_input("https://www.douyin.com/", source="manual")
+        self.assertEqual(target.kind, "recommendation")
+        self.assertEqual(target.value, "https://www.douyin.com/")
+        self.assertEqual(target.source, "manual")
+
     def test_parse_target_input_classifies_creator_url(self) -> None:
         module = importlib.import_module("Douyin")
         target = module.parse_target_input(
@@ -239,6 +368,73 @@ class DouyinModuleTests(unittest.TestCase):
                             browser_port=None,
                         )
 
+    def test_parse_aweme_items_extracts_author_info(self) -> None:
+        module = importlib.import_module("Douyin")
+        payload = {
+            "aweme_list": [
+                {
+                    "aweme_id": "7619989983668240802",
+                    "desc": "测试视频",
+                    "author": {
+                        "nickname": "测试博主",
+                        "uid": "123456789"
+                    },
+                    "video": {
+                        "play_addr": {
+                            "url_list": ["https://v26-web.douyinvod.com/example/video.mp4"]
+                        }
+                    },
+                }
+            ]
+        }
+        items = module.parse_aweme_items(payload)
+        self.assertEqual(len(items), 1)
+        self.assertEqual(items[0]["author_name"], "测试博主")
+        self.assertEqual(items[0]["author_id"], "123456789")
+
+    def test_parse_aweme_items_uses_play_addr_h264_when_play_addr_is_missing(self) -> None:
+        module = importlib.import_module("Douyin")
+        payload = {
+            "aweme_list": [
+                {
+                    "aweme_id": "7619989983668240802",
+                    "desc": "推荐视频",
+                    "video": {
+                        "play_addr_h264": {
+                            "url_list": ["https://v26-web.douyinvod.com/example/h264.mp4"]
+                        }
+                    },
+                }
+            ]
+        }
+        items = module.parse_aweme_items(payload)
+        self.assertEqual(len(items), 1)
+        self.assertEqual(items[0]["video_url"], "https://v26-web.douyinvod.com/example/h264.mp4")
+
+    def test_parse_aweme_items_uses_bit_rate_play_addr_when_top_level_addresses_are_missing(self) -> None:
+        module = importlib.import_module("Douyin")
+        payload = {
+            "aweme_list": [
+                {
+                    "aweme_id": "7619989983668240802",
+                    "desc": "推荐视频",
+                    "video": {
+                        "bit_rate": [
+                            {
+                                "format": "mp4",
+                                "play_addr": {
+                                    "url_list": ["https://v11-weba.douyinvod.com/example/bitrate.mp4"]
+                                },
+                            }
+                        ]
+                    },
+                }
+            ]
+        }
+        items = module.parse_aweme_items(payload)
+        self.assertEqual(len(items), 1)
+        self.assertEqual(items[0]["video_url"], "https://v11-weba.douyinvod.com/example/bitrate.mp4")
+
     def test_build_video_page_url_uses_aweme_id(self) -> None:
         module = importlib.import_module("Douyin")
         self.assertEqual(
@@ -246,6 +442,74 @@ class DouyinModuleTests(unittest.TestCase):
             "https://www.douyin.com/video/7619989983668240802",
         )
 
+    def test_build_search_page_url_encodes_keyword(self) -> None:
+        module = importlib.import_module("Douyin")
+        self.assertEqual(
+            module.build_search_page_url("猫咪"),
+            "https://www.douyin.com/search/%E7%8C%AB%E5%92%AA?type=general",
+        )
+
+    def test_parse_search_items_extracts_aweme_info(self) -> None:
+        module = importlib.import_module("Douyin")
+        payload = {
+            "data": [
+                {
+                    "type": 1,
+                    "aweme_info": {
+                        "aweme_id": "7319795133048769829",
+                        "desc": "猫咪视频",
+                        "author": {"nickname": "奶芙芙", "uid": "75478174642"},
+                        "video": {
+                            "play_addr_lowbr": {
+                                "url_list": ["https://v26-web.douyinvod.com/example/search.mp4"]
+                            }
+                        },
+                    },
+                }
+            ]
+        }
+        items = module.parse_search_items(payload)
+        self.assertEqual(len(items), 1)
+        self.assertEqual(items[0]["video_id"], "7319795133048769829")
+        self.assertEqual(items[0]["author_name"], "奶芙芙")
+        self.assertEqual(items[0]["video_url"], "https://v26-web.douyinvod.com/example/search.mp4")
+
+    def test_collect_recommendations_downloads_videos_with_author_prefix(self) -> None:
+        module = importlib.import_module("Douyin")
+        packet = FakePacket(
+            {
+                "aweme_list": [
+                    {
+                        "aweme_id": "7619989983668240802",
+                        "desc": "推荐视频1",
+                        "author": {"nickname": "博主A", "uid": "111"},
+                        "video": {
+                            "play_addr": {
+                                "url_list": ["https://v26-web.douyinvod.com/example/video1.mp4"]
+                            }
+                        },
+                    }
+                ]
+            }
+        )
+        page = FakeRuntimePage("https://www.douyin.com/", packet)
+        
+        with mock.patch.object(module, "import_runtime_dependencies", return_value=(object(), object(), object())):
+            with mock.patch.object(module, "create_page", return_value=page):
+                with mock.patch.object(module, "download_video") as mocked_download:
+                    with mock.patch.object(module, "human_like_scroll"):
+                        downloaded = module.collect_recommendations(
+                            max_videos=50,
+                            timeout=10,
+                            output_dir=module.Path("video"),
+                            browser_port=None,
+                        )
+        
+        self.assertEqual(downloaded, 1)
+        # 验证文件名包含博主前缀
+        call_kwargs = mocked_download.call_args[1]
+        self.assertIn("[博主A]", str(call_kwargs["output_path"]))
+
     def test_collect_single_video_downloads_exactly_one_file_for_video_url_target(self) -> None:
         module = importlib.import_module("Douyin")
         packet = FakePacket(
@@ -316,6 +580,54 @@ class DouyinModuleTests(unittest.TestCase):
         self.assertEqual(page.visited_urls, ["https://www.douyin.com/video/7619989983668240802"])
         mocked_download.assert_called_once()
 
+    def test_build_parser_has_max_videos_argument(self) -> None:
+        module = importlib.import_module("Douyin")
+        args = module.build_parser().parse_args(["--max-videos", "30"])
+        self.assertEqual(args.max_videos, 30)
+
+    def test_build_parser_has_human_scroll_arguments(self) -> None:
+        module = importlib.import_module("Douyin")
+        args = module.build_parser().parse_args(
+            [
+                "--scroll-mode",
+                "human",
+                "--min-wait",
+                "3",
+                "--max-wait",
+                "9",
+                "--reverse-scroll-probability",
+                "0.4",
+                "--max-runtime",
+                "600",
+            ]
+        )
+        self.assertEqual(args.scroll_mode, "human")
+        self.assertEqual(args.min_wait, 3)
+        self.assertEqual(args.max_wait, 9)
+        self.assertEqual(args.reverse_scroll_probability, 0.4)
+        self.assertEqual(args.max_runtime, 600)
+
+    def test_build_parser_has_search_keyword_argument(self) -> None:
+        module = importlib.import_module("Douyin")
+        args = module.build_parser().parse_args(["--search-keyword", "猫咪"])
+        self.assertEqual(args.search_keyword, "猫咪")
+
+    def test_main_dispatches_search_flow_for_search_keyword(self) -> None:
+        module = importlib.import_module("Douyin")
+        stdout = io.StringIO()
+        with redirect_stdout(stdout):
+            with mock.patch.object(module, "collect_search_results", return_value=7) as mocked_collect:
+                exit_code = module.main(["--search-keyword", "猫咪"])
+        self.assertEqual(exit_code, 0)
+        mocked_collect.assert_called_once_with(
+            keyword="猫咪",
+            max_videos=50,
+            timeout=10,
+            output_dir=module.Path("video"),
+            browser_port=9223,
+            scroll_settings=module.ScrollSettings(),
+        )
+
     def test_build_parser_defaults_to_zero_argument_current_page_flow(self) -> None:
         module = importlib.import_module("Douyin")
         args = module.build_parser().parse_args([])
@@ -331,6 +643,27 @@ class DouyinModuleTests(unittest.TestCase):
         self.assertEqual(target.aweme_id, "7619989983668240802")
         mocked_imports.assert_not_called()
 
+    def test_main_dispatches_recommendation_flow_for_recommendation_url(self) -> None:
+        module = importlib.import_module("Douyin")
+        stdout = io.StringIO()
+        recommendation_target = module.ResolvedTarget(
+            kind="recommendation",
+            value="https://www.douyin.com/",
+            source="current-page",
+        )
+        with redirect_stdout(stdout):
+            with mock.patch.object(module, "resolve_cli_target", return_value=recommendation_target):
+                with mock.patch.object(module, "collect_recommendations", return_value=5) as mocked_collect:
+                    exit_code = module.main([])
+        self.assertEqual(exit_code, 0)
+        mocked_collect.assert_called_once_with(
+            max_videos=50,
+            timeout=10,
+            output_dir=module.Path("video"),
+            browser_port=9223,
+            scroll_settings=module.ScrollSettings(),
+        )
+
     def test_main_without_target_dispatches_current_page_creator_flow(self) -> None:
         module = importlib.import_module("Douyin")
         stdout = io.StringIO()

test_learning_examples.py

@@ -299,6 +299,8 @@ class PlaywrightLearningHelperTests(unittest.TestCase):
                 "title": "Playwright 示例",
                 "video_id": "7619989983668240802",
                 "video_url": "https://v26-web.douyinvod.com/example/single.mp4",
+                "author_name": "unknown",
+                "author_id": "unknown",
             },
         )