🤝

DiscordとLINEをPython+FastAPI+Dockerで連携させる【その1】DiscordからLINEへテキストメッセージ

2023/01/19に公開

LINE

Python

Discord

tech

挨拶

qiitaにも同様の投稿をしています。

こんにちは。マグロです。
以前記事にしたものの続編です。

LINE Notify対応版が完成したので、記事にします。

順を追って解説できるように、記事をいくつか分割して解説します。
今回はDiscordからLINEへテキストメッセージを送ります。

すぐに使いたい場合は上記のリポジトリをフォークしてrailwayにデプロイしてください。

注意

以下の知識があることが前提となります。

Python
Discord.py+Pycord
LINE Message API
LINE Notify
Docker

また、前回と同様、railwayでホストすることを想定しております。
説明は省略いたしますので各自で調べてください。

LINE Notifyのメリット

送れるメッセージ上限が大幅に増えます。
以前の公式アカウントの場合、月1000件(23年6月以降は200件)しか送れず、機能しているとはいいがたいものでした。

ですがLINE Notifyの場合、1時間に1000件なので、ほぼ無制限といっても過言ではないでしょう。

これで課題だった上限による相互間のやり取りのしづらさが改善されました。

設計

DiscordBotは複数のサーバーでの運用を想定している。
それに伴い、それぞれのサーバーごとにLINEグループを作成する。
LINEグループには、それぞれのLINE公式アカウントを参加させる。
LINEグループ用のNotifyも登録しておく。

DiscordからLINEへ

1: メッセージをDiscordBotが読み込む。
2: 送られてきたメッセージから送信元のサーバーを特定し、対応付けられているLINEグループを判断する。
3: メッセージを変換し、LINE NotifyでメッセージをLINEグループに送信する。

LINEからDiscordへ

1: メッセージをLINEBotが読み込む。
2: 送信元のLINEBotから対応付けられているDiscordサーバーを特定する。
3: メッセージを変換し、DiscordBotからメッセージを送信する。

まとめると

DiscordBot:Discordのメッセージの受け取りとLINEからのメッセージを送信。
LINEBot:LINEグループのメッセージの受け取り。
LINE Notify:DiscordのメッセージをLINEグループに送信。

といった役割分担になります。

下準備

Discord DevloperサイトからBotを、LINE Devloperサイトからやり取りしたいサーバーの数までBotを発行します。

DiscordBotのトークン
LINEBotのトークン
LINEBotのシークレットキー

を控えておきましょう。
次にLINEグループを作成し、グループIDを取得します。
以下の記事を参考にするといいでしょう。

最後にLINE Notifyを作成したグループに登録します。
発行したトークンを控えましょう。
まとめると

DiscordBotのトークン
LINEBotのトークン
LINEBotのシークレットキー
LINEグループのid
LINE Notifyのトークン

この5つになります。

コーディング

ディレクトリ構成

$ tree
.
├── app
│   ├── cogs
│   │   └── mst_line.py  # DiscordからLINEへ
│   ├── core
│   │   └── start.py     # DiscordBot起動用
│   ├── message_type
│   │   ├── discord_type
│   │   │   └── message_creater.py  # DiscordAPIを直接叩く(今回は使いません)
│   │   └── line_type
│   │       ├── class_type.py       # LINEのプロフィールなどのクラス(今回は使いません)
│   │       ├── line_event.py       # LINEのイベントに関するクラス(今回は使いません)
│   │       └── line_message.py     # LINEのメッセージに関するクラス
│   └── main.py  
├── Dockerfile
├── Profile
└── requirements.txt

環境

Dockerを使って環境を構築し、cogでBotを稼働させます。

Dockerfile

FROM python:3.10.7
USER root

# ディレクトリ ./appに移動
WORKDIR /app

RUN apt-get -y update && apt-get -y install locales && apt-get -y upgrade && \
    localedef -f UTF-8 -i ja_JP ja_JP.UTF-8
ENV LANG ja_JP.UTF-8
ENV LANGUAGE ja_JP:ja
ENV LC_ALL ja_JP.UTF-8
ENV TZ JST-9
ENV TERM xterm

# ./root/src ディレクトリを作成 ホームのファイルをコピーして、移動
RUN mkdir -p /root/src
COPY . /root/src
WORKDIR /root/src

# Docker内で扱うffmpegをインストール
RUN apt-get install -y ffmpeg

# pipのアップグレード、requirements.txtから必要なライブラリをインストール
RUN pip install --upgrade pip
RUN pip install --upgrade setuptools
RUN pip install -r requirements.txt
# discord.pyをpycordにアップグレード
RUN pip install git+https://github.com/Pycord-Development/pycord

必要なライブラリをインストールするため、requirements.txtに書き込みます。

