x-dm-python 1.1.0

Twitter DM 批量并发发送库 (Rust 实现, 提供 Python 绑定)

Twitter DM Python

基于 Rust 开发的 Twitter 私信批量发送库，提供了单条和批量并发发送私信的功能。

✨ 功能特性

• 🚀 单条私信发送: 向指定用户发送单条私信
• 📦 批量并发发送: 同时向多个用户发送私信，提高效率
• 🔒 安全认证: 基于 cookies 的 Twitter 认证机制
• 📝 详细日志: 完整的发送日志和错误追踪
• ⚡ 高性能: 使用 Tokio 实现真正的并发请求
• 🐍 Python 绑定: 通过 PyO3 提供 Python 接口

📦 安装

从源码构建 Python 扩展

# 安装 maturin (Python 包构建工具)
pip install maturin

# 开发模式安装 (快速测试)
maturin develop

# 或者构建 release wheel
maturin build --release

使用 Rust

在 Cargo.toml 中添加依赖:

[dependencies]
x_dm_python = { path = "." }
tokio = { version = "1", features = ["full"] }

🚀 快速开始

Python 使用示例

import x_dm_python

# 初始化客户端 (需要有效的 cookies)
cookies = "ct0=your_csrf_token; auth_token=your_auth_token; twid=u%3D123456789"
client = x_dm_python.Twitter(cookies)

# 发送单条私信
result = client.send_direct_message("123456789", "Hello from Python!")
if result.success:
    print(f"私信发送成功! Event ID: {result.event_id}")
else:
    print(f"私信发送失败: {result.error_msg}")

# 批量发送私信
user_ids = ["user_id_1", "user_id_2", "user_id_3"]
message = "这是一条批量测试消息！"
batch_result = client.send_batch_direct_messages(user_ids, message)

print(f"成功: {batch_result.success_count}, 失败: {batch_result.failure_count}")
for res in batch_result.results:
    if res.success:
        print(f"用户 {res.user_id} 发送成功")
    else:
        print(f"用户 {res.user_id} 发送失败: {res.error_msg}")

Rust 使用示例

use x_dm_python::Twitter;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // 初始化客户端
    let cookies = "ct0=your_csrf_token; auth_token=your_auth_token; twid=u%3D123456789";
    let client = Twitter::new(cookies.to_string(), None)?;

    // 发送单条私信
    let result = client.send_direct_message("123456789", "Hello, World!").await?;
    if result.success {
        println!("私信发送成功!");
    }

    // 批量发送私信
    let user_ids = vec!["123456789".to_string(), "987654321".to_string()];
    let results = client.send_batch_direct_messages(user_ids, "批量消息", None).await?;

    println!("成功: {}, 失败: {}", results.success_count, results.failure_count);

    Ok(())
}

📖 API 文档

Twitter 类

Python API

• Twitter(cookies: str, proxy_url: Optional[str] = None): 创建 Twitter 客户端
• send_direct_message(user_id: str, message: str) -> DMResult: 发送单条私信
• send_batch_direct_messages(user_ids: List[str], message: str, client_transaction_ids: Optional[List[str]] = None) -> BatchDMResult: 批量发送私信
• validate_cookies() -> bool: 验证 cookies 是否有效
• get_cookies() -> str: 获取当前 cookies

Rust API

• Twitter::new(cookies: String, proxy_url: Option<String>) -> Result<Self>: 创建 Twitter 客户端
• async fn send_direct_message(&self, user_id: &str, message: &str) -> Result<DMResult>: 发送单条私信
• async fn send_batch_direct_messages(&self, user_ids: Vec<String>, message: &str, client_transaction_ids: Option<Vec<String>>) -> Result<BatchDMResult>: 批量发送私信
• fn validate_cookies(&self) -> bool: 验证 cookies 是否有效

DMResult 结构体

字段	类型	说明
success	bool	发送是否成功
user_id	str/String	目标用户ID
message	str/String	发送的消息内容
error_msg	str/String	错误信息(如果有)
http_status	int/u16	HTTP 状态码
event_id	Optional[str]/Option<String>	Twitter 响应中的事件ID

BatchDMResult 结构体

字段	类型	说明
success_count	int/usize	成功发送的数量
failure_count	int/usize	发送失败的数量
results	List[DMResult]/Vec<DMResult>	每个私信的详细发送结果

🔐 获取 Twitter Cookies

1. 在浏览器中登录 Twitter
2. 打开开发者工具 (F12)
3. 转到 Network 标签页
4. 发送一条私信
5. 在请求头中找到 Cookie 字段
6. 复制完整的 Cookie 值

重要: 请确保 cookies 包含以下必要字段：

• ct0: CSRF令牌
• auth_token: 认证令牌
• twid: 用户ID

⚠️ 注意事项

安全性

• 🔐 保护 cookies: 不要在代码中硬编码 cookies，使用环境变量或配置文件
• 🚫 避免滥用: 遵守 Twitter 的使用条款，避免发送垃圾信息
• ⏱️ 请求频率: 注意控制请求频率，避免触发反垃圾机制

性能优化

• 📊 并发控制: 批量发送时自动并发，无需手动控制
• ⏰ 超时设置: 默认请求超时 30 秒
• 📝 日志级别: 使用环境变量 RUST_LOG=debug 设置日志级别

🏗️ 架构设计

本项目采用组合模式 + Trait 抽象的模块化架构：

模块组织

• common 模块: 认证、HTTP 客户端、错误处理
• dm 模块: 私信发送功能
• tweet 模块: 帖子相关功能（计划中）
• lib.rs: 聚合导出层

使用方式

聚合使用（推荐）

use x_dm_python::Twitter;

let twitter = Twitter::new(cookies, proxy)?;
twitter.dm().send_message("user_id", "message").await?;

独立模块使用

use x_dm_python::dm::{DMClient, DMService};

let dm = DMClient::new(auth, client);
dm.send_message("user_id", "message").await?;

技术栈

• 🚀 并发: Tokio 异步运行时
• 🔌 Python 绑定: PyO3
• 📝 日志: tracing
• 🔒 错误处理: Result<T, E>
• 📦 JSON: serde_json
• 🌐 HTTP: reqwest

📝 开发

构建

# 构建 Rust 库
cargo build

# 构建带 Python 绑定的库
cargo build --features python

# 运行测试
cargo test

# 构建 Python wheel
maturin build --release

测试

# Rust 单元测试
cargo test

# Python 测试 (需先安装)
maturin develop
python -m pytest tests/

📜 许可证

MIT License

🔗 相关链接

• Rust 文档
• PyO3 文档
• Maturin 文档
• Tokio 文档