DM 私信模块示例

本文档提供私信（Direct Message）模块的完整使用示例。

目录

• 前置准备
• 发送单条私信
• 发送带图片的私信
• 批量发送相同内容
• 批量发送自定义文案
• 批量发送带图片的自定义私信
• 错误处理

前置准备

安装依赖

# 开发模式安装
maturin develop

# 或生产构建
maturin build --release
pip install target/wheels/x_api-*.whl

配置认证

import os

# 方式1：环境变量
cookies = os.getenv("TWITTER_COOKIES")

# 方式2：文件读取
from pathlib import Path
cookies = Path("cookies.txt").read_text().strip()

# cookies 格式
# ct0=xxx; auth_token=yyy; twid=u%3D123456789

创建客户端

import x_api

# 不使用代理
twitter = x_api.Twitter(cookies)

# 使用代理
twitter = x_api.Twitter(cookies, "http://127.0.0.1:7890")

发送单条私信

向单个用户发送纯文本私信。

import asyncio
import x_api

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

    # 发送纯文本私信
    result = await twitter.send_direct_message(
        user_id="123456789",
        text="你好！这是一条测试私信。"
    )

    if result.success:
        print(f"✅ 发送成功！")
        print(f"   Event ID: {result.event_id}")
        print(f"   消息内容: {result.message}")
    else:
        print(f"❌ 发送失败: {result.error_msg}")
        print(f"   HTTP 状态码: {result.http_status}")

asyncio.run(send_single_message())

输出示例：

✅ 发送成功！
   Event ID: 1234567890123456789
   消息内容: 你好！这是一条测试私信。

发送带图片的私信

先上传图片获取 media_id，再发送带图片的私信。

import asyncio
import x_api
from pathlib import Path

async def send_message_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)} 字节")

    # 步骤2：上传图片
    upload_result = await twitter.upload_image(
        image_bytes=image_bytes,
        media_category="dm_image"  # 私信图片类型
    )

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

    print(f"✅ 图片上传成功: {upload_result.media_id_string}")

    # 步骤3：发送带图片的私信
    result = await twitter.send_direct_message(
        user_id="123456789",
        text="看看这张图片！📷",
        media_id=upload_result.media_id_string
    )

    if result.success:
        print(f"✅ 私信发送成功！")
        print(f"   Event ID: {result.event_id}")
        print(f"   Media ID: {result.media_id}")
    else:
        print(f"❌ 发送失败: {result.error_msg}")

asyncio.run(send_message_with_image())

输出示例：

📖 读取图片: 102400 字节
✅ 图片上传成功: 1234567890123456789
✅ 私信发送成功！
   Event ID: 1234567890123456790
   Media ID: 1234567890123456789

批量发送相同内容

向多个用户发送相同的私信内容。

import asyncio
import x_api

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

    # 目标用户列表
    user_ids = [
        "123456789",
        "987654321",
        "555555555",
    ]

    # 发送相同的消息
    message = "这是一条群发消息，感谢关注！🎉"

    print(f"📤 开始批量发送私信到 {len(user_ids)} 个用户...")

    batch_result = await twitter.send_batch_direct_messages(
        user_ids=user_ids,
        text=message,
    )

    # 统计结果
    print(f"\n📊 发送统计:")
    print(f"   成功: {batch_result.success_count}/{len(user_ids)}")
    print(f"   失败: {batch_result.failure_count}/{len(user_ids)}")

    # 详细结果
    print(f"\n📋 详细结果:")
    for i, result in enumerate(batch_result.results, 1):
        status = "✅" if result.success else "❌"
        if result.success:
            print(f"   {status} {i}. 用户 {result.user_id} - 成功")
        else:
            print(f"   {status} {i}. 用户 {result.user_id} - 失败: {result.error_msg}")

asyncio.run(send_batch_messages())

输出示例：

📤 开始批量发送私信到 3 个用户...

📊 发送统计:
   成功: 3/3
   失败: 0/3

📋 详细结果:
   ✅ 1. 用户 123456789 - 成功
   ✅ 2. 用户 987654321 - 成功
   ✅ 3. 用户 555555555 - 成功

批量发送自定义文案

向不同用户发送不同的个性化内容。

import asyncio
import x_api

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

    # 用户ID 和对应的自定义文案
    user_ids = [
        "123456789",
        "987654321",
        "555555555",
    ]

    texts = [
        "你好，张三！很高兴认识你。",
        "你好，李四！感谢你的关注。",
        "你好，王五！欢迎加入我们。",
    ]

    print(f"📤 发送个性化私信到 {len(user_ids)} 个用户...")

    result = await twitter.send_batch_direct_messages_with_custom_texts(
        user_ids=user_ids,
        texts=texts,
    )

    print(f"\n✅ 发送完成！")
    print(f"   成功: {result.success_count} 条")
    print(f"   失败: {result.failure_count} 条")

    # 详细结果
    print("\n📊 详细结果：")
    for idx, dm_result in enumerate(result.results):
        status = "✅ 成功" if dm_result.success else "❌ 失败"
        if dm_result.success:
            detail = f"事件ID: {dm_result.event_id}"
        else:
            detail = dm_result.error_msg
        print(f"   [{idx + 1}] 用户 {dm_result.user_id} - {status} ({detail})")

