如何解决筛选的Twitter API流样本
我需要获取Twitter流的过滤样本
我正在使用tweepy 我检查了Stream类的函数以获取示例流并进行过滤
但是我不知道该如何设置课程
应该
stream.filter(track=['']).sample()
stream.sample().filter(track=[''])
或一行中的每一个或什么
如果您还有其他想法如何基于关键字过滤器获取示例流，请帮助
预先感谢
解决方法
我建议阅读tweepy的api文档。 Here，您可以了解如何根据需要过滤流。
通过阅读其他代码段，我相信应该这样做：
stream.filter(track=['Keyword'])
print(stream.sample())
据我了解，tweepy使用twitter v1.1 API，该API具有用于实时采样和过滤推文的单独API。
Twitter API参考。 v1 sample-realtime v1 filter-realtime
方法1：可以使用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收集数据，然后使用关键字进行过滤以存储相关数据。
,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）将收集带有cat或dog标签的流数据，并每100条tweet将其存储在json文件中。 （注意：这不包括随机抽样查询。）
services.AddDefaultIdentity<UserProfile>(options => {
options.SignIn.RequireConfirmedAccount = true;
options.SignIn.RequireConfirmedEmail = true;
options.Stores.MaxLengthForKeys = 64;
}).AddEntityFrameworkStores<OWSData>();
WebHostOptions webHost = configuration.GetSection("WebHost").Get<WebHostOptions>();
string root = webHost.Url;
services.AddIdentityServer((options) => {
options.PublicOrigin = root;
options.IssuerUri = root;
options.UserInteraction.LoginUrl = $"{root}/account/login";
options.UserInteraction.LogoutUrl = $"{root}/account/logout";
options.UserInteraction.ErrorUrl = $"{root}/account/error";
options.UserInteraction.ConsentUrl = $"{root}/account/terms-of-service";
options.UserInteraction.DeviceVerificationUrl = $"{root}/account/device-verification";
}).AddApiAuthorization<UserProfile,OWSData>(options => {
// Serilog.Log.Information("Clients: {@clients}",options.Clients);
});
services.AddScoped<ICurrentUser,CurrentUser>();
services.AddTransient<IReturnUrlParser,AuthReturnUrl>();
// Auth (JWT)
services.AddAuthentication(options => {
// options.DefaultAuthenticateScheme = CookieAuthenticationDefaults.AuthenticationScheme;
})
.AddCookie((options) => {
webHost.ConfigureCookie(CookieName,options.Cookie);
options.LoginPath = "/account/login";
options.AccessDeniedPath = "/account/denied";
options.LogoutPath = "/account/logout";
options.SlidingExpiration = true;
})
.AddIdentityServerJwt()
.AddGoogle(options => {
ConfigureOAuth(options,"Google",configuration,webHost);
})
.AddGitHub(options => {
ConfigureOAuth(options,"GitHub",webHost);
options.Scope.Add("user:email");
options.EnterpriseDomain = configuration["GitHub:EnterpriseDomain"];
});
services.AddAuthorization(options => {
// options.DefaultPolicy = new AuthorizationPolicy();
});
// Auth (Password)
services.Configure<IdentityOptions>(options => {
//Password settings
options.Password.RequireDigit = true;
options.Password.RequireLowercase = true;
options.Password.RequireNonAlphanumeric = true;
options.Password.RequireUppercase = true;
options.Password.RequiredLength = 6;
options.Password.RequiredUniqueChars = 1;
//Lockout settings
options.Lockout.DefaultLockoutTimeSpan = TimeSpan.FromMinutes(5);
options.Lockout.MaxFailedAccessAttempts = 5;
options.Lockout.AllowedForNewUsers = true;
//User settings
options.User.AllowedUserNameCharacters =
"abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-._@";
options.User.RequireUniqueEmail = false;
});
// Cookies
services.ConfigureApplicationCookie(options => {
webHost.ConfigureCookie(CookieName,options.Cookie);
});
然后，将熊猫数据框中的数据加载为
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",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",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,rules)
set = set_rules(headers,bearer_token)
get_stream(headers,bearer_token)
if __name__ == "__main__":
main()
。
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 dio@foxmail.com 举报，一经查实，本站将立刻删除。
