挨拶
こんにちは。マグロです。
こちらの続きとなります。
今回はLINEから画像をDiscordに送ります。
背景
こちらの記事の通り、LINEはアップロードされたファイルはバイナリデータでサーバーに保存され、一定時間で削除されます。
今回は上記の記事同様、Gyazoを使用してDiscord側に共有させます。
必須ではありませんが、記事の内容をやっておくと、流れを理解できて本稿の内容もすんなりとできると思います。(というかほぼまんま)
下準備
GyazoAPIキーの取得
GyazoAPIを使用するため、アカウント登録とAPIキーの取得をしておきましょう。
New Applicationを選択
名前とコールバックURLを指定します。
コールバックは使用しないの、名前と同様自由に決めてください。
submitを押すと、トークンが発行されます。
コイツを控えておきましょう。
環境変数
以下の環境変数を追加します。
GYAZO_TOKEN
Gyazoのトークン。
画像のアップロードに使用。
コーディング
ディレクトリ構成
$ tree
.
├── app
│ ├── cogs
│ │ └── mst_line.py # DiscordからLINEへ
│ ├── core
│ │ └── start.py # DiscordBot起動用
│ ├── message_type
│ │ ├── discord_type
│ │ │ ├── discord_type.py # Discordのサーバーに関するクラス
│ │ │ └── message_creater.py # DiscordAPIを直接叩く
│ │ └── line_type
│ │ ├── line_type.py # LINEのプロフィールなどのクラス
│ │ ├── line_event.py # LINEのイベントに関するクラス
│ │ └── line_message.py # LINEのメッセージに関するクラス
│ ├── server.py # サーバー立ち上げ
│ └── main.py
├── Dockerfile
├── Profile
└── requirements.txt
line_type.py
Gyazoのレスポンスもjsonとして受け取るため、前回と同様にクラスとして受け取ります。
line_type.py
import re
import json
class Base(object):
def __init__(self, **kwargs):
"""__init__ method.
:param kwargs:
"""
pass
def __str__(self):
"""__str__ method."""
return self.as_json_string()
def __repr__(self):
"""__repr__ method."""
return str(self)
def __eq__(self, other):
"""__eq__ method.
:param other:
"""
return other and self.as_json_dict() == other.as_json_dict()
def __ne__(self, other):
"""__ne__ method.
:param other:
"""
return not self.__eq__(other)
async def as_json_string(self):
"""jsonの文字列を返します。
:rtype: str
"""
return json.dumps(self.as_json_dict(), sort_keys=True)
async def as_json_dict(self):
"""このオブジェクトから辞書型を返します。
:return: dict
"""
data = {}
for key, value in self.__dict__.items():
camel_key = to_camel_case(key)
if isinstance(value, (list, tuple, set)):
data[camel_key] = list()
for item in value:
if hasattr(item, 'as_json_dict'):
data[camel_key].append(item.as_json_dict())
else:
data[camel_key].append(item)
elif hasattr(value, 'as_json_dict'):
data[camel_key] = value.as_json_dict()
elif value is not None:
data[camel_key] = value
return data
@classmethod
async def new_from_json_dict(cls, data:dict):
"""dict から新しいインスタンスを作成します。
:param data: JSONのディクショナリ
"""
new_data = {await to_snake_case(key): value
for key, value in data.items()}
return cls(**new_data)
async def to_snake_case(text:str):
"""スネークケースに変換する。
:param str text:
:rtype: str
"""
s1 = re.sub('(.)([A-Z])', r'\1_\2', text)
s2 = re.sub('(.)([0-9]+)', r'\1_\2', s1)
s3 = re.sub('([0-9])([a-z])', r'\1_\2', s2)
return s3.lower()
async def to_camel_case(text:str):
"""キャメルケースに変換する。
:param str text:
:rtype: str
"""
split = text.split('_')
return split[0] + "".join(x.title() for x in split[1:])
class Profile(Base):
"""
LINE Message APIのProfileクラス
user_id :LINEユーザーのid
display_name :LINEのユーザー名
picture_url :LINEのアイコンurl
status_message :LINEのプロフィール文
"""
def __init__(self,
user_id:str = None,
display_name:str = None,
picture_url:str = None,
status_message:str = None,
**kwargs
):
super(Profile, self).__init__(**kwargs)
self.user_id = user_id
self.display_name = display_name
self.picture_url = picture_url
self.status_message = status_message
+ class GyazoJson(Base):
+ """
+ Gyazoの画像クラス
+ image_id :画像id
+ permalink_url :画像のパーマリンク
+ thumb_url :サムネイル画像url
+ url :画像url
+ type :拡張子のタイプ
+ """
+ def __init__(self,
+ image_id:str=None,
+ permalink_url:str=None,
+ thumb_url:str=None,
+ url:str=None,
+ type:str=None,
+ **kwargs
+ ):
+ self.image_id = image_id
+ self.premalink_url = permalink_url
+ self.thumb_url = thumb_url
+ self.url = url
+ self.type = type
+ super(GyazoJson,self).__init__(**kwargs)
line_message.py
LINEから画像のバイナリデータを受け取り、Gyazoに送信する処理を追加します。
line_message.py
import json
import requests
from requests import Response
import datetime
import os
import asyncio
from functools import partial
import aiohttp
import subprocess
from typing import List
from dotenv import load_dotenv
load_dotenv()
+ from message_type.line_type.line_type import Profile,GyazoJson
NOTIFY_URL = 'https://notify-api.line.me/api/notify'
NOTIFY_STATUS_URL = 'https://notify-api.line.me/api/status'
LINE_BOT_URL = 'https://api.line.me/v2/bot'
LINE_CONTENT_URL = 'https://api-data.line.me/v2/bot'
# LINEのgetリクエストを行う
async def line_get_request(url: str, token: str) -> json:
async with aiohttp.ClientSession() as session:
async with session.get(
url = url,
headers = {'Authorization': 'Bearer ' + token}
) as resp:
return await resp.json()
# LINEのpostリクエストを行う
async def line_post_request(url: str, headers: dict, data: dict) -> json:
async with aiohttp.ClientSession() as session:
async with session.post(
url = url,
headers = headers,
data = data
) as resp:
return await resp.json()
class LineBotAPI:
def __init__(self, notify_token: str, line_bot_token: str, line_group_id: str) -> None:
self.notify_token = notify_token
self.line_bot_token = line_bot_token
self.line_group_id = line_group_id
self.loop = asyncio.get_event_loop()
# LINE Notifyでテキストメッセージを送信
async def push_message_notify(self, message: str) -> json:
data = {'message': f'message: {message}'}
return await line_post_request(
url = NOTIFY_URL,
headers = {'Authorization': f'Bearer {self.notify_token}'},
data = data
)
# LINE Messageing APIでテキストメッセージを送信
async def push_message(self,message_text:str) -> json:
data = {
'to':self.line_group_id,
'messages':[
{
'type':'text',
'text':message_text
}
]
}
return await line_post_request(
url = LINE_BOT_URL + "/message/push",
headers = {
'Authorization': 'Bearer ' + self.line_bot_token,
'Content-Type': 'application/json'
},
data = json.dumps(data)
)
# 送ったメッセージ数を取得
async def totalpush(self) -> int:
r = await line_get_request(
LINE_BOT_URL + "/message/quota/consumption",
self.line_bot_token
)
return int(r["totalUsage"])
# LINE Notifyのステータスを取得
async def notify_status(self) -> Response:
async with aiohttp.ClientSession() as session:
async with session.get(
url = NOTIFY_STATUS_URL,
headers = {'Authorization': 'Bearer ' + self.notify_token}
) as resp:
return resp
# LINE Notifyの1時間当たりの上限を取得
async def rate_limit(self) -> int:
resp = await self.notify_status()
ratelimit = resp.headers.get('X-RateLimit-Limit')
return int(ratelimit)
# LINE Notifyの1時間当たりの残りの回数を取得
async def rate_remaining(self) -> int:
resp = await self.notify_status()
ratelimit = resp.headers.get('X-RateLimit-Remaining')
return int(ratelimit)
# LINE Notifyの1時間当たりの画像送信上限を取得
async def rate_image_limit(self) -> int:
resp = await self.notify_status()
ratelimit = resp.headers.get('X-RateLimit-ImageLimit')
return int(ratelimit)
# LINE Notifyの1時間当たりの残り画像送信上限を取得
async def rate_image_remaining(self) -> int:
resp = await self.notify_status()
ratelimit = resp.headers.get('X-RateLimit-ImageRemaining')
return int(ratelimit)
# 友達数、グループ人数をカウント
async def friend(self) -> str:
# グループIDが有効かどうか判断
try:
r = await line_get_request(
LINE_BOT_URL + "/group/" + self.line_group_id + "/members/count",
self.line_bot_token,
)
return r["count"]
# グループIDなしの場合、友達数をカウント
except KeyError:
# 日付が変わった直後の場合、前日を参照
if datetime.datetime.now().strftime('%H') == '00':
before_day = datetime.date.today() + datetime.timedelta(days=-1)
url = LINE_BOT_URL + "/insight/followers?date=" + before_day.strftime('%Y%m%d')
else:
url = LINE_BOT_URL + "/insight/followers?date=" + datetime.date.today().strftime('%Y%m%d')
r = await line_get_request(
url,
self.line_bot_token,
)
return r["followers"]
# 当月に送信できるメッセージ数の上限目安を取得(基本1000,23年6月以降は200)
async def pushlimit(self) -> str:
r = await line_get_request(
LINE_BOT_URL + "/message/quota",
self.line_bot_token
)
return r["value"]
# LINEのユーザプロフィールから名前を取得
async def get_proflie(self, user_id: str) -> Profile:
# グループIDが有効かどうか判断
try:
r = await line_get_request(
LINE_BOT_URL + f"/group/{self.line_group_id}/member/{user_id}",
self.line_bot_token,
)
# グループIDが無効の場合、友達から判断
except KeyError:
r = await line_get_request(
LINE_BOT_URL + f"/profile/{user_id}",
self.line_bot_token,
)
return await Profile.new_from_json_dict(data=r)
+ # LINEから画像データを取得し、Gyazoにアップロード
+ async def image_upload(self, message_id: int) -> GyazoJson:
+ # 画像のバイナリデータを取得
+ async with aiohttp.ClientSession() as session:
+ async with session.get(
+ url = LINE_CONTENT_URL + f'/message/{message_id}/content',
+ headers={
+ 'Authorization': 'Bearer ' + self.line_bot_token
+ }
+ ) as bytes:
+ image_bytes = await bytes.read()
+
+ # Gyazoにアップロードする
+ async with aiohttp.ClientSession() as session:
+ async with session.post(
+ url = 'https://upload.gyazo.com/api/upload',
+ headers={
+ 'Authorization': 'Bearer ' + os.environ['GYAZO_TOKEN'],
+ },
+ data={
+ 'imagedata': image_bytes
+ }
+ ) as gyazo_image:
+ return await GyazoJson.new_from_json_dict(await gyazo_image.json())
server.py
受け取ったGyazoの画像URLをDiscordに送信します。
server.py
from fastapi import FastAPI,Depends,HTTPException,Request,Header,Response
from fastapi.responses import HTMLResponse
from threading import Thread
import uvicorn
import base64
import hashlib
import hmac
import re
from dotenv import load_dotenv
load_dotenv()
from message_type.line_type.line_event import Line_Responses
from message_type.discord_type.message_creater import ReqestDiscord
from message_type.line_type.line_message import LineBotAPI
import os
bots_name = os.environ['BOTS_NAME'].split(",")
TOKEN = os.environ['TOKEN']
app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
# LINE側のメッセージを受け取る
@app.post("/line_bot")
async def line_response(
response:Line_Responses,
byte_body:Request,
x_line_signature=Header(None)
):
"""
response:Line_Responses
LINEから受け取ったイベントの内容
jsonがクラスに変換されている。
byte_body:Request
LINEから受け取ったイベントのバイナリデータ。
LINEからのメッセージという署名の検証に必要。
x_line_signature:Header
LINEから受け取ったjsonのヘッダー。
こちらも署名に必要。
"""
# request.bodyを取得
boo = await byte_body.body()
body = boo.decode('utf-8')
# channel_secretからbotの種類を判別する
for bot_name in bots_name:
channel_secret = os.environ[f'{bot_name}_CHANNEL_SECRET']
# ハッシュ値を求める
hash = hmac.new(
channel_secret.encode('utf-8'),
body.encode('utf-8'),
hashlib.sha256
).digest()
# 結果を格納
signature = base64.b64encode(hash)
decode_signature = signature.decode('utf-8')
if decode_signature == x_line_signature:
channel_secret = os.environ[f'{bot_name}_CHANNEL_SECRET']
# Discordサーバーのクラスを宣言
discord_find_message = ReqestDiscord(
guild_id = int(os.environ[f'{bot_name}_GUILD_ID']),
limit = int(os.environ["USER_LIMIT"]),
token = TOKEN
)
# LINEのクラスを宣言
line_bot_api = LineBotAPI(
notify_token = os.environ.get(f'{bot_name}_NOTIFY_TOKEN'),
line_bot_token = os.environ[f'{bot_name}_BOT_TOKEN'],
line_group_id = os.environ.get(f'{bot_name}_GROUP_ID')
)
# メッセージを送信するDiscordのテキストチャンネルのID
channel_id = int(os.environ[f'{bot_name}_CHANNEL_ID'])
break
# ハッシュ値が一致しなかった場合エラーを返す
if decode_signature != x_line_signature:
raise Exception
# 応答確認の場合終了
if type(response.events) is list:
return HTMLResponse("OK")
# イベントの中身を取得
event = response.events
# LINEのプロフィールを取得(友達登録している場合)
profile_name = await line_bot_api.get_proflie(user_id=event.source.userId)
# テキストメッセージの場合
if event.message.type == 'text':
message = event.message.text
# Discordのメンバー、ロール、チャンネルの指定があるか取得する
"""
members_find
テキストメッセージからユーザーのメンションを検出し、変換する。
@ユーザー#0000#member → <@00000000000>
roles_find
テキストメッセージからロールのメンションを検出し、変換する。
@ロール#role → <@&0000000000>
channel_select
テキストメッセージから送信場所を検出し、送信先のチャンネルidを返す。
テキストチャンネルのみ送信可能。ただし、メッセージの先頭に書かれていなければ適用されない。
/チャンネル名#channel → 削除
"""
message = await discord_find_message.members_find(message=message)
message = await discord_find_message.roles_find(message=message)
channel_id, message = await discord_find_message.channel_select(channel_id=channel_id,message=message)
+ # 画像が送信された場合
+ if event.message.type == 'image':
+ # バイナリデータを取得しGyazoに送信
+ gyazo_json = await line_bot_api.image_upload(event.message.id)
+ # Gyazoのurlを返す
+ message = f"https://i.gyazo.com/{gyazo_json.image_id}.{gyazo_json.type}"
# LINEの名前 「メッセージ」の形式で送信
message = f'{profile_name.display_name} \n「 {message} 」'
await discord_find_message.send_discord(channel_id=channel_id, message=message)
# レスポンス200を返し終了
return HTMLResponse(content="OK")
def run():
uvicorn.run("server:app", host="0.0.0.0", port=int(os.getenv("PORT", default=5000)), log_level="info")
# DiscordBotと並列で立ち上げる
def keep_alive():
t = Thread(target=run)
t.start()
# ローカルで実行する際
if __name__ == '__main__':
uvicorn.run(app,host='localhost', port=8000)
ポイント
# 画像が送信された場合
if event.message.type == 'image':
# バイナリデータを取得しGyazoに送信
gyazo_json = await line_bot_api.image_upload(event.message.id)
# Gyazoのurlを返す
message = f"https://i.gyazo.com/{gyazo_json.image_id}.{gyazo_json.type}"
見ているとわかりますがテキストメッセージに直接画像のURLを貼り付けています。
Discordにはファイルのプレビュー機能があるため、テキストに直接貼り付けても画像がプレビューされるのです。
完成！！
試しに画像を送信してみましょう。
Discord側でプレビューできていればOKです！！
まとめ
お疲れ様でした。
あまり説明することもなかったので、結構内容は短めになってしまいました。
さて残りは
となります。いよいよ折り返しですね。
次回は
となります。
結構難関ですのでオウム返しで予習してみてください。