X API

https://x-api-rs.es007.com/dev
Admin
A high-performance Twitter (X) API Python client, implemented using Rust + PyO3, providing complete...
Tokens:145,693
Snippets:777
Trust Score:4.1
Update:9 months ago
Show doc for...
Tokens:
Raw
### Installation

Source: https://x-api-rs.es007.com/dev/examples/quickstart

Instructions for installing the X API Python client using pip and verifying the installation.

```APIDOC
## Installation

### Install using pip
```
pip install x-api-rs
```

### Verify Installation
```python
from x_api_rs import Twitter

# Verify module import
client = Twitter(cookies="test")
print("✅ x-api-rs installed successfully")
```

If the code runs without errors, the installation is successful.
```

--------------------------------

### Install X API Python Client

Source: https://x-api-rs.es007.com/dev/examples/quickstart

Installs the X API Python client library using pip. This is the first step to using the client for interacting with the X API.

```bash
pip install x-api-rs
```

--------------------------------

### Quick Start with X API Python Client

Source: https://x-api-rs.es007.com/dev/index

Demonstrates a quick start guide for using the X API Python client. It shows how to initialize the client with cookies, send a direct message, upload an image, create a tweet with media, and fetch user profile information. This example requires an `image.jpg` file to be present.

```python
import asyncio
from x_api_rs import Twitter

async def main():
    # 创建客户端（使用 cookies 认证）
    client = Twitter(cookies="your_cookies_here")

    # 发送私信
    result = await client.dm.send_message("123456", "Hello!")
    if result.success:
        print(f"发送成功，事件ID: {result.event_id}")

    # 上传图片
    with open("image.jpg", "rb") as f:
        image_bytes = f.read()
    upload_result = await client.upload.image(image_bytes, "tweet_image")

    # 发帖（带图片）
    tweet_result = await client.posts.create_tweet(
        text="Hello World!",
        media_ids=[upload_result.media_id_string]
    )

    # 获取用户资料
    user = await client.user.get_profile("elonmusk")
    print(f"用户: {user.name}, 粉丝: {user.followers_count}")

asyncio.run(main())

```

--------------------------------

### Verify X API Python Client Installation

Source: https://x-api-rs.es007.com/dev/examples/quickstart

Verifies the installation of the X API Python client by importing the Twitter class and creating a client instance. Successful execution indicates the library is installed correctly.

```python
from x_api_rs import Twitter

# Verify module import success
client = Twitter(cookies="test")
print("✅ x-api-rs Installation successful")
```

--------------------------------

### 带重试的高级错误处理

Source: https://x-api-rs.es007.com/dev/examples/quickstart

演示如何实现一个带指数退避策略的发送函数，以处理 API 速率限制和网络异常。此函数会根据 HTTP 状态码 429 进行重试，其他错误则直接返回。

```python
import asyncio

async def send_with_retry(client, user_id, text, max_retries=3):
    """带重试的发送函数"""
    for attempt in range(max_retries):
        try:
            result = await client.dm.send_message(user_id, text)

            if result.success:
                return result

            # 业务失败，检查是否需要重试
            if result.http_status == 429:  # 速率限制
                wait_time = 2 ** attempt  # 指数退避
                print(f"触发速率限制，等待 {wait_time} 秒...")
                await asyncio.sleep(wait_time)
                continue
            else:
                # 其他错误，不重试
                return result

        except Exception as e:
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(2 ** attempt)

    raise Exception(f"重试 {max_retries} 次后仍然失败")

```

--------------------------------

### 设置请求超时

Source: https://x-api-rs.es007.com/dev/examples/quickstart

演示了如何使用 `asyncio.wait_for` 来为私信发送操作设置超时限制。如果请求在指定时间内未完成，将抛出 `asyncio.TimeoutError`。

```python
import asyncio

try:
    result = await asyncio.wait_for(
        client.dm.send_message("user_id", "Hello!"),
        timeout=30.0  # 30秒超时
    )
except asyncio.TimeoutError:
    print("请求超时")

```

