我想嘗試製作一個簡單的Twitter客戶端,它可以瞭解我的口味,並自動查找朋友和有趣的推文,爲我提供相關信息。要開始,我需要獲得一個很好的隨機twitter消息流,所以我可以測試一些機器學習算法。從twitter獲取穩定的消息流
我應該使用哪些API方法?我是否需要定期輪詢以獲取消息,或者有沒有辦法讓twitter在發佈消息時推送消息?
我也有興趣瞭解任何類似的項目。
我想嘗試製作一個簡單的Twitter客戶端,它可以瞭解我的口味,並自動查找朋友和有趣的推文,爲我提供相關信息。要開始,我需要獲得一個很好的隨機twitter消息流,所以我可以測試一些機器學習算法。從twitter獲取穩定的消息流
我應該使用哪些API方法?我是否需要定期輪詢以獲取消息,或者有沒有辦法讓twitter在發佈消息時推送消息?
我也有興趣瞭解任何類似的項目。
我使用tweepy來訪問Twitter API並傾聽他們提供的public stream - 這應該是所有推文的百分之一樣本。這是我使用自己的示例代碼。您仍然可以使用流式傳輸的基本身份驗證機制,儘管它們很快就會改變。相應地更改USERNAME和PASSWORD變量,並確保尊重Twitter返回的錯誤代碼(此示例代碼可能不符合Twitter在某些情況下所需的指數退避機制)。
import tweepy
import time
def log_error(msg):
timestamp = time.strftime('%Y%m%d:%H%M:%S')
sys.stderr.write("%s: %s\n" % (timestamp,msg))
class StreamWatcherListener(tweepy.StreamListener):
def on_status(self, status):
print status.text.encode('utf-8')
def on_error(self, status_code):
log_error("Status code: %s." % status_code)
time.sleep(3)
return True # keep stream alive
def on_timeout(self):
log_error("Timeout.")
def main():
auth = tweepy.BasicAuthHandler(USERNAME, PASSWORD)
listener = StreamWatcherListener()
stream = tweepy.Stream(auth, listener)
stream.sample()
if __name__ == '__main__':
try:
main()
except KeyboardInterrupt:
break
except Exception,e:
log_error("Exception: %s" % str(e))
time.sleep(3)
我還設置了套接字模塊的超時時間,我相信我在Python中對默認的超時行爲有一些問題,所以要小心。
import socket
socket.setdefaulttimeout(timeout)
我不認爲你可以訪問世界Twitter的時間表。但是你當然可以看看你的朋友的推文和設置列表玩,我會推薦使用Twitter4J庫http://twitter4j.org/en/index.html
我可能被誤認爲,getPublicTimeline()可能是你想要的。
Twitter有一個streaming API就是爲了這個目的。他們提供了一個隨機發送給Twitter的所有消息的小樣本,並按照您所描述的「推送」方式不斷更新。如果你爲了某種崇高的目的而這樣做,那麼你可以從Twitter向request access更大的樣本。
從API文檔,你想statuses/sample
:
狀態/樣品
返回所有公共狀態的隨機抽樣 。 默認訪問級別'Spritzer' 提供了一小部分的流量,非常粗略地說,是所有 公共狀態的1%。 「Gardenhose」 訪問級別提供了一個比例更適合數據挖掘的 和 需要更大比例的研究應用程序是統計學上顯着的樣本 。目前 Gardenhose返回,非常粗略地,所有公共狀態的10% 。請注意, 這些比例受制於 突擊調整,因爲流量 體積不一。
網址:http://stream.twitter.com/1/statuses/sample.json
方法(S):GET
參數:count,分隔
返回:狀態元素的流
就個人而言,我已經取得了一些成功使用python庫tweepy來使用流API。
import tweepy, sys, time
ckey = ''
csecret = ''
atoken = ''
asecret = ''
def log_error(msg):
timestamp = time.strftime('%Y%m%d:%H%M:%S')
sys.stderr.write("%s: %s\n" % (timestamp,msg))
class StreamWatcherListener(tweepy.StreamListener):
def on_data(self, status):
try: #Some of the object are deletion of tweet, won't have 'text' in the dict
print getData['text']
except Exception, e:
pass
#print text.encode('utf-8')
def on_error(self, status_code):
log_error("Status code: %s." % status_code)
time.sleep(3)
return True # keep stream alive
def on_timeout(self):
log_error("Timeout.")
def main():
auth = tweepy.OAuthHandler(ckey, csecret)
auth.set_access_token(atoken, asecret)
listener = StreamWatcherListener()
stream = tweepy.Stream(auth, listener)
stream.sample()
if __name__ == '__main__':
try:
main()
except Exception,e:
log_error("Exception: %s" % str(e))
time.sleep(3)
Tweepy的BasicAuthHandler已被棄用。這是一套新的代碼。玩的開心!