mirror of
https://gitflic.ru/project/maks1ms/ocab.git
synced 2024-12-23 16:23:01 +03:00
Merged with private/new-module-system
This commit is contained in:
parent
370b4fc648
commit
eecc59ca94
@ -21,6 +21,7 @@ packages = [{include = "scripts"}]
|
||||
[tool.poetry.scripts]
|
||||
test = 'scripts.test:main'
|
||||
init = 'scripts.init:main'
|
||||
module = 'scripts.module:main'
|
||||
|
||||
[tool.poetry.dependencies]
|
||||
python = ">=3.11.6,<3.13"
|
||||
|
@ -6,12 +6,12 @@ def main():
|
||||
pwd = Path().cwd()
|
||||
dir_core = pwd / "src" / "ocab_core"
|
||||
dir_modules_standard = pwd / "src" / "ocab_modules" / "standard"
|
||||
dir_modules_custom = pwd / "src" / "ocab_modules" / "custom"
|
||||
dir_modules_external = pwd / "src" / "ocab_modules" / "external"
|
||||
|
||||
json = {
|
||||
"core": str(dir_core),
|
||||
"modules standard": str(dir_modules_standard),
|
||||
"modules custom": str(dir_modules_custom),
|
||||
"modules external": str(dir_modules_external),
|
||||
}
|
||||
with open("src/paths.json", "w", encoding="utf8") as f:
|
||||
f.write(dumps(json, indent=4))
|
||||
|
115
scripts/module.py
Normal file
115
scripts/module.py
Normal file
@ -0,0 +1,115 @@
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
|
||||
DEFAULTS = {
|
||||
"description": "Очень полезный модуль",
|
||||
"author": "OCAB Team",
|
||||
"version": "1.0.0",
|
||||
"privileged": "false",
|
||||
}
|
||||
|
||||
|
||||
def create_module(args):
|
||||
module_dir = os.path.join("src/ocab_modules/standard", args.module_name)
|
||||
os.makedirs(module_dir, exist_ok=True)
|
||||
|
||||
module_info = {
|
||||
"id": args.id,
|
||||
"name": args.name,
|
||||
"description": args.description,
|
||||
"author": args.author,
|
||||
"version": args.version,
|
||||
"privileged": args.privileged.lower() == "true",
|
||||
"dependencies": {},
|
||||
}
|
||||
|
||||
with open(os.path.join(module_dir, "info.json"), "w", encoding="utf-8") as f:
|
||||
json.dump(module_info, f, ensure_ascii=False, indent=4)
|
||||
|
||||
with open(os.path.join(module_dir, "__init__.py"), "w", encoding="utf-8") as f:
|
||||
f.write("# Init file for the module\n")
|
||||
|
||||
print(f"Module {args.module_name} created successfully.")
|
||||
|
||||
|
||||
def interactive_mode(args):
|
||||
def get_input(prompt, default=None):
|
||||
if default:
|
||||
value = input(f"{prompt} [{default}]: ")
|
||||
return value if value else default
|
||||
else:
|
||||
value = input(f"{prompt}: ")
|
||||
return value
|
||||
|
||||
module_name = get_input("Введите название модуля (папки)")
|
||||
module_id = get_input("Введите ID")
|
||||
name = get_input("Введите название модуля")
|
||||
description = get_input(
|
||||
"Введите описание модуля", args.description or DEFAULTS["description"]
|
||||
)
|
||||
author = get_input("Введите автора", args.author or DEFAULTS["author"])
|
||||
version = get_input("Введите версию", args.version or DEFAULTS["version"])
|
||||
privileged = get_input(
|
||||
"Модуль привилегированный (true/false)",
|
||||
args.privileged or DEFAULTS["privileged"],
|
||||
)
|
||||
|
||||
args = argparse.Namespace(
|
||||
command="create",
|
||||
module_name=module_name,
|
||||
id=module_id,
|
||||
name=name,
|
||||
description=description,
|
||||
author=author,
|
||||
version=version,
|
||||
privileged=privileged,
|
||||
dependencies="",
|
||||
)
|
||||
|
||||
create_module(args)
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Утилита для создания директории модуля с файлами."
|
||||
)
|
||||
subparsers = parser.add_subparsers(dest="command", required=True)
|
||||
|
||||
create_parser = subparsers.add_parser("create", help="Создать новый модуль")
|
||||
create_parser.add_argument("--module_name", help="Название директории модуля")
|
||||
create_parser.add_argument("--id", help="ID модуля")
|
||||
create_parser.add_argument("--name", help="Название модуля")
|
||||
create_parser.add_argument("--description", help="Описание модуля")
|
||||
create_parser.add_argument("--author", help="Автор модуля")
|
||||
create_parser.add_argument("--version", help="Версия модуля")
|
||||
create_parser.add_argument(
|
||||
"--privileged", help="Привилегированный модуль (true/false)"
|
||||
)
|
||||
create_parser.add_argument(
|
||||
"--dependencies", help="Список зависимостей в формате имя:версия через запятую"
|
||||
)
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.command == "create":
|
||||
if not all(
|
||||
[
|
||||
args.module_name,
|
||||
args.id,
|
||||
args.name,
|
||||
args.description,
|
||||
args.author,
|
||||
args.version,
|
||||
args.privileged,
|
||||
args.dependencies,
|
||||
]
|
||||
):
|
||||
print("Переход в интерактивный режим...")
|
||||
interactive_mode(args)
|
||||
else:
|
||||
create_module(args)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
@ -21,18 +21,9 @@ def setup_logger():
|
||||
)
|
||||
|
||||
|
||||
async def log(message):
|
||||
"""
|
||||
Функция для логирования сообщений
|
||||
|
||||
Она асинхронная, хотя таковой на самом деле не является.
|
||||
"""
|
||||
def log(message):
|
||||
if isinstance(message, Exception):
|
||||
error_message = f"Error: {str(message)}\n{traceback.format_exc()}"
|
||||
logging.error(error_message)
|
||||
else:
|
||||
logging.info(message)
|
||||
|
||||
|
||||
def log_new(message):
|
||||
logging.info(message)
|
||||
|
@ -3,7 +3,7 @@ import traceback
|
||||
|
||||
from aiogram import Bot, Dispatcher
|
||||
|
||||
from ocab_core.logger import setup_logger
|
||||
from ocab_core.logger import log, setup_logger
|
||||
from ocab_core.modules_system import ModulesManager
|
||||
from ocab_core.modules_system.loaders import FSLoader
|
||||
from ocab_core.modules_system.loaders.unsafe_fs_loader import UnsafeFSLoader
|
||||
@ -14,35 +14,47 @@ from service import paths
|
||||
bot_modules = [
|
||||
UnsafeFSLoader(f"{paths.modules_standard}/config"),
|
||||
UnsafeFSLoader(f"{paths.modules_standard}/database"),
|
||||
UnsafeFSLoader(f"{paths.modules_standard}/fsm_database_storage"),
|
||||
UnsafeFSLoader(f"{paths.modules_standard}/roles"),
|
||||
UnsafeFSLoader(f"{paths.modules_external}/yandexgpt"),
|
||||
FSLoader(f"{paths.modules_standard}/command_helper"),
|
||||
FSLoader(f"{paths.modules_standard}/info"),
|
||||
FSLoader(f"{paths.modules_standard}/filters"),
|
||||
FSLoader(f"{paths.modules_external}/create_report_apps"),
|
||||
FSLoader(f"{paths.modules_standard}/admin"),
|
||||
FSLoader(f"{paths.modules_standard}/message_processing"),
|
||||
]
|
||||
|
||||
|
||||
async def main():
|
||||
bot = None
|
||||
database = None
|
||||
setup_logger()
|
||||
|
||||
app = Singleton()
|
||||
|
||||
try:
|
||||
bot = Bot(token=get_telegram_token())
|
||||
|
||||
app.dp = Dispatcher()
|
||||
app.bot = Bot(token=get_telegram_token())
|
||||
app.modules_manager = ModulesManager()
|
||||
|
||||
for module_loader in bot_modules:
|
||||
app.modules_manager.load(module_loader)
|
||||
info = module_loader.info()
|
||||
log(f"Loading {info.name}({info.id}) module")
|
||||
await app.modules_manager.load(module_loader)
|
||||
|
||||
await app.dp.start_polling(bot)
|
||||
app.dp = Dispatcher(storage=app.storage["_fsm_storage"])
|
||||
|
||||
app.dp.include_routers(*app.storage["_routers"])
|
||||
|
||||
for middleware in app.storage["_outer_message_middlewares"]:
|
||||
app.dp.message.outer_middleware.register(middleware)
|
||||
|
||||
await app.modules_manager.late_init()
|
||||
await app.dp.start_polling(app.bot)
|
||||
except Exception:
|
||||
traceback.print_exc()
|
||||
finally:
|
||||
if bot is not None:
|
||||
await bot.session.close()
|
||||
if database is not None:
|
||||
database.close()
|
||||
await app.bot.session.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
@ -11,7 +11,7 @@ class ModuleInfo:
|
||||
name: str
|
||||
description: str
|
||||
version: str
|
||||
author: str
|
||||
author: str | list[str]
|
||||
privileged: bool
|
||||
dependencies: dict
|
||||
|
||||
|
@ -3,12 +3,12 @@ from pathlib import Path
|
||||
|
||||
from RestrictedPython import compile_restricted_exec
|
||||
|
||||
from ocab_core.modules_system.loaders.fs_loader.policy import (
|
||||
from ocab_core.modules_system.loaders.unsafe_fs_loader import UnsafeFSLoader
|
||||
from ocab_core.modules_system.safe.policy import (
|
||||
ALLOWED_IMPORTS,
|
||||
BUILTINS,
|
||||
RestrictedPythonPolicy,
|
||||
)
|
||||
from ocab_core.modules_system.loaders.unsafe_fs_loader import UnsafeFSLoader
|
||||
|
||||
|
||||
class FSLoader(UnsafeFSLoader):
|
||||
|
@ -49,9 +49,6 @@ class UnsafeFSLoader(AbstractLoader):
|
||||
# Добавляем директорию модуля в sys.path
|
||||
sys.path.insert(0, str(path))
|
||||
|
||||
print(full_path.parent.absolute())
|
||||
print(module_name)
|
||||
|
||||
# Загружаем спецификацию модуля
|
||||
spec = importlib.util.spec_from_file_location(module_name, full_path)
|
||||
|
||||
|
@ -23,20 +23,25 @@ def is_version_compatible(version, requirement):
|
||||
|
||||
class ModulesManager:
|
||||
def __init__(self):
|
||||
self.modules = {}
|
||||
self.modules = []
|
||||
|
||||
def load(self, loader: AbstractLoader):
|
||||
async def load(self, loader: AbstractLoader):
|
||||
info = loader.info()
|
||||
|
||||
if info.id in self.modules:
|
||||
# Check if the module is already loaded
|
||||
if any(mod["info"].id == info.id for mod in self.modules):
|
||||
return
|
||||
|
||||
# Check dependencies
|
||||
for dependency, version in info.dependencies.items():
|
||||
if dependency not in self.modules:
|
||||
loaded_dependency = next(
|
||||
(mod for mod in self.modules if mod["info"].id == dependency), None
|
||||
)
|
||||
if not loaded_dependency:
|
||||
raise Exception(
|
||||
f"Module {info.id} depends on {dependency}, but it is not loaded"
|
||||
)
|
||||
loaded_dependency_info = self.modules[dependency]["info"]
|
||||
loaded_dependency_info = loaded_dependency["info"]
|
||||
if not is_version_compatible(loaded_dependency_info.version, version):
|
||||
raise Exception(
|
||||
f"Module {info.id} depends on {dependency}, "
|
||||
@ -45,22 +50,36 @@ class ModulesManager:
|
||||
|
||||
module = loader.load()
|
||||
|
||||
self.modules[info.id] = {
|
||||
self.modules.append(
|
||||
{
|
||||
"info": info,
|
||||
"module": module,
|
||||
}
|
||||
)
|
||||
|
||||
if hasattr(module, "module_init"):
|
||||
module.module_init()
|
||||
await module.module_init()
|
||||
|
||||
async def late_init(self):
|
||||
for m in self.modules:
|
||||
module = m["module"]
|
||||
if hasattr(module, "module_late_init"):
|
||||
await module.module_late_init()
|
||||
|
||||
def get_by_id(self, module_id: str):
|
||||
if module_id not in self.modules:
|
||||
module = next(
|
||||
(mod for mod in self.modules if mod["info"].id == module_id), None
|
||||
)
|
||||
if not module:
|
||||
raise Exception(f"Module with id {module_id} not loaded")
|
||||
|
||||
return self.modules[module_id]["module"]
|
||||
return module["module"]
|
||||
|
||||
def get_info_by_id(self, module_id: str):
|
||||
if module_id not in self.modules:
|
||||
module = next(
|
||||
(mod for mod in self.modules if mod["info"].id == module_id), None
|
||||
)
|
||||
if not module:
|
||||
raise Exception(f"Module with id {module_id} not loaded")
|
||||
|
||||
return self.modules[module_id]["info"]
|
||||
return module["info"]
|
||||
|
@ -1,3 +1,10 @@
|
||||
from ocab_core.logger import log # noqa
|
||||
|
||||
from .public_api import Storage, get_module, register_router
|
||||
from .public_api import (
|
||||
Storage,
|
||||
get_fsm_context,
|
||||
get_module,
|
||||
log,
|
||||
register_outer_message_middleware,
|
||||
register_router,
|
||||
set_my_commands,
|
||||
)
|
||||
from .utils import Utils
|
||||
|
@ -1,14 +1,47 @@
|
||||
import types
|
||||
from typing import Any, Tuple, Union
|
||||
|
||||
from aiogram import Router
|
||||
from aiogram import BaseMiddleware, Router
|
||||
from aiogram.fsm.context import FSMContext
|
||||
from aiogram.fsm.storage.base import StorageKey
|
||||
|
||||
from ocab_core.logger import log
|
||||
from ocab_core.singleton import Singleton
|
||||
|
||||
|
||||
def register_router(router: Router):
|
||||
app = Singleton()
|
||||
app.dp.include_router(router)
|
||||
app.storage["_routers"].append(router)
|
||||
|
||||
|
||||
def register_outer_message_middleware(middleware: BaseMiddleware):
|
||||
app = Singleton()
|
||||
app.storage["_outer_message_middlewares"].append(middleware)
|
||||
|
||||
|
||||
async def set_my_commands(commands):
|
||||
app = Singleton()
|
||||
await app.bot.set_my_commands(commands)
|
||||
|
||||
|
||||
async def get_fsm_context(chat_id: int, user_id: int) -> FSMContext:
|
||||
dp = Singleton().dp
|
||||
bot = Singleton().bot
|
||||
|
||||
return FSMContext(
|
||||
storage=dp.storage,
|
||||
key=StorageKey(
|
||||
chat_id=chat_id,
|
||||
user_id=user_id,
|
||||
bot_id=bot.id,
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def set_fsm(storage):
|
||||
app = Singleton()
|
||||
log(storage)
|
||||
app.storage["_fsm_storage"] = storage
|
||||
|
||||
|
||||
def get_module(
|
||||
|
12
src/ocab_core/modules_system/public_api/utils.py
Normal file
12
src/ocab_core/modules_system/public_api/utils.py
Normal file
@ -0,0 +1,12 @@
|
||||
import re
|
||||
|
||||
CLEAN_HTML = re.compile("<.*?>")
|
||||
|
||||
|
||||
class Utils:
|
||||
@staticmethod
|
||||
def code_format(code: str, lang: str):
|
||||
if lang:
|
||||
return f'<pre><code class="language-{lang}">{code}</code></pre>'
|
||||
else:
|
||||
return f"<pre>{code}</pre>"
|
@ -1,13 +1,22 @@
|
||||
from _ast import AnnAssign
|
||||
from typing import Any
|
||||
|
||||
from aiogram import Bot
|
||||
from RestrictedPython import (
|
||||
RestrictingNodeTransformer,
|
||||
limited_builtins,
|
||||
safe_builtins,
|
||||
utility_builtins,
|
||||
)
|
||||
from RestrictedPython.Eval import default_guarded_getitem
|
||||
from RestrictedPython.Eval import default_guarded_getitem, default_guarded_getiter
|
||||
from RestrictedPython.Guards import (
|
||||
full_write_guard,
|
||||
guarded_unpack_sequence,
|
||||
safer_getattr,
|
||||
)
|
||||
|
||||
from ocab_core.logger import log
|
||||
from ocab_core.modules_system.safe.zope_guards import extra_safe_builtins
|
||||
|
||||
|
||||
class RestrictedPythonPolicy(RestrictingNodeTransformer):
|
||||
@ -77,22 +86,24 @@ ALLOWED_IMPORTS = [
|
||||
"warnings",
|
||||
]
|
||||
|
||||
|
||||
def safes_getattr(object, name, default=None, getattr=safer_getattr):
|
||||
if isinstance(object, Bot) and name == "token":
|
||||
log("Bot.token is not allowed")
|
||||
raise Exception("Bot.token is not allowed")
|
||||
|
||||
return getattr(object, name, default)
|
||||
|
||||
|
||||
BUILTINS = safe_builtins.copy()
|
||||
BUILTINS.update(utility_builtins)
|
||||
BUILTINS.update(limited_builtins)
|
||||
BUILTINS.update(extra_safe_builtins)
|
||||
BUILTINS["__metaclass__"] = _metaclass
|
||||
BUILTINS["_getitem_"] = default_guarded_getitem
|
||||
# BUILTINS["_write_"] = full_write_guard
|
||||
BUILTINS["_getattr_"] = safes_getattr
|
||||
BUILTINS["_getiter_"] = default_guarded_getiter
|
||||
BUILTINS["_write_"] = full_write_guard
|
||||
BUILTINS["_unpack_sequence_"] = guarded_unpack_sequence
|
||||
BUILTINS["staticmethod"] = staticmethod
|
||||
|
||||
|
||||
class GuardedDictType:
|
||||
def __call__(self, *args, **kwargs):
|
||||
return dict(*args, **kwargs)
|
||||
|
||||
@staticmethod
|
||||
def fromkeys(iterable, value=None):
|
||||
return dict.fromkeys(iterable, value)
|
||||
|
||||
|
||||
BUILTINS["dict"] = GuardedDictType()
|
||||
BUILTINS["tuple"] = tuple
|
225
src/ocab_core/modules_system/safe/zope_guards.py
Normal file
225
src/ocab_core/modules_system/safe/zope_guards.py
Normal file
@ -0,0 +1,225 @@
|
||||
#############################################################################
|
||||
#
|
||||
# Copyright (c) 2024 OCAB Team
|
||||
# Copyright (c) 2002 Zope Foundation and Contributors.
|
||||
#
|
||||
# This software includes a function derived from the software subject to the
|
||||
# provisions of the Zope Public License, Version 2.1 (ZPL). A copy of the ZPL
|
||||
# should accompany this distribution. THIS SOFTWARE IS PROVIDED "AS IS" AND ANY
|
||||
# AND ALL EXPRESS OR IMPLIED WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT
|
||||
# LIMITED TO, THE IMPLIED WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST
|
||||
# INFRINGEMENT, AND FITNESS FOR A PARTICULAR PURPOSE.
|
||||
#
|
||||
#
|
||||
##############################################################################
|
||||
|
||||
extra_safe_builtins = {}
|
||||
|
||||
|
||||
class GuardedDictType:
|
||||
def __call__(self, *args, **kwargs):
|
||||
return dict(*args, **kwargs)
|
||||
|
||||
def fromkeys(self, S, v=None):
|
||||
return dict.fromkeys(S, v)
|
||||
|
||||
|
||||
extra_safe_builtins["dict"] = GuardedDictType()
|
||||
|
||||
|
||||
ContainerAssertions = {
|
||||
type(()): 1,
|
||||
bytes: 1,
|
||||
str: 1,
|
||||
range: 1,
|
||||
}
|
||||
|
||||
Containers = ContainerAssertions.get
|
||||
|
||||
|
||||
def _error(index):
|
||||
raise Exception("unauthorized access to element")
|
||||
|
||||
|
||||
def guard(container, value, index=None):
|
||||
# if Containers(type(container)) and Containers(type(value)):
|
||||
# # Simple type. Short circuit.
|
||||
# return
|
||||
# I don't know how to do this.
|
||||
# if getSecurityManager().validate(container, container, index, value):
|
||||
# return
|
||||
# _error(index)
|
||||
return
|
||||
|
||||
|
||||
class SafeIter:
|
||||
__allow_access_to_unprotected_subobjects__ = 1
|
||||
|
||||
def __init__(self, ob, container=None):
|
||||
self._iter = iter(ob)
|
||||
if container is None:
|
||||
container = ob
|
||||
self.container = container
|
||||
|
||||
def __iter__(self):
|
||||
return self
|
||||
|
||||
def __next__(self):
|
||||
ob = next(self._iter)
|
||||
guard(self.container, ob)
|
||||
return ob
|
||||
|
||||
next = __next__
|
||||
|
||||
|
||||
class NullIter(SafeIter):
|
||||
def __init__(self, ob):
|
||||
self._iter = ob
|
||||
|
||||
def __next__(self):
|
||||
return next(self._iter)
|
||||
|
||||
next = __next__
|
||||
|
||||
|
||||
def guarded_iter(*args):
|
||||
if len(args) == 1:
|
||||
i = args[0]
|
||||
# Don't double-wrap
|
||||
if isinstance(i, SafeIter):
|
||||
return i
|
||||
if not isinstance(i, range):
|
||||
return SafeIter(i)
|
||||
# Other call styles / targets don't need to be guarded
|
||||
return NullIter(iter(*args))
|
||||
|
||||
|
||||
extra_safe_builtins["iter"] = guarded_iter
|
||||
|
||||
|
||||
def guarded_any(seq):
|
||||
return any(guarded_iter(seq))
|
||||
|
||||
|
||||
extra_safe_builtins["any"] = guarded_any
|
||||
|
||||
|
||||
def guarded_all(seq):
|
||||
return all(guarded_iter(seq))
|
||||
|
||||
|
||||
extra_safe_builtins["all"] = guarded_all
|
||||
|
||||
valid_inplace_types = (list, set)
|
||||
|
||||
inplace_slots = {
|
||||
"+=": "__iadd__",
|
||||
"-=": "__isub__",
|
||||
"*=": "__imul__",
|
||||
"/=": (1 / 2 == 0) and "__idiv__" or "__itruediv__",
|
||||
"//=": "__ifloordiv__",
|
||||
"%=": "__imod__",
|
||||
"**=": "__ipow__",
|
||||
"<<=": "__ilshift__",
|
||||
">>=": "__irshift__",
|
||||
"&=": "__iand__",
|
||||
"^=": "__ixor__",
|
||||
"|=": "__ior__",
|
||||
}
|
||||
|
||||
|
||||
def __iadd__(x, y):
|
||||
x += y
|
||||
return x
|
||||
|
||||
|
||||
def __isub__(x, y):
|
||||
x -= y
|
||||
return x
|
||||
|
||||
|
||||
def __imul__(x, y):
|
||||
x *= y
|
||||
return x
|
||||
|
||||
|
||||
def __idiv__(x, y):
|
||||
x /= y
|
||||
return x
|
||||
|
||||
|
||||
def __ifloordiv__(x, y):
|
||||
x //= y
|
||||
return x
|
||||
|
||||
|
||||
def __imod__(x, y):
|
||||
x %= y
|
||||
return x
|
||||
|
||||
|
||||
def __ipow__(x, y):
|
||||
x **= y
|
||||
return x
|
||||
|
||||
|
||||
def __ilshift__(x, y):
|
||||
x <<= y
|
||||
return x
|
||||
|
||||
|
||||
def __irshift__(x, y):
|
||||
x >>= y
|
||||
return x
|
||||
|
||||
|
||||
def __iand__(x, y):
|
||||
x &= y
|
||||
return x
|
||||
|
||||
|
||||
def __ixor__(x, y):
|
||||
x ^= y
|
||||
return x
|
||||
|
||||
|
||||
def __ior__(x, y):
|
||||
x |= y
|
||||
return x
|
||||
|
||||
|
||||
inplace_ops = {
|
||||
"+=": __iadd__,
|
||||
"-=": __isub__,
|
||||
"*=": __imul__,
|
||||
"/=": __idiv__,
|
||||
"//=": __ifloordiv__,
|
||||
"%=": __imod__,
|
||||
"**=": __ipow__,
|
||||
"<<=": __ilshift__,
|
||||
">>=": __irshift__,
|
||||
"&=": __iand__,
|
||||
"^=": __ixor__,
|
||||
"|=": __ior__,
|
||||
}
|
||||
|
||||
|
||||
def protected_inplacevar(op, var, expr):
|
||||
"""Do an inplace operation
|
||||
|
||||
If the var has an inplace slot, then disallow the operation
|
||||
unless the var an instance of ``valid_inplace_types``.
|
||||
"""
|
||||
if hasattr(var, inplace_slots[op]) and not isinstance(var, valid_inplace_types):
|
||||
try:
|
||||
cls = var.__class__
|
||||
except AttributeError:
|
||||
cls = type(var)
|
||||
raise TypeError(
|
||||
"Augmented assignment to %s objects is not allowed"
|
||||
" in untrusted code" % cls.__name__
|
||||
)
|
||||
return inplace_ops[op](var, expr)
|
||||
|
||||
|
||||
extra_safe_builtins["_inplacevar_"] = protected_inplacevar
|
@ -1,18 +0,0 @@
|
||||
from aiogram import Dispatcher
|
||||
|
||||
from src.ocab_modules.standard.admin.routers import router as admin_router
|
||||
|
||||
# from src.modules.standard.info.routers import router as info_router
|
||||
from src.ocab_modules.standard.message_processing.message_api import (
|
||||
router as process_message,
|
||||
)
|
||||
|
||||
|
||||
async def include_routers(dp: Dispatcher):
|
||||
"""
|
||||
Подключение роутеров в бота
|
||||
dp.include_router()
|
||||
"""
|
||||
# dp.include_router(info_router)
|
||||
dp.include_router(admin_router)
|
||||
dp.include_router(process_message)
|
@ -1,4 +1,5 @@
|
||||
from aiogram import Dispatcher
|
||||
from aiogram import Bot, Dispatcher
|
||||
from aiogram.fsm.storage.memory import MemoryStorage
|
||||
|
||||
from ocab_core.modules_system import ModulesManager
|
||||
|
||||
@ -14,6 +15,11 @@ class SingletonMeta(type):
|
||||
|
||||
|
||||
class Singleton(metaclass=SingletonMeta):
|
||||
bot: Bot
|
||||
dp: Dispatcher = None
|
||||
modules_manager: ModulesManager = None
|
||||
storage = dict()
|
||||
storage = {
|
||||
"_fsm_storage": MemoryStorage(),
|
||||
"_routers": [],
|
||||
"_outer_message_middlewares": [],
|
||||
}
|
||||
|
1
src/ocab_modules/external/create_report_apps/__init__.py
vendored
Normal file
1
src/ocab_modules/external/create_report_apps/__init__.py
vendored
Normal file
@ -0,0 +1 @@
|
||||
from .main import module_init
|
136
src/ocab_modules/external/create_report_apps/create_report.py
vendored
Normal file
136
src/ocab_modules/external/create_report_apps/create_report.py
vendored
Normal file
@ -0,0 +1,136 @@
|
||||
from aiogram import Bot, Router
|
||||
from aiogram.enums import ParseMode
|
||||
from aiogram.fsm.context import FSMContext
|
||||
from aiogram.fsm.state import State, StatesGroup
|
||||
from aiogram.types import (
|
||||
BufferedInputFile,
|
||||
KeyboardButton,
|
||||
Message,
|
||||
ReplyKeyboardMarkup,
|
||||
ReplyKeyboardRemove,
|
||||
)
|
||||
|
||||
from ocab_core.modules_system.public_api import Utils, get_fsm_context
|
||||
|
||||
from .report import Report
|
||||
|
||||
router = Router()
|
||||
|
||||
|
||||
class ReportState(StatesGroup):
|
||||
input_system_info = State()
|
||||
input_app_name = State()
|
||||
input_problem_step_by_step = State()
|
||||
input_actual_result = State()
|
||||
input_expected_result = State()
|
||||
input_additional_info = State()
|
||||
|
||||
|
||||
system_info_code = """echo "SESSION_TYPE: ${XDG_SESSION_TYPE:-Unknown}"
|
||||
[ -f /etc/os-release ] && grep "^PRETTY_NAME=" /etc/os-release | cut -d= -f2 \
|
||||
| tr -d '"' | xargs echo "OS: "
|
||||
echo "Kernel: $(uname -r)"
|
||||
echo "DE: ${XDG_CURRENT_DESKTOP:-Unknown}"
|
||||
grep "^model name" /proc/cpuinfo | head -n1 | cut -d: -f2 \
|
||||
| xargs echo "CPU: "
|
||||
lspci | grep "VGA compatible controller" | cut -d: -f3 \
|
||||
| xargs -I{} echo "GPU: {}"
|
||||
"""
|
||||
|
||||
|
||||
system_info_message = """Укажите параметры свой системы.
|
||||
Собрать информацию о системе можно с помощью данного скрипта:
|
||||
""" + Utils.code_format(
|
||||
system_info_code,
|
||||
"shell",
|
||||
)
|
||||
|
||||
|
||||
async def start_report(chat_id: int, bot: Bot):
|
||||
await bot.send_message(
|
||||
chat_id=chat_id,
|
||||
text=system_info_message,
|
||||
parse_mode=ParseMode.HTML,
|
||||
reply_markup=ReplyKeyboardRemove(),
|
||||
)
|
||||
state = await get_fsm_context(chat_id, chat_id)
|
||||
|
||||
await state.set_state(ReportState.input_system_info)
|
||||
|
||||
|
||||
app_info_message = """Укажите название и версию приложения.
|
||||
Узнать можно с помощью данной команды:""" + Utils.code_format(
|
||||
"rpm -qa | grep -i НАЗВАНИЕ_ПРИЛОЖЕНИЯ", "shell"
|
||||
)
|
||||
|
||||
|
||||
@router.message(ReportState.input_system_info)
|
||||
async def system_entered(message: Message, state: FSMContext):
|
||||
await state.update_data(system=message.text)
|
||||
await message.answer(
|
||||
text=app_info_message,
|
||||
parse_mode=ParseMode.HTML,
|
||||
)
|
||||
await state.set_state(ReportState.input_app_name)
|
||||
|
||||
|
||||
step_by_step_message = (
|
||||
"""Опиши проблему пошагово, что ты делал, что происходило, что не так."""
|
||||
)
|
||||
|
||||
|
||||
@router.message(ReportState.input_app_name)
|
||||
async def app_name_entered(message: Message, state: FSMContext):
|
||||
await state.update_data(app=message.text)
|
||||
await message.answer(text=step_by_step_message)
|
||||
await state.set_state(ReportState.input_problem_step_by_step)
|
||||
|
||||
|
||||
@router.message(ReportState.input_problem_step_by_step)
|
||||
async def problem_step_by_step_entered(message: Message, state: FSMContext):
|
||||
await state.update_data(problem_step_by_step=message.text)
|
||||
await message.answer(text="Опиши, что произошло (фактический результат).")
|
||||
await state.set_state(ReportState.input_actual_result)
|
||||
|
||||
|
||||
@router.message(ReportState.input_actual_result)
|
||||
async def actual_result_entered(message: Message, state: FSMContext):
|
||||
await state.update_data(actual=message.text)
|
||||
await message.answer(text="Опиши ожидаемый результат.")
|
||||
await state.set_state(ReportState.input_expected_result)
|
||||
|
||||
|
||||
@router.message(ReportState.input_expected_result)
|
||||
async def expected_result_entered(message: Message, state: FSMContext):
|
||||
await state.update_data(expected=message.text)
|
||||
await message.answer(
|
||||
text="Если есть дополнительная информация, то напиши ее.",
|
||||
reply_markup=ReplyKeyboardMarkup(
|
||||
resize_keyboard=True,
|
||||
keyboard=[
|
||||
[KeyboardButton(text="Дополнительной информации нет")],
|
||||
],
|
||||
),
|
||||
)
|
||||
await state.set_state(ReportState.input_additional_info)
|
||||
|
||||
|
||||
@router.message(ReportState.input_additional_info)
|
||||
async def additional_info_entered(message: Message, state: FSMContext):
|
||||
if message.text == "Дополнительной информации нет":
|
||||
additional_info = ""
|
||||
else:
|
||||
additional_info = message.text
|
||||
await state.update_data(additional=additional_info)
|
||||
await message.answer(
|
||||
text="Вот твой отчет сообщением, а также файлом:",
|
||||
reply_markup=ReplyKeyboardRemove(),
|
||||
)
|
||||
data = await state.get_data()
|
||||
|
||||
report = Report(data)
|
||||
file_report = report.export().encode()
|
||||
|
||||
await message.answer(text=report.export())
|
||||
await message.answer_document(document=BufferedInputFile(file_report, "report.txt"))
|
||||
await state.clear()
|
14
src/ocab_modules/external/create_report_apps/info.json
vendored
Normal file
14
src/ocab_modules/external/create_report_apps/info.json
vendored
Normal file
@ -0,0 +1,14 @@
|
||||
{
|
||||
"id": "external.create_report_apps",
|
||||
"name": "Create Report Apps",
|
||||
"description": "Модуль для создания отчетов о ошибках в приложениях",
|
||||
"author": [
|
||||
"OCAB Team",
|
||||
"Maxim Slipenko"
|
||||
],
|
||||
"version": "1.0.0",
|
||||
"privileged": false,
|
||||
"dependencies": {
|
||||
"standard.command_helper": "^1.0.0"
|
||||
}
|
||||
}
|
115
src/ocab_modules/external/create_report_apps/main.py
vendored
Normal file
115
src/ocab_modules/external/create_report_apps/main.py
vendored
Normal file
@ -0,0 +1,115 @@
|
||||
from typing import Union
|
||||
|
||||
from aiogram import Bot, F, Router
|
||||
from aiogram.exceptions import TelegramForbiddenError
|
||||
from aiogram.filters import BaseFilter, Command, CommandStart
|
||||
from aiogram.types import (
|
||||
CallbackQuery,
|
||||
InlineKeyboardButton,
|
||||
InlineKeyboardMarkup,
|
||||
Message,
|
||||
)
|
||||
|
||||
from ocab_core.modules_system.public_api import get_module, register_router
|
||||
|
||||
from .create_report import router as create_report_router
|
||||
from .create_report import start_report
|
||||
|
||||
register_command = get_module("standard.command_helper", "register_command")
|
||||
|
||||
router = Router()
|
||||
|
||||
|
||||
class ChatTypeFilter(BaseFilter):
|
||||
def __init__(self, chat_type: Union[str, list]):
|
||||
self.chat_type = chat_type
|
||||
|
||||
async def __call__(self, message: Message) -> bool:
|
||||
if isinstance(self.chat_type, str):
|
||||
return message.chat.type == self.chat_type
|
||||
return message.chat.type in self.chat_type
|
||||
|
||||
|
||||
@router.message(
|
||||
ChatTypeFilter(chat_type=["group", "supergroup"]), Command("create_report_apps")
|
||||
)
|
||||
async def create_report_apps_command_group(message: Message):
|
||||
keyboard = InlineKeyboardMarkup(
|
||||
inline_keyboard=[
|
||||
[
|
||||
InlineKeyboardButton(
|
||||
text="Да", callback_data=f"create_report:{message.from_user.id}"
|
||||
),
|
||||
InlineKeyboardButton(
|
||||
text="Нет", callback_data=f"cancel_report:{message.from_user.id}"
|
||||
),
|
||||
]
|
||||
]
|
||||
)
|
||||
await message.answer(
|
||||
"Я могу отправить тебе пару вопросов "
|
||||
"для помощи в составлении репорта личными "
|
||||
"сообщениями.",
|
||||
reply_markup=keyboard,
|
||||
)
|
||||
|
||||
|
||||
@router.message(
|
||||
ChatTypeFilter(chat_type=["private"]),
|
||||
CommandStart(deep_link=True, magic=F.args == "create_report_apps"),
|
||||
)
|
||||
@router.message(ChatTypeFilter(chat_type=["private"]), Command("create_report_apps"))
|
||||
async def create_report_apps_command(message: Message, bot: Bot):
|
||||
await start_report(message.from_user.id, bot)
|
||||
|
||||
|
||||
@router.callback_query(F.data.startswith("cancel_report"))
|
||||
async def cancel_report_callback(callback_query: CallbackQuery):
|
||||
callback_user_id = int(callback_query.data.split(":")[1])
|
||||
if callback_query.from_user.id != callback_user_id:
|
||||
await callback_query.answer("Эта кнопка не для вас.", show_alert=True)
|
||||
return
|
||||
|
||||
await callback_query.message.delete()
|
||||
|
||||
|
||||
@router.callback_query(F.data.startswith("create_report"))
|
||||
async def create_report_callback(callback_query: CallbackQuery, bot: Bot):
|
||||
callback_user_id = int(callback_query.data.split(":")[1])
|
||||
if callback_query.from_user.id != callback_user_id:
|
||||
await callback_query.answer("Эта кнопка не для вас.", show_alert=True)
|
||||
return
|
||||
|
||||
user_id = callback_query.from_user.id
|
||||
|
||||
async def on_chat_unavailable():
|
||||
await callback_query.message.edit_text(
|
||||
"Я в личных сообщениях задам тебе вопросы "
|
||||
"для помощи в составлении репорта. "
|
||||
'Но перед этим ты должен нажать кнопку "Запустить"'
|
||||
)
|
||||
info = await bot.get_me()
|
||||
await callback_query.answer(
|
||||
url=f"https://t.me/{info.username}?start=create_report_apps"
|
||||
)
|
||||
|
||||
try:
|
||||
chat_member = await bot.get_chat_member(chat_id=user_id, user_id=user_id)
|
||||
if chat_member.status != "left":
|
||||
await start_report(user_id, bot)
|
||||
await callback_query.message.edit_text(
|
||||
"Я в личных сообщениях задам тебе "
|
||||
"вопросы для помощи в составлении "
|
||||
"репорта."
|
||||
)
|
||||
else:
|
||||
await on_chat_unavailable()
|
||||
except TelegramForbiddenError:
|
||||
await on_chat_unavailable()
|
||||
|
||||
|
||||
async def module_init():
|
||||
router.include_router(create_report_router)
|
||||
|
||||
register_router(router)
|
||||
register_command("create_report_apps", "Написать репорт о приложении")
|
59
src/ocab_modules/external/create_report_apps/report.py
vendored
Normal file
59
src/ocab_modules/external/create_report_apps/report.py
vendored
Normal file
@ -0,0 +1,59 @@
|
||||
import aiogram
|
||||
|
||||
|
||||
class ReportFormatter:
|
||||
def __init__(self, html=True):
|
||||
self.html = html
|
||||
|
||||
def bold(self, string):
|
||||
if self.html:
|
||||
return f"<b>{self.text(string)}</b>"
|
||||
return self.text(string)
|
||||
|
||||
def text(self, string):
|
||||
if self.html:
|
||||
return aiogram.html.quote(string)
|
||||
return string
|
||||
|
||||
|
||||
class Report:
|
||||
def __init__(self, data: dict):
|
||||
self.data = data
|
||||
|
||||
def export(self):
|
||||
data = self.data
|
||||
|
||||
report = f"""
|
||||
Стенд с ошибкой:
|
||||
==============================
|
||||
|
||||
{data['system']}
|
||||
|
||||
Пакет:
|
||||
==============================
|
||||
|
||||
{data['app']}
|
||||
|
||||
Шаги, приводящие к ошибке:
|
||||
==============================
|
||||
|
||||
{data['problem_step_by_step']}
|
||||
|
||||
Фактический результат:
|
||||
==============================
|
||||
|
||||
{data['actual']}
|
||||
|
||||
Ожидаемый результат:
|
||||
==============================
|
||||
|
||||
{data['expected']}
|
||||
"""
|
||||
if data["additional"] != "":
|
||||
report += f"""
|
||||
Дополнительно:
|
||||
==============================
|
||||
|
||||
{data['additional']}
|
||||
"""
|
||||
return report
|
@ -0,0 +1 @@
|
||||
from .handlers import answer_to_message
|
@ -1,6 +1,4 @@
|
||||
# flake8: noqa
|
||||
import asyncio
|
||||
|
||||
from aiogram import Bot
|
||||
from aiogram.types import Message
|
||||
|
||||
@ -15,7 +13,7 @@ from ocab_modules.standard.database.db_api import add_message
|
||||
|
||||
async def answer_to_message(message: Message, bot: Bot):
|
||||
# print("answer_to_message")
|
||||
await log("answer_to_message")
|
||||
log("answer_to_message")
|
||||
yagpt = YandexGPT(get_yandexgpt_token(), get_yandexgpt_catalog_id())
|
||||
text = message.text
|
||||
prompt = get_yandexgpt_prompt()
|
||||
|
@ -1,6 +1,9 @@
|
||||
{
|
||||
"id": "external.yandexgpt",
|
||||
"name": "Yandex GPT",
|
||||
"description": "Модуль для работы с Yandex GPT",
|
||||
"author": "OCAB Team",
|
||||
"version": "1.0"
|
||||
"version": "1.0.0",
|
||||
"privileged": true,
|
||||
"dependencies": {}
|
||||
}
|
||||
|
@ -57,7 +57,7 @@ class YandexGPT:
|
||||
)
|
||||
except Exception as e: # TODO: Переделать обработку ошибок
|
||||
# print(e)
|
||||
await log(f"Error: {e}")
|
||||
log(f"Error: {e}")
|
||||
|
||||
continue
|
||||
if int(len(response["tokens"])) < (max_tokens - answer_token):
|
||||
@ -262,7 +262,7 @@ class YandexGPT:
|
||||
)
|
||||
elif type == "yandexgpt":
|
||||
# print("yandexgpt_request")
|
||||
await log("yandexgpt_request")
|
||||
log("yandexgpt_request")
|
||||
messages = await self.collect_messages(message_id, chat_id)
|
||||
return await self.async_yandexgpt(
|
||||
system_prompt=get_yandexgpt_prompt(),
|
||||
|
1
src/ocab_modules/legacy/moderation/__init__.py
Normal file
1
src/ocab_modules/legacy/moderation/__init__.py
Normal file
@ -0,0 +1 @@
|
||||
from .moderation import ban_user, unmute_user
|
@ -6,8 +6,8 @@ import time
|
||||
import aiogram
|
||||
import aiohttp
|
||||
|
||||
from ...standard.config.config import *
|
||||
from ...standard.roles.roles import *
|
||||
from ocab_modules.standard.config.config import *
|
||||
from ocab_modules.standard.roles.roles import *
|
||||
|
||||
|
||||
class Moderation:
|
@ -5,18 +5,12 @@ import random
|
||||
from threading import Thread
|
||||
|
||||
from aiogram import Bot
|
||||
from aiogram.types import Message
|
||||
from aiogram.types import inline_keyboard_button as types
|
||||
from aiogram.utils.keyboard import InlineKeyboardBuilder
|
||||
|
||||
from ocab_modules.legacy.moderation import ban_user, unmute_user
|
||||
from src.ocab_modules.standard.config.config import get_telegram_check_bot
|
||||
from src.ocab_modules.standard.database.db_api import *
|
||||
from src.ocab_modules.standard.moderation.moderation import (
|
||||
ban_user,
|
||||
mute_user,
|
||||
unmute_user,
|
||||
)
|
||||
from src.ocab_modules.standard.roles.roles import Roles
|
||||
|
||||
|
||||
async def create_math_task():
|
@ -1,6 +1,6 @@
|
||||
from aiogram import F, Router
|
||||
|
||||
from src.ocab_modules.standard.welcome.handlers import check_new_user
|
||||
from .handlers import check_new_user
|
||||
|
||||
router = Router()
|
||||
|
@ -1 +1 @@
|
||||
from . import routers
|
||||
from .main import module_init
|
||||
|
@ -1,10 +1,10 @@
|
||||
# flake8: noqa
|
||||
import time
|
||||
|
||||
from aiogram import Bot
|
||||
from aiogram.types import Message
|
||||
|
||||
from src.ocab_modules.standard.config.config import get_default_chat_tag
|
||||
from ocab_core.modules_system.public_api import get_module
|
||||
|
||||
get_default_chat_tag = get_module("standard.config", "get_default_chat_tag")
|
||||
|
||||
|
||||
async def delete_message(message: Message, bot: Bot):
|
||||
|
@ -1,6 +1,11 @@
|
||||
{
|
||||
"id": "standard.admin",
|
||||
"name": "Admin",
|
||||
"description": "Модуль для работы с админкой",
|
||||
"author": "OCAB Team",
|
||||
"version": "1.0"
|
||||
"version": "1.0.0",
|
||||
"privileged": false,
|
||||
"dependencies": {
|
||||
"standard.filters": "^1.0.0"
|
||||
}
|
||||
}
|
||||
|
7
src/ocab_modules/standard/admin/main.py
Normal file
7
src/ocab_modules/standard/admin/main.py
Normal file
@ -0,0 +1,7 @@
|
||||
from ocab_core.modules_system.public_api import register_router
|
||||
|
||||
from .routers import router
|
||||
|
||||
|
||||
async def module_init():
|
||||
register_router(router)
|
@ -1,15 +1,18 @@
|
||||
# flake8: noqa
|
||||
from aiogram import F, Router
|
||||
from aiogram.filters import Command
|
||||
|
||||
from src.ocab_modules.standard.admin.handlers import (
|
||||
from ocab_core.modules_system.public_api import get_module, log
|
||||
|
||||
from .handlers import (
|
||||
chat_not_in_approve_list,
|
||||
delete_message,
|
||||
error_access,
|
||||
get_chat_id,
|
||||
)
|
||||
from src.ocab_modules.standard.filters.filters import (
|
||||
ChatModerOrAdminFilter,
|
||||
ChatNotInApproveFilter,
|
||||
|
||||
(ChatModerOrAdminFilter, ChatNotInApproveFilter) = get_module(
|
||||
"standard.filters", ["ChatModerOrAdminFilter", "ChatNotInApproveFilter"]
|
||||
)
|
||||
|
||||
router = Router()
|
||||
@ -17,8 +20,8 @@ router = Router()
|
||||
# Если сообщение содержит какой либо текст и выполняется фильтр ChatNotInApproveFilter, то вызывается функция chat_not_in_approve_list
|
||||
router.message.register(chat_not_in_approve_list, ChatNotInApproveFilter(), F.text)
|
||||
|
||||
router.message.register(get_chat_id, ChatModerOrAdminFilter(), F.text == "/chatID")
|
||||
router.message.register(get_chat_id, ChatModerOrAdminFilter(), Command("chatID"))
|
||||
|
||||
router.message.register(delete_message, ChatModerOrAdminFilter(), F.text == "/rm")
|
||||
router.message.register(error_access, F.text == "/rm")
|
||||
router.message.register(error_access, F.text == "/chatID")
|
||||
router.message.register(delete_message, ChatModerOrAdminFilter(), Command("rm"))
|
||||
router.message.register(error_access, Command("rm"))
|
||||
router.message.register(error_access, Command("chatID"))
|
||||
|
1
src/ocab_modules/standard/command_helper/__init__.py
Normal file
1
src/ocab_modules/standard/command_helper/__init__.py
Normal file
@ -0,0 +1 @@
|
||||
from .main import module_init, register_command
|
12
src/ocab_modules/standard/command_helper/info.json
Normal file
12
src/ocab_modules/standard/command_helper/info.json
Normal file
@ -0,0 +1,12 @@
|
||||
{
|
||||
"id": "standard.command_helper",
|
||||
"name": "Command helper",
|
||||
"description": "Модуль для отображения команд при вводе '/'",
|
||||
"author": "OCAB Team",
|
||||
"version": "1.0.0",
|
||||
"privileged": false,
|
||||
"dependencies": {
|
||||
"standard.roles": "^1.0.0",
|
||||
"standard.database": "^1.0.0"
|
||||
}
|
||||
}
|
96
src/ocab_modules/standard/command_helper/main.py
Normal file
96
src/ocab_modules/standard/command_helper/main.py
Normal file
@ -0,0 +1,96 @@
|
||||
from typing import Any, Awaitable, Callable, Dict
|
||||
|
||||
from aiogram import BaseMiddleware
|
||||
from aiogram.types import BotCommand, TelegramObject
|
||||
|
||||
from ocab_core.modules_system.public_api import (
|
||||
get_module,
|
||||
register_outer_message_middleware,
|
||||
set_my_commands,
|
||||
)
|
||||
|
||||
commands = dict()
|
||||
|
||||
|
||||
db_api = get_module(
|
||||
"standard.database",
|
||||
"db_api",
|
||||
)
|
||||
|
||||
Roles = get_module("standard.roles", "Roles")
|
||||
|
||||
|
||||
def register_command(command, description, role="USER"):
|
||||
if role not in commands:
|
||||
commands[role] = dict()
|
||||
commands[role][command] = {
|
||||
"description": description,
|
||||
}
|
||||
|
||||
|
||||
class OuterMiddleware(BaseMiddleware):
|
||||
async def __call__(
|
||||
self,
|
||||
handler: Callable[[TelegramObject, Dict[str, Any]], Awaitable[Any]],
|
||||
event: TelegramObject,
|
||||
data: Dict[str, Any],
|
||||
):
|
||||
|
||||
# if not isinstance(event, Message):
|
||||
# return await handler(event, data)
|
||||
#
|
||||
# user = db_api.get_user(event.from_user.id)
|
||||
#
|
||||
# if user is None:
|
||||
# return
|
||||
#
|
||||
# roles = Roles()
|
||||
# role_name = await roles.get_role_name(role_id=user.user_role)
|
||||
#
|
||||
# if role_name not in commands:
|
||||
# return await handler(event, data)
|
||||
|
||||
# bot_commands = []
|
||||
|
||||
# for role_command in commands[role_name]:
|
||||
# bot_commands.append(
|
||||
# BotCommand(
|
||||
# command=role_command,
|
||||
# description=commands[role_name][role_command]["description"],
|
||||
# )
|
||||
# )
|
||||
|
||||
# await event.bot.set_my_commands(
|
||||
# bot_commands,
|
||||
# BotCommandScopeChatMember(
|
||||
# chat_id=event.chat.id,
|
||||
# user_id=event.from_user.id,
|
||||
# ),
|
||||
# )
|
||||
|
||||
return await handler(event, data)
|
||||
|
||||
|
||||
async def module_init():
|
||||
register_outer_message_middleware(OuterMiddleware())
|
||||
|
||||
|
||||
async def set_user_commands():
|
||||
bot_commands = []
|
||||
if "USER" in commands:
|
||||
user_commands = commands["USER"]
|
||||
for command in user_commands:
|
||||
bot_commands.append(
|
||||
BotCommand(
|
||||
command=command,
|
||||
description=user_commands[command]["description"],
|
||||
)
|
||||
)
|
||||
|
||||
await set_my_commands(
|
||||
bot_commands,
|
||||
)
|
||||
|
||||
|
||||
async def module_late_init():
|
||||
await set_user_commands()
|
@ -1 +1,7 @@
|
||||
from .config import config
|
||||
from .config import (
|
||||
get_approved_chat_id,
|
||||
get_default_chat_tag,
|
||||
get_roles,
|
||||
get_yandexgpt_in_words,
|
||||
get_yandexgpt_start_words,
|
||||
)
|
||||
|
@ -27,13 +27,17 @@ def get_telegram_check_bot() -> bool:
|
||||
return config["TELEGRAM"]["CHECK_BOT"]
|
||||
|
||||
|
||||
def get_aproved_chat_id() -> list:
|
||||
def get_approved_chat_id() -> list:
|
||||
# Возваращем сплитованный список id чатов в формате int
|
||||
return [
|
||||
int(chat_id) for chat_id in config["TELEGRAM"]["APPROVED_CHAT_ID"].split(" | ")
|
||||
]
|
||||
|
||||
|
||||
def get_roles():
|
||||
return config["ROLES"]
|
||||
|
||||
|
||||
def get_user_role_name(role_number) -> dict:
|
||||
# Возвращаем название роли пользвателя по номеру роли, если такой роли нет, возвращаем неизвестно
|
||||
return config["ROLES"].get(role_number, "Неизвестно")
|
||||
|
@ -1,5 +1,5 @@
|
||||
from . import db_api, models
|
||||
from . import db_api, models, repositories
|
||||
|
||||
|
||||
def module_init():
|
||||
async def module_init():
|
||||
db_api.connect_database()
|
||||
|
@ -7,6 +7,7 @@ from .exceptions import MissingModuleName, NotExpectedModuleName
|
||||
from .models.chat_stats import ChatStats
|
||||
from .models.chats import Chats
|
||||
from .models.db import database_proxy
|
||||
from .models.fsm_data import FSMData
|
||||
from .models.messages import Messages
|
||||
from .models.user_stats import UserStats
|
||||
from .models.users import Users
|
||||
@ -32,7 +33,7 @@ def connect_database(is_test: bool = False, module: str | None = None):
|
||||
|
||||
def create_tables(db: pw.SqliteDatabase):
|
||||
"""Создание таблиц"""
|
||||
for table in Chats, Messages, Users, UserStats, ChatStats:
|
||||
for table in Chats, Messages, Users, UserStats, ChatStats, FSMData:
|
||||
if not table.table_exists():
|
||||
db.create_tables([table])
|
||||
|
||||
|
@ -0,0 +1 @@
|
||||
from .fsm_data import FSMData
|
12
src/ocab_modules/standard/database/models/fsm_data.py
Normal file
12
src/ocab_modules/standard/database/models/fsm_data.py
Normal file
@ -0,0 +1,12 @@
|
||||
import peewee as pw
|
||||
|
||||
from .db import database_proxy
|
||||
|
||||
|
||||
class FSMData(pw.Model):
|
||||
class Meta:
|
||||
database = database_proxy
|
||||
|
||||
key = pw.CharField(primary_key=True)
|
||||
state = pw.CharField(null=True)
|
||||
data = pw.CharField(null=True)
|
@ -0,0 +1 @@
|
||||
from .fsm_data import FSMDataRepository
|
32
src/ocab_modules/standard/database/repositories/fsm_data.py
Normal file
32
src/ocab_modules/standard/database/repositories/fsm_data.py
Normal file
@ -0,0 +1,32 @@
|
||||
from peewee import fn
|
||||
|
||||
from ..models import FSMData
|
||||
|
||||
|
||||
class FSMDataRepository:
|
||||
def get(self, key: str):
|
||||
return FSMData.get_or_none(key=key)
|
||||
|
||||
def set_state(self, key: str, state: str):
|
||||
FSMData.insert(
|
||||
key=key,
|
||||
state=state,
|
||||
data=fn.COALESCE(
|
||||
FSMData.select(FSMData.data).where(FSMData.key == key), None
|
||||
),
|
||||
).on_conflict(
|
||||
conflict_target=[FSMData.key],
|
||||
update={FSMData.state: state},
|
||||
).execute()
|
||||
|
||||
def set_data(self, key: str, data: str):
|
||||
FSMData.insert(
|
||||
key=key,
|
||||
data=data,
|
||||
state=fn.COALESCE(
|
||||
FSMData.select(FSMData.state).where(FSMData.key == key), None
|
||||
),
|
||||
).on_conflict(
|
||||
conflict_target=[FSMData.key],
|
||||
update={FSMData.data: data},
|
||||
).execute()
|
@ -0,0 +1 @@
|
||||
from .filters import ChatModerOrAdminFilter, ChatNotInApproveFilter
|
@ -2,9 +2,10 @@ from aiogram import Bot
|
||||
from aiogram.filters import BaseFilter
|
||||
from aiogram.types import Message
|
||||
|
||||
from ocab_core.logger import log
|
||||
from ocab_modules.standard.config.config import get_aproved_chat_id
|
||||
from ocab_modules.standard.roles.roles import Roles
|
||||
from ocab_core.modules_system.public_api import get_module, log
|
||||
|
||||
get_approved_chat_id = get_module("standard.config", "get_approved_chat_id")
|
||||
Roles = get_module("standard.roles", "Roles")
|
||||
|
||||
|
||||
class ChatModerOrAdminFilter(BaseFilter):
|
||||
@ -21,14 +22,11 @@ class ChatModerOrAdminFilter(BaseFilter):
|
||||
|
||||
class ChatNotInApproveFilter(BaseFilter):
|
||||
async def __call__(self, message: Message, bot: Bot) -> bool:
|
||||
# print("chat_check")
|
||||
await log("chat_check")
|
||||
log("chat_check")
|
||||
chat_id = message.chat.id
|
||||
if chat_id in get_aproved_chat_id():
|
||||
# print(f"Chat in approve list: {chat_id}")
|
||||
await log(f"Chat in approve list: {chat_id}")
|
||||
if chat_id in get_approved_chat_id():
|
||||
log(f"Chat in approve list: {chat_id}")
|
||||
return False
|
||||
else:
|
||||
# print(f"Chat not in approve list: {chat_id}")
|
||||
await log(f"Chat not in approve list: {chat_id}")
|
||||
log(f"Chat not in approve list: {chat_id}")
|
||||
return True
|
||||
|
@ -1,6 +1,12 @@
|
||||
{
|
||||
"id": "standard.filters",
|
||||
"name": "Filters",
|
||||
"description": "Модуль с фильтрами",
|
||||
"author": "OCAB Team",
|
||||
"version": "1.0"
|
||||
"version": "1.0.0",
|
||||
"privileged": false,
|
||||
"dependencies": {
|
||||
"standard.roles": "^1.0.0",
|
||||
"standard.config": "^1.0.0"
|
||||
}
|
||||
}
|
||||
|
@ -0,0 +1 @@
|
||||
from .fsm import module_init
|
122
src/ocab_modules/standard/fsm_database_storage/fsm.py
Normal file
122
src/ocab_modules/standard/fsm_database_storage/fsm.py
Normal file
@ -0,0 +1,122 @@
|
||||
import json
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
from aiogram.fsm.state import State
|
||||
from aiogram.fsm.storage.base import BaseStorage, StorageKey
|
||||
|
||||
from ocab_core.modules_system.public_api import get_module, log
|
||||
from ocab_core.modules_system.public_api.public_api import set_fsm
|
||||
|
||||
FSMDataRepository = get_module("standard.database", "repositories.FSMDataRepository")
|
||||
|
||||
|
||||
def serialize_key(key: StorageKey) -> str:
|
||||
return f"{key.bot_id}:{key.chat_id}:{key.user_id}"
|
||||
|
||||
|
||||
def serialize_object(obj: object) -> str | None:
|
||||
try:
|
||||
return json.dumps(obj)
|
||||
except Exception as e:
|
||||
log(f"Serializing error! {e}")
|
||||
return None
|
||||
|
||||
|
||||
def deserialize_object(obj):
|
||||
try:
|
||||
return json.loads(obj)
|
||||
except Exception as e:
|
||||
log(f"Deserializing error! {e}")
|
||||
return None
|
||||
|
||||
|
||||
class SQLStorage(BaseStorage):
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.repo = FSMDataRepository()
|
||||
|
||||
async def set_state(self, key: StorageKey, state: State | None = None) -> None:
|
||||
"""
|
||||
Set state for specified key
|
||||
|
||||
:param key: storage key
|
||||
:param state: new state
|
||||
"""
|
||||
s_key = serialize_key(key)
|
||||
s_state = state.state if isinstance(state, State) else state
|
||||
|
||||
try:
|
||||
self.repo.set_state(s_key, s_state)
|
||||
except Exception as e:
|
||||
log(f"FSM Storage database error: {e}")
|
||||
|
||||
async def get_state(self, key: StorageKey) -> Optional[str]:
|
||||
"""
|
||||
Get key state
|
||||
|
||||
:param key: storage key
|
||||
:return: current state
|
||||
"""
|
||||
s_key = serialize_key(key)
|
||||
|
||||
try:
|
||||
s_state = self.repo.get(s_key)
|
||||
return s_state.state if s_state else None
|
||||
except Exception as e:
|
||||
log(f"FSM Storage database error: {e}")
|
||||
return None
|
||||
|
||||
async def set_data(self, key: StorageKey, data: Dict[str, Any]) -> None:
|
||||
"""
|
||||
Write data (replace)
|
||||
|
||||
:param key: storage key
|
||||
:param data: new data
|
||||
"""
|
||||
s_key = serialize_key(key)
|
||||
s_data = serialize_object(data)
|
||||
|
||||
try:
|
||||
self.repo.set_data(s_key, s_data)
|
||||
except Exception as e:
|
||||
log(f"FSM Storage database error: {e}")
|
||||
|
||||
async def get_data(self, key: StorageKey) -> Optional[Dict[str, Any]]:
|
||||
"""
|
||||
Get current data for key
|
||||
|
||||
:param key: storage key
|
||||
:return: current data
|
||||
"""
|
||||
s_key = serialize_key(key)
|
||||
|
||||
try:
|
||||
s_data = self.repo.get(s_key)
|
||||
return deserialize_object(s_data.data) if s_data else None
|
||||
except Exception as e:
|
||||
log(f"FSM Storage database error: {e}")
|
||||
return None
|
||||
|
||||
async def update_data(
|
||||
self, key: StorageKey, data: Dict[str, Any]
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Update data in the storage for key (like dict.update)
|
||||
|
||||
:param key: storage key
|
||||
:param data: partial data
|
||||
:return: new data
|
||||
"""
|
||||
current_data = await self.get_data(key=key)
|
||||
if not current_data:
|
||||
current_data = {}
|
||||
current_data.update(data)
|
||||
await self.set_data(key=key, data=current_data)
|
||||
return current_data.copy()
|
||||
|
||||
async def close(self) -> None: # pragma: no cover
|
||||
pass
|
||||
|
||||
|
||||
async def module_init():
|
||||
set_fsm(SQLStorage())
|
11
src/ocab_modules/standard/fsm_database_storage/info.json
Normal file
11
src/ocab_modules/standard/fsm_database_storage/info.json
Normal file
@ -0,0 +1,11 @@
|
||||
{
|
||||
"id": "standard.fsm_database_storage",
|
||||
"name": "FSM Database Storage",
|
||||
"description": "Очень полезный модуль",
|
||||
"author": "OCAB Team",
|
||||
"version": "1.0.0",
|
||||
"privileged": false,
|
||||
"dependencies": {
|
||||
"standard.database": "^1.0.0"
|
||||
}
|
||||
}
|
@ -1,14 +1,2 @@
|
||||
from aiogram import F, Router
|
||||
|
||||
from ocab_core.modules_system.public_api import register_router
|
||||
|
||||
from .handlers import get_chat_info, get_user_info
|
||||
|
||||
|
||||
def module_init():
|
||||
router = Router()
|
||||
|
||||
router.message.register(get_user_info, F.text.startswith("/info"))
|
||||
router.message.register(get_chat_info, F.text.startswith("/chatinfo"))
|
||||
|
||||
register_router(router)
|
||||
from .main import module_init
|
||||
|
@ -32,7 +32,7 @@ async def get_info_answer_by_id(message: Message, bot: Bot, user_id: int):
|
||||
|
||||
if user is None:
|
||||
await message.reply("Пользователь не найден")
|
||||
await log(f"Пользователь не найден: {user_id}, {user}")
|
||||
log(f"Пользователь не найден: {user_id}, {user}")
|
||||
return
|
||||
|
||||
roles = Roles()
|
||||
@ -69,7 +69,7 @@ async def get_user_info(message: Message, bot: Bot):
|
||||
" попробуйте запросить информацию о пользователе по его тегу или ответив на его сообщение"
|
||||
)
|
||||
# print(e)
|
||||
await log(e)
|
||||
log(e)
|
||||
|
||||
|
||||
async def get_chat_info(message: Message, bot: Bot):
|
||||
|
@ -7,6 +7,7 @@
|
||||
"privileged": false,
|
||||
"dependencies": {
|
||||
"standard.roles": "^1.0.0",
|
||||
"standard.database": "^1.0.0"
|
||||
"standard.database": "^1.0.0",
|
||||
"standard.command_helper": "^1.0.0"
|
||||
}
|
||||
}
|
||||
|
20
src/ocab_modules/standard/info/main.py
Normal file
20
src/ocab_modules/standard/info/main.py
Normal file
@ -0,0 +1,20 @@
|
||||
from aiogram import Router
|
||||
from aiogram.filters import Command
|
||||
|
||||
from ocab_core.modules_system.public_api import get_module, register_router
|
||||
|
||||
from .handlers import get_chat_info, get_user_info
|
||||
|
||||
register_command = get_module("standard.command_helper", "register_command")
|
||||
|
||||
|
||||
async def module_init():
|
||||
router = Router()
|
||||
|
||||
router.message.register(get_user_info, Command("info"))
|
||||
router.message.register(get_chat_info, Command("chatinfo"))
|
||||
|
||||
register_router(router)
|
||||
|
||||
register_command("info", "Информация о пользователе")
|
||||
register_command("chatinfo", "Информация о чате")
|
@ -0,0 +1 @@
|
||||
from .message_api import module_init
|
13
src/ocab_modules/standard/message_processing/info.json
Normal file
13
src/ocab_modules/standard/message_processing/info.json
Normal file
@ -0,0 +1,13 @@
|
||||
{
|
||||
"id": "standard.message_processing",
|
||||
"name": "Info",
|
||||
"description": "Модуль с информацией",
|
||||
"author": "OCAB Team",
|
||||
"version": "1.0.0",
|
||||
"privileged": false,
|
||||
"dependencies": {
|
||||
"standard.roles": "^1.0.0",
|
||||
"standard.database": "^1.0.0",
|
||||
"standard.command_helper": "^1.0.0"
|
||||
}
|
||||
}
|
@ -2,14 +2,45 @@
|
||||
|
||||
from aiogram import Bot, F, Router, types
|
||||
|
||||
from ocab_core.logger import log
|
||||
from ocab_modules.external.yandexgpt.handlers import answer_to_message
|
||||
from ocab_modules.standard.config.config import (
|
||||
get_aproved_chat_id,
|
||||
get_yandexgpt_in_words,
|
||||
get_yandexgpt_start_words,
|
||||
from ocab_core.modules_system.public_api import get_module, log, register_router
|
||||
|
||||
# from ocab_modules.standard.database.db_api import *
|
||||
|
||||
(get_approved_chat_id, get_yandexgpt_in_words, get_yandexgpt_start_words) = get_module(
|
||||
"standard.config",
|
||||
["get_approved_chat_id", "get_yandexgpt_in_words", "get_yandexgpt_start_words"],
|
||||
)
|
||||
|
||||
answer_to_message = get_module("external.yandexgpt", "answer_to_message")
|
||||
|
||||
(
|
||||
get_chat,
|
||||
add_chat,
|
||||
get_user,
|
||||
add_user,
|
||||
get_user_name,
|
||||
change_user_name,
|
||||
get_user_tag,
|
||||
change_user_tag,
|
||||
update_chat_all_stat,
|
||||
update_user_all_stat,
|
||||
add_message,
|
||||
) = get_module(
|
||||
"standard.database",
|
||||
[
|
||||
"db_api.get_chat",
|
||||
"db_api.add_chat",
|
||||
"db_api.get_user",
|
||||
"db_api.add_user",
|
||||
"db_api.get_user_name",
|
||||
"db_api.change_user_name",
|
||||
"db_api.get_user_tag",
|
||||
"db_api.change_user_tag",
|
||||
"db_api.update_chat_all_stat",
|
||||
"db_api.update_user_all_stat",
|
||||
"db_api.add_message",
|
||||
],
|
||||
)
|
||||
from ocab_modules.standard.database.db_api import *
|
||||
|
||||
|
||||
async def chat_check(message: types.Message):
|
||||
@ -17,17 +48,15 @@ async def chat_check(message: types.Message):
|
||||
# Если чата нет в базе данных, то проверяем его в наличии в конфиге и если он там есть то добавляем его в БД
|
||||
# Если чат есть в базе данных, то pass
|
||||
if get_chat(message.chat.id) is None:
|
||||
if message.chat.id in get_aproved_chat_id():
|
||||
if message.chat.id in get_approved_chat_id():
|
||||
# print(f"Chat in approve list: {message.chat.id} {message.chat.title}")
|
||||
await log(f"Chat in approve list: {message.chat.id} {message.chat.title}")
|
||||
log(f"Chat in approve list: {message.chat.id} {message.chat.title}")
|
||||
add_chat(message.chat.id, message.chat.title)
|
||||
# print(f"Chat added: {message.chat.id} {message.chat.title}")
|
||||
await log(f"Chat added: {message.chat.id} {message.chat.title}")
|
||||
log(f"Chat added: {message.chat.id} {message.chat.title}")
|
||||
else:
|
||||
# print(f"Chat not in approve list: {message.chat.id} {message.chat.title}")
|
||||
await log(
|
||||
f"Chat not in approve list: {message.chat.id} {message.chat.title}"
|
||||
)
|
||||
log(f"Chat not in approve list: {message.chat.id} {message.chat.title}")
|
||||
pass
|
||||
else:
|
||||
# Проверяем обновление названия чата
|
||||
@ -36,10 +65,10 @@ async def chat_check(message: types.Message):
|
||||
chat.chat_name = message.chat.title
|
||||
chat.save()
|
||||
# print(f"Chat updated: {message.chat.id} {message.chat.title}")
|
||||
await log(f"Chat updated: {message.chat.id} {message.chat.title}")
|
||||
log(f"Chat updated: {message.chat.id} {message.chat.title}")
|
||||
else:
|
||||
# print(f"Chat already exists: {message.chat.id} {message.chat.title}")
|
||||
await log(f"Chat already exists: {message.chat.id} {message.chat.title}")
|
||||
log(f"Chat already exists: {message.chat.id} {message.chat.title}")
|
||||
pass
|
||||
|
||||
|
||||
@ -62,23 +91,19 @@ async def user_check(message: types.Message):
|
||||
message.from_user.last_name,
|
||||
message.from_user.username,
|
||||
)
|
||||
# print(f"User added: {message.from_user.id} {message.from_user.first_name} {message.from_user.last_name}")
|
||||
await log(f"User added: {message.from_user.id} {current_user_name}")
|
||||
log(f"User added: {message.from_user.id} {current_user_name}")
|
||||
else:
|
||||
# print(f"User already exists: {message.from_user.id} {message.from_user.first_name} {message.from_user.last_name} {message.from_user.username}")
|
||||
await log(
|
||||
log(
|
||||
f"User already exists: {message.from_user.id} {current_user_name} {message.from_user.username}"
|
||||
)
|
||||
# Проверяем обновление имени пользователя
|
||||
if get_user_name(message.from_user.id) != current_user_name:
|
||||
change_user_name(message.from_user.id, current_user_name)
|
||||
# print(f"User updated: {message.from_user.id} {message.from_user.first_name} {message.from_user.last_name}")
|
||||
await log(f"User name updated: {message.from_user.id} {current_user_name}")
|
||||
log(f"User name updated: {message.from_user.id} {current_user_name}")
|
||||
# Проверяем обновление username пользователя
|
||||
if get_user_tag(message.from_user.id) != message.from_user.username:
|
||||
change_user_tag(message.from_user.id, message.from_user.username)
|
||||
# print(f"User updated: {message.from_user.id} {message.from_user.username}")
|
||||
await log(
|
||||
log(
|
||||
f"User tag updated: {message.from_user.id} {message.from_user.username}"
|
||||
)
|
||||
pass
|
||||
@ -101,17 +126,19 @@ async def message_processing(message: types.Message, bot: Bot):
|
||||
if (message.text.split(" ")[0] in get_yandexgpt_start_words()) or (
|
||||
any(word in message.text for word in get_yandexgpt_in_words())
|
||||
):
|
||||
# print("message_processing")
|
||||
await log("message_processing")
|
||||
log("message_processing")
|
||||
await answer_to_message(message, bot)
|
||||
|
||||
elif message.reply_to_message is not None:
|
||||
if message.reply_to_message.from_user.is_bot:
|
||||
# print("message_processing")
|
||||
await log("message_processing")
|
||||
log("message_processing")
|
||||
await answer_to_message(message, bot)
|
||||
|
||||
|
||||
router = Router()
|
||||
# Если сообщение содержит текст то вызывается функция message_processing
|
||||
router.message.register(message_processing, F.text)
|
||||
|
||||
|
||||
async def module_init():
|
||||
register_router(router)
|
||||
|
@ -1,7 +1,9 @@
|
||||
from ocab_core.modules_system.public_api import get_module
|
||||
|
||||
get_user_role = get_module("standard.database", "db_api.get_user_role")
|
||||
config: dict = get_module("standard.config", "config")
|
||||
get_roles = get_module("standard.config", "get_roles")
|
||||
|
||||
roles = get_roles()
|
||||
|
||||
|
||||
class Roles:
|
||||
@ -9,13 +11,12 @@ class Roles:
|
||||
moderator = "MODERATOR"
|
||||
admin = "ADMIN"
|
||||
bot = "BOT"
|
||||
__roles = config["ROLES"]
|
||||
|
||||
def __init__(self):
|
||||
self.user_role_id = self.__roles[self.user]
|
||||
self.moderator_role_id = self.__roles[self.moderator]
|
||||
self.admin_role_id = self.__roles[self.admin]
|
||||
self.bot_role_id = self.__roles[self.bot]
|
||||
self.user_role_id = roles[self.user]
|
||||
self.moderator_role_id = roles[self.moderator]
|
||||
self.admin_role_id = roles[self.admin]
|
||||
self.bot_role_id = roles[self.bot]
|
||||
|
||||
async def check_admin_permission(self, user_id):
|
||||
match get_user_role(user_id):
|
||||
|
@ -7,7 +7,7 @@ from json import loads
|
||||
class Path:
|
||||
core: str
|
||||
modules_standard: str
|
||||
modules_custom: str
|
||||
modules_external: str
|
||||
|
||||
|
||||
def _get_paths(path_to_json: str):
|
||||
@ -16,7 +16,7 @@ def _get_paths(path_to_json: str):
|
||||
return Path(
|
||||
core=paths["core"],
|
||||
modules_standard=paths["modules standard"],
|
||||
modules_custom=paths["modules custom"],
|
||||
modules_external=paths["modules external"],
|
||||
)
|
||||
|
||||
|
||||
|
Loading…
Reference in New Issue
Block a user