跳转至

模块总览

本文档提供 X API 各模块的功能概览、架构设计和模块关系说明。

目录

架构总览

X API 采用四层架构设计，实现了清晰的职责分离和高度的模块化：

┌──────────────────────────────────────────────────────────────┐
│                   Twitter (Facade 层)                         │
│  ┌──────┬────────┬────────┬──────────┬──────────────────┐    │
│  │  dm  │ posts  │  user  │  inbox   │  future modules  │    │
│  └───┬──┴────┬───┴────┬───┴────┬─────┴──────────────────┘    │
└──────┼───────┼────────┼────────┼─────────────────────────────┘
       │       │        │        │
       │       │        │        │     Service 层（业务模块）
   ┌───▼──┐ ┌──▼───┐ ┌─▼────┐ ┌▼────────┐
   │ DM   │ │Posts │ │User  │ │ Inbox   │
   │Client│ │Client│ │Client│ │ Client  │
   └───┬──┘ └──┬───┘ └─┬────┘ └┬────────┘
       │       │       │        │
       └───┬───┴───┬───┴────┬───┘
           │       │        │
      ┌────▼───────▼────────▼─────┐  Capability 层（共享能力）
      │  MediaUpload Capability   │
      │  Pagination Capability    │
      └────────────┬──────────────┘
                   │
         ┌─────────▼──────────────┐  Common 层（通用基础）
         │  Auth + HttpClient     │
         └────────────────────────┘

层级说明

层级 名称 职责 示例
Facade 门面层 统一入口，聚合所有模块 Twitter
Service 业务层 具体业务逻辑实现 DMClient, PostsClient
Capability 能力层 可复用的共享功能 MediaUploadCapability
Common 通用层 基础设施和抽象 AuthProvider, HttpClient

设计原则

  • 组合优于继承 - 使用组合模式组织模块
  • 依赖倒置 - 依赖抽象（Trait），而非具体实现
  • 单一职责 - 每个模块职责明确
  • 开放封闭 - 新增模块无需修改现有代码

模块列表

1. DM 模块 (私信)

职责: 私信发送和管理

主要功能: - 单条私信发送 - 批量并发发送（相同内容） - 批量自定义文案发送 - 媒体附件支持 - 消息编码/解码

Python 接口: 

from x_api_rs import Twitter

client = Twitter(cookies)

# 访问 DM 模块
await client.dm.send_message("user_id", "Hello!")
await client.dm.send_batch(user_ids, "批量消息")
await client.dm.send_batch_with_custom_texts(user_ids, texts)

详细文档: DM API 文档

2. Upload 模块 (媒体上传)

职责: 图片和视频上传管理

主要功能: - 图片上传（三阶段：INIT → APPEND → FINALIZE） - 视频上传（带处理状态查询） - 批量上传（获取多个 media_id） - 支持多种媒体类别（tweet_image, dm_image, banner_image, amplify_video, dm_video）

Python 接口: 

# 上传图片
result = await client.upload.image(image_bytes, "tweet_image")

# 上传视频
result = await client.upload.video(video_bytes, "amplify_video", duration_ms=10819.0)

# 批量上传同一张图片
result = await client.upload.image_multiple_times(image_bytes, "dm_image", count=3)

详细文档: Upload API 文档

3. Posts 模块 (帖子)

职责: 帖子发布和互动管理

主要功能: - 发布帖子（支持文字、图片、视频、回复、引用） - 删除帖子 - 点赞/取消点赞 - 转发/取消转发 - 获取用户帖子列表 - 获取点赞列表

Python 接口: 

# 发帖
result = await client.posts.create_tweet(
    text="Hello World!",
    media_ids=["123"],
    reply_to_tweet_id="456"
)

# 点赞
await client.posts.favorite_tweet("tweet_id")

# 转发
await client.posts.create_retweet("tweet_id")

# 获取帖子列表
result = await client.posts.get_tweets("user_id")

详细文档: Posts API 文档

4. User 模块 (用户)

职责: 用户资料查询和编辑

主要功能: - 获取用户资料（通过用户名或 ID） - 编辑用户资料（名称、简介等） - 更换头像 - 更换背景图（Banner） - 获取账号详细信息

Python 接口: 

# 获取用户资料
user = await client.user.get_profile("elonmusk")
user = await client.user.get_profile_by_id("44196397")

# 编辑资料
from x_api_rs.user import EditUserParams
params = EditUserParams(name="New Name", description="New bio")
result = await client.user.edit_profile(params)

# 更换头像
await client.user.change_profile_image(media_id)

# 更换背景图
await client.user.change_background_image(media_id)

详细文档: User API 文档

5. Inbox 模块 (收件箱)

职责: 私信收件箱查询

主要功能: - 获取私信更新 - 支持分页查询 - 活跃会话跟踪

Python 接口: 

# 获取收件箱更新
result = await client.inbox.get_user_updates()

