love_apples/maxapi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Добавлен types.InputMedia для простой загрузки медиафайлов

Browse Source
This commit is contained in:
Денис Семёнов 2025-06-20 02:24:14 +03:00
parent 85f58913c3
commit 1374d863f0
33 changed files with 332 additions and 262 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

151
example.py
View File

@ -1,151 +0,0 @@
import asyncio
import logging
from maxapi import Bot, Dispatcher, F
from maxapi.context import MemoryContext, State, StatesGroup
from maxapi.types import Command, MessageCreated, CallbackButton, MessageCallback, BotCommand
from maxapi.utils.inline_keyboard import InlineKeyboardBuilder
from for_example import router
logging.basicConfig(level=logging.INFO)
bot = Bot('f9LHodD0cOL5NY7All_9xJRh5ZhPw6bRvq_0Adm8-1bZZEHdRy6_ZHDMNVPejUYNZg7Zhty-wKHNv2X2WJBQ')
dp = Dispatcher()
dp.include_routers(router)
start_text = '''Пример чат-бота для MAX 💙
Мои команды:
/clear очищает ваш контекст
/state или /context показывают ваше контекстное состояние
/data показывает вашу контекстную память
'''
class Form(StatesGroup):
name = State()
age = State()
@dp.on_started()
async def _():
logging.info('Бот стартовал!')
@dp.message_created(Command('clear'))
async def hello(event: MessageCreated, context: MemoryContext):
await context.clear()
await event.message.answer(f"Ваш контекст был очищен!")
@dp.message_created(Command('data'))
async def hello(event: MessageCreated, context: MemoryContext):
data = await context.get_data()
await event.message.answer(f"Ваша контекстная память: {str(data)}")
@dp.message_created(Command('context'))
@dp.message_created(Command('state'))
async def hello(event: MessageCreated, context: MemoryContext):
data = await context.get_state()
await event.message.answer(f"Ваше контекстное состояние: {str(data)}")
@dp.message_created(Command('start'))
async def hello(event: MessageCreated):
builder = InlineKeyboardBuilder()
builder.row(
CallbackButton(
text='Ввести свое имя',
payload='btn_1'
),
CallbackButton(
text='Ввести свой возраст',
payload='btn_2'
)
)
builder.row(
CallbackButton(
text='Не хочу',
payload='btn_3'
)
)
await event.message.answer(
text=start_text,
attachments=[builder.as_markup()] # Для MAX клавиатура это вложение,
) # поэтому она в списке вложений
@dp.message_callback(F.callback.payload == 'btn_1')
async def hello(event: MessageCallback, context: MemoryContext):
await context.set_state(Form.name)
await event.message.delete()
await event.message.answer(f'Отправьте свое имя:')
@dp.message_callback(F.callback.payload == 'btn_2')
async def hello(event: MessageCallback, context: MemoryContext):
await context.set_state(Form.age)
await event.message.delete()
await event.message.answer(f'Отправьте ваш возраст:')
@dp.message_callback(F.callback.payload == 'btn_3')
async def hello(event: MessageCallback, context: MemoryContext):
await event.message.delete()
await event.message.answer(f'Ну ладно 🥲')
@dp.message_created(F.message.body.text, Form.name)
async def hello(event: MessageCreated, context: MemoryContext):
await context.update_data(name=event.message.body.text)
data = await context.get_data()
await event.message.answer(f"Приятно познакомиться, {data['name'].title()}!")
@dp.message_created(F.message.body.text, Form.age)
async def hello(event: MessageCreated, context: MemoryContext):
await context.update_data(age=event.message.body.text)
await event.message.answer(f"Ого! А мне всего пару недель 😁")
async def main():
await bot.set_my_commands(
BotCommand(
name='/start',
description='Перезапустить бота'
),
BotCommand(
name='/clear',
description='Очищает ваш контекст'
),
BotCommand(
name='/state',
description='Показывают ваше контекстное состояние'
),
BotCommand(
name='/data',
description='Показывает вашу контекстную память'
),
BotCommand(
name='/context',
description='Показывают ваше контекстное состояние'
)
)
await dp.start_polling(bot)
# await dp.handle_webhook(
# bot=bot,
# host='localhost',
# port=8080
# )
asyncio.run(main())

BIN
example/audio.mp3 Normal file
View File

Binary file not shown.