--------------------------------

### Prerequisites

Source: https://x-api-rs.es007.com/dev/examples/inbox_demo

Setup and client creation for interacting with the API.

```APIDOC
## Prerequisites

### Install Dependencies

```
maturin develop
```

### Create Client

```python
import x_api

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

# Or use a proxy
twitter = x_api.Twitter(cookies, "http://127.0.0.1:7890")
```
```

--------------------------------

### 并发发送多条私信

Source: https://x-api-rs.es007.com/dev/examples/quickstart

介绍了如何利用 `asyncio.gather` 并发发送多条私信。此方法允许同时执行多个 `client.dm.send_message` 任务，并收集所有结果或异常。

```python
import asyncio

# 创建任务列表
tasks = [
    client.dm.send_message("123", "消息1"),
    client.dm.send_message("456", "消息2"),
    client.dm.send_message("789", "消息3"),
]

# 并发执行
results = await asyncio.gather(*tasks, return_exceptions=True)

for result in results:
    if isinstance(result, Exception):
        print(f"异常: {result}")
    elif result.success:
        print(f"成功: {result.event_id}")

```

--------------------------------

### 发布带图片的推文

Source: https://x-api-rs.es007.com/dev/examples/quickstart

此代码片段演示了如何使用 X API RS 库上传图片并将其附加到推文中。它首先上传图片，然后使用返回的 media_id_string 创建带有文本和图片的推文。需要有效的 Twitter cookies 才能运行。

```python
async def post_tweet_with_image():
    client = Twitter(cookies="your_cookies_here")

    # 1. 上传图片
    with open("image.jpg", "rb") as f:
        upload_result = await client.upload.image(f.read(), "tweet_image")

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

    # 2. 发布带图片的帖子
    result = await client.posts.create_tweet(
        text="分享一张图片 📷",
        media_ids=[upload_result.media_id_string]
    )

    if result.success:
        print(f"✅ 发帖成功!")
        print(f"   Tweet ID: {result.tweet_id}")

asyncio.run(post_tweet_with_image())

```

--------------------------------

### 获取用户ID

Source: https://x-api-rs.es007.com/dev/examples/quickstart

展示了如何通过用户名查询用户资料并提取用户 ID。此功能依赖于 `client.user.get_profile` 方法，并检查返回结果的 `success` 属性。

```python
# 通过用户名获取资料（包含用户 ID）
result = await client.user.get_profile("username")
if result.success:
    user_id = result.rest_id
    print(f"用户 ID: {user_id}")

```

--------------------------------

### 运行完整示例说明

Source: https://x-api-rs.es007.com/dev/examples/quickstart

此部分提供了运行 X API RS 完整示例的说明。包括设置必要的 TWITTER_COOKIES 环境变量和执行 Python 脚本的命令。

```bash
# 设置环境变量
export TWITTER_COOKIES="ct0=xxx; auth_token=yyy; twid=u%3D123"

# 运行示例
python quickstart_full_example.py

```

--------------------------------

### Install Dependencies with Maturin

Source: https://x-api-rs.es007.com/dev/examples/batch_operations

This code snippet demonstrates how to install project dependencies using the 'maturin develop' command. Maturin is a tool for building and publishing Rust Python extensions.

```shell
maturin develop

```

--------------------------------

### Client Initialization

Source: https://x-api-rs.es007.com/dev/examples/quickstart

How to initialize the Twitter client with cookies, including proxy support and environment variable configuration.

```APIDOC
## Creating a Client

### Basic Usage
```python
from x_api_rs import Twitter

# Create client
client = Twitter(cookies="your_cookies_here")

# Validate cookies
if client.validate_cookies():
    print("✅ Cookies are valid")
else:
    print("❌ Cookies are invalid")
```

### Using Proxies
```python
# HTTP Proxy
client = Twitter(
    cookies="your_cookies_here",
    proxy_url="http://proxy:8080"
)

# SOCKS5 Proxy
client = Twitter(
    cookies="your_cookies_here",
    proxy_url="socks5://proxy:1080"
)
```

### Reading from Environment Variable
```python
import os
from x_api_rs import Twitter

