DiscordとLINEをPython+FastAPI+Dockerで連携させる【番外編】音声ファイルの送受信

# 挨拶
こんにちは、マグロです。
前回、一通りのやり取りを実装しました。

https://qiita.com/maguro-alternative/items/65231cc5bce28fbac98c

これで一応完成ではあるのですが、今回は番外編として音声ファイルの送受信を行います。

# 背景
LINEにはボイスメッセージ機能が存在し、APIの方でも送受信に関する方法が記載されています。

送信

https://developers.line.biz/ja/reference/messaging-api/#audio-message

受信

https://developers.line.biz/ja/reference/messaging-api/#wh-audio

ですが、連携をする際に以下のような問題に直面しました。
- Discordのファイルアップロード上限は8MB(画像はGyazo,動画はYouTubeを使用して回避)
- LINEに音声を送信できるファイルは**m4aのみ**
- 音声を保存し、共有できるサービスが見当たらない

上記に加えて音声の送受信は携帯で録音して送るということが多く、PCユーザーの多いDiscordでは需要がないと考え、実装を見送っていました。

しかし2~3ヶ月ほど前、Discordにこんなアナウンスが流れました。

https://support.discord.com/hc/en-us/articles/13091096725527-Voice-Messages-

Discord側がモバイル向けに音声メッセージの機能を付け加えるとの声明が出ました。
API対応は執筆時点でまだされていないようですが、音声ファイルの送受信に需要があると判断したのでしょう。
風向きが変わり始めました。

そして、、、

https://twitter.com/discord/status/1645522780337885184

**ついにファイルの送信上限が25MBに拡張されることがアナウンスされました。**
動画ファイルはともかく、音声ファイルの送受信には十分な容量です。

というわけで、音声ファイルの送受信を実装します。

**※2024/9/6追記**

送信上限が10MBに縮小されました。

https://support.discord.com/hc/en-us/articles/115000435108-What-are-Nitro-Nitro-Basic#:~:text=Q%3A%20Why%20do%20I%20need%20Nitro%20to%20upload%20more%20than%2010MB%20of%20files%3F

# 設計
今までで一番の曲者です。
まずLINEに送る場合は音声ファイルを```m4a```に変換する必要があります。
音声ファイルの変換には```ffmpeg```を使用します。
第一回でDockerでインストールさせてあるため、特にすることはありません。
流れは以下の通りです。

Discord→LINE
- 前回と同様にDiscord側のファイル拡張子から音声ファイルを判別。
- ffmpegで```m4a```に変換。
- 変換したm4aファイルをDiscordにアップロード。
- アップロードしたファイルのURLを取得し、LINEBotで送信。

何度も言いますがLINEはファイルをバイナリデータとして扱います。
ファイルの種類は```image```,```video```,```audio```といった形でLINE側から送られてきますが、拡張子まではわかりません。
ファイルの送信にはDiscordAPIを直接使用してアップロードしますが、**拡張子を含めた**ファイル名を送信の際に求められます。

ではどうするかというと、バイナリデータから判別します。
ファイルのバイナリにはマジックナンバーと呼ばれるファイルの拡張子を示すものが存在します。
詳しく説明すると大きく話がそれるので、気になった方は調べてみてください。
流れは以下のようになります。

LINE→Discord
- ファイルのバイナリデータを取得。
- バイナリデータからマジックナンバーを識別、該当する拡張子をファイル名の末尾に置く。
- DiscordAPIを使用してファイルをアップロード。

# コーデイング
## ディレクトリ構成
```
$ 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のメッセージに関するクラス
│   │   ├── file_type.py
│   │   └── youtube_upload.py
│   ├── server.py       # サーバー立ち上げ
│   └── main.py  
├── Dockerfile
├── Profile
└── requirements.txt  
```

