使用 Python 擷取 Twitter 資料

Akshay Nagpal2018-04-28 · 6 min read

Twitter 不僅僅是新聞的延伸來源，它絕對是世界上人們想法的最佳樣本之一。擁有超過 3.3 億活躍用戶，它是人們最喜歡分享想法的頂級平台之一。Twitter 數據可用於多種用途，例如研究, 消費者洞察, 人口統計洞察，以及更多。

因此，本教學的主要目的是教您如何獲取與您的專案或業務相關的 Twitter 數據樣本。

在繼續之前，請確保您已準備好所有這些變數：

Consumer Key
Consumer Secret
Access Token
Access Token Secret

如果您想知道如何獲取上述詳細資訊，請閱讀我的同事 Dattatray Upase 撰寫的那篇部落格文章.

現在讓我們來寫點程式碼吧！

定義輸入變數

首先，您必須定義程式所需的一些全域變數：

import sys

start_date = sys.argv[1] #"2018-01-09" 
end_date = sys.argv[2] #"2018-01-10" 

consumerKey="在此輸入您的_Consumer_Key"
consumerSecret="在此輸入您的_Consumer_Secret"
accessToken="在此輸入您的_Access_Token"
accessTokenSecret="在此輸入您的_Access_Token_Secret"

keyword= sys.argv[3] #"tcs"
lang="en" #查看 twitter 提供的語言過濾功能

data={}

import sys

start_date = sys.argv[1] #"2018-01-09"

end_date = sys.argv[2] #"2018-01-10"

consumerKey="在此輸入您的_Consumer_Key"

consumerSecret="在此輸入您的_Consumer_Secret"

accessToken="在此輸入您的_Access_Token"

accessTokenSecret="在此輸入您的_Access_Token_Secret"

keyword= sys.argv[3] #"tcs"

lang="en" #查看 twitter 提供的語言過濾功能

data={}

我正在匯入 ‘sys’ 來獲取命令列參數，因為我可能想要更改關鍵字、開始日期或結束日期。語言方面我選擇了英文，但您可能想查看支援哪些其他語言。結果將儲存在 ‘data’ 中。

因此，該指令碼的典型用法如下：

python script.py start_date end_date keyword

存取 Twitter API

import oauth2

req_count = 0

def oauth_req(url, http_method="GET", post_body=b"", http_headers=None):
    global req_count,consumerKey,consumerSecret,accessToken,accessTokenSecret
    req_count += 1
    
    consumer = oauth2.Consumer(key=consumerKey, secret=consumerSecret)
    token = oauth2.Token(key=accessToken, secret=accessTokenSecret)
    client = oauth2.Client(consumer, token)
    resp, content = client.request( url, method=http_method, body=post_body , headers=http_headers )
    return content

import oauth2

req_count = 0

def oauth_req(url, http_method="GET", post_body=b"", http_headers=None):

global req_count,consumerKey,consumerSecret,accessToken,accessTokenSecret

req_count += 1

consumer = oauth2.Consumer(key=consumerKey, secret=consumerSecret)

token = oauth2.Token(key=accessToken, secret=accessTokenSecret)

client = oauth2.Client(consumer, token)

resp, content = client.request( url, method=http_method, body=post_body , headers=http_headers )

return content

Twitter 允許透過 oauth2 授權框架使用其 API。他們在官方網站上提供了可用於過濾推文的說明文件。上面的這段程式碼將為您實現該功能。

該 req_count 變數是我在執行程式時使用該 API 的次數。使用給定的程式碼時，我遇到了以下錯誤：

TypeError: Unicode-objects must be encoded before hashing

為了避免這種情況，我將 post_body=”” 更改為 post_body=b”” ，這解決了問題。

Twitter API 使用與參考

是時候設定 API URL 來獲取 Twitter 數據了。我正在使用參數 min_faves。以下是該 URL 的說明以及一些優化技巧：

‘min_faves’ 是用來設定資料中推文應具備的最少最愛次數。這是一個非常實用的功能，但 Twitter API 文件中並未提及。

‘q’ 代表您想要輸入的查詢或關鍵字。在這裡，請務必確保提供儘可能少的關鍵字。例如，假設我想獲取關於 Facebook 和 Google 的推文。如果我將兩者都作為關鍵字（例如 FACEBOOK 和 GOOGLE），由於限制，它最多只會返回 100 條推文。但如果我執行兩次查詢——一次用 Facebook，一次用 Google，我總共可以獲得 200 條推文。簡而言之，每次查詢最好只使用一個關鍵字。

‘lang’ 代表篩選推文的語言。因為我想獲取英文推文，所以我將其設置為 ‘en’。

