awa

docs/doc/None.md

@@ -20,9 +20,9 @@ for plugin_file in confi["plugins"]["others_plugin"]:
             module = importlib.util.module_from_spec(spec)
             spec.loader.exec_module(module)
             # 检查是否实现了主方法
-            if not hasattr(module, "Plugin_Main"):
+            if not hasattr(module, "PluginMain"):
                 raise plugin_errors.NoMainMather("函数未实现主方法或者主方法名称错误")
-            plugin_main_class: plugin_main.Plugin_Main = module.Plugin_Main
+            plugin_main_class: plugin_main.PluginMain = module.PluginMain
 
             # 获取插件类型
             if plugin_main_class.plugin_type() == "message":

ictye-live-dm/__main__.py

@@ -49,25 +49,28 @@ def run_server():
 
     parse = argparse.ArgumentParser(description="一个基于python实现的模块化弹幕姬框架")
     parse.add_argument('-u', '--unportable', action='store_true', help='非便携性启动')
+    parse.add_argument("-cfg", "--config", default="", help='指定配置目錄')
     args = parse.parse_args()
 
-    unportable: bool=args.unportable
+    unportable: bool = args.unportable
     """便携启动开关"""
-
+    configdir: str = args.config
+    """配置目錄"""
 
     # 获取配置
-    config = configs.config()
+    config = configs.config(configdir)
+
     # 传递配置
     http_server.config = config
-    pluginsystem.confi = config
+    pluginsystem.global_config = config
     livewebsocket.config = config
+    # 获取logger
+    logger.setup_logging(config, unportable)
+    loggers = logging.getLogger(__name__)
     # 获取插件系统
     plugin_sys = pluginsystem.Plugin()
     livewebsocket.plugin_system = plugin_sys
     http_server.plugin_system = plugin_sys
-    # 获取logger
-    logger.setup_logging(config,unportable)
-    loggers = logging.getLogger(__name__)
 
     # 启动服务器
     loggers.info("project starting")

ictye-live-dm/config/system/config.yaml

@@ -20,6 +20,8 @@ plugins:
 #调试开关，非开发人员勿动(动吧，随便动（bushi)
 debug: 0
 
+dev: 1
+
 #日志级别
 # DEBUG:调试级别
 # INFO:一般运行

ictye-live-dm/depends/configs.py

@@ -1,9 +1,16 @@
 import yaml
 import os
 
+cfgdir: str = ""
 
-def config() -> dict:
-    with open("./config/system/config.yaml", "r", encoding="utf-8") as f:
+
+def config(cfg: str) -> dict:
+    cfgdir = cfg
+    if cfg:
+        cfgfile = cfg
+    else:
+        cfgfile = "./config/system/config.yaml"
+    with open(cfgfile, "r", encoding="utf-8") as f:
         configs = yaml.load(f.read(), Loader=yaml.FullLoader)
     if configs["debug"] == 1:
         print(f"log:already reading config file: {configs}\n")
@@ -34,7 +41,7 @@ def read_config(config_family: str) -> dict:
             configs = yaml.load(f.read(), Loader=yaml.FullLoader)
             return configs
     else:
-        os.makedirs(os.path.dirname(f"./config/plugin/{config_family}/config.yaml"),exist_ok=True)
+        os.makedirs(os.path.dirname(f"./config/plugin/{config_family}/config.yaml"), exist_ok=True)
         with open(f"./config/plugin/{config_family}/config.yaml", 'w+') as f:
             f.write(f"# config for {config_family}")
         return configs

ictye-live-dm/depends/connects.py

@@ -10,10 +10,10 @@
 #
 #   更多详情请参阅许可协议文档
 from websockets.server import WebSocketServerProtocol
+import aiohttp.web as web
 
 
 class connect_wrapper:
-
     """
     连接包装类
     """
@@ -29,4 +29,3 @@ def refresh(self):
         """
         self.id = self.__connect__.id
         self.open = self.__connect__.open
-

ictye-live-dm/depends/plugin_main.py → ictye-live-dm/depends/pluginmain.py

@@ -18,7 +18,7 @@
 from aiohttp import web
 
 
-class Plugin_Main:
+class PluginMain:
 
     @typing.final
     def __init__(self):
@@ -97,11 +97,10 @@ async def sprit_cgi(self, request):
         if self.sprit_cgi_support:
             raise plugin_errors.UnexpectedPluginMather("未实现的插件方法")
 
-    def dm_iter(self, params: dict, connect_waper: connects.connect_wrapper) -> object:
+    def dm_iter(self, params: dict) -> object:
         """
         返回弹幕迭代对象
         :param params: 前端的get参数
