Maks1mS/ocab
0
0
Fork 0
mirror of https://gitflic.ru/project/maks1ms/ocab.git synced 2024-12-24 00:33:05 +03:00
Code

wip

Browse Source
This commit is contained in:
Maxim Slipenko 2024-07-24 16:28:42 +03:00
parent 6aab1ee244
commit e48e83bf2c
15 changed files with 489 additions and 383 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

57
src/ocab_core/main.py
View File

@ -49,17 +49,48 @@ async def long_polling_mode():
async def webhook_mode():
singleton = Singleton()
app = FastAPI()
config = get_module("standard.config", "config")
app.mount("/webapp", singleton.storage["webapp"])
await register_bot_webhook(app, singleton.bot, singleton.dp)
await singleton.bot.set_webhook(
"https://mackerel-pumped-foal.ngrok-free.app/webhook"
await singleton.bot.set_webhook(config.get("core::webhook::public_url"))
hyperConfig = HyperConfig()
hyperConfig.bind = [f"0.0.0.0:{config.get("core::webhook::port")}"]
hyperConfig.logger_class = CustomLogger
await serve(app, hyperConfig)
def register_config():
config = get_module("standard.config", "config")
config.register(
"core::token",
"password",
visible=False,
)
config.register(
"core::mode",
"select",
options=["WEBHOOK", "LONG_POLLING"],
default_value="WEBHOOK",
visible=False,
)
config.register(
"core::webhook::port",
"int",
default_value=9000,
visible=False,
)
config.register(
"core::webhook::public_url",
"string",
visible=False,
)
config = HyperConfig()
config.bind = ["0.0.0.0:9000"]
config.logger_class = CustomLogger
await serve(app, config)
async def init_app():
@ -74,9 +105,13 @@ async def init_app():
log(f"Loading {info.name} ({info.id}) module")
await singleton.modules_manager.load(module_loader)
get_telegram_token = get_module("standard.config", ["get_telegram_token"])
register_config()
singleton.bot = Bot(token=get_telegram_token())
config = get_module("standard.config", "config")
config.load()
singleton.bot = Bot(token=config.get("core::token"))
singleton.dp = Dispatcher(storage=singleton.storage["_fsm_storage"])
singleton.dp.include_routers(*singleton.storage["_routers"])
@ -92,8 +127,12 @@ async def init_app():
async def main():
await init_app()
config = get_module("standard.config", "config")
if config.get("core::mode") == "WEBHOOK":
await webhook_mode()
# await long_polling_mode()
else:
await long_polling_mode()
if __name__ == "__main__":

46
src/ocab_modules/external/yandexgpt/main.py vendored
View File

@ -4,12 +4,42 @@ config = get_module("standard.config", "config")
def module_init():
config.register_setting(["YANDEXGPT", "TOKEN"], "", "string", is_private=True)
config.register_setting(["YANDEXGPT", "TOKEN_FOR_REQUEST"], 8000, "number")
config.register_setting(["YANDEXGPT", "TOKEN_FOR_ANSWER"], 2000, "number")
config.register_setting(["YANDEXGPT", "CATALOGID"], "", "string", is_private=True)
config.register_setting(["YANDEXGPT", "PROMPT"], "Ты чат-бот ...", "string")
config.register_setting(
["YANDEXGPT", "STARTWORD"], "Бот| Бот, | бот | бот,", "string"
config.register(
"yandexgpt::token",
"password",
required=True,
)
config.register(
"yandexgpt::token_for_request",
"int",
default_value=8000,
)
config.register(
"yandexgpt::token_for_answer",
"int",
default_value=2000,
)
config.register(
"yandexgpt::catalogid",
"password",
required=True,
)
config.register(
"yandexgpt::prompt",
"string",
default_value="Ты чат-бот ...",
)
config.register(
"yandexgpt::startword",
"string",
default_value="Бот| Бот, | бот | бот,",
)
config.register(
"yandexgpt::inword",
"string",
default_value="помогите | не работает",
)
config.register_setting(["YANDEXGPT", "INWORD"], "помогите | не работает", "string")

3
src/ocab_modules/standard/config/__init__.py
View File

@ -1,9 +1,6 @@
from .config import (
config,
get_approved_chat_id,
get_default_chat_tag,
get_roles,
get_telegram_token,
get_yandexgpt_in_words,
get_yandexgpt_prompt,
get_yandexgpt_start_words,

28
src/ocab_modules/standard/config/config.py
View File

@ -6,13 +6,16 @@ config = ConfigManager(
config_path="/home/maxim/dev/alt-gnome-infrastructure/ocab/src/ocab_core/config.yaml"
)
"""
def register_settings(settings_manager: ConfigManager):
# TELEGRAM settings
settings_manager.register_setting(["CORE", "TOKEN"], "", "string", is_private=True)
settings_manager.register_setting(
["TELEGRAM", "TOKEN"], "", "string", is_private=True
["CORE", "MODE"], "WEBHOOK", "string", options=["WEBHOOK", "LONG_POLLING"]
)
settings_manager.register_setting(
# TODO: вынести в конкретные модули
settings_manager.register(
["TELEGRAM", "APPROVED_CHAT_ID"],
"-123456789 | -012345678",
"string",
@ -32,13 +35,7 @@ def register_settings(settings_manager: ConfigManager):
pretty_name="Основной чат",
)
settings_manager.register_setting(["TELEGRAM", "CHECK_BOT"], True, "checkbox")
register_settings(config)
def get_telegram_token() -> str:
return config["TELEGRAM"]["TOKEN"]
"""
def get_telegram_check_bot() -> bool:
@ -52,15 +49,6 @@ def get_approved_chat_id() -> list:
]
def get_roles():
return config["ROLES"]
def get_user_role_name(role_number) -> dict:
# Возвращаем название роли пользвателя по номеру роли, если такой роли нет, возвращаем неизвестно
return config["ROLES"].get(role_number, "Неизвестно")
def get_default_chat_tag() -> str:
return config["TELEGRAM"]["DEFAULT_CHAT_TAG"]

349
src/ocab_modules/standard/config/config_manager.py
View File

@ -1,299 +1,92 @@
from typing import Any, Dict, List, Optional, Union
import inspect
from typing import Any, Dict, List
import flask
import yaml
try:
import dash_bootstrap_components as dbc
from dash_extensions.enrich import ALL, Input, Output, State, dcc, html
DASH_AVAILABLE = True
except ImportError:
DASH_AVAILABLE = False
class ConfigManager:
def __init__(self, config_path: str):
self._config_path = config_path
self._config = self.load_config()
self._registered_settings = dict()
self._registered_settings_meta = dict()
self._update_callbacks = []
self._update_required = False
self.config_path = config_path
@property
def config(self):
return self._config
self._config: Dict[str, Dict[str, Any]] = {}
@config.setter
def config(self, value):
self._config = value
self._metadata: Dict[str, Dict[str, Any]] = {}
def load_config(self) -> Dict[str, Any]:
with open(self._config_path, "r") as file:
return yaml.safe_load(file)
def load(self, file_path: str = ""):
if not file_path:
file_path = self.config_path
def save_config(self):
with open(self._config_path, "w") as file:
def build_key(prev, next):
if prev:
return f"{prev}::{next}"
return next
def recurse_set(value, key=""):
if isinstance(value, dict):
for k, v in value.items():
recurse_set(v, build_key(key, k))
return
if key in self._metadata:
self._config[key] = value
with open(file_path, "r", encoding="utf-8") as file:
data = yaml.safe_load(file)
recurse_set(data)
def save(self, file_path: str = ""):
if not file_path:
file_path = self.config_path
with open(file_path, "w", encoding="utf-8") as file:
yaml.dump(self._config, file, allow_unicode=True)
def register_setting(
def _check_rights(self, key, module_id, access_type="get"):
return
def get(self, key: str):
module_id = self._get_module_id()
self._check_rights(key, module_id)
return self._config.get(key, self._metadata.get(key).get("default_value"))
def get_meta(self, key: str):
module_id = self._get_module_id()
self._check_rights(key, module_id, "get_meta")
return self._metadata.get(key)
def _get_module_id(self):
caller_frame = inspect.currentframe().f_back.f_back
caller_globals = caller_frame.f_globals
module_id = caller_globals.get("__ocab_module_id__")
return module_id
def register(
self,
key: Union[str, List[str]],
default_value: Any,
setting_type: str,
is_private: bool = False,
pretty_name: str = None,
description: str = None,
options: Optional[List[str]] = None,
key: str,
value_type: str,
options: List[Any] = None,
default_value=None,
editable: bool = True,
shared: bool = False,
required: bool = False,
visible: bool = True,
pretty_name: str = "",
description: str = "",
):
if isinstance(key, str):
key = [key]
module_id = self._get_module_id()
current = self._registered_settings
for k in key[:-1]:
if k not in current:
current[k] = {}
current = current[k]
self._check_rights(key, module_id, "register")
current[key[-1]] = self.get_nested_setting(self._config, key, default_value)
if key in self._metadata:
raise ValueError("ERROR")
self.set_nested_setting(
self._registered_settings_meta,
key,
{
"type": setting_type,
"is_private": is_private,
self._metadata[key] = {
"type": value_type,
"options": options,
"default_value": default_value,
"visible": visible,
"editable": editable,
"shared": shared,
"required": required,
"pretty_name": pretty_name,
"description": description,
},
)
def get_nested_setting(
self, config: dict, keys: List[str], default: Any = None
) -> Any:
current = config
for key in keys:
if key in current:
current = current[key]
else:
return default
return current
def set_nested_setting(self, config: dict, keys: List[str], value: Any):
current = config
for key in keys[:-1]:
if key not in current:
current[key] = {}
current = current[key]
current[keys[-1]] = value
def __getitem__(self, key):
if key in self._registered_settings:
return self._registered_settings[key]
raise KeyError(key)
def update_setting(self, key: Union[str, List[str]], value: Any):
if isinstance(key, str):
key = [key]
self.set_nested_setting(self._registered_settings, key, value)
self.set_nested_setting(self._config, key, value)
self.save_config()
def get_settings_layout(self, prefix):
from dash_extensions.enrich import DashBlueprint
self._prefix = prefix
bp = DashBlueprint()
def create_layout():
def create_nested_layout(settings: dict, key_list=None):
if key_list is None:
key_list = []
components = []
for key, value in settings.items():
current_key_list = key_list.copy()
current_key_list.append(key)
if isinstance(value, dict):
nested = create_nested_layout(value, current_key_list)
if len(nested) > 0:
components.append(
dbc.Card(
[
dbc.CardHeader(html.H3(key, className="mb-0")),
dbc.CardBody(nested),
],
className="mb-3",
)
)
else:
meta = self.get_nested_setting(
self._registered_settings_meta, current_key_list
)
if not meta.get("is_private"):
row = []
label_text = meta.get("pretty_name", key)
if not label_text:
label_text = key
if meta.get("type") != "checkbox":
row.append(dbc.Label(label_text))
component_id = {
"type": "setting",
"key": "::".join(current_key_list),
"module_id": module_id,
}
if meta.get("type") == "string":
component = dbc.Input(
id=component_id,
type="text",
value=value,
)
elif meta.get("type") == "number":
component = dbc.Input(
id=component_id,
type="number",
value=value,
)
elif meta.get("type") == "checkbox":
component = dbc.Col(
[
dbc.Checkbox(
id=component_id,
value=value,
label=label_text,
)
]
)
elif meta.get("type") == "select":
options = [
{"label": opt, "value": opt}
for opt in meta.get("options", [])
]
component = dcc.Dropdown(
id=component_id,
options=options,
value=value,
)
else:
continue
row.append(component)
if meta.get("description"):
row.append(dbc.FormText(meta.get("description")))
components.append(dbc.Row(row, className="mb-3"))
return components
settings_components = create_nested_layout(self._registered_settings)
layout = html.Div(
[
html.H1("Настройки"),
dbc.Form(settings_components),
html.Div(id="save-confirmation"),
dbc.Button(
"Сохранить",
id="save-settings",
color="primary",
className="mt-3 w-100",
n_clicks=0,
),
html.Div(id="settings-update-trigger", style={"display": "none"}),
dcc.Store(id="settings-store"),
],
style={
"padding": "20px",
},
)
return layout
bp.layout = create_layout
self.setup_callbacks(bp)
return bp
def setup_callbacks(self, app):
@app.callback(
Output("save-confirmation", "children", allow_duplicate=True),
Output("settings-store", "data"),
Input("save-settings", "n_clicks"),
State({"type": "setting", "key": ALL}, "value"),
State({"type": "setting", "key": ALL}, "id"),
prevent_initial_call=True,
allow_duplicate=True,
# https://github.com/emilhe/dash-extensions/issues/344
# running=[
# (
# Output({
# id: "save-settings",
# 'n_clicks': MATCH
# }, "disabled"), True, False
# )
# ]
)
def save_settings(n_clicks, values, keys):
print(flask.g.user)
if n_clicks > 0:
updated_settings = {}
for value, id_dict in zip(values, keys):
key: str = id_dict["key"]
if self._prefix:
key = key.removeprefix(f"{self._prefix}-")
self.update_setting(key.split("::"), value)
updated_settings[key] = value
import datetime
import locale
locale.setlocale(locale.LC_TIME, "ru_RU.UTF-8")
now = datetime.datetime.now()
date_str = now.strftime("%H:%M:%S")
return (
dbc.Alert(
f"Настройки сохранены в {date_str}",
color="success",
duration=10000,
),
date_str,
)
app.clientside_callback(
"""
function(n_clicks) {
const buttonSelector = '#%s-save-settings';
if (n_clicks > 0) {
document.querySelector(buttonSelector).disabled = true;
}
}
"""
% (self._prefix),
Input("save-settings", "n_clicks"),
)
app.clientside_callback(
"""
function(data) {
const buttonSelector = '#%s-save-settings';
if (data) {
document.querySelector(buttonSelector).disabled = false;
}
}
"""
% (self._prefix),
Input("settings-store", "data"),
)

8
src/ocab_modules/standard/config/main.py
View File

@ -1,16 +1,20 @@
from ocab_core.modules_system.public_api import get_module, log
from .config import config
from .miniapp_ui import get_miniapp_blueprint
def register_settings_page():
try:
register_page = get_module("standard.miniapp", "register_page")
prefix = "settings"
register_page(
name="Настройки",
path="/settings",
blueprint=config.get_settings_layout("settings"),
prefix="settings",
blueprint=get_miniapp_blueprint(config, prefix),
prefix=prefix,
)
pass

223
src/ocab_modules/standard/config/miniapp_ui.py Normal file
View File

@ -0,0 +1,223 @@
from .config_manager import ConfigManager
try:
import dash_bootstrap_components as dbc
from dash_extensions.enrich import ALL, Input, Output, State, dcc, html
DASH_AVAILABLE = True
except ImportError:
DASH_AVAILABLE = False
def create_control(key, config):
value = config.get(key)
meta = config.get_meta(key)
component_id = {
"type": "setting",
"key": key,
}
label_text = meta.get("pretty_name") or key
if meta.get("type") in ["string", "int", "float", "password"]:
input_type = {
"string": "text",
"int": "number",
"float": "number",
"password": "password",
}.get(meta.get("type"), "text")
input_props = {
"id": component_id,
"type": input_type,
}
if meta.get("type") != "password":
input_props["value"] = value
if meta.get("type") == "int":
input_props["step"] = 1
input_props["pattern"] = r"\d+"
elif meta.get("type") == "float":
input_props["step"] = "any"
component = dbc.Input(**input_props, invalid=False)
elif meta.get("type") == "select":
options = [{"label": opt, "value": opt} for opt in meta.get("options", [])]
component = dcc.Dropdown(
id=component_id,
options=options,
value=value,
style={
"padding-left": 0,
"padding-right": 0,
},
)
elif meta.get("type") == "checkbox":
component = dbc.Checkbox(
id=component_id,
checked=value,
label=label_text,
)
else:
return None
row = []
if meta.get("type") != "checkbox":
row.append(dbc.Label(label_text))
row.append(component)
if meta.get("description"):
row.append(dbc.FormText(meta.get("description")))
return dbc.Row(row, className="mb-3 mx-1")
def build_settings_tree(config):
tree = {}
for key, value in config._metadata.items():
if not value["visible"]:
continue
parts = key.split("::")
control = create_control(key, config)
current = tree
for i, part in enumerate(parts[:-1]):
if part not in current:
current[part] = {"__controls": []}
current = current[part]
current["__controls"].append(control)
return tree
def create_card(category, controls):
return dbc.Card(
[
dbc.CardHeader(html.H3(category, className="mb-0")),
dbc.CardBody(controls),
],
className="mb-3",
)
def create_settings_components(tree, level=0):
components = []
for category, subtree in tree.items():
if category == "__controls":
continue
controls = subtree.get("__controls", [])
subcomponents = create_settings_components(subtree, level + 1)
if controls or subcomponents:
card_content = controls + subcomponents
card = create_card(category, card_content)
components.append(card)
return components
def get_miniapp_blueprint(config: ConfigManager, prefix: str):
import datetime
import locale
from dash_extensions.enrich import DashBlueprint
bp = DashBlueprint()
def create_layout():
settings_tree = build_settings_tree(config)
settings_components = create_settings_components(settings_tree)
layout = html.Div(
[
html.Script(),
html.H1("Настройки"),
dbc.Form(settings_components),
html.Div(id="save-confirmation"),
dbc.Button(
"Сохранить",
id="save-settings",
color="primary",
className="mt-3 w-100",
n_clicks=0,
),
html.Div(id="settings-update-trigger", style={"display": "none"}),
dcc.Store(id="settings-store"),
],
style={
"padding": "20px",
},
)
return layout
bp.layout = create_layout
@bp.callback(
Output("save-confirmation", "children"),
Output("settings-store", "data"),
Input("save-settings", "n_clicks"),
State({"type": "setting", "key": ALL}, "value"),
State({"type": "setting", "key": ALL}, "id"),
prevent_initial_call=True,
allow_duplicate=True,
)
def save_settings(n_clicks, values, keys):
if n_clicks > 0:
# TODO: добавить валидацию значений
updated_settings = {}
for value, id_dict in zip(values, keys):
key: str = id_dict["key"]
if prefix:
key = key.removeprefix(f"{prefix}-")
updated_settings[key] = value
locale.setlocale(locale.LC_TIME, "ru_RU.UTF-8")
now = datetime.datetime.now()
date_str = now.strftime("%H:%M:%S")
return (
dbc.Alert(
f"Настройки сохранены в {date_str}",
color="success",
duration=10000,
),
date_str,
)
bp.clientside_callback(
"""
function(n_clicks) {
const buttonSelector = '#%s-save-settings';
if (n_clicks > 0) {
document.querySelector(buttonSelector).disabled = true;
}
}
"""
% (prefix),
Input("save-settings", "n_clicks"),
)
bp.clientside_callback(
"""
function(data) {
const buttonSelector = '#%s-save-settings';
if (data) {
document.querySelector(buttonSelector).disabled = false;
}
}
"""
% (prefix),
Input("settings-store", "data"),
)
return bp

7
src/ocab_modules/standard/filters/__init__.py
View File

@ -1 +1,6 @@
from .filters import ChatModerOrAdminFilter, ChatNotInApproveFilter
from .filters import (
ChatModerOrAdminFilter,
ChatNotInApproveFilter,
chat_not_in_approve,
module_init,
)

32
src/ocab_modules/standard/filters/filters.py
View File

@ -4,10 +4,31 @@ from aiogram.types import Message
from ocab_core.modules_system.public_api import get_module, log
get_approved_chat_id = get_module("standard.config", "get_approved_chat_id")
config = get_module("standard.config", "config")
Roles = get_module("standard.roles", "Roles")
def module_init():
config.register("filters::approved_chat_id", "string")
def get_approved_chat_id() -> list:
# Возваращем сплитованный список id чатов в формате int
return [
int(chat_id) for chat_id in config.get("filters::approved_chat_id").split(" | ")
]
def chat_not_in_approve(message: Message) -> bool:
chat_id = message.chat.id
if chat_id in get_approved_chat_id():
log(f"Chat in approve list: {chat_id}")
return False
else:
log(f"Chat not in approve list: {chat_id}")
return True
class ChatModerOrAdminFilter(BaseFilter):
async def __call__(self, message: Message, bot: Bot) -> bool:
user_id = message.from_user.id
@ -22,11 +43,4 @@ class ChatModerOrAdminFilter(BaseFilter):
class ChatNotInApproveFilter(BaseFilter):
async def __call__(self, message: Message, bot: Bot) -> bool:
log("chat_check")
chat_id = message.chat.id
if chat_id in get_approved_chat_id():
log(f"Chat in approve list: {chat_id}")
return False
else:
log(f"Chat not in approve list: {chat_id}")
return True
return chat_not_in_approve(message)

8
src/ocab_modules/standard/message_processing/message_api.py
View File

@ -6,11 +6,13 @@ 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(
(get_yandexgpt_in_words, get_yandexgpt_start_words) = get_module(
"standard.config",
["get_approved_chat_id", "get_yandexgpt_in_words", "get_yandexgpt_start_words"],
["get_yandexgpt_in_words", "get_yandexgpt_start_words"],
)
chat_not_in_approve = get_module("standard.filters", ["chat_not_in_approve"])
answer_to_message = get_module("external.yandexgpt", "answer_to_message")
(
@ -48,7 +50,7 @@ async def chat_check(message: types.Message):
# Если чата нет в базе данных, то проверяем его в наличии в конфиге и если он там есть то добавляем его в БД
# Если чат есть в базе данных, то pass
if get_chat(message.chat.id) is None:
if message.chat.id in get_approved_chat_id():
if not chat_not_in_approve(message):
# print(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)

27
src/ocab_modules/standard/miniapp/dash_telegram_auth.py
View File

@ -1,8 +1,3 @@
import hashlib
import hmac
import time
from urllib.parse import parse_qsl
import flask
from aiogram.utils.web_app import safe_parse_webapp_init_data
from dash import Dash
@ -13,28 +8,6 @@ from flask import request
def get_auth_server(bot_token: str):
server = flask.Flask(__name__)
def validate_init_data(init_data):
try:
init_data = dict(parse_qsl(init_data))
received_hash = init_data.pop("hash")
data_check_string = "\n".join(
f"{k}={v}" for k, v in sorted(init_data.items())
)
secret_key = hmac.new(
b"WebAppData", bot_token.encode(), hashlib.sha256
).digest()
calculated_hash = hmac.new(
secret_key, data_check_string.encode(), hashlib.sha256
).hexdigest()
auth_date = int(init_data.get("auth_date", 0))
if (time.time() - auth_date) > 86400:
return False, "Init data is outdated"
return calculated_hash == received_hash, init_data
except Exception as e:
return False, str(e)
@server.before_request
def add_auth_data():
init_data = request.cookies.get("tg_init_data")

7
src/ocab_modules/standard/miniapp/lib.py
View File

@ -5,8 +5,7 @@ import dash_bootstrap_components as dbc
from dash_extensions.enrich import DashBlueprint, DashProxy, Input, Output, dcc, html
from dash_extensions.pages import setup_page_components
# TODO: заменить на get_module("standard.config")
from ocab_modules.standard.config.config import get_telegram_token
from ocab_core.modules_system.public_api import get_module
from .dash_telegram_auth import get_auth_server, setup_auth_clientcallback
@ -29,9 +28,11 @@ def register_home_page():
register_home_page()
config = get_module("standard.config", "config")
def create_dash_app(requests_pathname_prefix: str = None) -> dash.Dash:
server = get_auth_server(get_telegram_token())
server = get_auth_server(config.get("core::token"))
app = DashProxy(
pages_folder="",

28
src/ocab_modules/standard/miniapp/main.py
View File

@ -1,7 +1,13 @@
from aiogram import types
from fastapi.middleware.wsgi import WSGIMiddleware
from ocab_core.modules_system.public_api import Storage, set_chat_menu_button
from ocab_core.modules_system.public_api import (
Storage,
get_module,
set_chat_menu_button,
)
config = get_module("standard.config", "config")
def get_link():
@ -9,6 +15,20 @@ def get_link():
def module_init():
config.register(
"miniapp::prefix",
"string",
default_value="/webapp/",
visible=False,
)
config.register(
"miniapp::public_url",
"string",
visible=False,
)
pass
@ -19,13 +39,11 @@ def register_page():
async def module_late_init():
from .lib import create_dash_app
dash_app = create_dash_app(requests_pathname_prefix="/webapp/")
dash_app = create_dash_app(requests_pathname_prefix=config.get("miniapp::prefix"))
Storage.set("webapp", WSGIMiddleware(dash_app.server))
web_app_info = types.WebAppInfo(
url="https://mackerel-pumped-foal.ngrok-free.app/webapp"
)
web_app_info = types.WebAppInfo(url=config.get("miniapp::public_url"))
menu_button = types.MenuButtonWebApp(text="Меню", web_app=web_app_info)
await set_chat_menu_button(menu_button)

28
src/ocab_modules/standard/roles/main.py
View File

@ -4,9 +4,29 @@ from ocab_core.modules_system.public_api import get_module
def module_init():
config = get_module("standard.config", "config")
config.register_setting(["ROLES", "ADMIN"], 2, "number", is_private=True)
config.register_setting(["ROLES", "MODERATOR"], 1, "number", is_private=True)
config.register_setting(["ROLES", "USER"], 0, "number", is_private=True)
config.register_setting(["ROLES", "BOT"], 3, "number", is_private=True)
config.register(
"roles::admin",
"number",
default_value=2,
visible=False,
)
config.register(
"roles::moderator",
"number",
default_value=1,
visible=False,
)
config.register(
"roles::user",
"number",
default_value=0,
visible=False,
)
config.register(
"roles::bot",
"number",
default_value=3,
visible=False,
)
pass

11
src/ocab_modules/standard/roles/roles.py
View File

@ -1,7 +1,7 @@
from ocab_core.modules_system.public_api import get_module
get_user_role = get_module("standard.database", "db_api.get_user_role")
get_roles = get_module("standard.config", "get_roles")
config = get_module("standard.config", "config")
class Roles:
@ -14,11 +14,10 @@ class Roles:
pass
def update_roles(self):
roles = get_roles()
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]
self.user_role_id = config.get("roles::user")
self.moderator_role_id = config.get("roles::moderator")
self.admin_role_id = config.get("roles::admin")
self.bot_role_id = config.get("roles::bot")
async def check_admin_permission(self, user_id):
self.update_roles()