[PYTHON] J'ai essayé de supprimer régulièrement les mauvais tweets avec l'API AWS Lambda + Twitter

Image

―― Je veux faire en sorte que les mauvais tweets qui ne sont ni favoris ni RT disparaissent automatiquement. ――Je veux utiliser Lambda, qui a une image sympa. «Si vous dites" Je l'ai posté sur Qiita ", ce ne sera pas un mauvais tweet.

Les choses nécessaires

Inscription de l'application sur les développeurs Twitter

https://developer.twitter.com

Je vais omettre la méthode de demande d'utilisation de l'API Twitter. Cette fois, j'ai demandé aux développeurs Twitter de supprimer les tweets dans le but d'apprendre l'API, et j'ai obtenu les clés et jetons suivants nécessaires à l'utilisation de l'API Twitter.

• API key
• API secret key
• Access token
• Access token secret

Compte AWS

https://aws.amazon.com Les services utilisés sont les suivants.

--Lambda (accès à l'API Twitter) --Key Management Service (cryptage / décryptage des variables d'environnement qui stockent les clés API) --CloudWatch (Créer un événement pour le traitement périodique et enregistrer le journal d'exécution)

Tous ne sont appelés que des dizaines de fois par jour, donc je pense qu'ils rentreront dans le cadre gratuit à environ 100 yens par mois.

Environnement d'exécution local Python3

Il est utilisé lors du téléchargement d'un fichier de bibliothèque qui ne se trouve pas dans l'environnement Lambda dans l'environnement local une fois avec pip.

Créer une couche Lambda

Les bibliothèques supplémentaires telles que requests_oauthlib utilisées pour accéder à l'API Twitter à partir de Python doivent être téléchargées par vous-même dans l'environnement Lambda.

Vous pouvez télécharger le code de la bibliothèque avec le code principal, mais les "couches" de Lambda sont utiles car vous pouvez réutiliser la bibliothèque avec d'autres fonctions Lambda. Le contenu du fichier enregistré en tant que couche est combiné sous le répertoire / opt de l'environnement d'exécution Lambda.

Les bibliothèques supplémentaires utilisées cette fois sont les suivantes, et celles-ci seront enregistrées en tant que couches.

• requests_oauthlib
• python-dateutil
• aws_lambda_logging
• aws_lambda_powertools

création de couche twitter-api

Tout d'abord, pour utiliser l'API Twitter, créez une couche "twitter-api" qui contient requests_oauthlib et dateutil.

Téléchargez les bibliothèques supplémentaires dans le répertoire local python avec pip3 -t et rassemblez-les dans un fichier zip. Ceci est combiné sous / opt dans l'environnement d'exécution Lambda. (Cela devient / opt / python.)

bash

$ mkdir aws
$ cd aws
$ pip3 install requests_oauthlib python-dateutil -t ./python
$ zip -r twitter.zip ./python
$ rm -rf ./python

Ouvrez "Lambda> Calques" et cliquez sur "Créer un calque" pour ouvrir l'écran des paramètres du calque. Téléchargez "twitter.zip" créé par la commande ci-dessus avec "Upload .zip file", ajoutez "Python 3.7" à "Compatible runtime", et cliquez sur le bouton "Create".

Si vous obtenez une erreur d'option Distutils avec pip

Dans l'environnement macOS, l'ajout de l'option -t à pip peut provoquer DistutilsOptionError, alors créez ~ / .pydistutils.cfg pour le gérer.

$ echo -e "[install]\nprefix=" > ~/.pydistutils.cfg

création de couche aws-lambda-powertools

Suivez la même procédure pour créer une couche contenant aws_lambda_logging et aws_lambda_powertools.

bash

$ pip3 install aws_lambda_logging aws_lambda_powertools -t ./python
$ zip -r powertools.zip ./python
$ rm -rf ./python

Ce qui suit est omis

Créer une fonction Lambda

Cliquez sur "Créer une fonction" dans "Lambda> Fonction", entrez "Créer à partir de zéro" -> "Nom de la fonction" et sélectionnez Python 3.7 pour "Runtime" pour créer la fonction.

Ajouter une couche

Nous ajouterons des couches à la fonction créée.

Cliquez sur "Calques" dans "Designer" et le bouton "Ajouter un calque" apparaîtra. Cliquez dessus.

Le "twitter-api" que vous avez créé précédemment apparaîtra dans la couche compatible, alors sélectionnez-le et cliquez sur le bouton "Ajouter".

Ajoutez également la couche "aws-lambda-powertools".

Vous pouvez désormais appeler des bibliothèques supplémentaires à partir de votre code de fonction.

Code de fonction

J'écrirai un script en "code de fonction". La méthode spécifiée dans "Handler" est appelée par Lambda. Il s'appelle lambda_function.lambda_handler car il appelle la fonction lambda_handler dans lambda_function.py.

Voici le code source.

#Utilisé pour décrypter les variables d'environnement
import boto3
import os
from base64 import b64decode

#Utilisé pour utiliser l'API Twitter
from requests_oauthlib import OAuth1Session
import json

#Utilisé pour analyser et calculer la chaîne de date et d'heure d'un tweet
from datetime import datetime, timezone
from dateutil.parser import parse

#Utilisé pour laisser un message dans le journal CloudWatch
from aws_lambda_powertools.logging import logger_setup, logger_inject_lambda_context
logger = logger_setup()  

#Fonction de décryptage de la variable d'environnement
def decrypt(environ_key):
    encrypted = os.environ[environ_key]
    return boto3.client('kms').decrypt(CiphertextBlob=b64decode(encrypted))['Plaintext'].decode('utf-8')

#Fonction de suppression de Tweet
def delete_tweet(session, tweet_id):
    url = "https://api.twitter.com/1.1/statuses/destroy/{}.json".format(tweet_id)
    res = session.post(url)
    return res.status_code

#Fonction d'acquisition de la chronologie
def get_timeline(session, screen_name):
    url = "https://api.twitter.com/1.1/statuses/user_timeline.json"
    
    params = {
        "screen_name": screen_name,
        "count": 100,
        "exclude_replies": False,
        "include_rts": True,
    }

    res = session.get(url, params=params)
    res.raise_for_status()
    return json.loads(res.text)

#Décorateur pour sortie de journal avec fonction
@logger_inject_lambda_context
#Méthode Handler appelée depuis Lambda
def lambda_handler(event, context):
    #Obtenir la valeur de paramètre de la variable d'environnement
    screen_name = os.environ["SCREEN_NAME"]
    days_wait = int(os.environ["DAYS_WAIT"])
    favorite_passing_count = int(os.environ["FAVORITE_PASSING_COUNT"])
    retweet_passing_count = int(os.environ["RETWEET_PASSING_COUNT"])

    #Créer une session en récupérant / déchiffrant des clés et des jetons à partir de variables d'environnement
    session = OAuth1Session(
        decrypt("API_KEY"),
        decrypt("API_SECRET"),
        decrypt("ACCESS_TOKEN"),
        decrypt("ACCESS_TOKEN_SECRET")
        )
    
    #Obtenir la chronologie
    try:
        timeline = get_timeline(session, screen_name)
    except Exception as e:
        logger.exception(e)
        raise e

    #Traitement pour chaque tweet
    for tweet in timeline:
        tweet_id = tweet.get("id")

        user_id = tweet.get("user", {}).get("id")
        in_reply_to_user_id = tweet.get("in_reply_to_user_id")

        created_at = parse(tweet.get("created_at"))
        elapsed_td = datetime.now(timezone.utc) - created_at

        favorite_count = tweet.get("favorite_count", 0)
        retweet_count = tweet.get("retweet_count", 0)

        #Vrai pour les retweets
        is_quoted_rt = True if tweet.get("quoted_status") else False
        #Vrai pour les retweets réguliers
        is_rt = True if tweet.get("retweeted_status") and not is_quoted_rt else False
        #Vrai pour la réponse(Exclut les réponses à vous-même)
        is_reply = True if in_reply_to_user_id and in_reply_to_user_id != user_id else False
        #Vrai sinon mal
        is_popular = favorite_count >= favorite_passing_count or retweet_count >= retweet_passing_count

        #Une écriture qui devient vraie si tout est vrai
        if all([
                not is_rt,
                not is_reply,
                not is_popular,
                elapsed_td.days >= days_wait,
            ]):

            #Sortie de message de journal
            logger.info({
                "tweet_id": tweet_id,
                "created_at": created_at,
                "text": tweet.get("text"),
                "delete": delete_tweet(session, tweet_id),
            })

    #La méthode du gestionnaire Lambda nécessite un retour
    return

Il s'agit d'un code qui récupère vos tweets récents et supprime les tweets qui n'ont pas atteint le nombre spécifié de favoris ou de RT.

ʻAws_lambda_powertools.logging` est utilisé pour envoyer les tweets supprimés dans le journal CloudWatch. Si vous le laissez dans le journal CloudWatch, vous pouvez facilement le rechercher sur la console AWS.

La valeur de réglage est définie dans la variable d'environnement.

Les jetons d'accès, etc. sont stockés cryptés. J'utilise AWS KMS (Key Management Service), mais j'omettrai la méthode.

DAYS_WAIT est le nombre de jours jusqu'à ce qu'il soit supprimé, FAVORITE_PASSING_COUNT est la ligne de passe pour le nombre de favoris, RETWEET_PASSING_COUNT est la ligne de passe pour le nombre de retweets, et SCREEN_NAME est le nom de votre compte.

Ajout d'un déclencheur d'exécution périodique

Cliquez sur le bouton Ajouter un déclencheur dans Designer pour créer un déclencheur CloudWatch Events.

J'ai créé une règle "toutes les 1h" qui se déclenche toutes les heures avec rate (1 hour).

Vérification du journal

ʻAws_lambda_powertools.logging` laisse un message dans le journal CloudWatch afin que vous puissiez facilement le vérifier sur la console.

C'est une photo de Marky Mark Good Vibrations, mais elle a été effacée car je n'ai jamais eu de RT et de favoris.