LoginSignup
5
6

More than 5 years have passed since last update.

@bakira

Qiita記事をTwitterに投稿しSlackに通知するスクリプトをAWSで定期実行してみた

Posted at

概要

以下の記事でPythonスクリプトから各APIを操作する方法をを紹介しました。

今回は、

  1. QiitaAPIを使って記事一覧からランダムで取得
  2. TwitterAPIを使っていい感じに投稿する
  3. Incoming Webhookを使って実行結果をSlackに通知する
  4. AWS lambdaに乗せて実行をする
  5. AWS CloutWatchで定期実行する

という部分を実装してみようと思います。

手順

環境

  • Python 3.6.1
  • requests-oauthlib==0.8.0
  • requests==2.18.4

前提

以下のものは事前に用意しておいてください。

  • Python関連
    • requestsのインストール
    • requests-oauthlibのインストール
  • Twitter関連
    • Twitterアカウント
    • Consumer Key
    • Consumer Secret
    • Access Token
    • Access Token Secret
  • AWS関連
    • AWSアカウント
  • Qiita関連
    • Qiitaアカウント
    • Qiita記事
  • Slack関連
    • Incoming Webhookの設定
    • 通知先Chanel

また、今回はリファクタリング等はしていないので冗長的なコーディングになってしまっている部分がありますが、本質ではないので今回はそのままとさせていただきますorz

Twitterへの投稿

まずはQiitaAPIを利用して、ランダムで投稿記事を取得し、Twitterに投稿するところまで実装してみましょう。

config.py
CONSUMER_KEY = "{コンシューマーキー}"
CONSUMER_SECRET = "{コンシューマーシークレット}"
ACCESS_TOKEN = "{アクセストークン}"
ACCESS_TOKEN_SECRET = "{アクセストークンシークレット}"
script.py
import json, config, requests, random, sys  #必要なライブラリの読み込み 
from requests_oauthlib import OAuth1Session  #OAuthのライブラリの読み込み

CONSUMER_KEY = config.CONSUMER_KEY
CONSUMER_SECRET = config.CONSUMER_SECRET
ACCESS_TOKEN = config.ACCESS_TOKEN
ACCESS_TOKEN_SECRET = config.ACCESS_TOKEN_SECRET
twitter = OAuth1Session(CONSUMER_KEY, CONSUMER_SECRET, ACCESS_TOKEN, ACCESS_TOKEN_SECRET) #認証処理

TWITTER_API_HOST = "https://api.twitter.com/1.1"  #TwitterAPIホスト
STATUSES_UPDATE_ENDPOINT = "/statuses/update.json"  #ツイートエンドポイント

TARGET_USER_NAME = "{Qiitaユーザー名}"  #取得対象のQiitaユーザー名
QIITA_API_HOST = "https://qiita.com/api/v2"  #QiitaAPIホスト
USER_ENDPOINT = f"/users/{TARGET_USER_NAME}"  #ユーザー情報取得エンドポイント
USER_ITEMS_ENDPOINT = f"/users/{TARGET_USER_NAME}/items"  #ユーザー記事一覧取得エンドポイント
PAGE_PER_COUNT = 20  #1ページ辺りの記事取得件数

response = requests.get(
    QIITA_API_HOST + USER_ENDPOINT
).json()  #対象Qiitaアカウント情報をJsonで取得

totalItemCount = response['items_count']  #投稿総数を取得
pageCount = totalItemCount // PAGE_PER_COUNT  #1ページ辺りの記事取得件数で取得出来たページ数を取得
hasSurplus = totalItemCount % PAGE_PER_COUNT != 0  #余りの記事数を取得
totalPageCount = pageCount + 1 if hasSurplus else pageCount #余りの記事数があった場合、端数分のページ数を追加

params = {
    "page": random.randint(1,totalPageCount),  #ページ番号をランダム生成
    "per_page": PAGE_PER_COUNT
}  #記事一覧をページ番号と１ページ辺りの表示件数を指定するためのパラメータを生成

response = requests.get(
    QIITA_API_HOST + USER_ITEMS_ENDPOINT,
    params=params
).json()  #対象Qiitaアカウントの記事一覧を取得