85
example/example.py
View File

@ -3,19 +3,22 @@ import logging
from maxapi import Bot, Dispatcher, F
from maxapi.context import MemoryContext, State, StatesGroup
from maxapi.types import Command, MessageCreated, CallbackButton, MessageCallback
from maxapi.types import Command, MessageCreated, CallbackButton, MessageCallback, BotCommand
from maxapi.types.input_media import InputMedia
from maxapi.utils.inline_keyboard import InlineKeyboardBuilder
from example.for_example import router
from for_example import router
logging.basicConfig(level=logging.INFO)
bot = Bot('токен')
bot = Bot('f9LHodD0cOL5NY7All_9xJRh5ZhPw6bRvq_0Adm8-1bZZEHdRy6_ZHDMNVPejUYNZg7Zhty-wKHNv2X2WJBQ')
dp = Dispatcher()
dp.include_routers(router)
start_text = '''Мои команды:
start_text = '''Пример чат-бота для MAX 💙
Мои команды:
/clear очищает ваш контекст
/state или /context показывают ваше контекстное состояние
@ -34,26 +37,26 @@ async def _():
@dp.message_created(Command('clear'))
async def hello(obj: MessageCreated, context: MemoryContext):
async def hello(event: MessageCreated, context: MemoryContext):
await context.clear()
await obj.message.answer(f"Ваш контекст был очищен!")
await event.message.answer(f"Ваш контекст был очищен!")
@dp.message_created(Command('data'))
async def hello(obj: MessageCreated, context: MemoryContext):
async def hello(event: MessageCreated, context: MemoryContext):
data = await context.get_data()
await obj.message.answer(f"Ваша контекстная память: {str(data)}")
await event.message.answer(f"Ваша контекстная память: {str(data)}")
@dp.message_created(Command('context'))
@dp.message_created(Command('state'))
async def hello(obj: MessageCreated, context: MemoryContext):
async def hello(event: MessageCreated, context: MemoryContext):
data = await context.get_state()
await obj.message.answer(f"Ваше контекстное состояние: {str(data)}")
await event.message.answer(f"Ваше контекстное состояние: {str(data)}")
@dp.message_created(Command('start'))
async def hello(obj: MessageCreated):
async def hello(event: MessageCreated):
builder = InlineKeyboardBuilder()
builder.row(
@ -73,49 +76,73 @@ async def hello(obj: MessageCreated):
)
)
await obj.message.answer(
await event.message.answer(
text=start_text,
attachments=[builder.as_markup()] # Для MAX клавиатура это вложение,
) # поэтому она в списке вложений
attachments=[
builder.as_markup(),
] # Для MAX клавиатура это вложение,
) # поэтому она в списке вложений
@dp.message_callback(F.callback.payload == 'btn_1')
async def hello(obj: MessageCallback, context: MemoryContext):
async def hello(event: MessageCallback, context: MemoryContext):
await context.set_state(Form.name)
await obj.message.delete()
await obj.message.answer(f'Отправьте свое имя:')
await event.message.delete()
await event.message.answer(f'Отправьте свое имя:')
@dp.message_callback(F.callback.payload == 'btn_2')
async def hello(obj: MessageCallback, context: MemoryContext):
async def hello(event: MessageCallback, context: MemoryContext):
await context.set_state(Form.age)
await obj.message.delete()
await obj.message.answer(f'Отправьте ваш возраст:')
await event.message.delete()
await event.message.answer(f'Отправьте ваш возраст:')
@dp.message_callback(F.callback.payload == 'btn_3')
async def hello(obj: MessageCallback, context: MemoryContext):
await obj.message.delete()
await obj.message.answer(f'Ну ладно 🥲')
async def hello(event: MessageCallback, context: MemoryContext):
await event.message.delete()
await event.message.answer(f'Ну ладно 🥲')
@dp.message_created(F.message.body.text, Form.name)
async def hello(obj: MessageCreated, context: MemoryContext):
await context.update_data(name=obj.message.body.text)
async def hello(event: MessageCreated, context: MemoryContext):
await context.update_data(name=event.message.body.text)
data = await context.get_data()
await obj.message.answer(f"Приятно познакомиться, {data['name'].title()}!")
await event.message.answer(f"Приятно познакомиться, {data['name'].title()}!")
@dp.message_created(F.message.body.text, Form.age)
async def hello(obj: MessageCreated, context: MemoryContext):
await context.update_data(age=obj.message.body.text)
async def hello(event: MessageCreated, context: MemoryContext):
await context.update_data(age=event.message.body.text)
await obj.message.answer(f"Ого! А мне всего пару недель 😁")
await event.message.answer(f"Ого! А мне всего пару недель 😁")
async def main():
await bot.set_my_commands(
BotCommand(
name='/start',
description='Перезапустить бота'
),
BotCommand(
name='/clear',
description='Очищает ваш контекст'
),
BotCommand(
name='/state',
description='Показывают ваше контекстное состояние'
),
BotCommand(
name='/data',
description='Показывает вашу контекстную память'
),
BotCommand(
name='/context',
description='Показывают ваше контекстное состояние'
)
)
await dp.start_polling(bot)
# await dp.handle_webhook(
# bot=bot,

