Posts 模块 API 文档

Posts 模块提供完整的 Twitter 帖子管理功能，支持发帖、删帖、点赞、转发等操作。

目录

• 模块概述
• PostsClient 类
• 类型定义
• 使用示例
• 最佳实践
• 常见问题

模块概述

Posts 模块支持两种使用方式：

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

from x_api_rs import Twitter

client = Twitter(cookies)
result = await client.posts.create_tweet(text="Hello World!")

方式 2: 独立创建 PostsClient

from x_api_rs.posts import PostsClient

posts_client = PostsClient(cookies, proxy_url="http://proxy:8080")
result = await posts_client.create_tweet(text="Hello World!")

功能列表

功能	方法	说明
发帖	create_tweet()	发布新帖子（支持文字、图片、视频、回复、引用）
删帖	delete_tweet()	删除已发布的帖子
点赞	favorite_tweet()	点赞帖子
取消点赞	unfavorite_tweet()	取消点赞
转发	create_retweet()	转发帖子
取消转发	delete_retweet()	取消转发
获取帖子	get_tweets()	获取用户帖子列表
获取点赞	get_likes()	获取用户点赞列表

PostsClient 类

构造函数

PostsClient(
    cookies: str,
    proxy_url: str | None = None
)

创建新的 Posts 客户端实例。

参数:

• cookies (str): Twitter 账号的 cookies 字符串
• proxy_url (str | None): 可选的代理服务器 URL

返回: PostsClient 实例

异常: RuntimeError - 初始化失败

示例:

# 基础使用
client = PostsClient(cookies)

# 使用代理
client = PostsClient(cookies, proxy_url="http://proxy:8080")

---

create_tweet

async def create_tweet(
    text: str = "",
    media_ids: list[str] | None = None,
    reply_to_tweet_id: str | None = None,
    attachment_url: str | None = None,
    entity_id: str | None = None
) -> TweetResult

创建新帖子。

参数:

• text (str): 帖子文本内容（最多 280 字符，付费用户可更长）
• media_ids (list[str] | None): 可选的媒体 ID 列表（来自 upload.image 返回）
• reply_to_tweet_id (str | None): 可选的回复目标帖子 ID（用于回复）
• attachment_url (str | None): 可选的引用帖子 URL（用于引用转发）
• entity_id (str | None): 可选的社区 ID

返回: TweetResult 对象

异常: RuntimeError - 发帖失败

示例:

# 发布纯文本帖子
result = await client.posts.create_tweet(text="Hello World!")
if result.success:
    print(f"发帖成功！Tweet ID: {result.tweet_id}")

# 发布带图片的帖子
upload_result = await client.upload.image(image_bytes, "tweet_image")
result = await client.posts.create_tweet(
    text="分享一张照片 📸",
    media_ids=[upload_result.media_id_string]
)

# 回复帖子
result = await client.posts.create_tweet(
    text="这是我的回复",
    reply_to_tweet_id="1234567890"
)

# 引用转发
result = await client.posts.create_tweet(
    text="值得一看！",
    attachment_url="https://twitter.com/user/status/1234567890"
)

---

create_tweet_with_params

async def create_tweet_with_params(
    params: CreateTweetParams
) -> TweetResult

使用参数对象创建帖子（适用于复杂参数场景）。

参数:

• params (CreateTweetParams): 发帖参数对象

返回: TweetResult 对象

示例:

from x_api_rs.posts import CreateTweetParams

params = CreateTweetParams(
    text="Hello World!",
    media_ids=["123456"],
    reply_to_tweet_id="789012"
)
result = await client.posts.create_tweet_with_params(params)

---

delete_tweet

async def delete_tweet(
    tweet_id: str
) -> DeleteTweetResult

删除帖子。

参数:

• tweet_id (str): 要删除的帖子 ID

返回: DeleteTweetResult 对象

异常: RuntimeError - 删帖失败

