📂

DiscordとLINEをPython+FastAPI+Dockerで連携させる【その3】LINEからDiscordへの画像

2023/02/11に公開

挨拶

こんにちは。マグロです。
こちらの続きとなります。

https://zenn.dev/maguro_alterna/articles/65848d6f96ed30

今回はLINEから画像をDiscordに送ります。

背景

こちらの記事の通り、LINEはアップロードされたファイルはバイナリデータでサーバーに保存され、一定時間で削除されます。

https://zenn.dev/maguro_alterna/articles/21706a293a8eeb

今回は上記の記事同様、Gyazoを使用してDiscord側に共有させます。

必須ではありませんが、記事の内容をやっておくと、流れを理解できて本稿の内容もすんなりとできると思います。(というかほぼまんま)

下準備

GyazoAPIキーの取得

GyazoAPIを使用するため、アカウント登録とAPIキーの取得をしておきましょう。

https://gyazo.com/oauth/applications

New Applicationを選択
image.png

名前とコールバックURLを指定します。
コールバックは使用しないの、名前と同様自由に決めてください。
image.png
submitを押すと、トークンが発行されます。
コイツを控えておきましょう。
image.png

環境変数

以下の環境変数を追加します。

GYAZO_TOKEN

Gyazoのトークン。
画像のアップロードに使用。

コーディング

ディレクトリ構成

$ tree
.
├── app
│   ├── cogs
│   │   └── mst_line.py  # DiscordからLINEへ
│   ├── core
│   │   └── start.py     # DiscordBot起動用
│   ├── message_type
│   │   ├── discord_type
│   │   │   ├── discord_type.py     # Discordのサーバーに関するクラス
│   │   │   └── message_creater.py  # DiscordAPIを直接叩く
│   │   └── line_type
│   │       ├── line_type.py        # LINEのプロフィールなどのクラス
│   │       ├── line_event.py       # LINEのイベントに関するクラス
│   │       └── line_message.py     # LINEのメッセージに関するクラス
│   ├── server.py       # サーバー立ち上げ
│   └── main.py  
├── Dockerfile
├── Profile
└── requirements.txt  

line_type.py

Gyazoのレスポンスもjsonとして受け取るため、前回と同様にクラスとして受け取ります。

line_type.py
import re
import json

class Base(object):
    def __init__(self, **kwargs):
        """__init__ method.
        :param kwargs:
        """
        pass

    def __str__(self):
        """__str__ method."""
        return self.as_json_string()

    def __repr__(self):
        """__repr__ method."""
        return str(self)

    def __eq__(self, other):
        """__eq__ method.
        :param other:
        """
        return other and self.as_json_dict() == other.as_json_dict()

    def __ne__(self, other):
        """__ne__ method.
        :param other:
        """
        return not self.__eq__(other)

    async def as_json_string(self):
        """jsonの文字列を返します。 
        :rtype: str
        """
        return json.dumps(self.as_json_dict(), sort_keys=True)

    async def as_json_dict(self):
        """このオブジェクトから辞書型を返します。
        :return: dict
        """
        data = {}
        for key, value in self.__dict__.items():
            camel_key = to_camel_case(key)
            if isinstance(value, (list, tuple, set)):
                data[camel_key] = list()
                for item in value:
                    if hasattr(item, 'as_json_dict'):
                        data[camel_key].append(item.as_json_dict())
                    else:
                        data[camel_key].append(item)

            elif hasattr(value, 'as_json_dict'):
                data[camel_key] = value.as_json_dict()
            elif value is not None:
                data[camel_key] = value

        return data

    @classmethod
    async def new_from_json_dict(cls, data:dict):
        """dict から新しいインスタンスを作成します。
        :param data: JSONのディクショナリ
        """
        new_data = {await to_snake_case(key): value
                    for key, value in data.items()}

        return cls(**new_data)


async def to_snake_case(text:str):
    """スネークケースに変換する。
    :param str text:
    :rtype: str
    """
    s1 = re.sub('(.)([A-Z])', r'\1_\2', text)
    s2 = re.sub('(.)([0-9]+)', r'\1_\2', s1)
    s3 = re.sub('([0-9])([a-z])', r'\1_\2', s2)
    return s3.lower()

async def to_camel_case(text:str):
    """キャメルケースに変換する。
    :param str text:
    :rtype: str
    """
    split = text.split('_')
    return split[0] + "".join(x.title() for x in split[1:])

