[PYTHON] Ich habe versucht, schlechte Tweets regelmäßig mit der AWS Lambda + Twitter API zu löschen

Bild

――Ich möchte es so gestalten, dass schlechte Tweets, die weder Favorit noch RT sind, automatisch verschwinden. ――Ich möchte Lambda verwenden, das ein cooles Bild hat. ――Wenn Sie sagen "Ich habe es auf Qiita gepostet", wird es kein schlechter Tweet.

Dinge notwendig

App-Registrierung bei Twitter-Entwicklern

https://developer.twitter.com

Ich werde die Methode zur Beantragung der Verwendung der Twitter-API weglassen. Dieses Mal beantragte ich bei Twitter-Entwicklern, Tweets zu löschen, um die API zu lernen, und erhielt die folgenden Schlüssel und Token, die für die Verwendung der Twitter-API erforderlich sind.

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

AWS-Konto

https://aws.amazon.com Die verwendeten Dienste sind wie folgt.

• Lambda (Zugriff auf Twitter API) --Key Management Service (Ver- / Entschlüsselung von Umgebungsvariablen, in denen API-Schlüssel gespeichert sind) --CloudWatch (Ereignis für die regelmäßige Verarbeitung erstellen und Ausführungsprotokoll speichern)

Alle von ihnen werden nur Dutzende Male am Tag genannt, daher denke ich, dass sie bis etwa 100 Yen pro Monat in den freien Rahmen passen werden.

Lokale Python3-Ausführungsumgebung

Es wird verwendet, wenn eine Bibliotheksdatei, die sich nicht in der Lambda-Umgebung befindet, einmal mit pip in die lokale Umgebung heruntergeladen wird.

Erstellen Sie eine Lambda-Ebene

Zusätzliche Bibliotheken wie request_oauthlib, die für den Zugriff auf die Twitter-API von Python verwendet werden, müssen selbst in die Lambda-Umgebung hochgeladen werden.

Sie können den Bibliothekscode zusammen mit dem Hauptcode hochladen, aber die "Ebenen" von Lambda sind nützlich, da Sie die Bibliothek mit anderen Lambda-Funktionen wiederverwenden können. Der Inhalt der als Layer registrierten Datei wird im Verzeichnis / opt der Lambda-Ausführungsumgebung zusammengefasst.

Die zusätzlichen Bibliotheken, die dieses Mal verwendet werden, lauten wie folgt und werden als Ebenen registriert.

• requests_oauthlib
• python-dateutil
• aws_lambda_logging
• aws_lambda_powertools

Twitter-API-Layer-Erstellung

Um die Twitter-API zu verwenden, erstellen Sie zunächst eine Ebene "twitter-api", die request_oauthlib und dateutil enthält.

Verwenden Sie pip3 -t, um zusätzliche Bibliotheken in Ihr lokales Python-Verzeichnis herunterzuladen und sie zu einer Zip-Datei zu kombinieren. Dies wird in der Lambda-Ausführungsumgebung unter / opt kombiniert. (Es wird / opt / python.)

bash

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

Öffnen Sie "Lambda> Ebenen" und klicken Sie auf "Ebene erstellen", um den Bildschirm mit den Ebeneneinstellungen zu öffnen. Laden Sie "twitter.zip", das mit dem obigen Befehl erstellt wurde, mit "ZIP-Datei hochladen" hoch, fügen Sie "Python 3.7" zu "Kompatible Laufzeit" hinzu und klicken Sie auf die Schaltfläche "Erstellen".

Wenn Sie einen Distutils Option Error mit pip erhalten

In der macOS-Umgebung kann das Hinzufügen der Option "-t" zu pip zu "DistutilsOptionError" führen. Erstellen Sie daher "~ / .pydistutils.cfg", um damit umzugehen.

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

aws-lambda-powertools Ebenenerstellung

Gehen Sie genauso vor, um eine Ebene mit aws_lambda_logging und aws_lambda_powertools zu erstellen.

bash

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

Folgendes wird weggelassen

Erstellen einer Lambda-Funktion

Klicken Sie unter "Lambda> Funktion" auf "Funktion erstellen", geben Sie "Von Grund auf neu erstellen" -> "Funktionsname" ein und wählen Sie "Python 3.7" für "Laufzeit" aus, um die Funktion zu erstellen.

Ebene hinzufügen

Wir werden der erstellten Funktion Ebenen hinzufügen.

Klicken Sie in "Designer" auf "Ebenen" und die Schaltfläche "Ebene hinzufügen" wird angezeigt. Klicken Sie darauf.

Die zuvor erstellte "Twitter-API" wird in der kompatiblen Ebene angezeigt. Wählen Sie sie aus und klicken Sie auf die Schaltfläche "Hinzufügen".

Fügen Sie auch die Ebene "aws-lambda-powertools" hinzu.

