-
Notifications
You must be signed in to change notification settings - Fork 0
/
lambda_function.py
107 lines (80 loc) · 3.56 KB
/
lambda_function.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
import tweepy
from ArtInstance import Art
import SendEmail
import os
from datetime import date, timedelta
'''
lambda_function and package directory are zipped into lambda.zip.
we then use lambda.zip with AWS to run this program once a week for a year
lambda_function is the same as TwitterBot, just tweaked for AWSs
'''
def auth_client():
# authentification
client = tweepy.Client(
bearer_token=os.environ['BEARER_TOKEN'],
consumer_key=os.environ['API_KEY'],
consumer_secret=os.environ['API_SECRET'],
access_token=os.environ['ACCESS_TOKEN'],
access_token_secret=os.environ['TOKEN_SECRET'],
wait_on_rate_limit=True,
)
return client
def search_tweets():
client = auth_client()
query = "(@TrashTastePod OR #TrashTaste) (#trashtasteart OR #art OR #fanart OR #digitalart OR #illustration OR #drawing) -is:retweet has:media"
# with Essential access, can only search through past 7 days of tweets, max of 100
tweets = client.search_recent_tweets(query=query, tweet_fields=['entities'],
media_fields=['url'], expansions=['attachments.media_keys', 'author_id'],
max_results=100)
return tweets
def query_tweets(tweets, client, art_class):
media = {m.media_key: m for m in tweets.includes['media']}
users = {user.id: user.username for user in tweets.includes['users']}
try:
# query tweets
for tweet in tweets.data:
media_keys = tweet.attachments['media_keys']
media_urls = [media[key].url for key in media_keys]
media_types = [media[key]['type'] for key in media_keys]
author_id = tweet.author_id
username = users[author_id]
tweet_id = tweet.id
tweet_url = tweet.entities['urls'][0]['display_url']
# add data to dict
art_class.add_art(author_id=author_id, tweet_id=tweet_id,
media_keys=media_keys, media_urls=media_urls, media_types=media_types, tweet_url=tweet_url, username=username)
# liking and retweeting
client.like(tweet.id)
client.retweet(tweet.id)
# if dict is empty, will return err
except TypeError as e:
print(f'There are no recent fan arts, try again in 7 days: {e}')
else:
print('Query complete')
def save_tweet_media(art_class):
# saves the images into a folder
# saving images at end so the api doesnt stay open for long periods
art_class.save_all_users_media()
# saves author_id, tweet_id, media_keys, media_urls, media_types, tweet_url, username to a json
art_class.save_to_json()
def lambda_handler(event, context):
# instance
trash_taste_art = Art()
client = auth_client()
tweets = search_tweets()
query_tweets(tweets, client, trash_taste_art)
save_tweet_media(trash_taste_art)
# directory variables
tmp_folder = '../../tmp'
data_folder = os.path.join(tmp_folder, 'data')
zip_folder = os.path.join(tmp_folder, 'zip')
# compress data dir as zip file
today = date.today()
last_week = date.today() - timedelta(days=7)
zip_filename = f'TT-Data-{last_week}-{today}.zip'
SendEmail.check_folder_exists(zip_folder)
SendEmail.zip_dir(data_folder, zip_folder, zip_filename)
SendEmail.send_email(zip_folder, zip_filename,
os.environ['EMAIL_ADDRESS'], os.environ['GMAIL_APP_PASS'], os.environ['CC_EMAIL_ADDRESS'])
SendEmail.remove_folder_contents(data_folder)
SendEmail.remove_folder_contents(zip_folder)