Extrahovanie dát z Twitteru pomocou Pythonu

Akshay Nagpal2018-04-28 · 11 min read

Extrahovanie dát z Twitteru pomocou Pythonu

Twitter nie je len rozšíreným zdrojom správ, je to zďaleka jedna z najlepších vzoriek myšlienok sveta. S viac ako 330 miliónmi aktívnych používateľov je to jedna z popredných platforiem, kde ľudia radi zdieľajú svoje myšlienky. Údaje z Twitteru sa dajú použiť na rôzne účely, ako napríklad výskum, poznatky o spotrebiteľoch, demografické poznatky, a mnohé ďalšie.

Hlavným cieľom tohto návodu je preto naučiť vás, ako získať vzorku údajov z Twitteru relevantnú pre váš projekt alebo podnikanie.

Pred pokračovaním sa uistite, že máte po ruke všetky tieto premenné:

Consumer Key
Consumer Secret
Access Token
Access Token Secret

Ak chcete vedieť, ako získať vyššie uvedené údaje, prečítajte si tento príspevok na blogu, ktorý napísal môj kolega Dattatray Upase.

Teraz poďme trochu kódovať!

Definovanie vstupných premenných

Najprv musíte definovať niektoré globálne premenné, ktoré budete pre program potrebovať:

import sys

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

consumerKey="Sem_zadajte_svoj_Consumer_Key"
consumerSecret="Sem_zadajte_svoj_Consumer_Secret"
accessToken="Sem_zadajte_svoj_Access_Token"
accessTokenSecret="Sem_zadajte_svoj_Access_Token_Secret"

keyword= sys.argv[3] #"tcs"
lang="en" #pozrite sa, čo twitter ponúka na filtrovanie jazykov

data={}

import sys

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

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

consumerKey="Sem_zadajte_svoj_Consumer_Key"

consumerSecret="Sem_zadajte_svoj_Consumer_Secret"

accessToken="Sem_zadajte_svoj_Access_Token"

accessTokenSecret="Sem_zadajte_svoj_Access_Token_Secret"

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

lang="en" #pozrite sa, čo twitter ponúka na filtrovanie jazykov

data={}

Importujem ‘sys’ na získanie argumentov príkazového riadka, pretože by som mohol chcieť zmeniť kľúčové slová, počiatočný dátum (start-date) alebo koncový dátum (end-date). Pre jazyk som vybral angličtinu, ale možno budete chcieť skontrolovať, aké ďalšie jazyky sú podporované. Výsledky budú uložené v ‘data’ na konci.

V dôsledku toho by typické použitie skriptu vyzeralo takto:

python script.py start_date end_date keyword

Prístup k 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 umožňuje používanie svojho API prostredníctvom autorizačného rámca oauth2. Na svojej webovej stránke poskytujú dokumentáciu, pomocou ktorej je možné vykonávať filtrovanie tweetov. Vyššie uvedený kód by to pre vás implementoval.

Premenná req_count predstavuje počet, koľkokrát som použil API pri spúšťaní môjho programu. Pri uvedenom kóde sa stretávam s nasledujúcou chybou:

TypeError: Unicode-objects must be encoded before hashing

Aby som tomu predišiel, mením post_body=”” na post_body=b”” a to problém vyrieši.

Použitie a referenčná príručka Twitter API

Je čas nastaviť URL adresu API na získanie údajov z Twitteru. Používam parameter min_faves. Tu je vysvetlenie URL adresy a niekoľko optimalizačných trikov:

‘min_faves’ sa používa na nastavenie minimálneho počtu obľúbených položiek, ktoré by mal tweet v údajoch mať. Je to veľmi užitočná funkcia, ale nie je spomenutá v dokumentácii k Twitter API.

‘q’ predstavuje dopyt alebo kľúčové slová, ktoré chcete zadať. Tu je dôležité uistiť sa, že zadáte čo najmenej kľúčových slov. Predstavme si napríklad, že chcem tweety o Facebooku a Google. Ak zadám obe ako kľúčové slová, povedzme FACEBOOK a GOOGLE, vráti mi to maximálne 100 tweetov, keďže to je obmedzenie. Ale ak dopyt spustím dvakrát – raz s Facebookom a raz s Google, môžem získať celkovo 200 tweetov. Skrátka a dobre, je lepšie použiť jedno kľúčové slovo na dopyt.

‘lang’ predstavuje jazyk filtrovaných tweetov. Keďže chcem získať tweety v angličtine, nastavujem ho na ‘en’.

‘since’ je počiatočný dátum obdobia, od ktorého chcete vyhľadávať tweety. Tento počiatočný dátum by mal byť z posledných 7 dní. Toto je ďalšia funkcia, ktorá nie je zdokumentovaná v dokumentácii k Twitter API.

‘until’ predstavuje koncový dátum vami požadovaného obdobia. Logicky by mal byť tiež z posledných 7 dní. Taktiež nie je zdokumentovaný v dokumentácii k Twitter API.

‘result_type’ predstavuje druh tweetov, ktoré chcete. Má 3 hodnoty:

‘recent’ poskytuje najnovšie tweety, t. j. tweety na konci vybraného obdobia.

‘popular’ poskytuje najpopulárnejšie tweety, a preto vynecháva veľa tweetov. Vždy by ste dostali tweety s najvyšším počtom obľúbení (faves) a retweetov. Funkcia min_faves by tu bola zbytočná.

‘mixed’ poskytuje zmes najnovších a populárnych tweetov.

‘count’ predstavuje maximálny počet tweetov vo výsledku. Predvolená hodnota je nastavená na 15 a maximum je 100.

