💫

YouTubeを利用してLINEの動画オウム返しBotを作る

2022/12/17に公開

ご挨拶

本記事はQiitaに投稿したものと同様のものになります。
どうもこんにちは。STECH所属のマグロです。

STECHと愛知工業大学 システム工学研究会の共同アドベントカレンダーの記事となります。

今回はLINE APIを使用してYouTubeに動画をアップロードしてみます。

作った経緯

以前記事にしたDiscordとLINEを連携させる際、画像、動画はhttpの形式で送信する必要がありました。
Discordはファイルの保存にCloudflareを採用しており、httpでファイルの参照が可能でした。

しかしLINE側はバイナリデータで保存され、加えて一定時間で削除されてしまうそうです。
画像はともかく、動画は容量や負荷が大きく、共有は厳しそうと思っていました。

それでもめげずに調べてみると、YouTube Data APIというものを発見。
Pythonで動画のアップロードが可能ということがわかりました。

コイツをLINEBotに組み込み、アップロードした際にurlを返すことで共有が可能になるのです。
今回はそれを応用して、LINEに動画を挙げるとYouTubeの動画リンクを返すBotを作成します。

要するに動画版LINEのオウム返しBotになります。

環境

  • Python 3.8.6
  • flask
  • youtube-data-api
  • google-api-python-client
  • ngrok

下準備

GCPから認証情報を作成します。
以下の記事を参考に登録を進めてください。

https://qiita.com/ny7760/items/5a728fd9e7b40588237c

登録について

上記の記事と若干申請方法が変わったようなので、登録方法について書いておきます。
できている人はスキップしてください。

プロジェクトを新規作成

GCPにアクセスし、プロジェクトの選択をクリック。

https://console.cloud.google.com/

image.png

新しいプロジェクトをクリック。

image.png

プロジェクト名(わかれば何でもいい)、組織(標準のままで)を設定。
設定したら作成をクリック。

image.png

作成するとこんな感じの画面に遷移します。

スクリーンショット 2022-11-01 105309.png

YouTubeAPIの有効化

APIとサービスから有効なAPIとサービスを選択。
image.png

検索画面に遷移するので「YouTube data api v3」で検索。

スクリーンショット 2022-11-01 105345.png

出てきたYouTube Data API v3をクリック。

スクリーンショット 2022-11-01 105358.png

有効にするをクリック。

スクリーンショット 2022-11-01 105415.png

すると認証情報の登録を求められるので認証情報を作成をクリックする。

image.png

認証情報、同意画面の作成

上の認証情報を作成をクリック。
image.png

OAuthクライアントIDを選択。
image.png

同意画面の作成を求められるので作成します。

スクリーンショット 2022-11-01 105753.png

User Typeを外部に選択して作成をクリック。同意画面の情報を入力します。

スクリーンショット 2022-11-01 105817.png

アプリ名、サポート、デベロッパーメールアドレスの欄を入力。
ロゴに関しては任意。
入力したら保存して次へ。

image.png
image.png

OAuthクライアントIDを作成します。
アプリケーションの種類はデスクトップアプリにしましょう。
名前は何でもいいです。

スクリーンショット 2022-11-01 110032.png
スクリーンショット 2022-11-01 110054.png

これで作成を選択すると
スクリーンショット 2022-11-01 110124.png
OAuthクライアントが作成されます。
jsonをダウンロードして、上記の記事を参考に試しに動画をアップロードしましょう。

正常にアップロードされ、upload_video.py-oauth2.jsonが作成されていれば準備完了です。

設計

大雑把な流れとして以下の図のようになります。
image.png

ngrokでPCからサーバを立ち上げ、botをホストします。

lineへ動画を送ると、botが動画を保存します。
保存を終え次第、youtubeへのアップロードを開始いたします。
アップロードが完了すると、URLが出力されるのでそれを組み合わせて動画のURLを生成します。

それを返信することが一連の流れとなります。

作成

ディレクトリ構成

$ tree
.
├── movies
│   └── sample.mp4  # アップロード対象の動画
├── client_secrets.json
├── upload_video.py-oauth2.json
├── main.py  
├── line_api.py  # lineapiのclass
└── upload_video.py  # YouTubeアップロード用

main.py

main.py
from flask import Flask, request, abort,Response
import json
from line_api import LineMessageAPI


app = Flask(__name__)

ACCESSTOKEN = ''
GROUPID = ''

line_bot_api = LineMessageAPI(line_bot_token=ACCESSTOKEN,line_group_id=GROUPID)

