More than 3 years have passed since last update.

株式会社エーピーコミュニケーションズ

AWS LambdaとServerless Frameworkで爆速で作るTwitterbot

Last updated at 2020-08-29Posted at 2019-12-28

0. はじめに

ここ1年はStackstormばかり扱っているのですが、年末だし他の技術も触るかー！と思いたち、色々自分の作業ディレクトリを漁っていたところ、Twitterbotなるものを発掘しました。

Stackstorm？？？という方はこちらをご参照ください。(自演)
Dockerで始めるStackstorm再入門1/3(環境構築からOrquestaで書いたWorkflowの結果をslackに通知する)

話を戻します。
そのTwitterbotですが、私はvpsを使って運用していました。
ただ、そんなに頻繁に動かさないので、また勉強も兼ねて、AWS Lambda(以下、lambda)移行にチャレンジした次第です。

lambdaってなに？という方は、AWSがオフィシャルなハンズオンを公開しているので、そちらをご参照ください。

先日公開した AWS Hands-on for Beginners の紹介記事を書きました！よく頂くご質問についてもまとめています。最初のステップにぜひご活用いただければと思います🙏 / “実際に手を動かして学ぶ！AWS Hands-on for Beginners のご紹介 | Amazon Web Services ブログ” https://t.co/dasJxb5IDV
— ketancho 🙂｜Kei Kanazawa (@ketancho) November 11, 2019

本記事では、サンプルコードも用意しているので、ぜひお試しください。

1. 目次

1. 環境/バージョン情報
1. Serverless Frameworkとは
1. ローカル開発環境の準備
1. Twitterのtokenなどセキュアなパラメータの取り扱い
1. serverless.ymlを編集
1. lambdaの実行ファイルを編集
1. lambdaをローカルから実行
1. lambdaをデプロイ
1. デバッグコマンド
1. 参考

2020/05/23 16:10更新

P.P.S サンプルコード追加しました。

2.環境/バージョン情報

ローカル開発環境

Ubuntu: 19.04 (Disco Dingo)
npm: 6.13.4
Python: 3.7.3
- tweepy: 3.8.0
- oauthlib: 3.1.0
- requests: 2.22.0
- requests-oauthlib: 1.3.0他
serverless
- Framework Core: 1.60.4
- Plugin: 3.2.6
- SDK: 2.2.1
- Components Core: 1.1.2
- Components CLI: 1.4.0

AWS Lambda

python3.7

3. Serverless Frameworkとは

そもそもServerless Frameworkとはなんなのか。
公式ドキュメントでは、このように紹介されています。

The Serverless Framework consists of an open source CLI that makes it easy to develop, deploy and test serverless apps across different cloud providers, as well as a hosted Dashboard that includes features designed to further simplify serverless development, deployment, and testing, and enable you to easily secure and monitor your serverless apps.

参考: Serverless Framework Documentation

個人的に感じた特徴

node.js製フレームワーク
ローカルで開発したファンクションをserverlessコマンドを使って任意のプロバイダープラットフォーム(AWS, GCPなど)にデプロイ
デプロイの設定はserverless.ymlで定義

4. ローカル開発環境の準備

サンプルリポジトリをクローン
AWSクレデンシャルの設定
Serverless Frameworkのインストールとプロジェクトの作成
プラグインのインストール

サンプルリポジトリをクローン

gkz@localhost ~$ git clone https://github.com/gkzz/lambda_twbot.git \
&& cd lambda_twbot

AWSクレデンシャルの設定

gkz@localhost ~$ aws configure

~/.aws/config

[default]
region = ap-northeast-1
output = json

~/.aws/credentials

[default]
aws_access_key_id = xxxxxxxxxxx
aws_secret_access_key = yyyyyyyyyyyyy

Serverless Frameworkのインストールとプロジェクトの作成

gkz@localhost ~$ sudo npm init
(略)
gkz@localhost ~$ sudo npm install -g serverless
gkz@localhost ~$ serverless create \
> --template aws-python3 \
> --name src \
> --path src
Serverless: Generating boilerplate...

(略)

 _______                             __
|   _   .-----.----.--.--.-----.----|  .-----.-----.-----.
|   |___|  -__|   _|  |  |  -__|   _|  |  -__|__ --|__ --|
|____   |_____|__|  \___/|_____|__| |__|_____|_____|_____|
|   |   |             The Serverless Application Framework
|       |                           serverless.com, v1.60.4
 -------'

Serverless: Successfully generated boilerplate for template: "aws-python3"