‘since’ 是您想要尋找推文的期間起始日期。此起始日期應在過去 7 天內。這是另一個未記錄在 Twitter API 文件中的功能。

‘until’ 代表您期望期間的結束日期。邏輯上，它也應該在過去 7 天內。這同樣未記錄在 Twitter API 文件中。

‘result_type’ 代表您想要的推文類型。它有 3 個值：

‘recent’ 提供最新的推文，即所選期間結束時的推文。

‘popular’ 提供最熱門的推文，因此會遺漏很多推文。您總是會獲得獲得最多最愛和轉推的推文。而 min_faves 功能在這裡將毫無用處。

‘mixed’ 提供最新和熱門推文的混合內容。

‘count’ 代表結果中推文的最大數量。預設值為 15，最大值為 100。

配合混合的 result_type 以及使用 min_faves，我們可以透過多次執行查詢來獲取最多的推文。

def get_tweets(min_faves):
    global keyword, start_date, end_date, lang
    return oauth_req( 'https://api.twitter.com/1.1/search/tweets.json?' + '&amp;amp;q=' + keyword + '&amp;amp;lang=' + lang + '%20since%3A' + start_date + '%20until%3A' + end_date + '%20min_faves%3A' + str(min_faves) +'&amp;amp;result_type=mixed&amp;amp;count=100')

def get_tweets(min_faves):

global keyword, start_date, end_date, lang

return oauth_req( 'https://api.twitter.com/1.1/search/tweets.json?' + '&amp;q=' + keyword + '&amp;lang=' + lang + '%20since%3A' + start_date + '%20until%3A' + end_date + '%20min_faves%3A' + str(min_faves) +'&amp;result_type=mixed&amp;count=100')

如需了解更多已記錄的功能，您也可以查看 Twitter’s API 文件.

儲存/自動儲存獲取的推文資料

下一步，您需要定義一個具有參數 ‘saveOverride’ 的自動儲存/儲存方法。此步驟僅用於移除自動儲存時間限制並儲存檔案。為此，我建立了一個 t_last 來儲存程式的啟動時間。然後我在程式中存取該變數，並檢查自 t_last（上一次儲存時間）以來是否已過去 5 分鐘。如果已超過 5 分鐘，我會將 ‘saveStatus’ 標記為 True。

接下來，我檢查 ‘saveOverride’，這僅意味著我需要向我的程式發出指令，無論如何現在都應該儲存檔案。為此，我將 saveStatus 設置為 True。

然後，如果 ‘saveStatus’ 為 True，腳本將會把 t_last 更改為目前時間。之後，程式碼會建立一個字典物件並列印 “在 [time] 自動儲存”，以便您知道資料正在自動儲存。

接下來，我會檢查輸出檔案是否已存在。如果存在，我會將目前的資料與已儲存檔案中的資料進行合併。合併後，我會將其寫入同一個檔案。如果不存在，我會建立一個新檔案，然後將資料寫入該檔案。

import os
import json
import time
import datetime

t_last = time.time()

def autosave(saveOverride = False):
    global t_last
    saveStatus = (time.time() &amp;amp;gt; t_last + 300)
    if(saveOverride == True):en"
        saveStatus = True
        
    if(saveStatus):
        t_last=time.time()
        tmp = {}
        print("自動儲存於 " + str(datetime.datetime.now().strftime("%Y-%m-%d_%H:%M:%S")))
        fnamea = keyword + "-st-" + start_date + "-ed-"+ end_date + '.json'
        if os.path.exists(fnamea) == True:
            with open(fnamea,'r+') as f:
                tmp = json.load(f)
        for i in data.keys():
            tmp[i] = data[i]
        with open(fnamea,'w+') as f:
            json.dump(tmp,f)

import os

import json

import time

import datetime

t_last = time.time()

def autosave(saveOverride = False):

global t_last

saveStatus = (time.time() &amp;gt; t_last + 300)

if(saveOverride == True):en"

saveStatus = True

if(saveStatus):

t_last=time.time()

tmp = {}

print("自動儲存於 " + str(datetime.datetime.now().strftime("%Y-%m-%d_%H:%M:%S")))

fnamea = keyword + "-st-" + start_date + "-ed-"+ end_date + '.json'

if os.path.exists(fnamea) == True:

with open(fnamea,'r+') as f:

tmp = json.load(f)

for i in data.keys():

tmp[i] = data[i]

with open(fnamea,'w+') as f:

json.dump(tmp,f)

因此，我幾乎已經編寫了所有我想要的功能。

進一步最佳化

