dy/backend/rpa_engine/douyin_im/follower_poll.py

"""抖音网页版「粉丝列表」拉取，用于检测新粉丝（关注欢迎语功能）。

复用与 peer_profile / account_profile 相同的 a_bogus + msToken + cookie 签名方式，
调用 https://www.douyin.com/aweme/v1/web/user/follower/list/ 拉取本账号最近的粉丝。

返回的每个粉丝含：uid / sec_uid / nickname / follow_status / follower_status。
其中 follow_status 表示「我」与对方的关系：0=未关注 1=我已关注 2=互相关注（互关）。
"""

from __future__ import annotations

import asyncio
import logging
from typing import Any

from .dy_util import (
    DEFAULT_USER_AGENT,
    generate_a_bogus,
    generate_msToken,
    generate_webid,
    splice_url,
)
from .auth import DouyinAuth

logger = logging.getLogger("douyin_im.follower_poll")

FOLLOWER_LIST_URL = "https://www.douyin.com/aweme/v1/web/user/follower/list/"


def _requests_proxies() -> dict | None:
    try:
        from rpa_engine.runtime_config import requests_proxies

        return requests_proxies()
    except Exception:
        return None


def _to_int(value: Any) -> int:
    try:
        return int(value)
    except (TypeError, ValueError):
        return 0


def _extract_followers(data: dict[str, Any]) -> list[dict[str, Any]]:
    raw = data.get("followers")
    if not isinstance(raw, list):
        return []
    out: list[dict[str, Any]] = []
    for item in raw:
        if not isinstance(item, dict):
            continue
        uid = str(item.get("uid") or item.get("user_id") or "").strip()
        if not uid:
            continue
        out.append(
            {
                "uid": uid,
                "sec_uid": str(item.get("sec_uid") or item.get("sec_user_id") or "").strip(),
                "nickname": str(item.get("nickname") or item.get("nick_name") or "").strip(),
                # follow_status：我对对方的关系（2=互关）；follower_status：对方对我的关系
                "follow_status": _to_int(item.get("follow_status")),
                "follower_status": _to_int(item.get("follower_status")),
            }
        )
    return out


def fetch_recent_followers_sync(
    session,
    sec_user_id: str,
    count: int = 20,
    max_time: int = 0,
) -> list[dict[str, Any]]:
    """同步拉取最近粉丝（第一页）。失败返回 []，并在日志里写明原因。"""
    import requests

    sec_user_id = (sec_user_id or "").strip()
    if not sec_user_id:
        logger.warning("fetch followers skipped: 缺少本账号 sec_user_id")
        return []

    try:
        auth = DouyinAuth()
        auth.perepare_auth(session.cookie_header(), session.web_protect_str, session.keys_str)
    except Exception as exc:
        logger.warning("fetch followers: build auth failed: %s", exc)
        return []

    ua = session.user_agent or DEFAULT_USER_AGENT
    params = {
        "device_platform": "webapp",
        "aid": "6383",
        "channel": "channel_pc_web",
        "sec_user_id": sec_user_id,
        "count": str(count),
        "max_time": str(max_time),
        "min_time": "0",
        "offset": "0",
        "source_type": "1",
        "gps_access": "0",
        "address_book_access": "0",
        "is_top": "1",
        "update_version_code": "170400",
        "pc_client_type": "1",
        "version_code": "170400",
        "version_name": "17.4.0",
        "cookie_enabled": "true",
        "screen_width": "1536",
        "screen_height": "960",
        "browser_language": "zh-CN",
        "browser_platform": "Win32",
        "browser_name": "Chrome",
        "browser_version": "120.0.0.0",
        "browser_online": "true",
        "os_name": "Windows",
        "os_version": "10",
        "platform": "PC",
        "webid": generate_webid(auth, "https://www.douyin.com/"),
        "verifyFp": auth.cookie.get("s_v_web_id", "") if auth.cookie else "",
        "fp": auth.cookie.get("s_v_web_id", "") if auth.cookie else "",
        "msToken": auth.msToken or generate_msToken(),
    }
    query = splice_url(params)
    params["a_bogus"] = generate_a_bogus(query, user_agent=ua)

    headers = {
        "User-Agent": ua,
        "Referer": "https://www.douyin.com/",
        "Accept": "application/json, text/plain, */*",
    }
    try:
        resp = requests.get(
            FOLLOWER_LIST_URL,
            params=params,
            headers=headers,
            cookies=auth.cookie,
            timeout=15,
            verify=False,
            proxies=_requests_proxies(),
        )
        try:
            data = resp.json()
        except Exception:
            snippet = (resp.text or "")[:200].replace("\n", " ")
            logger.warning(
                "fetch followers: 非 JSON 响应 (HTTP %s): %s", resp.status_code, snippet
            )
            return []
        if not isinstance(data, dict):
            logger.warning("fetch followers: 响应不是 JSON 对象")
            return []
        status_code = data.get("status_code")
        if status_code not in (None, 0):
            logger.warning(
                "fetch followers: status_code=%s msg=%s",
                status_code,
                data.get("status_msg") or data.get("message") or "",
            )
            return []
        followers = _extract_followers(data)
        logger.info(
            "fetch followers ok: 拿到 %s 个粉丝 (has_more=%s total=%s)",
            len(followers),
            data.get("has_more"),
            data.get("total"),
        )
        return followers
    except Exception as exc:
        logger.warning("fetch followers failed: %s", exc)
        return []


async def fetch_recent_followers(
    session,
    sec_user_id: str,
    count: int = 20,
    max_time: int = 0,
) -> list[dict[str, Any]]:
    return await asyncio.to_thread(
        fetch_recent_followers_sync, session, sec_user_id, count, max_time
    )