Código para almacenar únicamente aquellos tweets que tengan geolocalización utilizando Tweepy y Python.

En este proyecto vamos a generar una pequeña investigación desde el punto de vista del Big Data y de las redes sociales. Exactamente, hemos utilizado Twitter como fuente textual, una red social en la que se nos permite recolectar tweets geolocalizados por zonas.

Para quien no lo conozca, Twitter permite el envío de mensajes en texto plano de corta longitud, con un máximo de 280 caracteres. Estos mensajes, llamados tweets, se muestran en la página principal del usuario y pueden ser capturados a través de una API proporcionada por la propia red social.

Este ejemplo va a consistir en la implementación de un script en Python que almacene continuamente todos los tweets que se escriben sobre el Coronavirus, en ingles y estén geolocalizados. De esta forma, podríamos investigar en que país o ciudad se escribe más sobre una determinada temática.

Para más información de cómo funciona Tweepy, les recomiendo que visiten la url: https://1938.com.es/app-coronavirus-twitter o si prefieres un vídeo tutorial visita la siguiente url: https://www.youtube.com/watch?v=vCFioQizM4w

El código completo lo puedes descargar en la siguiente url https://github.com/al118345/Tweepy_Example/blob/master/Tweepy_ejemplo_localizacion.py

Análisis del script.

import tweepy
import csv
import os
from autenticate import get_auth

