【WebAPI】YouTubeチャンネルを監視してみた

# 【補足】<> 内は適宜その説明どおりの情報を設定してください。

import json

# パラメータを定義する。
dict_channel_id = {'channel_id': '<アクセスするチャンネルのID>'}
dict_mail = {
                'smtp_host': '<SMTPホスト名>',
                'smtp_port': 587,
                'from_user': '<送信元ユーザ名>',
                'from_addr': '<送信元アドレス（適当でOK）>',
                'to_addr': '<送信先アドレス>'
            }
dict_excel = {'file_name': '<Excelファイル名>'}

# 定義したパラメータをJSONファイルへ書き込む。
with open('<YouTubeチェンネルIDを保存するJSONファイルのフルパス>', 'w') as f:
    json.dump(dict_channel_id, f, indent=4)

with open('<メール送信情報を保存するJSONファイルのフルパス>', 'w') as f:
    json.dump(dict_mail, f, indent=4)

with open('<Excel情報を保存するJSONファイルのフルパス>', 'w') as f:
    json.dump(dict_excel, f, indent=4)

import os

from googleapiclient.discovery import build

FNAME_PARAM_YOUTUBE_ID = "param_youtube_id.json"


class YouTubeWatcher(object):

    def __init__(self):
        self.developer_key = os.getenv("YOUTUBE_DEVELOPER_KEY")
        self.api_service_name = "youtube"
        self.api_version = "v3"

        # パラメータJSONファイルは当プログラムと同じ場所にあるものとする。
        with open(os.path.join(os.path.dirname(__file__), FNAME_PARAM_YOUTUBE_ID)) as f:
            param = json.load(f)
        self.param = param
        self.channel_id = param["channel_id"]
        self.info = {}

    def watch(self):
        try:
            youtube = build(self.api_service_name,
                            self.api_version,
                            developerKey=self.developer_key
                            )

            response = youtube.channels().list(part="snippet, statistics",
                                                id=self.channel_id).execute()

            self.info["title"] = response["items"][0]["snippet"]["localized"]["title"]
            self.info["id"] = self.channel_id
            self.info["videoCount"] = response["items"][0]["statistics"]["videoCount"]
            self.info["subscriberCount"] = response["items"][0]["statistics"]["subscriberCount"]
            self.info["viewCount"] = response["items"][0]["statistics"]["viewCount"]

        except Exception as e:
            print(f'YouTubeWatcher.watch() Failed. :{e}')
            self.info = None

        return self.info

import os
from datetime import datetime

import smtplib
from email.mime.text import MIMEText
from email.header import Header
from email.utils import formatdate

FNAME_PARAM_MAIL = "param_youtube_mail.json"


class Mailer(object):

    def __init__(self):
        self.from_pw = os.getenv("MAIL_FROM_PW")

        # パラメータJSONファイルは当プログラムと同じ場所にあるものとする。
        with open(os.path.join(os.path.dirname(__file__), FNAME_PARAM_MAIL)) as f:
            param = json.load(f)
        self.param = param
        self.smtp_obj = None
        self.smtp_host = param["smtp_host"]
        self.smtp_port = int(param["smtp_port"])
        self.from_user = param["from_user"]
        self.from_addr = param["from_addr"]
        self.to_addr = param["to_addr"]

    def send(self, info):

        # 本文を作成する。
        message = ''
        for k, v in info.items():
            message += str(k) + ': ' + str(v) + '\r\n'
        message += '\r\n\r\n' + formatdate()

        # 件名を作成する。
        subject = "youtube report " + datetime.now().strftime('(%Y.%m.%d %H:%M:%S)')

        # 送信情報を作成する。
        mime = MIMEText(message)
        mime['From'] = self.from_addr
        mime['To'] = self.to_addr
        mime['Subject'] = Header(subject, 'utf-8')

        # 作成したメールを送信する。
        try:
            self.smtp_obj = smtplib.SMTP(self.smtp_host, self.smtp_port)

            # EHLO(Extend Hello)を実行してSMTPサーバへ明示的に身元を明かす。
            self.smtp_obj.ehlo()
            # TLS(Transport Layer Security)に対応している場合は、SMTPコマンドを暗号化する。
            if self.smtp_obj.has_extn('STARTTLS'):
                self.smtp_obj.starttls()

            self.smtp_obj.login(self.from_user, self.from_pw)
            self.smtp_obj.send_message(mime)

        except Exception as e:
            print(f'Mailer.send() Failed. :{e}')

        finally:
            self.smtp_obj.quit()

import os
from datetime import datetime

import openpyxl as excel

FNAME_PARAM_EXCEL = "param_youtube_excel.json"


class Exceler(object):

    def __init__(self):
        # Sheetの1列目を項目名、2列目を項目値とする。
        self.column_item = 1
        self.column_val = 2

        # パラメータJSONファイルは当プログラムと同じ場所にあるものとする。
        with open(os.path.join(os.path.dirname(__file__), FNAME_PARAM_EXCEL)) as f:
            param = json.load(f)
        self.param = param
        self.path_output = os.path.join(os.path.dirname(__file__), param["file_name"])
        self.book = None

    def save(self, info):

        try:
            # YouTubeチャンネル情報を出力するシート名は、出力時の"日付_時刻"とする。
            sheetname = datetime.now().strftime('%Y%m%d_%H%M%S')

            if os.path.exists(self.path_output):
                self.book = excel.load_workbook(self.path_output)
            else:
                self.book = excel.Workbook()

            # シートをブックの先頭へ追加する。
            new_sheet = self.book.create_sheet(title=sheetname, index=0)

            # チャンネルタイトルをシートへ書き込む。
            new_sheet.cell(row=1, column=self.column_item, value="Title")
            new_sheet.cell(row=1, column=self.column_val, value=info["title"])

            # チャンネルIDをシートへ書き込む。
            new_sheet.cell(row=2, column=self.column_item, value="ID")
            new_sheet.cell(row=2, column=self.column_val, value=info["id"])

            # 動画数をシートへ書き込む。
            new_sheet.cell(row=3, column=self.column_item, value="VideoCount")
            new_sheet.cell(row=3, column=self.column_val, value=info["videoCount"])

            # 登録者数をシートへ書き込む。
            new_sheet.cell(row=4, column=self.column_item, value="SubscriberCount")
            new_sheet.cell(row=4, column=self.column_val, value=info["subscriberCount"])

            # 視聴回数をシートへ書き込む。
            new_sheet.cell(row=5, column=self.column_item, value="ViewCount")
            new_sheet.cell(row=5, column=self.column_val, value=info["viewCount"])

            self.book.save(self.path_output)

        except Exception as e:
            print(f'Exceler.save() Failed. :{e}')

        finally:
            self.book.close()

from youtubewatcher import YouTubeWatcher
from mailer import Mailer
from exceler import Exceler


def main():

    # 指定チャンネルの情報を取得する。
    watcher = YouTubeWatcher()
    info = watcher.watch()

    # 取得した情報をメールで送信する。
    mailer = Mailer()
    mailer.send(info)

    # 取得した情報をExcelファイルへ保存する。
    exceler = Exceler()
    exceler.save(info)


if __name__ == "__main__":
    main()