class Profile(Base):
    """
    LINE Message APIのProfileクラス
    user_id         :LINEユーザーのid
    display_name    :LINEのユーザー名
    picture_url     :LINEのアイコンurl
    status_message  :LINEのプロフィール文
    """
    def __init__(self,
        user_id:str = None,
        display_name:str = None,
        picture_url:str = None,
        status_message:str = None,
        **kwargs
    ):
        super(Profile, self).__init__(**kwargs)
        self.user_id = user_id
        self.display_name = display_name
        self.picture_url = picture_url
        self.status_message = status_message


+ class GyazoJson(Base):
+     """
+     Gyazoの画像クラス
+     image_id        :画像id
+     permalink_url   :画像のパーマリンク
+     thumb_url       :サムネイル画像url
+     url             :画像url
+     type            :拡張子のタイプ
+     """
+     def __init__(self,
+                  image_id:str=None,
+                  permalink_url:str=None,
+                  thumb_url:str=None,
+                  url:str=None,
+                  type:str=None,
+                  **kwargs
+     ):
+         self.image_id = image_id
+         self.premalink_url = permalink_url
+         self.thumb_url = thumb_url
+         self.url = url
+         self.type = type
+         super(GyazoJson,self).__init__(**kwargs)

line_message.py

LINEから画像のバイナリデータを受け取り、Gyazoに送信する処理を追加します。

line_message.py
import json
import requests
from requests import Response

import datetime

import os
import asyncio
from functools import partial

import aiohttp
import subprocess
from typing import List

from dotenv import load_dotenv
load_dotenv()

+ from message_type.line_type.line_type import Profile,GyazoJson

NOTIFY_URL = 'https://notify-api.line.me/api/notify'
NOTIFY_STATUS_URL = 'https://notify-api.line.me/api/status'
LINE_BOT_URL = 'https://api.line.me/v2/bot'
LINE_CONTENT_URL = 'https://api-data.line.me/v2/bot'

# LINEのgetリクエストを行う
async def line_get_request(url: str, token: str) -> json:
    async with aiohttp.ClientSession() as session:
        async with session.get(
            url = url,
            headers = {'Authorization': 'Bearer ' + token}
        ) as resp:
            return await resp.json()

# LINEのpostリクエストを行う
async def line_post_request(url: str, headers: dict, data: dict) -> json:
    async with aiohttp.ClientSession() as session:
        async with session.post(
            url = url,
            headers = headers,
            data = data
        ) as resp:
            return await resp.json()