if( response == [] ):  #記事一覧が空だった場合
    print("アイテムを取得出来ませんでしたので終了します。")
    sys.exit()  #処理終了


item = random.choice(response)  #取得した一覧からランダムで記事を取得
url = item["url"]  #記事URLを取得
title = item["title"]  #記事タイトルを取得
tags = map(
    lambda element: ("#" + element["name"]), 
    item["tags"] 
)  #タグ一覧をTwitterハッシュ化して配列として再生成
tagsStr = ' '.join(list(tags))  #タグ一覧を文字列結合

params = {
    "status" : "【Qiita】" +title + "\n" + tagsStr + "\n" + url  #ツイート内容を生成
}  #Postするパラメータを生成

res = twitter.post( 
    TWITTER_API_HOST + STATUSES_UPDATE_ENDPOINT ,
    params = params
)  #APIにPost通信

if res.status_code == 200: #正常投稿出来た場合
    print("Success.")
else: #正常投稿出来なかった場合
    print("Failed. : %s"% vars(res))

確認

command
python script.py

screencapture- 2018-01-15 22.57.30.png

それっぽくツイート出来てますね！

ただ、Qiitaのタグで「.」や「-」が含まれてると、Twitterのハッシュタグがうまく動きませんが、それは今回は妥協しましょうorz

Slackへの通知

次に処理結果をSlackに通知するところを実装してみましょう。

script.py
import json, config, requests, random, sys  #必要なライブラリの読み込み 
from requests_oauthlib import OAuth1Session  #OAuthのライブラリの読み込み

CONSUMER_KEY = config.CONSUMER_KEY
CONSUMER_SECRET = config.CONSUMER_SECRET
ACCESS_TOKEN = config.ACCESS_TOKEN
ACCESS_TOKEN_SECRET = config.ACCESS_TOKEN_SECRET
twitter = OAuth1Session(CONSUMER_KEY, CONSUMER_SECRET, ACCESS_TOKEN, ACCESS_TOKEN_SECRET) #認証処理

INCOMING_WEBHOOK_URL = config.INCOMING_WEBHOOK_URL

TWITTER_API_HOST = "https://api.twitter.com/1.1"  #TwitterAPIホスト
STATUSES_UPDATE_ENDPOINT = "/statuses/update.json"  #ツイートエンドポイント
TWITTER_USER_NAME = "Bakira_Tech_Bot"


QIITA_USER_NAME = "bakira"  #取得対象のQiitaアカウント名
QIITA_API_HOST = "https://qiita.com/api/v2"  #QiitaAPIホスト
USER_ENDPOINT = f"/users/{QIITA_USER_NAME}"  #ユーザー情報取得エンドポイント
USER_ITEMS_ENDPOINT = f"/users/{QIITA_USER_NAME}/items"  #ユーザー記事一覧取得エンドポイント
PAGE_PER_COUNT = 20  #1ページ辺りの記事取得件数

response = requests.get(
    QIITA_API_HOST + USER_ENDPOINT
).json()  #対象Qiitaアカウント情報をJsonで取得

totalItemCount = response['items_count']  #投稿総数を取得
pageCount = totalItemCount // PAGE_PER_COUNT  #1ページ辺りの記事取得件数で取得出来たページ数を取得
hasSurplus = totalItemCount % PAGE_PER_COUNT != 0  #余りの記事数を取得
totalPageCount = pageCount + 1 if hasSurplus else pageCount #余りの記事数があった場合、端数分のページ数を追加

params = {
    "page": random.randint(1,totalPageCount),  #ページ番号をランダム生成
    "per_page": PAGE_PER_COUNT
}  #記事一覧をページ番号と１ページ辺りの表示件数を指定するためのパラメータを生成

response = requests.get(
    QIITA_API_HOST + USER_ITEMS_ENDPOINT,
    params=params
).json()  #対象Qiitaアカウントの記事一覧を取得

if( response == [] ):  #記事一覧が空だった場合
    requests.post(
        INCOMING_WEBHOOK_URL, 
        data = json.dumps(
        {
           "attachments":[
              {
                 "fallback":"Qiita Article is Empty.",
                 "pretext":"Bot Result.",
                 "color":"#FF0000",
                 "fields":[
                    {
                       "title":"Qiita Item is Empty",
                       "value":"There was no Article"
                    }
                 ]
              }
           ]
        } 
    ))
    sys.exit()  #処理終了


