Upload 媒体上传模块示例¶

本文档提供媒体上传模块的完整使用示例。

目录¶

前置准备
上传单张图片
上传视频
批量上传图片
不同媒体类型
处理上传状态
错误处理

前置准备¶

安装依赖¶

# 开发模式安装
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")

上传单张图片¶

上传图片并获取 media_id。

import asyncio
import x_api
from pathlib import Path

async def upload_single_image():
    cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
    twitter = x_api.Twitter(cookies)

    # 读取图片文件
    image_path = Path("photo.jpg")
    image_bytes = image_path.read_bytes()
    print(f"📖 读取图片: {len(image_bytes)} 字节")

    # 上传图片
    result = await twitter.upload_image(
        image_bytes=image_bytes,
        media_category="tweet_image"  # 发帖用图片
    )

    if result.success:
        print(f"✅ 上传成功！")
        print(f"   Media ID: {result.media_id_string}")
        print(f"   过期时间: {result.expires_after_secs} 秒")
    else:
        print(f"❌ 上传失败: {result.error_msg}")
        print(f"   HTTP 状态码: {result.http_status}")

asyncio.run(upload_single_image())

输出示例：

📖 读取图片: 102400 字节
✅ 上传成功！
   Media ID: 1234567890123456789
   过期时间: 86400 秒

使用 UploadOptions 自定义上传¶

import asyncio
import x_api
from pathlib import Path

async def upload_with_options():
    cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
    twitter = x_api.Twitter(cookies)

    image_bytes = Path("photo.jpg").read_bytes()

    # 创建上传选项
    options = x_api.UploadOptions(
        media_category="tweet_image",
        additional_owners=["123456789", "987654321"],  # 共享给其他用户
    )

    result = await twitter.upload.upload(
        media_bytes=image_bytes,
        media_type="image/jpeg",
        options=options,
    )

    if result.success:
        print(f"✅ 上传成功: {result.media_id_string}")

asyncio.run(upload_with_options())

上传视频¶

视频上传支持分块上传和状态轮询。

import asyncio
import x_api
from pathlib import Path

async def upload_video():
    cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
    twitter = x_api.Twitter(cookies)

    # 读取视频文件
    video_path = Path("video.mp4")
    video_bytes = video_path.read_bytes()
    print(f"📹 读取视频: {len(video_bytes)} 字节")

    # 上传视频（用于发帖）
    result = await twitter.upload.video(
        video_bytes=video_bytes,
        media_category="amplify_video",  # 发帖视频
    )

    if result.success:
        print(f"✅ 视频上传成功！")
        print(f"   Media ID: {result.media_id_string}")

        # 检查处理状态
        if result.processing_info:
            info = result.processing_info
            print(f"   处理状态: {info.state}")
            if info.check_after_secs:
                print(f"   建议等待: {info.check_after_secs} 秒")
            if info.progress_percent is not None:
                print(f"   处理进度: {info.progress_percent}%")
    else:
        print(f"❌ 上传失败: {result.error_msg}")

asyncio.run(upload_video())

输出示例：

📹 读取视频: 5242880 字节
✅ 视频上传成功！
   Media ID: 1234567890123456789
   处理状态: pending
   建议等待: 5 秒

上传私信视频¶

import asyncio
import x_api
from pathlib import Path

async def upload_dm_video():
    cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
    twitter = x_api.Twitter(cookies)

    video_bytes = Path("video.mp4").read_bytes()

    # 使用 dm_video 类型
    result = await twitter.upload.video(
        video_bytes=video_bytes,
        media_category="dm_video",  # 私信视频
    )

    if result.success:
        print(f"✅ 私信视频上传成功: {result.media_id_string}")

        # 使用 media_id 发送私信
        dm_result = await twitter.send_direct_message(
            user_id="123456789",
            text="看看这个视频！",
            media_id=result.media_id_string,
        )
        print(f"   私信发送: {'成功' if dm_result.success else '失败'}")

asyncio.run(upload_dm_video())

批量上传图片¶

上传同一张图片多次，获取多个独立的 media_id。

import asyncio
import x_api
from pathlib import Path