-        :param connect_waper: 连接信息
         :return 消息迭代对象
 
         """
@@ -155,7 +154,7 @@ def plugin_getconfig(self) -> dict:
         """
         获取配置
         """
-        return configs.config()
+        return configs.config(configs.cfgdir)
 
     @typing.final
     def plugin_type(self) -> str:

ictye-live-dm/http_server.py

@@ -16,6 +16,7 @@
 import logging
 import os
 import pluginsystem
+import livewebsocket
 
 config = dict()
 plugin_system: pluginsystem.Plugin
@@ -40,7 +41,8 @@ async def http_handler(request):
 
 async def http_socket_get(request):
     log.info("return for socket")
-    return web.Response(text=json.dumps(msgs.socket_responce(config).to_dict()))
+    return web.Response(text=json.dumps({"code": 200, "local": "/ws"}) if config["dev"] else json.dumps(
+        msgs.socket_responce(config).to_dict()))
 
 
 async def http_websocket(request: web.Request):
@@ -97,7 +99,8 @@ async def http_cgi(request):
     try:
         if request.match_info["name"] in plugin_system.plugin_cgi_support:
             if request.match_info["page"] in plugin_system.plugin_cgi_support[request.match_info["name"]]:
-                req = await plugin_system.plugin_cgi_support[request.match_info["name"]][request.match_info["page"]](request)
+                req = await plugin_system.plugin_cgi_support[request.match_info["name"]][request.match_info["page"]](
+                    request)
             else:
                 req = web.Response(status=404, text="no such path")
         else:
@@ -121,7 +124,7 @@ async def http_server(configs):
                                   web.get("/js/{name}", http_js),
                                   web.get("/js/lib/{name}", http_lib),
                                   web.get("/js/script/{name}", http_script),
-                                  web.get("/websocket", http_websocket),
+                                  web.get("/ws", livewebsocket.aiohttp_ws),
                                   web.get("/api/plugin_list", http_api_plugin),
                                   web.get("/cgi/{name}/{page}", http_cgi)
                                   ]

ictye-live-dm/livewebsocket.py

@@ -4,6 +4,8 @@
 import logging
 from depends import msgs
 import pluginsystem
+from aiohttp import web
+import aiohttp
 
 plugin_system: pluginsystem.Plugin
 config: dict = {}
@@ -56,6 +58,54 @@ async def websockets(websocket: server.WebSocketServerProtocol):
         await plugin_system.remove_connect_in_id_dict(websocket.id)
 
 
+async def aiohttp_ws(request: web.Request):
+    """
+    aiohttp ws處理程序
+    """
+
+    ws = web.WebSocketResponse()
+    await ws.prepare(request)
+
+    # 连接检测
+    try:
+        message: aiohttp.WSMessage
+        async for message in ws:
+            loggers.info("receive a message" + message.data)
+
+            # 解码信息，注意信息必须是json格式
+            ret = json.loads(message.data)
+            if (ret["code"] == 200 and ret["msg"] == "ok") or ws in connect_list:  # 连接验证
+                loggers.info("connect with blower success")
+                await ws.send_str(json.dumps(msgs.connect_ok().to_dict()))
+
+                await asyncio.sleep(0.5)
+                dms: dict
+
+                while 1:
+                    await asyncio.sleep(0.5)
+                    async for dms in plugin_system.get_plugin_message_aiohttp(ret["param"], ws):
+                        # 过滤消息，并分析消息
+                        mdm = await plugin_system.message_filter(dms)
+                        await plugin_system.message_analyzer(mdm)
+
+                        loggers.debug(f"sending message {mdm}")
+
+                        # 发送消息
+                        await ws.send_str(json.dumps(mdm))
+
+
+            else:
+                # 认证失败就关闭连接
+                loggers.error("connect failed,unexpected client")
+                await ws.close()
+
+    finally:
+        # 后续的处理
+        await ws.close()
+        await plugin_system.remove_connect_in_id_dict_aiohttp(ws)
+    return ws
+
+
 async def websocket_main(configs):
     """
     websocket主函数，通过configs传递参数字典

ictye-live-dm/plugin/.gitignore