S kombinovaným result_type a použitím min_faves, môžeme získať maximálny počet tweetov spustením dopytu viackrát.

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')

Pre viac zdokumentovaných funkcií si môžete pozrieť aj dokumentáciu k Twitter API.

Ukladanie/automatické ukladanie získaných údajov o tweetoch

Ako ďalší krok musíte definovať metódu automatického ukladania/ukladania, ktorá má parameter ‘saveOverride’. Tento krok je potrebný jednoducho na odstránenie časového obmedzenia automatického ukladania a uloženie súboru. Aby som to urobil, vytvorím t_last na uloženie času spustenia programu. Potom k nemu pristúpim v programe a skontrolujem, či od t_last (posledného času uloženia) uplynulo viac ako 5 minút. Ak uplynulo viac ako 5 minút, označím ‘saveStatus’ ako True.

Ďalej skontrolujem ‘saveOverride’, čo jednoducho znamená, že musím dať svojmu programu inštrukciu, že bez ohľadu na čokoľvek sa má súbor teraz uložiť. Na tento účel nastavím saveStatus na True.

Potom, ak je ‘saveStatus’ True, skript zmení t_last na aktuálny čas. Potom kód vytvorí objekt slovníka a vypíše “Autosave v [time]” aby ste vedeli, že údaje sa automaticky ukladajú.

Ďalej kontrolujem, či výstupný súbor už existuje. Ak áno, spájam údaje a údaje z už uloženého súboru. Po spojení ich zapíšem do rovnakého súboru. Ak neexistuje, vytvorím nový súbor a potom doň zapíšem údaje.

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("Automatické ukladanie o " + 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("Automatické ukladanie o " + 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)

Výsledkom je, že som nakódoval takmer celú požadovanú funkcionalitu.

Ďalšia optimalizácia

Nakoniec je čas tieto funkcie využiť. Píšem cyklus while(1), čo znamená while(True). To v podstate spôsobuje, že inštrukcia beží navždy, kým sa nezavolá príkaz ‘break’ alebo sa nevyvolá nejaká výnimka.

Najprv zhromaždím dáta o tweetoch v slovníku ‘d’ pomocou json.loads, čo skonvertuje dáta do formátu slovníka. Potom spustím try-catch/except na bloku kódu, ktorý extrahuje stavy z dát. Používam try-catch, pretože niekedy Twitter API nevráti dáta, ale JSON s chybou. Nechcem, aby sa môj program v takýchto prípadoch zastavil. Taktiež chcem, aby zistil, pri ktorom čísle požiadavky k tomu dochádza, a uložil moje dáta z Twitteru pomocou príkazu autosave. Twitter nám umožňuje odoslať 180 požiadaviek za 15 minút. To je približne 12 požiadaviek za minútu alebo jedna požiadavka každých päť sekúnd. Pre istotu pridávam príkaz sleep, aby môj program po vykonaní jednej iterácie na 5 sekúnd zaspal.

Potom kód zobrazí počet tweetov, ktoré skript doteraz zhromaždil.

Nakoniec je čas na hlavný optimalizačný trik. Tento skript som testoval takmer týždeň a pre každú hodnotu min_faves som získal nasledujúci počet tweetov. Na jednu požiadavku môžem získať maximálne 100 tweetov a chcem ich získať čo najviac. V súčasnosti nie je veľa tweetov s vyššími hodnotami min_faves, ale chceme zohľadniť časy, keď je napríklad spoločnosť alebo kľúčové slovo populárne (trenduje). Maximálna hodnota min_faves môže byť 999999.

Hodnota min_faves	Počet tweetov
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

Preto používam logiku, ktorá by získala tweety od min_faves s hodnotou 60000 a potom ju zakaždým zníži o 10000, až kým nedosiahne 10000. Ale ak je napríklad kľúčové slovo populárne (trenduje) a získam 100 tweetov, keď pracujem s hodnotou min_faves nastavenou na 30000, zvýši to min_faves na 35000 a potom znova získať výsledky. Takže teraz je nová logika 5000 namiesto 10000. Ak sa však zmena zníži na menej ako 1000, požiadam o ignorovanie a pokračovanie s odčítaním 1000.

Určujem pevný interval 1000, o ktorý min_faves by sa mal znížiť, ak min_faves je menej alebo rovné 10000.

Na konci programu vás program upozorní, že práca je hotová, zobrazením ‘End’.

min_faves=60000
change=10000 #vysoké zníženie min_faves na získanie údajov
interval = 500 #bežné zníženie min_faves na získanie údajov
          
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("Chyba pri požiadavke : " + str(req_count))
        autosave(True)
      
        
    
    print("Pri požiadavke: " + str(req_count) + "  Celkovo zhromaždených tweetov: " + str(len(data)) + " s 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 #vysoké zníženie min_faves na získanie údajov

interval = 500 #bežné zníženie min_faves na získanie údajov

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("Chyba pri požiadavke : " + str(req_count))

autosave(True)

print("Pri požiadavke: " + str(req_count) + " Celkovo zhromaždených tweetov: " + str(len(data)) + " s 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")

Celý kód nájdete na GitHub.

To je všetko. V ďalšom návode na prácu s dátami z Twitteru vás naučím, ako získavať tweety v reálnom čase pomocou nástroja na spracovanie veľkých dát ‘Flume’. Zostaňte naladení!

Akshay Nagpal

Autor · CloudSigma

Preslav Dobrev je kreatívny dizajnér v spoločnosti CloudSigma, ktorý sa zameriava na konzistentnú firemnú identitu prostredníctvom tradičných a inovatívnych marketingových kanálov. Dokáže brilantne spájať umeleckú víziu so strategickým marketingom, čím vytvára pôsobivé príbehy značky.