asyncio.run(send_custom_messages())

输出示例：

📤 发送个性化私信到 3 个用户...

✅ 发送完成！
   成功: 3 条
   失败: 0 条

📊 详细结果：
   [1] 用户 123456789 - ✅ 成功 (事件ID: 1234567890123456789)
   [2] 用户 987654321 - ✅ 成功 (事件ID: 1234567890123456790)
   [3] 用户 555555555 - ✅ 成功 (事件ID: 1234567890123456791)

批量发送带图片的自定义私信

每个用户发送不同文案 + 独立的图片附件。

import asyncio
import x_api
from pathlib import Path

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

    # 用户配置
    user_ids = ["123456789", "987654321", "555555555"]
    texts = [
        "张三，这是给你的专属图片！",
        "李四，看看这个！",
        "王五，分享给你一张图片。",
    ]

    # 步骤1：读取图片
    image_bytes = Path("photo.jpg").read_bytes()
    print(f"📸 读取图片: {len(image_bytes)} 字节")

    # 步骤2：批量上传图片（获取多个独立的 media_id）
    # 注意：每个用户需要一个独立的 media_id
    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")

    # 步骤3：发送带图片的自定义私信
    result = await twitter.send_batch_direct_messages_with_custom_texts(
        user_ids=user_ids,
        texts=texts,
        media_ids=upload_result.media_ids,  # 传入 media_id 列表
    )

    print(f"\n📊 发送结果:")
    print(f"   成功: {result.success_count} 条")
    print(f"   失败: {result.failure_count} 条")

    for idx, dm_result in enumerate(result.results):
        status = "✅" if dm_result.success else "❌"
        print(f"   {status} 用户 {dm_result.user_id}: {texts[idx][:20]}...")

asyncio.run(send_custom_messages_with_images())

输出示例：

📸 读取图片: 102400 字节
✅ 图片上传成功，获得 3 个 media_id

📊 发送结果:
   成功: 3 条
   失败: 0 条
   ✅ 用户 123456789: 张三，这是给你的专属图片...
   ✅ 用户 987654321: 李四，看看这个！...
   ✅ 用户 555555555: 王五，分享给你一张图片。...

错误处理

处理常见的错误场景。

import asyncio
import x_api

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

    try:
        result = await twitter.send_direct_message(
            user_id="123456789",
            text="测试消息"
        )

        if result.success:
            print(f"✅ 发送成功")
        else:
            # 处理 API 返回的错误
            match result.http_status:
                case 403:
                    print("❌ 权限不足：对方可能不允许接收私信")
                case 404:
                    print("❌ 用户不存在")
                case 429:
                    print("❌ 请求过于频繁，请稍后重试")
                case _:
                    print(f"❌ 发送失败: {result.error_msg}")
                    print(f"   HTTP 状态码: {result.http_status}")

    except x_api.TwitterError as e:
        # 处理客户端异常
        print(f"❌ 客户端错误: {e}")

    except Exception as e:
        # 处理其他异常
        print(f"❌ 未知错误: {e}")

asyncio.run(handle_errors())

常见错误码

HTTP 状态码	说明	建议处理方式
400	请求参数错误	检查参数格式
401	认证失败	检查 cookies 是否过期
403	权限不足	对方可能不接受私信
404	用户不存在	验证 user_id 是否正确
429	请求频率限制	降低请求频率或等待
500	服务器内部错误	稍后重试

完整示例

将上述功能整合到一个完整的应用程序中。

#!/usr/bin/env python3
"""
DM 私信模块完整示例
"""

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 DMDemo:
    """DM 演示类"""

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

    async def send_single(self, user_id: str, text: str) -> bool:
        """发送单条私信"""
        result = await self.twitter.send_direct_message(user_id, text)
        return result.success

    async def send_with_image(
        self,
        user_id: str,
        text: str,
        image_path: str
    ) -> bool:
        """发送带图片的私信"""
        # 上传图片
        image_bytes = Path(image_path).read_bytes()
        upload = await self.twitter.upload_image(image_bytes, "dm_image")

        if not upload.success:
            return False

        # 发送私信
        result = await self.twitter.send_direct_message(
            user_id, text, upload.media_id_string
        )
        return result.success

    async def send_batch(
        self,
        user_ids: list[str],
        text: str
    ) -> tuple[int, int]:
        """批量发送，返回 (成功数, 失败数)"""
        result = await self.twitter.send_batch_direct_messages(user_ids, text)
        return result.success_count, result.failure_count

    async def send_personalized(
        self,
        user_ids: list[str],
        texts: list[str]
    ) -> tuple[int, int]:
        """发送个性化私信"""
        result = await self.twitter.send_batch_direct_messages_with_custom_texts(
            user_ids, texts
        )
        return result.success_count, result.failure_count


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

    proxy_url = os.getenv("PROXY_URL")

    # 创建演示实例
    demo = DMDemo(cookies, proxy_url)

    # 演示发送单条私信
    print("=== 发送单条私信 ===")
    success = await demo.send_single("123456789", "Hello!")
    print(f"结果: {'成功' if success else '失败'}")


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

相关链接

• 快速开始
• DM API 文档
• Upload API 文档
• 批量操作示例