Maks1mS/ocab
0
0
Fork 0
mirror of https://gitflic.ru/project/maks1ms/ocab.git synced 2024-12-23 16:23:01 +03:00
Code

wip

Browse Source
This commit is contained in:
Maxim Slipenko 2024-08-01 11:16:08 +03:00
parent 9fa23f18b9
commit 2becd33774
5 changed files with 241 additions and 53 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/ocab_core/ocab_core/main.py
View File

@ -78,7 +78,7 @@ class OCAB:
singleton = Singleton()
app = FastAPI()
config = get_module("standard.config", "config")
app.mount("/webapp", singleton.storage["webapp"])
app.mount(config.get("miniapp::prefix"), singleton.storage["webapp"])
await register_bot_webhook(app, singleton.bot, singleton.dp)
await singleton.bot.set_webhook(config.get("core::webhook::public_url"))
hyperConfig = HyperConfig()

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

@ -15,6 +15,7 @@ def register_settings_page():
path="/settings",
blueprint=get_miniapp_blueprint(config, prefix),
prefix=prefix,
role="ADMIN",
)
pass

43
src/ocab_modules/ocab_modules/standard/config/miniapp_ui.py
View File

@ -1,13 +1,22 @@
import asyncio
from typing import TYPE_CHECKING
from .config_manager import ConfigManager
try:
import dash_bootstrap_components as dbc
import flask
from dash_extensions.enrich import ALL, Input, Output, State, dcc, html
DASH_AVAILABLE = True
except ImportError:
DASH_AVAILABLE = False
from ocab_core.modules_system.public_api import get_module
if TYPE_CHECKING:
from ocab_modules.standard.roles import Roles as IRoles
def create_control(key: str, config: ConfigManager):
value = config.get(key)
@ -126,6 +135,10 @@ def create_settings_components(tree, level=0):
def get_miniapp_blueprint(config: ConfigManager, prefix: str):
Roles: "type[IRoles]" = get_module("standard.roles", "Roles")
roles = Roles()
import datetime
from dash_extensions.enrich import DashBlueprint
@ -171,6 +184,28 @@ def get_miniapp_blueprint(config: ConfigManager, prefix: str):
)
def save_settings(n_clicks, values, keys):
if n_clicks > 0:
user = getattr(flask.g, "user", None)
if user is None:
return (
dbc.Alert(
"Вы не авторизованы!",
color="danger",
duration=10000,
),
"-",
)
if not asyncio.run(roles.check_admin_permission(user["id"])):
return (
dbc.Alert(
"Вы не администратор!",
color="danger",
duration=10000,
),
"-",
)
# TODO: добавить валидацию значений
updated_settings = {}
@ -178,12 +213,18 @@ def get_miniapp_blueprint(config: ConfigManager, prefix: str):
key: str = id_dict["key"]
if prefix:
key = key.removeprefix(f"{prefix}-")
meta = config.get_meta(key)
if meta["type"] == "password":
if value: # Only update if a new value is provided
updated_settings[key] = value
else:
updated_settings[key] = value
config.mass_set(updated_settings)
config.save()
# locale.setlocale(locale.LC_TIME, "ru_RU.UTF-8")
now = datetime.datetime.now()
date_str = now.strftime("%H:%M:%S")

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

@ -1,13 +1,104 @@
import flask
from aiogram.utils.web_app import safe_parse_webapp_init_data
from dash import Dash
from dash_extensions.enrich import Input
from dash_extensions.enrich import Input, Output
from flask import request
# TODO: добавить прокидывание BASE_PATH, т.к. это параметр из настроек
WEBAPP_LOADER_TEMPLATE = """
<!DOCTYPE html>
<html lang="ru">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>OCAB</title>
<script src="https://telegram.org/js/telegram-web-app.js"></script>
<script>
window.addEventListener('message', function(event) {
if (event.origin !== window.location.origin) return;
if (event.data.type === 'iframe-url-changed') {
history.pushState(
null,
'',
window.BASE_PATH + event.data.pathname.substring(
window.INTERNAL_PATH.length
)
);
}
});
window.addEventListener('popstate', function(event) {
var iframe = document.getElementById('app-frame');
var iframeWindow = iframe.contentWindow;
iframeWindow.history.back();
});
</script>
<style>
#app-frame {
display:none;
width:100%;
height:100vh;
border:none;
position: absolute;
top: 0;
left: 0;
}
</style>
</head>
<body>
<div id="loading">Loading...</div>
<iframe id="app-frame"></iframe>
<script>
document.addEventListener('DOMContentLoaded', function() {
const tg = window.Telegram.WebApp;
document.cookie = `tg_init_data=${JSON.stringify(tg.initData)}; path=/`;
// if (!tg.initData) return;
const iframe = document.getElementById('app-frame');
// Константы для путей
const BASE_PATH = '/webapp';
const INTERNAL_PATH = '/webapp/_internal';
window.BASE_PATH = BASE_PATH;
window.INTERNAL_PATH = INTERNAL_PATH
// Текущий путь страницы
const currentPath = window.location.pathname;
// Формируем новый путь для iframe
let iframeSrc = INTERNAL_PATH;
// Если текущий путь начинается с BASE_PATH, убираем BASE_PATH из текущего пути
if (currentPath.startsWith(BASE_PATH)
&& currentPath.length > BASE_PATH.length) {
iframeSrc += currentPath.substring(BASE_PATH.length);
} else if (currentPath !== '/') {
iframeSrc += currentPath;
}
iframe.src = iframeSrc;
iframe.onload = function() {
document.getElementById('loading').style.display = 'none';
iframe.style.display = 'block';
};
});
</script>
</body>
</html>
"""
def get_auth_server(bot_token: str):
server = flask.Flask(__name__)
@server.route("/<path:rest>")
@server.route("/")
def webapp_loader(rest=None):
return flask.Response(WEBAPP_LOADER_TEMPLATE, mimetype="text/html")
@server.before_request
def add_auth_data():
init_data = request.cookies.get("tg_init_data")
@ -21,13 +112,35 @@ def get_auth_server(bot_token: str):
return server
def setup_auth_clientcallback(app: Dash):
def setup_auth_clientcallbacks(app: Dash):
app.clientside_callback(
"""
function(n_inervals) {
const tg = window.Telegram.WebApp;
document.cookie = `tg_init_data=${JSON.stringify(tg.initData)}; path=/`;
function(n_intervals) {
if (window.webAppData) {
return window.webAppData;
}
function receiveMessage(event) {
if (event.data.type === 'webAppData') {
window.webAppData = event.data.webApp;
window.removeEventListener('message', receiveMessage);
}
}
window.addEventListener('message', receiveMessage, false);
return window.dash_clientside.no_update;
}
""",
Input("init-telegram-interval", "n_intervals"),
Output("hidden-div", "children"),
Input("interval-component", "n_intervals"),
)
app.clientside_callback(
"""
function(pathname) {
window.parent.postMessage({ type: 'iframe-url-changed', pathname }, '*');
}
""",
Input("url", "pathname"),
)

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

