DEV Community

loading...
Cover image for Coletando dados do Twitter com Python

Coletando dados do Twitter com Python

Mayara Machado
I'm very passionate about programming and currently I do it mostly using python. I'm a pythonist.
Updated on ・6 min read

Eu iniciei um projeto para realizar a coleta de alguns tweets utilizando a biblioteca Tweepy, tratamento de dados utilizando Python e análise de polaridade dos textos escritos nos tweets. Irei relatar um pouco do processo, do código e dos resultados que obtive na série de dois textos, inicialmente iria ser somente um texto mas no decorrer do processo da escrita achei que seria mais interessante dividir essas em etapas.

Dessa forma, temos esse primeiro momento onde eu mostro os códigos usados para coleta dos tweets e o código para tratar os dados. No segundo momento irei mostrar os resultados e algumas visualizações dos dados obtidos. Então vamos lá!

Coleta e limpeza dos tweets

A gente precisa ter, primeiramente, uma conta ativa no ambiente do twitter para desenvolvedores. Uma vez que sua solicitação é aceita e você já possue sua conta de desenvolvimento ativa, você precisa ir no Dashboard de sua conta e criar um App. Ao criar um App vinculado a sua conta, o Twitter irá gerar credenciais que serão usadas para realizar as chamadas à API deles, então guardem bem essas credencias e não compartilhem na internet.

Coleta dos tweets

Utilizando as credenciais que obtivemos, a gente já pode estabelecer a conexão através da Tweetpy.

import tweetpy

CONSUMER_KEY = YOUR_CONSUMER_KEY
CONSUMER_SECRET = YOUR_CONSUMER_SECRET
ACCESS_TOKEN = YOUR_ACCESS_TOKEN
ACCESS_TOKEN_SECRET = YOUR_ACCESS_TOKEN_SECRET

auth = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET)
auth.set_access_token(ACCESS_TOKEN, ACCESS_TOKEN_SECRET)

conToken = tweepy.API(auth, wait_on_rate_limit=True, wait_on_rate_limit_notify=True, retry_count=5, retry_delay=10)
Enter fullscreen mode Exit fullscreen mode

Agora já com a conexão estabelecida, nós iremos escrever o código para obter os tweets. Para a proposta em questão, o interessante seria buscar tweets públicos através de uma hashtag ou palavra chave mesmo. O Twitter possui alguns endpoints disponíveis, nós podemos realizar coletas diretamente em perfis de usuários, no nosso próprio perfil ou utilizar o endpoint de pesquisa.

Caso esteja na dúvida de qual endpoint é o mais interessante para o seu projeto, eu recomento que seja realizado uma leitura na documentação da API do Twitter. O Tweepy já possui a implementação para realizar a conexão com esse endpoint, para isso nós precisamos selecionar apenas alguns parâmetros que serão utilizados pelo Tweepy no momento da busca.

tweets_iter = tweepy.Cursor(conToken.search,
                            q=KEYWORD, tweet_mode='extended',
                            rpp=5000, result_type='mixed',
                            since=datetime(2020,7,31,0,0,0).date(),
                            lang='en', include_entities=True).items(count)
Enter fullscreen mode Exit fullscreen mode

Vamos dar uma olhada nos parâmetros utilizados.

  • tweet_mode='extended' : passamos esse parâmetro para explicitar que nós queremos receber o texto completo do tweet, caso esse parâmetro não seja passado os tweets retornados terão até no máximo 140 caracteres de texto.
  • rpp=5000 : é a quantidade limite que nós queremos receber de tweets, em algumas requisições por questões de timeout, dependendo da quantidade que você utilizar, é possível que seja retornado menos que a quantidade solicitada.
  • result_type='mixed': É possível receber 3 tipos de tweets, os recentes, os que tiveram maiores engajamentos como curtidas e retweets, ou uma mistura de ambos. Para definir isso basta especificar o result_type dentre ('mixed', 'recent', 'popular'), caso esse campo não seja passado o default é sempre o mixed. Decidi explicitar aqui nesse caso qual era a minha intenção.
  • since=datetime(2020,7,31,0,0,0).date(): É possível também delimitar o tempo com os parâmetros since e until, no meu caso eu só delimitei a data a partir pois queria apenas a partir de uma data em específico.

