Извлечение данных Twitter с помощью Python

Akshay Nagpal2018-04-28 · 11 min read

Извлечение данных Twitter с помощью Python

Twitter — это не просто расширенный источник новостей, это, безусловно, один из лучших срезов мыслей всего мира. С более чем 330 миллионами активных пользователей это одна из ведущих платформ, где люди любят делиться своими мыслями. Данные 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 разрешает использование своего API через платформу авторизации oauth2. Они предоставляют документацию на своем веб-сайте, с помощью которой можно выполнять фильтрацию твитов. Приведенный выше фрагмент кода реализует это для вас.

Переменная req_count указывает на то, сколько раз я использовал API при выполнении своей программы. Я столкнулся со следующей ошибкой в предоставленном коде:

TypeError: Unicode-objects must be encoded before hashing

Чтобы избежать этого, я меняю post_body=”” на post_body=b”” , и это решает проблему.

Использование и справочник Twitter API

Пришло время настроить URL-адрес API для получения данных 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 API.

Сохранение/автосохранение полученных данных твитов

В качестве следующего шага вам нужно определить метод автосохранения/сохранения, который имеет параметр ‘saveOverride’. Этот шаг просто необходим для снятия ограничения по времени автосохранения и сохранения файла. Для этого я создаю t_last для сохранения времени запуска программы. Затем я обращаюсь к нему в программе и проверяю, прошло ли 5 минут с момента t_last (времени последнего сохранения). Если прошло более 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 позволяет нам делать 180 запросов каждые 15 минут. Это примерно 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. Но если, скажем, ключевое слово популярно и я получаю 100 твитов при работе со значением min_faves равным 30000, это увеличит 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("Ошибка в запросе: " + str(req_count))
        autosave(True)
      
        
    
    print("В запросе: " + str(req_count) + "  Всего собрано твитов: " + str(len(data)) + " с 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("Ошибка в запросе: " + str(req_count))

autosave(True)

print("В запросе: " + str(req_count) + " Всего собрано твитов: " + str(len(data)) + " с 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, сосредоточенный на формировании последовательного корпоративного образа с помощью традиционных и инновационных маркетинговых каналов. Он умело сочетает художественное видение со стратегическим маркетингом, создавая убедительные истории бренда.