It is still getting developed.

Changelog:
- Added all features of bozenka
- Added features_list.py
- Some refactors and starting writing code of registering all features handlers
This commit is contained in:
kittyneverdies 2024-02-08 16:51:08 +03:00
parent 4839b94cbc
commit 348a8c7245
19 changed files with 1189 additions and 136 deletions

View file

@ -1,11 +1,3 @@
from typing import Callable
from aiogram import Dispatcher
from aiogram.filters import CommandObject
from aiogram.types import InlineKeyboardMarkup, Message, CallbackQuery
from sqlalchemy.ext.asyncio import async_sessionmaker
from bozenka.database.tables.telegram import TelegramChatSettings
class BasicFeature:
@ -13,60 +5,29 @@ class BasicFeature:
A classic class of lineral (basic)
feature of bozenka. IN FUTURE!
"""
cmd_description: str = "Your description of command"
'''
Here is example of code you can add:
async def generate_telegram_inline_keyboard(self) -> InlineKeyboardMarkup:
"""
Generates a special telegram keyboard (menu)
:return: Inline Keyboard
"""
pass
async def telegram_command_handler(self, msg: Message, cmd: CommandObject, session_maker: async_sessionmaker) -> None:
"""
A special telegram handler for command (if exist)
It is just an example
:param msg: Telegram message object
:param cmd: Aiogram command object
:param session_maker: Async session maker object of SQLAlchemy
:return: Nothing
"""
pass
async def telegram_callback_handler(self, call: CallbackQuery, callback_data, session_maker: async_sessionmaker) -> None:
"""
A special telegram handler for command (if exist)
It is just an example
:param call: Telegram callbackquery object
:param callback_data: A custom callback data created by callback factory
:param session_maker: Async session maker object of SQLAlchemy
:return: Nothing
"""
pass
'''
def __init__(self):
"""
All information about feature
will be inside this function
"""
self.cmd_description: str = "Your description of command"
# Telegram feature settings
self.telegram_setting = TelegramChatSettings.text_generation
self.telegram_commands: list[str | None] = ["test"]
self.telegram_cmd_avaible = True # Is a feature have a commands
self.telegram_callback_factory = None
self.telegram_message_handlers = {
"""
Format is
Handler: [Filters]
"""
}
self.telegram_callback_handlers = {
"""
Format is
Handler: [Filters]
"""
# Telegram setting info
self.telegram_setting_in_list = False # Does feature shows in /setup list
self.telegram_setting_name = None # Setting title in /setup command
self.telegram_setting_description = None # Setting description in /setup command
self.telegram_db_name = None # Name of TelegramChatSettings column will be affected
# Telegram commands list of feature
self.telegram_commands: dict[str: str] = {
#
# Format is "CommandNameHere": "Command description is here"
#
"example": "Its an example"
}
self.telegram_cmd_avaible = True # Does this feature have a telegram commands
# All handlers
self.telegram_message_handlers = (
# Format is [Handler, [Filters]]
)
self.telegram_callback_handlers = (
# Format is [Handler, [Filters]]
)

View file

@ -0,0 +1,4 @@
from invite import Invite
from moderation import Moderation
from pins import Pins
from topics import Threads

View file

@ -0,0 +1,83 @@
import logging
from aiogram.enums import ChatMemberStatus
from aiogram.filters import Command
from aiogram.types import Message, CallbackQuery
from bozenka.database.tables.telegram import TelegramChatSettings
from bozenka.features import BasicFeature
from bozenka.instances.telegram.utils.callbacks_factory import RevokeCallbackData
from bozenka.instances.telegram.utils.keyboards import invite_keyboard
from bozenka.instances.telegram.utils.simpler import ru_cmds
class Invite(BasicFeature):
"""
A class with information about invite feature
All codes will be here
"""
@staticmethod
async def telegram_invite_command_handler(msg: Message) -> None:
"""
Generating invite to group by /invite command
:param msg: Message telegram object
:return: None
"""
logging.log(msg=f"Generating invite for user_id={msg.from_user.id}",
level=logging.INFO)
link = await msg.chat.create_invite_link()
await msg.answer(
f"<em> Держите ваше приглашение в чат, {msg.from_user.mention_html('пользователь')} 👋</em>",
reply_markup=invite_keyboard(link=str(link.invite_link), admin_id=msg.from_user.id,
chat_name=msg.chat.full_name)
)
@staticmethod
async def telegram_revoke_callback_handler(call: CallbackQuery, callback_data: RevokeCallbackData) -> None:
"""
Handler of CallbackQuery, revokes link after pressing button
:param call: CallbackQuery aioram object
:param callback_data: RevokeCallbackData object
:return: Nothing
"""
user_clicked = await call.message.chat.get_member(call.from_user.id)
if callback_data.admin_id != call.from_user.id and \
user_clicked.status != ChatMemberStatus.ADMINISTRATOR and user_clicked.status == ChatMemberStatus.CREATOR:
return
logging.log(msg=f"Revoking link for user_id={call.from_user.id}",
level=logging.INFO)
await call.message.chat.revoke_invite_link(invite_link="https://" + str(callback_data.link))
await call.answer("Удача ✅")
await call.message.delete()
def __init__(self):
"""
All information about feature
will be inside this function
"""
super().__init__()
"""
Telegram feature settings
"""
# Telegram setting info
self.telegram_setting_in_list = True
self.telegram_setting_name = "Приглашения в Чат ✉"
self.telegram_setting_description = "<b>Генератор приглашения в Чат ✉</b>\n" \
"Разрешает использование комманды <code>/invite</code> в чате, для созданния приглашений.\n" \
"Для исполнения <b>требует соответсвующих прав от пользователя и их наличие у бота.</b>"
self.telegram_db_name = TelegramChatSettings.invite_generator
# Telegram commands
self.telegram_commands: dict[str: str] = {"/invite": 'Generates invite into current chat'}
self.telegram_cmd_avaible = True # Is a feature have a commands
# List of aiogram handlers
self.telegram_message_handlers = (
# Format is [Handler, [Filters]]
[self.telegram_invite_command_handler, [Command(commands=["invite"])]]
)
self.telegram_callback_handlers = (
# Format is [Handler, [Filters]]
[self.telegram_revoke_callback_handler, [RevokeCallbackData.filter()]]
)

View file

@ -1,21 +1,18 @@
import logging
from aiogram import F
from aiogram.enums import ChatType, ChatMemberStatus
from aiogram.filters import Command, CommandObject
from aiogram.types import InlineKeyboardMarkup, Message, CallbackQuery, InlineKeyboardButton
from aiogram.utils.keyboard import InlineKeyboardBuilder
from aiogram.enums import ChatMemberStatus
from aiogram.filters import CommandObject, Command
from aiogram.types import Message, CallbackQuery
from sqlalchemy.ext.asyncio import async_sessionmaker
from bozenka.database.tables.telegram import get_chat_config_value
from bozenka.database.tables.telegram import get_chat_config_value, TelegramChatSettings
from bozenka.features import BasicFeature
from bozenka.instances.telegram.utils.callbacks_factory import HelpCategory, HelpBackCategory, HelpFeature, HelpBack, \
UnbanData, BanData, UnmuteData, MuteData
from bozenka.instances.telegram.utils.keyboards import help_category_keyboard, help_keyboard, \
help_feature_keyboard, gpt_categories_keyboard, ban_keyboard, delete_keyboard, mute_keyboard, unmute_keyboard, \
from bozenka.instances.telegram.utils.callbacks_factory import UnbanData, BanData, UnmuteData, MuteData
from bozenka.instances.telegram.utils.filters import IsAdminFilter, BotHasPermissions, UserHasPermissions
from bozenka.instances.telegram.utils.keyboards import ban_keyboard, delete_keyboard, mute_keyboard, unmute_keyboard, \
unban_keyboard
from bozenka.instances.telegram.utils.simpler import list_of_features, SolutionSimpler
from bozenka.instances.version import build, is_updated
class Moderation(BasicFeature):
@ -23,7 +20,6 @@ class Moderation(BasicFeature):
A class of moderation related feature
All staff related to it will be here
"""
cmd_description: str = "Basic command to show main menu"
@staticmethod
async def telegram_ban_callback_handler(call: CallbackQuery, callback_data: BanData,
@ -455,15 +451,49 @@ class Moderation(BasicFeature):
will be inside this function
"""
super().__init__()
self.cmd_description: str = "Your description of command"
# Telegram feature settings
self.telegram_setting = None
self.telegram_commands: list[str | None] = ["start"]
# Telegram setting info
self.telegram_setting_in_list = True
self.telegram_setting_name = "Модерация чата 🕵️"
self.telegram_setting_description = "<b>Модерация чата</b>🕵️\nДанная настройка включает следущие комманды:" \
"\n<pre>/ban [время блокировки] [причина блокировки] - блокировка пользователя" \
"\n/unban - разблокировка пользователя\n" \
"/mute [время мута] [причина мута] - мут пользователя\n" \
"/unmute - Размут пользователя</pre>\n" \
"Время обозначается как:" \
"<pre>1h - один час, " \
"1d - один день, " \
"1m - одна минута, " \
"1s - одна секунда</pre>\n" \
"Для того, " \
"чтобы выполнить одну из комманд по отношению к пользователю, " \
"ответьте на сообщение пользователя и используйте команду\n" \
"Для исполнения <b>требует соответсвующих прав от пользователя и их наличие у бота.</b>"
self.telegram_db_name = TelegramChatSettings.moderation
# Telegram commands
self.telegram_commands: dict[str: str] = {
"ban": "Command to ban user in chat",
'unban': 'Command to unban user in chat',
'mute': 'Command to mute user in chat',
'unmute': 'Command to unmute user in chat',
}
self.telegram_cmd_avaible = True # Is a feature have a commands
self.telegram_callback_factory = None
self.telegram_message_handlers = {
# Your message handlers
}
self.telegram_callback_handlers = {
# Start menu
}
# All handlers
self.telegram_message_handlers = (
# Format is [Handler, [Filters]]
[self.telegram_ban_cmd_handler, [Command(commands="ban"),
IsAdminFilter(True), F.reply_to_message.text]],
[self.telegram_unban_cmd_handler, [Command(commands="unban"),
IsAdminFilter(True), F.reply_to_message.text]],
[self.telegram_mute_cmd_handler, [Command(commands=["mute", "re"]),
UserHasPermissions(["can_restrict_members"]),
BotHasPermissions(["can_restrict_members"])]],
[self.telegram_unmute_cmd_handler, [Command(commands=["unmute"]),
UserHasPermissions(["can_restrict_members"]),
BotHasPermissions(["can_restrict_members"])]]
)
self.telegram_callback_handlers = (
# Format is [Handler, [Filters]]
[self.telegram_ban_callback_handler, [BanData.filter()]],
[self.telegram_unban_callback_handler, [UnbanData.filter()]],
[self.telegram_mute_callback_handler, [MuteData.filter()]],
[self.telegram_unmute_callback_handler, [UnmuteData.filter()]])

View file

@ -1,7 +1,11 @@
from aiogram import F
from aiogram.filters import Command
from aiogram.types import Message, CallbackQuery
from bozenka.database.tables.telegram import TelegramChatSettings
from bozenka.features import BasicFeature
from bozenka.instances.telegram.utils.callbacks_factory import PinMsg, UnpinMsg
from bozenka.instances.telegram.utils.filters import UserHasPermissions, BotHasPermissions, IsAdminFilter
from bozenka.instances.telegram.utils.keyboards import unpin_msg_keyboard, delete_keyboard, pin_msg_keyboard
from bozenka.instances.telegram.utils.simpler import SolutionSimpler
@ -92,13 +96,31 @@ class Pins(BasicFeature):
super().__init__()
self.cmd_description: str = "Your description of command"
# Telegram feature settings
self.telegram_setting = None
self.telegram_commands: list[str | None] = ["start"]
self.telegram_setting_in_list = True
self.telegram_setting_name = "Закреп 📌"
self.telegram_setting_description = "<b>Закреп</b>📌" \
"\nДанная функция включает команды:" \
"<pre>/pin - закрепляет сообщение\n" \
"/unpin - открепляет сообщение\n" \
"/unpin_all - открепляет все сообщения, которые видит бот</pre>\n" \
"Для исполнения <b>требует соответсвующих прав от пользователя и их наличие у бота.</b>"
self.telegram_db_name = TelegramChatSettings.pins
# Telegram commands
self.telegram_commands: dict[str: str] = {
'pin': 'Pin fast any message in chat',
'unpin': 'Unpin fast any message in chat',
}
self.telegram_cmd_avaible = True # Is a feature have a commands
self.telegram_callback_factory = None
# Telegram Handler
self.telegram_message_handlers = {
self.telegram_pin_cmd: [Command(commands="pin"), UserHasPermissions(["can_pin_messages"]),
BotHasPermissions(["can_pin_messages"]), F.reply_to_message],
self.telegram_unpin_cmd: [Command(commands="unpin"), UserHasPermissions(["can_pin_messages"]),
BotHasPermissions(["can_pin_messages"]), F.reply_to_message],
self.telegram_unpinall_cmd: [Command(commands="unpin_all"), IsAdminFilter(True),
BotHasPermissions(["can_pin_messages"]), F.reply_to_message.text],
}
self.telegram_callback_handlers = {
self.telegram_pin_callback_handler: [PinMsg.filter()],
self.telegram_unpin_callback_handler: [UnpinMsg.filter()]
}

View file

@ -1,17 +1,13 @@
from aiogram import F, Bot
from aiogram.enums import ChatType
from aiogram import F
from aiogram.filters import Command
from aiogram.types import InlineKeyboardMarkup, Message, CallbackQuery, InlineKeyboardButton
from aiogram.utils.keyboard import InlineKeyboardBuilder
from aiogram.types import Message, CallbackQuery
from bozenka.database.tables.telegram import TelegramChatSettings
from bozenka.features import BasicFeature
from bozenka.instances.telegram.utils.callbacks_factory import HelpCategory, HelpBackCategory, HelpFeature, HelpBack, \
PinMsg, UnpinMsg, CloseThread, OpenThread
from bozenka.instances.telegram.utils.keyboards import help_category_keyboard, help_keyboard, \
help_feature_keyboard, gpt_categories_keyboard, unpin_msg_keyboard, delete_keyboard, pin_msg_keyboard, \
close_thread_keyboard, open_thread_keyboard
from bozenka.instances.telegram.utils.simpler import list_of_features, SolutionSimpler
from bozenka.instances.version import build, is_updated
from bozenka.instances.telegram.utils.callbacks_factory import CloseThread, OpenThread
from bozenka.instances.telegram.utils.filters import UserHasPermissions, BotHasPermissions
from bozenka.instances.telegram.utils.keyboards import delete_keyboard, close_thread_keyboard, open_thread_keyboard
from bozenka.instances.telegram.utils.simpler import SolutionSimpler
class Threads(BasicFeature):
@ -21,11 +17,10 @@ class Threads(BasicFeature):
"""
@staticmethod
async def telegram_close_topic_cmd_handler(msg: Message, bot: Bot) -> None:
async def telegram_close_topic_cmd_handler(msg: Message) -> None:
"""
/close command function. Closing thread
:param msg: Message telegram object
:param bot: Object of telegram bot
:return: Nothing
"""
config = await SolutionSimpler.close_topic(msg=msg)
@ -34,11 +29,10 @@ class Threads(BasicFeature):
if config[1] else delete_keyboard(msg.from_user.id))
@staticmethod
async def telegram_reopen_topic_cmd_handler(msg: Message, bot: Bot) -> None:
async def telegram_reopen_topic_cmd_handler(msg: Message) -> None:
"""
/open command function. Opens thread
:param msg: Message telegram object
:param bot: Object of telegram bot
:return: Nothing
"""
config = await SolutionSimpler.open_topic(msg=msg)
@ -47,11 +41,10 @@ class Threads(BasicFeature):
if config[1] else delete_keyboard(msg.from_user.id))
@staticmethod
async def telegram_close_general_topic_cmd_handler(msg: Message, bot: Bot) -> None:
async def telegram_close_general_topic_cmd_handler(msg: Message) -> None:
"""
/close_general command function. Closes general thread
:param msg: Message telegram object
:param bot: Object of telegram bot
:return: Nothing
"""
config = await SolutionSimpler.close_general_topic(msg=msg)
@ -60,11 +53,10 @@ class Threads(BasicFeature):
if config[1] else delete_keyboard(msg.from_user.id))
@staticmethod
async def telegram_reopen_general_topic_cmd(msg: Message, bot: Bot) -> None:
async def telegram_reopen_general_topic_cmd(msg: Message) -> None:
"""
/open_general command function. Opens general thread
:param msg: Message telegram object
:param bot: Object of telegram bot
:return: Nothing
"""
config = await SolutionSimpler.open_general_topic(msg=msg)
@ -73,11 +65,10 @@ class Threads(BasicFeature):
if config[1] else delete_keyboard(msg.from_user.id))
@staticmethod
async def telegram_hide_general_topic_cmd_handler(msg: Message, bot: Bot) -> None:
async def telegram_hide_general_topic_cmd_handler(msg: Message) -> None:
"""
/hide_general command function. Hides general thread
:param msg: Message telegram object
:param bot: Object of telegram bot
:return: Nothing
"""
config = await SolutionSimpler.hide_general_topic(msg=msg)
@ -85,11 +76,10 @@ class Threads(BasicFeature):
reply_markup=delete_keyboard(msg.from_user.id))
@staticmethod
async def telegram_unhide_general_topic_cmd(msg: Message, bot: Bot) -> None:
async def telegram_unhide_general_topic_cmd(msg: Message) -> None:
"""
/show_general command function. Shows back general thread.
:param msg: Message telegram object
:param bot: Object of telegram bot
:return: Nothing
"""
config = await SolutionSimpler.show_general_topic(msg=msg)
@ -138,15 +128,51 @@ class Threads(BasicFeature):
will be inside this function
"""
super().__init__()
self.cmd_description: str = "Your description of command"
# Telegram feature settings
self.telegram_setting = None
self.telegram_commands: list[str | None] = ["start"]
# Telegram setting info
self.telegram_setting_in_list = True
self.telegram_setting_name = "Работа с Форумом 💬"
self.telegram_setting_description = "<b>Работа с Форумом</b>💬\nДанная настройка включает следущие комманды:\n" \
"<pre>/open - открывают тему форума\n" \
"/close - закрывают тему форума\n" \
"/open_general - открывают основную тему форума\n" \
"/close_general - закрывает основную тему форума\n" \
"/hide_general - прячет основную тему форума\n" \
"/show_general - показывает основную тему форума</pre>\n" \
"Для исполнения <b>требует соответсвующих прав от пользователя и их наличие у бота. Также должен быть" \
"включен форум</b>"
self.telegram_db_name = TelegramChatSettings.topics
# Telegram commands
self.telegram_commands: dict[str: str] = {
'close': 'Close fast topic (not general) in chat',
'open': 'Open fast topic (not general) in chat',
'hide_general': 'Hide general topic in chat',
'show_general': 'Show general topic in chat',
"close_general": 'Closes general topic in chat',
"open_general": 'Opens general topic in chat',
}
self.telegram_cmd_avaible = True # Is a feature have a commands
self.telegram_callback_factory = None
# All handlers
self.telegram_message_handlers = {
self.telegram_close_topic_cmd_handler: [Command(commands=["close_topic", "close"]),
UserHasPermissions(["can_manage_topics"]),
BotHasPermissions(["can_manage_topics"]), F.chat.is_forum],
self.telegram_reopen_topic_cmd_handler: [Command(commands=["reopen_topic", "open_topic", "open"]),
UserHasPermissions(["can_manage_topics"]),
BotHasPermissions(["can_manage_topics"]), F.chat.is_forum],
self.telegram_close_general_topic_cmd_handler: [Command(commands=["close_general"]),
UserHasPermissions(["can_manage_topics"]),
BotHasPermissions(["can_manage_topics"]), F.chat.is_forum],
self.telegram_reopen_general_topic_cmd: [Command(commands=["reopen_general", "open_general"]),
UserHasPermissions(["can_manage_topics"]),
BotHasPermissions(["can_manage_topics"]), F.chat.is_forum],
self.telegram_hide_general_topic_cmd_handler: [Command(commands=["hide_general"]),
UserHasPermissions(["can_manage_topics"]),
BotHasPermissions(["can_manage_topics"]), F.chat.is_forum],
self.telegram_unhide_general_topic_cmd: [Command(commands=["unhide_general", "show_general"]),
UserHasPermissions(["can_manage_topics"]),
BotHasPermissions(["can_manage_topics"]), F.chat.is_forum]
}
self.telegram_callback_handlers = {
self.telegram_close_thread_callback_handler: [CloseThread.filter()],
self.telegram_reopen_topic_cmd_handler: [OpenThread.filter()]
}

View file

@ -0,0 +1,2 @@
from setup import Setup
from start import Start

View file

@ -0,0 +1,103 @@
from aiogram import F
from aiogram.enums import ChatType
from aiogram.filters import Command
from aiogram.types import Message, CallbackQuery
from sqlalchemy import Update
from sqlalchemy.ext.asyncio import async_sessionmaker
from bozenka.database.tables.telegram import get_chat_config_value, TelegramChatSettings
from bozenka.features import BasicFeature
from bozenka.instances.telegram.utils.callbacks_factory import SetupAction, SetupFeature, SetupCategory
from bozenka.instances.telegram.utils.keyboards import setup_keyboard, setup_category_keyboard, setup_feature_keyboard
from bozenka.instances.telegram.utils.simpler import list_of_features
class Setup(BasicFeature):
"""
A class of /setup command
All staff related to it will be here
"""
@staticmethod
async def telegram_setup_cmd_handler(msg: Message) -> None:
"""
/setup telegram handler
:param msg: Telegram message object
:return: Nothing
"""
await msg.answer("Привет владелец чата 👋\n"
"Чтобы меня настроить, используй меню под данным сообщением",
reply_markup=setup_keyboard())
@staticmethod
async def telegram_setup_categories_handler(call: CallbackQuery, callback_data: SetupCategory | SetupAction):
"""
Query, what shows list of features to enable.
:param call:
:param callback_data:
:return:
"""
await call.message.edit_text("Выберите настройку, которую хотите изменить",
reply_markup=setup_category_keyboard(category=callback_data.category_name))
@staticmethod
async def telegram_setup_edit_feature_handler(call: CallbackQuery, callback_data: SetupFeature, session_maker: async_sessionmaker):
"""
Query, what shows menu to enable / disable feature
:param call:
:param callback_data:
:param session_maker:
:return:
"""
is_enabled = await get_chat_config_value(
chat_id=call.message.chat.id,
session=session_maker,
setting=list_of_features[callback_data.feature_category][callback_data.feature_index]
)
await call.message.edit_text(
list_of_features[callback_data.feature_category][callback_data.feature_index].description,
reply_markup=await setup_feature_keyboard(category=callback_data.feature_category,
index=callback_data.feature_index,
is_enabled=is_enabled))
@staticmethod
async def telegram_features_edit_handler(call: CallbackQuery, callback_data: SetupAction, session_maker: async_sessionmaker):
"""
Query, what shows menu to enable / disable feature
after editing
:param call:
:param callback_data:
:param session_maker:
:return:
"""
async with session_maker() as session:
async with session.begin():
await session.execute(Update(TelegramChatSettings)
.values(
{list_of_features[callback_data.category_name][callback_data.feature_index].settings_name: callback_data.action == "enable"})
.where(TelegramChatSettings.chat_id == call.message.chat.id))
await call.message.edit_text(
list_of_features[callback_data.category_name][callback_data.feature_index].description,
reply_markup=await setup_feature_keyboard(category=callback_data.category_name,
index=callback_data.afeature_index,
is_enabled=callback_data.action == "enable"))
def __init__(self):
"""
All information about feature
will be inside this function
"""
super().__init__()
# Telegram feature settings
self.telegram_setting_in_list = False
self.telegram_commands = {"setup": 'Command to setup bozenka features in chat'}
self.telegram_cmd_avaible = True
self.telegram_message_handlers = {
self.telegram_setup_cmd_handler: [Command(commands=["setup"]), ~(F.chat.type == ChatType.PRIVATE)]
}
self.telegram_callback_handlers = {
self.telegram_features_edit_handler: [SetupAction.filter(F.action == "disable" or F.action == "enable")],
self.telegram_setup_edit_feature_handler: [SetupFeature.filter()],
self.telegram_setup_categories_handler: [SetupAction.filter(F.action == "back")]
}

View file

@ -201,10 +201,9 @@ class Start(BasicFeature):
will be inside this function
"""
super().__init__()
self.cmd_description: str = "Your description of command"
# Telegram feature settings
self.telegram_setting = None
self.telegram_commands: list[str | None] = ["start"]
self.telegram_commands: dict[str: str] = {"start": "Command to start work with bozenka the bot"}
self.telegram_cmd_avaible = True # Is a feature have a commands
self.telegram_callback_factory = None
self.telegram_message_handlers = {

View file

@ -0,0 +1,30 @@
from aiogram import Dispatcher
from bozenka.features import BasicFeature
from bozenka.features.admin import *
from bozenka.features.user import *
from bozenka.features.basic import *
features_list = [
# Admin related category
Moderation,
Invite,
Pins,
Threads,
# User related category
ImageGeneratrion,
TextGeneratrion,
Welcome,
# Basic Functions
Setup,
Start
]
def register_all_features(features_list: list[BasicFeature], dispatcher: Dispatcher) -> None:
"""
Registers all features / handlers avaible in bozenka
:param features_list: List of features
:return: None
"""
pass

View file

@ -0,0 +1,3 @@
from image_generation import ImageGeneratrion
from text_generation import TextGeneratrion
from welcome import Welcome

View file

@ -0,0 +1,157 @@
import logging
from typing import Callable
from aiogram import Dispatcher
from aiogram.filters import CommandObject, Command
from aiogram.fsm.context import FSMContext
from aiogram.types import InlineKeyboardMarkup, Message, CallbackQuery, FSInputFile
from sqlalchemy.ext.asyncio import async_sessionmaker
from bozenka.database.tables.telegram import TelegramChatSettings
from bozenka.features import BasicFeature
from bozenka.generative.kadinsky import kadinsky_gen
from bozenka.instances.telegram.utils.callbacks_factory import ImageGenerationCategory, ImageGeneration
from bozenka.instances.telegram.utils.keyboards import image_resolution_keyboard, delete_keyboard, \
image_generation_keyboard, image_response_keyboard
from bozenka.instances.telegram.utils.simpler import GeneratingImages
class ImageGeneratrion(BasicFeature):
"""
A classic class of lineral (basic)
feature of bozenka. IN FUTURE!
"""
cmd_description: str = "Your description of command"
@staticmethod
async def telegram_select_image_size_handler(call: CallbackQuery, callback_data: ImageGenerationCategory,
state: FSMContext) -> None:
"""
Query, what shows menu for image size to generate in
:param call: CallbackQuery object
:param callback_data: ImageGenerationCategory
:param state: FSMContext aiogram object
:return: None
"""
if call.from_user.id != callback_data.user_id:
return
await state.update_data(set_category=callback_data.category)
await state.set_state(GeneratingImages.set_size)
await call.message.edit_text("Пожалуста, выберите размер изображения 🖼",
reply_markup=image_resolution_keyboard(user_id=call.from_user.id,
category=callback_data.category))
@staticmethod
async def telegram_end_generation_handler(call: CallbackQuery, callback_data: ImageGeneration, state: FSMContext) -> None:
"""
Query, what shows menu for image size to generate in
:param call:
:param callback_data:
:param state:
:return: None
"""
if call.from_user.id != callback_data.user_id:
return
await state.update_data(set_size=callback_data.size)
await state.set_state(GeneratingImages.ready_to_generate)
await call.message.edit_text(
f"Вы выбрали {callback_data.category} для генерации изображений в размере {callback_data.size}.\n"
"Напишите /cancel для отмены",
reply_markup=delete_keyboard(admin_id=call.from_user.id))
@staticmethod
async def telegram_already_generating_handler(msg: Message, state: FSMContext) -> None:
"""
Giving response, if generating image for user right now,
but user still asks something
:param msg: Message telegram object
:param state: FSM state of bot
:return:
"""
await msg.answer(
"Подождите пожалуйста, мы уже генерируем изображение для вас, подождите, когда мы ответим на ваш передыдущий вопрос",
reply_markup=delete_keyboard(admin_id=msg.from_user.id))
@staticmethod
async def telegram_imagine_handler(msg: Message, state: FSMContext) -> None:
"""
/imagine command handler, start menu
:param msg: Message telegram object
:param state: FSM state of bot
:return:
"""
if await state.get_state():
return
await msg.answer("Пожалуста, выберите сервис / модель для генерации изображений",
reply_markup=image_generation_keyboard(user_id=msg.from_user.id))
@staticmethod
async def telegram_kadinsky_generating_handler(msg: Message, state: FSMContext) -> None:
"""
Message handler for kandinsky to generate image by text from message
:param msg: Message telegram object
:param state: FSM state of bot
:return:
"""
await state.set_state(GeneratingImages.generating)
message = await msg.answer("Пожалуйста подождите, мы генерируем изображение ⏰\n"
"Если что-то пойдет не так, мы вам сообщим 👌")
data = await state.get_data()
try:
model_id = kadinsky_gen.get_model()
width, height = data["set_size"].split("x")
generating = kadinsky_gen.generate(model=model_id,
prompt=msg.text,
width=int(width),
height=int(height))
result = kadinsky_gen.check_generation(request_id=generating)
if result:
path = kadinsky_gen.save_image(result[0], f"telegram_{msg.from_user.id}")
photo = FSInputFile(path)
await msg.answer_photo(photo=photo,
caption=msg.text,
reply_markup=image_response_keyboard(user_id=msg.from_user.id))
await message.delete()
else:
await message.edit_text("Простите, произошла ошибка 😔\n"
"Убедитесь, что севрера kadinsky работают и ваш промт не является неподобающим и неприемлимым\n"
"Если это продолжается, пожалуйста используйте /cancel",
reply_markup=image_response_keyboard(user_id=msg.from_user.id))
except Exception as ex:
logging.log(msg=f"Get an exception for generating answer={ex}",
level=logging.ERROR)
finally:
logging.log(msg=f"Generated image for user_id={msg.from_user.id} with promt",
level=logging.INFO)
await state.set_state(GeneratingImages.ready_to_generate)
def __init__(self):
"""
All information about feature
will be inside this function
"""
super().__init__()
# Telegram feature settings
self.telegram_setting = TelegramChatSettings.image_generation
self.telegram_commands: dict[str: str] = {'imagine', 'Starts conversation with image generative ai'}
self.telegram_setting_in_list = True
self.telegram_setting_name = "Генерация изображений 📸"
self.telegram_setting_description = "<b>Генерация изображений </b>🤖" \
"\nНаходится в разработке.\n" \
"На текущий момент есть поддержка:\n" \
"- Kadinksy\n" \
" Следите за обновлениями 😘"
self.telegram_cmd_avaible = True # Is a feature have a commands
self.telegram_message_handlers = {
self.telegram_kadinsky_generating_handler: [GeneratingImages.ready_to_generate, ~Command(commands=["cancel"])],
self.telegram_imagine_handler: [Command(commands=["imagine"])]
}
self.telegram_callback_handlers = {
self.telegram_select_image_size_handler: [ImageGenerationCategory.filter()],
self.telegram_end_generation_handler: [ImageGeneration.filter()]
}

View file

@ -0,0 +1,559 @@
import logging
import g4f
from aiogram import F
from aiogram.filters import Command
from aiogram.fsm.context import FSMContext
from aiogram.types import Message, CallbackQuery
from gpt4all import GPT4All
from bozenka.database.tables.telegram import TelegramChatSettings
from bozenka.features import BasicFeature
from bozenka.generative.gpt4all import model_path, check
from bozenka.generative.gpt4free import generate_gpt4free_providers
from bozenka.instances.telegram.utils.callbacks_factory import Gpt4FreeProvsModelPage, Gpt4FreeProviderPage, \
Gpt4AllSelect, Gpt4AllModel, GptCategory, Gpt4freeResult, \
Gpt4FreeProvider, GptBackMenu, Gpt4FreeModel, Gpt4FreeCategory, Gpt4FreeModelPage, GptStop
from bozenka.instances.telegram.utils.keyboards import delete_keyboard, \
text_response_keyboard, gpt_categories_keyboard, \
gpt4free_models_by_provider_keyboard, gpt4free_providers_keyboard, gpt4all_model_menu, generate_gpt4all_page, \
gpt4free_categories_keyboard, gpt4free_models_keyboard
from bozenka.instances.telegram.utils.simpler import AnsweringGPT4Free, AnsweringGpt4All
class TextGeneratrion(BasicFeature):
"""
A class, what have inside all handlers / functions
related to text generation of bozenka
"""
cmd_description: str = "Your description of command"
@staticmethod
async def telegram_already_answering_handler(msg: Message, state: FSMContext) -> None:
"""
Giving response, if answering user now,
but he still asks something
:param msg: Message telegram object
:param state: FSM state of bot
:return: Nothing
"""
await msg.answer(
"Подождите пожалуйста, мы уже генерируем ответ для вас, подождите, когда мы ответим на ваш передыдущий вопрос",
reply_markup=delete_keyboard(admin_id=msg.from_user.id))
@staticmethod
async def telegram_conversation_cmd_handler(msg: Message, state: FSMContext) -> None:
"""
/conversation command handler, start
:param msg: Message telegram object
:param state: FSM state of bot
:return:
"""
if await state.get_state():
return
await msg.answer("Пожалуста, выберите сервис для ИИ.",
reply_markup=gpt_categories_keyboard
(user_id=msg.from_user.id))
@staticmethod
async def telegram_cancel_cmd_handler(msg: Message, state: FSMContext) -> None:
"""
Canceling dialog with generative model
Works on command /cancel
:param msg: Message telegram object
:param state: FSM state of bot
:return:
"""
current = await state.get_state()
if current is None:
return
await state.clear()
await msg.answer("Удача ✅\n"
"Диалог отменён!", reply_markup=delete_keyboard(admin_id=msg.from_user.id))
# G4F telegram category
# All handlers and other stuff
# All code and comments
@staticmethod
async def telegram_g4f_generate_handler(msg: Message, state: FSMContext) -> None:
"""
Generating answer if GPT4Free model and provider has been selected
:param msg: Message telegram object
:param state: FSM state of bot
:return:
"""
await state.set_state(AnsweringGPT4Free.answering)
info = await state.get_data()
providers = generate_gpt4free_providers()
reply = await msg.reply("Пожалуйста подождите, мы генерируем ответ от провайдера ⏰\n"
"Если что-то пойдет не так, мы вам сообщим 👌")
current_messages = []
if info.get("ready_to_answer"):
for message in info["ready_to_answer"]:
current_messages.append(message)
if not info.get("set_provider"):
info["set_provider"] = None
current_messages.append({"role": "user", "content": msg.text})
response = ""
try:
response = await g4f.ChatCompletion.create_async(
model=info["set_model"],
messages=current_messages,
provider=None if info["set_provider"] is None else providers[info["set_provider"]],
stream=False
)
except NameError or SyntaxError:
try:
response = g4f.ChatCompletion.create(
model=info["set_model"],
messages=current_messages,
provider=None if info["set_provider"] is None else providers[info["set_provider"]],
stream=False
)
except Exception as S:
response = "Простите, произошла ошибка 😔\nЕсли это продолжается, пожалуйста используйте /cancel"
logging.log(msg=f"Get an exception for generating answer={S}",
level=logging.ERROR)
except Exception as S:
response = "Простите, произошла ошибка 😔\nЕсли это продолжается, пожалуйста используйте /cancel"
logging.log(msg=f"Get an exception for generating answer={S}",
level=logging.ERROR)
finally:
await reply.edit_text(text=response, reply_markup=text_response_keyboard(user_id=msg.from_user.id))
current_messages.append({"role": "assistant", "content": response})
await state.update_data(ready_to_answer=current_messages)
await state.set_state(AnsweringGPT4Free.ready_to_answer)
@staticmethod
async def telegram_instart_conversation_handler(call: CallbackQuery, callback_data: GptBackMenu, state: FSMContext) -> None:
"""
Query, what shows when clicking on button in /start menu
:param call: CallbackQuery class
:param state: FSMContext aiogram class
:param callback_data: GptBackMenu class
:return: Nothing
"""
if call.from_user.id != callback_data.user_id or await state.get_state():
return
await call.message.edit_text("Пожалуста, выберите сервис для ИИ.",
reply_markup=gpt_categories_keyboard(user_id=call.from_user.id))
@staticmethod
async def telegram_g4f_providers_handlers(call: CallbackQuery, callback_data: Gpt4FreeCategory,
state: FSMContext) -> None:
"""
Query, what creating providers selecting menu.
:param call: CallbackQuery class
:param state: FSMContext aiogram class
:param callback_data: Gpt4FreeCategory
return: Nothing
"""
if call.from_user.id != callback_data.user_id:
return
logging.log(msg=f"Selected gpt4free category by user_id={call.from_user.id}",
level=logging.INFO)
await call.answer("Вы выбрали провайдеры 🤖")
await state.set_state(AnsweringGPT4Free.set_provider)
await call.message.edit_text("Выберите пожалуйста одного из провайдеров 👨‍💻",
reply_markup=gpt4free_providers_keyboard(user_id=call.from_user.id, page=0))
@staticmethod
async def telegram_g4f_models_handler(call: CallbackQuery, callback_data: GptCategory, state: FSMContext) -> None:
"""
Query, what creating models selecting menu
:param call: CallbackQuery class
:param state: FSMContext aiogram class
:param callback_data: GptCategory
return: Nothing
"""
if call.from_user.id != callback_data.user_id:
return
await state.set_state(AnsweringGPT4Free.set_model)
await call.answer("Вы выбрали модели 🤖")
await call.message.edit_text("Выберите модель, с которой будете общаться 🤖",
reply_markup=gpt4free_models_keyboard(user_id=call.from_user.id, page=0))
@staticmethod
async def telegram_end_g4f_model_handler(call: CallbackQuery, callback_data: Gpt4FreeModel, state: FSMContext) -> None:
"""
Query, what ending g4f model selecting
:param call: CallbackQuery class
:param callback_data: Gpt4FreeModel model
:param state: FSMContext aiogram class
:return: Nothing
"""
if call.from_user.id != callback_data.user_id:
return
await state.update_data(set_model=callback_data.model)
await state.set_state(AnsweringGPT4Free.ready_to_answer)
await call.answer("Вы можете начать общаться 🤖")
await call.message.edit_text("Удача ✅\n"
"Вы теперь можете спокойно вести диалог 🤖\n"
f"Вы выбрали модель <b>{callback_data.model}</b>👾\n"
"Чтобы прекратить общение, используйте /cancel ",
reply_markup=delete_keyboard(admin_id=call.from_user.id))
@staticmethod
async def telegram_g4f_next_model_handler(call: CallbackQuery, callback_data: Gpt4FreeModelPage,
state: FSMContext) -> None:
"""
Query, what creating models selecting menu
:param state: FSMContext aiogram class
:param call: CallbackQuery class
:param callback_data: Gpt4FreeModelPage class
:return: None
"""
if call.from_user.id != callback_data.user_id:
return
await call.answer(f"Вы перелистнули на страницу {callback_data.page + 1}📄")
await call.message.edit_text("Выберите модель, с которой будете общаться 🤖",
reply_markup=gpt4free_models_keyboard(user_id=call.from_user.id,
page=callback_data.page))
@staticmethod
async def telegram_g4f_category_handler(call: CallbackQuery, callback_data: GptCategory, state: FSMContext) -> None:
"""
Query, what creating providers selecting menu.
:param state: FSMContext aiogram class
:param call: CallbackQuery class
:param callback_data: GptCategory class
:return: None
"""
if call.from_user.id != callback_data.user_id:
return
logging.log(msg=f"Selected gpt4free category by user_id={call.from_user.id}",
level=logging.INFO)
await state.update_data(set_category=callback_data.category)
await call.answer("Вы выбрали Gpt4Free 🤖")
await call.message.edit_text("Выберите, по какому пункту мы будем вести диалог с нейронной сети 🤖",
reply_markup=gpt4free_categories_keyboard(user_id=call.from_user.id))
await call.answer("Выберите, по какому пункту мы будем вести диалог с нейронной сети 🤖")
@staticmethod
async def telegram_g4f_back_provider_handler(call: CallbackQuery, callback_data: GptBackMenu,
state: FSMContext) -> None:
"""
Query, what creating providers selecting menu.
:param state: FSMContext aiogram class
:param call: CallbackQuery telegram class
:param callback_data: GptBackMenu class
:return: None
"""
if call.from_user.id != callback_data.user_id:
return
logging.log(msg=f"Back to providers menu by user_id={call.from_user.id}",
level=logging.INFO)
await state.set_state(AnsweringGPT4Free.set_provider)
await call.message.edit_text("Выберите пожалуйста одного из провайдеров 👨‍💻",
reply_markup=gpt4free_providers_keyboard(page=0, user_id=callback_data.user_id))
await call.answer("Выберите пожалуйста одного из провайдеров 👨‍💻")
@staticmethod
async def inline_g4f_provider_models(call: CallbackQuery, callback_data: Gpt4FreeProvider,
state: FSMContext) -> None:
"""
Query, what creating models selecting menu.
:param state: FSMContext aiogram class
:param call: CallbackQuery telegram class
:param callback_data: Gpt4FreeProvider Class
:return: None
"""
if call.from_user.id != callback_data.user_id:
return
logging.log(msg=f"Selected gpt4free provider {callback_data.provider} by user_id={call.from_user.id}",
level=logging.INFO)
await state.update_data(set_provider=callback_data.provider)
await state.set_state(AnsweringGPT4Free.set_model)
await call.message.edit_text("Выберите пожалуйста модель ИИ 👾",
reply_markup=gpt4free_models_by_provider_keyboard(
user_id=callback_data.user_id,
provider=callback_data.provider,
page=0
))
await call.answer("Выберите пожалуйста модель ИИ 👾")
@staticmethod
async def telegram_g4f_ready_handler(call: CallbackQuery, callback_data: Gpt4freeResult, state: FSMContext) -> None:
"""
Query, what says about getting ready to questions for ChatGPT from Gpt4Free.
:param state: FSMContext aiogram class
:param call: CallbackQuery telegram class
:param callback_data: Gpt4freeResult
:return: None
"""
if call.from_user.id != callback_data.user_id:
return
logging.log(msg=f"Selected gpt4free model {callback_data.model} by user_id={call.from_user.id}",
level=logging.INFO)
await state.update_data(set_model=callback_data.model)
await state.set_state(AnsweringGPT4Free.ready_to_answer)
logging.log(msg=f"Loaded GPT answering status for user_id={call.from_user.id}",
level=logging.INFO)
await call.message.edit_text("Удача ✅\n"
"Вы теперь можете спокойно вести диалог 🤖\n"
f"Вы выбрали модель <b>{callback_data.model}</b>👾, от провайдера <b>{callback_data.provider}</b>👨‍💻\n"
"Чтобы прекратить общение, используйте /cancel ",
reply_markup=delete_keyboard(admin_id=callback_data.user_id))
await call.answer("Вы теперь можете спокойно вести диалог 🤖")
@staticmethod
async def telegram_g4f_models_by_provider_handler(call: CallbackQuery, callback_data: Gpt4FreeProvsModelPage,
state: FSMContext) -> None:
"""
Query, what generates a next page of models for user.
:param state: FSMContext aiogram class
:param call: CallbackQuery telegram class
:param callback_data: Gpt4FreeProvsModelPage
:return: None
"""
if call.from_user.id != callback_data.user_id:
return
logging.log(msg=f"Changed page to {str(callback_data.page + 1)} user_id={call.from_user.id}",
level=logging.INFO)
await call.message.edit_text(call.message.text,
reply_markup=gpt4free_models_by_provider_keyboard(
user_id=callback_data.user_id,
provider=callback_data.provider,
page=callback_data.page
))
await call.answer(f"Вы перелистали на страницу {callback_data.page + 1}📄")
@staticmethod
async def telegram_next_g4f_providers_handler(call: CallbackQuery, callback_data: Gpt4FreeProviderPage,
state: FSMContext) -> None:
"""
Query, what generates a next page of providers for user
:param state: FSMContext aiogram class
:param call: CallbackQuery telegram class
:param callback_data: Gpt4FreeProviderPage class
:return: None
"""
if call.from_user.id != callback_data.user_id:
return
logging.log(msg=f"Changed page to {str(callback_data.page + 1)} user_id={call.from_user.id}",
level=logging.INFO)
await call.message.edit_text(call.message.text,
reply_markup=gpt4free_providers_keyboard(user_id=callback_data.user_id,
page=callback_data.page))
await call.answer(f"Вы перелистнули на страницу {callback_data.page + 1}📄")
# G4A telegram handlers section
# All code and commentaries here
# All handlers here
@staticmethod
async def telegram_g4a_generate_handler(msg: Message, state: FSMContext) -> None:
"""
Generating answer if Gpt4All has been selected
:param msg: Message telegram object
:param state: FSM state of bot
:return:
"""
await state.set_state(AnsweringGpt4All.answering)
models = GPT4All.list_models()
info = await state.get_data()
answer = ""
main_msg = await msg.answer("Пожалуйста подождите, мы генерируем вам ответ ⏰\n"
"Если что-то пойдет не так, мы вам сообщим 👌",
reply_markup=text_response_keyboard(user_id=msg.from_user.id))
if not check(models[info["set_model"]]["filename"]):
main_msg = await main_msg.edit_text(main_msg.text + "\nПодождите пожалуста, мы скачиваем модель...",
reply_markup=main_msg.reply_markup)
try:
# Setting Gpt4All model
model = GPT4All(model_name=models[info['set_model']]['filename'],
model_path=model_path,
allow_download=True)
# Setting our chat session if exist
model.current_chat_session = [] if not info.get("ready_to_answer") else info["ready_to_answer"]
# Generating answer
with model.chat_session():
answer = model.generate(msg.text)
await state.update_data(ready_to_answer=model.current_chat_session)
except Exception as S:
answer = "Простите, произошла ошибка 😔\nЕсли это продолжается, пожалуйста используйте /cancel"
logging.log(msg=f"Get an exception for generating answer={S}",
level=logging.ERROR)
finally:
await main_msg.edit_text(answer, reply_markup=text_response_keyboard(user_id=msg.from_user.id))
await state.set_state(AnsweringGpt4All.ready_to_answer)
@staticmethod
async def telegram_g4a_handler(call: CallbackQuery, callback_data: GptCategory, state: FSMContext) -> None:
"""
Query, what shows list for gpt4all models
:param state: FSMContext aiogram class
:param call: CallbackQuery telegram class
:param callback_data: GptCategory class
:return: None
"""
if callback_data.user_id != call.from_user.id:
return
await state.set_state(AnsweringGpt4All.set_model)
await call.message.edit_text("Выберите пожалуйста модель ИИ 👾",
reply_markup=generate_gpt4all_page(user_id=call.from_user.id))
@staticmethod
async def telegram_g4a_back_handler(call: CallbackQuery, callback_data: GptCategory, state: FSMContext) -> None:
"""
Query, what shows list for gpt4all models back
:param state: FSMContext aiogram class
:param call: CallbackQuery telegram class
:param callback_data: GptCategory class
:return: None
"""
if callback_data.user_id != call.from_user.id:
return
await state.set_state(AnsweringGpt4All.set_model)
await call.message.edit_text("Выберите пожалуйста модель ИИ 👾",
reply_markup=generate_gpt4all_page(user_id=call.from_user.id))
@staticmethod
async def telegram_g4a_infomration_handler(call: CallbackQuery, callback_data: Gpt4AllModel, state: FSMContext) -> None:
"""
Query, what show information about clicked gpt4all model from list
:param state: FSMContext aiogram class
:param call: CallbackQuery telegram class
:param callback_data: Gpt4AllModel class
:return: None
"""
if callback_data.user_id != call.from_user.id:
return
models = GPT4All.list_models()
name = models[callback_data.index]['name']
await call.message.edit_text(f"{name}\n"
f"Обученно на основе {models[callback_data.index]['parameters']} строк 👨‍💻",
reply_markup=gpt4all_model_menu(user_id=call.from_user.id,
index=callback_data.index))
@staticmethod
async def telegram_g4a_end_handler(call: CallbackQuery, callback_data: Gpt4AllSelect,
state: FSMContext) -> None:
"""
Query, what says about getting ready for question for Gpt4All model
:param state: FSMContext aiogram class
:param call: CallbackQuery telegram class
:param callback_data: Gpt4AllSelect class
:return: None
"""
if callback_data.user_id != call.from_user.id:
return
await state.update_data(set_model=callback_data.index)
await state.set_state(AnsweringGpt4All.ready_to_answer)
models = GPT4All.list_models()
await call.message.edit_text("Удача ✅\n"
"Вы теперь можете спокойно вести диалог 🤖\n"
f"Вы выбрали модель <b>{models[callback_data.index]['name']}</b>👾 от Gpt4All\n"
"Чтобы прекратить общение, используйте /cancel ",
reply_markup=delete_keyboard(admin_id=callback_data.user_id))
@staticmethod
async def telegram_pages_handler(call: CallbackQuery) -> None:
"""
Query, made for helping purposes.
Shows current page.
:param call: CallbackQuery telegram class
:return: None
"""
logging.log(msg=f"Showed helping info for user_id={call.from_user.id}",
level=logging.INFO)
await call.answer("Здесь расположается текущая странница 📃")
@staticmethod
async def telegram_stop_dialog_handler(call: CallbackQuery, callback_data: GptStop, state: FSMContext) -> None:
"""
Query, what stops dialog
:param call: CallbackQuery telegram class
:param callback_data: GptStop class
:param state: None
"""
# Checking user_id of user
if callback_data.user_id != call.from_user.id:
return
# Clearing state
await state.clear()
# Answering something
await call.answer("Хорошо ✅")
if await state.get_state() == AnsweringGPT4Free.ready_to_answer or await state.get_state() == AnsweringGpt4All.answering:
await call.message.edit_text(text=call.message.text + "\n\nДиалог остановлен ✅\n",
reply_markup=delete_keyboard(admin_id=call.from_user.id))
else:
await call.message.delete()
def __init__(self):
"""
All information about feature
will be inside this function
"""
super().__init__()
# Telegram feature settings
self.telegram_setting = TelegramChatSettings.text_generation
self.telegram_setting_in_list = True
self.telegram_setting_name = "ИИ ЧатБот 🤖"
self.telegram_setting_description = "<b>ИИ ЧатБот </b>🤖" \
"\nЕсть поддержка:\n" \
"- Моделей Gpt4All\n" \
"- Провайдеров Gpt4Free и моделей\n" \
"Для использования:\n" \
"<pre>/conversations</pre>" \
"\nНаходится в разработке, планируется в будущем. Следите за обновлениями 😘"
self.telegram_commands: dict[str: str] = {
'conversation': 'Starts conversation with text generative ai'
}
self.telegram_cmd_avaible = True # Is a feature have a commands
self.telegram_message_handlers = {
self.telegram_conversation_cmd_handler: [Command(commands=["conversation"])],
self.telegram_g4a_generate_handler: [AnsweringGpt4All.ready_to_answer, ~Command(commands=["cancel"])],
self.telegram_g4f_generate_handler: [AnsweringGPT4Free.ready_to_answer, ~Command(commands=["cancel"])],
self.telegram_already_answering_handler: [AnsweringGPT4Free.answering, AnsweringGpt4All.answering]
}
self.telegram_callback_handlers = {
# g4a
self.telegram_g4a_handler: [GptCategory.filter(F.category == "Gpt4All")],
self.telegram_g4a_infomration_handler: [Gpt4AllModel.filter()],
self.telegram_g4a_end_handler: [Gpt4AllSelect.filter()],
# g4f
self.telegram_g4f_category_handler: [GptCategory.filter(F.category == "Gpt4Free")],
self.telegram_g4f_models_handler: [Gpt4FreeCategory.filter(F.category == "models")],
self.telegram_g4f_providers_handlers: [Gpt4FreeCategory.filter(F.category == "providers")],
self.telegram_g4f_models_by_provider_handler: [Gpt4FreeProvider.filter()],
self.telegram_next_g4f_providers_handler: [Gpt4FreeProviderPage.filter()],
# stop talking
self.telegram_stop_dialog_handler: [GptStop.filter()]
}

View file

@ -0,0 +1,82 @@
import logging
from aiogram import Bot
from aiogram.types import Message, CallbackQuery
from sqlalchemy.ext.asyncio import async_sessionmaker
from bozenka.database.tables.telegram import TelegramChatSettings
from bozenka.features import BasicFeature
from bozenka.instances.telegram.utils.callbacks_factory import PinMsg, UnpinMsg
from bozenka.instances.telegram.utils.keyboards import unpin_msg_keyboard, delete_keyboard, pin_msg_keyboard
from bozenka.instances.telegram.utils.simpler import SolutionSimpler
class Welcome(BasicFeature):
"""
A class of pins related commands
All staff related to it will be here
"""
@staticmethod
async def telegram_join_handler(msg: Message, session_maker: async_sessionmaker) -> None:
"""
Message handler.
Send welcome message, after adding new member to chat.
Also works on adding bot to chat and sending welcome message.
:param msg: Message telegram object
:param session_maker: AsyncSessionmaker object
:return: None
"""
for new in msg.new_chat_members:
if new.id != msg.bot.id:
logging.log(msg=f"Saing welcome for user_id={new.id}, chat_id={msg.chat.id}",
level=logging.INFO)
await msg.answer(
f"Пользователь {new.mention_html()} переехал в конфу, благодаря {msg.from_user.mention_html()}👋",
)
await msg.delete()
else:
logging.log(msg=f"Saing welcome to administrators for chat_id={msg.chat.id}",
level=logging.INFO)
await msg.answer("Здраствуйте администраторы чата 👋\n"
"Я - <b>бозенька</b>, мультифункциональный бот, разрабатываемый Bozo Developement\n"
"Выдайте мне <b>полные права администратора</b> для моей полной работы.\n"
"Чтобы настроить функционал, используйте /setup или кнопку под сообщением", )
await SolutionSimpler.auto_settings(msg=msg, session=session_maker)
@staticmethod
async def telegram_leave_handler(msg: Message, bot: Bot) -> None:
"""
Sens goodbye message, after deleting member from chat
:param msg: Message telegram object
:param bot: Aiogram bot object
:return: Nothing
"""
await msg.delete()
if msg.from_user.id == bot.id:
return
logging.log(msg=f"Saing goodbye for user_id={msg.left_chat_member.id}, chat_id={msg.chat.id}",
level=logging.INFO)
await msg.answer(
f"Пользователь {msg.left_chat_member.mention_html()} съехал с конфы, благодаря {msg.from_user.mention_html()}👋"
)
def __init__(self):
"""
All information about feature
will be inside this function
"""
super().__init__()
self.cmd_description: str = "Your description of command"
# Telegram feature settings
self.telegram_setting = TelegramChatSettings.welcome_messages
self.telegram_commands: dict[str: str] = {}
self.telegram_setting_in_list = True
self.telegram_setting_name = "Приветсвенные сообщения 👋"
self.telegram_setting_description = "<b>Приветсвенные сообщения 👋</b>" \
"\nПриветсвенные сообщения новым и ушедшим пользователям.",
self.telegram_cmd_avaible = False # Is a feature have a commands
self.telegram_message_handlers = {
}
self.telegram_callback_handlers = {}

View file

@ -23,8 +23,6 @@ def register_admin_cmd(router: Router) -> None:
logging.log(msg="Registering administrator commands", level=logging.INFO)
# Ban / Unban commands handler
router.message.register(ban_command, Command(commands="ban"),
IsAdminFilter(True))
router.message.register(ban_command, Command(commands="ban"),
IsAdminFilter(True), F.reply_to_message.text)
router.message.register(unban_command, Command(commands="unban"),

View file

@ -1,18 +1,12 @@
import logging
import os
import g4f
import base64
from gpt4all import GPT4All
from aiogram.fsm.context import FSMContext
from aiogram.types import Message as Message, FSInputFile
from bozenka.generative.kadinsky import kadinsky_gen
from bozenka.instances.telegram.utils.keyboards import gpt_categories_keyboard, delete_keyboard, text_response_keyboard, \
image_generation_keyboard, image_response_keyboard
from bozenka.instances.telegram.utils.simpler import AnsweringGpt4All, \
AnsweringGPT4Free, GeneratingImages
from bozenka.generative.gpt4free import generate_gpt4free_providers
from bozenka.instances.telegram.utils.keyboards import delete_keyboard, image_generation_keyboard, \
image_response_keyboard
from bozenka.instances.telegram.utils.simpler import GeneratingImages
async def already_generating(msg: Message, state: FSMContext):

View file

@ -27,7 +27,7 @@ async def inline_setup_category_back(call: CallbackQuery, callback_data: SetupAc
:return:
"""
await call.message.edit_text("Выберите настройку, которую хотите изменить",
reply_markup=setup_category_keyboard(category=callback_data.feature_category))
reply_markup=setup_category_keyboard(category=callback_data.category_name))
async def inline_edit_feature(call: CallbackQuery, callback_data: SetupFeature, session_maker: async_sessionmaker):
@ -63,11 +63,11 @@ async def inline_feature_edited(call: CallbackQuery, callback_data: SetupAction,
async with session_maker() as session:
async with session.begin():
await session.execute(Update(TelegramChatSettings)
.values({list_of_features[callback_data.feature_category][callback_data.feature_index].settings_name:
.values({list_of_features[callback_data.category_name][callback_data.feature_index].settings_name:
callback_data.action == "enable"})
.where(TelegramChatSettings.chat_id == call.message.chat.id))
await call.message.edit_text(
list_of_features[callback_data.feature_category][callback_data.feature_index].description,
reply_markup=await setup_feature_keyboard(category=callback_data.feature_category,
list_of_features[callback_data.category_name][callback_data.feature_index].description,
reply_markup=await setup_feature_keyboard(category=callback_data.category_name,
index=callback_data.feature_index,
is_enabled=callback_data.action == "enable"))

View file

@ -28,7 +28,7 @@ class SetupAction(CallbackData, prefix="sa"):
Callback with information to do with a feature
"""
action: str
feature_category: str
category_name: str
feature_index: int

View file

@ -92,7 +92,7 @@ list_of_features = {
BaseFeature(
name="Приглашения в Чат ✉",
description="<b>Генератор приглашения в Чат ✉</b>\n"
"Разрешает использование комманды /invite в чате, для созданния приглашений.\n"
"Разрешает использование комманды <code>/invite</code> в чате, для созданния приглашений.\n"
"Для исполнения <b>требует соответсвующих прав от пользователя и их наличие у бота.</b>",
callback_name="invites",
settings_name="invite_generator",
@ -288,7 +288,7 @@ list_of_commands = {
("open_general", 'Opens general topic in chat'),
# AI related
('conversation', 'Starts conversation with text generative ai'),
('imagine', 'Starts conversation with image generative ai'), \
('imagine', 'Starts conversation with image generative ai'),
# Basic features
('invite', 'Generates invite into current chat'),
('about', 'Sends information about bozenka'),