[PYTHON] Créez un robot de réponse automatique en temps réel à l'aide de l'API Twitter Streaming

introduction

J'écrirai en détail sur la partie API Twitter de ce que j'ai fait dans cet article.

Lors du développement du robot de réponse automatique Twitter, je n'ai pas trouvé d'article de référence avec un contenu tel que "Obtenir des tweets (réponses) en temps réel" pour la version 2020, je voudrais donc résumer ce que j'ai recherché cette fois. Je pense.

Voici les spécifications du Bot que j'ai réalisé cette fois.

Twitter API

Obtenir la clé API Twitter

En août 2020, la méthode d'obtention de la clé API est légèrement différente, mais j'ai fait référence à cet article pour savoir comment demander son utilisation. ・ Explication détaillée de la phrase d'exemple de l'application d'utilisation de l'API Twitter version 2020 à l'acquisition de la clé API

Une fois votre candidature approuvée, vous avez effectué les opérations suivantes sur le portail des développeurs:

--Générer un jeton d'accès et un secret (requis pour utiliser l'API)

Twitter Streaming API

Apparemment, l'API User Streams a été utilisée pour obtenir des tweets sur la chronologie en temps réel. Cependant, il semble que cette API ait été supprimée en août 2018, et comme substitut maintenant, Statuts / filtre des Tweets en temps réel du filtre / tweets / filter-realtime / api-reference / post-statuses-filter) peut être utilisé.

Pour plus de détails, j'ai renvoyé à cet article. ・ À propos de l'API Streaming de Twitter - journal d'activité de kivantium

Dans l'article ci-dessus, il semble que tweepy soit utilisé, mais j'ai personnellement semblé utiliser la bibliothèque twitter, donc j'ai utilisé la bibliothèque twitter. Installez avec la commande suivante. (Exécuter dans un environnement virtuel)

(venv)$ pip install twitter

Pour la bibliothèque twitter, les sites suivants ont été utiles. ・ Twitter 1.18.0Notes sur l'utilisation des outils Twitter PythonMémo de remplacement du flux utilisateur Twitter pour Python

Comment utiliser la bibliothèque Twitter

Préparation des objets d'interface pour l'API Twitter

import twitter

#Création d'un objet d'authentification OAuth
oauth = twitter.OAuth('Your Access token',
                      'Your Access token secret',
                      'Your API key',
                      'Your API secret key')

#Objet d'utilisation de l'API REST Twitter
twitter_api = twitter.Twitter(auth=oauth)
#Objets pour l'utilisation de l'API Twitter Streaming
twitter_stream = twitter.TwitterStream(auth=oauth)

Envoyez une réponse

API d'envoi de tweets états POST / mise à jour Est une API REST, utilisez donc l'objet twitter_api préparé ci-dessus.

text = 'tester'

#Pour les tweets réguliers
twitter_api.statuses.update(status=text)

#Lorsque vous répondez à un tweet spécifique
reply_text = '@username' + '\n'
reply_text += text
twitter_api.statuses.update(status=reply_text, in_reply_to_status_id='ID du tweet auquel répondre')