item = random.choice(response)  #取得した一覧からランダムで記事を取得
url = item["url"]  #記事URLを取得
title = item["title"]  #記事タイトルを取得
tags = map(
    lambda element: ("#" + element["name"]), 
    item["tags"] 
)  #タグ一覧をTwitterハッシュ化して配列として再生成
tagsStr = ' '.join(list(tags))  #タグ一覧を文字列結合

params = {
    "status" : "【Qiita】" +title + "\n" + tagsStr + "\n" + url  #ツイート内容を生成
}  #Postするパラメータを生成

res = twitter.post( 
    TWITTER_API_HOST + STATUSES_UPDATE_ENDPOINT ,
    params = params
)  #APIにPost通信
if res.status_code == 200: #正常投稿出来た場合
    resDic = json.loads(res.text)
    id = resDic["id"]
    requests.post(
        INCOMING_WEBHOOK_URL, 
        data = json.dumps(
        {
           "attachments":[
              {
                 "fallback":"Tweet Success.",
                 "pretext":"Bot Result.",
                 "color":"#00FF00",
                 "fields":[
                    {
                       "title":"Tweet Success.",
                       "value":f"Tweet URL : https://twitter.com/{TWITTER_USER_NAME}/status/{id}"
                    }
                 ]
              }
           ]
        } 
    ))
else: #正常投稿出来なかった場合
    requests.post(
        INCOMING_WEBHOOK_URL, 
        data = json.dumps(
        {
           "attachments":[
              {
                 "fallback":"Tweet Failed.",
                 "pretext":"Bot Result.",
                 "color":"#FF0000",
                 "fields":[
                    {
                       "title":"Tweet Failed.",
                       "value":f"Error.[{res.status_code}]"
                    }
                 ]
              }
           ]
        } 
    ))

screencapture- 2018-01-15 23.35.53.png

細かい部分は今後の課題として、一旦しっかりSlackへの通知が出来ていそうです。

AWS-lambdaへ登録

Scriptをlambda向けに修正

先ほど作ったスクリプトはコマンドから実行可能ですが、lambda上では動かないので少し手を加えます。

修正点としては、

  • ファイル名をlambda_function.pyに変更
  • 全体をlambda_handler(event, context)関数にする
  • 設定値を環境変数化
    • ※環境変数の指定の仕方は後述します。

修正後のファイルは以下となります。

lambda_function.py
import json, os, requests, random, sys  #必要なライブラリの読み込み 
from requests_oauthlib import OAuth1Session  #OAuthのライブラリの読み込み

