User 用户模块示例¶
本文档提供 User 用户模块的完整使用示例。
目录¶
前置准备¶
安装依赖¶
创建客户端¶
import x_api
cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
twitter = x_api.Twitter(cookies)
# 或使用代理
twitter = x_api.Twitter(cookies, "http://127.0.0.1:7890")
查询用户资料¶
通过用户名查询用户的详细资料。
import asyncio
import x_api
async def get_user_profile():
cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
twitter = x_api.Twitter(cookies)
# 查询用户资料(不需要 @ 符号)
result = await twitter.user.get_profile("elonmusk")
if result.success:
print(f"👤 用户资料")
print(f" ID: {result.user_id}")
print(f" 用户名: @{result.screen_name}")
print(f" 显示名: {result.name}")
print(f" 简介: {result.description}")
print(f" 位置: {result.location}")
print(f" 网站: {result.url}")
print(f" 粉丝数: {result.followers_count:,}")
print(f" 关注数: {result.following_count:,}")
print(f" 私密账号: {result.protected}")
print(f" 我是否关注: {result.following}")
else:
print(f"❌ 查询失败: {result.error_msg}")
print(f" HTTP 状态码: {result.http_status}")
asyncio.run(get_user_profile())
输出示例:
👤 用户资料
ID: 44196397
用户名: @elonmusk
显示名: Elon Musk
简介: Read @America
位置: None
网站: None
粉丝数: 210,345,678
关注数: 876
私密账号: False
我是否关注: True
获取头像和背景图 URL¶
import asyncio
import x_api
async def get_user_images():
cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
twitter = x_api.Twitter(cookies)
result = await twitter.user.get_profile("elonmusk")
if result.success:
print(f"🖼️ 用户图片")
print(f" 头像: {result.profile_image_url}")
print(f" 背景图: {result.background_image}")
asyncio.run(get_user_images())
通过用户 ID 查询¶
如果你知道用户的数字 ID,可以直接通过 ID 查询。
import asyncio
import x_api
async def get_profile_by_id():
cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
twitter = x_api.Twitter(cookies)
# Elon Musk 的用户 ID
user_id = "44196397"
result = await twitter.user.get_profile_by_id(user_id)
if result.success:
print(f"✅ 找到用户: @{result.screen_name}")
print(f" 显示名: {result.name}")
print(f" 粉丝数: {result.followers_count:,}")
else:
print(f"❌ 查询失败: {result.error_msg}")
asyncio.run(get_profile_by_id())
从用户名获取用户 ID¶
import asyncio
import x_api
async def username_to_id():
cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
twitter = x_api.Twitter(cookies)
username = "elonmusk"
result = await twitter.user.get_profile(username)
if result.success:
print(f"@{username} 的用户 ID 是: {result.user_id}")
asyncio.run(username_to_id())
获取账号详细信息¶
获取账号的验证状态、注册地等详细信息。
import asyncio
import x_api
async def get_account_details():
cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
twitter = x_api.Twitter(cookies)
result = await twitter.user.get_about_account("elonmusk")
if result.success:
print(f"📋 账号详情")
print(f" 用户 ID: {result.rest_id}")
# 验证状态
if result.is_blue_verified:
print(f" ✅ 蓝V认证")
else:
print(f" ❌ 未蓝V认证")
if result.is_identity_verified:
print(f" ✅ 身份认证")
else:
print(f" ❌ 未身份认证")
# 账号信息
if result.account_based_in:
print(f" 📍 账号来源: {result.account_based_in}")
if result.username_change_count:
print(f" 🔄 用户名修改次数: {result.username_change_count}")
else:
print(f"❌ 获取失败: {result.error_msg}")
asyncio.run(get_account_details())
输出示例:
编辑个人资料¶
更新当前登录账号的个人资料信息。
import asyncio
import x_api
from x_api_rs.user import EditUserParams
async def edit_my_profile():
cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
twitter = x_api.Twitter(cookies)
# 创建编辑参数
params = EditUserParams(
name="Python Developer 🐍",
description="Building amazing things with Python and Rust | Tech enthusiast",
location="San Francisco, CA",
url="https://github.com/myusername"
)
result = await twitter.user.edit_profile(params)
if result.success:
print(f"✅ 资料更新成功!")
print(f" 新名称: {result.name}")
print(f" 新简介: {result.description}")
print(f" 新位置: {result.location}")
print(f" 新网站: {result.url}")
else:
print(f"❌ 更新失败: {result.error_msg}")
asyncio.run(edit_my_profile())
只更新部分字段¶
import asyncio
import x_api
from x_api_rs.user import EditUserParams
async def update_bio_only():
cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
twitter = x_api.Twitter(cookies)
# 只更新简介
params = EditUserParams(
description="New bio text 🚀"
)
result = await twitter.user.edit_profile(params)
if result.success:
print(f"✅ 简介已更新: {result.description}")
asyncio.run(update_bio_only())
清空字段¶
import asyncio
import x_api
from x_api_rs.user import EditUserParams
async def clear_location():
cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
twitter = x_api.Twitter(cookies)
# 清空位置信息(传入空字符串)
params = EditUserParams(location="")
result = await twitter.user.edit_profile(params)
if result.success:
print(f"✅ 位置已清空")
asyncio.run(clear_location())
更换头像¶
上传新头像并更换。
import asyncio
import x_api
from pathlib import Path
async def change_avatar():
cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
twitter = x_api.Twitter(cookies)
# 步骤1:读取头像图片
avatar_path = Path("avatar.jpg")
if not avatar_path.exists():
print("❌ 头像文件不存在")
return
avatar_bytes = avatar_path.read_bytes()
print(f"📖 读取头像: {len(avatar_bytes)} 字节")
# 步骤2:上传头像(使用 banner_image 类别)
upload_result = await twitter.upload.image(
image_bytes=avatar_bytes,
media_category="banner_image" # 头像和背景图都用这个类别
)
if not upload_result.success:
print(f"❌ 头像上传失败: {upload_result.error_msg}")
return
print(f"✅ 头像上传成功: {upload_result.media_id_string}")
# 步骤3:更换头像
result = await twitter.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}")
asyncio.run(change_avatar())
输出示例:
📖 读取头像: 102400 字节
✅ 头像上传成功: 1234567890123456789
✅ 头像更换成功!
新头像 URL: https://pbs.twimg.com/profile_images/...
头像尺寸建议¶
- 推荐尺寸:400x400 像素
- 最小尺寸:200x200 像素
- 格式:JPEG、PNG
- Twitter 会将头像裁剪为圆形显示
更换背景图¶
上传新背景图(Banner)并更换。
import asyncio
import x_api
from pathlib import Path
async def change_banner():
cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
twitter = x_api.Twitter(cookies)
# 步骤1:读取背景图
banner_path = Path("banner.jpg")
if not banner_path.exists():
print("❌ 背景图文件不存在")
return
banner_bytes = banner_path.read_bytes()
print(f"📖 读取背景图: {len(banner_bytes)} 字节")
# 步骤2:上传背景图
upload_result = await twitter.upload.image(
image_bytes=banner_bytes,
media_category="banner_image"
)
if not upload_result.success:
print(f"❌ 背景图上传失败: {upload_result.error_msg}")
return
print(f"✅ 背景图上传成功: {upload_result.media_id_string}")
# 步骤3:更换背景图
result = await twitter.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}")
asyncio.run(change_banner())
背景图尺寸建议¶
- 推荐尺寸:1500x500 像素(3:1 比例)
- 最小尺寸:600x200 像素
- 格式:JPEG、PNG
- 文件大小:建议 < 2MB
同时更换头像和背景图¶
import asyncio
import x_api
from pathlib import Path
async def update_all_images():
cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
twitter = x_api.Twitter(cookies)
# 更换头像
avatar_bytes = Path("avatar.jpg").read_bytes()
avatar_upload = await twitter.upload.image(avatar_bytes, "banner_image")
if avatar_upload.success:
avatar_result = await twitter.user.change_profile_image(
avatar_upload.media_id_string
)
if avatar_result.success:
print(f"✅ 头像更换成功")
# 更换背景图
banner_bytes = Path("banner.jpg").read_bytes()
banner_upload = await twitter.upload.image(banner_bytes, "banner_image")
if banner_upload.success:
banner_result = await twitter.user.change_background_image(
banner_upload.media_id_string
)
if banner_result.success:
print(f"✅ 背景图更换成功")
print("\n🎉 个人形象更新完成!")
asyncio.run(update_all_images())
批量查询用户¶
批量查询多个用户的资料。
import asyncio
import x_api
async def batch_get_profiles():
cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
twitter = x_api.Twitter(cookies)
usernames = [
"elonmusk",
"BillGates",
"satloading",
"jack",
"sundarpichai"
]
users = []
print(f"📥 查询 {len(usernames)} 个用户...")
for username in usernames:
result = await twitter.user.get_profile(username)
if result.success:
users.append({
"username": result.screen_name,
"name": result.name,
"followers": result.followers_count or 0,
"following": result.following_count or 0,
})
print(f" ✅ @{username}")
else:
print(f" ❌ @{username}: {result.error_msg}")
# 避免限流
await asyncio.sleep(0.5)
# 按粉丝数排序
users.sort(key=lambda x: x["followers"], reverse=True)
print(f"\n📊 粉丝排行榜:")
for i, user in enumerate(users, 1):
print(f" {i}. @{user['username']}: {user['followers']:,} 粉丝")
asyncio.run(batch_get_profiles())
输出示例:
📥 查询 5 个用户...
✅ @elonmusk
✅ @BillGates
✅ @satloading
✅ @jack
✅ @sundarpichai
📊 粉丝排行榜:
1. @elonmusk: 210,345,678 粉丝
2. @BillGates: 65,123,456 粉丝
3. @sundarpichai: 12,345,678 粉丝
4. @jack: 6,789,012 粉丝
5. @satloading: 123,456 粉丝
检查用户验证状态¶
import asyncio
import x_api
async def check_verification_status():
cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
twitter = x_api.Twitter(cookies)
usernames = ["elonmusk", "BillGates", "jack"]
print(f"🔍 检查用户验证状态...")
for username in usernames:
result = await twitter.user.get_about_account(username)
if result.success:
blue = "✅" if result.is_blue_verified else "❌"
identity = "✅" if result.is_identity_verified else "❌"
print(f" @{username}: 蓝V {blue} | 身份认证 {identity}")
else:
print(f" @{username}: 查询失败")
await asyncio.sleep(0.5)
asyncio.run(check_verification_status())
错误处理¶
处理用户模块操作中的常见错误。
import asyncio
import x_api
async def handle_user_errors():
cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
twitter = x_api.Twitter(cookies)
try:
result = await twitter.user.get_profile("some_username")
if result.success:
print(f"✅ 找到用户: @{result.screen_name}")
# 检查是否为私密账号
if result.protected:
print("⚠️ 这是私密账号,部分信息可能不可见")
# 检查账号状态
if result.profile_interstitial_type:
print(f"⚠️ 账号状态异常: {result.profile_interstitial_type}")
else:
# 处理 API 返回的错误
match result.http_status:
case 400:
print("❌ 请求参数错误")
case 401:
print("❌ 认证失败,请检查 cookies")
case 403:
print("❌ 权限不足")
case 404:
print("❌ 用户不存在或已被封禁")
case 429:
print("❌ 请求过于频繁,请稍后重试")
case _:
print(f"❌ 查询失败: {result.error_msg}")
except x_api.TwitterError as e:
print(f"❌ 客户端错误: {e}")
except Exception as e:
print(f"❌ 未知错误: {e}")
asyncio.run(handle_user_errors())
常见错误码¶
| HTTP 状态码 | 说明 | 处理建议 |
|---|---|---|
| 400 | 请求参数错误 | 检查用户名格式 |
| 401 | 认证失败 | 检查 cookies 是否过期 |
| 403 | 权限不足 | 账号可能被限制 |
| 404 | 用户不存在 | 检查用户名是否正确 |
| 429 | 请求频率限制 | 降低请求频率 |
完整示例¶
#!/usr/bin/env python3
"""
User 用户模块完整示例
"""
import asyncio
import os
import sys
from pathlib import Path
try:
import x_api
from x_api_rs.user import EditUserParams
except ImportError:
print("❌ 请先安装 x_api: maturin develop")
sys.exit(1)
class UserDemo:
"""用户演示类"""
def __init__(self, cookies: str, proxy_url: str | None = None):
self.twitter = x_api.Twitter(cookies, proxy_url)
async def get_user(self, username: str) -> dict | None:
"""获取用户信息"""
result = await self.twitter.user.get_profile(username)
if result.success:
return {
"user_id": result.user_id,
"username": result.screen_name,
"name": result.name,
"bio": result.description,
"followers": result.followers_count,
"following": result.following_count,
}
return None
async def update_profile(
self,
name: str | None = None,
bio: str | None = None,
location: str | None = None
) -> bool:
"""更新个人资料"""
params = EditUserParams(
name=name,
description=bio,
location=location
)
result = await self.twitter.user.edit_profile(params)
return result.success
async def change_avatar(self, image_path: str) -> bool:
"""更换头像"""
image_bytes = Path(image_path).read_bytes()
upload = await self.twitter.upload.image(image_bytes, "banner_image")
if not upload.success:
return False
result = await self.twitter.user.change_profile_image(
upload.media_id_string
)
return result.success
async def is_verified(self, username: str) -> tuple[bool, bool]:
"""检查用户验证状态,返回 (蓝V, 身份认证)"""
result = await self.twitter.user.get_about_account(username)
if result.success:
return (
result.is_blue_verified or False,
result.is_identity_verified or False
)
return False, False
async def main():
# 读取配置
cookies = os.getenv("TWITTER_COOKIES")
if not cookies:
cookies_file = Path("cookies.txt")
if cookies_file.exists():
cookies = cookies_file.read_text().strip()
else:
print("❌ 请设置 TWITTER_COOKIES 或创建 cookies.txt")
return
demo = UserDemo(cookies)
# 演示获取用户信息
print("=== 获取用户信息 ===")
user = await demo.get_user("elonmusk")
if user:
print(f"✅ 找到用户: @{user['username']}")
print(f" 粉丝: {user['followers']:,}")
else:
print("❌ 未找到用户")
# 检查验证状态
print("\n=== 检查验证状态 ===")
blue, identity = await demo.is_verified("elonmusk")
print(f"蓝V: {'✅' if blue else '❌'}")
print(f"身份认证: {'✅' if identity else '❌'}")
if __name__ == "__main__":
asyncio.run(main())