acquire-lead/collectors/douyin/navigate.py

"""步骤：在搜索结果中切换下一条视频。"""

from __future__ import annotations

from playwright.sync_api import Page

from collectors.douyin.common import human_pause
from collectors.douyin.panel import (
    _ensure_video_page_focus,
    close_comment_panel,
)

def _ensure_search_video_modal(page: Page) -> None:
    """确认仍在搜索结果的视频弹层，而不是作者作品流。"""
    url = page.url or ""
    if "/search/" not in url:
        raise RuntimeError(f"当前不在搜索结果页，无法在搜索列表中切视频: {url}")
    if "modal_id=" not in url and "/video/" not in url:
        # 有的状态只有 aweme 节点；仍要求停留在 search
        print(f"[切视频] 警告：URL 无 modal_id，但仍在搜索页: {url}")


def _focus_search_video_player(page: Page) -> None:
    """把焦点放到搜索弹层左侧视频区，避免方向键作用在评论/AI 输入框。"""
    try:
        box = page.locator(
            '[data-e2e="feed-active-video"], [data-e2e="modal-video-container"]'
        ).first.bounding_box()
        if box:
            page.mouse.click(
                box["x"] + min(box["width"] * 0.35, 220),
                box["y"] + box["height"] * 0.45,
            )
            human_pause(0.4, 0.8)
            return
    except Exception:
        pass
    page.mouse.click(360, 360)
    human_pause(0.3, 0.6)


def _read_current_video_id(page: Page) -> str:
    """读取当前活跃视频 ID（modal_id / aweme-id）。"""
    return (
        page.evaluate(
            """() => {
                const modalId = new URLSearchParams(location.search).get('modal_id') || '';
                const root = document.querySelector('[data-e2e="feed-active-video"]')
                    || document.querySelector('[data-e2e="modal-video-container"]')
                    || document;
                const infoEl = root.querySelector('[data-e2e="video-info"][data-e2e-aweme-id]')
                    || root.querySelector('[data-e2e="video-info"]');
                const awemeId = infoEl ? (infoEl.getAttribute('data-e2e-aweme-id') || '') : '';
                return awemeId || modalId || '';
            }"""
        )
        or ""
    )


def go_to_next_video(page: Page, prev_video_id: str = "") -> str:
    """
    在「搜索结果」视频弹层中切换到下一条（不是作者作品流）。
    返回新 video_id；失败抛错。
    """
    _ensure_search_video_modal(page)
    close_comment_panel(page)
    human_pause(0.8, 1.5)
    _focus_search_video_player(page)

    before = (prev_video_id or "").strip() or _read_current_video_id(page)

    def _changed() -> str:
        _ensure_search_video_modal(page)
        now = _read_current_video_id(page)
        return now if now and now != before else ""

    page.keyboard.press("ArrowDown")
    human_pause(1.5, 2.5)
    new_id = _changed()
    if new_id:
        print(f"已切换到下一条搜索视频: {new_id}")
        _ensure_video_page_focus(page)
        return new_id

    page.keyboard.press("ArrowDown")
    human_pause(1.5, 2.5)
    new_id = _changed()
    if new_id:
        print(f"已切换到下一条搜索视频(二次方向键): {new_id}")
        _ensure_video_page_focus(page)
        return new_id

    # 滚轮只作用在左侧视频区
    try:
        box = page.locator(
            '[data-e2e="feed-active-video"], [data-e2e="modal-video-container"]'
        ).first.bounding_box()
        if box:
            page.mouse.move(box["x"] + box["width"] * 0.3, box["y"] + box["height"] * 0.5)
            page.mouse.wheel(0, 1400)
            human_pause(1.5, 2.5)
    except Exception:
        pass

    new_id = _changed()
    if new_id:
        print(f"已切换到下一条搜索视频(滚轮): {new_id}")
        _ensure_video_page_focus(page)
        return new_id

    raise RuntimeError(f"未能在搜索结果中切换到下一条视频（仍停留在 {before or '未知'}）")