Twitter V2过滤流异常求助：连接重置后无法添加规则

1. 解决HTTP 400规则添加错误 #

报错信息Cannot deserialize value of type ... AddRule from String value明确说明规则格式不符合API要求，结合代码分析，需修正两个核心问题：

（1）补全API请求的HTTPS前缀 #

代码中所有Twitter API的URL都缺少https://前缀，导致requests无法正确识别远程服务器，请求格式被破坏。需将所有API地址修改为完整格式：

• 将"api.twitter.com/2/tweets/search/stream/rules"改为"https://api.twitter.com/2/tweets/search/stream/rules"
• 将"api.twitter.com/2/tweets/search/stream"改为"https://api.twitter.com/2/tweets/search/stream"

（2）确保规则格式严格符合要求 #

Twitter过滤流规则要求add参数必须是包含规则字典的数组，每个规则字典需包含value字段（规则表达式），可选tag字段用于标记规则。检查set_rules函数中的sample_rules：

• 错误示例：直接传入字符串"dog"或单个字典（无数组包裹）
• 正确示例：

  sample_rules = [
      {"value": "dog", "tag": "dog-related tweets"}  # tag可选，可省略
  ]
  

2. 修复ConnectionResetError连接异常 #

流连接被重置通常是网络波动或服务器主动断开导致，需添加异常捕获与重试逻辑，确保流自动恢复：

修改get_stream函数，添加异常处理：

def get_stream(set):
    while True:
        try:
            response = requests.get(
                "https://api.twitter.com/2/tweets/search/stream", 
                auth=bearer_oauth, 
                stream=True,
                timeout=30  # 添加超时，避免无限等待
            )
            print(response.status_code)
            if response.status_code != 200:
                raise Exception(
                    "Cannot get stream (HTTP {}): {}".format(
                        response.status_code, response.text
                    )
                )
            for response_line in response.iter_lines():
                if response_line:
                    json_response = json.loads(response_line)
                    print(json.dumps(json_response, indent=4, sort_keys=True))
        except (ConnectionResetError, requests.exceptions.RequestException) as e:
            print(f"Stream connection failed: {e}, reconnecting in 5 seconds...")
            time.sleep(5)  # 重试前短暂等待，避免频繁请求
            continue  # 发生异常时重试连接

3. 完整修正后的代码 #

import requests
import os
import json
import time

bearer_token = "my token"  # 确保此处token有效且未过期

def bearer_oauth(r):
    r.headers["Authorization"] = f"Bearer {bearer_token}"
    r.headers["User-Agent"] = "v2FilteredStreamPython"
    return r

def get_rules():
    response = requests.get(
        "https://api.twitter.com/2/tweets/search/stream/rules", auth=bearer_oauth
    )
    if response.status_code != 200:
        raise Exception(
            "Cannot get rules (HTTP {}): {}".format(response.status_code, response.text)
        )
    print(json.dumps(response.json()))
    return response.json()

def delete_all_rules(rules):
    if rules is None or "data" not in rules:
        return None

    ids = list(map(lambda rule: rule["id"], rules["data"]))
    payload = {"delete": {"ids": ids}}
    response = requests.post(
        "https://api.twitter.com/2/tweets/search/stream/rules",
        auth=bearer_oauth,
        json=payload
    )
    if response.status_code != 200:
        raise Exception(
            "Cannot delete rules (HTTP {}): {}".format(
                response.status_code, response.text
            )
        )
    print(json.dumps(response.json()))

def set_rules(delete):
    sample_rules = [
        {"value": "dog", "tag": "dog-related tweets"}  # 可按需修改规则
    ]
    payload = {"add": sample_rules}
    response = requests.post(
        "https://api.twitter.com/2/tweets/search/stream/rules",
        auth=bearer_oauth,
        json=payload,
    )
    if response.status_code != 201:
        raise Exception(
            "Cannot add rules (HTTP {}): {}".format(response.status_code, response.text)
        )
    print(json.dumps(response.json()))

def get_stream(set):
    while True:
        try:
            response = requests.get(
                "https://api.twitter.com/2/tweets/search/stream", 
                auth=bearer_oauth, 
                stream=True,
                timeout=30
            )
            print(response.status_code)
            if response.status_code != 200:
                raise Exception(
                    "Cannot get stream (HTTP {}): {}".format(
                        response.status_code, response.text
                    )
                )
            for response_line in response.iter_lines():
                if response_line:
                    json_response = json.loads(response_line)
                    print(json.dumps(json_response, indent=4, sort_keys=True))
        except (ConnectionResetError, requests.exceptions.RequestException) as e:
            print(f"Stream connection failed: {e}, reconnecting in 5 seconds...")
            time.sleep(5)
            continue

def main():
    rules = get_rules()
    delete = delete_all_rules(rules)
    set = set_rules(delete)
    get_stream(set)

if __name__ == "__main__":
    main()

额外检查项 #

• 确认bearer_token未过期，且拥有Twitter API v2过滤流的访问权限
• 规则表达式value需符合Twitter过滤流语法，避免使用无效运算符或格式错误

内容的提问来源于stack exchange，提问作者si-young Lee