Twitter 的替代品 API
Alternative to Twitter API
我正在做一个项目,我从 Twitter API 流式传输推文,然后应用情绪分析并在交互式彩色地图上可视化结果。
我试过 python 中的 'tweepy' 库,但问题是它只能检索少量推文(10 条或更少)。
此外,我将指定语言和位置,这意味着我可能会收到更少的推文!我需要 hundred/thousands 推文的实时流式传输。
这是我试过的代码(以防万一):
import os
import tweepy
from textblob import TextBlob
port = os.getenv('PORT', '8080')
host = os.getenv('IP', '0.0.0.0')
# Step 1 - Authenticate
consumer_key= 'xx'
consumer_secret= 'xx'
access_token='xx'
access_token_secret='xx'
auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
auth.set_access_token(access_token, access_token_secret)
api = tweepy.API(auth)
#Step 3 - Retrieve Tweets
public_tweets = api.search('school')
for tweet in public_tweets:
print(tweet.text)
analysis = TextBlob(tweet.text)
print(analysis)
有没有更好的选择?我发现 "PubNub" 是 JavaScript API 但现在我想要 python 中的东西,因为它对我来说更容易。
谢谢
如果您想要大量推文,我建议您使用 Twitter 的流媒体 API 使用 tweepy
:
#Create a stream listner:
import tweepy
tweets = []
class MyStreamListener(tweepy.StreamListener):
#The next function defines what to do when a tweet is parsed by the streaming API
def on_status(self, status):
tweets.append(status.text)
#Create a stream:
myStreamListener = MyStreamListener()
myStream = tweepy.Stream(auth = api.auth, listener=myStreamListener)
#Filter streamed tweets by the keyword 'school':
myStream.filter(track=['school'], languages=['en'])
请注意,此处使用的跟踪过滤器是标准的免费过滤 API,其中还有另一个 API 称为 PowerTrack,它是为有更多要求和规则过滤的企业构建的。
参考:https://developer.twitter.com/en/docs/tweets/filter-realtime/overview/statuses-filter
否则,如果你想坚持search
方法,你可以通过添加count
查询最多100条推文,并在解析的最大id上使用since_id
来获取新的推文,您可以将这些属性添加到 search
方法,如下所示:
public_tweets = []
max_id = 0
for i in range(10): #This loop will run 10 times you can play around with that
public_tweets.extend(api.search(q='school', count=100, since_id=max_id))
max_id = max([tweet.id for tweet in public_tweets])
#To make sure you only got unique tweets, you can do:
unique_tweets = list({tweet._json['id']:tweet._json for tweet in public_tweets}.values())
这样你就必须小心 API 的限制,你必须在初始化 API 时通过启用 wait_on_rate_limit
属性来处理它:api = tweepy.API(auth,wait_on_rate_limit=True)
我正在做一个项目,我从 Twitter API 流式传输推文,然后应用情绪分析并在交互式彩色地图上可视化结果。
我试过 python 中的 'tweepy' 库,但问题是它只能检索少量推文(10 条或更少)。
此外,我将指定语言和位置,这意味着我可能会收到更少的推文!我需要 hundred/thousands 推文的实时流式传输。
这是我试过的代码(以防万一):
import os
import tweepy
from textblob import TextBlob
port = os.getenv('PORT', '8080')
host = os.getenv('IP', '0.0.0.0')
# Step 1 - Authenticate
consumer_key= 'xx'
consumer_secret= 'xx'
access_token='xx'
access_token_secret='xx'
auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
auth.set_access_token(access_token, access_token_secret)
api = tweepy.API(auth)
#Step 3 - Retrieve Tweets
public_tweets = api.search('school')
for tweet in public_tweets:
print(tweet.text)
analysis = TextBlob(tweet.text)
print(analysis)
有没有更好的选择?我发现 "PubNub" 是 JavaScript API 但现在我想要 python 中的东西,因为它对我来说更容易。
谢谢
如果您想要大量推文,我建议您使用 Twitter 的流媒体 API 使用 tweepy
:
#Create a stream listner:
import tweepy
tweets = []
class MyStreamListener(tweepy.StreamListener):
#The next function defines what to do when a tweet is parsed by the streaming API
def on_status(self, status):
tweets.append(status.text)
#Create a stream:
myStreamListener = MyStreamListener()
myStream = tweepy.Stream(auth = api.auth, listener=myStreamListener)
#Filter streamed tweets by the keyword 'school':
myStream.filter(track=['school'], languages=['en'])
请注意,此处使用的跟踪过滤器是标准的免费过滤 API,其中还有另一个 API 称为 PowerTrack,它是为有更多要求和规则过滤的企业构建的。
参考:https://developer.twitter.com/en/docs/tweets/filter-realtime/overview/statuses-filter
否则,如果你想坚持search
方法,你可以通过添加count
查询最多100条推文,并在解析的最大id上使用since_id
来获取新的推文,您可以将这些属性添加到 search
方法,如下所示:
public_tweets = []
max_id = 0
for i in range(10): #This loop will run 10 times you can play around with that
public_tweets.extend(api.search(q='school', count=100, since_id=max_id))
max_id = max([tweet.id for tweet in public_tweets])
#To make sure you only got unique tweets, you can do:
unique_tweets = list({tweet._json['id']:tweet._json for tweet in public_tweets}.values())
这样你就必须小心 API 的限制,你必须在初始化 API 时通过启用 wait_on_rate_limit
属性来处理它:api = tweepy.API(auth,wait_on_rate_limit=True)