跳转至

User 用户模块示例

本文档提供 User 用户模块的完整使用示例。

目录

前置准备

安装依赖

maturin develop

创建客户端

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())

输出示例:

📋 账号详情
   用户 ID: 44196397
   ✅ 蓝V认证
   ❌ 未身份认证
   📍 账号来源: United States
   🔄 用户名修改次数: 0

编辑个人资料

更新当前登录账号的个人资料信息。

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())

相关链接