@app.route("/", methods=['POST'])
async def callback():
    # 送られてきたリクエストを展開
    requests = request.get_json()
    
    # eventsの中身が空(応答確認)の場合
    if len(requests['events']) == 0:
        return abort(Response("ok"))

    # eventsの中身を展開
    data = requests['events'][0]
    
    if data['message']['type'] == 'video':
        # message_idから動画のデータをクラスごと取得
        message_content = await line_bot_api.movie_upload(message_id=data['message']['id'],display_name=await line_bot_api.get_proflie(user_id=data['source']['userId']))
        await line_bot_api.reply(reply_token=data['replyToken'],message=message_content)
        return abort(Response("ok"))


if __name__ == "__main__":
    app.run("0.0.0.0", port=8080)

line_api.py

line_api.py
import subprocess
import requests

import asyncio

from functools import partial

class LineMessageAPI:
    def __init__(self, line_bot_token: str, line_group_id: str) -> None:
        self.line_group_id = line_group_id
        self.line_bot_token = line_bot_token

    # LINEのユーザプロフィールから名前を取得
    async def get_proflie(self, user_id: str):
        # グループIDが有効かどうか判断
        try:
            return await linereq(
                f"https://api.line.me/v2/bot/group/{self.line_group_id}/member/{user_id}",
                self.line_bot_token,
                "displayName"
            )
        # グループIDが無効の場合、友達から判断
        except KeyError:
            return await linereq(
                f"https://api.line.me/v2/bot/profile/{user_id}",
                self.line_bot_token,
                "displayName"
            )

    # LINEから受け取った動画を保存し、YouTubeに限定公開でアップロード
    async def movie_upload(self, message_id: int, display_name: str):
        # 動画のバイナリデータを取得
        movies_bytes = requests.get(
            f'https://api-data.line.me/v2/bot/message/{message_id}/content',
            headers={
                'Authorization': 'Bearer ' + self.line_bot_token
            }
        ).iter_content()

        # mp4で保存
        with open("./movies/sample.mp4", 'wb') as fd:
            for chunk in movies_bytes:
                fd.write(chunk)

        # subprocessでupload_video.pyを実行、動画がYouTubeに限定公開でアップロードされる
        youtube_id = subprocess.run(
            ['python', 'upload_video.py','--file=./movies/sample.mp4', f'--title="{display_name}の動画"', '--description="LINEからの動画"','--privacyStatus=unlisted'],
            capture_output=True)

        # 出力されたidを当てはめ、YouTubeの限定公開リンクを作成
        return f"https://youtu.be/{youtube_id.stdout.decode()}"

    async def reply(self, reply_token:str,message:str):
        body = {
            'replyToken': reply_token,
            'messages': message
        }
        return await requests.post(url="https://api.line.me/v2/bot/message/reply",headers={'Authorization': 'Bearer ' + self.line_bot_token},data=body)

# getリクエストを送り、指定されたデータ列を返す。
async def linereq(url: str, token: str, jsonkey: str):
    r = requests.get(url=url,headers={'Authorization': 'Bearer ' + token})
    return r.json()[jsonkey]

upload_video.py

import http.client  # httplibはPython3はhttp.clientへ移行
import httplib2
import os
import random
import sys
import time

from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from googleapiclient.http import MediaFileUpload
from oauth2client.client import flow_from_clientsecrets
from oauth2client.file import Storage
from oauth2client.tools import argparser, run_flow


httplib2.RETRIES = 1
MAX_RETRIES = 10
RETRIABLE_EXCEPTIONS = (httplib2.HttpLib2Error,
                        IOError,
                        http.client.NotConnected,
                        http.client.IncompleteRead,
                        http.client.ImproperConnectionState,
                        http.client.CannotSendRequest,
                        http.client.CannotSendHeader,
                        http.client.ResponseNotReady,
                        http.client.BadStatusLine)
RETRIABLE_STATUS_CODES = [500, 502, 503, 504]
CLIENT_SECRETS_FILE = "client_secret.json"       # 生成されたjsonのファイル名に書き換えること
MISSING_CLIENT_SECRETS_MESSAGE = """
WARNING: Please configure OAuth 2.0

To make this sample run you will need to populate the client_secrets.json file
found at:

   %s

with information from the API Console
https://console.developers.google.com/

For more information about the client_secrets.json file format, please visit:
https://developers.google.com/api-client-library/python/guide/aaa_client_secrets
""" % os.path.abspath(os.path.join(os.path.dirname(__file__),
                                   CLIENT_SECRETS_FILE))

YOUTUBE_UPLOAD_SCOPE = "https://www.googleapis.com/auth/youtube.upload"
YOUTUBE_API_SERVICE_NAME = "youtube"
YOUTUBE_API_VERSION = "v3"


VALID_PRIVACY_STATUSES = ("public", "private", "unlisted")

#  python upload_video.py --file="./movies/sample.mp4" --title="Sample Movie" --description="This is a sample movie." --category="22" --privacyStatus="unlisted"

def get_authenticated_service(args):
    flow = flow_from_clientsecrets(CLIENT_SECRETS_FILE,
                                   scope=YOUTUBE_UPLOAD_SCOPE,
                                   message=MISSING_CLIENT_SECRETS_MESSAGE)

    storage = Storage("%s-oauth2.json" % sys.argv[0])
    credentials = storage.get()

    if credentials is None or credentials.invalid:
        credentials = run_flow(flow, storage, args)

    return build(YOUTUBE_API_SERVICE_NAME,
                 YOUTUBE_API_VERSION,
                 http=credentials.authorize(httplib2.Http()))


def initialize_upload(youtube, options):
    tags = None
    if options.keywords:
        tags = options.keywords.split(",")

    body = dict(
        snippet=dict(
            title=options.title,
            description=options.description,
            tags=tags,
            categoryId=options.category
        ),
        status=dict(
            privacyStatus=options.privacyStatus
        )
    )

    insert_request = youtube.videos().insert(
        part=",".join(body.keys()),
        body=body,
        media_body=MediaFileUpload(options.file, chunksize=-1, resumable=True)
    )

    resumable_upload(insert_request)


def resumable_upload(insert_request):
    response = None
    error = None
    retry = 0
    while response is None:
        try:
            # print("Uploading file...")  # print文
            status, response = insert_request.next_chunk()
            if response is not None:
                if 'id' in response:
                    print(response['id'])     # 動画のid これを組み合わせて動画のurlを生成する。
                else:
                    exit("The upload failed with an unexpected response: %s" % response)
        except HttpError as e:
            if e.resp.status in RETRIABLE_STATUS_CODES:
                error = "A retriable HTTP error %d occurred:\n%s" % \
                        (e.resp.status, e.content)
            else:
                raise
        except RETRIABLE_EXCEPTIONS as e:
            error = "A retriable error occurred: %s" % e
        if error is not None:
            print(error)
            retry += 1
            if retry > MAX_RETRIES:
              exit("No longer attempting to retry.")
            max_sleep = 2 ** retry
            sleep_seconds = random.random() * max_sleep
            print("Sleeping %f seconds and then retrying..." % sleep_seconds)
            time.sleep(sleep_seconds)


if __name__ == '__main__':
    # コマンドライン引数からアップロードする動画、タイトル、公開形式などを設定(大体はデフォルトで定めてる)
    argparser.add_argument("--file", help="Video file to upload",default="./movies/sample.mp4")
    argparser.add_argument("--title", help="Video title", default="Test Title")
    argparser.add_argument("--description",
                           help="Video description",
                           default="Test Description")
    argparser.add_argument("--category", default="22",
                           help="Numeric video category. " +
                                "See https://developers.google.com/youtube/v3/docs/videoCategories/list")
    argparser.add_argument("--keywords", help="Video keywords, comma separated",
                           default="")
    argparser.add_argument("--privacyStatus", choices=VALID_PRIVACY_STATUSES,
                           default=VALID_PRIVACY_STATUSES[0],
                           help="Video privacy status.")
    args = argparser.parse_args()

    if not os.path.exists(args.file):
        exit("Please specify a valid file using the --file= parameter.")

    youtube = get_authenticated_service(args)
    try:
        initialize_upload(youtube, args)
    except HttpError as e:
        print("An HTTP error %d occurred:\n%s" % (e.resp.status, e.content))

ngrok起動

8080ポートで起動します。

ngrok http 8080

起動したら、表示されるhttps://*********.ngrok.ioをLINEBotのWebHookに忘れず設定しておきましょう。
次にlinebotを起動します。

python main.py

接続テストで成功が出れば、これで完成です。

送信

送信してみましょう。
image.png
image.png

URLが返信されて、動画が見れれば完成です。
処理に時間がかかり、しばらく見れない場合もあります。

最後に

お疲れさまでした。
ここまで読んでいただきありがとうございます。

最初に話したDiscordとの連携ですが、Notifyに対応させるため作り直してます。
現在多忙で完成時期は未定ですが、完成したら記事にしようかと思います。

それではまた次回の記事でお会いいたしましょう。

Discussion