※ プロジェクト名をsrcとする場合、サンプルコードで用意しているプロジェクトディレクトリと名前が被るので、サンプルコードのプロジェクトデイレクトリのsrcを任意の名前に変更してください。

gkz@localhost ~$ mv src src.old # src.oldという名前に変更している

プラグインのインストール

今回は以下の3つのプラグインをインストールしました。

プラグインの名称	目的
serverless-python-requirements	ライブラリをインストールするため
serverless-dotenv-plugin	環境構築を読み込むため
serverless-offline	ローカル開発環境でlambdaを実行するため

参考までにserverless-python-requirementsをインストールした際のコマンドを貼りますが、他の2つも同様にインストールしています。

gkz@localhost ~$ sudo npm install --save serverless-python-requirements

プロジェクト構成

serverless,ymlを編集する前にディレクトリ構成を確認しておきましょう。
特に確認したい点は以下の2点です。

serverless.ymlからみた.envの配置場所です。
- Twitterのtokenは./config/.envから読み取ります。
requirements.txtはserverless.ymlと同じ階層に配置します。

gkz@localhost ~$ cat src/handler.py.tmpl > src/handler.py
gkz@localhost ~$ cat config/.env.tmpl > config/.env
gkz@localhost ~$ tree -L 2
.
└── src                    # プロジェクトフォルダ
    ├── 37                 # python -m venv $nameで作ったpython3.7の仮想環境の名称
    ├── config             # .envなど変数が記載されたファイルの親ディレクトリ
    ├── handler.py         # lambdaの実行関数
    ├── node_modules
    ├── package.json
    ├── package-lock.json
    ├── __pycache__
    ├── requirements.txt   # tweepyなど必要なライブラリが記載
    ├── serverless.yml     # lambdaをローカル開発環境からデプロイするのに使う
    └── serverless.yml.org # プロジェクトフォルダを作る際に生成されたserverless.ymlのサンプルファイル

5 directories, 6 files

プロジェクトディレクトリが生成されたことが確認できたら、そのディレクトリに移動してください。

5. Twitterのtokenなどセキュアなパラメータの取り扱い

ここでは、Twitterのtokenなど必要かつセキュアなパラメータは環境変数として取り扱うようにし、./lambda_twbot/src/config/.envに記載することとしています。
.envファイルの配置場所を変える場合、後述する/lambda_twbot/serverless.ymlで記載するbasePath: ./config/ の箇所を適宜変更してください。
なお、サンプルコードではtokenの他に環境変数として$STAGEと$REGIONを用意していますが、こちらも後述する/lambda_twbot/serverless.ymlにベタ書きでもいいと思います。

./lambda_twbot/src/config/.env


# base
STAGE=dev
REGION=ap-northeast-1

# twitter
CONSUMER_KEY=xxxxxxxxxxxxxxxxxxxxxxxxxxx
CONSUMER_SECRET=yyyyyyyyyyyyyyyyyyyyyyyyyyy
ACCESS_TOKEN=aaaaaaaaaaaaaaaaaaaaaaaa
ACCESS_TOKEN_SECRET=bbbbbbbbbbbbbbbbbbb

6. serverless.ymlを編集

./lambda_twbot/serverless.yml

service: src

custom:
  dotenv:
    basePath: ./config/  # ./lambda_twbot/src/.config/
  stage: ${env:STAGE}    # basepath/.envから環境変数STAGEを読み込む
  region: ${env:REGION}　# 同様に環境変数REGIONを読み込む
  pythonRequirements:
    dockerizePip: non-linux

provider:
  name: aws
  runtime: python3.7
  stage: ${self:custom.stage}
  region: ${self:custom.region}

plugins:
  - serverless-python-requirements
  - serverless-dotenv-plugin
  - serverless-offline

functions:
  rtweet:
    handler: handler.rtweet  # $filename.$function(handler.pyのrtweet関数を実行)
    memorySize: 256
    timeout: 90s
    events:
      - schedule: cron(0/20 * * * ? *)  # 毎日20分おき
  
  fav:
    handler: handler.fav
    memorySize: 256
    timeout: 90s
    events:
      - schedule: cron(0/600 * * * ? *)　# 毎日600分おき

7. lambdaの実行ファイルを編集

./lambda_twbot/src/handler.py

# coding: UTF-8
try:
    import unzip_requirements
except ImportError:
    pass

import tweepy
import time
import os
import random
import traceback

def _set_token():
    """ set twitter token """

    CK = os.environ['CONSUMER_KEY']
    CS = os.environ['CONSUMER_SECRET']
    AT = os.environ['ACCESS_TOKEN']
    AS = os.environ['ACCESS_TOKEN_SECRET']

    auth = tweepy.OAuthHandler(CK, CS)
    auth.set_access_token(AT, AS)
    return tweepy.API(auth)