# Set environment variable
# export TWITTER_COOKIES="ct0=xxx; auth_token=yyy; twid=u%3D123"

cookies = os.getenv("TWITTER_COOKIES")
if not cookies:
    raise ValueError("Please set the TWITTER_COOKIES environment variable")

client = Twitter(cookies)
```
```

--------------------------------

### Full Media Upload Example with x-api-rs in Python

Source: https://x-api-rs.es007.com/dev/examples/upload_demo

A comprehensive Python example demonstrating various media upload functionalities using x-api-rs. This includes uploading single images, batch uploading multiple images, and uploading videos, along with setup and configuration details.

```python
#!/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())

```

--------------------------------

### X API RS 快速开始完整示例

Source: https://x-api-rs.es007.com/dev/examples/quickstart

这是一个综合性的 Python 示例，演示了 X API RS 库的多个核心功能。它包括初始化客户端、发送私信、上传图片、发布文本或图片推文、执行批量私信操作以及获取用户资料。运行前需要设置 TWITTER_COOKIES 环境变量。

```python
#!/usr/bin/env python3
"""
X API 快速开始完整示例

演示私信、上传、发帖等核心功能
"""

import asyncio
import os
from x_api_rs import Twitter

async def main():
    # 1. 初始化客户端
    print("=" * 50)
    print("🔐 初始化 Twitter 客户端")
    print("=" * 50)

    cookies = os.getenv("TWITTER_COOKIES")
    if not cookies:
        print("❌ 请设置 TWITTER_COOKIES 环境变量")
        print("   示例: export TWITTER_COOKIES='ct0=xxx; auth_token=yyy; twid=u%3D123'")
        return

    client = Twitter(cookies)

    # 验证 cookies
    if not client.validate_cookies():
        print("❌ Cookies 验证失败!")
        return

    print("✅ Cookies 验证成功\n")

    # 2. 发送私信
    print("=" * 50)
    print("📨 发送私信测试")
    print("=" * 50)

    user_id = "123456789"  # 替换为真实用户 ID
    dm_result = await client.dm.send_message(
        user_id=user_id,
        text="Hello! 这是一条测试私信"
    )

    if dm_result.success:
        print(f"✅ 私信发送成功")
        print(f"   事件 ID: {dm_result.event_id}\n")
    else:
        print(f"❌ 私信发送失败: {dm_result.error_msg}\n")

    # 3. 上传图片
    print("=" * 50)
    print("📤 上传图片测试")
    print("=" * 50)

    # 检查图片文件是否存在
    image_path = "test_image.jpg"
    if os.path.exists(image_path):
        with open(image_path, "rb") as f:
            upload_result = await client.upload.image(f.read(), "tweet_image")

        if upload_result.success:
            print(f"✅ 图片上传成功")
            print(f"   Media ID: {upload_result.media_id_string}\n")
            media_id = upload_result.media_id_string
        else:
            print(f"❌ 图片上传失败: {upload_result.error_msg}\n")
            media_id = None
    else:
        print(f"⚠️  图片文件不存在: {image_path}")
        print("   跳过上传测试\n")
        media_id = None

    # 4. 发布帖子
    print("=" * 50)
    print("📝 发布帖子测试")
    print("=" * 50)

    tweet_text = "Hello from X API! 🚀"
    if media_id:
        # 带图片发帖
        tweet_result = await client.posts.create_tweet(
            text=tweet_text,
            media_ids=[media_id]
        )
    else:
        # 纯文本发帖
        tweet_result = await client.posts.create_tweet(text=tweet_text)

    if tweet_result.success:
        print(f"✅ 帖子发布成功")
        print(f"   Tweet ID: {tweet_result.tweet_id}")
        print(f"   链接: https://twitter.com/i/web/status/{tweet_result.tweet_id}\n")
    else:
        print(f"❌ 帖子发布失败: {tweet_result.error_msg}\n")

    # 5. 批量操作
    print("=" * 50)
    print("📦 批量操作测试")
    print("=" * 50)

    user_ids = ["123456789", "987654321", "111222333"]
    batch_result = await client.dm.send_batch(
        user_ids=user_ids,
        text="批量测试消息"
    )

    print(f"批量发送完成:")
    print(f"   成功: {batch_result.success_count} 条")
    print(f"   失败: {batch_result.failure_count} 条\n")

    # 6. 获取用户资料
    print("=" * 50)
    print("👤 获取用户资料测试")
    print("=" * 50)

    user_result = await client.user.get_profile("elonmusk")
    if user_result.success:
        print(f"✅ 用户资料获取成功")
        print(f"   用户名: {user_result.name}")
        print(f"   用户 ID: {user_result.rest_id}")
        print(f"   粉丝数: {user_result.followers_count:,}")
        print(f"   关注数: {user_result.following_count:,}\n")
    else:
        print(f"❌ 用户资料获取失败: {user_result.error_msg}\n")

    # 完成
    print("=" * 50)
    print("✨ 快速开始示例运行完成!")
    print("=" * 50)

if __name__ == "__main__":
    # 运行主函数
    asyncio.run(main())

```

