LoginSignup
6
0
お題は不問!Qiita Engineer Festa 2023で記事投稿!

DiscordとLINEをPython+FastAPI+Dockerで連携させる【その5】LINEからDiscordへのスタンプ&DiscordからLINEへの画像、動画、スタンプ

Last updated at Posted at 2023-06-18

はじめに

こんにちは。マグロです。
前回の続きです。

今回は

  • LINEからDiscordへのスタンプ
  • DiscordからLINEへの画像、動画、スタンプ

を説明します。

LINEスタンプ

LINEの代表的なコミュニケーション手段ですね。
やはりLINEとの連携なのでどうしても実装したいですね。

というわけで、LINEスタンプの画像をどうにかしてDiscordに持っていきましょう。

LINEストアにアクセスします。

スタンプのページに遷移して「Ctrl + Shift + I」を押します。
そしてスタンプの要素を見てみましょう。

image.png

スタンプ画像の該当箇所に
<span class="mdCMN09Image" style="background-image:url(https://stickershop.line-scdn.net/stickershop/v1/sticker/331531470/android/sticker.png?v=1);"></span>
とあります。

アクセスしてみると、、、、
image.png

png画像で出てきた!!!!
どうにか持ってこれないか!??!

URLの内容を調べます。
https://stickershop.line-scdn.net/stickershop/v1/sticker/331531470/android/sticker.png
どうやら331531470はLINEスタンプのIDらしいです。
LINE公式のリファレンスによると、スタンプが送信されるとそのイベントの中にIDが含まれるそうです。

URLのID部分に、受け取ったスタンプのIDを当てはめればよさそうです。

またandroidの部分はiPhoneにもできます。
androidの場合はpngのサイズが200x200で、iPhoneの場合は100x100になるそうです。

android一択!!と思っていたのですが、
どうやら古いスタンプの画像には対応していないらしく、404エラーが出てくるスタンプ画像もありました。
なのでiPhoneにします。

というわけで、このスタンプURLをそのままDiscordに送信します。

余談ですが、動くスタンプはcanvasタグで動かしているらしく、動きまでは持ってこれなさそうです。
ですがpngであることに変わりなく、LINEでのプレビュー通りに持ってこれそうです。

コーデイング(LINEスタンプ)

ディレクトリ構成

$ 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のメッセージに関するクラス
│   │   └── youtube_upload.py
│   ├── server.py       # サーバー立ち上げ
│   └── main.py  
├── Dockerfile
├── Profile
└── requirements.txt  

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)

+   # スタンプが送信された場合
+   if event.message.type == 'sticker':
+       # スタンプのurlを代入
+       message = f"https://stickershop.line-scdn.net/stickershop/v1/sticker/{event.message.stickerId}/iPhone/sticker_key@2x.png"

    # 画像が送信された場合
    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}"

    # 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 == 'sticker':
        # スタンプのurlを代入
        message = f"https://stickershop.line-scdn.net/stickershop/v1/sticker/{event.message.stickerId}/iPhone/sticker_key@2x.png"

event.message.typestickerの場合、スタンプが送信されたことになるので、
event.message.stickerIdからスタンプIDを取り出します。
スタンプIDをURLに当てはめ、それをテキストメッセージとしてDiscordに送信します。

画像の時にも解説しましたが、Discordには画像のプレビュー機能があるため、URLを送るだけで画像が見れます。

これでスタンプは実装出来ました。実行して確認してみましょう。

実行(LINEスタンプ)

画像のようにDiscordにプレビューされていれば成功です。
image.png

DiscordからLINEへの画像、動画、スタンプ

LINEのスタンプ実装は以上です。
ただあまりにも短いのでついでにLINEへの画像、動画、スタンプも実装しちゃいます。

DiscordはLINEとは違いCDN方式でファイルコンテンツを保存しているため、ファイルをhttps形式で参照することができます。

またLINEBotは、画像や動画などのファイルを送信する場合はhttps形式で送付する必要があるため、このDIscordのファイルURLを送付するだけで送信できます。

画像はLINE Notify、動画はLINEBotで送信させます。

コーデイング(LINEへの画像、動画、スタンプ)

line_message.py

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
from message_type.youtube_upload import YouTubeUpload

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

    # 当月に送信できるメッセージ数の上限目安を取得(基本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"]

解説は後述。

mst_line.py

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 = []
+
+       # ユーザーネームの空白文字を削除
+       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)
+
+           messagetext += "が、"
+
+           # 送信された動画と画像の数を格納
+           if len(imagelist) > 0:
+               messagetext += f"画像を{len(imagelist)}枚、"
+
+           if len(videolist) > 0:
+               messagetext += f"動画を{len(videolist)}個、"
+
+
+           # 画像と動画以外のファイルがある場合、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(imagelist) + len(videolist) == 0:
+           await line_bot_api.push_message_notify(message=messagetext)
+
-       # メッセージをLINEに送信する
-       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

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

ポイント

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

message.attachmentsにはDiscordのメッセージに含まれているファイルの情報が、配列で格納されています。
どんなファイルが入っているか確認するため、image_checkervideo_checkerでチェックを行います。

# 画像を識別
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

imageにはタプル形式で画像ファイルの拡張子が入っています。
ファイルのURLの末尾に拡張子があるので、そこから画像かどうか判別します。

画像ファイルと判別した場合、image_urlsに格納し、attachmentsから削除します。
動画も同様です。

        # スタンプが送付されている場合
        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)

message.stickersにはmessage.attachments同様、スタンプの情報が配列で格納されています。
が、Discordは一度に送られるスタンプは1個までなので、最初の配列を参照するだけで済みます。

スタンプは画像なので、image_checkerで画像として送信させるためにimagelistに格納させます。

LINE同様、動くスタンプもありますが、どうやらjsonで動きを設定しているらしく(おそらくAPNG?)、送信不可なので終了させます。テキストやファイルを同時に送付することもできないようなのでreturnさせてます。

        # 画像を一個ずつ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)

NotifyはLINEBotのように一回に複数画像を送信させることはできないので、画像の数だけメッセージを送信します。
Notifyは動画を扱えないので、LINEBotに送信させます。

また、動画を送信する際、動画のプレビュー画像(いわゆるサムネ)を用意する必要があります。
サーバーアイコンをサムネイルにしていますが、サーバーアイコンが設定されていない場合はユーザアイコンがサムネイルになります。

実行(LINEへの画像、動画、スタンプ)

画像、動画をDiscordに送信してみましょう。
image.png
image.png

画像のように送信されていれば成功です。

終わりに

はい!これでDiscordとLINEの連携は完成です!!
お疲れ様でした!!

それと更新が遅れて申し訳ありませんでした。就活が忙しくてモチベーションダダ下がりだったのですが、先月参加した技育博での反応がいいモチベーションになりました。ありがとうございます。

また、前の記事でもお話しした通り、音声ファイルの送受信も可能になりました。
それはまた別途記事を書こうかと思います。

それではまた次の記事で会いましょう!!

6
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
6
0