Upload 媒体上传模块示例

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

目录

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

前置准备

安装依赖

# 开发模式安装
maturin develop

创建客户端

from x_api_rs.web import Client

cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
twitter = await Client.create(cookies)

# 或使用代理
twitter = await Client.create(cookies, "http://127.0.0.1:7890")

上传单张图片

上传图片并获取 media_id。

import asyncio
from x_api_rs.web import Client
from pathlib import Path

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

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

    # 上传图片
    result = await twitter.upload.upload_from_bytes(
        data=image_bytes,
        category="tweet_image"  # 发帖用图片
    )

    if result.success:
        print(f"✅ 上传成功！")
        print(f"   Media ID: {result.media_id_string}")
    else:
        print(f"❌ 上传失败: {result.error_msg}")

asyncio.run(upload_single_image())

输出示例：

📖 读取图片: 102400 字节
✅ 上传成功！
   Media ID: 1234567890123456789

单次上传覆盖代理

import asyncio
from x_api_rs.web import Client
from pathlib import Path

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

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

    # 仅本次上传直连；也可使用 proxy_url="http://host:port"
    result = await twitter.upload.upload_from_bytes(
        data=image_bytes,
        category="tweet_image",
        disable_proxy=True,
    )

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

asyncio.run(upload_with_options())

上传视频

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

import asyncio
from x_api_rs.web import Client
from pathlib import Path

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

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

    # 上传视频（用于发帖）
    result = await twitter.upload.upload_from_bytes(
        data=video_bytes,
        category="amplify_video",  # 发帖视频，时长由 core 自动解析
    )

    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.get("check_after_secs"):
                print(f"   建议等待: {info['check_after_secs']} 秒")
            if info.get("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
from x_api_rs.web import Client
from pathlib import Path

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

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

    # 使用 dm_video 类型
    result = await twitter.upload.upload_from_bytes(
        data=video_bytes,
        category="dm_video",  # 私信视频，时长由 core 自动解析
    )

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

        # 使用 media_id 发送私信
        dm_result = await twitter.dm.send_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
from x_api_rs.web import Client
from pathlib import Path

async def batch_upload():
    cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
    twitter = await Client.create(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.upload_multiple_times(
        data=image_bytes,
        count=upload_count,
        category="dm_image",
    )

    # 显示结果
    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.get('media_id_string')}")
        if upload_result.get("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
from x_api_rs.web import Client
from pathlib import Path

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

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

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

    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.dm.send_batch(
        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
from x_api_rs.web import Client
from pathlib import Path

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

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

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

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

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

asyncio.run(upload_different_types())

处理上传状态

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

ProcessingInfo 状态说明

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

轮询处理状态

import asyncio
from x_api_rs.web import Client
from pathlib import Path

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

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

    # 上传视频
    result = await twitter.upload.upload_from_bytes(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']}")

        # 根据状态处理
        state = info["state"]
        if state == "succeeded":
            print("✅ 视频处理完成，可以使用")
        elif state in {"pending", "in_progress"}:
            wait_time = info.get("check_after_secs") or 5
            print(f"⏳ 视频处理中，建议等待 {wait_time} 秒后检查")
        elif state == "failed" and info.get("error"):
            print(f"❌ 处理失败: {info['error']['message']}")
            print(f"   错误码: {info['error']['code']}")

asyncio.run(upload_video_with_polling())

错误处理

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

import asyncio
from x_api_rs import XApiError
from x_api_rs.web import Client
from pathlib import Path

async def handle_upload_errors():
    cookies = "ct0=xxx; auth_token=yyy; twid=u%3D123456789"
    twitter = await Client.create(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.upload_from_bytes(image_bytes, "tweet_image")

        if result.success:
            print(f"✅ 上传成功: {result.media_id_string}")
        else:
            print(f"❌ 上传失败: {result.error_msg}")

    except XApiError as e:
        print(f"❌ X API 错误: {e}")
    except OSError 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

完整示例

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

import asyncio
import os
import sys
from pathlib import Path
from typing import List, Optional

try:
    from x_api_rs.web import Client
except ImportError:
    print("❌ 请先安装 x_api: maturin develop")
    sys.exit(1)


class UploadDemo:
    """上传演示类"""

    def __init__(self, client: Client):
        self.client = client

    async def upload_image(
        self,
        image_path: str,
        category: str = "tweet_image"
    ) -> Optional[str]:
        """上传单张图片，返回 media_id"""
        image_bytes = Path(image_path).read_bytes()
        result = await self.client.upload.upload_from_bytes(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.client.upload.upload_multiple_times(
            image_bytes, category, count
        )
        return result.media_ids

    async def upload_video(
        self,
        video_path: str,
        category: str = "amplify_video"
    ) -> Optional[str]:
        """上传视频，返回 media_id"""
        video_bytes = Path(video_path).read_bytes()
        result = await self.client.upload.upload_from_bytes(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

    client = await Client.create(cookies)
    demo = UploadDemo(client)

    # 演示上传图片
    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 API 文档
• DM 私信示例
• 批量操作示例