Posts 帖子模块示例¶
本文档提供 Posts 帖子模块的完整使用示例。
目录¶
前置准备¶
安装依赖¶
创建客户端¶
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 create_text_tweet():
cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
twitter = x_api.Twitter(cookies)
# 发布纯文本帖子
result = await twitter.posts.create_tweet(text="Hello World! 👋")
if result.success:
print(f"✅ 发帖成功！")
print(f" Tweet ID: {result.tweet_id}")
else:
print(f"❌ 发帖失败: {result.error_msg}")
print(f" HTTP 状态码: {result.http_status}")
asyncio.run(create_text_tweet())
输出示例：
发布长文本帖子¶
Twitter Blue 用户可以发布长文本（最多 25000 字符）。
import asyncio
import x_api
async def create_long_tweet():
cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
twitter = x_api.Twitter(cookies)
long_text = """
这是一篇长文章的开头...
第一段内容：详细的说明和解释。
第二段内容：更多的信息和数据。
第三段内容：总结和结论。
感谢阅读！🙏
"""
result = await twitter.posts.create_tweet(text=long_text.strip())
if result.success:
print(f"✅ 长文发布成功: {result.tweet_id}")
asyncio.run(create_long_tweet())
发布带图片的帖子¶
先上传图片，再发布带图片的帖子。
import asyncio
import x_api
from pathlib import Path
async def create_tweet_with_image():
cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
twitter = x_api.Twitter(cookies)
# 步骤1：读取并上传图片
image_bytes = Path("photo.jpg").read_bytes()
print(f"📖 读取图片: {len(image_bytes)} 字节")
upload_result = await twitter.upload.image(
image_bytes=image_bytes,
media_category="tweet_image" # 发帖用图片
)
if not upload_result.success:
print(f"❌ 图片上传失败: {upload_result.error_msg}")
return
print(f"✅ 图片上传成功: {upload_result.media_id_string}")
# 步骤2：发布带图片的帖子
result = await twitter.posts.create_tweet(
text="分享一张照片 📸",
media_ids=[upload_result.media_id_string]
)
if result.success:
print(f"✅ 发帖成功: {result.tweet_id}")
else:
print(f"❌ 发帖失败: {result.error_msg}")
asyncio.run(create_tweet_with_image())
输出示例：
发布多图帖子¶
发布包含多张图片的帖子（最多 4 张）。
import asyncio
import x_api
from pathlib import Path
async def create_tweet_with_multiple_images():
cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
twitter = x_api.Twitter(cookies)
# 图片文件列表
image_files = ["photo1.jpg", "photo2.jpg", "photo3.jpg", "photo4.jpg"]
media_ids = []
print("📤 上传图片...")
# 上传所有图片
for i, file in enumerate(image_files, 1):
image_path = Path(file)
if not image_path.exists():
print(f" ⚠️ 文件不存在: {file}")
continue
image_bytes = image_path.read_bytes()
result = await twitter.upload.image(image_bytes, "tweet_image")
if result.success:
media_ids.append(result.media_id_string)
print(f" ✅ {i}. {file} -> {result.media_id_string}")
else:
print(f" ❌ {i}. {file} 上传失败: {result.error_msg}")
if not media_ids:
print("❌ 没有成功上传的图片")
return
# 发布帖子（最多 4 张图片）
print(f"\n📝 发布帖子（{len(media_ids)} 张图片）...")
result = await twitter.posts.create_tweet(
text="多图分享 🖼️",
media_ids=media_ids[:4] # 最多 4 张
)
if result.success:
print(f"✅ 发帖成功: {result.tweet_id}")
else:
print(f"❌ 发帖失败: {result.error_msg}")
asyncio.run(create_tweet_with_multiple_images())
输出示例：
📤 上传图片...
✅ 1. photo1.jpg -> 1234567890123456789
✅ 2. photo2.jpg -> 1234567890123456790
✅ 3. photo3.jpg -> 1234567890123456791
✅ 4. photo4.jpg -> 1234567890123456792
📝 发布帖子（4 张图片）...
✅ 发帖成功: 1234567890123456800
回复帖子¶
回复其他用户的帖子。
import asyncio
import x_api
async def reply_to_tweet():
cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
twitter = x_api.Twitter(cookies)
# 要回复的原帖 ID
original_tweet_id = "1234567890123456789"
result = await twitter.posts.create_tweet(
text="这是我的回复！感谢分享 🙏",
reply_to_tweet_id=original_tweet_id
)
if result.success:
print(f"✅ 回复成功！")
print(f" 回复 ID: {result.tweet_id}")
print(f" 原帖 ID: {original_tweet_id}")
else:
print(f"❌ 回复失败: {result.error_msg}")
asyncio.run(reply_to_tweet())
带图片的回复¶
import asyncio
import x_api
from pathlib import Path
async def reply_with_image():
cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
twitter = x_api.Twitter(cookies)
original_tweet_id = "1234567890123456789"
# 上传图片
image_bytes = Path("reply_image.jpg").read_bytes()
upload_result = await twitter.upload.image(image_bytes, "tweet_image")
if not upload_result.success:
print(f"❌ 图片上传失败")
return
# 带图片回复
result = await twitter.posts.create_tweet(
text="看看这个！",
reply_to_tweet_id=original_tweet_id,
media_ids=[upload_result.media_id_string]
)
if result.success:
print(f"✅ 带图回复成功: {result.tweet_id}")
asyncio.run(reply_with_image())
引用转发¶
引用其他用户的帖子并添加评论。
import asyncio
import x_api
async def quote_tweet():
cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
twitter = x_api.Twitter(cookies)
# 要引用的帖子 URL
tweet_url = "https://twitter.com/elonmusk/status/1234567890123456789"
result = await twitter.posts.create_tweet(
text="这条推文非常有意思！值得一看 👀",
attachment_url=tweet_url
)
if result.success:
print(f"✅ 引用转发成功！")
print(f" Tweet ID: {result.tweet_id}")
else:
print(f"❌ 引用转发失败: {result.error_msg}")
asyncio.run(quote_tweet())
点赞和取消点赞¶
对帖子进行点赞和取消点赞操作。
点赞帖子¶
import asyncio
import x_api
async def like_tweet():
cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
twitter = x_api.Twitter(cookies)
tweet_id = "1234567890123456789"
result = await twitter.posts.favorite_tweet(tweet_id)
if result.success:
print(f"✅ 点赞成功 ❤️")
print(f" Tweet ID: {result.tweet_id}")
else:
print(f"❌ 点赞失败: {result.error_msg}")
asyncio.run(like_tweet())
取消点赞¶
import asyncio
import x_api
async def unlike_tweet():
cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
twitter = x_api.Twitter(cookies)
tweet_id = "1234567890123456789"
result = await twitter.posts.unfavorite_tweet(tweet_id)
if result.success:
print(f"✅ 取消点赞成功")
else:
print(f"❌ 取消点赞失败: {result.error_msg}")
asyncio.run(unlike_tweet())
批量点赞¶
import asyncio
import x_api
async def batch_like():
cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
twitter = x_api.Twitter(cookies)
tweet_ids = [
"1111111111111111111",
"2222222222222222222",
"3333333333333333333",
"4444444444444444444",
"5555555555555555555",
]
success_count = 0
failure_count = 0
print(f"🚀 批量点赞 {len(tweet_ids)} 条帖子...")
for i, tweet_id in enumerate(tweet_ids, 1):
result = await twitter.posts.favorite_tweet(tweet_id)
if result.success:
print(f" ✅ {i}. {tweet_id}")
success_count += 1
else:
print(f" ❌ {i}. {tweet_id} - {result.error_msg}")
failure_count += 1
# 避免限流
await asyncio.sleep(1)
print(f"\n📊 结果: {success_count} 成功, {failure_count} 失败")
asyncio.run(batch_like())
转发和取消转发¶
转发帖子和取消转发。
转发帖子¶
import asyncio
import x_api
async def retweet():
cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
twitter = x_api.Twitter(cookies)
tweet_id = "1234567890123456789"
result = await twitter.posts.create_retweet(tweet_id)
if result.success:
print(f"✅ 转发成功 🔁")
print(f" 原帖 ID: {result.source_tweet_id}")
print(f" 转发 ID: {result.retweet_id}")
else:
print(f"❌ 转发失败: {result.error_msg}")
asyncio.run(retweet())
取消转发¶
import asyncio
import x_api
async def unretweet():
cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
twitter = x_api.Twitter(cookies)
# 注意：使用原帖 ID，不是 retweet_id
original_tweet_id = "1234567890123456789"
result = await twitter.posts.delete_retweet(original_tweet_id)
if result.success:
print(f"✅ 取消转发成功")
else:
print(f"❌ 取消转发失败: {result.error_msg}")
asyncio.run(unretweet())
删除帖子¶
删除自己发布的帖子。
import asyncio
import x_api
async def delete_tweet():
cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
twitter = x_api.Twitter(cookies)
tweet_id = "1234567890123456789"
# 确认删除（实际应用中应该添加确认逻辑）
print(f"⚠️ 即将删除帖子: {tweet_id}")
result = await twitter.posts.delete_tweet(tweet_id)
if result.success:
print(f"✅ 帖子已删除")
else:
print(f"❌ 删除失败: {result.error_msg}")
asyncio.run(delete_tweet())
获取用户帖子¶
获取指定用户的帖子列表。
获取第一页¶
import asyncio
import x_api
async def get_user_tweets():
cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
twitter = x_api.Twitter(cookies)
# Elon Musk 的用户 ID
user_id = "44196397"
result = await twitter.posts.get_tweets(user_id)
if result.success:
print(f"📋 获取到 {len(result)} 条帖子")
print(f" 是否有更多: {result.has_more}")
for i, tweet in enumerate(result.tweets[:5], 1):
print(f"\n--- 帖子 {i} ---")
print(f"ID: {tweet.tweet_id}")
print(f"作者: @{tweet.author_screen_name}")
print(f"内容: {tweet.text[:100]}..." if tweet.text and len(tweet.text) > 100 else f"内容: {tweet.text}")
print(f"点赞: {tweet.favorite_count}, 转发: {tweet.retweet_count}")
else:
print(f"❌ 获取失败: {result.error_msg}")
asyncio.run(get_user_tweets())
分页获取所有帖子¶
import asyncio
import x_api
async def get_all_user_tweets():
cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
twitter = x_api.Twitter(cookies)
user_id = "44196397"
all_tweets = []
cursor = None
max_pages = 5 # 限制最大页数
print(f"📥 获取用户 {user_id} 的帖子...")
for page in range(max_pages):
result = await twitter.posts.get_tweets(user_id, cursor)
if not result.success:
print(f"❌ 第 {page + 1} 页获取失败: {result.error_msg}")
break
all_tweets.extend(result.tweets)
print(f" 第 {page + 1} 页: {len(result.tweets)} 条")
if not result.has_more:
print(" 已获取全部帖子")
break
cursor = result.next_cursor
await asyncio.sleep(1) # 避免限流
print(f"\n📊 共获取 {len(all_tweets)} 条帖子")
# 显示统计信息
if all_tweets:
total_likes = sum(t.favorite_count or 0 for t in all_tweets)
total_retweets = sum(t.retweet_count or 0 for t in all_tweets)
print(f" 总点赞: {total_likes:,}")
print(f" 总转发: {total_retweets:,}")
asyncio.run(get_all_user_tweets())
输出示例：
📥 获取用户 44196397 的帖子...
第 1 页: 20 条
第 2 页: 20 条
第 3 页: 20 条
第 4 页: 15 条
已获取全部帖子
📊 共获取 75 条帖子
总点赞: 1,234,567
总转发: 234,567
获取用户点赞¶
获取用户点赞过的帖子列表。
import asyncio
import x_api
async def get_user_likes():
cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
twitter = x_api.Twitter(cookies)
# 获取自己的点赞列表
result = await twitter.posts.get_likes()
if result.success:
print(f"❤️ 获取到 {len(result)} 条点赞")
for i, tweet in enumerate(result.tweets[:5], 1):
print(f"\n{i}. @{tweet.author_screen_name}")
text_preview = tweet.text[:80] + "..." if tweet.text and len(tweet.text) > 80 else tweet.text
print(f" {text_preview}")
else:
print(f"❌ 获取失败: {result.error_msg}")
asyncio.run(get_user_likes())
获取指定用户的点赞¶
import asyncio
import x_api
async def get_other_user_likes():
cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
twitter = x_api.Twitter(cookies)
# 获取其他用户的点赞列表
user_id = "44196397"
result = await twitter.posts.get_likes(user_id)
if result.success:
print(f"❤️ 用户 {user_id} 点赞了 {len(result)} 条帖子")
for tweet in result.tweets[:3]:
print(f" - {tweet.text[:50]}...")
asyncio.run(get_other_user_likes())
错误处理¶
处理帖子操作中的常见错误。
import asyncio
import x_api
async def handle_post_errors():
cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
twitter = x_api.Twitter(cookies)
try:
result = await twitter.posts.create_tweet(text="测试帖子")
if result.success:
print(f"✅ 发帖成功: {result.tweet_id}")
else:
# 处理 API 返回的错误
match result.http_status:
case 400:
print("❌ 请求参数错误")
case 401:
print("❌ 认证失败，请检查 cookies")
case 403:
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_post_errors())
带重试的发帖¶
import asyncio
import x_api
async def create_tweet_with_retry(text: str, max_retries: int = 3):
"""带重试机制的发帖"""
cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
twitter = x_api.Twitter(cookies)
for attempt in range(max_retries):
result = await twitter.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
return result
async def main():
result = await create_tweet_with_retry("Hello with retry!")
if result.success:
print(f"✅ 发帖成功: {result.tweet_id}")
else:
print(f"❌ 最终失败: {result.error_msg}")
asyncio.run(main())
常见错误码¶
| HTTP 状态码 | 说明 | 处理建议 |
|---|---|---|
| 400 | 请求参数错误 | 检查文本长度、media_id 格式 |
| 401 | 认证失败 | 检查 cookies 是否过期 |
| 403 | 权限不足 | 账号可能被限制 |
| 404 | 帖子不存在 | 检查 tweet_id 是否正确 |
| 429 | 请求频率限制 | 降低请求频率或等待 |
| 500 | 服务器错误 | 稍后重试 |
完整示例¶
#!/usr/bin/env python3
"""
Posts 帖子模块完整示例
"""
import asyncio
import os
import sys
from pathlib import Path
try:
import x_api
except ImportError:
print("❌ 请先安装 x_api: maturin develop")
sys.exit(1)
class PostsDemo:
"""帖子演示类"""
def __init__(self, cookies: str, proxy_url: str | None = None):
self.twitter = x_api.Twitter(cookies, proxy_url)
async def create_tweet(self, text: str) -> str | None:
"""发布帖子，返回 tweet_id"""
result = await self.twitter.posts.create_tweet(text=text)
return result.tweet_id if result.success else None
async def create_tweet_with_image(
self,
text: str,
image_path: str
) -> str | None:
"""发布带图片的帖子"""
# 上传图片
image_bytes = Path(image_path).read_bytes()
upload = await self.twitter.upload.image(image_bytes, "tweet_image")
if not upload.success:
return None
# 发帖
result = await self.twitter.posts.create_tweet(
text=text,
media_ids=[upload.media_id_string]
)
return result.tweet_id if result.success else None
async def like_and_retweet(self, tweet_id: str) -> tuple[bool, bool]:
"""点赞并转发，返回 (点赞成功, 转发成功)"""
like_result = await self.twitter.posts.favorite_tweet(tweet_id)
retweet_result = await self.twitter.posts.create_retweet(tweet_id)
return like_result.success, retweet_result.success
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 = PostsDemo(cookies)
# 演示发帖
print("=== 发布帖子 ===")
tweet_id = await demo.create_tweet("Hello from Python! 🐍")
if tweet_id:
print(f"✅ 发帖成功: {tweet_id}")
else:
print("❌ 发帖失败")
if __name__ == "__main__":
asyncio.run(main())