iTranslated by AI
The content below is an AI-generated translation. This is an experimental feature, and may contain errors. View original article
📂
Integrating Discord and LINE with Python, FastAPI, and Docker [Part 3]: Sending Images from LINE to Discord
Greeting
Hello. I'm Maguro.
This is a continuation of the following article.
This time, we will send images from LINE to Discord.
Background
As mentioned in this article, files uploaded to LINE are stored as binary data on the server and are deleted after a certain period.
In this article, just like the one above, we will use Gyazo to share them on the Discord side.
It is not mandatory, but if you have followed the previous article's content, you will likely understand the flow and be able to implement this article's content smoothly. (Actually, it's almost the same thing.)
Preparation
Obtaining the Gyazo API Key
To use the Gyazo API, make sure to register an account and obtain an API key.
Select "New Application"
Specify the name and callback URL.
Since we won't be using the callback, you can decide it freely, just like the name.
Once you press submit, a token will be issued.
Make sure to keep a note of it.
Environment Variables
Add the following environment variables.
GYAZO_TOKEN
Gyazo token.
Used for uploading images.
Coding
Directory Structure
$ tree
.
├── app
│ ├── cogs
│ │ └── mst_line.py # From Discord to LINE
│ ├── core
│ │ └── start.py # For starting the Discord Bot
│ ├── message_type
│ │ ├── discord_type
│ │ │ ├── discord_type.py # Class for Discord server
│ │ │ └── message_creater.py # Directly calling the Discord API
│ │ └── line_type
│ │ ├── line_type.py # Classes for LINE profiles, etc.
│ │ ├── line_event.py # Classes for LINE events
│ │ └── line_message.py # Classes for LINE messages
│ ├── server.py # Server startup
│ └── main.py
├── Dockerfile
├── Profile
└── requirements.txt
line_type.py
Since Gyazo's response is also received as JSON, we will handle it as a class just like before.
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):
"""Returns the JSON string.
:rtype: str
"""
return json.dumps(self.as_json_dict(), sort_keys=True)
async def as_json_dict(self):
"""Returns a dictionary from this object.
: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):
"""Creates a new instance from a dict.
:param data: JSON dictionary
"""
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):
"""Convert to snake case.
: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):
"""Convert to camel case.
:param str text:
:rtype: str
"""
split = text.split('_')
return split[0] + "".join(x.title() for x in split[1:])
class Profile(Base):
"""
Profile class for LINE Messaging API
user_id :LINE user id
display_name :LINE user name
picture_url :LINE icon URL
status_message :LINE profile status message
"""
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 class
+ image_id :Image ID
+ permalink_url :Image permalink
+ thumb_url :Thumbnail image URL
+ url :Image URL
+ type :Extension 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
Add the process to receive binary image data from LINE and send it to 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'
# Perform a GET request to LINE
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()
# Perform a POST request to LINE
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()
# Send a text message via 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
)
# Send a text message via LINE Messaging 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)
)
# Get the number of messages sent
async def totalpush(self) -> int:
r = await line_get_request(
LINE_BOT_URL + "/message/quota/consumption",
self.line_bot_token
)
return int(r["totalUsage"])
# Get the status of 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
# Get the hourly limit of LINE Notify
async def rate_limit(self) -> int:
resp = await self.notify_status()
ratelimit = resp.headers.get('X-RateLimit-Limit')
return int(ratelimit)
# Get the hourly remaining number of requests for LINE Notify
async def rate_remaining(self) -> int:
resp = await self.notify_status()
ratelimit = resp.headers.get('X-RateLimit-Remaining')
return int(ratelimit)
# Get the hourly image transmission limit for LINE Notify
async def rate_image_limit(self) -> int:
resp = await self.notify_status()
ratelimit = resp.headers.get('X-RateLimit-ImageLimit')
return int(ratelimit)
# Get the hourly remaining image transmission limit for LINE Notify
async def rate_image_remaining(self) -> int:
resp = await self.notify_status()
ratelimit = resp.headers.get('X-RateLimit-ImageRemaining')
return int(ratelimit)
# Count friends and group members
async def friend(self) -> str:
# Check if the group ID is valid
try:
r = await line_get_request(
LINE_BOT_URL + "/group/" + self.line_group_id + "/members/count",
self.line_bot_token,
)
return r["count"]
# If no group ID is found, count the number of friends
except KeyError:
# If it is right after the date has changed, refer to the previous day
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"]
# Get the estimated maximum number of messages that can be sent in the current month (usually 1000, 200 after June 2023)
async def pushlimit(self) -> str:
r = await line_get_request(
LINE_BOT_URL + "/message/quota",
self.line_bot_token
)
return r["value"]
# Get the name from the LINE user profile
async def get_proflie(self, user_id: str) -> Profile:
# Check if the group ID is valid
try:
r = await line_get_request(
LINE_BOT_URL + f"/group/{self.line_group_id}/member/{user_id}",
self.line_bot_token,
)
# If the group ID is invalid, determine from friends
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)
+ # Retrieve image data from LINE and upload it to Gyazo
+ async def image_upload(self, message_id: int) -> GyazoJson:
+ # Retrieve the binary data of the image
+ 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()
+
+ # Upload to 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
Send the received Gyazo image URL to 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)
# Receive messages from LINE
@app.post("/line_bot")
async def line_response(
response:Line_Responses,
byte_body:Request,
x_line_signature=Header(None)
):
"""
response:Line_Responses
Content of the event received from LINE
JSON has been converted into a class.
byte_body:Request
Binary data of the event received from LINE.
Required for verifying the signature as a message from LINE.
x_line_signature:Header
JSON header received from LINE.
Also required for the signature.
"""
# Get request.body
boo = await byte_body.body()
body = boo.decode('utf-8')
# Determine the bot type from channel_secret
for bot_name in bots_name:
channel_secret = os.environ[f'{bot_name}_CHANNEL_SECRET']
# Calculate the hash value
hash = hmac.new(
channel_secret.encode('utf-8'),
body.encode('utf-8'),
hashlib.sha256
).digest()
# Store the result
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']
# Declare the Discord server class
discord_find_message = ReqestDiscord(
guild_id = int(os.environ[f'{bot_name}_GUILD_ID']),
limit = int(os.environ["USER_LIMIT"]),
token = TOKEN
)
# Declare the LINE class
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')
)
# ID of the Discord text channel to send the message to
channel_id = int(os.environ[f'{bot_name}_CHANNEL_ID'])
break
# Return an error if the hash values do not match
if decode_signature != x_line_signature:
raise Exception
# End if it is a connectivity test
if type(response.events) is list:
return HTMLResponse("OK")
# Get the content of the event
event = response.events
# Get the LINE profile (if they have added the bot as a friend)
profile_name = await line_bot_api.get_proflie(user_id=event.source.userId)
# In case of a text message
if event.message.type == 'text':
message = event.message.text
# Check if Discord members, roles, or channels are specified
"""
members_find
Detect user mentions from text messages and convert them.
@User#0000#member → <@00000000000>
roles_find
Detect role mentions from text messages and convert them.
@Role#role → <@&0000000000>
channel_select
Detect the destination from text messages and return the destination channel ID.
Only text channels can be sent to. However, it won't apply unless it's written at the beginning of the message.
/channel_name#channel → Delete
"""
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 an image is sent
+ if event.message.type == 'image':
+ # Get binary data and send to Gyazo
+ gyazo_json = await line_bot_api.image_upload(event.message.id)
+ # Return the Gyazo URL
+ message = f"https://i.gyazo.com/{gyazo_json.image_id}.{gyazo_json.type}"
# Send in the format: LINE Name 「 Message 」
message = f'{profile_name.display_name} \n「 {message} 」'
await discord_find_message.send_discord(channel_id=channel_id, message=message)
# Return response 200 and end
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")
# Start in parallel with the Discord Bot
def keep_alive():
t = Thread(target=run)
t.start()
# When running locally
if __name__ == '__main__':
uvicorn.run(app,host='localhost', port=8000)
Key Points
# If an image is sent
if event.message.type == 'image':
# Get binary data and send to Gyazo
gyazo_json = await line_bot_api.image_upload(event.message.id)
# Return the Gyazo URL
message = f"https://i.gyazo.com/{gyazo_json.image_id}.{gyazo_json.type}"
As you can see, the image URL is pasted directly into the text message.
Since Discord has a file preview feature, the image will be previewed even if the URL is pasted directly into the text.
Finished!!
Let's try sending an image. If it's previewed on the Discord side, it's a success!!
Summary
Great job! Since there wasn't much to explain, the content ended up being quite short.
Now, the remaining tasks are:
- Videos from LINE to Discord
- Stickers from LINE to Discord
- Images, videos, and stickers from Discord to LINE
We are finally at the halfway point. Next time will be:
- Videos from LINE to Discord
It's quite a challenge, so please try preparing by practicing with an echo bot.
Discussion