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): 可选的代理服务器 URLenable_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¶
通过用户名查询用户资料。
参数:
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¶
通过用户 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¶
获取账号详细信息,包括验证状态、账号来源地等。
参数:
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¶
编辑当前认证用户的资料信息。
参数:
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¶
更换用户头像。需要先上传图片获取 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¶
更换用户背景图(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¶
获取指定用户的粉丝列表(支持分页)。
参数:
user_id(str): 目标用户的 Twitter IDcursor(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¶
获取指定用户的关注列表(支持分页)。
参数:
user_id(str): 目标用户的 Twitter IDcursor(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¶
关注指定用户。
参数:
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¶
取消关注指定用户。
参数:
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): 用户 IDscreen_name(str | None): 用户名(不含 @)name(str | None): 显示名称description(str | None): 个人简介location(str | None): 位置url(str | None): 个人网站 URLprofile_image_url(str | None): 头像 URLbackground_image(str | None): 背景图 URLfollowing_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 IDaccount_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): 更新后的网站 URLerror_msg(str): 错误消息http_status(int): HTTP 状态码
ChangeProfileImageResult¶
更换头像结果。
属性:
success(bool): 请求是否成功profile_image_url(str | None): 新头像 URLerror_msg(str): 错误消息http_status(int): HTTP 状态码
ChangeBannerResult¶
更换背景图结果。
属性:
success(bool): 请求是否成功banner_url(str | None): 新背景图 URLerror_msg(str): 错误消息http_status(int): HTTP 状态码
FollowUser¶
粉丝/关注用户信息。
属性:
rest_id(str): 用户 REST IDname(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): 头像 URLdescription(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): 目标用户 IDerror_msg(str): 错误消息http_status(int): HTTP 状态码
UnfollowResult¶
取消关注结果。
属性:
success(bool): 请求是否成功user_id(str): 目标用户 IDerror_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: 编辑资料时如何清空某个字段?¶
传递空字符串即可:
Q5: 为什么 following 字段有时为 False?¶
following 表示当前认证用户是否关注目标用户:
True: 你正在关注该用户False: 你没有关注该用户
如果查询的是自己的资料,该字段可能没有意义。
Q6: 如何判断用户是否被封禁?¶
被封禁的用户查询时可能返回:
success = Falseerror_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)} 个用户")
下一步¶
- 查看 Upload 模块文档 了解如何上传头像图片
- 查看 Posts 模块文档 了解帖子功能
- 查看 DM 模块文档 了解如何给粉丝发送私信
- 查看 用户采集示例 了解完整的数据采集流程
- 查看 示例代码 了解更多用法