跳转至

User 模块 API 文档

User 模块提供完整的 Twitter 用户资料管理功能,支持查询用户信息、编辑个人资料、更换头像和背景图等操作。

目录

模块概述

User 模块支持两种使用方式:

方式 1: 通过 Twitter 主入口访问(推荐)

from x_api_rs import Twitter

client = await Twitter.create(cookies)
result = await client.user.get_profile("elonmusk")

方式 2: 独立创建 UserClient

from x_api_rs.user import UserClient

# 异步创建客户端
user_client = await UserClient.create(cookies, proxy_url="http://proxy:8080")
result = await user_client.get_profile("elonmusk")

功能列表

功能 方法 说明
查询用户资料 get_profile() 通过用户名查询用户信息
通过 ID 查询 get_profile_by_id() 通过用户 ID 查询用户信息
获取账号详情 get_about_account() 获取账号验证状态等详细信息
编辑资料 edit_profile() 更新当前用户的资料信息
更换头像 change_profile_image() 更换用户头像
更换背景图 change_background_image() 更换用户背景图(Banner)
获取粉丝列表 get_followers() 获取指定用户的粉丝列表(支持分页)
获取关注列表 get_following() 获取指定用户的关注列表(支持分页)
关注 follow() 关注指定用户
取消关注 unfollow() 取消关注指定用户

UserClient 类

创建方法

@staticmethod
async def create(
    cookies: str,
    proxy_url: str | None = None,
    enable_ja3: bool = True
) -> UserClient

异步创建新的 User 客户端实例。

参数:

  • cookies (str): Twitter 账号的 cookies 字符串
  • proxy_url (str | None): 可选的代理服务器 URL
  • enable_ja3 (bool): 是否启用 JA3/TLS 指纹模拟(默认 True)
  • True: 使用 Chrome 136 TLS 指纹模拟,增强反检测能力
  • False: 使用 rquest 默认 TLS 配置(无指纹模拟)

返回: UserClient 实例

异常: RuntimeError - 初始化失败

示例:

# 基础使用(启用 JA3,默认行为)
client = await UserClient.create(cookies)

# 使用代理
client = await UserClient.create(cookies, proxy_url="http://proxy:8080")

# 禁用 JA3 指纹模拟
client = await UserClient.create(cookies, enable_ja3=False)

get_profile

async def get_profile(
    screen_name: str
) -> UserResp

通过用户名查询用户资料。

参数:

  • screen_name (str): Twitter 用户名(不含 @ 符号)

返回: UserResp 对象,包含用户详细信息

异常: RuntimeError - 查询失败

示例:

result = await client.user.get_profile("elonmusk")

if result.success:
    print(f"用户名: @{result.screen_name}")
    print(f"显示名: {result.name}")
    print(f"简介: {result.description}")
    print(f"粉丝数: {result.followers_count}")
    print(f"关注数: {result.following_count}")
else:
    print(f"查询失败: {result.error_msg}")

get_profile_by_id

async def get_profile_by_id(
    rest_id: str
) -> UserResp

通过用户 ID 查询用户资料。

参数:

  • rest_id (str): Twitter 用户 REST ID

返回: UserResp 对象

异常: RuntimeError - 查询失败

示例:

# Elon Musk 的用户 ID
result = await client.user.get_profile_by_id("44196397")

if result.success:
    print(f"用户: @{result.screen_name} ({result.name})")

get_about_account

async def get_about_account(
    screen_name: str
) -> AboutAccountResult

获取账号详细信息,包括验证状态、账号来源地等。

参数:

  • screen_name (str): Twitter 用户名(不含 @ 符号)

返回: AboutAccountResult 对象

异常: RuntimeError - 查询失败

示例:

result = await client.user.get_about_account("elonmusk")

if result.success:
    print(f"用户 ID: {result.rest_id}")
    print(f"蓝V认证: {result.is_blue_verified}")
    print(f"身份认证: {result.is_identity_verified}")
    print(f"账号来源地: {result.account_based_in}")
    print(f"用户名修改次数: {result.username_change_count}")

edit_profile

async def edit_profile(
    params: EditUserParams
) -> EditUserResult

编辑当前认证用户的资料信息。

参数:

  • params (EditUserParams): 编辑参数对象,包含要更新的字段

返回: EditUserResult 对象

异常: RuntimeError - 编辑失败

示例:

from x_api_rs.user import EditUserParams

