Lambda+DynamoDB+CloudWatchでQiitaのいいね数監視を自動化してみる

この記事は リンク情報システム の「2020新春アドベントカレンダー TechConnect!」のリレー記事です。
TechConnect! は勝手に始めるアドベントカレンダーとして、engineer.hanzomon という勝手に作ったグループによってリレーされます。
（リンク情報システムのFacebookはこちらから）

本記事は7日目、1/15(水)分です。

---

過去記事でもやってますが弊社のアドカレのいいね数とかをなんとなーく集計する係みたいになってました。(その前のアドカレは記事にしてませんがshell芸で収集したりしてます)
んで、どうせなら自動化していいね数が一定超えたらQiitaのマイルストーンみたいに通知したら楽できるいいんじゃないかなと思いまして、AWS Lambdaで実行させてみることにしました。

実装としては過去記事同様アドカレトップページをスクレイピングして記事URLを収集、Qiita APIでいいね数を取得していく感じです。
まずは記事IDを収集するLambda関数から

コード

import os
import requests
import boto3
from selenium import webdriver
from bs4 import BeautifulSoup
from urllib.parse import urljoin

def lambda_handler(event, context):
    api_endpoint = 'https://qiita.com/api/v2/'

    try:
        dynamoDB = boto3.resource("dynamodb")
        advent_calendar = dynamoDB.Table("advent_calendar")

        options = webdriver.ChromeOptions()
        options.binary_location = "/opt/bin/headless-chromium"

        options.add_argument("--headless")
        options.add_argument("--disable-gpu")
        options.add_argument("--window-size=1280x1696")
        options.add_argument("--disable-application-cache")
        options.add_argument("--disable-infobars")
        options.add_argument("--no-sandbox")
        options.add_argument("--hide-scrollbars")
        options.add_argument("--enable-logging")
        options.add_argument("--log-level=0")
        options.add_argument("--single-process")
        options.add_argument("--ignore-certificate-errors")
        options.add_argument("--homedir=/tmp")

        driver = webdriver.Chrome(executable_path="/opt/bin/chromedriver", options=options)
        driver.get(os.environ['TARGET_URL'])
        soup = BeautifulSoup(driver.page_source, 'html.parser')

        item = soup.find('div', id='personal-public-article-body')
        tables = item.find_all('tbody')
        for table in tables:
            rows = table.find_all('tr')
            for row in rows:
                user_id = row.find_all('td')[1].text
                tmp = row.find_all('td')[2].find('a')['href']
                item_id = tmp[tmp.find('items/'):]
                response = advent_calendar.get_item(
                    Key={
                        'user_id': user_id,
                        'item_id': item_id
                    }
                )
                if 'Item' not in response:
                    advent_calendar.put_item(
                        Item = {
                            "user_id": user_id,
                            "item_id": item_id,
                            'likes': 0
                        }
                    )
    except Exception as e:
        print(e)
    finally:
        driver.quit()
    return

必要なライブラリなりchromedriverなりはあらかじめLambda Layersに登録しておきます。 取得した記事のIDはDynamoDBに保管しておきます。いいね数もここで初期化。 このLambda関数をCloudWatch Eventで毎時実行します。

んで、この収集した記事IDに対しQiita API発行していいね数を取得していく関数が以下

コード

import os
import boto3
import requests
from urllib.parse import urljoin
import smtplib
from email.message import EmailMessage

def lambda_handler(event, context):
    api_endpoint = 'https://qiita.com/api/v2/'
    headers = {'Authorization': 'Bearer ' + os.environ['QIITA_AUTH']}

    dynamoDB = boto3.resource("dynamodb")
    advent_calendar = dynamoDB.Table("advent_calendar")

    try:
        smtp = smtplib.SMTP_SSL(os.environ['SMTP_HOST'], int(os.environ['SMTP_PORT']))
        smtp_user = os.environ['SMTP_USER']
        smtp_pass = os.environ['SMTP_PASS']
        message = EmailMessage()
        message['From'] = os.environ['FROM_ADDRESS']
        message['To'] = os.environ['TO_ADDRESS']
        message['Subject'] = 'アドカレいいね監視'
        smtp.login(smtp_user, smtp_pass)

        response = advent_calendar.scan()
        for i in response['Items']:
            user_id = i['user_id']
            item_id = i['item_id']
            old_likes = int(i['likes'])
            item_url = urljoin(api_endpoint, item_id)
            item_detail = requests.get(item_url, headers=headers).json()
            
            title = item_detail['title']
            url = item_detail['url']
            new_likes = int(item_detail['likes_count'])
            comments = int(item_detail['comments_count'])
            stockers_url = urljoin(api_endpoint, item_id + '/stockers?per_page=100')
            stockers = len(requests.get(stockers_url, headers=headers).json())
            
            if old_likes < 100 and new_likes >= 100:
                message.set_content(user_id+"さんの記事「"+title+"("+url+")」が100いいねを超えました")
                smtp.send_message(message)
            elif old_likes < 50 and new_likes >= 50:
                message.set_content(user_id+"さんの記事「"+title+"("+url+")」が50いいねを超えました")
                smtp.send_message(message)
            elif old_likes < 30 and new_likes >= 30:
                message.set_content(user_id+"さんの記事「"+title+"("+url+")」が30いいねを超えました")
                smtp.send_message(message)
            elif old_likes < 10 and new_likes >= 10:
                message.set_content(user_id+"さんの記事「"+title+"("+url+")」が10いいねを超えました")
                smtp.send_message(message)
            
            advent_calendar.put_item(
                Item = {
                    "user_id": user_id,
                    "item_id": item_id,
                    "likes" : new_likes,
                    "comments" : comments,
                    "stockers" : stockers
                }
            )
    except Exception as e:
        print(e)
    finally:
        smtp.close()
    return

DynamoDBをscanして記事IDを取得、それに対しQiita APIを発行し前回取得したいいね数と比較、判定部分が非常に雑ですが閾値超えてたらメール送信します。~~100以降は通知ないですがどうせそんなにいいねつかないでしょう~~ こいつはCloudWatch Eventで毎分実行。

本当は弊社の社内チャットとして使用しているMicrosoft Teamsのほうに通知投げたかったんですが、Office365の認証でうちの二段階認証に引っかかって実現できず…現状自分のメールアドレス宛に飛んでるだけです。
Outlookのほうで自動転送してやろうかとも思いましたがそっちはそっちで権限不足で転送できず。なんだかなー。

---

若干片手落ち感がありありですが一応いいね数収集の自動化はできました。カレンダー終わってひと段落ついたらDynamoDBのデータ引っ張って最終結果出すかな。
最初はZABBIXのHTTP Agent使ってやろうと思っていましたが、EC2の無料枠が無くなったのでLambda+DynamoDBにしました。無料枠最高。

明日は@h-yamasakiさんです。

---

1/17 ついでにコメント数とストック数も収集するよう修正、API発行回数が増えたので1時間1000回の上限引っ掛かりそうだったのでCloudWatch Eventの監視間隔を1分→5分に変更