Ekstrakcja danych z Twittera za pomocą Pythona

Akshay Nagpal2018-04-28 · 11 min read

Ekstrakcja danych z Twittera za pomocą Pythona

Twitter to nie tylko rozszerzone źródło wiadomości, to zdecydowanie jedna z najlepszych próbek myśli świata. Z ponad 330 milionami aktywnych użytkowników jest to jedna z najpopularniejszych platform, na których ludzie chętnie dzielą się swoimi przemyśleniami. Dane z Twittera mogą być wykorzystywane do różnych celów, takich jak badania, analizy konsumenckie, analizy demograficzne, i wiele innych.

Dlatego głównym celem tego samouczka jest nauczenie Cię, jak pozyskać próbkę danych z Twittera istotnych dla Twojego projektu lub biznesu.

Przed kontynuowaniem upewnij się, że masz pod ręką wszystkie te zmienne:

Consumer Key
Consumer Secret
Access Token
Access Token Secret

Jeśli chcesz dowiedzieć się, jak uzyskać wyżej wymienione dane, przeczytaj ten wpis na blogu napisany przez mojego kolegę Dattatray Upase.

Teraz czas na kodowanie!

Definiowanie zmiennych wejściowych

Najpierw musisz zdefiniować niektóre zmienne globalne, które będą potrzebne w programie:

import sys

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

consumerKey="Enter_Your_Consumer_Key_Here"
consumerSecret="Enter_Your_Consumer_Secret_Here"
accessToken="Enter_Your_Access_Token_Here"
accessTokenSecret="Enter_Your_Access_Token_Secret_Here"

keyword= sys.argv[3] #"tcs"
lang="en" #zobacz, co twitter oferuje w zakresie filtrowania języków

data={}

import sys

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

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

consumerKey="Enter_Your_Consumer_Key_Here"

consumerSecret="Enter_Your_Consumer_Secret_Here"

accessToken="Enter_Your_Access_Token_Here"

accessTokenSecret="Enter_Your_Access_Token_Secret_Here"

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

lang="en" #zobacz, co twitter oferuje w zakresie filtrowania języków

data={}

Importuję ‘sys’ w celu pobrania argumentów wiersza poleceń, ponieważ mogę chcieć zmienić słowa kluczowe, datę początkową lub końcową. Jako język wybrałem angielski, ale możesz sprawdzić, jakie inne języki są obsługiwane. Wyniki zostaną zapisane w ‘data’ na końcu.

W rezultacie typowe użycie skryptu wyglądałoby następująco:

python script.py start_date end_date keyword

Dostęp do API Twittera

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żliwia korzystanie ze swojego API poprzez architekturę autoryzacji oauth2. Na swojej stronie internetowej udostępniają dokumentację, za pomocą której można filtrować tweety. Powyższy fragment kodu zaimplementuje to dla Ciebie.

Zmienna req_count określa liczbę przypadków użycia API podczas wykonywania mojego programu. W podanym kodzie napotykam następujący błąd:

TypeError: Unicode-objects must be encoded before hashing

Aby tego uniknąć, zmieniam post_body=”” na post_body=b”” i to rozwiązuje problem.

Użycie i dokumentacja API Twittera

Czas ustawić adres URL API, aby pobrać dane z Twittera. Używam parametru min_faves. Oto wyjaśnienie adresu URL i kilka wskazówek optymalizacyjnych:

‘min_faves’ służy do ustawiania minimalnej liczby polubień, jaką tweet powinien mieć w danych. To bardzo przydatna funkcja, ale nie jest wspomniana w dokumentacji Twitter API.

‘q’ reprezentuje zapytanie lub słowa kluczowe, które chcesz wprowadzić. Tutaj ważne jest, aby upewnić się, że podajesz jak najmniej słów kluczowych. Na przykład wyobraźmy sobie, że chcę tweety o Facebooku i Google. Jeśli podam oba jako słowa kluczowe, powiedzmy FACEBOOK i GOOGLE, zwróci mi to maksymalnie 100 tweetów, ponieważ takie jest ograniczenie. Ale jeśli uruchomię zapytanie dwukrotnie – raz z Facebookiem i raz z Google, mogę uzyskać łącznie 200 tweetów. Krótko mówiąc, lepiej używać jednego słowa kluczowego na zapytanie.

‘lang’ reprezentuje język filtrowanych tweetów. Ponieważ chcę otrzymywać tweety w języku angielskim, ustawiam go na ‘en’.

‘since’ to data początkowa okresu, z którego chcesz szukać tweetów. Ta data początkowa powinna mieścić się w ciągu ostatnich 7 dni. To kolejna funkcja, która nie jest udokumentowana w dokumentacji Twitter API.

‘until’ reprezentuje datę końcową żądanego okresu. Logicznie rzecz biorąc, powinna ona również mieścić się w ciągu ostatnich 7 dni. Nie jest ona również udokumentowana w dokumentacji Twitter API.

‘result_type’ reprezentuje rodzaj tweetów, które chcesz otrzymać. Ma 3 wartości:

‘recent’ zwraca najnowsze tweety, tj. tweety z końca wybranego okresu.

‘popular’ zwraca najpopularniejsze tweety, a co za tym idzie, pomija wiele tweetów. Zawsze otrzymasz tweety z największą liczbą polubień i retweetów. Funkcja min_faves byłaby tutaj bezużyteczna.

‘mixed’ daje mieszankę najnowszych i popularnych tweetów.

‘count’ reprezentuje maksymalną liczbę tweetów w wyniku. Domyślnie ustawiona jest na 15, a maksymalna to 100.

Dzięki mieszanemu result_type i użyciu min_faves, możemy uzyskać maksymalną liczbę tweetów, uruchamiając zapytanie wielokrotnie.

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