--------------------------------

### 验证 Cookies

Source: https://x-api-rs.es007.com/dev/examples/quickstart

说明了如何使用 `client.validate_cookies()` 方法来检查当前配置的 Cookies 是否有效。该方法返回一个布尔值，指示 Cookies 是否已过期。

```python
if client.validate_cookies():
    print("Cookies 有效")
else:
    print("Cookies 无效或已过期")

```

--------------------------------

### 基础错误处理

Source: https://x-api-rs.es007.com/dev/examples/quickstart

此代码片段展示了使用 X API RS 库进行私信发送时的基础错误处理。它使用 try-except 块来捕获潜在的异常（如网络错误或超时），并检查 `send_message` 方法返回结果中的 `success` 属性和 `error_msg`。

```python
async def safe_send():
    client = Twitter(cookies)

    try:
        result = await client.dm.send_message("user_id", "Hello!")

        # 检查业务结果
        if result.success:
            print(f"成功: {result.event_id}")
        else:
            print(f"失败: {result.error_msg} (HTTP {result.http_status})")

    except Exception as e:
        # 捕获异常（网络错误、超时等）
        print(f"异常: {e}")

asyncio.run(safe_send())

```

--------------------------------

### Initialize X API Client

Source: https://x-api-rs.es007.com/dev/examples/quickstart

Initializes the X API client with provided cookies. It includes a method to validate the cookies to ensure they are still active and usable.

```python
from x_api_rs import Twitter

# Create client
client = Twitter(cookies="your_cookies_here")

# Validate cookies
if client.validate_cookies():
    print("✅ Cookies are valid")
else:
    print("❌ Cookies are invalid")
```

--------------------------------

### Obtaining Cookies

Source: https://x-api-rs.es007.com/dev/examples/quickstart

Methods to obtain necessary cookies for authenticating with the Twitter API.

```APIDOC
## Obtaining Cookies

### Method 1: Browser Developer Tools
1. Log in to twitter.com using Chrome/Firefox.
2. Press `F12` to open Developer Tools.
3. Navigate to the `Network` tab.
4. Refresh the page or perform an action.
5. Select any request and view `Request Headers`.
6. Find the `Cookie` field and copy its entire content.

**Example**: 
```
ct0=your_csrf_token_here; auth_token=your_auth_token_here; twid=u%3D1234567890
```

### Method 2: Browser Extensions
Install a cookie export extension (e.g., EditThisCookie) and export cookies with one click.

### Method 3: Using `auth_token` Conversion
If you only have `auth_token`, you can use the conversion feature:
```python
import asyncio
from x_api_rs import Twitter

async def main():
    result = await Twitter.auth_token_to_cookies(
        "your_auth_token_here",
        proxy_url="http://proxy:8080"  # Optional
    )

    print(f"User ID: {result.user_id}")
    print(f"Cookies: {result.cookies}")

    # Use converted cookies
    client = Twitter(result.cookies)

