mirror of
https://gitflic.ru/project/maks1ms/ocab.git
synced 2025-04-05 09:03:48 +03:00
144 lines
4.0 KiB
Python
144 lines
4.0 KiB
Python
import asyncio
|
|
import traceback
|
|
from typing import TYPE_CHECKING
|
|
|
|
from aiogram import Bot, Dispatcher
|
|
from fastapi import FastAPI
|
|
from hypercorn.asyncio import serve
|
|
from hypercorn.config import Config as HyperConfig
|
|
|
|
from ocab_core.lib import get_module_directory, register_bot_webhook
|
|
from ocab_core.logger import CustomLogger, log, setup_logger
|
|
from ocab_core.modules_system import ModulesManager
|
|
from ocab_core.modules_system.loaders import FSLoader, UnsafeFSLoader
|
|
from ocab_core.modules_system.public_api import get_module
|
|
from ocab_core.singleton import Singleton
|
|
|
|
if TYPE_CHECKING:
|
|
from ocab_modules.standard.config import IConfig
|
|
|
|
ocab_modules_path = get_module_directory("ocab_modules")
|
|
|
|
|
|
def ocab_modules_loader(namespace: str, module_name: str, safe=True):
|
|
if not safe:
|
|
return UnsafeFSLoader(f"{ocab_modules_path}/{namespace}/{module_name}")
|
|
else:
|
|
return FSLoader(f"{ocab_modules_path}/{namespace}/{module_name}")
|
|
|
|
|
|
bot_modules = [
|
|
ocab_modules_loader("standard", "config", safe=False),
|
|
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=False),
|
|
#
|
|
ocab_modules_loader("standard", "command_helper"),
|
|
ocab_modules_loader("standard", "info"),
|
|
ocab_modules_loader("standard", "filters"),
|
|
ocab_modules_loader("external", "create_report_apps"),
|
|
ocab_modules_loader("standard", "admin"),
|
|
ocab_modules_loader("standard", "message_processing"),
|
|
ocab_modules_loader("standard", "miniapp", safe=False),
|
|
]
|
|
|
|
|
|
async def long_polling_mode():
|
|
singleton = Singleton()
|
|
await singleton.bot.delete_webhook()
|
|
await singleton.dp.start_polling(singleton.bot)
|
|
|
|
|
|
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(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: "IConfig" = 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,
|
|
)
|
|
|
|
|
|
async def init_app():
|
|
setup_logger()
|
|
singleton = Singleton()
|
|
|
|
try:
|
|
singleton.modules_manager = ModulesManager()
|
|
|
|
for module_loader in bot_modules:
|
|
info = module_loader.info()
|
|
log(f"Loading {info.name} ({info.id}) module")
|
|
await singleton.modules_manager.load(module_loader)
|
|
|
|
register_config()
|
|
|
|
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"])
|
|
|
|
for middleware in singleton.storage["_outer_message_middlewares"]:
|
|
singleton.dp.message.outer_middleware.register(middleware)
|
|
|
|
await singleton.modules_manager.late_init()
|
|
|
|
except Exception:
|
|
traceback.print_exc()
|
|
raise
|
|
|
|
|
|
async def main():
|
|
await init_app()
|
|
config = get_module("standard.config", "config")
|
|
|
|
if config.get("core::mode") == "WEBHOOK":
|
|
await webhook_mode()
|
|
else:
|
|
await long_polling_mode()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|