class LineBotAPI:
    def __init__(self, notify_token: str, line_bot_token: str, line_group_id: str) -> None:
        self.notify_token = notify_token
        self.line_bot_token = line_bot_token
        self.line_group_id = line_group_id
        self.loop = asyncio.get_event_loop()

    # LINE Notifyでテキストメッセージを送信
    async def push_message_notify(self, message: str) -> json:
        data = {'message': f'message: {message}'}
        return await line_post_request(
            url = NOTIFY_URL, 
            headers = {'Authorization': f'Bearer {self.notify_token}'}, 
            data = data
        )
        
    # LINE Messageing APIでテキストメッセージを送信
    async def push_message(self,message_text:str) -> json:
        data = {
            'to':self.line_group_id,
            'messages':[
                {
                    'type':'text',
                    'text':message_text
                }
            ]
        }
        return await line_post_request(
            url = LINE_BOT_URL + "/message/push",
            headers = {
                'Authorization': 'Bearer ' + self.line_bot_token,
                'Content-Type': 'application/json'
            },
            data = json.dumps(data)
        )

    # 送ったメッセージ数を取得
    async def totalpush(self) -> int:
        r = await line_get_request(
            LINE_BOT_URL + "/message/quota/consumption",
            self.line_bot_token
        )
        return int(r["totalUsage"])

    # LINE Notifyのステータスを取得
    async def notify_status(self) -> Response:
        async with aiohttp.ClientSession() as session:
            async with session.get(
                url = NOTIFY_STATUS_URL,
                headers = {'Authorization': 'Bearer ' + self.notify_token}
            ) as resp:
                return resp

    # LINE Notifyの1時間当たりの上限を取得
    async def rate_limit(self) -> int:
        resp = await self.notify_status()
        ratelimit = resp.headers.get('X-RateLimit-Limit')
        return int(ratelimit)

    # LINE Notifyの1時間当たりの残りの回数を取得
    async def rate_remaining(self) -> int:
        resp = await self.notify_status()
        ratelimit = resp.headers.get('X-RateLimit-Remaining')
        return int(ratelimit)

    # LINE Notifyの1時間当たりの画像送信上限を取得
    async def rate_image_limit(self) -> int:
        resp = await self.notify_status()
        ratelimit = resp.headers.get('X-RateLimit-ImageLimit')
        return int(ratelimit)

    # LINE Notifyの1時間当たりの残り画像送信上限を取得
    async def rate_image_remaining(self) -> int:
        resp = await self.notify_status()
        ratelimit = resp.headers.get('X-RateLimit-ImageRemaining')
        return int(ratelimit)

    # 友達数、グループ人数をカウント
    async def friend(self) -> str:
        # グループIDが有効かどうか判断
        try:
            r = await line_get_request(
                LINE_BOT_URL + "/group/" + self.line_group_id + "/members/count",
                self.line_bot_token,
            )
            return r["count"]
        # グループIDなしの場合、友達数をカウント
        except KeyError:
            # 日付が変わった直後の場合、前日を参照
            if datetime.datetime.now().strftime('%H') == '00':
                before_day = datetime.date.today() + datetime.timedelta(days=-1)
                url = LINE_BOT_URL + "/insight/followers?date=" + before_day.strftime('%Y%m%d')
            else:
                url = LINE_BOT_URL + "/insight/followers?date=" + datetime.date.today().strftime('%Y%m%d')
            r = await line_get_request(
                url,
                self.line_bot_token,
            )
            return r["followers"] 

    # 当月に送信できるメッセージ数の上限目安を取得(基本1000,23年6月以降は200)
    async def pushlimit(self) -> str:
        r = await line_get_request(
            LINE_BOT_URL + "/message/quota",
            self.line_bot_token
        )
        return r["value"]

    # LINEのユーザプロフィールから名前を取得
    async def get_proflie(self, user_id: str) -> Profile:
        # グループIDが有効かどうか判断
        try:
            r = await line_get_request(
                LINE_BOT_URL + f"/group/{self.line_group_id}/member/{user_id}",
                self.line_bot_token,
            )
        # グループIDが無効の場合、友達から判断
        except KeyError:
            r = await line_get_request(
                LINE_BOT_URL + f"/profile/{user_id}",
                self.line_bot_token,
            )
        return await Profile.new_from_json_dict(data=r)


+     # LINEから画像データを取得し、Gyazoにアップロード
+     async def image_upload(self, message_id: int) -> GyazoJson:
+         # 画像のバイナリデータを取得
+         async with aiohttp.ClientSession() as session:
+             async with session.get(
+                     url = LINE_CONTENT_URL + f'/message/{message_id}/content',
+                     headers={
+                         'Authorization': 'Bearer ' + self.line_bot_token
+                     }
+             ) as bytes:
+                 image_bytes = await bytes.read()
+ 
+                 # Gyazoにアップロードする
+                 async with aiohttp.ClientSession() as session:
+                     async with session.post(
+                         url = 'https://upload.gyazo.com/api/upload',
+                         headers={
+                             'Authorization': 'Bearer ' + os.environ['GYAZO_TOKEN'],
+                         },
+                         data={
+                             'imagedata': image_bytes
+                         }
+                     ) as gyazo_image:
+                         return await GyazoJson.new_from_json_dict(await gyazo_image.json())

server.py

受け取ったGyazoの画像URLをDiscordに送信します。

server.py
from fastapi import FastAPI,Depends,HTTPException,Request,Header,Response
from fastapi.responses import HTMLResponse
from threading import Thread
import uvicorn

import base64
import hashlib
import hmac
import re

from dotenv import load_dotenv
load_dotenv()


from message_type.line_type.line_event import Line_Responses
from message_type.discord_type.message_creater import ReqestDiscord
from message_type.line_type.line_message import LineBotAPI


import os

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

app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)