'''
Código base para la obtención de los tweets.
'''
class MyStreamListener(tweepy.StreamListener):
    def on_status(self, status):
        # if status is not False and status.text is not None and
        if status.coordinates is not None:
            try:
                texto = status.extended_tweet["full_text"]
            except AttributeError:
                texto = status.text
            texto=texto.replace('
', ' ')
            print(texto)
            linea = [status.created_at,
                     status.id, texto, status.source, status.truncated,
                     status.in_reply_to_status_id, status.in_reply_to_user_id,
                     status.in_reply_to_screen_name, status.geo,
                     status.coordinates,
                     status.place, status.contributors, status.lang,
                     status.retweeted]
            linea = linea
            csvFile = open('resultado_geolocalizado.csv', 'a', encoding= 'utf-8', newline='')
            csvWriter = csv.writer(csvFile)
            csvWriter.writerow(linea)
            csvFile.close()
            print("Almacenamos Tweet")

        # print("fin")

    def on_error(self, status_code):
        print(status_code)
        return False


if __name__ == '__main__':
    print("===== Captador de tweets =====")
    # Get an API item using tweepy
    auth = get_auth()  # Retrieve an auth object using the function 'get_auth' above
    api = tweepy.API(auth)  # Build an API object.

    # Connect to the stream
    myStreamListener = MyStreamListener()
    while True:
        try:
            if os.path.isfile(
                    'resultado_geolocalizado.csv'):
                print('Preparado el fichero')
            else:
                print('El no archivo existe.');
                csvFile = open('resultado_geolocalizado.csv', 'w', encoding= 'utf-8', newline='')
                csvWriter = csv.writer(csvFile)
                cabecera = ['Fecha_creación', 'Id', 'Texto', 'Fuente',
                            'Truncado'
                    , 'Respuesta_al_tweet', 'Respuesta_al_usuario_id'
                    , 'Respuesta_al_usuario_nombre'
                    , 'Localización'
                    , 'Coordenadas'
                    , 'Ciudad'
                    , 'Contribuciones'
                    , 'Idioma'
                    , 'Retweeted'

                            ]
                csvWriter.writerow(cabecera)
                csvFile.close()
                print("Creación de la cabecera")
            myStream = tweepy.Stream(auth=api.auth, listener=myStreamListener)
            myStream.filter(track=['Coronavirus'])
        except:
            continue
    print("Terminado")
  

Como podéis observar, el código no tiene ningún misterio. El objetivo de esta implementación es buscar todos los tweets en el idioma inglés (en), que contengan la palabra Coronavirus y almacenar únicamente aquellos que tengan información sobre la geolocalización en el momento que fueron escrito.

Dicho esto y antes de empezar con las partes más importantes del código, voy a mostrar en la siguientes tabla que información geolocalizada se puede obtener de un tweet.

Cómo se puede comprobar, en un tweet se almacena la localización en diferentes formatos y tipo de coordenadas. Por una parte, en la propiedad place podemos obtener el nombre completo de la ciudad, estado o país dónde se localiza el usuario en el momento de escribir el tweet y, por otra parte, en coordinates o geo tenemos los puntos coordinados.

Pero no todos los tweets escritos tienen activada esta funcionalidad, es decir, de cada 100 tweets escritos sobres el coronavirus únicamente 1 posee esta información sobre su localización.

Para únicamente obtener el tweet que disponga de información geografica, debemos fijarnos en la línea 15 del código mostrado en la parte superior. Exactamente la siguiente linea.

if status.coordinates is not None: 

Esta linea solamente permite almacenar tweets que tengan el campo coordinate rellenado. Podéis contextualizar la funcionalidad expuesta dentro de la siguiente función, cuya única finalidad es, dado un tweet (status ) comprobar si esta geoposicionado. En caso afirmativo lo almacena en el fichero, en caso negativo, continuamos.

def on_status(self, status):
        # if status is not False and status.text is not None and
        if status.coordinates is not None:
            try:
                texto = status.extended_tweet["full_text"]
            except AttributeError:
                texto = status.text
            print(texto)
            texto=texto.replace('
', ' ')
            linea = [status.created_at,
                     status.id, texto, status.source, status.truncated,
                     status.in_reply_to_status_id, status.in_reply_to_user_id,
                     status.in_reply_to_screen_name, status.geo,
                     status.coordinates,
                     status.place, status.contributors, status.lang,
                     status.retweeted]
            csvFile = open('resultado_geolocalizado.csv', 'a', encoding= 'utf-8', newline='')
            csvWriter = csv.writer(csvFile)
            csvWriter.writerow(linea)
            csvFile.close()
            print("Almacenamos Tweet")
  

El encargado de proporcionar los tweets a la función analizada es el fragmento de código expuesto a continuación. Su única función es activar el listener encargado de ir obteniendo los diferentes tweets publicados por una temática/idioma determinado.

if __name__ == '__main__':
    print("===== Captador de tweets =====")
    # Get an API item using tweepy
    auth = get_auth()  # Retrieve an auth object using the function 'get_auth' above
    api = tweepy.API(auth)  # Build an API object.

    # Connect to the stream
    myStreamListener = MyStreamListener()
    while True:
        try:
            if os.path.isfile(
                    'resultado_geolocalizado.csv'):
                print('Preparado el fichero')
            else:
                print('El no archivo existe.');
                csvFile = open('resultat.csv', 'w', encoding= 'utf-8', newline='')
                csvWriter = csv.writer(csvFile)
                cabecera = ['Fecha_creación', 'Id', 'Texto', 'Fuente',
                            'Truncado'
                    , 'Respuesta_al_tweet', 'Respuesta_al_usuario_id'
                    , 'Respuesta_al_usuario_nombre'
                    , 'Localización'
                    , 'Coordenadas'
                    , 'Ciudad'
                    , 'Contribuciones'
                    , 'Idioma'
                    , 'Retweeted'

                            ]
                csvWriter.writerow(cabecera)
                csvFile.close()
                print("Creación de la cabecera")
            myStream = tweepy.Stream(auth=api.auth, listener=myStreamListener)
            myStream.filter(track=['Coronavirus'])
        except:
            continue
    print("Terminado")
    

Por último, recordar que es necesario obtener y rellenar las credenciales de acceso para que funcione correctamente el script. Para ello recomiendo visitar la siguiente url https://1938.com.es/app-coronavirus-twitter dónde se expone cómo se puede obtener dicha información.

 auth = get_auth()  # Retrieve an auth object using the function 'get_auth' above
 api = tweepy.API(auth)  # Build an API object.
    

Explicación en Youtube