## file_type.py
```python:file_type.py
import io
import os

class Audio_Files:
    """
    ファイルのバイナリのデータを格納するクラス
    
    param:
    byte:bytes
    ファイルのバイトデータ

    filename:str
    ファイル名、拡張子は付けても付けなくてもいい

    付かなかった場合マジックナンバーから推測する
    """
    def __init__(
        self,
        byte:bytes,
        filename:str = None
    )-> None:
        """
        Discordのファイルの送信を行う際のクラス

        param:
        byte:bytes
        ファイルのバイナリデータ

        iobyte:io.BytesIO
        ファイルのバイナリデータ

        filename:str
        ファイル名、拡張子も付ける

        ついていない場合はマジックナンバーから推測する
        """
        self.byte = byte
        self.iobyte = io.BytesIO(byte)
        if len(os.path.splitext(filename)[1]) == 0:
            extension = self.detect_audio_file()
            self.filename = filename + extension.capitalize()
        else:
            self.filename = filename
        
    def detect_audio_file(self) -> str:
        """
        バイナリデータのマジックナンバーから音声ファイルの拡張子を識別する。

        param:
        file_byte:io.BytesIO
        ファイルのバイナリデータ

        return
        拡張子の文字列:str
        """
        header = self.iobyte.read(12)
            
        # AIFFファイルのマジックナンバー
        if header.startswith(b'FORM') and header[8:12] == b'AIFF':
            return '.aiff'
            
        # AIFF-Cファイルのマジックナンバー
        if header.startswith(b'FORM') and header[8:12] == b'AIFC':
            return '.aifc'
            
        # WAVEファイルのマジックナンバー
        if header.startswith(b'RIFF') and header[8:12] == b'WAVE':
            return '.wav'
            
        # MP3ファイルのマジックナンバー
        if header.startswith(b'\xFF\xFB') or header.startswith(b'\xFF\xF3') or \
        header.startswith(b'\xFF\xF2') or header.startswith(b'\xFF\xF4'):
            return '.mp3'

        # FLACファイルのマジックナンバー
        if header.startswith(b'fLaC'):
            return '.flac'

        # OGG Vorbisファイルのマジックナンバー
        if header.startswith(b'OggS') and header[28:31] == b'vorb':
            return '.ogg'

        # AACファイルのマジックナンバー
        if header.startswith(b'\xFF\xF1') or header.startswith(b'\xFF\xF9'):
            return '.aac'

        # AC-3ファイルのマジックナンバー
        if header.startswith(b'\x0B\x77') or header.startswith(b'\x77\x0B'):
            return '.ac3'

        # AMRファイルのマジックナンバー
        if header.startswith(b'#!AMR'):
            return '.amr'

        # GSMファイルのマジックナンバー
        if header.startswith(b'\x00\x01\x00\x01'):
            return '.gsm'
        
        # M4Aのマジックナンバー(LINEボイスメッセージの標準規格)
        if header.startswith(b'\x00\x00\x00\x1c') and header[8:12] == b'M4A ':
            return '.m4a'

```

音声ファイルのマジックナンバーは先頭の数バイトから識別することができます。
マジックナンバーの詳細は省きます。

## message_creater.py
```diff_python:message_creater.py
import os
import re
+ import io

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
+ from message_type.file_type import Audio_Files

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',
        }
+       # ファイルアップロードの際にContent-Typeが邪魔になるので取り除く
+       self.no_content_headers = {
+           'Authorization': f'Bot {token}'
+       }

    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 def send_discord_file(
+       self, 
+       channel_id: int, 
+       message: str, 
+       fileobj:Audio_Files
+   ) -> Dict:
+       """
+       Discordへファイル付きのメッセージを送信する。
+       
+       channel_id  :int
+           Discordのテキストチャンネルのid
+       message     :str
+           テキストメッセージ
+       fileobj     :Audio_Files
+           音声ファイルのオブジェクト
+       """
+
+       with aiohttp.MultipartWriter("form-data") as mpwriter:
+           # ファイルを送付
+           mpwriter.append(
+               obj=io.BytesIO(fileobj.byte)
+           ).set_content_disposition(
+               disptype='form-data', 
+               name=fileobj.filename, 
+               filename=fileobj.filename
+           )
+
+           # テキストメッセージを送付
+           mpwriter.append(
+               obj=message
+           ).set_content_disposition(
+               disptype='form-data',
+               name="content"
+           )
+
+           # Discordにファイルとメッセージを送信
+           # 'Content-Type': 'application/x-www-form-urlencoded'が邪魔なので取り除く
+           async with aiohttp.ClientSession() as session:
+               async with session.post(
+                   url = f'{DISCORD_BASE_URL}/channels/{channel_id}/messages',
+                   headers = self.no_content_headers,
+                   data = mpwriter
+               ) as resp:
+                   return await resp.json()
```
ポイント
```python
        with aiohttp.MultipartWriter("form-data") as mpwriter:
            # ファイルを送付
            mpwriter.append(
                obj=io.BytesIO(fileobj.byte)
            ).set_content_disposition(
                disptype='form-data', 
                name=fileobj.filename, 
                filename=fileobj.filename
            )

            # テキストメッセージを送付
            mpwriter.append(
                obj=message
            ).set_content_disposition(
                disptype='form-data',
                name="content"
            )
```
```with aiohttp.MultipartWriter("form-data") as mpwriter:```で非同期でフォームデータの作成をしています。
```obj=fileobj.iobyte```はファイルのバイナリデータで、```set_content_disposition```で関連する情報を追加しています。```name```,```filename```にはファイル名(拡張子含む)を入れます。

