[PYTHON] Create a bot to retweet coronavirus information

Deliverables

Twitterbot that automatically retweets coronavirus-related tweets sent by Japanese public institutions

https://twitter.com/CovidJaGovRT

Background and purpose

Is there anything I can do on my own with growing anxiety about the coronavirus every day? I thought. Therefore, I recently created an app that uses the twitter API, and I think it can be used to spread correct coronavirus information. However, since I don't have any knowledge of viruses, I can't judge what is the correct information, let alone spread the wrong information and confuse it, so I want to avoid it. Among them, I came up with the idea that I could use the coronavirus information sent by local governments and ministries that are public institutions. I want you to be accurate and to know more about the information provided by people who work for residents in public institutions. With that in mind, I created a bot that retweets tweets about coronavirus information from public institutions.

Main technologies used

• AWS Lambda
• Serverless Framework
• python
• tweepy(twitter API)

Overview

--Get a tweet with "Corona" or "COVID" from your timeline on Twitter and like RT --Follow only public institutions so that only public institutions can be output on the timeline --Matching is done with regular expressions --Start every 20 minutes using AWSCloudWatchEvents cron

Applicable source

serverless.yml

service: gov-info-COVID-19

plugins:
  - serverless-python-requirements

provider:
  name: aws
  runtime: python3.8
  stage: prod
  region: ap-northeast-1

custom:
  defaultStage: prod
  profiles:
    prod: sls
  otherfile:
    environment:
      prod: ${file(env.yml)}

functions:
  RT_bot:
    handler: handler.main
    events:
      - schedule: cron(*/20 * * * ? *)
    environment:
      CONSUMER_KEY: ${self:custom.otherfile.environment.${self:provider.stage}.CONSUMER_KEY}
      CONSUMER_SECRET: ${self:custom.otherfile.environment.${self:provider.stage}.CONSUMER_SECRET}
      ACCESS_TOKEN: ${self:custom.otherfile.environment.${self:provider.stage}.ACCESS_TOKEN}
      ACCESS_TOKEN_SECRET: ${self:custom.otherfile.environment.${self:provider.stage}.ACCESS_TOKEN_SECRET}

handler.py

import configparser
import json
import os
import sys
import re
import logging
import traceback
sys.path.append(os.path.join(os.path.dirname(__file__), 'lib'))
from lib import tweepy

#For local development
OAUTH_INI = configparser.ConfigParser()
OAUTH_INI.read('oauth.ini', encoding='utf-8')
CONSUMER_KEY = os.environ['CONSUMER_KEY']
CONSUMER_SECRET = os.environ['CONSUMER_SECRET']
ACCESS_TOKEN = os.environ['ACCESS_TOKEN']
ACCESS_TOKEN_SECRET = os.environ['ACCESS_TOKEN_SECRET']


def main(event, context):
    tweepy_api = tweepy_oath()
    retweet(tweepy_api)

#twitter API authentication
def tweepy_oath():
    auth = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET)
    auth.set_access_token(ACCESS_TOKEN, ACCESS_TOKEN_SECRET)
    return tweepy.API(auth)

#Retweet tweets containing the target word from your timeline
def retweet(tweepy_api):
    exclude_exp_obj = re.compile(r'.*(corona|COVID)(.|\s)*$')

    for tweet in tweepy_api.home_timeline(count=200):
        if exclude_exp_obj.match(tweet.text):
            id = tweet.id

            try:
                tweepy_api.create_favorite(id)
                tweepy_api.retweet(id)
            #Exception occurrence leaves only log and does not stop processing
            except:
                logging.error(traceback.format_exc())
                pass

in conclusion

I sincerely hope that the coronavirus will settle down and return to peace as soon as possible.