I tried to delete bad tweets regularly with AWS Lambda + Twitter API
Image
――I want to make it so that bad tweets that are neither favorite nor RT disappear automatically. ――I want to use Lambda, which has a cool image. ――If you say "I posted it on Qiita", it won't be a bad tweet.
Things necessary
App registration on Twitter Developers
https://developer.twitter.com
I will omit the method of applying for the use of Twitter API. This time, I applied to Twitter Developers to delete tweets for the purpose of learning the API, and obtained the following keys and tokens necessary for using the Twitter API.
- API key
- API secret key
- Access token
- Access token secret
AWS account
https://aws.amazon.com The services used are as follows.
--Lambda (access to Twitter API) --Key Management Service (encryption / decryption of environment variables that store API keys) --CloudWatch (Create event for periodic processing and save execution log)
All of them are called only dozens of times a day, so I think it will fit in the free frame to about 100 yen a month.
Local Python3 execution environment
It is used when downloading a library file that is not in the Lambda environment to the local environment once with pip.
Create Lambda layer
Additional libraries such as requests_oauthlib used to access the Twitter API from Python must be uploaded to the Lambda environment by yourself.
You can upload the library code along with the main code, but Lambda's "layers" are useful because you can reuse the library with other Lambda functions. The contents of the file registered as a layer are combined under the / opt directory of the Lambda execution environment.
The additional libraries used this time are as follows, and these will be registered as layers.
- requests_oauthlib
- python-dateutil
- aws_lambda_logging
- aws_lambda_powertools
twitter-api layer creation
First, in order to use the Twitter API, create a layer "twitter-api" that contains requests_oauthlib and dateutil.
Use pip3 -t to download additional libraries in your local python directory and zip them together. This will be combined under / opt in the Lambda runtime environment. (It becomes / opt / python.)
bash
$ mkdir aws
$ cd aws
$ pip3 install requests_oauthlib python-dateutil -t ./python
$ zip -r twitter.zip ./python
$ rm -rf ./python
Open "Lambda> Layers" and click "Create Layer" to open the layer settings screen.
Upload "twitter.zip" created by the above command with "Upload .zip file", add "Python 3.7" to "Compatible runtime", and click the "Create" button.
If you get DistutilsOptionError with pip
In the macOS environment, adding the -t option to pip may cause DistutilsOptionError, so create ~ / .pydistutils.cfg to handle it.
$ echo -e "[install]\nprefix=" > ~/.pydistutils.cfg
aws-lambda-powertools layer creation
Follow the same procedure to create a layer containing aws_lambda_logging and aws_lambda_powertools.
bash
$ pip3 install aws_lambda_logging aws_lambda_powertools -t ./python
$ zip -r powertools.zip ./python
$ rm -rf ./python
The following is omitted
Creating a Lambda function
Click "Create Function" from "Lambda> Function", enter "Create from scratch"-> "Function Name", and select Python 3.7 for "Runtime" to create the function.
Add layer
We will add layers to the created function.
Click "Layers" in "Designer" and the "Add Layer" button will appear. Click it.
The "twitter-api" you created earlier will appear in the compatible layer, so select it and click the "Add" button.
Add the "aws-lambda-powertools" layer as well.
You can now call additional libraries from your function code.
Function code
I will write a script in "function code". The method specified in "Handler" is called by Lambda. It's called lambda_function.lambda_handler because it calls the lambda_handler function in lambda_function.py.
Below is the source code.
#Used to decrypt environment variables
import boto3
import os
from base64 import b64decode
#Used to use Twitter API
from requests_oauthlib import OAuth1Session
import json
#Used to parse and calculate the date and time string of a tweet
from datetime import datetime, timezone
from dateutil.parser import parse
#Used to leave a message in the CloudWatch log
from aws_lambda_powertools.logging import logger_setup, logger_inject_lambda_context
logger = logger_setup()
#Environment variable decoding function
def decrypt(environ_key):
encrypted = os.environ[environ_key]
return boto3.client('kms').decrypt(CiphertextBlob=b64decode(encrypted))['Plaintext'].decode('utf-8')
#Tweet delete function
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 acquisition function
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)
#Decorator for log output with function
@logger_inject_lambda_context
#Handler method called from Lambda
def lambda_handler(event, context):
#Get the setting value from the environment variable
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"])
#Create a session by getting / decrypting the key and token from the environment variables
session = OAuth1Session(
decrypt("API_KEY"),
decrypt("API_SECRET"),
decrypt("ACCESS_TOKEN"),
decrypt("ACCESS_TOKEN_SECRET")
)
#Get the timeline
try:
timeline = get_timeline(session, screen_name)
except Exception as e:
logger.exception(e)
raise e
#Processing for each 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)
#True for quote retweets
is_quoted_rt = True if tweet.get("quoted_status") else False
#True for regular retweets
is_rt = True if tweet.get("retweeted_status") and not is_quoted_rt else False
#True for reply(Excludes replies to yourself)
is_reply = True if in_reply_to_user_id and in_reply_to_user_id != user_id else False
#True if not bad
is_popular = favorite_count >= favorite_passing_count or retweet_count >= retweet_passing_count
#If everything is True, it will be True.
if all([
not is_rt,
not is_reply,
not is_popular,
elapsed_td.days >= days_wait,
]):
#Log message output
logger.info({
"tweet_id": tweet_id,
"created_at": created_at,
"text": tweet.get("text"),
"delete": delete_tweet(session, tweet_id),
})
#Lambda handler method requires return
return
This is a code that retrieves your recent tweets and deletes tweets that have not reached the specified number of favorites or RTs.
ʻAws_lambda_powertools.logging` is used to send which tweets have been deleted to the CloudWatch log. If you leave it in the CloudWatch log, you can easily search it on the AWS console.
The setting value is set in the environment variable.
Access tokens are encrypted and stored. I use AWS KMS (Key Management Service), but I will omit the method.
DAYS_WAIT is the number of days until it is deleted, FAVORITE_PASSING_COUNT is the pass line for the number of favorites, RETWEET_PASSING_COUNT is the pass line for the number of retweets, and SCREEN_NAME is your account name.
Addition of periodic trigger
In Designer, click the Add Trigger button to create a CloudWatch Events trigger.
I created an "every-1h" rule that triggers every hour with rate (1 hour).
Log check
ʻAws_lambda_powertools.logging` leaves a message in the CloudWatch log so you can easily check it on the console.
This is a picture of Marky Mark Good Vibrations tweeted, but it was deleted because I never had RT and favorites.
Recommended Posts