Twitter API v2 如何获取所有推文的列表而不是流式传输它们

Question

我是 Twitter API 的新手，我只是在使用 python 对其进行测试。这是我正在使用的代码（我从 Twitter 的 Github 获得的）：

import requests
import os
import json

def create_headers(bearer_token):
    headers = {"Authorization": "Bearer {}".format(bearer_token)}
    return headers


def get_rules(headers, bearer_token):
    response = requests.get(
        "https://api.twitter.com/2/tweets/search/stream/rules", headers=headers
    )
    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(headers, bearer_token, 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",
        headers=headers,
        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(headers, delete, bearer_token):
    # You can adjust the rules if needed
    sample_rules = [
        {"value": "dog has:images", "tag": "dog pictures"},
        {"value": "cat has:images -grumpy", "tag": "cat pictures"},
    ]
    payload = {"add": sample_rules}
    response = requests.post(
        "https://api.twitter.com/2/tweets/search/stream/rules",
        headers=headers,
        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(headers, set, bearer_token):
    response = requests.get(
        "https://api.twitter.com/2/tweets/search/stream", headers=headers, stream=True,
    )
    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)
            to_be_parsed = json.dumps(json_response, indent=4, sort_keys=True)
            a = json.loads(to_be_parsed)
            print(a['data']['text'])


def main():
    bearer_token = '<BEARER_TOKEN>'
    headers = create_headers(bearer_token)
    rules = get_rules(headers, bearer_token)
    delete = delete_all_rules(headers, bearer_token, rules)
    set = set_rules(headers, delete, bearer_token)
    get_stream(headers, set, bearer_token)


if __name__ == "__main__":
    main()

并且我想将获得的推文用于情绪分析项目，那么有什么方法可以获取基于某个关键字和某个时间范围的推文列表，而不是我得到的恒定流使用此代码（有点像 getoldtweets3 库）？提前感谢您的帮助。

Answer 1

访问 Twitter 的 API 有两种主要格式：

• 实时（流式传输）：这是您建立连接并继续监听之后发生的所有事情的地方。
• 历史 (RESTful)：您可以在此处查询当前或过去发生的内容，然后停止。

您使用流式连接所做的事情是说，请向我发送从现在开始发生的有关此模式/主题/查询的所有推文。它不允许你回顾过去。

您要求的是“从某个时间范围”向后看的能力。在 Twitter API v1.1 和现在的 Twitter API v2 中（更多即将推出），当前的选项是搜索与您的查询匹配的推文。搜索 API 最多支持过去 7 天，因此您不能要求，例如，从 2019 年 1 月到 2019 年 3 月的所有推文。为此，您需要查看完整存档高级搜索等商业 API。将来，v2 可能会启用更大的历史范围。

在今天的 API v2 中，您可以使用 Recent Search API sample 获取过去 7 天的推文。一些第三方 Python API 库现在也支持 API v2。

GetOldTweets 库使用网络抓取来获取数据，这正式违反了 Twitter 的服务条款。最好使用官方API，这是一种支持的访问方式。