forked from capedayo/minya_hotomoe
103 lines
3 KiB
Python
103 lines
3 KiB
Python
from pyfiglet import Figlet
|
|
|
|
f = Figlet(font="small")
|
|
|
|
import os
|
|
import importlib.util
|
|
import re
|
|
import inspect
|
|
from dotenv import load_dotenv
|
|
|
|
from module_interface import ModuleInterface, ModuleManager
|
|
|
|
import asyncio
|
|
|
|
from aiohttp import ClientWebSocketResponse
|
|
from mipac.models.notification import NotificationNote
|
|
from mipa.ext.commands.bot import Bot
|
|
|
|
|
|
def load_modules():
|
|
module_dir = "modules"
|
|
modules = []
|
|
for filename in os.listdir(module_dir):
|
|
if filename.endswith(".py"):
|
|
module_name = os.path.splitext(filename)[0]
|
|
module_path = os.path.join(module_dir, filename)
|
|
|
|
spec = importlib.util.spec_from_file_location(module_name, module_path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
|
|
for attribute_name in dir(module):
|
|
attribute = getattr(module, attribute_name)
|
|
if (
|
|
inspect.isclass(attribute)
|
|
and issubclass(attribute, ModuleInterface)
|
|
and attribute != ModuleInterface
|
|
):
|
|
modules.append(attribute())
|
|
return modules
|
|
|
|
|
|
class MiNya(Bot):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.__modules = []
|
|
self.__manager = ModuleManager(self)
|
|
|
|
async def _connect_channel(self):
|
|
await self.router.connect_channel(["main"])
|
|
|
|
async def on_reconnect(self, ws: ClientWebSocketResponse):
|
|
await self._connect_channel()
|
|
|
|
async def on_ready(self, ws: ClientWebSocketResponse):
|
|
await self._connect_channel()
|
|
print("[Main] Bot System Ready.")
|
|
print("")
|
|
print("[Main] Loading Modules...")
|
|
print("")
|
|
self.__modules = load_modules()
|
|
print("")
|
|
print(f"[Main] Loaded Modules : {len(self.__modules)}")
|
|
print("")
|
|
|
|
print("[Main] Registering Modules...")
|
|
|
|
for module in self.__modules:
|
|
self.__manager.register_module(module)
|
|
for module in self.__modules:
|
|
module.module_ready()
|
|
print("")
|
|
|
|
async def on_mention(self, notice: NotificationNote):
|
|
print("[Main] Got Mentioned!")
|
|
for _, module in self.__manager.modules.items():
|
|
if module.regex_pattern and re.match(
|
|
module.regex_pattern, notice.note.content
|
|
):
|
|
print("[Main] Regex Matching Module found : ", module.name)
|
|
if await module.execute_module(notice.note):
|
|
break
|
|
|
|
|
|
def main():
|
|
print(f.renderText("MiNya V1"))
|
|
print('"MIsskey Networked Yet Ambitious"')
|
|
print("")
|
|
print("[Main] Minya V1 Bootstap...")
|
|
print("[Main] Loading settings...")
|
|
load_dotenv()
|
|
instance = os.environ.get("BOT_INSTANCE")
|
|
key = os.environ.get("BOT_KEY")
|
|
|
|
if not (instance and key):
|
|
print("[MAIN] ERR! please check .env or os environment!")
|
|
return
|
|
bot = MiNya()
|
|
asyncio.run(bot.start(f"wss://{instance}/streaming", key))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|