16
example/for_example.py
View File

@ -1,10 +1,24 @@
from maxapi import F, Router
from maxapi.types import Command, MessageCreated
from maxapi.types import InputMedia
router = Router()
file = __file__.split('\\')[-1]
@router.message_created(Command('router'))
async def hello(obj: MessageCreated):
file = __file__.split('\\')[-1]
await obj.message.answer(f"Пишу тебе из роута {file}")
# новая команда для примера, /media,
# пример использования: /media image.png (медиафайл берется указанному пути)
@router.message_created(Command('media'))
async def hello(event: MessageCreated):
await event.message.answer(
attachments=[
InputMedia(
path=event.message.body.text.replace('/media ', '')
)
]
)

10
for_example.py
View File

@ -1,10 +0,0 @@
from maxapi import F, Router
from maxapi.types import Command, MessageCreated
router = Router()
@router.message_created(Command('router'))
async def hello(obj: MessageCreated):
file = __file__.split('\\')[-1]
await obj.message.answer(f"Пишу тебе из роута {file}")

11
maxapi/bot.py
View File

@ -1,6 +1,7 @@
from datetime import datetime
from typing import Any, Dict, List, TYPE_CHECKING
from .methods.get_upload_url import GetUploadURL
from .methods.get_updates import GetUpdates
from .methods.remove_member_chat import RemoveMemberChat
from .methods.add_admin_chat import AddAdminChat
@ -30,6 +31,7 @@ from .methods.send_message import SendMessage
from .enums.parse_mode import ParseMode
from .enums.sender_action import SenderAction
from .enums.upload_type import UploadType
from .types.attachments.attachment import Attachment
from .types.attachments.image import PhotoAttachmentRequestPayload
@ -337,6 +339,15 @@ class Bot(BaseConnection):
bot=self,
).request()
async def get_upload_url(
self,
type: UploadType
):
return await GetUploadURL(
bot=self,
type=type
).request()
async def set_my_commands(
self,
*commands: BotCommand

60
maxapi/connection/base.py
View File

@ -1,10 +1,17 @@
import os
from typing import TYPE_CHECKING
import aiohttp
from pydantic import BaseModel
from ..types.errors import Error
from ..enums.http_method import HTTPMethod
from ..enums.api_path import ApiPath
from ..loggers import logger_bot
from ..enums.upload_type import UploadType
from ..loggers import logger_bot, logger_connection
if TYPE_CHECKING:
from ..bot import Bot
class BaseConnection:
@ -12,14 +19,14 @@ class BaseConnection:
API_URL = 'https://botapi.max.ru'
def __init__(self):
self.bot = None
self.session = None
self.bot: 'Bot' = None
self.session: aiohttp.ClientSession = None
async def request(
self,
method: HTTPMethod,
path: ApiPath,
model: BaseModel,
model: BaseModel = None,
is_return_raw: bool = False,
**kwargs
):
@ -27,15 +34,18 @@ class BaseConnection:
if not self.bot.session:
self.bot.session = aiohttp.ClientSession(self.bot.API_URL)
r = await self.bot.session.request(
method=method.value,
url=path.value if isinstance(path, ApiPath) else path,
**kwargs
)
try:
r = await self.bot.session.request(
method=method.value,
url=path.value if isinstance(path, ApiPath) else path,
**kwargs
)
except aiohttp.ClientConnectorDNSError as e:
return logger_connection.error(f'Ошибка при отправке запроса: {e}')
if not r.ok:
raw = await r.text()
error = Error(code=r.status, text=raw)
raw = await r.json()
error = Error(code=r.status, raw=raw)
logger_bot.error(error)
return error
@ -54,3 +64,31 @@ class BaseConnection:
model.bot = self.bot
return model
async def upload_file(
self,
url: str,
path: str,
type: UploadType
):
with open(path, 'rb') as f:
file_data = f.read()
basename = os.path.basename(path)
name, ext = os.path.splitext(basename)
form = aiohttp.FormData()
form.add_field(
name='data',
value=file_data,
filename=basename,
content_type=f"{type.value}/{ext.lstrip('.')}"
)
async with aiohttp.ClientSession() as session:
response = await session.post(
url=url,
data=form
)
return await response.text()

5
maxapi/dispatcher.py
View File

@ -3,6 +3,7 @@ from typing import Callable, List
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from uvicorn import Config, Server
from aiohttp import ClientConnectorDNSError
from .filters.handler import Handler
@ -121,7 +122,9 @@ class Dispatcher:
try:
await self.handle(event)
except Exception as e:
logger_dp.error(f"Ошибка при обработке события: {events['update_type']}: {e}")
logger_dp.error(f"Ошибка при обработке события: {event.update_type}: {e}")
except ClientConnectorDNSError:
logger_dp.error(f'Ошибка подключения: {e}')
except Exception as e:
logger_dp.error(f'Общая ошибка при обработке событий: {e}')

1
maxapi/enums/api_path.py
View File

@ -11,3 +11,4 @@ class ApiPath(str, Enum):
PIN = '/pin'
MEMBERS = '/members'
ADMINS = '/admins'
UPLOADS = '/uploads'

9
maxapi/enums/chat_status.py Normal file
View File

@ -0,0 +1,9 @@
from enum import Enum
class ChatStatus(str, Enum):
ACTIVE = 'active'
REMOVED = 'removed'
LEFT = 'left'
CLOSED = 'closed'
SUSPENDED = 'suspended'

8
maxapi/enums/upload_type.py Normal file
View File

@ -0,0 +1,8 @@
from enum import Enum
class UploadType(str, Enum):
IMAGE = 'image'
VIDEO = 'video'
AUDIO = 'audio'
FILE = 'file'

2
maxapi/filters/handler.py
View File

@ -29,7 +29,7 @@ class Handler:
elif isinstance(arg, State):
self.state = arg
elif isinstance(arg, Command):
self.filters.insert(0, F.message.body.text == arg.command)
self.filters.insert(0, F.message.body.text.startswith(arg.command))
else:
logger_dp.info(f'Обнаружен неизвестный фильтр `{arg}` при '
f'регистрации функции `{func_event.__name__}`')

1
maxapi/loggers.py
View File

@ -1,4 +1,5 @@
import logging
logger_bot = logging.getLogger('bot')
logger_connection = logging.getLogger('connection')
logger_dp = logging.getLogger('dispatcher')

3
maxapi/methods/add_admin_chat.py
View File

@ -1,6 +1,3 @@
from re import findall
from typing import TYPE_CHECKING, List
from .types.added_admin_chat import AddedListAdminChat

3
maxapi/methods/add_members_chat.py
View File

@ -1,6 +1,3 @@
from re import findall
from typing import TYPE_CHECKING, List
from ..methods.types.added_members_chat import AddedMembersChat

2
maxapi/methods/change_info.py
View File

@ -1,5 +1,3 @@
from typing import Any, Dict, List, TYPE_CHECKING
from ..types.users import User

5
maxapi/methods/get_chat_by_id.py
View File

@ -1,12 +1,7 @@
from re import findall
from typing import TYPE_CHECKING
from ..types.chats import Chat
from ..types.users import User
from ..enums.http_method import HTTPMethod
from ..enums.api_path import ApiPath

4
maxapi/methods/get_chat_by_link.py
View File

@ -1,12 +1,8 @@
from re import findall
from typing import TYPE_CHECKING
from ..types.chats import Chat
from ..types.users import User
from ..enums.http_method import HTTPMethod
from ..enums.api_path import ApiPath

2
maxapi/methods/get_chats.py
View File

@ -4,8 +4,6 @@ from typing import TYPE_CHECKING
from ..types.chats import Chats
from ..types.users import User
from ..enums.http_method import HTTPMethod
from ..enums.api_path import ApiPath

3
maxapi/methods/get_list_admin_chat.py
View File

@ -1,6 +1,3 @@
from re import findall
from typing import TYPE_CHECKING
from ..methods.types.getted_list_admin_chat import GettedListAdminChat

4
maxapi/methods/get_me_from_chat.py
View File

@ -2,9 +2,7 @@
from typing import TYPE_CHECKING
from ..types.chats import ChatMember, Chats
from ..types.users import User
from ..types.chats import ChatMember
from ..enums.http_method import HTTPMethod
from ..enums.api_path import ApiPath

3
maxapi/methods/get_members_chat.py
View File

@ -1,6 +1,3 @@
from re import findall
from typing import TYPE_CHECKING, List
from ..methods.types.getted_members_chat import GettedMembersChat

35
maxapi/methods/get_upload_url.py Normal file
View File

@ -0,0 +1,35 @@
from typing import TYPE_CHECKING
from ..methods.types.getted_upload_url import GettedUploadUrl
from ..enums.http_method import HTTPMethod
from ..enums.api_path import ApiPath
from ..enums.upload_type import UploadType
from ..connection.base import BaseConnection
if TYPE_CHECKING:
from ..bot import Bot
class GetUploadURL(BaseConnection):
def __init__(
self,
bot: 'Bot',
type: UploadType
):
self.bot = bot
self.type = type
async def request(self) -> GettedUploadUrl:
params = self.bot.params.copy()
params['type'] = self.type.value
return await super().request(
method=HTTPMethod.POST,
path=ApiPath.UPLOADS,
model=GettedUploadUrl,
params=params,
)

86
maxapi/methods/send_message.py
View File

@ -1,20 +1,33 @@
import asyncio
from typing import List, TYPE_CHECKING
from json import loads as json_loads
from ..enums.upload_type import UploadType
from ..types.attachments.upload import AttachmentPayload, AttachmentUpload
from ..types.errors import Error
from .types.sended_message import SendedMessage
from ..types.message import NewMessageLink
from ..types.input_media import InputMedia
from ..types.attachments.attachment import Attachment
from ..enums.parse_mode import ParseMode
from ..enums.http_method import HTTPMethod
from ..enums.api_path import ApiPath
from ..connection.base import BaseConnection
from ..loggers import logger_bot
if TYPE_CHECKING:
from ..bot import Bot
class UploadResponse:
token: str = None
class SendMessage(BaseConnection):
def __init__(
self,
@ -23,7 +36,7 @@ class SendMessage(BaseConnection):
user_id: int = None,
disable_link_preview: bool = False,
text: str = None,
attachments: List[Attachment] = None,
attachments: List[Attachment | InputMedia] = None,
link: NewMessageLink = None,
notify: bool = True,
parse_mode: ParseMode = None
@ -38,10 +51,41 @@ class SendMessage(BaseConnection):
self.notify = notify
self.parse_mode = parse_mode
async def __process_input_media(
self,
att: InputMedia
):
upload = await self.bot.get_upload_url(att.type)
upload_file_response = await self.upload_file(
url=upload.url,
path=att.path,
type=att.type
)
if att.type in (UploadType.VIDEO, UploadType.AUDIO):
token = upload.token
elif att.type == UploadType.FILE:
json_r = json_loads(upload_file_response)
token = json_r['token']
elif att.type == UploadType.IMAGE:
json_r = json_loads(upload_file_response)
json_r_keys = list(json_r['photos'].keys())
token = json_r['photos'][json_r_keys[0]]['token']
return AttachmentUpload(
type=att.type,
payload=AttachmentPayload(
token=token
)
)
async def request(self) -> SendedMessage:
params = self.bot.params.copy()
json = {}
json = {'attachments': []}
if self.chat_id: params['chat_id'] = self.chat_id
elif self.user_id: params['user_id'] = self.user_id
@ -49,17 +93,37 @@ class SendMessage(BaseConnection):
json['text'] = self.text
json['disable_link_preview'] = str(self.disable_link_preview).lower()
if self.attachments: json['attachments'] = \
[att.model_dump() for att in self.attachments]
if self.attachments:
for att in self.attachments:
if isinstance(att, InputMedia):
input_media = await self.__process_input_media(att)
json['attachments'].append(
input_media.model_dump()
)
else:
json['attachments'].append(att.model_dump())
if not self.link is None: json['link'] = self.link.model_dump()
if not self.notify is None: json['notify'] = self.notify
if not self.parse_mode is None: json['format'] = self.parse_mode.value
return await super().request(
method=HTTPMethod.POST,
path=ApiPath.MESSAGES,
model=SendedMessage,
params=params,
json=json
)
response = None
for attempt in range(5):
response = await super().request(
method=HTTPMethod.POST,
path=ApiPath.MESSAGES,
model=SendedMessage,
params=params,
json=json
)
if isinstance(response, Error):
if response.raw.get('code') == 'attachment.not.ready':
logger_bot.info(f'Ошибка при отправке загруженного медиа, попытка {attempt+1}, жду 2 секунды')
await asyncio.sleep(2)
continue
return response
return response

9
maxapi/methods/types/getted_upload_url.py Normal file
View File

@ -0,0 +1,9 @@
from typing import Any, Optional
from pydantic import BaseModel
from ...types.message import Message
class GettedUploadUrl(BaseModel):
url: Optional[str] = None
token: Optional[str] = None

5
maxapi/methods/types/upload_file_response.py Normal file
View File

@ -0,0 +1,5 @@
from pydantic import BaseModel
class UploadFileResponse(BaseModel):
...

3
maxapi/types/__init__.py
View File

@ -23,7 +23,10 @@ from ..types.attachments.buttons.request_geo_location_button import RequestGeoLo
from ..types.command import Command, BotCommand
from input_media import InputMedia
__all__ = [
InputMedia,
BotCommand,
CallbackButton,
ChatButton,

3
maxapi/types/attachments/attachment.py
View File

@ -1,6 +1,8 @@
from typing import List, Optional, Union
from pydantic import BaseModel
from ...types.attachments.upload import AttachmentUpload
from ...types.attachments.buttons import InlineButtonUnion
from ...types.users import User
from ...enums.attachment import AttachmentType
@ -36,6 +38,7 @@ class ButtonsPayload(BaseModel):
class Attachment(BaseModel):
type: AttachmentType
payload: Optional[Union[
AttachmentUpload,
PhotoAttachmentPayload,
OtherAttachmentPayload,
ContactAttachmentPayload,

14
maxapi/types/attachments/upload.py Normal file
View File

@ -0,0 +1,14 @@
from pydantic import BaseModel
from ...enums.upload_type import UploadType
class AttachmentPayload(BaseModel):
token: str
class AttachmentUpload(BaseModel):
type: UploadType
payload: AttachmentPayload

4
maxapi/types/callback.py
View File

@ -1,10 +1,8 @@
from typing import List, Optional, Union
from typing import Optional
from pydantic import BaseModel
from ..types.users import User
from ..types.users import User
class Callback(BaseModel):
timestamp: int

13
maxapi/types/chats.py
View File

@ -1,23 +1,14 @@
from pydantic import BaseModel, field_validator
from typing import Dict, List, Optional
from enum import Enum
from datetime import datetime
from ..enums.chat_status import ChatStatus
from ..enums.chat_type import ChatType
from ..enums.chat_permission import ChatPermission
from ..types.users import User
from ..types.message import Message
class ChatType(str, Enum):
DIALOG = "dialog"
CHAT = "chat"
class ChatStatus(str, Enum):
ACTIVE = "active"
REMOVED = "removed"
LEFT = "left"
CLOSED = "closed"
SUSPENDED = "suspended"
class Icon(BaseModel):
url: str

2
maxapi/types/errors.py
View File

@ -3,4 +3,4 @@ from pydantic import BaseModel
class Error(BaseModel):
code: int
text: str
raw: dict

24
maxapi/types/input_media.py Normal file
View File

@ -0,0 +1,24 @@
import mimetypes
from ..enums.upload_type import UploadType
class InputMedia:
def __init__(self, path: str):
self.path = path
self.type = self.__detect_file_type(path)
def __detect_file_type(self, path: str) -> UploadType:
mime_type, _ = mimetypes.guess_type(path)
if mime_type is None:
return UploadType.FILE
if mime_type.startswith('video/'):
return UploadType.VIDEO
elif mime_type.startswith('image/'):
return UploadType.IMAGE
elif mime_type.startswith('audio/'):
return UploadType.AUDIO
else:
return UploadType.FILE