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

问题描述

我正在尝试使用以下代码向 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 提出请求。我不确定我做错了什么。我非常感谢您的帮助。

解决方法

编辑：这是一个代码示例的重写，希望能让这更容易理解。

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",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",)
        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",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",]
        :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",],breaker=3)

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

我不太确定您要做什么，因此很难说得更具体，但我会先粘贴他们的示例代码并使用您的不记名令牌。