A mesure que Twitter s’est imposé dans le paysage numérique et médiatique, on a découvert à quel point cette plateforme sociale pouvait être chronophage et parfois contraignante. Dans ce cas, le recours à une forme d’automatisation est tentant, et on enverrait bien un robot dans les méandres de Twitter pour nous mâcher le travail. C’est ainsi que des applications telles que threadreaderapp ou blockpartyapp sont aujourd’hui à disposition. Ces solutions, comme celles que vous imaginerez pour tirer le meilleur de Twitter ont en commun de solliciter l’API de la plateforme. Nous allons donc, dans cet article, détailler les étapes nécessaire à l’élaboration d’un bot Twitter en Python. Il ne vous restera plus qu’à imaginer l’usage que vous pourrez en faire.
Pour bien comprendre le principe de fonctionnement de ce bot Twitter, nous allons procéder par étapes. Nous allons dans un premier temps solliciter l’API Twitter de manière ponctuelle, pour recueillir les informations d’un tweet en particulier. Nous allons ensuite collecter ce type d’informations mais sur un ensemble de tweets sur la base d’un filtre. Enfin, nous allons automatiser des actions à mener sur les tweets ainsi sélectionner, à savoir les envoyer dans un mail ou y répondre.
Quelques pré-requis
Pour parvenir à faire fonctionner un programme de manière automatique en sollicitant l’API de Twitter, vous aurez besoin de quelques accès et compétences :
- Un compte Twitter (forcément) qui vous permettra d’accéder à votre compte développeur . C’est là que vous indiquerez votre volonté d’utiliser l’API, on vous assignera un Bearer Token. Gardez-le précieusement, car vous en aurez besoin pour faire vos requêtes plus tard.
- Être familier avec les notions propres aux API, notamment celle d’endpoint.
- Une petite maîtrise de Python et notamment de l’usage des bibliothèques qui vont nous être indispensables.
Un premier contact avec l’API de Twitter
L’objectif de ce premier contact avec l’API va être de récupérer des informations sur un tweet en particulier sans ouvrir Twitter. Pour cela nous allons indiquer l’identifiant de ce tweet à notre programme (si vous vous demandez comment obtenir un identifiant de tweet, cliquez sur un tweet depuis un navigateur web, observez l’url, après la partie status, la valeur numérique qui suit est son identifiant), nous identifier et demander à l’API différentes informations, chacune correspondant à un “endpoint” particulier. La liste des références à ces “endpoints” est disponible dans la documentation de l’API.
Par défaut, le programme nous donne l’auteur et le texte du tweet. Nous allons utiliser la bibliothèque requests pour effectuer les calls HTTP et pprint pour afficher (“de manière élégante” nous dit la documentation) le fichier json obtenu :
import os
from pprint import pprint
import requests
params = {'ids': ['1588915242490560512']}
# Le bearer token est stocké dans une variable d'environnement BEARER_TOKEN
headers = {'Authorization': f"Bearer {os.getenv('BEARER_TOKEN')}"}
r = requests.get('https://api.twitter.com/2/tweets', params=params, headers=headers)
pprint(r.json())
# exemple de réponse qu'on aura { 'data': [ { 'edit_history_tweet_ids': ['1588915242490560512'], 'id': '1588915242490560512', 'text': 'in what is becoming a tradition in odd point release...' } ] }
Vous remarquerez que les données se trouvent dans la partie data et que chaque objet représente un tweet. Ici nous n’en avons sélectionné qu’un seul. Pour afficher plus d’informations sur ce tweet, en plus de son auteur et le texte du tweet, il faudra utiliser un champ supplémentaire de requête tweet.fields. Ce sera la même chose avec les autres types d’objets, si vous voulez obtenir plus d’informations sur un utilisateur, il faudra configurer le champ user.fields.
Revenons à notre exemple et disons qu’on va afficher en plus la date de création du tweet et le l’id de l’auteur.
import os from pprint import pprint import requests params = { 'ids': '1588915242490560512', # on ajoute les champs supplémentaires qu'on veut 'tweet.fields': 'created_at,author_id' } headers = {'Authorization': f"Bearer {os.getenv('BEARER_TOKEN')}"} r = requests.get('https://api.twitter.com/2/tweets', params=params, headers=headers) pprint(r.json(), indent=4)
# exemple de réponse { 'data': [ { 'author_id': '1037022474762768384', 'created_at': '2022-11-05T15:24:49.000Z', 'edit_history_tweet_ids': ['1588915242490560512'], 'id': '1588915242490560512', 'text': 'in what is becoming a tradition in odd point release...', } ] }
Surveiller une thématique avec notre bot Twitter en Python
Maintenant, nous entrons dans le vif du sujet, nous allons voir comment observer des événements spécifiques qui se passent sur Twitter et réaliser une action adéquate. Cherchons dans un premier temps les tweets récents qui répondent à nos critères de recherche.
Consulter les tweets récents
Pour rassembler les tweets récents répondant à certains critères, nous allons conjointement faire deux opérations :
1. Appliquer un filtre de recherche
Ce filtre accepte un certain nombre de caractères (512 si vous débutez et avez un accès de type “Essential”) et qui vous permettra de combiner avec des opérateurs booléens des mots-clés et des critères comme le compte émetteur (“from”), le compte destinataire si c’est une réponse (“to”), des informations sur le contenu, images, liens… (“has”) ou des éléments contextuels sur la nature du tweet ou de son auteur (“is:retweet”, “is:verified”… etc). La liste complète des filtres vous laisse entrevoir les possibilités de cet outil.
Pour notre exemple, on se propose de rechercher les promos gandi sur des noms de domaines. On aura la requête suivante:
from:gandi_net #promo has:links -is:retweet.
Vous pouvez intuitivement comprendre ce que fait cette requête, mais récapitulons, on recherche donc :
- des tweets provenant de l’utiliser gandi from:gandi_net
- ET qui comportent le hashtag #promo
- ET qui a un lien (histoire qu’on sache où aller pour faire nos emplettes) has:links
- ET qui n’est pas un retweet -is:retweet. Ce dernier filtre est très important et même recommandé dans la documentation officielle parce que beaucoup de tweets sont souvent des retweets, ça permet d’éviter d’ajouter du bruit inutile dans nos réponses.
2. Obtenir des données historiques
Ici, la route qui nous intéresse permet de rechercher les tweets sur les 7 derniers jours. On pourrait continuer à travailler avec la bibliothèque requests mais il serait fastidieux d’analyser les résultats, gérer la pagination, etc. c’est pourquoi nous allons utiliser une bibliothèque spécialisée pour l’api Twitter j’ai nommé tweepy. Nous utiliserons sa classe tweepy.Client pour manipuler les endpoints l’api v2 de Twitter. Pour connaître une correspondance entre les méthodes de cette classe et les endpoints api, référez-vous à cette page. Dans notre cas, la méthode qui nous intéresse est search_recent_tweets.
import tweepy client = tweepy.Client(os.getenv('BEARER_TOKEN')) response = client.search_recent_tweets( 'from:gandi_net #promo has:links -is:retweet', max_results=100, tweet_fields=['created_at'] ) if response.data is not None: for tweet in response.data: print(tweet.id, tweet.text) # les infos de pagination sont listées ici print(response.meta)
Quelques précisions :
- L’utilisation est simple, ici on a spécifié le filtre qui est le seul argument obligatoire. Ensuite on a spécifié le nombre d’éléments qu’on voulait afficher et les champs additionnels. En général, les arguments de méthodes sont les mêmes que la route api peut prendre. Pour connaître tous les arguments possibles de passer à cette méthode, referez-vous à sa documentation
- l’objet réponse retournée est un namedtuple qui contient quatre clés data, includes, meta et errors. Ce sont les mêmes informations que lorsqu’on effectue la requête soi-même avec requests par exemple
- Dans response.data, on a la liste des objets demandés. En règle générale, si vous connaissez la liste des champs de l’objet en question, alors vous pouvez les utiliser comme propriété comme fait dans l’exemple ci-dessus. Néanmoins, ils sont tous documentés dans la documentation de tweepy.
Signalons ici sans le développer qu’une autre façon d’utiliser la route des tweets récents est le polling, c’est un peu un mode temps réel. L’idée est de chercher tous les tweets qui correspondent au filtre, mais non plus dans le passé, mais à partir d’un tweet précis. Cela demande donc de connaître un tweet (récent) qui répond déjà à notre critère et d’itérer à partir de là.
Obtenir un flux filtré
Une méthode possible pour rechercher des tweets en temps réel est le flux filtré. Sa documentation se retrouve ici. Elle comprend 3 endpoints à exploiter:
- Un pour ajouter ou supprimer des filtres.
- Un pour lister les filtres.
- Un pour pour rechercher les tweets.
Contrairement à l’endpoint précédent, on peut rajouter plusieurs règles de filtrage, le nombre dépend du type d’accès qu’on a.
- Pour un accès Essential, on a droit à 5 règles de 512 caractères chacune. (Probablement votre cas si vous débutez avec l’api Twitter)
- Pour un accès Elevated, on a droit à 25 règles de 512 caractères chacune.
- Pour un accès Academic on a droit à 1000 règles de 1024 caractères chacune.
L’idée ici est simple, on ajoute une ou plusieurs règles de filtrage et on recherche des tweets récents qui sont liés à ses filtres. Si une seule règle matche pour un tweet, il sera retourné. Pour manipuler ses endpoints avec tweepy, on utilisera la classe tweepy.StreamingClient.
Pour ajouter, lire et supprimer des filtres on pourra écrire :
import os import tweepy client = tweepy.StreamingClient(os.getenv('BEARER_TOKEN')) rules = [ # on ajoute nos règles ici tweepy.StreamRule('from:gandi_net #promo has:links -is:retweet', tag='gandi promo') tweepy.StreamRule('from:gandi_net #certificat -is:retweet', tag='gandi certificat') ] client.add_rules(rules) # on affiche nos règles response = client.get_rules() for rule in response.data: print(rule) # On supprime une ou plusieurs règles en passant leur id client.delete_rules(['158939726852798054'])
Note: Vous pouvez passer l’argument dry_run=True aux méthodes add_rules et delete_rules histoire de tester la requête pour vérifier qu’elle est correcte sans réellement l’exécuter côté serveur.
Note: Lorsqu’on définit plusieurs règles, il est recommandé d’associer un tag pour se rappeler ce que fait le filtre. En effet un filtre peut être assez complexe à lire 🙂
Une fois qu’on a créé nos règles, il ne nous reste plus qu’à appeler l’endpoint permettant de lister les tweets. Normalement la méthode à utiliser est listen sauf que si vous l’appeler en l’état, vous ne verrez rien parce que par défaut StreamingClient ne fait rien avec les tweets qu’il récupère. Il faut donc hériter de la classe et surcharger certaines de ces méthodes.
import os import tweepy class IDPrinter(tweepy.StreamingClient): # vous pouvez obtenir un objet réponse complète def on_response(self, response): # il a la structure StreamResponse(tweet, includes, errors, matching_rules) # pour chaque tweet, on a donc l'ensemble des règles qui ont matché print(response) # ou vous pouvez juste obtenir le tweet def on_tweet(self, tweet): print(tweet.id, tweet.text) def on_errors(self, errors): print(errors) def on_connection_error(self): # ce qu'on doit faire en cas d'erreur de connexion réseau self.disconnect() def on_request_error(self, status_code): # ce qu'on doit faire si le status code de la réponse HTTP est >= 400 pass printer = IDPrinter("Bearer Token here") printer.filter()
Notes:
- En surchargeant on_response, on a en paramètre un objet StreamingResponse qui contient le tweet et l’ensemble des règles qui ont matché ce tweet.
- On peut passer l’argument threaded=True à filter pour ne pas bloquer le programme et récupérer le thread créé pour ensuite le fermer plus tard.
- Il existe une version asynchrone de cette classe de streaming AsyncStreamingClient qui permet de manipuler des coroutines au lieu des threads. Je n’en parlerais pas ici vu que c’est un sujet avancé. Pour les plus courageux, vous pouvez lire cet article.
- Il existe une autre méthode de la classe StreamingClient, à savoir sample qui est lié à cette route api. Elle ne prend pas en compte les filtres créés et retourne 1% des nouveaux tweets enregistrés sur la plateforme Twitter. Cela vous permettrait par exemple avec des algorithmes d’analyse naturelle de texte de détecter les tendances Twitter comme ceux qu’on voit afficher sur la droite sur l’interface web. Prenez garde néanmoins en utilisant cette route, elle peut vite finir votre limite de tweets à lire par mois. Il faut avoir un objectif clair avant de l’utiliser et ne le faire que dans un laps de temps court.
Automatiser des tâches grâce au bot Twitter
Maintenant que l’on sait converser avec l’API et lui demander des choses très précises, il est temps d’envoyer notre bot twitter au travail et lui faire faire des actions concrètes
Envoi hebdomadaire des résultats de recherche par email
Nous verrons dans un premier temps comment envoyer un email avec pièce jointe des différents tweets résultant de la recherche de promotions, chaque lundi matin à 9h, sur les résultats de la semaine écoulée.
Nous utiliserons les bibliothèques python suivantes:
- apscheduler: pour programmer des tâches répétitives.
- emails: pour envoyer des emails.
Voici le code résultant que nous commenterons juste après:
import os import emails import tempfile import json from pathlib import Path import tweepy from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.cron import CronTrigger def send_mail_with_tweets(): # vous devez définir les variables d'environnement suivantes: # SMTP_HOST: le serveur smtp où d'où envoyer le message # SMTP_PORT: le port utilisé par le serveur SMTP # SMTP_TLS: "true" si on veut utiliser TLS "false" sinon # SMTP_USER: optionel, l'utilisateur qui va envoyer le mail # SMTP_PASSWORD: optionel, le mot de passe de l'utilisateur smtp = { 'host': os.getenv('SMTP_HOST'), 'port': os.getenv('SMTP_PORT'), } if os.getenv('SMTP_TLS', '').lower() == 'true': smtp.update({ 'ssl': True, 'user': os.getenv('SMTP_USER'), 'password': os.getenv('SMTP_PASSWORD') }) client = tweepy.Client(os.getenv('BEARER_TOKEN')) with tempfile.TemporaryDirectory() as tmp_dir: # vous pourriez avoir envie de remplacer l'extension ".jl" # par "txt" pour ne pas être bloqué par certains fournisseurs de mail path = Path(tmp_dir) / 'tweets.jl' with path.open('w') as f: for tweet in tweepy.Paginator( client.search_recent_tweets, 'from:gandi_net #promo has:links -is:retweet', max_results=100 ).flatten(): data = { 'id': tweet.id, 'text': tweet.text, 'twitter_url': f'https://twitter/gandi_net/status/{tweet.id}' } f.write(f'{json.dumps(data)}\n') message = emails.Message( subject='Promotions gandi', text='En pièce jointe vous trouverez un fichier avec tous les tweets liés à des promotions.', mail_from=('Twitter Bot', 'twitter@bot.com') ) message.attach(filename=path.name, content_disposition='inline', data=open(path, 'rb')) response = message.send(to='foo@bar.com', smtp=smtp) # loguer le fait qu'un email n'ait pas pu être envoyé if response.status_code != 250: print("l'email n' a pas pu être envoyé") # Pour un exemple plus industriel il faudrait configurer un jobstore pour conserver # les informations de job en db, comme cela s'il y a un crash du serveur et qu'il # doit redémarrer, vous le scheduler reprendra où il s'est arrêté scheduler = BlockingScheduler(timezone='utc') # on va utiliser la notation crontab pour déclencher notre action # on le fera chaque lundi à 9h du matin scheduler.add_job(send_mail_with_tweets, CronTrigger.from_crontab('0 9 * * 1')) # le scheduler démarre ici et son appel est bloquant scheduler.start()
Notes:
- Il y a quelques variables d’environnement à créer pour configurer l’envoi d’emails. Le login et password ne sont nécessaires que pour l’usage de TLS, ce qui sera le cas la plupart du temps.
- Pour tester l’envoi d’emails en local, vous pouvez utiliser le service mailhog.
- Lignes 52 à 64, pour stocker les tweets dans un fichier, j’ai utilisé le format json lines. Il est pratique pour sauvegarder beaucoup de données sans bouffer toute la mémoire.
- Lignes 66 à 76, on définit les informations de l’email (vous pouvez les modifier à votre guise), la pièce jointe et on l’envoie. J’ai affiché un message dans la console comme quoi l’envoi d’email s’est mal passé, mais vous pouvez configurer le logging à la place. D’ailleurs pour essayer de déboguer l’erreur, vous devriez configurer le logging pour afficher les messages de la bibliothèque emails au niveau debug.
- Ligne 83, on définit le scheduler, on utilise la version Blocking qui est adapté dans notre cas, mais suivant le style de programme que vous écrivez, vous pourriez être en mesure d’utiliser la version Threading ou Asyncio. Je vous laisse regarder la documentation pour plus de détails.
- Ligne 87 on ajoute notre job en utilisant la crontab comme déclencheur. Si vous voulez vérifier la syntaxe de la crontab, vous pouvez utiliser ce site web.
- Ligne 90, notre scheduler va s’exécuter en mode démon.
Créer un tweet avec notre bot Twitter
Pour pouvoir faire la deuxième application qu’on s’est promise au début de cet article, il va falloir apprendre à créer un tweet par API. Hélas, on ne pourra plus utiliser le bearer token comme avec les routes précédentes. Il nous faudra un access token avec des droits bien spécifiques. Si l’on regarde la documentation de la route en question, on aura besoin des permissions tweet:read, tweet:write et users:read. Le workflow pour obtenir un token est défini ici. Je vais néanmoins vous montrer comment procéder avec tweepy. Il va falloir une api ou plus exactement une url de redirection pour reprendre les termes techniques. Cette url servira pour envoyer un code nécessaire à l’obtention du token d’accès. Sur cette url, deux informations seront passées en paramètres de requête:
- state: qui est une valeur aléatoire de sécurité. Plus d’informations sur le jargon technique oauth2 sur cette page.
- code: valeur définie par l’api Twitter.
Un exemple de serveur fastapi que vous pourrez mettre en place pour l’url de redirection.
import logging from fastapi import FastAPI, Response app = FastAPI() logger = logging.getLogger(__name__) @app.get('/') def get_code(code: str, state: str): logger.info('code: %s, state: %s', code, state) return Response(status_code=200)
Une fois cela fait, il va falloir retourner sur le portail développeur, au niveau de votre projet, vous avez une section « User authentication settings », cliquez dessus. Vous pouvez zapper la première partie « App permissions » qui concerne oauth1 que nous n’utilisons pas. Sur la deuxième partie « Type of App », choisissez bien « Web app, Automated App or bot ». Dans la partie « App info », renseignez l’url correspondant à votre serveur. Vous devrez renseigner aussi une url personnelle (vous pouvez mettre n’importe quoi tant que c’est lié à vous) à vous et d’autres informations optionnelles si vous le voulez. Une fois cela fait, vous obtiendrez un client_id et un client_secret nécessaires pour l’authentification oauth2. Gardez-les en lieu sûr. Nous utiliserons dans les scripts à venir les variables d’environnement CLIENT_ID et CLIENT_SECRET qui devront contenir ces valeurs. Pour ce qui est du workflow pour obtenir un token d’accès avec tweepy, il se présente de la manière suivante:
1 – Utiliser la classe tweepy.OAuth2UserHandler en renseignant toutes les informations nécessaires.
import os import tweepy oauth2_user_handler = tweepy.OAuth2UserHandler( client_id=os.getenv('CLIENT_ID'), client_secret=os.getenv('CLIENT_SECRET'), redirect_uri='votre url de redirection ici', # les permissions ici scope=['tweet.read', 'tweet.write', 'users.read', 'offline.access'], ) # il va génerer l'url d'autorisation qu'on devra lancer dans notre navigateur web. print(oauth2_user_handler.get_authorization_url())
Vous remarquerez pour l’argument scope que j’ai rajouté la permission offline.access. Il permettra de rafraîchir le token d’accès sans passer par une intervention manuelle, je l’explique plus tard. Une fois que vous avez l’url d’autorisation, copiez-la dans la barre d’url de votre navigateur préféré et suivez-la (tapez sur Enter). Vous allez être amené à autoriser votre bot d’avoir accès à votre compte, une fois que c’est fait vous serez rediriger sur votre l’url que vous avez défini comme url de redirection.
2 – Copiez l’url de redirection dans le navigateur qui doit contenir le code qui vous a été attribué, et utilisez une méthode de OAuth2UserHandler qui va récupérer le token d’accès.
access_token = oauth2_user_handler.fetch_token( 'url de redirection ici' )
3 – Une fois cela fait vous pouvez réutiliser le client tweepy comme d’habitude, sauf qu’au lieu du bearer token, ce sera le token d’accès qui sera utilisé.
... client = tweepy.Client('access token')
Ensuite vous pourrez créer un tweet comme ceci :
client.create_tweet(text='hello from bot', user_auth=False)
Le user_auth=False est important sinon tweepy va essayer une authentication oauth1 et la requête va échouer. C’est un peu étrange comme api, mais c’est l’héritage de l’ancienne api. Pour info, un token d’accès a une validité de deux heures. Et là certains petits malins doivent se demander s’il va falloir répéter manuellement l’opération avec le navigateur… Ça ne fait pas très automatisé tout ça. Vous vous souvenez de la permission offline.access qu’on a utilisé au début pour avoir l’url d’autorisation? Elle nous servira à rafraîchir notre token avant son expiration. En fait, au moment où on a récupéré le token d’accès, notre permission nous a aussi permis de récupérer un token de rafraîchissement (refresh token). C’est ce dernier token qui est utilisé pour le rafraîchissement. Elle est conservée en interne par tweepy. Pour rafraîchir le token d’accès avec tweepy, voici comment on procédera :
import tweepy token_info = oauth2_user_handler.refresh_token( 'https://api.twitter.com/2/oauth2/token' ) client = tweepy.Client(token['access_token'])
On doit passer l’url pour refresh un token. Si vous vous demandez où j’ai trouvé cette url, elle est disponible sur cette page sous la section « Step 5… ». Si vous avez un refresh_token que vous avez obtenu d’une autre façon que par tweepy, vous pouvez le passer avec l’argument refresh_token. Dans token_info on a un ensemble d’information dont le token d’accès et le nouveau token de rafraîchissement (là encore sauvegardé par tweepy pour une utilisation ultérieure). On peut donc instancier un nouveau client avec le nouveau token d’accès sans avoir à faire une intervention manuelle. Et pour automatiser le tout, on peut utiliser apscheduler que nous connaissons bien désormais. Un exemple de code que vous pourrez écrire :
import os import tweepy from apscheduler.schedulers.blocking import BlockingScheduler scheduler = BlockingScheduler() oauth2_user_handler = tweepy.OAuth2UserHandler( client_id=os.getenv('CLIENT_ID'), client_secret=os.getenv('CLIENT_SECRET'), redirect_uri='votre url de redirection ici', scope=['tweet.read', 'tweet.write', 'users.read', 'offline.access'], ) def refresh_token(): token_data = oauth2_user_handler.refresh_token( 'https://api.twitter.com/2/oauth2/token' ) # sauvegarder le token où vous voulez os.environ['ACCESS_TOKEN'] = token_data['access_token'] # Penser à ajouter ce job la première fois que vous obtenez un token d'accès scheduler.add_job(refresh_token, 'interval', hours=1, minutes=55) scheduler.start()
Ici on utilise le déclencheur interval pour rafraîchir le token après une heure et cinquante-cinq minutes, vu qu’un token dure deux heures.
(bonus) Envoi d’un mème en réponse à chaque nouvelle promotion de Gandi
On arrive donc au moment le plus amusant. On va reprendre notre dernier exemple qui affichait dans la console et répondre avec le meme « take my money ».
import os import tweepy client = tweepy.Client(os.getenv('ACCESS_TOKEN')) class MemePromoAnswer(tweepy.StreamingClient): def on_tweet(self, tweet): print(tweet.id, tweet.text) client.create_tweet(media_ids=['1590404643045216256'], in_reply_to_tweet_id=tweet.id) def on_errors(self, errors): print(errors) def on_connection_error(self): self.disconnect() meme = MemePromoAnswer(os.getenv('BEARER_TOKEN')) meme.filter()
Précisions :
- Vous devez avoir gardé le filtre des promotions Gandi pour que ça marche.
- Dans l’utilisation de create_tweets, je passe un tableau à une seule valeur qui correspond à l’id du GIF « take my money ». Si vous vous demandez comment j’ai fait pour l’obtenir, eh ben j’ai créé un tweet avec ce meme et j’ai récupéré les infos du tweet. Ensuite vu qu’on répond à un tweet, on utilise l’argument in_reply_to_tweet_id et on spécifie l’id du tweet ciblé.
Mission accomplie. C’était long, mais admettez que ça valait le coup.