def lambda_handler(event, context):
    CONSUMER_KEY = os.environ.get('CONSUMER_KEY')
    CONSUMER_SECRET = os.environ.get('CONSUMER_SECRET')
    ACCESS_TOKEN = os.environ.get('ACCESS_TOKEN')
    ACCESS_TOKEN_SECRET = os.environ.get('ACCESS_TOKEN_SECRET')
    twitter = OAuth1Session(CONSUMER_KEY, CONSUMER_SECRET, ACCESS_TOKEN, ACCESS_TOKEN_SECRET) #認証処理

    INCOMING_WEBHOOK_URL = os.environ.get('INCOMING_WEBHOOK_URL')

    TWITTER_API_HOST = "https://api.twitter.com/1.1"  #TwitterAPIホスト
    STATUSES_UPDATE_ENDPOINT = "/statuses/update.json"  #ツイートエンドポイント
    TWITTER_USER_NAME = os.environ.get('TWITTER_USER_NAME')


    QIITA_USER_NAME = os.environ.get('QIITA_USER_NAME')  #取得対象のQiitaアカウント名
    QIITA_API_HOST = "https://qiita.com/api/v2"  #QiitaAPIホスト
    USER_ENDPOINT = f"/users/{QIITA_USER_NAME}"  #ユーザー情報取得エンドポイント
    USER_ITEMS_ENDPOINT = f"/users/{QIITA_USER_NAME}/items"  #ユーザー記事一覧取得エンドポイント
    PAGE_PER_COUNT = 20  #1ページ辺りの記事取得件数

    response = requests.get(
        QIITA_API_HOST + USER_ENDPOINT
    ).json()  #対象Qiitaアカウント情報をJsonで取得

    totalItemCount = response['items_count']  #投稿総数を取得
    pageCount = totalItemCount // PAGE_PER_COUNT  #1ページ辺りの記事取得件数で取得出来たページ数を取得
    hasSurplus = totalItemCount % PAGE_PER_COUNT != 0  #余りの記事数を取得
    totalPageCount = pageCount + 1 if hasSurplus else pageCount #余りの記事数があった場合、端数分のページ数を追加

    params = {
        "page": random.randint(1,totalPageCount),  #ページ番号をランダム生成
        "per_page": PAGE_PER_COUNT
    }  #記事一覧をページ番号と１ページ辺りの表示件数を指定するためのパラメータを生成

    response = requests.get(
        QIITA_API_HOST + USER_ITEMS_ENDPOINT,
        params=params
    ).json()  #対象Qiitaアカウントの記事一覧を取得

    if( response == [] ):  #記事一覧が空だった場合
        requests.post(
            INCOMING_WEBHOOK_URL, 
            data = json.dumps(
            {
               "attachments":[
                  {
                     "fallback":"Qiita Article is Empty.",
                     "pretext":"Bot Result.",
                     "color":"#FF0000",
                     "fields":[
                        {
                           "title":"Qiita Item is Empty",
                           "value":"There was no Article"
                        }
                     ]
                  }
               ]
            } 
        ))
        sys.exit()  #処理終了


    item = random.choice(response)  #取得した一覧からランダムで記事を取得
    url = item["url"]  #記事URLを取得
    title = item["title"]  #記事タイトルを取得
    tags = map(
        lambda element: ("#" + element["name"]), 
        item["tags"] 
    )  #タグ一覧をTwitterハッシュ化して配列として再生成
    tagsStr = ' '.join(list(tags))  #タグ一覧を文字列結合

    params = {
        "status" : "【Qiita】" +title + "\n" + tagsStr + "\n" + url  #ツイート内容を生成
    }  #Postするパラメータを生成

    res = twitter.post( 
        TWITTER_API_HOST + STATUSES_UPDATE_ENDPOINT ,
        params = params
    )  #APIにPost通信
    if res.status_code == 200: #正常投稿出来た場合
        resDic = json.loads(res.text)
        id = resDic["id"]
        requests.post(
            INCOMING_WEBHOOK_URL, 
            data = json.dumps(
            {
               "attachments":[
                  {
                     "fallback":"Tweet Success.",
                     "pretext":"Bot Result.",
                     "color":"#00FF00",
                     "fields":[
                        {
                           "title":"Tweet Success.",
                           "value":f"Tweet URL : https://twitter.com/{TWITTER_USER_NAME}/status/{id}"
                        }
                     ]
                  }
               ]
            } 
        ))
    else: #正常投稿出来なかった場合
        requests.post(
            INCOMING_WEBHOOK_URL, 
            data = json.dumps(
            {
               "attachments":[
                  {
                     "fallback":"Tweet Failed.",
                     "pretext":"Bot Result.",
                     "color":"#FF0000",
                     "fields":[
                        {
                           "title":"Tweet Failed.",
                           "value":f"Error.[{res.status_code}]"
                        }
                     ]
                  }
               ]
            } 
        ))

ライブラリを同階層にインストール

lambdaではpipコマンドでモジュールをインストールすることが出来ないので、サードパーティ製のモジュールが必要な場合はzip化してアップロードする必要があります。

今回は

  • requests
  • requests_oauthlib

をzipに含める必要があります。

作業ディレクトリで以下のコマンドを実行してください。

command
pip install requests -t .
pip install requests_oauthlib -t .

そうすると以下のように展開されます。

tree
.
├── certifi
├── certifi-2017.11.5.dist-info
├── chardet
├── chardet-3.0.4.dist-info
├── idna
├── idna-2.6.dist-info
├── oauthlib
├── oauthlib-2.0.6.dist-info
├── requests
├── requests-2.18.4.dist-info
├── requests_oauthlib
├── requests_oauthlib-0.8.0.dist-info
├── lambda_function.py
├── urllib3
└── urllib3-1.22.dist-info