absl-py==0.15.0
aiohttp==3.7.4.post0
aiofiles==22.1.0
anyio==3.5.0
appdirs==1.4.4
asgiref==3.5.2
astroid==2.4.2
astunparse==1.6.3
async-timeout==3.0.1
attrs==21.2.0
decorator==5.1.0
discord.py==1.7.3
docutils==0.19
fastapi==0.88.0
ffmpeg-python==0.2.0
git-lfs==1.6
google-api-core==1.31.1
google-api-python-client==2.15.0
google-auth==1.34.0
google-auth-httplib2==0.1.0
google-auth-oauthlib==0.4.6
google-pasta==0.2.0
googleapis-common-protos==1.53.0
gunicorn==20.1.0
httpcore==0.13.7
httplib2==0.19.1
httpx==0.19.0
hyperframe==6.0.1
idna==3.2
importlib-metadata==4.10.1
joblib==1.1.0
librosa==0.9.2
lxml==4.8.0
numpy==1.23.0
oauth2client==4.1.3
oauthlib==3.2.0
opt-einsum==3.3.0
pyasn1==0.4.8
pyasn1-modules==0.2.8
pydub==0.25.1
pydantic==1.9.1
PyNaCl==1.4.0
python-dotenv
requests==2.26.0
requests-oauthlib==1.3.1
rsa==4.7.2
scipy==1.9.3
SoundFile==0.10.3.post1
threadpoolctl==3.0.0
typed-ast==1.5.4
typer==0.3.2
typing_extensions==4.1.1
tzdata==2022.4
uritemplate==3.0.1
urllib3==1.26.6
uvicorn==0.20.0
Werkzeug==2.0.2
wrapt==1.12.1
yarl==1.6.3
youtube-dl==2021.12.17
zipp==3.7.0

環境変数

これから書くコードには、以下の環境変数が含まれています。

TOKEN

DiscordBotのトークンです。

GAME_NAME

DiscordBotがプレイするゲームの名前です。
ない場合僕の大好きなゲーム、「senran kagura」になってしまいます。

WEBHOOK

DiscordBotが429エラーを吐いた際に、メッセージを送信するWebHookです。
よっぽどのことでもない限り起きないので、設定しなくても問題ないです。

BOTS_NAME

DiscordのサーバーとLINEのグループを識別するための名前です。
カンマ区切りで設定し、環境変数の名前として使用します。
例として以下の様に設定した場合、

BOTS_NAME=a,b

次に示す環境変数の名前は

a_BOT_TOKEN
b_BOT_TOKEN

と設定する必要があります。

_BOT_TOKEN

LINEBotのアクセストークン。

_CHANNEL_SECRET

LINEBotのチャンネルシークレット。

_GROUP_ID

LINEのグループトークID。

_GUILD_ID

DiscordのサーバーID。

_CHANNEL_ID

Discordにメッセージを送信するテキストチャンネルのID。雑談とかにおすすめ。

_NG_CHANNEL

LINE側に送りたくないDiscordチャンネルの名前。BOTS_NAMEと同様にカンマ区切りで複数指定可能。

_NOTIFY_TOKEN

LINE Notifyのトークン。メッセージの送信に使用する。

main.py

Bot全体を起動させます。

main.py

from core.start import DBot
import discord
import os

from dotenv import load_dotenv
load_dotenv()

Token=os.environ['TOKEN']

# Bot立ち上げ
DBot(Token,discord.Intents.all()).run()

start.py

DiscordBotを起動します。

start.py

import discord
from discord.ext import tasks
import os
import datetime
import traceback
import requests,json

from dotenv import load_dotenv
load_dotenv()


class DBot(discord.AutoShardedBot):
    def __init__(self, token, intents):
        self.token = token
        super().__init__(intents = intents)
        self.load_cogs()

    async def on_ready(self):
        print('起動しました')
        game_name = os.environ.get('GAME_NAME')
        if game_name == None:
            game_name = 'senran kagura'
        await self.change_presence(activity = discord.Game(name = game_name))

    def load_cogs(self):
        for file in os.listdir("./cogs"): 
            if file.endswith(".py"): 
                cog = file[:-3] 
                self.load_extension(f"cogs.{cog}")
                print(cog + "をロードしました")

    # 起動用の補助関数です
    def run(self):
        try:
            self.loop.run_until_complete(self.start(self.token))
        except discord.LoginFailure:
            print("Discord Tokenが不正です")
        except KeyboardInterrupt:
            print("終了します")
            self.loop.run_until_complete(self.close())
        except discord.HTTPException as e:
            traceback.print_exc()
            if e.status == 429:
                main_content = {'content': 'DiscordBot 429エラー\n直ちにDockerファイルを再起動してください。'}
                headers      = {'Content-Type': 'application/json'}

                response     = requests.post(os.environ["WEBHOOK"], json.dumps(main_content), headers=headers)
                
        except:
            traceback.print_exc()

load_cogsについて解説

    def load_cogs(self):
        for file in os.listdir("./cogs"): 
            if file.endswith(".py"): 
                cog = file[:-3] 
                self.load_extension(f"cogs.{cog}")
                print(cog + "をロードしました")

for文でcogs下のファイルを一つ一つ参照します。
拡張子が「.py」の場合、「cogs.{ファイル名}」としてインポートします。

これによりcogs下のpythonファイルはDiscordBotの機能を含むpythonファイルとして適用されます。
次のmst_line.pyはDiscordのメッセージを受け取り、LINEに送る機能を書き込みます。