@ -1,26 +1,30 @@
import asyncio
from collections import OrderedDict
from typing import TYPE_CHECKING
import dash
import dash_bootstrap_components as dbc
import flask
from dash_extensions.enrich import DashBlueprint, DashProxy, Input, Output, dcc, html
from dash_extensions.pages import setup_page_components
from ocab_core.modules_system.public_api import get_module
from ocab_core.modules_system.public_api import get_module, log
from .dash_telegram_auth import get_auth_server, setup_auth_clientcallback
from .dash_telegram_auth import get_auth_server, setup_auth_clientcallbacks
if TYPE_CHECKING:
from ocab_modules.standard.config import IConfig
from ocab_modules.standard.roles import Roles as IRoles
pages = OrderedDict()
def register_page(name, path, blueprint, prefix=""):
def register_page(name, path, blueprint, prefix="", role="USER"):
pages[path] = {
"name": name,
"blueprint": blueprint,
"prefix": prefix,
"role": role,
}
@ -33,9 +37,14 @@ def register_home_page():
register_home_page()
config: "IConfig" = get_module("standard.config", "config")
Roles: "type[IRoles]" = get_module("standard.roles", "Roles")
def create_dash_app(requests_pathname_prefix: str = None) -> dash.Dash:
log(requests_pathname_prefix)
real_prefix = f"{requests_pathname_prefix}_internal/"
server = get_auth_server(config.get("core::token"))
app = DashProxy(
@ -47,10 +56,13 @@ def create_dash_app(requests_pathname_prefix: str = None) -> dash.Dash:
dbc.icons.BOOTSTRAP,
],
external_scripts=[
#
"https://telegram.org/js/telegram-web-app.js"
], # Add Telegram Mini Apps script to <head>
],
server=server,
requests_pathname_prefix=requests_pathname_prefix,
requests_pathname_prefix=real_prefix,
routes_pathname_prefix="/_internal/",
# requests_pathname_prefix=requests_pathname_prefix,
meta_tags=[
{"name": "viewport", "content": "width=device-width, initial-scale=1"},
],
@ -63,30 +75,8 @@ def create_dash_app(requests_pathname_prefix: str = None) -> dash.Dash:
# Register pages
for path, page in pages.items():
# dash.register_page(page["name"], path=path, layout=page["layout"])
page["blueprint"].register(app, path, prefix=page["prefix"])
# Create sidebar
sidebar = dbc.Offcanvas(
id="offcanvas",
title="Меню",
is_open=False,
children=[
dbc.Nav(
[
dbc.NavLink(
page["name"],
href=f"{requests_pathname_prefix}{path.lstrip('/')}",
id={"type": "nav-link", "index": path},
)
for path, page in pages.items()
],
vertical=True,
pills=True,
),
],
)
# Create navbar
navbar = dbc.Navbar(
dbc.Container(
@ -104,8 +94,47 @@ def create_dash_app(requests_pathname_prefix: str = None) -> dash.Dash:
dark=True,
)
# Define app layout
app.layout = html.Div(
roles = Roles()
def create_layout():
user = getattr(flask.g, "user", None)
if not user:
return html.Div()
user_id = user["id"]
user_permission = asyncio.run(roles.get_user_permission(user_id)) or "USER"
available_pages = {
path: page
for path, page in pages.items()
if (isinstance(page["role"], list) and user_permission in page["role"])
or page["role"] == user_permission
or page["role"] == "USER"
}
# Create sidebar
sidebar = dbc.Offcanvas(
id="offcanvas",
title="Меню",
is_open=False,
children=[
dbc.Nav(
[
dbc.NavLink(
page["name"],
href=f"{real_prefix}/{path.lstrip('/')}",
id={"type": "nav-link", "index": path},
)
for path, page in available_pages.items()
],
vertical=True,
pills=True,
),
],
)
layout = html.Div(
[
dcc.Location(id="url", refresh=False),
dcc.Interval(
@ -121,7 +150,11 @@ def create_dash_app(requests_pathname_prefix: str = None) -> dash.Dash:
]
)
setup_auth_clientcallback(app)
return layout
app.layout = create_layout
setup_auth_clientcallbacks(app)
# Открытие на кнопку меню
app.clientside_callback(