zip化

次にlambdaにアップロードするためにzip化しましょう。
作業ディレクトリで以下を実行してください。

command
zip -r upload.zip *

そうすると、upload.zipというファイルが出来ていると思います。
これをアップロードします。

AWS Lambdaの設定

Lambda関数の作成

次にAWSにログインし、コンピューティングメニューのLambdaを選択します。

screencapture- 2018-01-15 23.53.05.png

関数の作成ボタンをクリックします。

screencapture- 2018-01-15 23.53.17.png

一から作成を選択し、以下を設定の上、

  • 名前　：　任意の名称
  • ランタイム　：　Python 3.6
  • ロール　：　テンプレートから新しいロールを作成
  • ロール名　：　任意の名称
  • ポリシーテンプレート　：　プルダウンからBasic Edge Lambda アクセス権限を選択

関数の作成ボタンをクリックしてください。

screencapture- 2018-01-16 0.27.19.png

そうすると、Lambda関数が作成されるので詳細を設定していきます。

screencapture- 2018-01-16 0.27.53.png

関数コード

コードエントリータイプを.ZIPファイルをアップロード、ランタイムをPython 3.6、ハンドラをlambda_function.lambda_handler、先ほど作成したzipファイルを選択します。

screencapture- 2018-01-16 0.29.56.png

環境変数

Lambda関数単位で環境変数を指定することが出来ます。
今回は先ほどconfig.pyに記載していたものを環境変数として設定してみましょう。

screencapture- 2018-01-16 0.32.35.png

テスト

設定が出来たらテスト実行してみましょう。

screencapture- 2018-01-16 0.45.43.png

このような表示がされたら無事にスクリプトが動いたという事になります。

screencapture- 2018-01-16 0.45.50.png

Slackにも通知が来ていますね。

Lambda関数を使うメリット

サーバーを用意する必要がないのでサーバーレス構成を実現することが可能です。
Lambdaは他にもAWSサービス群への操作をトリガーに処理を実行することが出来るので非常に便利です！！
また、ログファイルについてもデフォルトでCloudWatch Logsに吐き出してくれるのでとても助かりますね。

screencapture- 2018-01-16 17.53.29.png

AWS CloudWatchの設定

スケジューリング登録

次はこのスクリプトを定期的に実行するように仕込みます。

今回は毎日00:00に投稿するようにスケジューリングしてみましょう。
実現するにはCloudWatch Eventsを利用します。

管理ツールメニューのCloudWatchを選択します。

screencapture- 2018-01-16 0.50.59.png

イベントからルールの作成ボタンをクリックします。

screencapture- 2018-01-16 0.51.22.png

イベントソースのスケジュールを選択し、Cron式に0 0 * * ? 0を入力します。
ターゲットはLambda関数にし、機能は先ほど作ったLambda関数を選択します。

そして、設定の詳細ボタンをクリックします。

screencapture- 2018-01-16 8.33.45.png

名前を任意の名称にしてルールの作成ボタンをクリックします。

screencapture- 2018-01-16 0.52.20.png

これでルールが作成されたのでスケジューリングは完了です。

screencapture- 2018-01-16 0.52.27.png

JSTとUTCの違い

ここですっかり次の日の00:00にLambda関数が実行されると思って油断していたら、、、

screencapture- 2018-01-16 11.21.10.png

むむ。。。

9:00にLambda関数が実行されていました。

CloudWatchで指定するCronはUTC(GMT)で実行されるので、JSTに置き換えた時間指定をする必要があります。

厳密には9時間のズレがあるので-9時間をすれば良いのです。
今回の場合00:00に実行したいので、0 15 * * ? 0と指定します。

screencapture- 2018-01-16 11.36.56.png

これで正常に毎日00:00にLambda関数が実行されるようになりました♪
※処理のラグで多少通知時間がズレてますが良しとしましょう！！笑

screencapture- 2018-01-17 0.09.59.png

参考にさせていただいた記事

5
6
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
5
6