挨拶
こんにちは。マグロです。
前回の続きとなります。
今回はLINEからDiscordへテキストメッセージを送れるようにします。
設計
LINEBotはメッセージを受け取った場合、Developerサイトで設定したWebHookに内容が送信されます。
そのためFastAPIでサーバーを立ち上げ、WebHookを設定します。
しかしここで一つ問題があります。
そこからどうやってDiscordに送る？？
Discord.pyやPycordを使って送ればいいと思いますが、Botの起動が優先されてしまい、サーバーが立ち上がりません。
DiscordBot内にFastAPIを組み込む方法はあるようですが、cogでは使用不可の模様。
こうなると並列でサーバーを起動させるしか方法はありません。
そうなるとライブラリは使用不可となります....
じゃあDiscordAPI直接叩くか！！
というわけで設計です。
まずFastAPIを並列で実行して、サーバーを立ち上げます。
そのあとBotを起動します。
サーバーはLINEから受け取ったイベントの受け皿となり、イベントの種類がテキストメッセージだった場合、Discordにメッセージを送信します。
また前回説明しましたが、複数のサーバーでの運用を想定しているため、どのLINEBotから送信されたか判別する必要があります。これは署名を使用します。(詳細は後ほど)
コーディング
ディレクトリ構成
$ 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のプロフィールなどのクラス(前回のclass_type.py)
│ │ ├── line_event.py # LINEのイベントに関するクラス
│ │ └── line_message.py # LINEのメッセージに関するクラス
│ ├── server.py # サーバー立ち上げ
│ └── main.py
├── Dockerfile
├── Profile
└── requirements.txt
line_type.py
LINEからイベントを受け取る際、json形式で受け取ります。
pythonでは受け取ったjsonをdict型で扱います。
そのまま使っても問題ないのですが、可読性が下がるのでクラスにしてしまいます。
line-bot-sdkのBase.pyを参考(ほぼ丸パクリ)に書きます。
line_type.py
import re
import json
from typing import List
from message_type.line_type.line_event import Line_Events
class Base(object):
def __init__(self, **kwargs):
"""__init__ method.
:param kwargs:
"""
pass
# 文字列が参照された場合
def __str__(self):
"""__str__ method."""
# jsonを文字列として返す
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():
# キャメルケースに変換する(user_id→userId)
camel_key = to_camel_case(key)
# 値がlist,tuple,setのいずれかの場合
if isinstance(value, (list, tuple, set)):
# 型の中身をlistにする
data[camel_key] = list()
for item in value:
# 中にjsonがある場合、as_json_dictを実行
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のディクショナリ
"""
# スネークケースに変換(userId→user_id)
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
詳細を説明すると長くなるため割愛します。(そもそもあまりに理解できてない)
ここではLINEのプロフィール名を取得するProfileを使用します。
line_event.py
FastAPIはpydanticを使用することで受け取るイベントの型の定義ができます。
LINEのメッセージイベントの型を定義します。
line_event.py
from pydantic import BaseModel,validator
from typing import List,Optional,Union
class ContentProvider(BaseModel):
"""
コンテンツファイルのクラス
type :str
ファイルの提供元。
originalContentUrl :Optional[str]
ファイルのURL。基本的には含まれない。
previewImageUrl :Optional[str]
プレビュー画像のURL。基本的には含まれない。
"""
type:str
originalContentUrl:Optional[str]
previewImageUrl:Optional[str]
class ImageSet(BaseModel):
"""
画像のクラス
id :int
画像セットID。複数の画像を同時に送信した場合のみ含まれる。
total :Optional[float]
同時に送信した画像の総数。
index :Optional[float]
同時に送信した画像セットの中で、何番目の画像かを示す1から始まるインデックス。
画像が届く順番が不定なので、付けられている。
"""
id:int
total:Optional[float]
index:Optional[float]
class Message(BaseModel):
"""
メッセージの内容を含むオブジェクト。
詳細は以下の公式リファレンスを参照
https://developers.line.biz/ja/reference/messaging-api/#message-event
text :Optional[str]
メッセージのテキスト
id :int
メッセージID(公式リファレンスでは文字列だが、すべて整数なのでint)
type :str
メッセージの種類(テキストか画像か)
imageSet :Optional[ImageSet]
画像のセットを表すクラス。複数の画像を同時に送信した場合のみ含まれる。
contentProvider :Optional[ContentProvider]
画像、動画、音声ファイルの提供元。
duration :Optional[int]
動画、音声ファイルの長さ（ミリ秒）
fileName :Optional[str]
ファイル名
fileSize :Optional[int]
ファイルサイズ（バイト）
title :Optional[str]
位置情報タイトル
address :Optional[str]
住所
latitude :Optional[float]
緯度
longitude :Optional[float]
経度
packageId :Optional[str]
スタンプのパッケージID
stickerId :Optional[str]
スタンプID
stickerResourceType :Optional[str]
スタンプのリソースタイプ。
keywords :Optional[List[str]]
スタンプを表すキーワード。
"""
text:Optional[str]
id:int
type:str
imageSet:Optional[ImageSet]
contentProvider:Optional[ContentProvider]
duration:Optional[int]
fileName:Optional[str]
fileSize:Optional[int]
title:Optional[str]
address:Optional[str]
latitude:Optional[float]
longitude:Optional[float]
packageId:Optional[str]
stickerId:Optional[str]
stickerResourceType:Optional[str]
keywords:Optional[List[str]]
class Source(BaseModel):
"""
イベントの送信元情報を含むユーザー、グループトーク、または複数人トーククラス。
groupId :Optional[str]
送信元グループトークのグループID
userId :str
送信元ユーザーのID。
type :str
送信元のタイプ(ユーザー、グループ)
"""
groupId:Optional[str]
userId:str
type:str
class DeliveryContext(BaseModel):
"""
Webhookイベントが再送されたものかどうかを表すクラス。
isRedelivery:bool
再送されたものかどうか
"""
isRedelivery:bool
class Line_Events(BaseModel):
"""
LINEのイベントクラス
詳細は以下の公式リファレンスを参照
https://developers.line.biz/ja/reference/messaging-api/#common-properties
timestamp :float
イベントが送られてきた時間（ミリ秒）
mode :str
チャネルの状態。
replyToken :str
このイベントに対して応答メッセージを送る際に使用する応答トークン
deliveryContext :DeliveryContext
Webhookイベントが再送されたものかどうか。
webhookEventId :str
WebhookイベントID。Webhookイベントを一意に識別するためのID。ULID形式の文字列になる。
type :str
イベントのタイプを表す識別子
message :Message
メッセージの内容を含むオブジェクト。
source :Source
イベントの送信元情報を含むユーザー、グループトーク、または複数人トークオブジェクト。
"""
timestamp:float
mode:str
replyToken:str
deliveryContext:DeliveryContext
webhookEventId:str
type:str
message:Message
source:Source
class Line_Responses(BaseModel):
"""
LINEのイベントクラス
destination:str
BotのユーザID
events:List[Line_Events] or Line_Events
LINE側でのイベント内容
応答確認の場合はlistで返す。
それ以外の場合はlistの中身を返す。
"""
destination:str
events:Union[List[Line_Events],Line_Events]
@validator("events")
def validate_hoge(cls, value):
# 応答確認の場合
if len(value) == 0:
return value
# Listの中身を返す。
value:Optional[Line_Events]
return value[0]
line_message.py
2023/2/11追記
LINEのプロフィールからユーザー名を取得する関数を加え忘れてました！！
送信元が誰かを判別させるため、LINEユーザー名を取得する関数を作成します。
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
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)
discord_type.py
上記のline_type.pyと同様に受け取ったjsonをクラスに変換します。
discord_type.py
from typing import List
import re
import json
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()
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 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)
def as_json_string(self):
"""jsonの文字列を返します。
:rtype: str
"""
return json.dumps(self.as_json_dict(), sort_keys=True, ensure_ascii=False)
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
def new_from_json_dict(cls, data:dict):
"""dict から新しいインスタンスを作成します。
:param data: JSONのディクショナリ
"""
new_data = {to_snake_case(key): value
for key, value in data.items()}
return cls(**new_data)
class User(Base):
"""
DiscordのUserクラス
id :ユーザーid
username :ユーザー名
avatar :ユーザーのアバターハッシュ
avatar_decoration :ユーザーのアバターのデコレーション
discriminator :4桁のユーザー番号
public_flags :ユーザーアカウントの公開フラグ
bot :botかどうか
"""
def __init__(
self,
id:int = None,
username:str = None,
avatar:str = None,
avatar_decoration:str = None,
discriminator:int = None,
public_flags:int = None,
bot:bool = None,
**kwargs
):
self.id = id
self.username = username
self.avater = avatar
self.avater_decoration = avatar_decoration
self.discreminator = discriminator
self.public_flags = public_flags
self.bot = bot
super().__init__(**kwargs)
class Permission(Base):
"""
Discordのチャンネルの権限のクラス
上書きする際に使用
id :チャンネルのid
type :チャンネルのタイプ
allow :許可されている権限
deny :禁止されている権限
allow_new :新たなに許可する権限
deny_new :新たに禁止する権限
"""
def __init__(
self,
id:int = None,
type:str = None,
allow:int = None,
deny:int = None,
allow_new:int = None,
deny_new:int = None,
**kwargs
):
self.id = id
self.type = type
self.allow = allow
self.deny = deny
self.allow_new = allow_new
self.deny_new = deny_new
super().__init__(**kwargs)
class Discord_Member(Base):
"""
Discordのユーザーのサーバーでのステータス
user :Discordのユーザークラス
nick :ニックネーム
is_pending :用途不明 https://github.com/discord/discord-api-docs/issues/2235
flags :こちらも用途不明
avatar :ユーザーのアバターハッシュ
roles :サーバーで割り当てられているロール
joined_at :参加した日付
deaf :スピーカーミュートしているか
mute :マイクミュートしているか
"""
def __init__(
self,
user:User = None,
nick:str = None,
is_pending:bool = None,
flags:int = None,
avatar:str = None,
roles:List[int] = None,
joined_at:str = None,
deaf:bool = None,
mute:bool = None,
**kwargs
):
self.user = User.new_from_json_dict(user)
self.nick = nick
self.is_pending = is_pending
self.flags = flags
self.avatar = avatar
self.roles = roles
self.joined_at = joined_at
self.deaf = deaf
self.mute = mute
super().__init__(**kwargs)
class Discord_Role(Base):
"""
id :ロールid
name :ロール名
description :ロールの説明
permissions :ロールに割り当てられている権限
position :ロールの順番
color :ロールの色
hoist :オンラインメンバーとは別に表示するか
managed :管理者権限?
mentionable :メンション可能かどうか
icon :サーバーにギルドアイコン機能がある場合、その画像
unicode_emoji :ギルドアイコン機能での絵文字
flags :用途不明
permissions_new :新たに設定する権限
"""
def __init__(
self,
id:int = None,
name:str = None,
description:str = None,
permissions:int = None,
position:int = None,
color:int = None,
hoist:bool = None,
managed:bool = None,
mentionable:bool = None,
icon:str = None,
unicode_emoji:str = None,
flags:int = None,
permissions_new:int = None,
**kwargs
):
self.id = id
self.name = name
self.description = description
self.permissions = permissions
self.position = position
self.color = color
self.hoist = hoist
self.managed = managed
self.mentionable = mentionable
self.icon = icon
self.unicode_emoji = unicode_emoji
self.flags = flags
self.permissions_new = permissions_new
super().__init__(**kwargs)
class Discord_Channel(Base):
"""
Discordのチャンネルのクラス
id :チャンネルid
last_message_id :最後に発言されたメッセージのid
type :チャンネルのタイプ(0の場合、テキストチャンネル)
name :チャンネル名
position :チャンネルの順番
flags :用途不明
parent_id :親チャンネルのid
bitrate :音声のビットレート
user_limit :ボイスチャンネルのユーザーの上限
rtc_region :音声のリージョン
topic :チャンネルのトピックス
guild_id :サーバーid
premission_overwrites :新たに設定する権限
rate_limit_per_user :低速モードで再び発言できるまでの秒数
nsfw :閲覧注意チャンネルかどうか
"""
def __init__(
self,
id:int = None,
last_message_id:int = None,
type:int = None,
name:str = None,
position:int = None,
flags:int = None,
parent_id:str = None,
bitrate:int = None,
user_limit:int = None,
rtc_region:str = None,
topic:str = None,
guild_id:int = None,
permission_overwrites:List[Permission] = None,
rate_limit_per_user:int = None,
nsfw:bool = None,
**kwargs
):
self.id = id
self.last_message_id = last_message_id
self.type = type
self.name = name
self.position = position
self.flags = flags
self.parent_id = parent_id
self.bitrate = bitrate
self.user_limit = user_limit
self.rtc_region = rtc_region
self.topic = topic
self.guild_id = guild_id
self.permission_overwrites = permission_overwrites
self.rate_limit_per_user = rate_limit_per_user
self.nsfw = nsfw
super().__init__(**kwargs)
message_creater.py
DiscordAPIを直接叩きます。
ユーザー、ロール、チャンネルの取得とメッセージの送信を行います。
message_creater.py
import os
import re
import aiohttp
import requests
import asyncio
import time
from dotenv import load_dotenv
load_dotenv()
from functools import partial
from typing import List,Tuple
import asyncio
from message_type.discord_type.discord_type import Discord_Member,Discord_Role,Discord_Channel
class ReqestDiscord:
def __init__(self, guild_id: int, limit: int, token: str) -> None:
self.guild_id = guild_id
self.limit = limit
self.headers = {
'Authorization': f'Bot {token}',
'Content-Type': 'application/x-www-form-urlencoded',
}
async def member_get(self) -> List[Discord_Member]:
"""
サーバーのユーザーを取得する。
戻り値
Discord_Member
"""
async with aiohttp.ClientSession() as session:
async with session.get(
url = f'https://discordapp.com/api/guilds/{self.guild_id}/members?limit={self.limit}',
headers = self.headers
) as resp:
# 取得したユーザー情報を展開
res = await resp.json()
member_list = []
for member in res:
r = Discord_Member.new_from_json_dict(member)
member_list.append(r)
return member_list
async def role_get(self) -> List[Discord_Role]:
"""
ロールを取得する。
戻り値
Discord_Role
"""
async with aiohttp.ClientSession() as session:
async with session.get(
url = f'https://discordapp.com/api/guilds/{self.guild_id}/roles',
headers = self.headers
) as resp:
# 取得したロール情報を取得
res = await resp.json()
role_list = []
for role in res:
r = Discord_Role.new_from_json_dict(role)
role_list.append(r)
return role_list
async def channel_get(self) -> List[Discord_Channel]:
"""
チャンネルを取得する。
戻り値
Discord_Channel
"""
async with aiohttp.ClientSession() as session:
async with session.get(
url = f'https://discordapp.com/api/guilds/{self.guild_id}/channels',
headers = self.headers
) as resp:
# 取得したチャンネルを展開
res = await resp.json()
channel_list = []
for channel in res:
r = Discord_Channel.new_from_json_dict(channel)
channel_list.append(r)
return channel_list
async def members_find(self, message: str) -> str:
"""
テキストメッセージのメンションを変換する。
@ユーザ名#4桁の数字#member → @ユーザ名
戻り値
message 変更後の文字列: str
"""
# @{空白以外の0文字以上}#{0以上の数字}#member
member_mention_list = re.findall("@\S*?#\d*?#member",message,re.S)
if not member_mention_list:
return message
get_member_list = await self.member_get()
for member in get_member_list:
# ユーザー名の空白文字を削除
member.user.username = re.sub("[\u3000 \t]", "",member.user.username)
# メッセージに「@{ユーザー名}#{4桁の数字}member」が含まれていた場合
if f'@{member.user.username}#{member.user.discreminator}#member' in member_mention_list:
message = message.replace(f'@{member.user.username}#{member.user.discreminator}#member',f'<@{member.user.id}>')
member_mention_list = [
user for user in member_mention_list
if user != f'@{member.user.username}#{member.user.discreminator}#member'
]
if not member_mention_list:
return message
return message
async def roles_find(self, message: str) -> str:
"""
テキストメッセージのメンションを変換する。
@ロール名#role → @ロール名
戻り値
message 変更後の文字列: str
"""
role_list = re.findall("@\S*?#role",message,re.S)
if not role_list:
return message
get_role_list = await self.role_get()
for role in get_role_list:
# ロール名の空白文字を削除
role.name = re.sub("[\u3000 \t]", "",role.name)
# メッセージに「@{ロール名}#role」が含まれていた場合
if f'@{role.name}#role' in role_list:
message = message.replace(f'@{role.name}#role',f'<@&{role.id}>')
role_list = [
rolename for rolename in role_list
if rolename != f'@{role.name}#role'
]
if not role_list:
return message
return message
async def channel_select(self, channel_id: int, message: str) -> Tuple[int,str]:
"""
テキストメッセージから送信場所を読み取り変更する。
テキストチャンネルのみ可能。
/チャンネル名#channel → 削除
戻り値
channel_id 送信先のチャンネル :id
message 指定したチャンネル名 :str
"""
channel_list = re.findall("\A/\S*?#channel",message,re.S)
if not channel_list or message.find('/') != 0:
return channel_id, message
get_channel_list = await self.channel_get()
for channel in get_channel_list:
# チャンネル名の空白文字を削除
channel.name = re.sub("[\u3000 \t]", "",channel.name)
# メッセージの先頭に「/{チャンネル名}#channel」が含まれていた場合
if message.find(f'/{channel.name}#channel') == 0 and channel.type == 0:
message = message.lstrip(f'/{channel.name}#channel')
channel_id = channel.id
return channel_id, message
return channel_id, message
async def send_discord(self, channel_id: int, message: str):
"""
Discordへメッセージを送信する。
channel_id :int
Discordのテキストチャンネルのid
message :str
テキストメッセージ
"""
async with aiohttp.ClientSession() as session:
async with session.post(
url = f'https://discordapp.com/api/channels/{channel_id}/messages',
headers = self.headers,data = {'content': f'{message}'}
) as resp:
return await resp.json()
ポイントというか解説。
async with aiohttp.ClientSession() as session:
async with session.get(
url = f'https://discordapp.com/api/guilds/{self.guild_id}/members?limit={self.limit}',
headers = self.headers
) as resp:
aiohttpはdiscord.pyに標準でついてくるライブラリで、非同期でリクエストを行えます。
同期でリクエストを行うと遅延や処理が正しく行われない場合があるので、必ずaiohttpでリクエストします。(discord.py公式でも推奨されています。)
async def members_find(self, message: str) -> str:
async def roles_find(self, message: str) -> str:
async def channel_select(self, channel_id: int, message: str) -> Tuple[int,str]:
この三つの関数はそれぞれ
- ユーザーへのメンション
- ロールへのメンション
- 送信先チャンネルの指定
ができます。
テキストメッセージに以下のような記述をすると適用されます。
@ユーザー名#ユーザーの4桁の数字#member(4桁の数字はフレンド申請に使用する数字)
@ロール名#role
/チャンネル名#channel(メッセージの先頭に記述すること)
また空白文字は取り除かれます。(後述)
例:ユーザー名がi am the storm that is approaching#0000の場合、
@iamthestormthatisapproaching#0000#memberとすることでメンションできます。
例としてユーザー名のメンションを見ていきます。
# @{空白以外の0文字以上}#{0以上の数字}#member
member_mention_list = re.findall("@\S*?#\d*?#member",message,re.S)
if not member_mention_list:
return message
get_member_list = await self.member_get()
正規表現でテキストメッセージ内に@ユーザー名#ユーザーの4桁の数字#memberがあるか判別します。
ある場合listとして格納され、対応するユーザーがいるか調べます。
for member in get_member_list:
# ユーザー名の空白文字を削除
member.user.username = re.sub("[\u3000 \t]", "",member.user.username)
# メッセージに「@{ユーザー名}#{4桁の数字}member」が含まれていた場合
if f'@{member.user.username}#{member.user.discreminator}#member' in member_mention_list:
message = message.replace(f'@{member.user.username}#{member.user.discreminator}#member',f'<@{member.user.id}>')
member_mention_list = [
user for user in member_mention_list
if user != f'@{member.user.username}#{member.user.discreminator}#member'
]
if not member_mention_list:
return message
return message
取得したユーザー情報をfor文で一つ一つ参照します。
(ユーザー名に空白文字がある場合消します。)
該当するユーザーがいた場合、メンション形式に書き換えます。その後、書き換えたユーザーをlistから削除します。
listが空になり次第、終了します。
ここで、正規表現で\S(空白文字以外の文字)を使用している理由を説明します。
空白文字を許容したとして、以下のようなテキストメッセージが送られてきたとします。
@悪魔#role @i am the storm that is approaching#0000#member
members_findで空白文字を含む場合、member_mention_listの中身はこうなります。
member_mention_list = ["@悪魔#role @i am the storm that is approaching#0000#member"]
、、、あれ？？
本来はこうなるはずですが、、
member_mention_list = ["@i am the storm that is approaching#0000#member"]
そう、空白文字を許容した場合悪魔#role @i am the storm that is approachingをユーザー名と判断してしまいます。
そのため空白文字を消しています。
main.py
並列でサーバーを起動させるため、起動用関数のkeep_aliveを追加します。
main.py
from core.start import DBot
import discord
import os
from dotenv import load_dotenv
load_dotenv()
+ from server import keep_alive
# サーバー立ち上げ
+ keep_alive()
Token=os.environ['TOKEN']
# Bot立ち上げ
DBot(Token,discord.Intents.all()).run()
server.py
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)
# 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)
ポイントとなる点はこちら。
# 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')
LINE Developerには署名について以下のように記載されています。
リクエストがLINEプラットフォームから送られたことを確認するために、ボットサーバーでリクエストヘッダーのx-line-signatureに含まれる署名を検証します。
チャネルシークレットを秘密鍵として、HMAC-SHA256アルゴリズムを使用してリクエストボディのダイジェスト値を取得します。
ダイジェスト値をBase64エンコードした値と、リクエストヘッダーのx-line-signatureに含まれる署名が一致することを確認します。
上記のコードはダイジェスト値の取得を行っています。
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
署名が一致した場合、対応するLINEBotとDiscordBotのクラスを宣言します。
# ハッシュ値が一致しなかった場合エラーを返す
if decode_signature != x_line_signature:
raise Exception
# 応答確認の場合終了
if type(response.events) is list:
return HTMLResponse("OK")
ハッシュ値が一致しない場合は終了します。
応答確認の場合イベントはlist型で来るため、その場合終了します。
デプロイ
railwayにデプロイします。
こちらが参考になります。
またProfileに起動用のコマンドを記述します。
/bin/bash -c "cd app && python -u main.py"
デプロイが完了し、URLが生成されたら、LINE DeveloperのWebHookを忘れずに登録しましょう。
https://******railway.app/line_botがエンドポイントです。
応答確認もしておきましょう。
完成！
試しにメンションしてみます。
まとめ
お疲れ様でした。
これによりテキスト間でLINEとDiscordのやり取りができるようになりました。
残りは
の4つとなります。
次回は
を解説します。