66 lines
2 KiB
Python
66 lines
2 KiB
Python
from datetime import datetime
|
|
from io import BytesIO
|
|
from typing import Optional
|
|
|
|
from aiogram import Bot
|
|
from aiogram.enums import ContentType
|
|
from aiogram.types import User, PhotoSize, Message, BufferedInputFile
|
|
|
|
import ai_agent
|
|
import utils
|
|
|
|
|
|
async def get_user_name_for_ai(user: User):
|
|
if user.first_name and user.last_name:
|
|
return "{} {}".format(user.first_name, user.last_name)
|
|
elif user.first_name:
|
|
return user.first_name
|
|
elif user.username:
|
|
return user.username
|
|
else:
|
|
return str(user.id)
|
|
|
|
|
|
async def download_photo(photo: PhotoSize, bot: Bot) -> bytes:
|
|
# noinspection PyTypeChecker
|
|
photo_bytes: BytesIO = await bot.download(photo.file_id)
|
|
return photo_bytes.getvalue()
|
|
|
|
|
|
def get_message_text(message: Message) -> Optional[str]:
|
|
if message.content_type == ContentType.TEXT:
|
|
return message.text
|
|
elif message.content_type == ContentType.PHOTO:
|
|
return message.caption
|
|
else:
|
|
return None
|
|
|
|
|
|
async def create_ai_message(message: Message, bot: Bot) -> ai_agent.Message:
|
|
ai_message = ai_agent.Message()
|
|
ai_message.message_id = message.message_id
|
|
ai_message.user_name = await get_user_name_for_ai(message.from_user)
|
|
if message.content_type == ContentType.TEXT:
|
|
ai_message.text = message.text
|
|
elif message.content_type == ContentType.PHOTO:
|
|
if message.media_group_id is None:
|
|
ai_message.text = message.caption
|
|
ai_message.image = await download_photo(message.photo[-1], bot)
|
|
else:
|
|
raise utils.UnsupportedContentType()
|
|
else:
|
|
raise utils.UnsupportedContentType()
|
|
return ai_message
|
|
|
|
|
|
def wrap_photo(image: bytes) -> BufferedInputFile:
|
|
return BufferedInputFile(image, 'image.jpg')
|
|
|
|
|
|
def wrap_document(document: bytes, name_prefix: str, extension: str) -> BufferedInputFile:
|
|
name = "{}_{}.{}".format(name_prefix, datetime.now().strftime("%Y%m%d_%H%M%S"), extension)
|
|
return BufferedInputFile(document, name)
|
|
|
|
|
|
def trim_caption(caption: str) -> str:
|
|
return caption[:1024]
|