acquire-lead/collectors/douyin/search.py

"""步骤：搜索关键词 / 视频筛选 / 打开首个视频。"""

from __future__ import annotations

import random
from urllib.parse import quote

from playwright.sync_api import Page

from collectors.douyin.common import human_pause

def search_keyword(page: Page, keyword: str) -> None:
    search = page.get_by_placeholder("搜索你感兴趣的内容").or_(
        page.locator('input[type="search"], input[enterkeyhint="search"]').first
    ).first
    search.wait_for(state="visible", timeout=15_000)

    human_pause(0.8, 1.6)
    search.click()
    human_pause(0.5, 1.2)
    search.fill("")
    human_pause(0.3, 0.8)
    search.press_sequentially(keyword, delay=random.randint(80, 180))
    print(f"已输入关键词: {keyword}")
    human_pause(1.2, 2.5)
    search.press("Enter")
    page.wait_for_load_state("domcontentloaded")
    print(f"已搜索关键词: {keyword}")


def click_video_filter(page: Page, keyword: str) -> None:
    page.wait_for_function(
        """() => {
            const spans = Array.from(document.querySelectorAll('span'));
            return spans.some(s => {
                const t = (s.innerText || '').trim();
                const y = s.getBoundingClientRect().y;
                return t === '视频' && y > 40 && y < 120;
            });
        }""",
        timeout=20_000,
    )
    human_pause(0.8, 1.6)

    clicked = page.evaluate(
        """() => {
            const spans = Array.from(document.querySelectorAll('span'));
            const el = spans.find(s => {
                const t = (s.innerText || '').trim();
                const rect = s.getBoundingClientRect();
                return t === '视频' && rect.width > 0 && rect.y > 40 && rect.y < 120;
            });
            if (!el) return false;
            el.click();
            return true;
        }"""
    )
    if not clicked:
        url = f"https://www.douyin.com/search/{quote(keyword)}?type=video"
        print(f"未点到「视频」栏，改为直接打开: {url}")
        page.goto(url, wait_until="domcontentloaded")
    else:
        print("已点击顶部「视频」筛选项")
        page.wait_for_load_state("domcontentloaded")
    human_pause(2.5, 4.0)


def open_first_video(page: Page) -> None:
    page.wait_for_selector('a[href*="/video/"]', timeout=20_000)
    human_pause(1.0, 2.0)

    first_video = page.locator('a[href*="/video/"]:visible').first
    first_video.wait_for(state="visible", timeout=10_000)
    href = first_video.get_attribute("href") or ""
    print(f"准备打开第一个视频: {href}")
    human_pause(0.8, 1.5)

    try:
        first_video.click(timeout=5_000)
    except Exception:
        print("普通点击失败，改为 JS 点击第一个视频")
        page.evaluate(
            """() => {
                const a = document.querySelector('a[href*="/video/"]');
                if (a) a.click();
            }"""
        )

    page.wait_for_load_state("domcontentloaded")
    print(f"已进入视频页: {page.url}")