asyncio.run(main())
```
```

--------------------------------

### Initialize X API Client with Proxy

Source: https://x-api-rs.es007.com/dev/examples/quickstart

Initializes the X API client while configuring proxy settings. Supports both HTTP and SOCKS5 proxy protocols for network requests.

```python
# HTTP Proxy
client = Twitter(
    cookies="your_cookies_here",
    proxy_url="http://proxy:8080"
)

# SOCKS5 Proxy
client = Twitter(
    cookies="your_cookies_here",
    proxy_url="socks5://proxy:1080"
)
```

--------------------------------

### Media Upload API

Source: https://x-api-rs.es007.com/dev/examples/quickstart

Endpoints for uploading media files, such as images, with different categories for various uses.

```APIDOC
## Uploading Images

### Upload Single Image
```python
import asyncio
from x_api_rs import Twitter

async def upload_image():
    client = Twitter(cookies="your_cookies_here")

    # Read image file
    with open("image.jpg", "rb") as f:
        image_bytes = f.read()

    # Upload image
    result = await client.upload.image(
        image_bytes=image_bytes,
        media_category="tweet_image"  # For posting tweets
    )

    if result.success:
        print(f"✅ Upload successful!")
        print(f"   Media ID: {result.media_id_string}")
        return result.media_id_string
    else:
        print(f"❌ Upload failed: {result.error_msg}")
        return None

asyncio.run(upload_image())
```

### Media Category Explanation
Category | Usage  
---|---
`tweet_image` | For posting tweets  
`dm_image` | For direct message attachments  
`banner_image` | For user banner images  

### Sending Direct Message with Image
```python
async def send_image_dm():
    client = Twitter(cookies="your_cookies_here")

    # 1. Upload image
    with open("product.jpg", "rb") as f:
        upload_result = await client.upload.image(f.read(), "dm_image")

    if not upload_result.success:
        print(f"Upload failed: {upload_result.error_msg}")
        return

    # 2. Send direct message with image
    result = await client.dm.send_message(
        user_id="123456789",
        text="Please see this image!",
        media_id=upload_result.media_id_string
    )

    if result.success:
        print(f"✅ Sent successfully!")
    else:
        print(f"❌ Failed to send: {result.error_msg}")

asyncio.run(send_image_dm())
```
```

--------------------------------

### Direct Message API

Source: https://x-api-rs.es007.com/dev/examples/quickstart

Endpoints for sending direct messages, including single and batch sending.

```APIDOC
## Sending a Direct Message

### Send Single Message
```python
import asyncio
from x_api_rs import Twitter

async def main():
    client = Twitter(cookies="your_cookies_here")

    # Send direct message
    result = await client.dm.send_message(
        user_id="123456789",  # Replace with actual user ID
        text="Hello! This is my first direct message 🎉"
    )

    # Check result
    if result.success:
        print(f"✅ Sent successfully!")
        print(f"   Event ID: {result.event_id}")
        print(f"   User ID: {result.user_id}")
    else:
        print(f"❌ Failed to send: {result.error_msg}")
        print(f"   HTTP Status Code: {result.http_status}")

# Run
asyncio.run(main())
```

### Batch Sending
```python
import asyncio
from x_api_rs import Twitter

async def batch_send():
    client = Twitter(cookies="your_cookies_here")

    # Prepare user ID list
    user_ids = [
        "123456789",
        "987654321",
        "111222333"
    ]

    # Send the same content in batch
    result = await client.dm.send_batch(
        user_ids=user_ids,
        text="Batch sending test message!"
    )

    # View results
    print(f"📊 Batch sending completed:")
    print(f"   Success: {result.success_count} messages")
    print(f"   Failure: {result.failure_count} messages")

    # View detailed results
    for item in result.results:
        if item.success:
            print(f"   ✅ User {item.user_id}: Success")
        else:
            print(f"   ❌ User {item.user_id}: {item.error_msg}")

