Twitter API v2 如何获取所有推文的列表而不是流式传输它们
Twitter API v2 How to get a list of all tweets instead of streaming them
我是 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 库)?预先感谢您的帮助。
访问 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,这是一种支持的访问方式。
我是 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 库)?预先感谢您的帮助。
访问 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,这是一种支持的访问方式。