# 创建编辑参数(只需提供要修改的字段)
params = EditUserParams(
    name="New Display Name",
    description="Updated bio text",
    location="San Francisco, CA",
    url="https://example.com"
)

result = await client.user.edit_profile(params)

if result.success:
    print(f"资料已更新!")
    print(f"新名称: {result.name}")
    print(f"新简介: {result.description}")
else:
    print(f"更新失败: {result.error_msg}")

change_profile_image

async def change_profile_image(
    media_id: str
) -> ChangeProfileImageResult

更换用户头像。需要先上传图片获取 media_id

参数:

  • media_id (str): 已上传图片的 media_id(来自 upload.image 返回的 media_id_string

返回: ChangeProfileImageResult 对象

异常: RuntimeError - 更换失败

示例:

# 1. 上传头像图片
with open("avatar.jpg", "rb") as f:
    image_bytes = f.read()

upload_result = await client.upload.image(image_bytes, "banner_image")

if upload_result.success:
    # 2. 更换头像
    result = await client.user.change_profile_image(upload_result.media_id_string)

    if result.success:
        print(f"头像已更换!")
        print(f"新头像 URL: {result.profile_image_url}")
    else:
        print(f"更换失败: {result.error_msg}")

change_background_image

async def change_background_image(
    media_id: str
) -> ChangeBannerResult

更换用户背景图(Banner)。需要先上传图片获取 media_id

参数:

  • media_id (str): 已上传图片的 media_id(来自 upload.image 返回的 media_id_string

返回: ChangeBannerResult 对象

异常: RuntimeError - 更换失败

示例:

# 1. 上传背景图
with open("banner.jpg", "rb") as f:
    image_bytes = f.read()

upload_result = await client.upload.image(image_bytes, "banner_image")

if upload_result.success:
    # 2. 更换背景图
    result = await client.user.change_background_image(upload_result.media_id_string)

    if result.success:
        print(f"背景图已更换!")
        print(f"新背景图 URL: {result.banner_url}")
    else:
        print(f"更换失败: {result.error_msg}")

get_followers

async def get_followers(
    user_id: str,
    cursor: str | None = None
) -> FollowListResult

获取指定用户的粉丝列表(支持分页)。

参数:

  • user_id (str): 目标用户的 Twitter ID
  • cursor (str | None): 分页游标,用于获取下一页数据(首次请求传 None)

返回: FollowListResult 对象

异常: RuntimeError - 查询失败

示例:

# 获取第一页粉丝
result = await client.user.get_followers("44196397")

if result.success:
    print(f"获取到 {len(result.users)} 个粉丝")

    for user in result.users:
        print(f"  @{user.screen_name} - {user.name}")
        print(f"    粉丝: {user.followers_count:,} | 可私信: {user.can_dm}")

    # 检查是否有更多数据
    if result.has_more():
        # 获取下一页
        next_result = await client.user.get_followers(
            "44196397",
            cursor=result.next_cursor
        )

    # 检查 API 限制
    if result.api_limit:
        print(f"剩余请求: {result.api_limit.remaining}/{result.api_limit.limit}")
else:
    if result.is_abnormal:
        print("无法获取粉丝列表(可能是私密账号)")
    else:
        print(f"查询失败: {result.error_msg}")

get_following

async def get_following(
    user_id: str,
    cursor: str | None = None
) -> FollowListResult

获取指定用户的关注列表(支持分页)。

参数:

  • user_id (str): 目标用户的 Twitter ID
  • cursor (str | None): 分页游标,用于获取下一页数据(首次请求传 None)

返回: FollowListResult 对象

异常: RuntimeError - 查询失败

示例:

# 获取第一页关注
result = await client.user.get_following("44196397")

if result.success:
    print(f"获取到 {len(result.users)} 个关注用户")

    for user in result.users:
        print(f"  @{user.screen_name} - {user.name}")
        print(f"    位置: {user.location or 'N/A'}")
        print(f"    简介: {user.description or 'N/A'}")

    # 检查是否有更多数据
    if result.has_more():
        next_result = await client.user.get_following(
            "44196397",
            cursor=result.next_cursor
        )

follow

async def follow(
    user_id: str
) -> FollowResult

关注指定用户。

参数:

  • user_id (str): 目标用户的 Twitter ID

返回: FollowResult 对象

异常: RuntimeError - 关注失败

示例:

result = await client.user.follow("123456789")

if result.success:
    print(f"已关注用户: {result.user_id}")
else:
    print(f"关注失败: {result.error_msg}")

unfollow

async def unfollow(
    user_id: str
) -> UnfollowResult

取消关注指定用户。

参数:

  • user_id (str): 目标用户的 Twitter ID

返回: UnfollowResult 对象

异常: RuntimeError - 取消关注失败

示例:

result = await client.user.unfollow("123456789")

if result.success:
    print(f"已取消关注用户: {result.user_id}")
else:
    print(f"取消关注失败: {result.error_msg}")

类型定义

UserResp

用户资料响应。

属性:

  • success (bool): 请求是否成功
  • user_id (str | None): 用户 ID
  • screen_name (str | None): 用户名(不含 @)
  • name (str | None): 显示名称
  • description (str | None): 个人简介
  • location (str | None): 位置
  • url (str | None): 个人网站 URL
  • profile_image_url (str | None): 头像 URL
  • background_image (str | None): 背景图 URL
  • following_count (int | None): 关注数
  • followers_count (int | None): 粉丝数
  • following (bool): 是否正在关注该用户
  • protected (bool | None): 是否为私密账号
  • profile_interstitial_type (str | None): 资料插页类型
  • error_msg (str): 错误消息
  • http_status (int): HTTP 状态码

示例:

result = await client.user.get_profile("elonmusk")

if result.success:
    print(f"User ID: {result.user_id}")
    print(f"Screen Name: @{result.screen_name}")
    print(f"Display Name: {result.name}")
    print(f"Bio: {result.description}")
    print(f"Location: {result.location}")
    print(f"Website: {result.url}")
    print(f"Followers: {result.followers_count}")
    print(f"Following: {result.following_count}")
    print(f"Is Private: {result.protected}")
    print(f"Am I following: {result.following}")

AboutAccountResult

账号详细信息结果。

属性:

  • success (bool): 请求是否成功
  • rest_id (str | None): 用户 REST ID
  • account_based_in (str | None): 账号来源地
  • location_accurate (bool | None): 位置是否准确
  • learn_more_url (str | None): 了解更多链接
  • source (str | None): 来源信息
  • username_change_count (str | None): 用户名修改次数
  • username_last_changed_at_msec (str | None): 最后修改用户名的时间戳
  • is_identity_verified (bool | None): 是否身份认证
  • is_blue_verified (bool | None): 是否蓝V认证
  • error_msg (str): 错误消息
  • http_status (int): HTTP 状态码

示例:

result = await client.user.get_about_account("elonmusk")

if result.success:
    print(f"蓝V: {result.is_blue_verified}")
    print(f"身份认证: {result.is_identity_verified}")
    print(f"账号来源: {result.account_based_in}")

EditUserParams

编辑用户资料参数。

属性:

  • name (str | None): 显示名称
  • description (str | None): 个人简介
  • location (str | None): 位置
  • url (str | None): 个人网站 URL

示例:

from x_api_rs.user import EditUserParams

# 只更新部分字段
params = EditUserParams(name="New Name")

# 更新多个字段
params = EditUserParams(
    name="New Name",
    description="New bio",
    location="New York",
    url="https://example.com"
)

EditUserResult

编辑用户资料结果。

属性:

  • success (bool): 请求是否成功
  • name (str | None): 更新后的显示名称
  • description (str | None): 更新后的个人简介
  • location (str | None): 更新后的位置
  • url (str | None): 更新后的网站 URL
  • error_msg (str): 错误消息
  • http_status (int): HTTP 状态码

ChangeProfileImageResult

更换头像结果。

属性:

  • success (bool): 请求是否成功
  • profile_image_url (str | None): 新头像 URL
  • error_msg (str): 错误消息
  • http_status (int): HTTP 状态码

ChangeBannerResult

更换背景图结果。

属性:

  • success (bool): 请求是否成功
  • banner_url (str | None): 新背景图 URL
  • error_msg (str): 错误消息
  • http_status (int): HTTP 状态码

FollowUser

粉丝/关注用户信息。

属性:

  • rest_id (str): 用户 REST ID
  • name (str): 显示名称
  • screen_name (str): 用户名(不含 @)
  • location (str | None): 位置
  • can_dm (bool): 是否可以发送私信
  • followers_count (int): 粉丝数
  • following_count (int): 关注数
  • media_count (int): 媒体数量
  • protected (bool): 是否受保护(私密账号)
  • profile_image_url (str | None): 头像 URL
  • description (str | None): 个人简介

FollowListResult

粉丝/关注列表响应。

属性:

  • success (bool): 请求是否成功
  • users (list[FollowUser]): 用户列表
  • next_cursor (str | None): 下一页游标(如果为 None 或以 "0|" 开头表示没有更多数据)
  • is_abnormal (bool): 是否异常(如私密账号无法获取列表)
  • api_limit (ApiLimit | None): API 速率限制信息
  • error_msg (str): 错误消息
  • http_status (int): HTTP 状态码

方法:

  • has_more() -> bool: 判断是否还有更多数据

ApiLimit

API 速率限制信息。

属性:

  • limit (int | None): 速率限制上限
  • remaining (int | None): 剩余请求次数
  • reset (int | None): 重置时间(Unix 时间戳)

FollowResult

关注结果。

属性:

  • success (bool): 请求是否成功
  • user_id (str): 目标用户 ID
  • error_msg (str): 错误消息
  • http_status (int): HTTP 状态码

UnfollowResult

取消关注结果。

属性:

  • success (bool): 请求是否成功
  • user_id (str): 目标用户 ID
  • error_msg (str): 错误消息
  • http_status (int): HTTP 状态码

使用示例

查询用户资料

import asyncio
from x_api_rs import Twitter

async def main():
    client = await Twitter.create(cookies)

    # 通过用户名查询
    result = await client.user.get_profile("elonmusk")

    if result.success:
        print(f"👤 @{result.screen_name}")
        print(f"📛 {result.name}")
        print(f"📝 {result.description}")
        print(f"📍 {result.location}")
        print(f"👥 粉丝: {result.followers_count:,}")
        print(f"👁️ 关注: {result.following_count:,}")

asyncio.run(main())

批量查询用户

async def batch_get_profiles():
    client = await Twitter.create(cookies)

    usernames = ["elonmusk", "BillGates", "satloading"]
    users = []

    for username in usernames:
        result = await client.user.get_profile(username)
        if result.success:
            users.append({
                "username": result.screen_name,
                "name": result.name,
                "followers": result.followers_count
            })
        await asyncio.sleep(0.5)  # 避免限流

    # 按粉丝数排序
    users.sort(key=lambda x: x["followers"] or 0, reverse=True)

    for user in users:
        print(f"@{user['username']}: {user['followers']:,} followers")

编辑个人资料

async def update_my_profile():
    client = await Twitter.create(cookies)

    from x_api_rs.user import EditUserParams

    # 更新资料
    params = EditUserParams(
        name="Python Developer 🐍",
        description="Building cool stuff with Python and Rust",
        location="San Francisco, CA"
    )

    result = await client.user.edit_profile(params)

    if result.success:
        print("资料更新成功!")
        print(f"新名称: {result.name}")
        print(f"新简介: {result.description}")
    else:
        print(f"更新失败: {result.error_msg}")

更换头像和背景图

async def update_profile_images():
    client = await Twitter.create(cookies)

    # 更换头像
    with open("avatar.jpg", "rb") as f:
        avatar_bytes = f.read()

    avatar_upload = await client.upload.image(avatar_bytes, "banner_image")
    if avatar_upload.success:
        avatar_result = await client.user.change_profile_image(
            avatar_upload.media_id_string
        )
        if avatar_result.success:
            print(f"头像更换成功: {avatar_result.profile_image_url}")

    # 更换背景图
    with open("banner.jpg", "rb") as f:
        banner_bytes = f.read()

    banner_upload = await client.upload.image(banner_bytes, "banner_image")
    if banner_upload.success:
        banner_result = await client.user.change_background_image(
            banner_upload.media_id_string
        )
        if banner_result.success:
            print(f"背景图更换成功: {banner_result.banner_url}")

检查用户认证状态

async def check_verification():
    client = await Twitter.create(cookies)

    result = await client.user.get_about_account("elonmusk")

    if result.success:
        print(f"用户 ID: {result.rest_id}")

        if result.is_blue_verified:
            print("✅ 已蓝V认证")
        else:
            print("❌ 未蓝V认证")

        if result.is_identity_verified:
            print("✅ 已身份认证")
        else:
            print("❌ 未身份认证")

        if result.account_based_in:
            print(f"📍 账号来源: {result.account_based_in}")

获取粉丝列表

async def get_all_followers():
    client = await Twitter.create(cookies)
    user_id = "44196397"  # Elon Musk
    all_followers = []
    cursor = None

    while True:
        result = await client.user.get_followers(user_id, cursor)

        if not result.success:
            if result.is_abnormal:
                print("无法获取粉丝列表(私密账号)")
            else:
                print(f"获取失败: {result.error_msg}")
            break

        print(f"本次获取 {len(result.users)} 个粉丝")
        all_followers.extend(result.users)

        # 检查 API 限制
        if result.api_limit:
            print(f"API 限制: {result.api_limit.remaining}/{result.api_limit.limit}")
            print(f"重置时间: {result.api_limit.reset}")

        # 检查是否还有更多数据
        if not result.has_more():
            print("已获取所有粉丝")
            break

        cursor = result.next_cursor
        await asyncio.sleep(2)  # 避免限流

    print(f"\n总计获取 {len(all_followers)} 个粉丝")

    # 筛选可私信的粉丝
    dm_available = [u for u in all_followers if u.can_dm]
    print(f"可私信粉丝: {len(dm_available)} 人")

获取关注列表

async def get_all_following():
    client = await Twitter.create(cookies)
    user_id = "44196397"
    all_following = []
    cursor = None

    while True:
        result = await client.user.get_following(user_id, cursor)

        if not result.success:
            print(f"获取失败: {result.error_msg}")
            break

        all_following.extend(result.users)

        if not result.has_more():
            break

        cursor = result.next_cursor
        await asyncio.sleep(2)

    print(f"关注总数: {len(all_following)}")

    # 按粉丝数排序
    all_following.sort(key=lambda u: u.followers_count, reverse=True)

    print("\n粉丝最多的 10 个关注用户:")
    for i, user in enumerate(all_following[:10], 1):
        print(f"{i}. @{user.screen_name}: {user.followers_count:,} 粉丝")

批量取消关注

async def unfollow_inactive_users():
    client = await Twitter.create(cookies)

    # 获取当前用户的关注列表
    my_id = "your_user_id"
    result = await client.user.get_following(my_id)

    if not result.success:
        print("获取关注列表失败")
        return

    # 筛选需要取消关注的用户(例如:6个月未发帖)
    to_unfollow = [
        user for user in result.users
        if user.media_count == 0  # 示例条件
    ]

    print(f"将取消关注 {len(to_unfollow)} 个用户")

    for user in to_unfollow:
        unfollow_result = await client.user.unfollow(user.rest_id)

        if unfollow_result.success:
            print(f"✅ 已取消关注 @{user.screen_name}")
        else:
            print(f"❌ 取消关注失败: @{user.screen_name}")

        await asyncio.sleep(1)  # 避免限流

导出用户数据到 CSV

import csv
import asyncio
from x_api_rs import Twitter

async def export_followers_to_csv():
    client = await Twitter.create(cookies)
    user_id = "44196397"

    # 获取粉丝列表
    result = await client.user.get_followers(user_id)

    if not result.success:
        print("获取粉丝失败")
        return

    # 导出到 CSV
    with open("followers.csv", "w", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerow([
            "User ID", "Username", "Name", "Followers",
            "Following", "Can DM", "Location", "Bio"
        ])

        for user in result.users:
            writer.writerow([
                user.rest_id,
                user.screen_name,
                user.name,
                user.followers_count,
                user.following_count,
                user.can_dm,
                user.location or "",
                user.description or ""
            ])

    print(f"已导出 {len(result.users)} 个粉丝到 followers.csv")

最佳实践

1. 验证操作结果

result = await client.user.get_profile("username")

# 始终检查 success 状态
if result.success:
    # 成功处理
    print(f"Found: @{result.screen_name}")
else:
    # 失败处理
    print(f"Error: {result.error_msg}")
    print(f"HTTP Status: {result.http_status}")

2. 处理私密账号

result = await client.user.get_profile("username")

if result.success:
    if result.protected:
        print("这是一个私密账号,部分信息可能受限")
    else:
        print(f"粉丝数: {result.followers_count}")

3. 上传头像/背景图时使用正确的类别

# 头像和背景图都使用 banner_image 类别
upload_result = await client.upload.image(image_bytes, "banner_image")

# 不要使用其他类别
# ❌ await client.upload.image(image_bytes, "tweet_image")  # 错误
# ❌ await client.upload.image(image_bytes, "dm_image")     # 错误

4. 批量操作添加延迟

usernames = ["user1", "user2", "user3", ...]

for username in usernames:
    result = await client.user.get_profile(username)
    # 处理结果...
    await asyncio.sleep(0.5)  # 避免限流

5. 编辑资料时只传递需要修改的字段

from x_api_rs.user import EditUserParams

# ✅ 好:只传递需要修改的字段
params = EditUserParams(name="New Name")

# ✅ 也可以:传递多个字段
params = EditUserParams(
    name="New Name",
    description="New bio"
)

# 不需要传递空字符串来"清空"字段

6. 采集数据时注意 API 限制

# 检查并处理 API 限制
result = await client.user.get_followers(user_id)

if result.api_limit:
    if result.api_limit.remaining == 0:
        # 计算需要等待的时间
        import time
        wait_time = result.api_limit.reset - int(time.time())
        print(f"API 限制已用完,需等待 {wait_time} 秒")
        await asyncio.sleep(wait_time + 1)
    elif result.api_limit.remaining < 5:
        print(f"警告:仅剩 {result.api_limit.remaining} 次请求")

7. 处理私密账号和异常状态

result = await client.user.get_followers(user_id)

if result.is_abnormal:
    # 私密账号或无权限
    print("该账号设置了隐私保护,无法获取列表")
elif not result.success:
    # 其他错误
    print(f"获取失败: {result.error_msg}")

8. 分页获取时合理控制间隔

# ✅ 推荐:添加延迟避免限流
cursor = None
while True:
    result = await client.user.get_followers(user_id, cursor)
    # 处理数据...
    if not result.has_more():
        break
    cursor = result.next_cursor
    await asyncio.sleep(2)  # 每页间隔 2 秒

# ❌ 不推荐:连续高速请求容易触发限流

常见问题

Q1: 如何获取用户 ID?

可以通过 get_profile() 获取用户 ID:

result = await client.user.get_profile("elonmusk")
if result.success:
    user_id = result.user_id  # "44196397"

Q2: 查询用户时出现 "User not found" 错误?

可能的原因:

  • 用户名拼写错误
  • 用户已被封禁或注销
  • 用户设置了隐私保护

Q3: 头像/背景图上传后尺寸不对?

Twitter 会自动调整图片尺寸:

  • 头像:建议 400x400 像素,会被裁剪为圆形
  • 背景图:建议 1500x500 像素(3:1 比例)

Q4: 编辑资料时如何清空某个字段?

传递空字符串即可:

params = EditUserParams(
    location=""  # 清空位置信息
)
result = await client.user.edit_profile(params)

Q5: 为什么 following 字段有时为 False?

following 表示当前认证用户是否关注目标用户:

  • True: 你正在关注该用户
  • False: 你没有关注该用户

如果查询的是自己的资料,该字段可能没有意义。

Q6: 如何判断用户是否被封禁?

被封禁的用户查询时可能返回:

  • success = False
  • error_msg 包含相关错误信息
  • profile_interstitial_type 有特定值
result = await client.user.get_profile("username")

if not result.success:
    print("用户可能被封禁或不存在")
elif result.profile_interstitial_type:
    print(f"账号状态异常: {result.profile_interstitial_type}")

Q7: 获取粉丝/关注列表的速率限制是多少?

Twitter API 对粉丝/关注列表的速率限制通常为:

  • 每 15 分钟最多 15 次请求
  • 每次最多返回 200 个用户

建议:

  • 每页间隔至少 2-3 秒
  • 监控 api_limit 字段判断剩余请求次数
  • 达到限制后等待重置时间

Q8: 如何判断用户是否可以发送私信?

检查 FollowUser.can_dm 字段:

result = await client.user.get_followers(user_id)

for user in result.users:
    if user.can_dm:
        print(f"@{user.screen_name} 可以发送私信")

Q9: 为什么私密账号无法获取粉丝列表?

私密账号的粉丝/关注列表受保护,除非你关注了该账号且对方接受。检查方法:

result = await client.user.get_followers(user_id)

if result.is_abnormal:
    print("该账号可能是私密账号或设置了隐私保护")

Q10: 如何高效获取大量用户数据?

建议策略:

# 1. 使用生成器分批处理
async def fetch_followers_batch(user_id, batch_size=1000):
    cursor = None
    batch = []

    while True:
        result = await client.user.get_followers(user_id, cursor)

        if not result.success:
            break

        batch.extend(result.users)

        if len(batch) >= batch_size or not result.has_more():
            yield batch
            batch = []

        if not result.has_more():
            break

        cursor = result.next_cursor
        await asyncio.sleep(2)

# 2. 分批处理和存储
async for batch in fetch_followers_batch("44196397", batch_size=500):
    # 处理或存储这批数据
    save_to_database(batch)
    print(f"已处理 {len(batch)} 个用户")

下一步