BotForge/quickbot
2
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactoring

Browse Source
This commit is contained in:
Alexander Kalinovsky
2025-01-09 13:11:10 +01:00
parent 7793a0cb77
commit 3898a333fa
29 changed files with 1065 additions and 381 deletions
Download Patch File Download Diff File Expand all files Collapse all files

214
bot/handlers/common/__init__.py
View File

@@ -1,24 +1,35 @@
from types import NoneType, UnionType
from aiogram import Router, F
from aiogram.fsm.context import FSMContext
from aiogram.types import Message, CallbackQuery, InlineKeyboardButton
from aiogram.utils.keyboard import InlineKeyboardBuilder
from babel.support import LazyProxy
from inspect import signature
from sqlmodel.ext.asyncio.session import AsyncSession
from typing import Any, get_args, get_origin
from typing import Any, get_args, get_origin, TYPE_CHECKING
import ujson as json
from ..context import ContextData, CallbackCommand, CommandContext
from ...command_context_filter import CallbackCommandFilter
from ....model.user import UserBase
from ....model.settings import Settings
from ....main import QBotApp
from ....model.bot_entity import BotEntity
from ....model.bot_enum import BotEnum
from ....model.view_setting import ViewSetting
from ....utils import get_local_text, deserialize
from ....model.descriptors import (EntityFieldDescriptor,
EntityDescriptor,
EntityCaptionCallable,
EntityItemCaptionCallable,
EntityFieldCaptionCallable)
if TYPE_CHECKING:
from ....main import QBotApp
router = Router()
def get_send_message(message: Message | CallbackQuery):
if isinstance(message, Message):
@@ -27,52 +38,43 @@ def get_send_message(message: Message | CallbackQuery):
return message.message.edit_text
def get_local_text(text: str, lang: str):
try:
text_obj = json.loads(text) #@IgnoreException
return text_obj.get(lang, text_obj[list(text_obj.keys())[0]])
except:
return text
# def get_local_text(text: str, lang: str):
# try:
# text_obj = json.loads(text) #@IgnoreException
# return text_obj.get(lang, text_obj[list(text_obj.keys())[0]])
# except:
# return text
def get_value_repr(value: Any, field_descriptor: EntityFieldDescriptor, locale: str | None = None) -> str:
type_ = field_descriptor.type_
origin = get_origin(type_)
type_ = field_descriptor.type_base
if value is None:
return ""
if origin == UnionType:
args = get_args(type_)
if args[1] == NoneType:
type_ = args[0]
if isinstance(value, bool):
return "【✔︎】" if value else "【 】"
elif origin == list:
arg_type = None
args = get_args(type_)
if args:
arg_type = args[0]
if arg_type and issubclass(arg_type, BotEntity):
if locale and arg_type.bot_entity_descriptor.fields_descriptors["name"].localizable:
return "[" + ", ".join([get_local_text(value = item.name, locale = locale) for item in value]) + "]"
elif field_descriptor.is_list:
if issubclass(type_, BotEntity):
if locale and type_.bot_entity_descriptor.fields_descriptors["name"].localizable:
return "[" + ", ".join([get_local_text(text = item.name, locale = locale) for item in value]) + "]"
else:
return "[" + ", ".join([str(item.name) for item in value]) + "]"
elif arg_type and issubclass(arg_type, BotEnum):
elif issubclass(type_, BotEnum):
return "[" + ", ".join(item.localized(locale) for item in value) + "]"
elif arg_type == str:
elif type_ == str:
return "[" + ", ".join([f"\"{item}\"" for item in value]) + "]"
else:
return "[" + ", ".join([str(item) for item in value]) + "]"
elif issubclass(type_, BotEntity):
if type_.bot_entity_descriptor.fields_descriptors["name"].localizable:
return get_local_text(value = value.name, locale = locale)
return get_local_text(text = value.name, locale = locale)
return value.name
elif issubclass(type_, BotEnum):
return value.localized(locale)
elif isinstance(value, str):
if field_descriptor and field_descriptor.localizable:
return get_local_text(value, locale)
return get_local_text(text = value, locale = locale)
return value
elif isinstance(value, int):
return str(value)
@@ -92,7 +94,13 @@ def get_callable_str(callable_str: str | LazyProxy | EntityCaptionCallable | Ent
elif isinstance(callable_str, LazyProxy):
return callable_str.value
elif callable(callable_str):
return callable_str(*(descriptor, entity, value))
args = signature(callable_str).parameters
if len(args) == 1:
return callable_str(descriptor)
elif len(args) == 2:
return callable_str(descriptor, entity)
elif len(args) == 3:
return callable_str(descriptor, entity, value)
async def authorize_command(user: UserBase,
@@ -100,24 +108,24 @@ async def authorize_command(user: UserBase,
if (callback_data.command == CallbackCommand.MENU_ENTRY_PARAMETERS or
callback_data.context == CommandContext.SETTING_EDIT):
allowed_roles = (await Settings.get(Settings.SECURITY_SETTINGS_ROLES))
allowed_roles = (await Settings.get(Settings.SECURITY_PARAMETERS_ROLES))
return any(role in user.roles for role in allowed_roles)
return False
def get_entity_descriptor(app: QBotApp, callback_data: ContextData) -> EntityDescriptor | None:
def get_entity_descriptor(app: "QBotApp", callback_data: ContextData) -> EntityDescriptor | None:
if callback_data.entity_name:
return app.entity_metadata.entity_descriptors[callback_data.entity_name]
return None
def get_field_descriptor(app: QBotApp, callback_data: ContextData) -> EntityFieldDescriptor | None:
def get_field_descriptor(app: "QBotApp", callback_data: ContextData) -> EntityFieldDescriptor | None:
if callback_data.context == CommandContext.SETTING_EDIT:
return Settings.list_params()[callback_data.field_name]
elif callback_data.context in [CommandContext.ENTITY_CREATE, CommandContext.ENTITY_EDIT]:
elif callback_data.context in [CommandContext.ENTITY_CREATE, CommandContext.ENTITY_EDIT, CommandContext.ENTITY_FIELD_EDIT]:
entity_descriptor = get_entity_descriptor(app, callback_data)
if entity_descriptor:
return entity_descriptor.fields_descriptors.get(callback_data.field_name)
@@ -192,4 +200,144 @@ def add_pagination_controls(keyboard_builder: InlineKeyboardBuilder,
data = str(total_pages) if page != total_pages else "skip",
save_state = True).pack()))
keyboard_builder.row(*navigation_buttons)
keyboard_builder.row(*navigation_buttons)
def add_filter_controls(keyboard_builder: InlineKeyboardBuilder,
entity_descriptor: EntityDescriptor,
filter: str = None,
page: int = 1):
field_name_descriptor = entity_descriptor.fields_descriptors["name"]
if field_name_descriptor.caption:
caption = get_callable_str(field_name_descriptor.caption, field_name_descriptor)
else:
caption = field_name_descriptor.name
keyboard_builder.row(
InlineKeyboardButton(
text = f"🔎 {caption}{f": \"{filter}\"" if filter else ""}",
callback_data = ContextData(
command = CallbackCommand.VIEW_FILTER_EDIT,
entity_name = entity_descriptor.name,
data = str(page)).pack()))
@router.callback_query(ContextData.filter(F.command == CallbackCommand.VIEW_FILTER_EDIT))
async def view_filter_edit(query: CallbackQuery, **kwargs):
callback_data: ContextData = kwargs["callback_data"]
state: FSMContext = kwargs["state"]
state_data = await state.get_data()
kwargs["state_data"] = state_data
args = callback_data.data.split("&")
page = int(args[0])
cmd = None
if len(args) > 1:
cmd = args[1]
db_session: AsyncSession = kwargs["db_session"]
app: "QBotApp" = kwargs["app"]
user: UserBase = kwargs["user"]
entity_descriptor = get_entity_descriptor(app = app, callback_data = callback_data)
if cmd in ["cancel", "clear"]:
if cmd == "clear":
await ViewSetting.set_filter(session = db_session, user_id = user.id, entity_name = entity_descriptor.class_name, filter = None)
context_data_bak = state_data.pop("context_data_bak")
if context_data_bak:
state_data["context_data"] = context_data_bak
context_data = ContextData.unpack(context_data_bak)
field_descriptor = get_field_descriptor(app, context_data)
edit_prompt = state_data["edit_prompt"]
current_value = await deserialize(session = db_session,
type_ = field_descriptor.type_,
value = state_data["value"])
page = int(state_data.pop("page"))
kwargs.pop("callback_data")
return await render_entity_picker(field_descriptor = field_descriptor,
message = query,
callback_data = context_data,
current_value = current_value,
edit_prompt = edit_prompt,
page = page,
**kwargs)
else:
return await route_callback(message = query, back = False, **kwargs)
#await save_navigation_context(callback_data = callback_data, state = state)
old_context_data = state_data.get("context_data")
await state.update_data({"context_data": callback_data.pack(),
"context_data_bak": old_context_data,
"page": page})
send_message = get_send_message(query)
await send_message(text = await Settings.get(Settings.APP_STRINGS_VIEW_FILTER_EDIT_PROMPT),
reply_markup = InlineKeyboardBuilder().row(
InlineKeyboardButton(
text = await Settings.get(Settings.APP_STRINGS_CANCEL_BTN),
callback_data = ContextData(
command = CallbackCommand.VIEW_FILTER_EDIT,
entity_name = entity_descriptor.name,
data = f"{page}&cancel").pack()),
InlineKeyboardButton(
text = await Settings.get(Settings.APP_STRINGS_CLEAR_BTN),
callback_data = ContextData(
command = CallbackCommand.VIEW_FILTER_EDIT,
entity_name = entity_descriptor.name,
data = f"{page}&clear").pack())).as_markup())
@router.message(CallbackCommandFilter(command = CallbackCommand.VIEW_FILTER_EDIT))
async def view_filter_edit_input(message: Message, **kwargs):
state: FSMContext = kwargs["state"]
state_data = await state.get_data()
kwargs["state_data"] = state_data
callback_data = ContextData.unpack(state_data["context_data"])
db_session: AsyncSession = kwargs["db_session"]
user: UserBase = kwargs["user"]
app: "QBotApp" = kwargs["app"]
entity_descriptor = get_entity_descriptor(app = app, callback_data = callback_data)
filter = message.text
await ViewSetting.set_filter(session = db_session, user_id = user.id, entity_name = entity_descriptor.class_name, filter = filter)
#state_data.pop("context_data")
#return await route_callback(message = message, back = False, **kwargs)
context_data_bak = state_data.pop("context_data_bak")
if context_data_bak:
state_data["context_data"] = context_data_bak
context_data = ContextData.unpack(context_data_bak)
field_descriptor = get_field_descriptor(app, context_data)
edit_prompt = state_data["edit_prompt"]
current_value = await deserialize(session = db_session,
type_ = field_descriptor.type_,
value = state_data["value"])
page = int(state_data.pop("page"))
return await render_entity_picker(field_descriptor = field_descriptor,
message = message,
callback_data = context_data,
current_value = current_value,
edit_prompt = edit_prompt,
page = page,
**kwargs)
else:
return await route_callback(message = message, back = False, **kwargs)
from ..navigation import route_callback, save_navigation_context, clear_state, get_navigation_context
from ..editors.entity import render_entity_picker