示例:

result = await client.posts.delete_tweet("1234567890")
if result.success:
    print("删除成功")
else:
    print(f"删除失败: {result.error_msg}")

---

favorite_tweet

async def favorite_tweet(
    tweet_id: str
) -> LikeResult

点赞帖子。

参数:

• tweet_id (str): 要点赞的帖子 ID

返回: LikeResult 对象

异常: RuntimeError - 点赞失败

示例:

result = await client.posts.favorite_tweet("1234567890")
if result.success:
    print("点赞成功 ❤️")

---

unfavorite_tweet

async def unfavorite_tweet(
    tweet_id: str
) -> LikeResult

取消点赞。

参数:

• tweet_id (str): 要取消点赞的帖子 ID

返回: LikeResult 对象

异常: RuntimeError - 取消点赞失败

示例:

result = await client.posts.unfavorite_tweet("1234567890")
if result.success:
    print("取消点赞成功")

---

create_retweet

async def create_retweet(
    tweet_id: str
) -> RetweetResult

转发帖子。

参数:

• tweet_id (str): 要转发的帖子 ID

返回: RetweetResult 对象

异常: RuntimeError - 转发失败

示例:

result = await client.posts.create_retweet("1234567890")
if result.success:
    print(f"转发成功！Retweet ID: {result.retweet_id}")

---

delete_retweet

async def delete_retweet(
    tweet_id: str
) -> DeleteTweetResult

取消转发。

参数:

• tweet_id (str): 原帖的 ID（不是 retweet_id）

返回: DeleteTweetResult 对象

异常: RuntimeError - 取消转发失败

示例:

result = await client.posts.delete_retweet("1234567890")
if result.success:
    print("取消转发成功")

---

get_tweets

async def get_tweets(
    user_id: str,
    cursor: str | None = None
) -> GetTweetsResult

获取用户帖子列表。

参数:

• user_id (str): 用户 ID
• cursor (str | None): 可选的分页游标

返回: GetTweetsResult 对象

异常: RuntimeError - 获取失败

示例:

# 获取第一页
result = await client.posts.get_tweets("44196397")
for tweet in result.tweets:
    print(f"{tweet.author_screen_name}: {tweet.text}")

# 获取下一页
if result.has_more:
    next_page = await client.posts.get_tweets("44196397", result.next_cursor)

---

get_likes

async def get_likes(
    user_id: str | None = None,
    cursor: str | None = None
) -> GetLikesResult

获取用户点赞列表。

参数:

• user_id (str | None): 可选的用户 ID，默认为当前登录用户
• cursor (str | None): 可选的分页游标

返回: GetLikesResult 对象

异常: RuntimeError - 获取失败

示例:

# 获取自己的点赞列表
result = await client.posts.get_likes()

# 获取指定用户的点赞列表
result = await client.posts.get_likes("44196397")

for tweet in result.tweets:
    print(f"点赞了: {tweet.text[:50]}...")

---

类型定义

CreateTweetParams

发帖参数对象。

属性:

• text (str): 帖子文本内容
• media_ids (list[str] | None): 媒体 ID 列表
• reply_to_tweet_id (str | None): 回复目标帖子 ID
• attachment_url (str | None): 引用转发 URL
• entity_id (str | None): 社区 ID

示例:

from x_api_rs.posts import CreateTweetParams

params = CreateTweetParams(
    text="Hello World!",
    media_ids=["123456", "789012"],
    reply_to_tweet_id="111222333"
)

---

TweetResult

发帖结果。

属性:

• success (bool): 是否发帖成功
• tweet_id (str | None): 帖子 ID（成功时有值）
• error_msg (str): 错误消息（失败时有值）
• http_status (int): HTTP 状态码

示例:

result = await client.posts.create_tweet(text="Hello!")

print(result.success)      # True
print(result.tweet_id)     # "1234567890123456789"
print(result.http_status)  # 200