こうすることでバイナリデータのまま、Discordに音声ファイルを送信できます。

テキストメッセージは```obj=message```にテキストを入れ、```set_content_disposition```で```name="content"```と指定することでテキストを送付できます。

```python
            # Discordにファイルとメッセージを送信
            # 'Content-Type': 'application/x-www-form-urlencoded'が邪魔なので取り除く
            async with aiohttp.ClientSession() as session:
                async with session.post(
                    url = f'{DISCORD_BASE_URL}/channels/{channel_id}/messages',
                    headers = self.no_content_headers,
                    data = mpwriter
                ) as resp:
                    return await resp.json()
```
aiohttpで非同期でメッセージを送信します。
```application/x-www-form-urlencoded```はテキストを送信する形式なのでファイル送信には使用できません。
```content-type```なしで送信させます。

## line_message.py
```diff_python
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
from message_type.youtube_upload import YouTubeUpload
+ from message_type.file_type import Audio_Files

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

+ class Voice_File:
+   """
+   Discordの音声ファイルのURLと秒数を格納するクラス
+
+   param:
+   url     :str
+   音声ファイルのURL
+
+   second  :float
+   音声ファイルの秒数(秒)
+   """
+   def __init__(self,url:str,second:float) -> None:
+       self.url = url
+       self.second = second


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

    
    # LINE Notifyで画像を送信
    async def push_image_notify(self, message: str, image_url: str) -> Dict:
        """
        LINE Notifyで画像を送信

        param
        message:str
        送信するテキストメッセージ

        image_url:str
        送信する画像のURL

        return
        resp.json:json
        レスポンス
        """
        if len(message) == 0:
            message = "画像を送信しました。"
        data = {
            'imageThumbnail': f'{image_url}',
            'imageFullsize': f'{image_url}',
            'message': f'{message}',
        }
        return await line_post_request(
            url = NOTIFY_URL, 
            headers = {'Authorization': f'Bearer {self.notify_token}'}, 
            data = data
        )

    # 動画の送信(動画のみ)
    async def push_movie(self, preview_image: str, movie_urls: List[str]) -> Dict:
        """
        LINEBotで動画の送信(動画のみ)

        param
        preview_image:str
        プレビュー画像のURL

        movie_urls:List[str]
        動画のURL(複数)

        return
        resp.json:json
        レスポンス
        """
        data = []
        # 動画を1個ずつ格納
        for movie_url in movie_urls:
            data.append({
                "type": "video",
                "originalContentUrl": movie_url,
                "previewImageUrl": preview_image
            })
        datas = {
            "to": self.line_group_id,
            "messages": data
        }
        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(datas)
        )
    
+   async def push_voice(self,voice_file:List[Voice_File]) -> Dict:
+       """
+       LINEBotで音声の送信
+
+       param
+       voice_file:List[Voice_File]
+       送信する音声ファイルのクラス
+
+       return
+       Dict
+       レスポンス
+       """
+       data = []
+       for voice in voice_file:
+           data.append({
+               'type':'audio',
+               'originalContentUrl':voice.url,
+               'duration':int(voice.second * 1000)
+           })
+       datas = {
+           'to': self.line_group_id,
+           'messages': data
+       }
+       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(datas)
+       )

    # 送ったメッセージ数を取得
    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)

    
    # LINEのユーザプロフィールから名前を取得
    async def get_proflie(self, user_id: str) -> Profile:
        """
        LINEのユーザプロフィールから名前を取得

        param
        user_id:str
        LINEのユーザーid

        return
        profile:Profile
        LINEユーザーのプロフィールオブジェクト
        """
        # グループIDが有効かどうか判断
        
        r = await line_get_request(
            LINE_BOT_URL + f"/group/{self.line_group_id}/member/{user_id}",
            self.line_bot_token,
        )
        
        # グループIDが無効の場合、友達から判断
        if r.get('message') != None:
            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:
        """
        LINEから画像データを取得し、Gyazoにアップロードする

        param
        message_id:int
        LINEのメッセージのid

        return
        gayzo:GyazoJson
        Gyazoの画像のオブジェクト
        """
        # 画像のバイナリデータを取得
        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())
        
    # LINEから受け取った動画を保存し、YouTubeに限定公開でアップロード
    async def movie_upload(self, message_id: int, display_name: str) -> str:
        """
        LINEから受け取った動画を保存し、YouTubeに限定公開でアップロード

        param
        message_id:int
        LINEのメッセージのid

        display_name:str
        LINEのユーザー名

        return
        youtube_id:str
        YouTubeの動画id
        """
        # 動画のバイナリデータを取得
        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:

                video_bytes = await bytes.read()

                youtube_video = YouTubeUpload(
                    title=f"{display_name}からの動画",
                    description="LINEからの動画",
                    privacy_status="unlisted"
                )

                youtube = await youtube_video.get_authenticated_service()

                return await youtube_video.byte_upload(
                    video_bytes=io.BytesIO(video_bytes),
                    youtube=youtube
                )

+   # LINEから受け取った音声データを取得し、Discordにアップロード
+   async def voice_get(self ,message_id: int) -> Audio_Files:
+       """
+       LINEから受け取った音声データを取得し、Discordにアップロード
+
+       param:
+       message_id:int
+       LINEのメッセージのid
+
+       return
+       Audio_File
+       アップロードする音声データのクラス
+       """
+       # 音声のバイナリデータを取得
+       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:
+
+               voice_bytes = await bytes.read()
+
+               # アップロードするファイルを指定する
+               return Audio_Files(
+                   byte=voice_bytes,
+                   filename='line_audio'
+               )

    # 当月に送信できるメッセージ数の上限目安を取得(基本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"]
```
ポイント
```python
class Voice_File:
    """
    Discordの音声ファイルのURLと秒数を格納するクラス

    param:
    url     :str
    音声ファイルのURL

    second  :float
    音声ファイルの秒数(秒)
    """
    def __init__(self,url:str,second:float) -> None:
        self.url = url
        self.second = second
```
Discordの音声ファイルのURLと秒数を格納するクラスです。
詳細は後述。
```python
    async def push_voice(self,voice_file:List[Voice_File]) -> Dict:
        """
        LINEBotで音声の送信

        param
        voice_file:List[Voice_File]
        送信する音声ファイルのクラス

        return
        Dict
        レスポンス
        """
        data = []
        for voice in voice_file:
            data.append({
                'type':'audio',
                'originalContentUrl':voice.url,
                'duration':int(voice.second * 1000)
            })
        datas = {
            'to': self.line_group_id,
            'messages': data
        }
        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(datas)
        )
```
LINEに音声を送信させます。
動画と同じく、リストで複数の音声を送信できます。
```'duration':int(voice.second * 1000)```は音声の長さ(ミリ秒)を示しています。
これがないと送信できないようです。
```python
    # LINEから受け取った音声データを取得し、Discordにアップロード
    async def voice_get(self ,message_id: int) -> Audio_Files:
        """
        LINEから受け取った音声データを取得し、Discordにアップロード

        param:
        message_id:int
        LINEのメッセージのid

        return
        Audio_File
        アップロードする音声データのクラス
        """
        # 音声のバイナリデータを取得
        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:

                voice_bytes = await bytes.read()

                # アップロードするファイルを指定する
                return Audio_Files(
                    byte=voice_bytes,
                    filename='line_audio'
                )
```
LINEから受け取った音声データを取得し、```Audio_Files```としてインスタンスを作ります。