# 分页查询
result = await client.inbox.get_user_updates(cursor="next_cursor")

# 指定活跃会话
result = await client.inbox.get_user_updates(active_conversation_id="conv_id")

详细文档: Inbox API 文档

模块依赖关系

依赖图

Twitter (主入口)
├── DMClient
│   └── MediaUploadCapability (可选，用于媒体附件)
├── UploadClient
│   └── MediaUploadCapability
├── PostsClient
│   └── MediaUploadCapability (可选，用于图片/视频帖子)
├── UserClient
│   └── MediaUploadCapability (用于头像/背景图上传)
└── InboxClient
    └── PaginationCapability (可选，用于分页)

共享基础设施:
├── AuthProvider (认证)
├── HttpClient (HTTP 请求)
└── TwitterError (错误处理)

模块间通信

  • 独立使用: 每个模块可以独立创建和使用
  • 通过 Twitter 访问: 推荐通过 Twitter 主入口访问各模块
  • 共享能力: 通过 Capability 层共享通用功能（如媒体上传）
  • 无环依赖: 模块间无循环依赖，保证清晰的层级结构

使用指南

基础用法

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

from x_api_rs import Twitter

# 创建主客户端
client = Twitter(cookies, proxy_url="http://proxy:8080")

# 访问各模块
await client.dm.send_message("user_id", "Hello!")
await client.upload.image(image_bytes, "tweet_image")
await client.posts.create_tweet(text="Hello!")
await client.user.get_profile("elonmusk")
await client.inbox.get_user_updates()

优点: - 统一的认证和配置管理 - 自动共享连接池和缓存 - 代码简洁，易于维护

方式 2: 独立创建模块客户端

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

# 独立创建各模块客户端
dm_client = DMClient(cookies)
upload_client = UploadClient(cookies)

await dm_client.send_message("user_id", "Hello!")
await upload_client.image(image_bytes, "tweet_image")

使用场景: - 只需要单一模块功能 - 需要不同的配置（如不同代理） - 微服务架构中的独立服务

高级用法

批量操作

# 批量私信
user_ids = ["123", "456", "789"]
result = await client.dm.send_batch(user_ids, "批量消息")

# 批量自定义文案
texts = ["你好，张三！", "你好，李四！"]
result = await client.dm.send_batch_with_custom_texts(user_ids, texts)

# 批量上传
result = await client.upload.image_multiple_times(image_bytes, "dm_image", count=5)

组合操作

# 上传图片 + 发送带图私信
upload_result = await client.upload.image(image_bytes, "dm_image")
if upload_result.success:
    dm_result = await client.dm.send_message(
        user_id="123",
        text="请看这张图片",
        media_id=upload_result.media_id_string
    )

# 上传图片 + 发布带图帖子
upload_result = await client.upload.image(image_bytes, "tweet_image")
if upload_result.success:
    tweet_result = await client.posts.create_tweet(
        text="分享一张图片",
        media_ids=[upload_result.media_id_string]
    )

错误处理

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

使用 auth_token

# 将 auth_token 转换为 cookies
result = await Twitter.auth_token_to_cookies(
    "your_auth_token",
    proxy_url="http://proxy:8080"
)

# 使用转换后的 cookies
client = Twitter(result.cookies)

模块扩展

添加新模块的步骤

  1. 定义 Service Trait (src/x/newmodule/service.rs)
  2. 实现 Client (src/x/newmodule/client.rs)
  3. 定义类型 (src/x/newmodule/types.rs)
  4. 创建 Python 绑定 (src/python/newmodule/)
  5. 在 Twitter 中注册 (src/lib.rs 和 src/python/mod.rs)
  6. 编写测试 (tests/x/newmodule/)
  7. 编写文档 (docs/api/newmodule.md)

详细步骤请参考：开发文档

常见问题

Q1: 如何选择使用方式（主入口 vs 独立模块）？

建议: 优先使用 Twitter 主入口。只有在以下情况下使用独立模块： - 只需要单一模块功能 - 需要不同的配置（如不同账号、不同代理） - 内存受限的环境

Q2: 如何处理并发请求？

所有 API 都是异步的，可以使用 asyncio.gather 并发执行：

import asyncio

# 并发发送多条私信
tasks = [
    client.dm.send_message("123", "消息1"),
    client.dm.send_message("456", "消息2"),
    client.dm.send_message("789", "消息3"),
]
results = await asyncio.gather(*tasks)

Q3: 批量操作会被限流吗？

会。Twitter API 有速率限制。建议： - 控制并发数量（推荐 3-5 个并发） - 添加延迟（每批次间隔 1-2 秒） - 监听 HTTP 429 错误并实施指数退避

Q4: 如何启用详细日志？

# 运行前设置环境变量
export RUST_LOG=debug

# 或在 Python 代码中
import os
os.environ['RUST_LOG'] = 'debug'

日志规范请参考：日志文档

下一步