iTranslated by AI
Building a LINE Video Echo Bot using YouTube
Introduction
This article is the same as the one I posted on Qiita.
Hello. I'm Maguro from STECH.
This is an article for the STECH and Aichi Institute of Technology Systems Engineering Research Group Joint Advent Calendar.
In this article, I will try uploading a video to YouTube using the LINE API.
Why I Built It
When I previously wrote an article about linking Discord and LINE, images and videos had to be sent in HTTP format.
Discord uses Cloudflare for file storage, so files can be referenced via HTTP.
However, on the LINE side, files are saved as binary data and are reportedly deleted after a certain period.
Setting images aside, I thought sharing videos would be difficult due to the large capacity and load involved.
Nevertheless, I kept looking and discovered the YouTube Data API.
I found that it is possible to upload videos using Python.
By incorporating this into a LINE Bot and returning a URL when it's uploaded, sharing becomes possible.
This time, I will apply that to create a Bot that returns a YouTube video link when a video is uploaded to LINE.
In short, it will be a video version of a LINE echo bot.
Environment
- Python 3.8.6
- flask
- youtube-data-api
- google-api-python-client
- ngrok
Preparation
Create credentials from Google Cloud Platform (GCP).
Please proceed with registration by referring to the following article.
About Registration
Since the application method seems to have changed slightly from the article above, I will write about the registration method here.
If you have already done this, please skip this section.
Creating a New Project
Access GCP and click "Select a project".
Click "New Project".
Set the Project name (any name you like) and Organization (leave as default).
Once configured, click "Create".
Once created, you will be redirected to a screen like this.
Enabling the YouTube API
Select "Enabled APIs & services" from "APIs & Services".
You will be redirected to the search screen, so search for "YouTube data api v3".
Click on the "YouTube Data API v3" that appears.
Click "Enable".
You will then be prompted to register credentials, so click "Create Credentials".
Creating Credentials and Consent Screen
Click "Create Credentials" at the top.
Select "OAuth client ID".
You will be prompted to create a consent screen, so click create.
Select "External" for User Type and click "Create". Enter the information for the consent screen.
Fill in the App name, User support email, and Developer contact information fields.
The logo is optional.
Once filled in, click "Save and Continue".
Now, create an OAuth client ID.
Set the Application type to "Desktop app".
The name can be anything you like.
When you select "Create" here...
The OAuth client will be created.
Download the JSON and try uploading a video following the article mentioned earlier.
If the upload completes successfully and upload_video.py-oauth2.json is generated, the preparation is complete.
Design
The overall flow is shown in the diagram below.
Host the bot by starting a server from your PC using ngrok.
When a video is sent to LINE, the bot saves the video.
Once the saving is finished, it starts uploading to YouTube.
After the upload is complete, an ID is output, which is then used to generate the video URL.
The sequence of events concludes by replying with that URL.
Implementation
Directory Structure
$ tree
.
├── movies
│ └── sample.mp4 # Video to be uploaded
├── client_secrets.json
├── upload_video.py-oauth2.json
├── main.py
├── line_api.py # lineapi class
└── upload_video.py # For YouTube upload
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():
# Parse the incoming request
requests = request.get_json()
# If the events content is empty (connection verification)
if len(requests['events']) == 0:
return abort(Response("ok"))
# Parse the contents of events
data = requests['events'][0]
if data['message']['type'] == 'video':
# Retrieve video data by class using 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
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
# Get name from LINE user profile
async def get_proflie(self, user_id: str):
# Determine if Group ID is valid
try:
return await linereq(
f"https://api.line.me/v2/bot/group/{self.line_group_id}/member/{user_id}",
self.line_bot_token,
"displayName"
)
# If Group ID is invalid, determine from friends
except KeyError:
return await linereq(
f"https://api.line.me/v2/bot/profile/{user_id}",
self.line_bot_token,
"displayName"
)
# Save the video received from LINE and upload it to YouTube as unlisted
async def movie_upload(self, message_id: int, display_name: str):
# Get video binary data
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()
# Save as mp4
with open("./movies/sample.mp4", 'wb') as fd:
for chunk in movies_bytes:
fd.write(chunk)
# Execute upload_video.py using subprocess; the video will be uploaded to YouTube as unlisted
youtube_id = subprocess.run(
['python', 'upload_video.py','--file=./movies/sample.mp4', f'--title="{display_name}\'s Video"', '--description="Video from LINE"','--privacyStatus=unlisted'],
capture_output=True)
# Use the outputted ID to create a YouTube unlisted link
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)
# Send a GET request and return the specified data column.
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 has moved to http.client in Python 3
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" # Rewrite to the generated JSON filename
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 statement
status, response = insert_request.next_chunk()
if response is not None:
if 'id' in response:
print(response['id']) # Video ID; use this to generate the video 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__':
# Set the video to upload, title, visibility, etc., from command line arguments (mostly defaults are defined)
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))
Starting ngrok
Start it on port 8080.
ngrok http 8080
Once started, don't forget to set the displayed https://*********.ngrok.io as the Webhook URL for your LINE Bot.
Next, start the LINE Bot.
python main.py
If the connection test is successful, you're all set.
Sending
Let's try sending it.
It's complete if a URL is returned and you can watch the video.
It may take some time to process, so it might not be viewable immediately.
Conclusion
Great job.
Thank you for reading this far.
Regarding the Discord integration mentioned earlier, I'm currently rebuilding it to support Notify.
I'm currently busy, so the completion date is undecided, but I'll write an article once it's finished.
See you in the next article.
Discussion