acquire-lead/collectors/douyin/profile.py

"""步骤：打开用户/作者主页并读取抖音号。"""

from __future__ import annotations

import re

from playwright.sync_api import Page

from collectors.douyin.common import human_pause

def _to_abs_url(href: str) -> str:
    if href.startswith("http"):
        return href
    if href.startswith("//"):
        return f"https:{href}"
    if href.startswith("/"):
        return f"https://www.douyin.com{href}"
    return f"https://www.douyin.com/{href}"


def _normalize_user_key(href: str) -> str:
    """用 /user/xxx 路径作为同一用户缓存键，避免 query 差异导致重复开页。"""
    if not href:
        return ""
    m = re.search(r"/user/[^/?#]+", href)
    return m.group(0) if m else href.strip()


def _extract_douyin_id(profile: Page) -> str:
    """从用户主页提取抖音号；先等到页面出现「抖音号」文案，并做二次确认。"""
    try:
        profile.wait_for_function(
            """() => /抖音号/.test(document.body.innerText || '')""",
            timeout=20_000,
        )
    except Exception:
        human_pause(2.5, 4.0)

    def _read_once() -> str:
        return (
            profile.evaluate(
                """() => {
                    const bodyText = document.body.innerText || '';
                    const m = bodyText.match(/抖音号[：:\\s]*([A-Za-z0-9_.\\-]+)/);
                    if (m) return m[1];
                    const spans = Array.from(document.querySelectorAll('span, p'));
                    for (const el of spans) {
                        const t = (el.innerText || '').trim();
                        const mm = t.match(/^抖音号[：:\\s]*([A-Za-z0-9_.\\-]+)$/);
                        if (mm) return mm[1];
                    }
                    return '';
                }"""
            )
            or ""
        ).strip()

    douyin_id = _read_once()
    if not douyin_id:
        human_pause(1.5, 2.5)
        douyin_id = _read_once()
    if not douyin_id:
        raise RuntimeError("主页上未解析到抖音号")

    # 二次读取，避免偶发读到未渲染完的占位
    human_pause(0.4, 0.8)
    again = _read_once()
    if again and again != douyin_id:
        print(f"[抖音号校验] 两次读取不一致: {douyin_id} vs {again}，采用第二次")
        douyin_id = again
    return douyin_id


def _open_profile_by_url(page: Page, url: str) -> Page:
    profile = page.context.new_page()
    profile.goto(url, wait_until="domcontentloaded")
    profile.wait_for_load_state("domcontentloaded")
    return profile


def _open_author_profile(page: Page, author_link: str) -> Page:
    """
    在新标签打开作者主页，避免同页跳进作者作品流，
    保证后续仍在「搜索结果」视频弹层里上下切换。
    """
    return _open_profile_by_url(page, author_link)

def _open_user_profile(
    page: Page,
    href: str,
    avatar_x: int | None = None,
    avatar_y: int | None = None,
) -> Page:
    """
    打开评论者主页。
    一律用主页链接在新标签打开，避免点错头像/坐标过期导致抖音号错配。
    """
    _ = avatar_x, avatar_y
    url = _to_abs_url(href)
    if "/user/" not in url:
        raise RuntimeError(f"无效用户主页链接: {url}")
    profile = page.context.new_page()
    profile.goto(url, wait_until="domcontentloaded")
    profile.wait_for_load_state("domcontentloaded")
    return profile