跳转至

API 文档索引

本目录包含 X API 各模块的详细 API 文档。

文档列表

模块 文档 描述
DM dm.md 私信发送和管理 API
Upload upload.md 图片和视频上传 API
Posts posts.md 帖子发布和互动 API
User user.md 用户资料查询和编辑 API
Inbox inbox.md 收件箱查询 API

快速导航

按功能分类

私信相关

媒体上传

帖子操作

用户管理

收件箱

使用指南

选择访问方式

所有模块都支持两种访问方式：

方式 1: 通过 Twitter 主入口（推荐）

from x_api_rs import Twitter

client = Twitter(cookies)

# 访问各模块
await client.dm.send_message(...)
await client.upload.image(...)
await client.posts.create_tweet(...)
await client.user.get_profile(...)
await client.inbox.get_user_updates(...)

优点： - 统一的认证和配置 - 自动共享连接池 - 代码简洁

方式 2: 独立模块客户端

from x_api_rs.dm import DMClient
from x_api_rs.upload import UploadClient

dm = DMClient(cookies)
upload = UploadClient(cookies)

适用场景： - 只需单一模块功能 - 需要不同配置（如不同代理）

通用参数

所有客户端构造函数都支持以下参数：

  • cookies (str): Twitter cookies字符串，必需
  • proxy_url (str | None): 代理服务器URL，可选

异步操作

所有 I/O 操作都是异步的，必须使用 await：

import asyncio

async def main():
    client = Twitter(cookies)
    result = await client.dm.send_message("user_id", "Hello!")

asyncio.run(main())

错误处理

所有 API 都可能抛出异常，建议使用 try-except：

try:
    result = await client.dm.send_message("user_id", "Hello!")
    if result.success:
        print("成功")
    else:
        print(f"失败: {result.error_msg}")
except Exception as e:
    print(f"异常: {e}")

类型系统

所有 API 都提供完整的类型提示支持。可以使用 IDE 的自动补全功能查看参数和返回值类型。

常用类型

类型 用途 模块
DMResult 单条私信结果 dm
BatchDMResult 批量私信结果 dm
UploadResult 上传结果 upload
TweetResult 发帖结果 posts
UserResp 用户资料 user

完整示例

查看各模块文档中的示例代码，或访问 examples 目录。

最佳实践

1. 使用主入口

# 推荐
client = Twitter(cookies)
await client.dm.send_message(...)

# 不推荐（除非有特殊需求）
dm = DMClient(cookies)
upload = UploadClient(cookies)

2. 复用客户端实例

# 推荐：复用客户端
client = Twitter(cookies)
await client.dm.send_message(...)
await client.posts.create_tweet(...)

# 不推荐：重复创建
for i in range(10):
    client = Twitter(cookies)  # 浪费资源
    await client.dm.send_message(...)

3. 批量操作优化

# 推荐：使用批量 API
user_ids = ["123", "456", "789"]
await client.dm.send_batch(user_ids, "消息")

# 不推荐：循环调用单条 API
for user_id in user_ids:
    await client.dm.send_message(user_id, "消息")  # 效率低

4. 错误重试

import asyncio

async def send_with_retry(client, user_id, text, max_retries=3):
    for attempt in range(max_retries):
        try:
            result = await client.dm.send_message(user_id, text)
            if result.success:
                return result
            await asyncio.sleep(2 ** attempt)  # 指数退避
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(2 ** attempt)

性能建议

并发控制

import asyncio

# 限制并发数
semaphore = asyncio.Semaphore(5)  # 最多5个并发

async def send_limited(user_id, text):
    async with semaphore:
        return await client.dm.send_message(user_id, text)

tasks = [send_limited(uid, "消息") for uid in user_ids]
results = await asyncio.gather(*tasks)

批量处理

from itertools import islice

def batched(iterable, n):
    it = iter(iterable)
    while True:
        batch = list(islice(it, n))
        if not batch:
            return
        yield batch

# 分批处理
for batch in batched(user_ids, 10):
    result = await client.dm.send_batch(batch, "消息")
    await asyncio.sleep(1)  # 批次间延迟

速率限制

Twitter API 有速率限制，建议：

  • 控制请求频率（建议每秒不超过 3 个请求）
  • 批量操作时添加延迟
  • 监听 HTTP 429 错误
  • 实施指数退避策略

调试

启用详细日志

export RUST_LOG=debug
python your_script.py

使用 Python 日志

import logging

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

result = await client.dm.send_message(...)
logging.info(f"Result: {result}")

常见问题

Q: 如何获取 cookies？

  1. 浏览器登录 Twitter
  2. 打开开发者工具（F12）
  3. Network 标签页
  4. 找到任意请求的 Cookie 头
  5. 复制包含 ct0、auth_token、twid 的完整 cookies

Q: 如何使用代理？

# HTTP 代理
client = Twitter(cookies, proxy_url="http://proxy:8080")

# SOCKS5 代理
client = Twitter(cookies, proxy_url="socks5://proxy:1080")

Q: 如何处理超时？

import asyncio

try:
    result = await asyncio.wait_for(
        client.dm.send_message("user_id", "Hello!"),
        timeout=30.0  # 30秒超时
    )
except asyncio.TimeoutError:
    print("请求超时")

更多资源