Sie können jetzt zusätzliche Bibliotheken aus Ihrem Funktionscode aufrufen.

Funktionscode

Ich werde ein Skript in "Funktionscode" schreiben. Die in "Handler" angegebene Methode wird von Lambda aufgerufen. Es heißt lambda_function.lambda_handler, weil es die Funktion lambda_handler in lambda_function.py aufruft.

Unten ist der Quellcode.

#Wird zum Entschlüsseln von Umgebungsvariablen verwendet
import boto3
import os
from base64 import b64decode

#Wird für die Verwendung der Twitter-API verwendet
from requests_oauthlib import OAuth1Session
import json

#Wird verwendet, um die Datums- und Zeitzeichenfolge eines Tweets zu analysieren und zu berechnen
from datetime import datetime, timezone
from dateutil.parser import parse

#Wird verwendet, um eine Nachricht im CloudWatch-Protokoll zu hinterlassen
from aws_lambda_powertools.logging import logger_setup, logger_inject_lambda_context
logger = logger_setup()  

#Entschlüsselungsfunktion der Umgebungsvariablen
def decrypt(environ_key):
    encrypted = os.environ[environ_key]
    return boto3.client('kms').decrypt(CiphertextBlob=b64decode(encrypted))['Plaintext'].decode('utf-8')

#Tweet Löschfunktion
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

#Timeline-Erfassungsfunktion
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)

#Dekorateur für Protokollausgabe mit Funktion
@logger_inject_lambda_context
#Handler-Methode von Lambda aufgerufen
def lambda_handler(event, context):
    #Ruft den Einstellungswert aus der Umgebungsvariablen ab
    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"])

    #Erstellen Sie eine Sitzung, indem Sie Schlüssel und Token aus Umgebungsvariablen abrufen / entschlüsseln
    session = OAuth1Session(
        decrypt("API_KEY"),
        decrypt("API_SECRET"),
        decrypt("ACCESS_TOKEN"),
        decrypt("ACCESS_TOKEN_SECRET")
        )
    
    #Holen Sie sich die Zeitleiste
    try:
        timeline = get_timeline(session, screen_name)
    except Exception as e:
        logger.exception(e)
        raise e

    #Verarbeitung für jeden 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)

        #Richtig für Retweets
        is_quoted_rt = True if tweet.get("quoted_status") else False
        #Richtig für reguläre Retweets
        is_rt = True if tweet.get("retweeted_status") and not is_quoted_rt else False
        #Richtig für die Antwort(Schließt Antworten auf sich selbst aus)
        is_reply = True if in_reply_to_user_id and in_reply_to_user_id != user_id else False
        #Richtig, wenn nicht schlecht
        is_popular = favorite_count >= favorite_passing_count or retweet_count >= retweet_passing_count

        #Schreiben, das wahr wird, wenn alle wahr sind
        if all([
                not is_rt,
                not is_reply,
                not is_popular,
                elapsed_td.days >= days_wait,
            ]):

            #Protokollnachrichtenausgabe
            logger.info({
                "tweet_id": tweet_id,
                "created_at": created_at,
                "text": tweet.get("text"),
                "delete": delete_tweet(session, tweet_id),
            })

    #Die Lambda-Handler-Methode erfordert eine Rückgabe
    return

Dies ist ein Code, der Ihre letzten Tweets abruft und Tweets löscht, die die angegebene Anzahl von Favoriten oder RTs nicht erreicht haben.

Ich verwende "aws_lambda_powertools.logging", um die von mir gelöschten Tweets an das CloudWatch-Protokoll zu senden. Wenn Sie es im CloudWatch-Protokoll belassen, können Sie es problemlos in der AWS-Konsole suchen.

Der Einstellwert wird in der Umgebungsvariablen festgelegt.

Zugriffstoken usw. werden verschlüsselt gespeichert. Ich verwende AWS KMS (Key Management Service), werde die Methode jedoch weglassen.

DAYS_WAIT ist die Anzahl der Tage bis zum Löschen, FAVORITE_PASSING_COUNT ist die Passzeile für die Anzahl der Favoriten, RETWEET_PASSING_COUNT ist die Passzeile für die Anzahl der Retweets und SCREEN_NAME ist Ihr Kontoname.

Hinzufügen eines Auslösers für die periodische Ausführung

Klicken Sie in Designer auf die Schaltfläche Trigger hinzufügen, um einen CloudWatch-Ereignis-Trigger zu erstellen.

Ich habe eine "alle 1 Stunde" -Regel erstellt, die jede Stunde mit "Rate (1 Stunde)" ausgelöst wird.

Protokollprüfung

aws_lambda_powertools.logging hinterlässt eine Nachricht im CloudWatch-Protokoll, sodass Sie sie einfach auf der Konsole überprüfen können.

Dies ist ein Bild von Marky Mark Good Vibrations, aber es wurde gelöscht, weil ich nie RT und Favoriten hatte.