Limpeza e preparo dos tweets

Após obtenção de um objeto Iterator no código anterior, nós vamos iterar para acessar cada tweet obtido e realizar a limpeza. Inicialmente precisamos tratar os textos dos tweets, é nele que vamos realizar a extração de itens que não são necessariamente interessantes para a nossa análise.

Para essa remoção, eu utilizei regex simples que removem links que podem estar associados ao tweet, removem as menções a outros usuários. Além disso, a requisição retorna muitos dados sobre o tweet e não precisamos de tudo isso. Nesse exemplo eu deixei algumas colunas que podem ser interessantes para análises futuras.

import re

def clean_tweet(self, tweets_text):

    clean_text = re.sub(r'RT+', '', tweets_text) 
    clean_text = re.sub(r'@\S+', '', clean_text)  
    clean_text = re.sub(r'https?\S+', '', clean_text) 
    clean_text = clean_text.replace("\n", " ")

    return clean_text

def prepare_tweets_list(self, tweets_iter):

  tweets_data_list = []
  for tweet in tweets_iter:
    if not 'retweeted_status' in dir(tweet):
      tweet_text = clean_tweet(tweet.full_text)
      tweets_data = {
          'len' : len(tweet_text),
          'ID' : tweet.id,
          'User' : tweet.user.screen_name,
          'UserName' : tweet.user.name,
          'UserLocation' : tweet.user.location,
          'TweetText' : tweet_text,
          'Language' : tweet.user.lang,
          'Date' : tweet.created_at,
          'Source': tweet.source,
          'Likes' : tweet.favorite_count,
          'Retweets' : tweet.retweet_count,
          'Coordinates' : tweet.coordinates,
          'Place' : tweet.place 
      }

      tweets_data_list.append(tweets_data)

  return tweets_data_list
Enter fullscreen mode Exit fullscreen mode

Polaridade dos tweets

Por fim, nós também podemos utilizar a biblioteca TextBlob para realizar a análise de polaridade nos tweets que obtivemos. O TextBlob recebe uma string, que pode ser uma palavra ou uma frase completa, e retorna um inteiro no intervalo entre 0 e 1. Esse inteiro diz respeito a polaridade encontrada pela biblioteca.

Para melhorar a visualização e entendimento, eu realizei a substituição do valor da polaridade por uma string indicando em qual espectro aquele tweet se encaixava. Dessa forma valores menores que zero receberam Negative, valores maiores que zero receberam Positive. As vezes a biblioteca não consegue determinar uma polaridade a partir da string e retorna o valor zero, para esses casos eu considerei como Neutral.

from textblob import TextBlob

def sentiment_polarity(self, tweets_text_list):
      tweets_sentiments_list = []

      for tweet in tweets_text_list:
        polarity = TextBlob(tweet).sentiment.polarity
        if polarity > 0:
          tweets_sentiments_list.append('Positive')
        elif polarity < 0:
          tweets_sentiments_list.append('Negative')
        else:
          tweets_sentiments_list.append('Neutral')

      return tweets_sentiments_list
Enter fullscreen mode Exit fullscreen mode

Código final

Por fim, nós podemos organizar todo esse código escrito até então em uma única classe, a classe responsável pelos tweets que realiza coleta, tratamento de tweets.

import re
import tweepy
from datetime import datetime
from textblob import TextBlob

CONSUMER_KEY = YOUR_CONSUMER_KEY
CONSUMER_SECRET = YOUR_CONSUMER_SECRET
ACCESS_TOKEN = YOUR_ACCESS_TOKEN
ACCESS_TOKEN_SECRET = YOUR_ACCESS_TOKEN_SECRET

keyword = ("'Black is King' OR 'black is king' OR 'Beyonce' OR 'beyonce' OR #blackisking OR '#BlackIsKing' OR 'black is king beyonce'")
count = 5000