asyncio.run(batch_send())
```
```

--------------------------------

### Error Handling

Source: https://x-api-rs.es007.com/dev/examples/quickstart

Demonstrates basic error handling using try-except blocks for API calls and checking the success field in the response.

```APIDOC
## Error Handling

### Description
This section outlines how to handle potential errors during API interactions, including network issues and business logic failures.

### Basic Error Handling Example
```python
async def safe_send():
    client = Twitter(cookies)

    try:
        # Attempt to send a direct message
        result = await client.dm.send_message("user_id", "Hello!")

        # Check the business logic result from the API response
        if result.success:
            print(f"Successfully sent message with Event ID: {result.event_id}")
        else:
            # Handle API-specific errors reported in the response body
            print(f"Failed to send message: {result.error_msg} (HTTP Status: {result.http_status})")

    except Exception as e:
        # Catch exceptions related to network errors, timeouts, etc.
        print(f"An exception occurred: {e}")

asyncio.run(safe_send())
```

### Key Points
- **Try-Except Blocks**: Use `try...except` to catch network-level errors or unexpected exceptions during the request.
- **Response `success` Field**: Always check the `success` field in the API response to determine if the business operation was completed successfully.
- **Error Messages**: Utilize `error_msg` and `http_status` from the response for detailed error diagnostics.
```

--------------------------------

### POST /posts/create_tweet

Source: https://x-api-rs.es007.com/dev/examples/quickstart

Allows posting tweets with or without media. Media can be uploaded separately and then referenced by its media ID.

```APIDOC
## POST /posts/create_tweet

### Description
This endpoint allows you to create a new tweet. You can include text and optionally attach media that has been previously uploaded.

### Method
POST

### Endpoint
/posts/create_tweet

### Parameters
#### Query Parameters
None

#### Request Body
- **text** (string) - Required - The main content of the tweet.
- **media_ids** (array of strings) - Optional - A list of media IDs to attach to the tweet.

### Request Example
```json
{
  "text": "Sharing a photo! 📷",
  "media_ids": ["your_media_id_string"]
}
```

### Response
#### Success Response (200)
- **success** (boolean) - Indicates if the tweet creation was successful.
- **tweet_id** (string) - The unique identifier of the created tweet.

#### Response Example
```json
{
  "success": true,
  "tweet_id": "1234567890123456789"
}
```

#### Error Response
- **success** (boolean) - False if an error occurred.
- **error_msg** (string) - A message describing the error.
- **http_status** (integer) - The HTTP status code of the error.

```json
{
  "success": false,
  "error_msg": "Invalid media ID provided",
  "http_status": 400
}
```
```

--------------------------------

### Full Upload Example

Source: https://x-api-rs.es007.com/dev/examples/upload_demo

A comprehensive example demonstrating the usage of the UploadDemo class for uploading images and videos via the X API.

```APIDOC
## Full Example

```python
#!/usr/bin/env python3
"""
Upload media upload module full example
"""

import asyncio
import os
import sys
from pathlib import Path

try:
    import x_api
except ImportError:
    print("❌ Please install x_api first: maturin develop")
    sys.exit(1)


class UploadDemo:
    """Upload demo class"""

    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:
        """Upload a single image, return 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]:
        """Batch upload images, return media_id list"""
        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:
        """Upload video, return 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():
    # Read configuration
    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("❌ Please set TWITTER_COOKIES or create cookies.txt")
            return

    demo = UploadDemo(cookies)

    # Demonstrate image upload
    if Path("photo.jpg").exists():
        print("=== Uploading image ===")
        media_id = await demo.upload_image("photo.jpg")
        if media_id:
            print(f"✅ Upload successful: {media_id}")
        else:
            print("❌ Upload failed")


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

```

### Related Links

* Quick Start
* Upload API Documentation
* DM Direct Message Example
* Batch Operations Example
```

--------------------------------

### Install X API Python Client

Source: https://x-api-rs.es007.com/dev/index

Installs the X API Python client using pip. This is the primary method for obtaining the library for use in Python projects.