@@ -0,0 +1 @@
+./*

ictye-live-dm/pluginsystem.py

@@ -11,15 +11,16 @@
 #   更多详情请参阅许可协议文档
 
 import asyncio
-from depends import plugin_main, plugin_errors
+from depends import pluginmain, plugin_errors
 import logging
 import os
 import importlib
 import importlib.util
 from websockets.server import WebSocketServerProtocol
+import aiohttp.web as web
 from depends import connects
 
-confi = {}  # 配置
+global_config = {}  # 配置
 
 
 class Plugin:
@@ -33,28 +34,33 @@ def __init__(self):
         self.connect_id_dict: dict[any, list] = {}  # 连接id——消息对象，为了防止重复向插件申请迭代对象
         self.plugin_cgi_support: dict = {}  # 消息插件cgi
         self.plugin_js_support: dict = {}  # js支持字典
+        self.connect_id_dict_aiohttp = {}  # 连接id——消息对象，为了防止重复向插件申请迭代对象(aiohttp)
 
-        plugin_name = ""
         # 加载默认插件目录
-        for plugin_file in os.listdir(confi['plugins']['default_path']):
+        self.__lod_init_plugin__()
+
+    def __lod_init_plugin__(self):
+        plugin_name = ""
+        for plugin_file in os.listdir(global_config['plugins']['default_path']):
             try:
                 # 排除一些非法的文件和缓存目录，还要同时保障能够加载软件包
                 if os.path.splitext(plugin_file)[1] == ".py" or os.path.isdir(
-                        os.path.join(confi['plugins']['default_path'],
+                        os.path.join(global_config['plugins']['default_path'],
                                      plugin_file)) and not plugin_file == "__pycache__":
+
                     plugin_name = os.path.splitext(plugin_file)[0]
-                    pathname = os.path.basename(confi['plugins']['default_path'])
+                    path_name = os.path.basename(global_config['plugins']['default_path'])
 
-                    mlogger.info(f"found a plugin '{plugin_name}' in {pathname}")
+                    self.logger.info(f"found a plugin '{plugin_name}' in dir {path_name}")
 
-                    plugin_module = importlib.import_module(f'{pathname}.{plugin_name}')
+                    plugin_module = importlib.import_module(f'{path_name}.{plugin_name}')
 
                     # 合法性检查
-                    if not hasattr(plugin_module, "Plugin_Main"):
+                    if not hasattr(plugin_module, "PluginMain"):
                         raise plugin_errors.NoMainMather("函数未实现主方法或者主方法名称错误")
 
-                    plugin_class = getattr(plugin_module, "Plugin_Main")
-                    plugin_interface: plugin_main.Plugin_Main = plugin_class()
+                    plugin_class = getattr(plugin_module, "PluginMain")
+                    plugin_interface: pluginmain.PluginMain = plugin_class()
 
                     # 获取插件类型
                     if plugin_interface.plugin_type() == "message":
@@ -72,7 +78,7 @@ def __init__(self):
                         self.plugin_js_support[plugin_interface.plugin_name] = plugin_interface.plugin_js_sprit
 
             except IndexError as e:
-                mlogger.error(f"failed to import plugin :\n{plugin_name} {str(e)}")
+                self.logger.error(f"failed to import plugin :\n{plugin_name} {str(e)}")
 
     async def remove_connect_in_id_dict(self, id):
         """
@@ -84,6 +90,16 @@ async def remove_connect_in_id_dict(self, id):
                 await i.callback()
         return self.connect_id_dict.pop(id, False)
 
+    async def remove_connect_in_id_dict_aiohttp(self, id):
+        """
+        当连接关闭时，移除连接
+        :param id 连接id
+        """
+        for i in self.connect_id_dict_aiohttp[id]:
+            if hasattr(i, "callback"):
+                await i.callback()
+        return self.connect_id_dict_aiohttp.pop(id, False)
+
     async def get_plugin_message(self, params, connect: WebSocketServerProtocol):
         """
         弹幕对象迭代器，迭代对应参数的弹幕
@@ -98,7 +114,7 @@ async def get_plugin_message(self, params, connect: WebSocketServerProtocol):
             # 获取未缓存的消息迭代器
             self.connect_id_dict[connect.id] = []
             for plugin in self.message_plugin_list:
-                dm = plugin.dm_iter(params, connects.connect_wrapper(connect))
+                dm = plugin.dm_iter(params)
                 if dm is None:
                     continue
                 self.connect_id_dict[connect.id].append(dm)
@@ -107,6 +123,29 @@ async def get_plugin_message(self, params, connect: WebSocketServerProtocol):
                     self.logger.debug("get a dm:", _dm)
                     yield _dm
 
+    async def get_plugin_message_aiohttp(self, params, connect: web.WebSocketResponse):
+        """
+        弹幕对象迭代器，迭代对应参数的弹幕
+        """
+        if connect in self.connect_id_dict_aiohttp.keys():
+            # 已经缓存消息迭代器
+            for dm_iter in self.connect_id_dict_aiohttp[connect]:
+                async for _dm in dm_iter:
+                    self.logger.debug("get a dm:", _dm)
+                    yield _dm
+        else:
+            # 获取未缓存的消息迭代器
+            self.connect_id_dict_aiohttp[connect] = []
+            for plugin in self.message_plugin_list:
+                dm = plugin.dm_iter(params)
+                if dm is None:
+                    continue
+                self.connect_id_dict_aiohttp[connect].append(dm)
+
+                async for _dm in dm:
+                    self.logger.debug("get a dm:", _dm)
+                    yield _dm
+
     async def message_analyzer(self, message):
         # 消息分析插件
         for plugins in self.analyzer_plugin_list:

requirements.txt

@@ -1,7 +1,7 @@
-aiohttp==3.8.6
+aiohttp==3.9.2
 aiowebsocket==1.0.0.dev2
 Brotli==1.1.0
-pycryptodome==3.19.0
+pycryptodome==3.19.1
 PyYAML==6.0.1
 Requests==2.31.0
 websockets==11.0.3