---

DeleteTweetResult

删帖结果。

属性:

• success (bool): 是否删除成功
• tweet_id (str): 帖子 ID
• error_msg (str): 错误消息
• http_status (int): HTTP 状态码

---

LikeResult

点赞/取消点赞结果。

属性:

• success (bool): 是否操作成功
• tweet_id (str): 帖子 ID
• error_msg (str): 错误消息
• http_status (int): HTTP 状态码

---

RetweetResult

转发结果。

属性:

• success (bool): 是否转发成功
• source_tweet_id (str): 原帖 ID
• retweet_id (str | None): 转发后的帖子 ID（成功时有值）
• error_msg (str): 错误消息
• http_status (int): HTTP 状态码

---

TweetInfo

帖子详细信息。

属性:

• tweet_id (str): 帖子 ID
• text (str | None): 帖子文本内容
• author_id (str | None): 作者 ID
• author_screen_name (str | None): 作者用户名
• created_at (str | None): 创建时间
• retweet_count (int | None): 转发数
• favorite_count (int | None): 点赞数
• reply_count (int | None): 回复数
• quote_count (int | None): 引用数
• favorited (bool): 是否已点赞
• retweeted (bool): 是否已转发
• media_urls (list[str] | None): 媒体 URL 列表

示例:

result = await client.posts.get_tweets("44196397")

for tweet in result.tweets:
    print(f"ID: {tweet.tweet_id}")
    print(f"作者: @{tweet.author_screen_name}")
    print(f"内容: {tweet.text}")
    print(f"点赞: {tweet.favorite_count}, 转发: {tweet.retweet_count}")
    print(f"已点赞: {tweet.favorited}, 已转发: {tweet.retweeted}")
    print("---")

---

GetTweetsResult

获取帖子列表结果。

属性:

• success (bool): 是否获取成功
• tweets (list[TweetInfo]): 帖子列表
• next_cursor (str | None): 下一页游标
• has_more (bool): 是否有更多数据
• error_msg (str): 错误消息
• http_status (int): HTTP 状态码

方法:

• __len__() -> int: 返回帖子数量

示例:

result = await client.posts.get_tweets("44196397")

print(f"获取到 {len(result)} 条帖子")
print(f"是否有更多: {result.has_more}")

if result.has_more:
    next_result = await client.posts.get_tweets("44196397", result.next_cursor)

---

GetLikesResult

获取点赞列表结果（结构与 GetTweetsResult 相同）。

属性:

• success (bool): 是否获取成功
• tweets (list[TweetInfo]): 点赞的帖子列表
• next_cursor (str | None): 下一页游标
• has_more (bool): 是否有更多数据
• error_msg (str): 错误消息
• http_status (int): HTTP 状态码

---

使用示例

基础发帖

import asyncio
from x_api_rs import Twitter

async def main():
    client = Twitter(cookies)

    # 发布纯文本帖子
    result = await client.posts.create_tweet(text="Hello Twitter! 👋")

    if result.success:
        print(f"发帖成功！Tweet ID: {result.tweet_id}")
    else:
        print(f"发帖失败: {result.error_msg}")

asyncio.run(main())

发布带图片的帖子

async def tweet_with_image():
    client = Twitter(cookies)

    # 1. 上传图片
    with open("photo.jpg", "rb") as f:
        image_bytes = f.read()

    upload_result = await client.upload.image(image_bytes, "tweet_image")
    if not upload_result.success:
        print(f"上传失败: {upload_result.error_msg}")
        return

    # 2. 发帖
    result = await client.posts.create_tweet(
        text="分享一张照片 📸",
        media_ids=[upload_result.media_id_string]
    )

    if result.success:
        print(f"发帖成功！Tweet ID: {result.tweet_id}")

发布带多张图片的帖子