```bash
pip install x-api-rs

```

--------------------------------

### Posts API - Full Example

Source: https://x-api-rs.es007.com/dev/examples/posts_demo

A comprehensive example demonstrating various functionalities of the Posts API, including creating tweets, creating tweets with images, and performing likes and retweets.

```APIDOC
## Posts API - Full Example

### Description
This example showcases the full capabilities of the Posts API module within the x-api-rs library. It includes methods for creating simple text tweets, tweets with images, liking tweets, and retweeting.

### Method
Various (GET, POST)

### Endpoint
Multiple endpoints covered by the `x_api.Twitter.posts` and `x_api.Twitter.upload` objects.

### Parameters
- **cookies** (string) - Required - Authentication cookies for the Twitter API.
- **proxy_url** (string) - Optional - URL for a proxy server to use for requests.
- **text** (string) - Required - The content for a tweet.
- **image_path** (string) - Required - The file path to the image to be uploaded.
- **tweet_id** (string) - Required - The ID of the tweet to interact with (like/retweet).

### Request Example
```python
#!/usr/bin/env python3

import asyncio
import os
import sys
from pathlib import Path

try:
    import x_api
except ImportError:
    print("❌ Please install 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:
        """Posts a tweet, returns the 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:
        """Posts a tweet with an image"""
        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]:
        """Likes and retweets a tweet, returns (like_success, retweet_success)"""
        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("❌ Please set TWITTER_COOKIES or create cookies.txt")
            return

    demo = PostsDemo(cookies)

    print("=== Posting a tweet ===")
    tweet_id = await demo.create_tweet("Hello from Python! 🐍")
    if tweet_id:
        print(f"✅ Tweet posted successfully: {tweet_id}")
    else:
        print("❌ Failed to post tweet")

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

### Response
#### Success Response (200)
- **tweet_id** (string) - The ID of the created tweet.
- **success** (boolean) - Indicates if the operation was successful.

#### Response Example
```json
{
  "success": true,
  "tweet_id": "1789012345678901234"
}
```

### Related Links
- [Quick Start](link-to-quick-start)
- [Posts API Documentation](link-to-posts-api-docs)
- [Upload API Documentation](link-to-upload-api-docs)
- [DM Direct Message Example](link-to-dm-example)
```

--------------------------------

### GET /user/get_profile

Source: https://x-api-rs.es007.com/dev/examples/quickstart

Retrieves the profile information for a given user. You can specify the username to fetch details.

```APIDOC
## GET /user/get_profile

### Description
Fetches detailed profile information for a specified user, identified by their username.

### Method
GET

### Endpoint
/user/get_profile

### Parameters
#### Path Parameters
None

#### Query Parameters
- **username** (string) - Required - The username of the user whose profile to retrieve.

### Request Example
```
GET /user/get_profile?username=elonmusk
```

### Response
#### Success Response (200)
- **success** (boolean) - Indicates if the user profile was retrieved successfully.
- **name** (string) - The display name of the user.
- **rest_id** (string) - The unique internal ID of the user.
- **followers_count** (integer) - The number of followers the user has.
- **following_count** (integer) - The number of users the user is following.

#### Response Example
```json
{
  "success": true,
  "name": "Elon Musk",
  "rest_id": "123456789",
  "followers_count": 150000000,
  "following_count": 500
}
```

#### Error Response
- **success** (boolean) - False if an error occurred.
- **error_msg** (string) - A message describing the error.

```json
{
  "success": false,
  "error_msg": "User not found"
}
```
```

--------------------------------

### Get Latest Direct Messages

Source: https://x-api-rs.es007.com/dev/examples/inbox_demo

Fetches the most recent messages from the inbox.

```APIDOC
## Get Latest Direct Messages

Fetches the most recent messages from the inbox.

```python
import asyncio
import x_api

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

    result = await twitter.inbox.get_user_updates()

    if result.success:
        print(f"📬 Inbox Updates")
        print(f"   Data Source: {result.data_source}")
        print(f"   Message Count: {len(result.entries)}")
        print(f"   User Count: {len(result.users)}")

        # Iterate through messages
        print(f"\n📨 Latest Messages:")
        for entry in result.entries:
            if entry.message:
                msg = entry.message.message_data

                # Get sender info
                sender = result.users.get(msg.sender_id)
                sender_name = f"@{sender.screen_name}" if sender else msg.sender_id

                print(f"   [{msg.time}] {sender_name}: {msg.text[:50]}...")
    else:
        print(f"❌ Failed to get: {result.error_msg}")
        print(f"   HTTP Status Code: {result.http_status}")

asyncio.run(get_latest_messages())

```

**Output Example:**
```
📬 Inbox Updates
   Data Source: rpc
   Message Count: 15
   User Count: 8

📨 Latest Messages:
   [1705123456789] @user1: Hello! How are you doing lately?...
   [1705123456000] @user2: Got it, I'll reply tomorrow...
   [1705123400000] @user3: Thank you for your help!...

```

### Simple Message List

```python
import asyncio
import x_api

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

    result = await twitter.inbox.get_user_updates()

    if result.success:
        for i, entry in enumerate(result.entries[:10], 1):
            if entry.message:
                msg = entry.message.message_data
                sender = result.users.get(msg.sender_id)
                name = sender.screen_name if sender else "Unknown"
                print(f"{i}. @{name}: {msg.text}")

asyncio.run(list_messages())

```
```

--------------------------------

### Get Specific Conversation Messages

Source: https://x-api-rs.es007.com/dev/examples/inbox_demo

Fetches messages from a conversation with a specific user.

```APIDOC
## Get Specific Conversation Messages

Fetches messages from a conversation with a specific user.

```python
import asyncio
import x_api

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

    # Specify conversation ID
    conversation_id = "123456789-987654321"

    result = await twitter.inbox.get_user_updates(
        active_conversation_id=conversation_id
    )

    if result.success:
        print(f"💬 Messages in conversation {conversation_id[:16]}...:")

        for entry in result.entries:
            if entry.message:
                if entry.message.conversation_id == conversation_id:
                    msg = entry.message.message_data
                    sender = result.users.get(msg.sender_id)
                    name = sender.screen_name if sender else msg.sender_id
                    print(f"   @{name}: {msg.text}")
    else:
        print(f"❌ Failed to get: {result.error_msg}")

asyncio.run(get_conversation_messages())

```
```

--------------------------------

### Initialize Twitter Client with Main Entry Point (Python)

Source: https://x-api-rs.es007.com/dev/api

Demonstrates initializing the main Twitter client using the recommended entry point. This method offers unified authentication, configuration, and connection pooling for accessing various X API modules.

```python
from x_api_rs import Twitter

client = Twitter(cookies)

# Accessing various modules
await client.dm.send_message(...)
await client.upload.image(...)
await client.posts.create_tweet(...)
await client.user.get_profile(...)
await client.inbox.get_user_updates(...)
```

--------------------------------

### DMClient Constructor Examples (Python)

Source: https://x-api-rs.es007.com/dev/api/dm

Provides examples of initializing the DMClient with different configurations, including basic initialization, using a proxy server, and disabling JA3 fingerprint simulation for advanced use cases.

```python
# Basic usage
client = DMClient(cookies)

# Using a proxy
client = DMClient(cookies, proxy_url="http://proxy:8080")

# Disabling JA3 fingerprint simulation
client = DMClient(cookies, enable_ja3=False)
```

--------------------------------

### Recommended Client Initialization vs. Not Recommended (Python)

Source: https://x-api-rs.es007.com/dev/api

Highlights the recommended practice of using the main `Twitter` client entry point for unified access versus initializing individual module clients, which is generally not recommended unless specific configuration needs arise.

```python
# Recommended
client = Twitter(cookies)
await client.dm.send_message(...)

# Not recommended (unless for special needs)
dm = DMClient(cookies)
upload = UploadClient(cookies)
```
Tokens:
Raw