matrix_chatgpt_bot/send_image.py
2023-04-13 13:41:47 +08:00

63 lines
1.9 KiB
Python

"""
code derived from:
https://matrix-nio.readthedocs.io/en/latest/examples.html#sending-an-image
"""
import os
import aiofiles.os
import magic
from PIL import Image
from nio import AsyncClient, UploadResponse
from log import getlogger
logger = getlogger()
async def send_room_image(client: AsyncClient,
room_id: str, image: str):
"""
image: image path
"""
mime_type = magic.from_file(image, mime=True) # e.g. "image/jpeg"
im = Image.open(image)
(width, height) = im.size # im.size returns (width,height) tuple
# first do an upload of image, then send URI of upload to room
file_stat = await aiofiles.os.stat(image)
async with aiofiles.open(image, "r+b") as f:
resp, maybe_keys = await client.upload(
f,
content_type=mime_type, # image/jpeg
filename=os.path.basename(image),
filesize=file_stat.st_size,
)
if not isinstance(resp, UploadResponse):
logger.warning(f"Failed to generate image. Failure response: {resp}")
await client.room_send(
room_id,
message_type="m.room.message",
content={"msgtype": "m.text",
"body": f"Failed to generate image. Failure response: {resp}", },
ignore_unverified_devices=True,
)
return
content = {
"body": os.path.basename(image), # descriptive title
"info": {
"size": file_stat.st_size,
"mimetype": mime_type,
"w": width, # width in pixel
"h": height, # height in pixel
},
"msgtype": "m.image",
"url": resp.content_uri,
}
try:
await client.room_send(room_id, message_type="m.room.message", content=content)
except Exception as e:
logger.error(
f"Image send of file {image} failed.\n Error: {e}", exc_info=True)
raise Exception(e)