def _find_tweet(token):
    found = []

    kws = [
        '#python 本',
        '#lambda　楽しい',
    ]

    for kw in kws:
        for tweet in tweepy.Cursor(token.search, kw).items(5):
            found.append(tweet)
    
    return found


def rtweet(event, context):
    """ entry point of rtweet """

    counter = 0

    token = _set_token()
    tweets = _find_tweet(token)

    for tweet in tweets:
        try:
            #print('\nRetweet Bot found tweet by @' + tweet.user.screen_name + '. ' + 'Attempting to retweet.')
            tweet.retweet()
            logger.info(tweet.retweet())
            counter+=1
            #print('Retweet published successfully.')
    
            # Where sleep(10), sleep is measured in seconds.
            # Change 10 to amount of seconds you want to have in-between retweets.
            # Read Twitter's rules on automation. Don't spam!
            time.sleep(random.randint(2,5))
            
        # Some basic error handling. Will print out why retweet failed, into your terminal.
        except tweepy.TweepError as error:
            print('''Retweet not successful. 
            Reason: {r}
            '''.format(r=error.reason))
    
        except StopIteration:
            print(traceback.format_exc())
            break
    
    return "RtweetConts: {num}".format(num=counter)

    

def fav(event, context):
    """ entry point of fav """

    counter = 0

    token = _set_token()
    tweets = _find_tweet(token)

    for tweet in tweets:
        try:
            tweet.favorite()
    
            # Where sleep(10), sleep is measured in seconds.
            # Change 10 to amount of seconds you want to have in-between retweets.
            # Read Twitter's rules on automation. Don't spam!
            time.sleep(random.randint(2,5))
            
        # Some basic error handling. Will print out why retweet failed, into your terminal.
        except tweepy.TweepError as error:
            print('''Fav not successful. 
            Reason: {r}
            '''.format(r=error.reason))
    
        except StopIteration:
            print(traceback.format_exc())
            break
    
    return "FavConts: {num}".format(num=counter)
    


#if __name__ == "__main__":
#    rtweet('', '')

./lambda_twbot/src/requirements.txt

certifi==2019.11.28
chardet==3.0.4
idna==2.8
oauthlib==3.1.0
PySocks==1.7.1
requests==2.22.0
requests-oauthlib==1.3.0
six==1.13.0
tweepy==3.8.0
urllib3==1.25.7
```

## 8. lambdaをローカルから実行
実行する前に`requirements.txtからライブラリをインストール`することと、`-f`あるいは`-function`で実行関数を指定することを忘れないでください。

```
gkz@localhost ~$ python3.7 -m venv 37 && \
> source 37/bin/activate \
> pip install -r requirements.txt 
gkz@localhost ~$ serverless invoke local --function rtweet
gkz@localhost ~$ serverless invoke local --f fav
```

## 9. lambdaをデプロイ

デプロイは以下のコマンドで実行できます。

```
gkz@localhost ~$ serverless deploy -v
gkz@localhost ~$ sls deploy -v
```
ファイルを編集した際には一度削除してから改めてデプロイします。
やりかたは`deployをremoveに変える`だけです。

```
gkz@localhost ~$ serverless remove
gkz@localhost ~$ sls remove
```

## 10. デバッグコマンド

serverless.ymlが読み込まれているかデバッグさせるコマンドもあります。

```
gkz@localhost ~$ sls print
gkz@localhost ~$ serverless print
Serverless: DOTENV: Loading environment variables from ./config/.env:
Serverless: 	 - STAGE
Serverless: 	 - REGION
Serverless: 	 - CONSUMER_KEY
Serverless: 	 - CONSUMER_SECRET
Serverless: 	 - ACCESS_TOKEN
Serverless: 	 - ACCESS_TOKEN_SECRET
service: src
custom:
  dotenv:
    basePath: ./config/
  stage: dev
  region: ap-northeast-1
  pythonRequirements:
    dockerizePip: non-linux
provider:
  stage: dev
  region: ap-northeast-1
  name: aws
  runtime: python3.7
plugins:
  - serverless-python-requirements
  - serverless-dotenv-plugin
  - serverless-offline
functions:
  rtweet:
    handler: handler.rtweet
    memorySize: 256
    timeout: 90s
    events:
      - schedule: cron(0/20 * * * ? *)
    name: src-dev-rtweet
  fav:
    handler: handler.fav
    memorySize: 256
    timeout: 90s
    events:
      - schedule: cron(0/600 * * * ? *)
    name: src-dev-fav