async def batch_upload():
    cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
    twitter = x_api.Twitter(cookies)

    # 读取图片
    image_bytes = Path("photo.jpg").read_bytes()
    print(f"📖 读取图片: {len(image_bytes)} 字节")

    # 批量上传 5 次
    upload_count = 5
    print(f"\n🚀 开始批量上传 {upload_count} 次...")

    result = await twitter.upload_image_multiple_times(
        image_bytes=image_bytes,
        media_category="dm_image",
        count=upload_count,
    )

    # 显示结果
    print(f"\n📊 批量上传结果:")
    print(f"   成功: {result.success_count}/{upload_count}")
    print(f"   失败: {result.failure_count}/{upload_count}")

    print(f"\n📋 获取的 Media IDs:")
    for i, media_id in enumerate(result.media_ids, 1):
        print(f"   {i}. {media_id}")

    # 显示详细结果
    print(f"\n📝 详细上传结果:")
    for i, upload_result in enumerate(result.results, 1):
        status = "✅" if upload_result.success else "❌"
        print(f"   {i}. {status} media_id: {upload_result.media_id_string}")
        if upload_result.error_msg:
            print(f"      错误: {upload_result.error_msg}")

asyncio.run(batch_upload())

输出示例：

📖 读取图片: 102400 字节

🚀 开始批量上传 5 次...

📊 批量上传结果:
   成功: 5/5
   失败: 0/5

📋 获取的 Media IDs:
   1. 1234567890123456789
   2. 1234567890123456790
   3. 1234567890123456791
   4. 1234567890123456792
   5. 1234567890123456793

📝 详细上传结果:
   1. ✅ media_id: 1234567890123456789
   2. ✅ media_id: 1234567890123456790
   3. ✅ media_id: 1234567890123456791
   4. ✅ media_id: 1234567890123456792
   5. ✅ media_id: 1234567890123456793

批量上传用于批量私信¶

import asyncio
import x_api
from pathlib import Path

async def batch_upload_for_dm():
    cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
    twitter = x_api.Twitter(cookies)

    # 目标用户
    user_ids = ["111111111", "222222222", "333333333"]

    # 批量上传图片
    image_bytes = Path("photo.jpg").read_bytes()
    upload_result = await twitter.upload_image_multiple_times(
        image_bytes=image_bytes,
        media_category="dm_image",
        count=len(user_ids),
    )

    if upload_result.success_count != len(user_ids):
        print(f"❌ 上传不完整: {upload_result.success_count}/{len(user_ids)}")
        return

    print(f"✅ 图片上传完成，获得 {len(upload_result.media_ids)} 个 media_id")

    # 批量发送私信
    dm_result = await twitter.send_batch_direct_messages(
        user_ids=user_ids,
        text="看看这张图片！📷",
        media_ids=upload_result.media_ids,
    )

    print(f"\n📊 私信发送结果:")
    print(f"   成功: {dm_result.success_count}/{len(user_ids)}")
    print(f"   失败: {dm_result.failure_count}/{len(user_ids)}")

asyncio.run(batch_upload_for_dm())

不同媒体类型¶

根据用途选择正确的媒体类型。

媒体类型对照表¶

媒体类型	用途	说明
`tweet_image`	发帖图片	用于创建推文时附带的图片
`dm_image`	私信图片	用于私信中发送的图片
`banner_image`	个人横幅	用于更新用户的个人主页横幅
`amplify_video`	发帖视频	用于发帖时附带的视频
`dm_video`	私信视频	用于私信中发送的视频

使用示例¶

import asyncio
import x_api
from pathlib import Path

async def upload_different_types():
    cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
    twitter = x_api.Twitter(cookies)

    image_bytes = Path("photo.jpg").read_bytes()

    # 1. 发帖图片
    tweet_image = await twitter.upload_image(image_bytes, "tweet_image")
    print(f"发帖图片: {tweet_image.media_id_string}")

    # 2. 私信图片
    dm_image = await twitter.upload_image(image_bytes, "dm_image")
    print(f"私信图片: {dm_image.media_id_string}")

    # 3. 个人横幅
    banner = await twitter.upload_image(image_bytes, "banner_image")
    print(f"个人横幅: {banner.media_id_string}")

asyncio.run(upload_different_types())

处理上传状态¶

视频上传需要处理异步处理状态。

ProcessingInfo 状态说明¶

状态	说明
`pending`	等待处理
`in_progress`	处理中
`succeeded`	处理成功
`failed`	处理失败

轮询处理状态¶

import asyncio
import x_api
from pathlib import Path