mst_line.py

Discordのメッセージを受け取り、LINEに送ります。

mst_line.py

import discord
from discord.ext import commands
import os
from typing import List,Tuple,Union

from dotenv import load_dotenv
load_dotenv()

from message_type.line_type.line_message import LineBotAPI
from core.start import DBot

class mst_line(commands.Cog):
    def __init__(self, bot : DBot):
        self.bot = bot

    # DiscordからLINEへ
    @commands.Cog.listener(name='on_message')
    async def on_message(self, message:discord.Message):

        # メッセージがbot、閲覧注意チャンネル、ピン止め、ボイスチャンネルの場合終了
        if (message.author.bot is True or
            message.channel.nsfw is True or
            message.type == discord.MessageType.pins_add or
            message.channel.type == discord.ChannelType.voice):
            return

        # FIVE_SECONDs,FIVE_HOUR
        # ACCESS_TOKEN,GUILD_ID,TEMPLE_ID (それぞれ最低限必要な環境変数)
        bots_name=os.environ['BOTS_NAME'].split(",")

        for bot_name in bots_name:
            # メッセージが送られたサーバーを探す
            if os.environ.get(f"{bot_name}_GUILD_ID") == str(message.guild.id):
                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')
                )
                break

        # line_bot_apiが定義されなかった場合、終了
        # 主な原因はLINEグループを作成していないサーバーからのメッセージ
        if not bool('line_bot_api' in locals()):
            return

        # 送信NGのチャンネル名の場合、終了
        ng_channel = os.environ.get(f"{bot_name}_NG_CHANNEL").split(",")
        if message.channel.name in ng_channel:
            return

        # テキストメッセージ
        messagetext=f"{message.channel.name}にて、{message.author.name}"

        if message.type == discord.MessageType.new_member:
            messagetext=f"{message.author.name}が参加しました。"

        if message.type == discord.MessageType.premium_guild_subscription:
            messagetext=f"{message.author.name}がサーバーブーストしました。"

        if message.type == discord.MessageType.premium_guild_tier_1:
            messagetext=f"{message.author.name}がサーバーブーストし、レベル1になりました！！！！！！！！"

        if message.type == discord.MessageType.premium_guild_tier_2:
            messagetext=f"{message.author.name}がサーバーブーストし、レベル2になりました！！！！！！！！！"

        if message.type == discord.MessageType.premium_guild_tier_3:
            messagetext=f"{message.author.name}がサーバーブーストし、レベル3になりました！！！！！！！！！！！"

        # メッセージをLINEに送信する
        await line_bot_api.push_message_notify(message=messagetext)

def setup(bot:DBot):
    return bot.add_cog(mst_line(bot))

ポイントとなる点

        bots_name=os.environ['BOTS_NAME'].split(",")

        for bot_name in bots_name:
            # メッセージが送られたサーバーを探す
            if os.environ.get(f"{bot_name}_GUILD_ID") == str(message.guild.id):
                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')
                )
                break

環境変数の項目でも言及した通り、os.environ['BOTS_NAME'].split(",")で名前を区切り、サーバーを分別します。
forで回し、os.environ.get(f"{bot_name}_GUILD_ID")で送信元のサーバーと一致した場合、LINEBotのクラスを宣言します。

        # 送信NGのチャンネル名の場合、終了
        ng_channel = os.environ.get(f"{bot_name}_NG_CHANNEL").split(",")
        if message.channel.name in ng_channel:
            return

上記と同様に、カンマでチャンネル名を区切ります。
送信元のチャンネル名がNGに該当する場合、送信せず終了します。

line_message.py

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()


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"]

いろいろ宣言していますが、使うのはpush_message_notifyのみです。

        # メッセージをLINEに送信する
        await line_bot_api.push_message_notify(message=messagetext)

これでLINEにメッセージを送れるようになります。

動作確認

Dockerから/bin/bash -c "cd app && python -u main.py"でDiscordBotを起動しましょう。
ローカル環境で問題ないです。

画像のようにDiscordBotがメッセージを読み取って、LINE Notifyにメッセージを送信できていれば完成です。

まとめ

DiscordからLINEへのテキストメッセージの送信が可能になりました。
ですが、

LINEからDiscordへのテキストメッセージ
LINEからDiscordへの画像
LINEからDiscordへの動画
LINEからDiscordへのスタンプ
DiscordからLINEへの画像、動画、スタンプ

が実装できていません。

次回は

LINEからDiscordへのテキストメッセージ

を説明します。

挨拶

注意

LINE Notifyのメリット

設計

DiscordからLINEへ

LINEからDiscordへ

下準備

コーディング

ディレクトリ構成

環境

環境変数

TOKEN

GAME_NAME

WEBHOOK

BOTS_NAME

_BOT_TOKEN

_CHANNEL_SECRET

_GROUP_ID

_GUILD_ID

_CHANNEL_ID

_NG_CHANNEL

_NOTIFY_TOKEN

main.py

start.py

mst_line.py

line_message.py

動作確認

まとめ

Discussion