# LINE側のメッセージを受け取る
@app.post("/line_bot")
async def line_response(
    response:Line_Responses,
    byte_body:Request, 
    x_line_signature=Header(None)
):
    """
    response:Line_Responses
    LINEから受け取ったイベントの内容
    jsonがクラスに変換されている。
    
    byte_body:Request
    LINEから受け取ったイベントのバイナリデータ。
    LINEからのメッセージという署名の検証に必要。

    x_line_signature:Header
    LINEから受け取ったjsonのヘッダー。
    こちらも署名に必要。
    """

    # request.bodyを取得
    boo = await byte_body.body()
    body = boo.decode('utf-8')

    # channel_secretからbotの種類を判別する
    for bot_name in bots_name:
        channel_secret = os.environ[f'{bot_name}_CHANNEL_SECRET']
        # ハッシュ値を求める
        hash = hmac.new(
            channel_secret.encode('utf-8'),
            body.encode('utf-8'), 
            hashlib.sha256
        ).digest()

        # 結果を格納
        signature = base64.b64encode(hash)
        decode_signature = signature.decode('utf-8')

        if decode_signature == x_line_signature:
            channel_secret = os.environ[f'{bot_name}_CHANNEL_SECRET']
            # Discordサーバーのクラスを宣言
            discord_find_message = ReqestDiscord(
                guild_id = int(os.environ[f'{bot_name}_GUILD_ID']),
                limit = int(os.environ["USER_LIMIT"]), 
                token = TOKEN
            )
            # LINEのクラスを宣言
            line_bot_api = LineBotAPI(
                notify_token = os.environ.get(f'{bot_name}_NOTIFY_TOKEN'),
                line_bot_token = os.environ[f'{bot_name}_BOT_TOKEN'],
                line_group_id = os.environ.get(f'{bot_name}_GROUP_ID')
            )
            # メッセージを送信するDiscordのテキストチャンネルのID
            channel_id = int(os.environ[f'{bot_name}_CHANNEL_ID'])
            break

    # ハッシュ値が一致しなかった場合エラーを返す
    if decode_signature != x_line_signature: 
        raise Exception

    # 応答確認の場合終了
    if type(response.events) is list:
        return HTMLResponse("OK")

    # イベントの中身を取得
    event = response.events

    # LINEのプロフィールを取得(友達登録している場合)
    profile_name = await line_bot_api.get_proflie(user_id=event.source.userId)

    # テキストメッセージの場合
    if event.message.type == 'text':
        message = event.message.text
        # Discordのメンバー、ロール、チャンネルの指定があるか取得する
        """
        members_find
        テキストメッセージからユーザーのメンションを検出し、変換する。
        @ユーザー#0000#member → <@00000000000>
        roles_find
        テキストメッセージからロールのメンションを検出し、変換する。
        @ロール#role → <@&0000000000>
        channel_select
        テキストメッセージから送信場所を検出し、送信先のチャンネルidを返す。
        テキストチャンネルのみ送信可能。ただし、メッセージの先頭に書かれていなければ適用されない。
        /チャンネル名#channel → 削除
        """

        message = await discord_find_message.members_find(message=message)
        message = await discord_find_message.roles_find(message=message)
        channel_id, message = await discord_find_message.channel_select(channel_id=channel_id,message=message)

+     # 画像が送信された場合
+     if event.message.type == 'image':
+         # バイナリデータを取得しGyazoに送信
+         gyazo_json = await line_bot_api.image_upload(event.message.id)
+         # Gyazoのurlを返す
+         message = f"https://i.gyazo.com/{gyazo_json.image_id}.{gyazo_json.type}"

    # LINEの名前 「メッセージ」の形式で送信
    message = f'{profile_name.display_name} \n「 {message} 」'
    await discord_find_message.send_discord(channel_id=channel_id, message=message)

    # レスポンス200を返し終了
    return HTMLResponse(content="OK")

def run():
    uvicorn.run("server:app",  host="0.0.0.0", port=int(os.getenv("PORT", default=5000)), log_level="info")

# DiscordBotと並列で立ち上げる
def keep_alive():
    t = Thread(target=run)
    t.start()

# ローカルで実行する際
if __name__ == '__main__':
    uvicorn.run(app,host='localhost', port=8000)

ポイント

    # 画像が送信された場合
    if event.message.type == 'image':
        # バイナリデータを取得しGyazoに送信
        gyazo_json = await line_bot_api.image_upload(event.message.id)
        # Gyazoのurlを返す
        message = f"https://i.gyazo.com/{gyazo_json.image_id}.{gyazo_json.type}"

見ているとわかりますがテキストメッセージに直接画像のURLを貼り付けています。
Discordにはファイルのプレビュー機能があるため、テキストに直接貼り付けても画像がプレビューされるのです。

完成!!

試しに画像を送信してみましょう。
Discord側でプレビューできていればOKです!!
スクリーンショット 2022-09-16 083039.png

まとめ

お疲れ様でした。
あまり説明することもなかったので、結構内容は短めになってしまいました。
さて残りは

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

となります。いよいよ折り返しですね。
次回は

  • LINEからDiscordへの動画

となります。
結構難関ですのでオウム返しで予習してみてください。

https://qiita.com/maguro-alternative/items/7c9f1d3024bec6082ff0

Discussion