async def upload_video_with_polling():
    cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
    twitter = x_api.Twitter(cookies)

    video_bytes = Path("video.mp4").read_bytes()

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

    if not result.success:
        print(f"❌ 上传失败: {result.error_msg}")
        return

    print(f"✅ 上传成功: {result.media_id_string}")

    # 检查处理状态
    if result.processing_info:
        info = result.processing_info
        print(f"📊 处理状态: {info.state}")

        # 根据状态处理
        match info.state:
            case "succeeded":
                print("✅ 视频处理完成，可以使用")
            case "pending" | "in_progress":
                wait_time = info.check_after_secs or 5
                print(f"⏳ 视频处理中，建议等待 {wait_time} 秒后检查")
                # 实际应用中可实现轮询逻辑
            case "failed":
                if info.error:
                    print(f"❌ 处理失败: {info.error.message}")
                    print(f"   错误码: {info.error.code}")

asyncio.run(upload_video_with_polling())

错误处理¶

处理上传过程中的常见错误。

import asyncio
import x_api
from pathlib import Path

async def handle_upload_errors():
    cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
    twitter = x_api.Twitter(cookies)

    image_path = Path("photo.jpg")

    # 检查文件是否存在
    if not image_path.exists():
        print(f"❌ 文件不存在: {image_path}")
        return

    # 检查文件大小
    file_size = image_path.stat().st_size
    max_size = 5 * 1024 * 1024  # 5MB 限制
    if file_size > max_size:
        print(f"❌ 文件过大: {file_size} > {max_size}")
        return

    try:
        image_bytes = image_path.read_bytes()

        result = await twitter.upload_image(image_bytes, "tweet_image")

        if result.success:
            print(f"✅ 上传成功: {result.media_id_string}")
        else:
            # 处理 API 返回的错误
            match result.http_status:
                case 400:
                    print("❌ 请求参数错误，检查图片格式")
                case 401:
                    print("❌ 认证失败，检查 cookies")
                case 413:
                    print("❌ 文件过大")
                case 429:
                    print("❌ 请求频率限制，请稍后重试")
                case _:
                    print(f"❌ 上传失败: {result.error_msg}")

    except x_api.TwitterError as e:
        print(f"❌ Twitter 错误: {e}")
    except IOError as e:
        print(f"❌ 文件读取错误: {e}")
    except Exception as e:
        print(f"❌ 未知错误: {e}")

asyncio.run(handle_upload_errors())

文件大小限制¶

类型	最大大小
图片 (JPEG/PNG/GIF)	5 MB
GIF 动图	15 MB
视频	512 MB

支持的格式¶

类型	支持格式
图片	JPEG, PNG, GIF, WEBP
视频	MP4, MOV

完整示例¶

#!/usr/bin/env python3
"""
Upload 媒体上传模块完整示例
"""

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 UploadDemo:
    """上传演示类"""

    def __init__(self, cookies: str, proxy_url: str | None = None):
        self.twitter = x_api.Twitter(cookies, proxy_url)

    async def upload_image(
        self,
        image_path: str,
        category: str = "tweet_image"
    ) -> str | None:
        """上传单张图片，返回 media_id"""
        image_bytes = Path(image_path).read_bytes()
        result = await self.twitter.upload_image(image_bytes, category)
        return result.media_id_string if result.success else None

    async def upload_images_batch(
        self,
        image_path: str,
        count: int,
        category: str = "dm_image"
    ) -> list[str]:
        """批量上传图片，返回 media_id 列表"""
        image_bytes = Path(image_path).read_bytes()
        result = await self.twitter.upload_image_multiple_times(
            image_bytes, category, count
        )
        return result.media_ids

    async def upload_video(
        self,
        video_path: str,
        category: str = "amplify_video"
    ) -> str | None:
        """上传视频，返回 media_id"""
        video_bytes = Path(video_path).read_bytes()
        result = await self.twitter.upload.video(video_bytes, category)
        return result.media_id_string if result.success else None


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 = UploadDemo(cookies)

    # 演示上传图片
    if Path("photo.jpg").exists():
        print("=== 上传图片 ===")
        media_id = await demo.upload_image("photo.jpg")
        if media_id:
            print(f"✅ 上传成功: {media_id}")
        else:
            print("❌ 上传失败")


if __name__ == "__main__":
    asyncio.run(main())

Upload 媒体上传模块示例¶

目录¶

前置准备¶

安装依赖¶

创建客户端¶

上传单张图片¶

使用 UploadOptions 自定义上传¶

上传视频¶

上传私信视频¶

批量上传图片¶

批量上传用于批量私信¶

不同媒体类型¶

媒体类型对照表¶

使用示例¶

处理上传状态¶

ProcessingInfo 状态说明¶

轮询处理状态¶

错误处理¶

文件大小限制¶

支持的格式¶

完整示例¶

相关链接¶