## mst_line.py
```diff_python: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に送信する動画、画像、音声ファイルのリスト
        imagelist = []
        videolist = []
+       voicelist = []

        # ユーザーネームの空白文字を削除
        user_name = re.sub("[\u3000 \t]", "",message.author.name)


        # 送付ファイルがあった場合
        if message.attachments:
            # 画像か動画であるかをチェック
            imagelist, message.attachments = await image_checker(message.attachments)
            videolist, message.attachments = await video_checker(message.attachments)
+           voicelist, message.attachments = await voice_checker(message.attachments,message)

            messagetext += "が、"

            # 送信された動画と画像の数を格納
            if len(imagelist) > 0:
                messagetext += f"画像を{len(imagelist)}枚、"

            if len(videolist) > 0:
                messagetext += f"動画を{len(videolist)}個、"

+           if len(voicelist) > 0:
+               messagetext += f"音声を{len(voicelist)}個、"

            # 画像と動画以外のファイルがある場合、urlを直接書き込む
            if message.attachments:
                for attachment in message.attachments:
                    messagetext += f"\n{attachment.url} "

            messagetext += "送信しました。"

        # メッセージ本文を書き込む
        messagetext += f"「 {message.clean_content} 」"

        # スタンプが送付されている場合
        if message.stickers:
            # 動くスタンプは送信不可のため終了
            if message.stickers[0].url.endswith(".json"):
                return
            # 画像として送信
            else:
                messagetext = f'{messagetext} スタンプ:{message.stickers[0].name}'
                imagelist, message.stickers = await image_checker(message.stickers)

        # 画像を一個ずつNotifyで送信
        if len(imagelist) > 0:
            for image in imagelist:
                await line_bot_api.push_image_notify(message=messagetext,image_url=image)

        # 動画を送信
        if len(videolist) > 0:
            if hasattr(message.guild.icon,'url'):
                icon_url = message.guild.icon.url
            else:
                icon_url = message.author.display_avatar.url

            await line_bot_api.push_message_notify(message=messagetext)
            await line_bot_api.push_movie(preview_image=icon_url,movie_urls=videolist)

+       # 音声を送信
+       if len(voicelist) > 0:
+           await line_bot_api.push_message_notify(message=messagetext)
+           await line_bot_api.push_voice(voice_file=voicelist)

        # ファイルなしの場合、テキストを送信
-       if len(imagelist) + len(videolist) == 0:
+       if len(imagelist) + len(videolist) + len(voicelist) == 0:
            await line_bot_api.push_message_notify(message=messagetext)



# 画像を識別
async def image_checker(
    attachments:List[discord.Attachment]
) -> Tuple[
    List[str],
    Union[List[discord.Attachment],List[discord.StickerItem]]
]:
    """
    Discordの送付ファイルから、画像を抜き出す。
    引数:      attachments:    Discordの送付ファイル
    戻り値:    image_urls:     画像かスタンプのurlリスト
               attachments:    画像を抜き出したDiscordの送付ファイル
    """
    image = (".jpg", ".png", ".JPG", ".PNG", ".jpeg", ".gif", ".GIF")
    image_urls = []
    for attachment in attachments[:]:
        # 画像があった場合、urlを画像のリストに追加し、送付ファイルのリストから削除
        if attachment.url.endswith(image):
            image_urls.append(attachment.url)
            attachments.remove(attachment)

    return image_urls, attachments

# 動画を識別
async def video_checker(
    attachments:List[discord.Attachment]
) -> Tuple[
    List[str],
    List[discord.Attachment]
]:
    """
    Discordの送付ファイルから、動画を抜き出す。
    引数:      attachments:    Discordの送付ファイル
    戻り値:    video_urls:     動画のurlリスト
               attachments:    動画を抜き出したDiscordの送付ファイル
    """
    video = (".mp4", ".MP4", ".MOV", ".mov", ".mpg", ".avi", ".wmv")
    video_urls = []
    for attachment in attachments[:]:
        # 動画があった場合、urlを動画のリストに追加し、送付ファイルのリストから削除
        if attachment.url.endswith(video):
            video_urls.append(attachment.url)
            attachments.remove(attachment)

    return video_urls, attachments

+ # 音声を識別
+ async def voice_checker(
+   attachments:List[discord.Attachment],
+   message:discord.Message
+) -> Tuple[
+   List[Voice_File],
+   List[discord.Attachment]
+]:
+   """
+   Discordの送付ファイルから、音声を抜き出す。
+   m4a以外のファイルは、ffmpegで変換しDiscordに送信する。
+   引数:      attachments:    Discordの送付ファイル
+   戻り値:    video_urls:     音声のurlリスト
+              attachments:    音声を抜き出したDiscordの送付ファイル
+   """
+   voice = (".wav",".mp3",".flac",".aif",".m4a",".oga",".ogg")
+   voice_files = []
+   loop = asyncio.get_event_loop()
+   for attachment in attachments[:]:
+       # 動画があった場合、urlを動画のリストに追加し、送付ファイルのリストから削除
+       if attachment.url.endswith(voice):
+           # 音声ファイルをダウンロードする
+           await attachment.save(attachment.filename)
+
+           # m4aの場合はそのまま格納
+           if attachment.url.endswith('.m4a'):
+               voice_url = attachment.url
+               attachments.remove(attachment)
+           else:
+               # ffmpegを使用して音声ファイルをm4aに変換する
+               output_filename = f"{os.path.splitext(attachment.filename)[0]}.m4a"
+               await loop.run_in_executor(
+                   None,
+                   partial(
+                       subprocess.run,['ffmpeg', '-i', attachment.filename, output_filename],
+                       check=True
+                   )
+               )
+
+               # aiofilesで非同期でファイルを開く
+               async with aiofiles.open(output_filename, 'rb') as f:
+                   m4a_data = await f.read()
+                   # Discordにファイルを送信する
+                   m4a_file = discord.File(
+                       fp=io.BytesIO(m4a_data), 
+                       filename=output_filename
+                   )
+                   m4a_file_message = await message.channel.send(f"m4aファイルに変換します。: {attachment.filename} -> {output_filename}", file=m4a_file)
+
+               voice_url = m4a_file_message.attachments[0].url
+               # 変換したm4aファイルのurl
+               attachments.remove(attachment)
+
+           # m4aファイルの秒数を計算
+           ogg_sound = AudioSegment.from_file(output_filename,format="m4a")
+           sound_second = ogg_sound.duration_seconds
+
+           voice_files.append(Voice_File(
+               url=voice_url,
+               second=sound_second
+           ))
+
+           # 変換したファイルを削除する
+           os.remove(output_filename)
+           os.remove(attachment.filename)
+
+   return voice_files, attachments


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

```
ポイント
```python
# 音声を識別
async def voice_checker(
    attachments:List[discord.Attachment],
    message:discord.Message
) -> Tuple[
    List[Voice_File],
    List[discord.Attachment]
]:
    """
    Discordの送付ファイルから、音声を抜き出す。
    m4a以外のファイルは、ffmpegで変換しDiscordに送信する。
    引数:      attachments:    Discordの送付ファイル
    戻り値:    video_urls:     音声のurlリスト
               attachments:    音声を抜き出したDiscordの送付ファイル
    """
    voice = (".wav",".mp3",".flac",".aif",".m4a",".oga",".ogg")
    voice_files = []
    loop = asyncio.get_event_loop()
    for attachment in attachments[:]:
        # 動画があった場合、urlを動画のリストに追加し、送付ファイルのリストから削除
        if attachment.url.endswith(voice):
            # 音声ファイルをダウンロードする
            await attachment.save(attachment.filename)
            
            # m4aの場合はそのまま格納
            if attachment.url.endswith('.m4a'):
                voice_url = attachment.url
                attachments.remove(attachment)
            else:
                # ffmpegを使用して音声ファイルをm4aに変換する
                output_filename = f"{os.path.splitext(attachment.filename)[0]}.m4a"
                await loop.run_in_executor(
                    None,
                    partial(
                        subprocess.run,['ffmpeg', '-i', attachment.filename, output_filename],
                        check=True
                    )
                )

                # aiofilesで非同期でファイルを開く
                async with aiofiles.open(output_filename, 'rb') as f:
                    m4a_data = await f.read()
                    # Discordにファイルを送信する
                    m4a_file = discord.File(
                        fp=io.BytesIO(m4a_data), 
                        filename=output_filename
                    )
                    m4a_file_message = await message.channel.send(f"m4aファイルに変換します。: {attachment.filename} -> {output_filename}", file=m4a_file)

                voice_url = m4a_file_message.attachments[0].url
                # 変換したoggファイルのurl
                attachments.remove(attachment)

            # m4aファイルの秒数を計算
            ogg_sound = AudioSegment.from_file(output_filename,format="m4a")
            sound_second = ogg_sound.duration_seconds

            voice_files.append(Voice_File(
                url=voice_url,
                second=sound_second
            ))

            # 変換したファイルを削除する
            os.remove(output_filename)
            os.remove(attachment.filename)

    return voice_files, attachments
```
基本的には```image_checker```と```video_checker```と同じく、拡張子から音声ファイルかどうかを認識していますが、音声ファイルはいったんローカルに保存します。
拡張子が```m4a```の場合はそのまま格納しますが、それ以外の場合は```ffmpeg```で```m4a```に変換します。

