我需要获取经过过滤的 twitter 流样本
我正在使用 tweepy 我检查了 Stream 类的函数以获取样本流并进行过滤
但我不知道应该如何设置课程
应该是
stream.filter(track=['']).sample()
stream.sample().filter(track=[''])
或每一个在一行中或什么
如果您对如何根据关键字过滤器获取示例流有其他想法，请提供帮助
提前致谢
【问题讨论】：
标签： python tweepy twitter-streaming-api
Twitter v2 API 包括随机抽样端点和过滤推文端点。
后者允许在查询中指定随机样本百分比 (for example, sample:10 will return a random 10% sample)。
请注意，v2 API 仍然是新的，目前有 a cap of 500k tweets per month。
作为后者的示例，以下代码（modified version of this、see this doc）将收集带有猫或狗标签的流数据，并将其存储在每 100 条推文的 json 文件中。 （注意：这不包括随机抽样查询。）
import requests
import os
import json
import pandas as pd
# To set your enviornment variables in your terminal run the following line:
# export 'BEARER_TOKEN'='<your_bearer_token>'
data = []
counter = 0
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):
global data, counter
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)
print(json.dumps(json_response, indent=4, sort_keys=True))
data.append(json_response['data'])
if len(data) % 100 == 0:
print('storing data')
pd.read_json(json.dumps(data), orient='records').to_json(f'tw_example_{counter}.json', orient='records')
data = []
counter +=1
def main():
bearer_token = os.environ.get("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()
然后，将 pandas 数据框中的数据加载为
df = pd.read_json('tw_example.json', orient='records').
【讨论】：
我建议阅读 tweepy 的 api 文档。 Here你可以看到如何过滤你想要的流。
通过阅读其他代码 sn-ps，我相信应该这样做：
stream.filter(track=['Keyword'])
print(stream.sample())
【讨论】：
据我了解，tweepy 使用 twitter v1.1 API，它具有用于实时采样和过滤推文的单独 API。
Twitter API 参考。 v1 sample-realtime v1 filter-realtime
方法一：可以使用stream.filter(track=['Keyword1', 'keyord2'])等获取过滤后的流数据，然后从收集的数据中采样记录。
class StreamListener(tweepy.StreamListener):
def on_status(self, status):
# do data processing and storing here
参见https://www.storybench.org/how-to-collect-tweets-from-the-twitter-streaming-api-using-python/Ignoring Retweets When Streaming Twitter Tweets等示例
方法 2：可以编写以随机时间间隔启动和停止流式传输的程序（例如，每 15 分钟随机采样 3 分钟间隔）。
方法3：可以改为使用采样API收集数据，然后用关键字过滤以存储相关数据。
【讨论】：