流式传输

X API 通过诸如 Filtered Stream Endpoint 等端点支持实时 data，并在发生时推送匹配的 Post。这需要建立持久的 HTTP 连接。

​

设置与基础流

​

同步

from xdk import Client

# 初始化客户端
client = Client(bearer_token="your_bearer_token")

# 流式处理 Post（请先确保已配置规则）
for post_response in client.stream.posts():
    data = post_response.model_dump()
    
    if 'data' in data and data['data']:
        tweet = data['data']
        print(f"Post： {tweet.get('text', '')}")

​

异步

import asyncio
from asyncio import Queue
import threading
from xdk import Client

async def stream_posts_async(client: Client):
    queue = Queue()
    loop = asyncio.get_event_loop()
    stop = threading.Event()
    
    def run_stream():
        for post in client.stream.posts():
            if stop.is_set():
                break
            asyncio.run_coroutine_threadsafe(queue.put(post), loop)
        asyncio.run_coroutine_threadsafe(queue.put(None), loop)
    
    threading.Thread(target=run_stream, daemon=True).start()
    
    while True:
        post = await queue.get()
        if post is None:
            break
        data = post.model_dump()
        if 'data' in data and data['data']:
            print(f"Post: {data['data'].get('text', '')}")
    stop.set()

async def main():
    client = Client(bearer_token="your_bearer_token")
    await stream_posts_async(client)

asyncio.run(main())

​

规则管理

规则用于定义筛选条件，以确定你希望获取的特定 data（例如关键词、用户等）。你可以参阅本指南了解如何构建规则。

添加规则：

from xdk.stream.models import UpdateRulesRequest

# Add a rule
add_rules = {
    "add": [
        {"value": "from:xdevelopers", "tag": "official_updates"}
    ]
}

request_body = UpdateRulesRequest(**add_rules)
response = client.stream.update_rules(body=request_body)

删除规则：

from xdk.stream.models import UpdateRulesRequest

delete_rules = {
    "delete": {
        "ids": ["rule_id_1", "rule_id_2"]
    }
}

request_body = UpdateRulesRequest(**delete_rules)
response = client.stream.update_rules(body=request_body)

上架规则：

response = client.stream.get_rules()

# 打印规则
for rule in response.data:
    print(f"ID：{rule.id}，值：{rule.value}，标签：{rule.tag}")

有关完整规则语法，请参阅X Streaming Rules 文档。

​

故障排查

• 403 Forbidden：身份验证无效或权限不足。
• 420 Enhance Your Calm：触发频率限制；请等待后重试。
• No Data：使用 get_rules() 检查规则；确保存在匹配的 Post。

有关更多示例和 API 参考，请参见内联文档字符串（例如 help(client.tweets.search_recent)）或源码中的生成桩。通过 GitHub 仓库 提交反馈。