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

добавил типизацию

This commit is contained in:
Maxim Slipenko 2024-07-27 22:23:42 +03:00
parent e48e83bf2c
commit 5b02c2ce6a
20 changed files with 132 additions and 170 deletions

View File

@ -28,7 +28,7 @@ bot_modules = [
ocab_modules_loader("standard", "database", safe=False),
ocab_modules_loader("standard", "fsm_database_storage", safe=False),
ocab_modules_loader("standard", "roles", safe=False),
ocab_modules_loader("external", "yandexgpt", safe=True),
ocab_modules_loader("external", "yandexgpt", safe=False),
#
ocab_modules_loader("standard", "command_helper"),
ocab_modules_loader("standard", "info"),

View File

@ -142,3 +142,4 @@ BUILTINS["_write_"] = write_guard()
BUILTINS["_unpack_sequence_"] = guarded_unpack_sequence
BUILTINS["staticmethod"] = staticmethod
BUILTINS["tuple"] = tuple
BUILTINS["reversed"] = reversed

View File

@ -1,4 +1,6 @@
# flake8: noqa
from typing import TYPE_CHECKING
from aiogram import Bot
from aiogram.types import Message
@ -6,12 +8,29 @@ from ocab_core.modules_system.public_api import get_module, log
from .yandexgpt import YandexGPT
(get_yandexgpt_catalog_id, get_yandexgpt_token, get_yandexgpt_prompt) = get_module(
if TYPE_CHECKING:
from ocab_modules.standard.config import IConfig
from ocab_modules.standard.database.db_api import add_message as IAddMessage
config: "IConfig" = get_module(
"standard.config",
["get_yandexgpt_catalog_id", "get_yandexgpt_token", "get_yandexgpt_prompt"],
"config",
)
add_message = get_module("standard.database", "db_api.add_message")
def get_yandexgpt_catalog_id():
return config.get("yandexgpt::catalogid")
def get_yandexgpt_token():
return config.get("yandexgpt::token")
def get_yandexgpt_prompt():
return config.get("yandexgpt::prompt")
add_message: "IAddMessage" = get_module("standard.database", "db_api.add_message")
async def answer_to_message(message: Message, bot: Bot):

View File

@ -1,6 +1,11 @@
from typing import TYPE_CHECKING
from ocab_core.modules_system.public_api import get_module
config = get_module("standard.config", "config")
if TYPE_CHECKING:
from ocab_modules.standard.config import IConfig
config: "IConfig" = get_module("standard.config", "config")
def module_init():

View File

@ -1,25 +1,31 @@
# flake8: noqa
import json
from typing import TYPE_CHECKING
import aiohttp
import requests
from ocab_core.modules_system.public_api import get_module, log
db_api = get_module("standard.database", "db_api")
if TYPE_CHECKING:
from ocab_modules.standard.config import IConfig
from ocab_modules.standard.database import db_api as IDbApi
(
get_yandexgpt_token_for_answer,
get_yandexgpt_prompt,
get_yandexgpt_token_for_request,
) = get_module(
"standard.config",
[
"get_yandexgpt_token_for_answer",
"get_yandexgpt_prompt",
"get_yandexgpt_token_for_request",
],
)
db_api: "IDbApi" = get_module("standard.database", "db_api")
config: "IConfig" = get_module("standard.config", "config")
def get_yandexgpt_token_for_answer():
return config.get("yandexgpt::token_for_answer")
def get_yandexgpt_token_for_request():
return config.get("yandexgpt::token_for_request")
def get_yandexgpt_prompt():
return config.get("yandexgpt::prompt")
class YandexGPT:

View File

@ -1,10 +1,19 @@
# flake8: noqa
from typing import TYPE_CHECKING
from aiogram import Bot
from aiogram.types import Message
from ocab_core.modules_system.public_api import get_module
get_default_chat_tag = get_module("standard.config", "get_default_chat_tag")
if TYPE_CHECKING:
from ocab_modules.standard.config import IConfig
config: "IConfig" = get_module("standard.config", "config")
def get_default_chat_tag():
return config.get("filters::default_chat_tag")
async def delete_message(message: Message, bot: Bot):

View File

@ -1,4 +1,4 @@
from typing import Any, Awaitable, Callable, Dict
from typing import TYPE_CHECKING, Any, Awaitable, Callable, Dict
from aiogram import BaseMiddleware
from aiogram.types import BotCommand, TelegramObject
@ -9,15 +9,18 @@ from ocab_core.modules_system.public_api import (
set_my_commands,
)
if TYPE_CHECKING:
from ocab_modules.standard.database import db_api as IDbApi
from ocab_modules.standard.roles import Roles as IRoles
commands = dict()
db_api = get_module(
db_api: "IDbApi" = get_module(
"standard.database",
"db_api",
)
Roles = get_module("standard.roles", "Roles")
Roles: "IRoles" = get_module("standard.roles", "Roles")
def register_command(command, description, role="USER"):

View File

@ -1,10 +1,2 @@
from .config import (
config,
get_default_chat_tag,
get_yandexgpt_in_words,
get_yandexgpt_prompt,
get_yandexgpt_start_words,
get_yandexgpt_token_for_answer,
get_yandexgpt_token_for_request,
)
from .config import IConfig, config
from .main import module_late_init

View File

@ -2,84 +2,16 @@
from .config_manager import ConfigManager
config = ConfigManager(
IConfig = ConfigManager
config: ConfigManager = ConfigManager(
config_path="/home/maxim/dev/alt-gnome-infrastructure/ocab/src/ocab_core/config.yaml"
)
"""
def register_settings(settings_manager: ConfigManager):
settings_manager.register_setting(["CORE", "TOKEN"], "", "string", is_private=True)
settings_manager.register_setting(
["CORE", "MODE"], "WEBHOOK", "string", options=["WEBHOOK", "LONG_POLLING"]
)
# TODO: вынести в конкретные модули
settings_manager.register(
["TELEGRAM", "APPROVED_CHAT_ID"],
"-123456789 | -012345678",
"string",
pretty_name="ID разрешенных чатов",
description='Чаты, в которых будет работать бот. "|" - разделитель',
)
settings_manager.register_setting(
["TELEGRAM", "ADMINCHATID"],
-12345678,
"number",
pretty_name="ID чата администраторов",
)
settings_manager.register_setting(
["TELEGRAM", "DEFAULT_CHAT_TAG"],
"@alt_gnome_chat",
"string",
pretty_name="Основной чат",
)
settings_manager.register_setting(["TELEGRAM", "CHECK_BOT"], True, "checkbox")
"""
def get_telegram_check_bot() -> bool:
return config["TELEGRAM"]["CHECK_BOT"]
def get_approved_chat_id() -> list:
# Возваращем сплитованный список id чатов в формате int
return [
int(chat_id) for chat_id in config["TELEGRAM"]["APPROVED_CHAT_ID"].split(" | ")
]
def get_default_chat_tag() -> str:
return config["TELEGRAM"]["DEFAULT_CHAT_TAG"]
def get_yandexgpt_token() -> str:
return config["YANDEXGPT"]["TOKEN"]
def get_yandexgpt_catalog_id() -> str:
return config["YANDEXGPT"]["CATALOGID"]
def get_yandexgpt_prompt() -> str:
return config["YANDEXGPT"]["PROMPT"]
def get_yandexgpt_start_words() -> list:
return config["YANDEXGPT"]["STARTWORD"].split(" | ")
def get_yandexgpt_in_words() -> list:
return config["YANDEXGPT"]["INWORD"].split(" | ")
def get_yandexgpt_token_for_request() -> int:
return config["YANDEXGPT"]["TOKEN_FOR_REQUEST"]
def get_yandexgpt_token_for_answer() -> int:
return config["YANDEXGPT"]["TOKEN_FOR_ANSWER"]
def get_access_rights() -> dict:
return config["ACCESS_RIGHTS"]

View File

@ -9,7 +9,7 @@ except ImportError:
DASH_AVAILABLE = False
def create_control(key, config):
def create_control(key: str, config: ConfigManager):
value = config.get(key)
meta = config.get_meta(key)
@ -76,7 +76,7 @@ def create_control(key, config):
return dbc.Row(row, className="mb-3 mx-1")
def build_settings_tree(config):
def build_settings_tree(config: ConfigManager):
tree = {}
for key, value in config._metadata.items():

View File

@ -158,7 +158,7 @@ def get_user_name(user_id):
return user.user_name if user else None
def get_user_role(user_id):
def get_user_role(user_id: str):
user = Users.get_or_none(Users.id == user_id)
return user.user_role if user else None

View File

@ -9,7 +9,8 @@ Roles = get_module("standard.roles", "Roles")
def module_init():
config.register("filters::approved_chat_id", "string")
config.register("filters::approved_chat_id", "string", shared=True)
config.register("filters::default_chat_tag", "string", shared=True)
def get_approved_chat_id() -> list:

View File

@ -1,5 +1,5 @@
import json
from typing import Any, Dict, Optional
from typing import TYPE_CHECKING, Any, Dict, Optional
from aiogram.fsm.state import State
from aiogram.fsm.storage.base import BaseStorage, StorageKey
@ -7,7 +7,14 @@ 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")
if TYPE_CHECKING:
from ocab_modules.standard.database.repositories import (
FSMDataRepository as IFSMDataRepository,
)
FSMDataRepository: "type[IFSMDataRepository]" = get_module(
"standard.database", "repositories.FSMDataRepository"
)
def serialize_key(key: StorageKey) -> str:

View File

@ -1,19 +1,22 @@
# flake8: noqa
from typing import Type
from typing import TYPE_CHECKING
from aiogram import Bot
from aiogram.types import Message
from ocab_core.modules_system.public_api import get_module, log
from .interfaces import IDbApi, IRoles
if TYPE_CHECKING:
from ocab_modules.standard.database import db_api as IDbApi
from ocab_modules.standard.roles import Roles as IRoles
db_api: Type[IDbApi] = get_module(
db_api: "IDbApi" = get_module(
"standard.database",
"db_api",
)
Roles: Type[IRoles] = get_module("standard.roles", "Roles")
Roles: "type[IRoles]" = get_module("standard.roles", "Roles")
async def get_info_answer_by_id(message: Message, bot: Bot, user_id: int):

View File

@ -1,47 +0,0 @@
from typing import Any
class IRoles:
user: str
moderator: str
admin: str
bot: str
def __init__(self):
pass
async def check_admin_permission(self, user_id: int) -> bool:
pass
async def check_moderator_permission(self, user_id) -> bool:
pass
async def get_role_name(self, role_id) -> str:
pass
async def get_user_permission(self, user_id) -> str | None:
pass
class IUsers:
user_id: int
user_name: str
user_role: int
user_tag: str
user_stats: int
user_rep: int
class IDbApi:
@staticmethod
def get_message_ai_model(message_chat_id: Any, message_id: Any) -> Any | None:
pass
@staticmethod
def get_user(user_id: int) -> IUsers | None:
pass
@staticmethod
def get_chat_all_stat(chat_id: int) -> int:
pass

View File

@ -1,15 +1,24 @@
# flake8: noqa
from typing import TYPE_CHECKING
from aiogram import Bot, F, Router, types
from ocab_core.modules_system.public_api import get_module, log, register_router
# from ocab_modules.standard.database.db_api import *
if TYPE_CHECKING:
from ocab_modules.standard.config import IConfig
config: "IConfig" = get_module("standard.config", "config")
def get_yandexgpt_start_words():
return config.get("yandexgpt::startword").split(" | ")
def get_yandexgpt_in_words():
return config.get("yandexgpt::inword").split(" | ")
(get_yandexgpt_in_words, get_yandexgpt_start_words) = get_module(
"standard.config",
["get_yandexgpt_in_words", "get_yandexgpt_start_words"],
)
chat_not_in_approve = get_module("standard.filters", ["chat_not_in_approve"])

View File

@ -1,4 +1,5 @@
from collections import OrderedDict
from typing import TYPE_CHECKING
import dash
import dash_bootstrap_components as dbc
@ -9,6 +10,9 @@ from ocab_core.modules_system.public_api import get_module
from .dash_telegram_auth import get_auth_server, setup_auth_clientcallback
if TYPE_CHECKING:
from ocab_modules.standard.config import IConfig
pages = OrderedDict()
@ -28,7 +32,7 @@ def register_home_page():
register_home_page()
config = get_module("standard.config", "config")
config: "IConfig" = get_module("standard.config", "config")
def create_dash_app(requests_pathname_prefix: str = None) -> dash.Dash:

View File

@ -1,3 +1,5 @@
from typing import TYPE_CHECKING
from aiogram import types
from fastapi.middleware.wsgi import WSGIMiddleware
@ -7,7 +9,10 @@ from ocab_core.modules_system.public_api import (
set_chat_menu_button,
)
config = get_module("standard.config", "config")
if TYPE_CHECKING:
from ocab_modules.standard.config import IConfig
config: "IConfig" = get_module("standard.config", "config")
def get_link():

View File

@ -1,8 +1,13 @@
from typing import TYPE_CHECKING
from ocab_core.modules_system.public_api import get_module
if TYPE_CHECKING:
from ocab_modules.standard.config import IConfig
def module_init():
config = get_module("standard.config", "config")
config: "IConfig" = get_module("standard.config", "config")
config.register(
"roles::admin",

View File

@ -1,7 +1,15 @@
from typing import TYPE_CHECKING
from ocab_core.modules_system.public_api import get_module
get_user_role = get_module("standard.database", "db_api.get_user_role")
config = get_module("standard.config", "config")
if TYPE_CHECKING:
from ocab_modules.standard.config import IConfig
from ocab_modules.standard.database.db_api import get_user_role as IGetUserRoleType
get_user_role: "IGetUserRoleType" = get_module(
"standard.database", "db_api.get_user_role"
)
config: "IConfig" = get_module("standard.config", "config")
class Roles: