2023/04/09追記
GAS対応版書きました。
ご挨拶
どうもこんにちは。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から認証情報を作成します。
以下の記事を参考に登録を進めてください。
登録について
上記の記事と若干申請方法が変わったようなので、登録方法について書いておきます。
できている人はスキップしてください。
プロジェクトを新規作成
GCPにアクセスし、プロジェクトの選択をクリック。
新しいプロジェクトをクリック。
プロジェクト名(わかれば何でもいい)、組織(標準のままで)を設定。
設定したら作成をクリック。
作成するとこんな感じの画面に遷移します。
YouTubeAPIの有効化
APIとサービスから有効なAPIとサービスを選択。
検索画面に遷移するので「YouTube data api v3」で検索。
出てきたYouTube Data API v3をクリック。
有効にするをクリック。
すると認証情報の登録を求められるので認証情報を作成をクリックする。
認証情報、同意画面の作成
上の認証情報を作成をクリック。
OAuthクライアントIDを選択。
同意画面の作成を求められるので作成します。
User Typeを外部に選択して作成をクリック。同意画面の情報を入力します。
アプリ名、サポート、デベロッパーメールアドレスの欄を入力。
ロゴに関しては任意。
入力したら保存して次へ。
OAuthクライアントIDを作成します。
アプリケーションの種類はデスクトップアプリにしましょう。
名前は何でもいいです。
これで作成を選択すると
OAuthクライアントが作成されます。
jsonをダウンロードして、上記の記事を参考に試しに動画をアップロードしましょう。
正常にアップロードされ、upload_video.py-oauth2.jsonが作成されていれば準備完了です。
設計
大雑把な流れとして以下の図のようになります。
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
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
接続テストで成功が出れば、これで完成です。
送信
送信してみましょう。
URLが返信されて、動画が見れれば完成です。
処理に時間がかかり、しばらく見れない場合もあります。
最後に
お疲れさまでした。
ここまで読んでいただきありがとうございます。
最初に話したDiscordとの連携ですが、Notifyに対応させるため作り直してます。
現在多忙で完成時期は未定ですが、完成したら記事にしようかと思います。
それではまた次回の記事でお会いいたしましょう。