Więcej udokumentowanych funkcji można znaleźć w Dokumentacji API Twittera.

Zapisywanie/Autozapis pobranych danych tweetów

W kolejnym kroku musisz zdefiniować metodę autozapisu/zapisu, która posiada parametr ‘saveOverride’. Ten krok jest po prostu potrzebny, aby usunąć ograniczenie czasowe autozapisu i zapisać plik. Aby to zrobić, tworzę t_last, aby zapisać czas rozpoczęcia programu. Następnie uzyskuję do niego dostęp w programie i sprawdzam, czy minęło 5 minut od t_last (czasu ostatniego zapisu). Jeśli minęło więcej niż 5 minut, oznaczam ‘saveStatus’ jako True.

Następnie sprawdzam ‘saveOverride’, co oznacza po prostu, że muszę wydać programowi instrukcję, aby bez względu na wszystko zapisał plik w tym momencie. W tym celu ustawiam saveStatus na True.

Następnie, jeśli ‘saveStatus’ ma wartość True, skrypt zmieni t_last na aktualny czas. Następnie kod tworzy obiekt słownika i wypisuje “Autozapis o [time]”, abyś wiedział, że dane są automatycznie zapisywane.

Następnie sprawdzam, czy plik wyjściowy już istnieje. Jeśli tak, łączę dane z danymi z już zapisanego pliku. Po połączeniu zapisuję je do tego samego pliku. Jeśli nie istnieje, tworzę nowy plik, a następnie zapisuję do niego dane.

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("Autozapis 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("Autozapis 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)

W rezultacie zakodowałem prawie całą funkcjonalność, którą chciałem.

Dalsza optymalizacja

W końcu nadszedł czas na skorzystanie z tych funkcji. Piszę pętlę while(1), co oznacza while(True). To w zasadzie sprawia, że instrukcja działa w nieskończoność, dopóki nie zostanie wywołana instrukcja ‘break’ lub nie zostanie zgłoszony wyjątek.

Najpierw zbieram dane o tweetach w słowniku ‘d’ za pomocą json.loads co konwertuje dane na format słownika. Następnie uruchamiam try-catch/except na bloku kodu wyodrębniającym statusy z danych. Używam try-catch, ponieważ czasami Twitter API nie zwraca danych, lecz plik JSON zawierający informację o błędzie. Nie chcę, aby mój program zatrzymywał się w takich przypadkach. Chcę również, aby pobierał informację, przy którym numerze żądania to się dzieje, i zapisywał moje dane z Twittera za pomocą polecenia autosave. Twitter pozwala nam na wykonanie 180 żądań na 15 minut. To daje około 12 żądań na minutę lub jedno żądanie co pięć sekund. Na wszelki wypadek dodaję polecenie sleep, aby uśpić program na 5 sekund po wykonaniu jednej iteracji.

Następnie kod wyświetli liczbę tweetów, które skrypt zebrał do tej pory.

W końcu nadszedł czas na główną sztuczkę optymalizacyjną. Testowałem ten skrypt przez prawie tydzień i otrzymałem następującą liczbę tweetów dla każdej wartości min_faves. Mogę uzyskać maksymalnie 100 tweetów na żądanie i chcę uzyskać ich jak najwięcej. Obecnie nie ma zbyt wielu tweetów o wyższych wartościach min_faves, ale chcemy uwzględnić sytuacje, w których być może firma lub słowo kluczowe zyskuje na popularności. Maksymalna wartość min_faves może wynosić 999999.

Wartość min_faves	Liczba tweetów
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

Dlatego używam logiki, która pobierałaby tweety od min_faves o wartości 60000, a następnie zmniejsza ją o 10000 za każdym razem, aż osiągnie 10000. Ale jeśli powiedzmy, słowo kluczowe jest popularne i otrzymam 100 tweetów, gdy pracuję z wartością min_faves wynoszącą 30000, to zwiększy to min_faves do 35000, a następnie ponownie pobrać wyniki. Teraz nowa logika to 5000 zamiast 10000. Jeśli jednak zmiana spadnie poniżej 1000, każę programowi to zignorować i przejść do odejmowania 1000.

Określam stały interwał wynoszący 1000, o który min_faves powinien się zmniejszyć, jeśli min_faves jest mniejsze lub równe 10000.

Na końcu programu poinformuje on o zakończeniu pracy, wyświetlając ‘End’.

min_faves=60000
change=10000 #duża redukcja min_faves w celu wyodrębnienia danych
interval = 500 #normalna redukcja min_faves w celu wyodrębnienia danych
          
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("Błąd przy żądaniu : " + str(req_count))
        autosave(True)
      
        
    
    print("Przy żądaniu: " + str(req_count) + "  Łącznie zebranych tweetów: " + str(len(data)) + " z 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 #duża redukcja min_faves w celu wyodrębnienia danych

interval = 500 #normalna redukcja min_faves w celu wyodrębnienia danych

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("Błąd przy żądaniu : " + str(req_count))

autosave(True)

print("Przy żądaniu: " + str(req_count) + " Łącznie zebranych tweetów: " + str(len(data)) + " z 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")

Cały kod można znaleźć na GitHub.

To wszystko. W kolejnym samouczku dotyczącym danych z Twittera nauczę Cię, jak pobierać tweety w czasie rzeczywistym za pomocą narzędzia do big data ‘Flume’. Bądź na bieżąco!

Akshay Nagpal

Autor · CloudSigma

Preslav Dobrev jest projektantem kreatywnym w CloudSigma, skupiającym się na spójnej tożsamości biznesowej przy wykorzystaniu tradycyjnych i innowacyjnych kanałów marketingowych. Biegle łączy wizję artystyczną ze strategicznym marketingiem, tworząc wywierające wpływ narracje marki.