最後，是時候使用這些函式了。我正在編寫一個 while(1) 迴圈，這意味著 while(True)。這基本上會讓指令永遠執行，直到呼叫了某個 ‘break’ 陳述式或引發了某個例外。

首先，我將推文資料收集到字典 ‘d’ 中，使用 json.loads ，這會將資料轉換為字典格式。然後，我執行一個 try-catch/except 在從資料中擷取狀態的程式碼區塊上。我使用 try-catch 是因為有時 Twitter API 不會回傳資料，而是回傳一個提及錯誤的 JSON。我不希望我的程式在這種情況下停止。此外，我希望它能取得這是在第幾次請求時發生的，並使用自動儲存指令來儲存我的 Twitter 資料。Twitter 允許我們每 15 分鐘發送 180 次請求。這相當於每分鐘 12 次請求，或者每五秒一次請求。為了安全起見，我加入了一個 sleep 指令，讓我的程式在執行一次迭代後休眠 5 秒。

之後，程式碼將顯示該腳本目前為止收集到的推文數量。

最後，是時候使用主要的最佳化技巧了。我測試這個腳本將近一個星期，並獲得了以下針對每個 min_faves 值的推文數量。我每次請求最多可以獲取 100 則推文，並且我想盡可能獲取更多。目前，具有較高 min_faves 值的推文並不多，但我們希望考慮到公司或關鍵字可能正在流行的時候。 min_faves 的最大值可以是 999999。

min_faves 值	推文數量
100,000	1
90,000	1
80,000	1
70,000	2
60,000	3
50,000	6
40,000	6
30,000	12
25,000	12

因此，我使用了一種邏輯，該邏輯會從 min_faves 值為 60000 開始獲取推文，然後每次減少 10000，直到達到 10000。但如果說，該關鍵字很熱門，而當我處理 min_faves 值為 30000 時獲得了 100 則推文，它將會增加 min_faves 至 35000，然後再次獲取結果。所以現在的新邏輯是 5000 而不是 10000。然而，如果變化量減少到小於 1000，我會要求它忽略並繼續減去 1000。

我正在指定一個固定的間隔 1000，min_faves 應該減少，如果 min_faves 小於或等於 10000。

在程式結束時，程式會透過顯示 ‘End’ 來讓您知道工作已完成。

min_faves=60000
change=10000 #大幅減少 min_faves 以提取數據
interval = 500 #正常減少 min_faves 以提取數據
          
while(1): 
    d = json.loads(get_tweets(min_faves))
    try:
        for i in d['statuses']:
            data[i['id']] = i
        c = len(d['statuses'])
    except Exception as e:
        print("Error at request : " + str(req_count))
        autosave(True)
      
        
    
    print("At request: " + str(req_count) + "  Total Tweets Collected: " + str(len(data)) + " with Min Faves: " + str(min_faves) )
    if c==100 and min_faves&amp;gt;10000:
        if (change&amp;gt;1000):
            change /= 2
            min_faves += change
        else:
            min_faves -= change
        
    elif min_faves&amp;gt;10000:
        min_faves -= change
    
    else:
        min_faves -= interval
        if(min_faves &amp;lt; 0):
            fnamea = keyword + '.json'
            autosave(True)
            break
    autosave()
    time.sleep(5)
    
print("End")

min_faves=60000

change=10000 #大幅減少 min_faves 以提取數據

interval = 500 #正常減少 min_faves 以提取數據

while(1):

d = json.loads(get_tweets(min_faves))

try:

for i in d['statuses']:

data[i['id']] = i

c = len(d['statuses'])

except Exception as e:

print("Error at request : " + str(req_count))

autosave(True)

print("At request: " + str(req_count) + " Total Tweets Collected: " + str(len(data)) + " with Min Faves: " + str(min_faves) )

if c==100 and min_faves&gt;10000:

if (change&gt;1000):

change /= 2

min_faves += change

else:

min_faves -= change

elif min_faves&gt;10000:

min_faves -= change

else:

min_faves -= interval

if(min_faves &lt; 0):

fnamea = keyword + '.json'

autosave(True)

break

autosave()

time.sleep(5)

print("End")

您可以在 GitHub.

以上就是全部內容。在下一個 twitter 數據教學中，我將教您如何使用大數據工具 ‘Flume’ 獲取即時推文。敬請期待！

Akshay Nagpal

作者 · CloudSigma

Preslav Dobrev 是 CloudSigma 的創意設計師，專注於透過傳統與創新行銷渠道建立一致的企業形象。他擅長將藝術願景與策略行銷相融合，創造具有影響力的品牌敘事。