```

## 11. 参考
-  公式ドキュメント
  - [Serverless Framework Documentation](https://serverless.com/framework/docs/)  
  - [How to Handle your Python packaging in Lambda with Serverless plugins](https://serverless.com/blog/serverless-python-packaging/)
- サンプルコード
  - [serverless/examples: Serverless Examples – A collection of boilerplates and examples of serverless architectures built with the Serverless Framework on AWS Lambda, Microsoft Azure, Google Cloud Functions, and more.](https://github.com/serverless/examples)
-  環境変数の取り扱い
  - [Managing multiple environments with Express & Serverless Framework](https://sangeeta.io/2019/05/01/express-serverless/#disqus_thread)
- [Rate または Cron を使用したスケジュール式](https://docs.aws.amazon.com/ja_jp/lambda/latest/dg/tutorial-scheduled-events-schedule-expressions.html)
  - [cronの書き方確認サイト](https://crontab.guru)
- サンプルリポジトリ
 - [lambda_twbot](https://github.com/gkzz/lambda_twbot)
- Twitterbotのサンプルコード
  - [Create a Twitter Bot in Python Using Tweepy](https://medium.com/free-code-camp/creating-a-twitter-bot-in-python-with-tweepy-ac524157a607) 　

 
## P.S. Twitterもやってるのでフォローしていただけると泣いて喜びます:)
[@gkzvoice](https://twitter.com/gkzvoice)


## P.P.S サンプルコード追加

- [event-news-bot](https://github.com/gkzz/event-news-bot)
  - イベント情報ポータルサイトの[connpass](https://connpass.com/)さんから「オンライン」関連のイベントをTwitterに定期的に投稿するBOTを作ったので、そのソースコードを公開します。
  - BOTの[@EventNewsBot](https://twitter.com/EventNewsBot)もフォローしていただけるとうれしいです。

Tweetしたい場合、Tweepyのupdate_statusメソッドを使います。

>API.update_status(status[, in_reply_to_status_id][, lat][, long][, source][, place_id])
Update the authenticated user’s status. Statuses that are duplicates or too long will be silently ignored.

出所：[tweepy 3.5.0 documentation](http://docs.tweepy.org/en/v3.5.0/api.html)

```event-news-bot/src/handler.py

# coding: UTF-8
try:
    import unzip_requirements
except ImportError:
    pass

import tweepy
import time
import os
import random
import traceback
import requests


def _set_token():
    """ set twitter token """

    CK = os.environ['CONSUMER_KEY']
    CS = os.environ['CONSUMER_SECRET']
    AT = os.environ['ACCESS_TOKEN']
    AS = os.environ['ACCESS_TOKEN_SECRET']

    auth = tweepy.OAuthHandler(CK, CS)
    auth.set_access_token(AT, AS)
    return tweepy.API(auth)
    


def _get_content(keyword, number_events):
    """ get content

    Args:
        keyword {type: string, default: online}: keyword to search events
        number_events {type: int, default: 2}: number of events
    
    Returns:
        content {type: list}: contents is list of some dict, which'keys are title and url

    """

    BASE_URL = 'https://connpass.com/api/v1/event/'

    params = {
        'keyword': keyword,
        'count': 30,
    }
    response = requests.get(BASE_URL, params=params)
    #return [{'title': r['title'], 'url': r['event_url']} for r in response.json()['events']]

    content = []
    for r in random.sample(response.json()['events'], number_events):
        content.append({
            'title': r['title'], 'started_at': r['started_at'], 
            'event_url': r['event_url']
        })
    return content




def tweet(event, context):
    """ entry point of tweet """

    response = []

    try:
        token = _set_token()
        response = _get_content(keyword='online', number_events=2)


        for resp in response:
            tw = """
            イベント名：{title}\n開始時間： {started_at}\n{event_url}
            """.format(
                title=resp['title'], started_at=resp['started_at'], 
                event_url=resp['event_url']
            )

            token.update_status(tw)
            
            # Where sleep(10), sleep is measured in seconds.
            # Change 10 to amount of seconds you want to have in-between retweets.
            # Read Twitter's rules on automation. Don't spam!
            time.sleep(random.randint(2,5))

    except tweepy.TweepError as error:
            print("""Tweet not successful. 
            Reason: {r}
            """.format(r=error.reason))

    
    except:
        return traceback.format_exc()
    
    return response




#if __name__ == "__main__":
#    import pdb; pdb.set_trace()
#    tweet('', '')


```

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up