Lors de l'envoi d'une réponse, il semble que vous deviez non seulement spécifier in_reply_to_status_id, mais également inclure @ username (@ + nom d'utilisateur du tweet auquel répondre) dans le texte.

Obtenez la chronologie en temps réel

[Comme ci-dessus](# twitter-streaming-api), l'API Streaming est utilisée, utilisez donc l'objet twitter_stream préparé ci-dessus.

#Chaîne de caractères à surveiller
tracking_text = '@my_test du nom d'utilisateur' #ET recherchez avec un espace demi-largeur.','OU recherche par délimiteur

for tweet in twitter_stream.statuses.filter(language='ja', track=tracking_text):
    print(tweet)

Si vous faites cela, vous obtiendrez un objet tweet tweet contenant à la fois les chaînes" @ my_username` "et" test ". (Pour tous les tweets du compte public) Veuillez vérifier le contenu. Utilisez les informations contenues dans cet objet «tweet» pour répondre.

Il existe également un argument de filtre appelé follow, et vous pouvez limiter les utilisateurs à surveiller en passant des" ID utilisateur séparés par ",". La raison pour laquelle je n'ai pas spécifié follow cette fois est que ce sera une recherche OR au lieu d'une recherche AND avec la condition track. En d'autres termes, le tweet de l'utilisateur spécifié par «follow» sera acquis quel que soit le contenu. Dans cette spécification, la spécification de «track» peut réduire les tweets à acquérir plutôt que de spécifier «follow», de sorte que seul «track» est spécifié.

Code source

Structure du répertoire

Cette fois, j'ai créé un écran de gestion pour enregistrer et éditer le texte de réponse avec flask, donc la structure est similaire à flask. Il n'y a aucun problème même si vous les rassemblez dans un seul fichier.

app/
├── .env             #Préparez la clé API Twitter, etc.
├── bot.py           #Dossier d'exécution
└── bot/             #Traitement de la base de données pour le texte de réponse / la classe de service omis
    ├── __init__.py
    ├── config.py
    └── tweet.py

Dossier d'exécution

Il lit simplement le fichier et exécute la fonction.

bot.py


import bot

if __name__ == '__main__':
    bot.start()

fichier de configuration

Nous préparons un objet pour lire les variables d'environnement et utiliser l'API Twitter.

config.py


import os
from os.path import join, dirname
from dotenv import load_dotenv

import twitter

dotenv_path = join(dirname(__file__), '../.env')
load_dotenv(dotenv_path)


oauth = twitter.OAuth(os.environ.get('ACCESS_TOKEN_KEY'),
                      os.environ.get('ACCESS_TOKEN_SECRET'),
                      os.environ.get('CONSUMER_KEY'),
                      os.environ.get('CONSUMER_SECRET'))

twitter_api = twitter.Twitter(auth=oauth)
twitter_stream = twitter.TwitterStream(auth=oauth)
SCREEN_NAME = os.environ.get('TWITTER_SCREEN_NAME') # @nom d'utilisateur votre propre nom d'utilisateur

Traitement principal

Il surveille la chronologie en temps réel et répond automatiquement aux réponses contenant des mots spécifiques. J'aurais pu les mettre ensemble dans ce fichier, mais j'ai préparé une autre classe pour gérer l'objet Tweet récupéré et séparé le traitement.

__init__.py


import logging
from bot.config import twitter_stream, SCREEN_NAME
from bot.tweet import Tweet


def start():

    #paramètres de journalisation
    formatter = '%(levelname)s : %(asctime)s : %(message)s'
    logging.basicConfig(level=logging.INFO, format=formatter)

    REPLY_TEXT = 'Bonjour'

    replied_user_list = [] #Gérer les utilisateurs qui répondent dans une liste

    tracking_text = '@'+ SCREEN_NAME + 'Bonjour'
    for tweet in twitter_stream.statuses.filter(language='ja', track=tracking_text):
        tweet_obj = Tweet(tweet)

        #Si la clé requise n'est pas incluse
        required_keys_list = [
            'id_str',
            'text',
            'user'
        ]
        if not tweet_obj.has_required_keys(required_keys_list):
            logging.warning('FALSE->required key is empty')
            print(tweet_obj.tweet)
            continue

        #Pour les retweets (suivi_Il réagit s'il correspond au texte)
        if tweet_obj.is_retweet():
            logging.warning('%s\n [user]: %s\n [tweet]: %s', 'FALSE->is retweet', tweet_obj.get_user_screenname(), tweet_obj.get_tweet_text())
            continue

        #Pour les utilisateurs qui ont répondu dans le passé
        user_id = tweet_obj.get_user_id()
        if user_id in replied_user_list:
            logging.warning('%s\n [user]: %s\n [tweet]: %s', 'FALSE->has already replied', tweet_obj.get_user_screenname(), tweet_obj.get_tweet_text())
            continue

        #Pour vos propres tweets
        if tweet_obj.is_reply_from_me():
            tweet_obj.reply(REPLY_TEXT)
            replied_user_list.append(user_id)
            logging.info('%s\n [user]: %s\n [tweet]: %s', 'SUCCESS->self tweet', tweet_obj.get_user_screenname(), tweet_obj.get_tweet_text())
            continue

        #Si ce n'est pas le tweet d'un abonné
        if not tweet_obj.is_reply_from_follower():
            logging.warning('%s\n [user]: %s\n [tweet]: %s', 'FALSE->not follwer', tweet_obj.get_user_screenname(), tweet_obj.get_tweet_text())
            continue

        #Système normal
        tweet_obj.reply(REPLY_TEXT)
        replied_user_list.append(user_id)
        logging.info('%s\n [user]: %s\n [tweet]: %s', 'SUCCESS', tweet_obj.get_user_screenname(), tweet_obj.get_tweet_text())

Cette fois, l'explication du traitement de la classe de service pour le texte de réponse est omise, elle est donc remplacée par le texte de réponse fixe «REPLY_TEXT». Pour une réponse par jour, cliquez ici (https://qiita.com/YiwaiY/items/ce790838725aba54c035#1%E3%82%A2%E3%82%AB%E3%82%A6%E3%83% B3% E3% 83% 88% E3% 81% AB% E3% 81% A4% E3% 81% 8D1% E6% 97% A51% E3% 83% AA% E3% 83% 97% E3% 83% A9% E3% 82% A4% E3% 82% 92% E3% 81% A9% E3% 81% 86% E5% AE% 9F% E8% A3% 85% E3% 81% 99% E3% 82% 8B% E3% Veuillez vous référer à 81% 8B).

Classe qui gère les objets tweet

Le traitement qui utilise l'API REST Twitter et la méthode d'évaluation des conditions sont résumés ici.

tweet.py


from bot.config import twitter_api, SCREEN_NAME


class Tweet():

    def __init__(self, tweet):
        self.tweet = tweet

    def has_required_keys(self, required_keys_list):
        for required_key in required_keys_list:
            if required_key not in self.tweet:
                return False
        return True

    def reply(self, text):
        status = '@' + self.get_user_screenname() + '\n'
        status += text
        return twitter_api.statuses.update(status=status, in_reply_to_status_id=self.tweet['id_str'])

    def is_retweet(self):
        return 'retweeted_status' in self.tweet

    def is_reply_from_me(self):
        return self.get_user_screenname() == SCREEN_NAME

    def is_reply_from_follower(self):
        followers = twitter_api.followers.ids(screen_name=SCREEN_NAME, count=5000) #Tableau d'ID des abonnés de type int
        return self.get_user_id() in followers['ids']

    def get_user_id(self):
        return self.tweet['user']['id']

    def get_user_name(self):
        return self.tweet['user']['name']

    def get_user_screenname(self):
        return self.tweet['user']['screen_name']

    def get_tweet_text(self):
        return self.tweet['text']

Résumé

Au début, je voulais faire un Bot de réponse automatique sur Twitter, puis je faisais des recherches sur diverses choses, mais je ne pouvais pas tout à fait atteindre ces articles, et je me demandais si je ne pouvais plus les faire. C'était ma première expérience de faire quelque chose par moi-même, alors je suis tombé sur diverses choses, mais je suis content que ce soit terminé pour le moment.

Je suis encore dans ma deuxième année de programmation, et je pense que le code Python du type qui utilise habituellement PHP n'est pas tout à fait là, mais j'espère que cela aidera quelqu'un qui a des problèmes comme moi. ..

Je l'ai présenté au début, mais j'apprécierais également votre faveur dans cet article. ・ [Premier développement personnel] L'histoire du déploiement de l'application Flask et du robot de réponse automatique Twitter sur Heroku

Recommended Posts

Créez un robot de réponse automatique en temps réel à l'aide de l'API Twitter Streaming
Essayez d'utiliser l'API Twitter
Essayez d'utiliser l'API Twitter
Créez un Twitter BOT avec le SDK GoogleAppEngine pour Python
Créer une API CRUD à l'aide de l'API rapide
Créer un graphique à l'aide du module Sympy
Créer une application à l'aide de l'API Spotify
Créez un bot qui stimule les tendances Twitter
Créez un service Twitter BOT avec GAE / P + Tweepy + RIOT API! (Partie 1)
Créez un service Twitter BOT avec GAE / P + Tweepy + RIOT API! (Partie 2)
Créons une API REST en utilisant SpringBoot + MongoDB
Étapes pour créer un bot Twitter avec Python
Créez une interface graphique sur le terminal à l'aide de curses
J'ai créé un LINE BOT qui renvoie une image de riz terroriste en utilisant l'API Flickr
Créez une API REST à l'aide du modèle appris dans Lobe et TensorFlow Serving.
Créer un bot slack
Créer un bot de collecte de données en Python à l'aide de Selenium
[CRUD] [Django] Créer un site CRUD en utilisant le framework Python Django ~ 1 ~
Créez une carte thermique de tweet avec l'API Google Maps
Publiez sur votre compte en utilisant l'API sur Twitter
Créer un pseudo serveur d'API REST à l'aide de pages GitHub
[CRUD] [Django] Créer un site CRUD en utilisant le framework Python Django ~ 2 ~
[CRUD] [Django] Créer un site CRUD en utilisant le framework Python Django ~ 3 ~
[CRUD] [Django] Créer un site CRUD en utilisant le framework Python Django ~ 4 ~
[CRUD] [Django] Créer un site CRUD en utilisant le framework Python Django ~ 5 ~
Rechercher la table à l'aide de sqlalchemy et créer un dictionnaire
Utilisez l'API Twitter pour réduire le temps nécessaire à Twitter (créer une chronologie de mise en évidence (comme))
Essayez d'utiliser l'API PeeringDB 2.0
Créer un diagramme de corrélation à partir de l'historique des conversations de Twitter
L'histoire de la création d'une base de données à l'aide de l'API Google Analytics
Créer une API qui renvoie les données d'un modèle à l'aide de turicreate
Exporter le contenu acquis par Twitter Streaming API au format JSON
Utilisez sqlalchemy pour rechercher la table DB et créer un Dataflame pour les pandas
J'ai créé un bot Twitter avec Go x Qiita API x Lambda
[Ev3dev] Créez un programme qui capture LCD (écran) en utilisant python
Créez facilement un TalkBot en utilisant Discord.py et l'API Talk d'A3RT (pya3rt).
[LINE Messaging API] Créez un BOT qui se connecte à quelqu'un avec Python
Créer une API REST pour faire fonctionner dynamodb avec le Framework Django REST
Créer une interface graphique python à l'aide de tkinter
Créer un dictionnaire imbriqué à l'aide de defaultdict
Créez une API en utilisant hug avec mod_wsgi
Accéder à l'API Twitter avec Python
J'ai essayé d'utiliser l'API checkio
Créez un wrapper de langage C à l'aide de Boost.Python
Créer un bot LINE avec Django
[Python] J'ai créé un bot Twitter qui génère des tweets semblables à des amis avec des règles de chaîne de Markov.
Créer un enregistrement avec des pièces jointes dans KINTONE à l'aide du module de requêtes Python
Hanashi qui est devenu un BOT en lecture en essayant de faire un BOT d'analyse morphologique en utilisant la populaire API LINE BOT
Tweet la météo avec bot
Créer un bot pour retweeter les informations sur le virus corona
Essayez d'utiliser l'API Wunderlist en Python
Connaissance lors de la création d'un bot en utilisant discord.py
Classe qui atteint l'API de DMM
Essayez d'utiliser l'API Kraken avec Python
[Python] Créer un environnement Batch à l'aide d'AWS-CDK
[Python] [LINE Bot] Créer un robot LINE de retour de perroquet
Créer un bloc de données à partir d'Excel à l'aide de pandas