class TweetAnalyzer():

  def __init__(self, consumer_key, consumer_secret, access_token, access_token_secret):
    '''
      Conectar com o tweepy
    '''
    auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
    auth.set_access_token(access_token, access_token_secret)

    self.conToken = tweepy.API(auth, wait_on_rate_limit=True, wait_on_rate_limit_notify=True, retry_count=5, retry_delay=10)

  def __clean_tweet(self, tweets_text):
    '''
    Tweet cleansing.
    '''

    clean_text = re.sub(r'RT+', '', tweets_text) 
    clean_text = re.sub(r'@\S+', '', clean_text)  
    clean_text = re.sub(r'https?\S+', '', clean_text) 
    clean_text = clean_text.replace("\n", " ")

    return clean_text

  def search_by_keyword(self, keyword, count=10, result_type='mixed', lang='en', tweet_mode='extended'):
    '''
      Search for the twitters thar has commented the keyword subject.
    '''
    tweets_iter = tweepy.Cursor(self.conToken.search,
                          q=keyword, tweet_mode=tweet_mode,
                          rpp=count, result_type=result_type,
                          since=datetime(2020,7,31,0,0,0).date(),
                          lang=lang, include_entities=True).items(count)

    return tweets_iter

  def prepare_tweets_list(self, tweets_iter):
    '''
      Transforming the data to DataFrame.
    '''

    tweets_data_list = []
    for tweet in tweets_iter:
      if not 'retweeted_status' in dir(tweet):
        tweet_text = self.__clean_tweet(tweet.full_text)
        tweets_data = {
            'len' : len(tweet_text),
            'ID' : tweet.id,
            'User' : tweet.user.screen_name,
            'UserName' : tweet.user.name,
            'UserLocation' : tweet.user.location,
            'TweetText' : tweet_text,
            'Language' : tweet.user.lang,
            'Date' : tweet.created_at,
            'Source': tweet.source,
            'Likes' : tweet.favorite_count,
            'Retweets' : tweet.retweet_count,
            'Coordinates' : tweet.coordinates,
            'Place' : tweet.place 
        }

        tweets_data_list.append(tweets_data)

    return tweets_data_list

  def sentiment_polarity(self, tweets_text_list):
      tweets_sentiments_list = []

      for tweet in tweets_text_list:
        polarity = TextBlob(tweet).sentiment.polarity
        if polarity > 0:
          tweets_sentiments_list.append('Positive')
        elif polarity < 0:
          tweets_sentiments_list.append('Negative')
        else:
          tweets_sentiments_list.append('Neutral')

      return tweets_sentiments_list
Enter fullscreen mode Exit fullscreen mode

Agora basta realizar as chamadas das classes para desfrutar das funções.

analyzer = TweetAnalyzer(consumer_key = CONSUMER_KEY, consumer_secret = CONSUMER_SECRET, 
access_token = ACCESS_TOKEN, access_token_secret=ACCESS_TOKEN_SECRET)

# Para realizar as buscas usando a keyword e quantidade predefinida e converter em uma lista.
tweets_iter = analyzer.search_by_keyword(keyword, count)
tweets_list = analyzer.prepare_tweets_list(tweets_iter)

# Uso de um Dataframe para melhor manipulação
tweets_df = pd.DataFrame(tweets_list)

# Chamar a função para análise de sentimentos
tweets_df['Sentiment'] = analyzer.sentiment_polarity(tweets_df['TweetText'])

Enter fullscreen mode Exit fullscreen mode

Uma vez que temos uma estrutura Dataframe com essas informações, nós podemos explorar outras tópicos também como abordei neste post aqui. Além disso, você pode ver o notebook com código desenvolvido nesse repositório.

Considerações finais e próximos passos

No texto de hoje eu demonstrei como podemos realizar a coleta de tweets, tratamento dos tweets e análise deles utilizando Python, Tweepy e TextBlob. Essas bibliotecas são muito interessantes para quem está iniciando com a coleta e processamento de tweets, são muito práticas e permitem uma introdução interessante para atividades envolvidas no processo de coleta e análise de dados.

No próximo texto irei compartilhar quais foram os resultados obtidos através dessa coleta. Até lá, pessoal!

Discussion (0)