尝试向过滤流 Twitter API 发出请求时出现未经授权的消息

Question

我正在尝试使用以下代码向Filtered stream Twitter API 服务发出请求：

def filtered_stream(token):

    'token input is the Bearer Token'
    
    # endpoint URL
    url = 'https://api.twitter.com/2/tweets/search/stream?'
        
    # parameters
    params = { 
        'Authorization': "Bearer " + token,
        'tweet.fields': 'created_at'
    }
    
    res = requests.get(url = url, params = params, stream=True)
    response = res.json()
    
    return response

但我收到以下回复：

{'title': 'Unauthorized',
 'type': 'about:blank',
 'status': 401,
 'detail': 'Unauthorized'}

我不确定为什么会出现该错误。我的应用程序属于一个项目，可以使用 V2 提出请求。我不确定我做错了什么。非常感谢您的帮助。

Answer 1

编辑：这是对代码示例的重写，希望它更容易理解。

import requests
import json
from dataclasses import dataclass


@dataclass
class TwitterStream:
    bearer_token: str = "<YOUR BEARER TOKEN GOES HERE>"

    @classmethod
    def bearer_oauth(cls, request_object):
        """
        Method used to authenticated get requests made to the api.
        :param request_object: requests object, used to specify headers for authentication.
        :return request_object: same object with bearer token and app identifiable user agent string.
        """

        request_object.headers["Authorization"] = "Bearer {}".format(cls.bearer_token)
        request_object.headers["User-Agent"] = "v2FilteredStreamPython"
        return request_object

    @classmethod
    def get_rules(cls):
        """ "
        Specify rules for searching the streams.
        :return json_response: JSON, API response that contains the rules to call the stream endpoint.
        """
        response = requests.get(
            "https://api.twitter.com/2/tweets/search/stream/rules",
            auth=cls.bearer_oauth,
        )
        if response.status_code != 200:
            raise Exception(
                "Cannot get rules (HTTP {}): {}".format(
                    response.status_code, response.text
                )
            )
        json_response = response.json()
        return json_response

    @classmethod
    def delete_all_rules(cls, rules):
        """
        Deletes existing rules.
        :param rules: json response, rules returned from get rules method above.
        """

        if rules or "data" in rules:
            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=cls.bearer_oauth,
                json=payload,
            )
            if response.status_code != 200:
                raise Exception(
                    "Cannot delete rules (HTTP {}): {}".format(
                        response.status_code, response.text
                    )
                )

    @classmethod
    def set_rules(cls, rules):
        """
        Sets rules for calling the stream endpoint.
        :param rules: list of dicts containing rules. Sample syntax:
        [
            {"value": "dog has:images", "tag": "dog pictures"},
            {"value": "cat has:images -grumpy", "tag": "cat pictures"},
        ]
        """

        payload = {"add": rules}
        response = requests.post(
            "https://api.twitter.com/2/tweets/search/stream/rules",
            auth=cls.bearer_oauth,
            json=payload,
        )
        if response.status_code != 201:
            raise Exception(
                "Cannot add rules (HTTP {}): {}".format(
                    response.status_code, response.text
                )
            )

    @classmethod
    def get_stream(cls, breaker):
        """
        Call the stream endpoint.
        :param breaker: int, number of successful calls to make to the stream endpoint.
        :return json_responses: list of json reponses.
        """
        response = requests.get(
            "https://api.twitter.com/2/tweets/search/stream",
            auth=cls.bearer_oauth,
            stream=True,
        )
        if response.status_code != 200:
            raise Exception(
                "Cannot get stream (HTTP {}): {}".format(
                    response.status_code, response.text
                )
            )

        count = 0
        json_responses = []
        for response_line in response.iter_lines():
            if response_line:
                json_response = json.loads(response_line)
                json_responses.append(json_response)
                count += 1
            if count == breaker:
                break

        if json_responses:
            return json_responses
        else:
            raise Exception(
                "No responses returned from stream ({}".format(
                    [x for x in response.iter_lines()]
                )
            )

    @classmethod
    def main(cls, my_rules, breaker):
        """
        Main function for calling the endpoint.
        :param my_rules: list of dicts containing rules. Sample syntax:
        [
            {"value": "dog has:images", "tag": "dog pictures"},
            {"value": "cat has:images -grumpy", "tag": "cat pictures"},
        ]
        :param breaker: int, number of successful calls to make to the stream endpoint.
        :return: output, list of json responses. Note the code will raise an error if no ouptut as defined in the above
        method.
        """
        existing_rules = cls.get_rules()
        cls.delete_all_rules(rules=existing_rules)
        cls.set_rules(rules=my_rules)
        output = cls.get_stream(breaker=breaker)
        return output


if __name__ == "__main__":
    TwitterStream = TwitterStream()
    result = TwitterStream.main(my_rules=[
            {"value": "dog has:images", "tag": "dog pictures"},
            {"value": "cat has:images -grumpy", "tag": "cat pictures"},
        ], breaker=3)

我认为您向 API 发出请求的语法是错误的 - 请查看他们的示例代码 here 用于此端点。

我不太确定您要做什么，因此要更具体一些比较棘手，但我会先粘贴他们的示例代码并使用您的不记名令牌。