変換後、Discord上にm4aファイルをアップロードします。

```m4a_file_message```にアップロードされたファイルのurlがあるのでそれを取得します。
pydubを使用しダウンロード、もしくは変換したm4aの秒数を計測した後、urlとその秒数をもとに```Voice_File```を定義し格納します。

これでLINEへ音声ファイルを送信する準備が整います。

## server.py
```diff_python: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}"


    # 動画が送信された場合
    if event.message.type == 'video':
        # 動画をYouTubeにアップロードし、urlを返す
        youtube_id = await line_bot_api.movie_upload(
            message_id=event.message.id,
            display_name=profile_name.display_name
        )
        message = f"https://youtu.be/{youtube_id}"

+   # 音声が送信された場合
+   if event.message.type == 'audio':
+       # 音声ファイルのデータを取得し、Discordに送信
+       fileobj = await line_bot_api.voice_get(message_id=event.message.id)
+       await discord_find_message.send_discord_file(
+           channel_id=channel_id,
+           message=f'{profile_name.display_name}',
+           fileobj=fileobj
+       )
+       # レスポンス200を返し終了
+       return HTMLResponse(content="OK")


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

```

# 実行
以下のように送受信できていれば成功です。
念のため音声が聞こえるかも確認しましょう。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2853914/a9545660-95fa-0945-d69b-285f7c67d8b8.png)
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2853914/c4e4dc8d-826c-9279-781d-b757d4015a1e.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2853914/a1b0e810-205b-d66c-e49b-45a02da1bff6.png)



# 終わりに
全5+1回に及ぶDiscordとLINE連携もこれで終わりです。本当にお疲れ様でした。
音声も追加したことで、よりコミュニケーション表現の幅が広くなったといえます。

本Botはこれにいくつか機能を追加して身内で使用しているのですが、改良を重ねており、本記事のLINE連携をWebUI上で制御できるようにしています。
以下のように実装しています。

https://qiita.com/maguro-alternative/items/f7eec5f11bf888328cd6

https://github.com/maguro-alternative/discordfast

FastAPIとjinja2を使用してフルスタックアプリとして運用していますが、フロントと分けた方がよいと考えたためまた改良しようと考えています。

完成して、やる気があればまた記事にしたいと思います。

ここまで読んでいただきありがとうございます。