async def tweet_with_multiple_images():
    client = Twitter(cookies)

    # 上传多张图片
    image_files = ["photo1.jpg", "photo2.jpg", "photo3.jpg"]
    media_ids = []

    for file in image_files:
        with open(file, "rb") as f:
            image_bytes = f.read()
        result = await client.upload.image(image_bytes, "tweet_image")
        if result.success:
            media_ids.append(result.media_id_string)

    # 发帖（最多 4 张图片）
    result = await client.posts.create_tweet(
        text="多图分享 🖼️",
        media_ids=media_ids[:4]
    )

回复帖子

async def reply_to_tweet():
    client = Twitter(cookies)

    original_tweet_id = "1234567890"

    result = await client.posts.create_tweet(
        text="这是我的回复内容",
        reply_to_tweet_id=original_tweet_id
    )

    if result.success:
        print(f"回复成功！Tweet ID: {result.tweet_id}")

引用转发

async def quote_tweet():
    client = Twitter(cookies)

    # 引用转发
    result = await client.posts.create_tweet(
        text="这条推文很有意思！",
        attachment_url="https://twitter.com/elonmusk/status/1234567890"
    )

    if result.success:
        print(f"引用转发成功！Tweet ID: {result.tweet_id}")

批量点赞

async def batch_like():
    client = Twitter(cookies)

    tweet_ids = ["111", "222", "333", "444", "555"]

    for tweet_id in tweet_ids:
        result = await client.posts.favorite_tweet(tweet_id)
        if result.success:
            print(f"✓ 点赞成功: {tweet_id}")
        else:
            print(f"✗ 点赞失败: {tweet_id} - {result.error_msg}")
        await asyncio.sleep(1)  # 避免限流

获取用户时间线

async def get_user_timeline():
    client = Twitter(cookies)

    user_id = "44196397"  # Elon Musk 的用户 ID
    all_tweets = []

    # 获取第一页
    result = await client.posts.get_tweets(user_id)
    if result.success:
        all_tweets.extend(result.tweets)

    # 获取更多页（最多 5 页）
    page = 1
    while result.has_more and page < 5:
        result = await client.posts.get_tweets(user_id, result.next_cursor)
        if result.success:
            all_tweets.extend(result.tweets)
        page += 1
        await asyncio.sleep(1)

    print(f"共获取 {len(all_tweets)} 条帖子")

    for tweet in all_tweets[:10]:
        print(f"- {tweet.text[:50]}...")

互动操作组合

async def interact_with_tweet():
    client = Twitter(cookies)

    tweet_id = "1234567890"

    # 点赞
    like_result = await client.posts.favorite_tweet(tweet_id)
    if like_result.success:
        print("点赞成功")

    # 转发
    retweet_result = await client.posts.create_retweet(tweet_id)
    if retweet_result.success:
        print(f"转发成功，Retweet ID: {retweet_result.retweet_id}")

    # 回复
    reply_result = await client.posts.create_tweet(
        text="精彩的内容！",
        reply_to_tweet_id=tweet_id
    )
    if reply_result.success:
        print(f"回复成功，Reply ID: {reply_result.tweet_id}")

---

最佳实践

1. 验证操作结果

result = await client.posts.create_tweet(text="Hello!")

# 始终检查 success 状态
if result.success:
    # 成功处理
    print(f"Tweet ID: {result.tweet_id}")
else:
    # 失败处理
    print(f"Error: {result.error_msg}")
    print(f"HTTP Status: {result.http_status}")

2. 控制操作频率

import asyncio

# 批量操作时添加延迟
for tweet_id in tweet_ids:
    await client.posts.favorite_tweet(tweet_id)
    await asyncio.sleep(1)  # 每次操作后等待 1 秒

# 或使用批量操作时分批处理
batch_size = 10
for i in range(0, len(tweet_ids), batch_size):
    batch = tweet_ids[i:i+batch_size]
    tasks = [client.posts.favorite_tweet(tid) for tid in batch]
    await asyncio.gather(*tasks)
    await asyncio.sleep(5)  # 每批次后等待 5 秒

