acquire-lead/collectors/douyin/runner.py

"""编排入口：按步骤跑完整采集流程。"""

from __future__ import annotations

from playwright.sync_api import sync_playwright

from config import settings
from collectors.douyin.auth import click_login_if_needed, wait_until_logged_in
from collectors.douyin.comments import collect_comments_and_save
from collectors.douyin.common import SkipVideoError, human_pause, run_step
from collectors.douyin.navigate import go_to_next_video
from collectors.douyin.search import (
    click_video_filter,
    open_first_video,
    search_keyword,
)
from collectors.douyin.video import collect_and_save_video_meta


def run() -> None:
    settings.BROWSER_DATA_DIR.mkdir(parents=True, exist_ok=True)
    keyword = settings.SEARCH_KEYWORD
    limit = settings.COMMENT_LIMIT
    video_limit = settings.VIDEO_LIMIT
    max_attempts = max(video_limit * 3, video_limit + 5)

    with sync_playwright() as p:
        context = p.chromium.launch_persistent_context(
            user_data_dir=str(settings.BROWSER_DATA_DIR),
            headless=False,
            viewport={"width": 1280, "height": 800},
            locale="zh-CN",
            args=["--disable-blink-features=AutomationControlled"],
        )
        page = context.pages[0] if context.pages else context.new_page()
        page.goto(settings.DOUYIN_URL, wait_until="domcontentloaded")
        print(f"已打开: {page.url}")

        human_pause(2.0, 3.5)
        run_step("点击登录", lambda: click_login_if_needed(page))

        try:
            wait_until_logged_in(page)
        except Exception as exc:
            print(f"[等待登录] 超时或失败: {exc}")
            print("登录未完成，后续步骤将跳过；浏览器保持打开，可手动继续。")
        else:
            human_pause(2.0, 4.0)
            if not run_step("搜索关键词", lambda: search_keyword(page, keyword)):
                pass
            elif not run_step("点击视频筛选", lambda: click_video_filter(page, keyword)):
                pass
            elif not run_step("打开第一个视频", lambda: open_first_video(page)):
                pass
            else:
                human_pause(2.0, 4.0)
                prev_video_id = ""
                success = 0
                attempt = 0
                need_switch = False

                while success < video_limit and attempt < max_attempts:
                    attempt += 1
                    print("=" * 48)
                    print(
                        f"尝试采集第 {attempt} 个候选 "
                        f"(已成功 {success}/{video_limit})"
                    )
                    print("=" * 48)

                    if need_switch:
                        switched = {"id": ""}

                        def _switch() -> None:
                            switched["id"] = go_to_next_video(page, prev_video_id)

                        if not run_step("切换到下一条搜索视频", _switch):
                            print("无法继续在搜索结果中切视频，结束采集。")
                            break
                        if switched["id"]:
                            prev_video_id = switched["id"]
                        human_pause(2.0, 3.5)
                    need_switch = True

                    video_meta_box: dict = {}
                    try:
                        print("[获取作者信息/AI总结并入库] 开始")
                        video_meta_box["meta"] = collect_and_save_video_meta(
                            page, keyword, limit
                        )
                        print("[获取作者信息/AI总结并入库] 完成")
                    except SkipVideoError as exc:
                        print(f"[跳过视频] {exc}")
                        human_pause(1.0, 2.0)
                        continue
                    except Exception as exc:
                        print(f"[获取作者信息/AI总结并入库] 失败: {exc}")
                        print("准备重试一次...")
                        human_pause(2.0, 4.0)
                        try:
                            video_meta_box["meta"] = collect_and_save_video_meta(
                                page, keyword, limit
                            )
                            print("[获取作者信息/AI总结并入库] 完成")
                        except SkipVideoError as exc2:
                            print(f"[跳过视频] {exc2}")
                            continue
                        except Exception as exc2:
                            print(f"[获取作者信息/AI总结并入库] 重试仍失败: {exc2}")
                            continue

                    prev_video_id = video_meta_box["meta"].get("video_id") or prev_video_id
                    human_pause(1.5, 2.5)

                    try:
                        print("[打开评论区并按序采集入库] 开始")
                        collect_comments_and_save(
                            page, video_meta_box["meta"], limit
                        )
                        print("[打开评论区并按序采集入库] 完成")
                        success += 1
                    except SkipVideoError as exc:
                        print(f"[跳过视频] {exc}")
                    except Exception as exc:
                        print(f"[打开评论区并按序采集入库] 失败: {exc}")
                        print("准备重试一次...")
                        human_pause(2.0, 4.0)
                        try:
                            collect_comments_and_save(
                                page, video_meta_box["meta"], limit
                            )
                            print("[打开评论区并按序采集入库] 完成")
                            success += 1
                        except SkipVideoError as exc2:
                            print(f"[跳过视频] {exc2}")
                        except Exception as exc2:
                            print(f"[打开评论区并按序采集入库] 重试仍失败: {exc2}")

                    human_pause(1.5, 3.0)

                print(f"多视频采集结束：成功 {success}/{video_limit}，尝试 {attempt} 次。")

        print("流程结束（失败步骤已跳过）。浏览器保持打开，关闭窗口或按 Ctrl+C 结束。")
        try:
            page.wait_for_event("close")
        except Exception:
            pass
        context.close()