3. 正确使用媒体

# 发帖用 tweet_image
upload_result = await client.upload.image(image_bytes, "tweet_image")
await client.posts.create_tweet(
    text="分享",
    media_ids=[upload_result.media_id_string]
)

# 多张图片时，确保都是 tweet_image 类别
media_ids = []
for img in images:
    result = await client.upload.image(img, "tweet_image")
    if result.success:
        media_ids.append(result.media_id_string)

# 最多 4 张图片
await client.posts.create_tweet(text="多图", media_ids=media_ids[:4])

4. 分页获取数据

async def get_all_tweets(user_id: str, max_pages: int = 10):
    """获取用户所有帖子（带分页限制）"""
    client = Twitter(cookies)
    all_tweets = []
    cursor = None

    for _ in range(max_pages):
        result = await client.posts.get_tweets(user_id, cursor)
        if not result.success:
            break

        all_tweets.extend(result.tweets)

        if not result.has_more:
            break

        cursor = result.next_cursor
        await asyncio.sleep(1)  # 避免限流

    return all_tweets

5. 错误处理和重试

import asyncio

async def create_tweet_with_retry(text: str, max_retries: int = 3):
    """带重试的发帖"""
    client = Twitter(cookies)

    for attempt in range(max_retries):
        try:
            result = await client.posts.create_tweet(text=text)
            if result.success:
                return result

            # 如果是限流错误，等待后重试
            if result.http_status == 429:
                wait_time = 2 ** attempt * 10  # 指数退避
                print(f"限流，等待 {wait_time} 秒...")
                await asyncio.sleep(wait_time)
                continue

            # 其他错误直接返回
            return result

        except Exception as e:
            if attempt < max_retries - 1:
                await asyncio.sleep(2 ** attempt)
            else:
                raise

    return result

---

常见问题

Q1: 帖子文本长度限制是多少？

普通用户最多 280 字符，Twitter Blue 用户最多 25000 字符。中文、日文、韩文等字符按 1 个字符计算。

Q2: 一条帖子最多可以附加多少张图片？

最多 4 张图片，或 1 个视频/GIF。图片和视频不能混合。

Q3: 如何发布带视频的帖子？

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

# 2. 等待处理完成
if result.processing_info and result.processing_info.state.is_pending():
    await asyncio.sleep(result.processing_info.check_after_secs or 5)

# 3. 发帖
await client.posts.create_tweet(
    text="分享一个视频",
    media_ids=[result.media_id_string]
)

Q4: 删除帖子后能恢复吗？

不能。删除操作是不可逆的，请谨慎使用。

Q5: 取消转发时应该传什么 ID？

传原帖的 tweet_id，不是转发后获得的 retweet_id。

# 转发
retweet_result = await client.posts.create_retweet("original_tweet_id")

# 取消转发（使用原帖 ID）
await client.posts.delete_retweet("original_tweet_id")

Q6: 如何判断帖子是否已点赞/转发？

通过 get_tweets() 获取的 TweetInfo 对象包含 favorited 和 retweeted 属性：

result = await client.posts.get_tweets(user_id)

for tweet in result.tweets:
    if tweet.favorited:
        print(f"{tweet.tweet_id} 已点赞")
    if tweet.retweeted:
        print(f"{tweet.tweet_id} 已转发")

Q7: 发帖时出现 "duplicate content" 错误？

Twitter 不允许短时间内发布重复内容。解决方案：

• 修改帖子内容（添加时间戳、emoji 等）
• 等待一段时间后再发

import time

# 添加时间戳避免重复
text = f"Hello! {int(time.time())}"
await client.posts.create_tweet(text=text)

---

下一步

• 查看 Upload 模块文档 了解如何上传媒体
• 查看 DM 模块文档 了解私信功能
• 查看 示例代码 了解更多用法