From 7777fb73cda90d9318148d665de6ec75e984b11d Mon Sep 17 00:00:00 2001 From: 1-6 Date: Thu, 13 Jul 2023 20:27:10 +0800 Subject: [PATCH] x --- adapters/pagermaid/main.py | 144 ++++++++++++++++++++++++++++++++++++ core/carry.go | 12 +-- core/function.go | 8 +- core/plugin_utils.go | 17 +++++ core/plugin_web_response.go | 6 +- core/websocket.go | 28 ++++--- 6 files changed, 195 insertions(+), 20 deletions(-) create mode 100644 adapters/pagermaid/main.py diff --git a/adapters/pagermaid/main.py b/adapters/pagermaid/main.py new file mode 100644 index 0000000..edfe0b5 --- /dev/null +++ b/adapters/pagermaid/main.py @@ -0,0 +1,144 @@ +import contextlib +import json + +from asyncio import sleep + +from pagermaid import log +from pagermaid.hook import Hook +from pagermaid.listener import listener +from pagermaid.enums import Message +from pagermaid.services import bot +from pagermaid.single_utils import sqlite +from pagermaid.utils import pip_install + +pip_install("aiohttp") + +import aiohttp + +uri="" + +class WebSocket: + def __init__(self): + self.uri = uri + self.loop = bot.loop + self.client = aiohttp.ClientSession(loop=self.loop) + self.need_stop = False + self.ws = None + self.connection = None + + @staticmethod + def database_have_uri(): + return uri != "" + + def restore_uri(self): + self.uri = uri + + async def set_uri(self, uri): + await self.disconnect() + self.uri = uri + sqlite["websocket_uri"] = uri + + def is_connected(self): + return self.connection is not None + + async def connect(self): + if self.is_connected(): + await self.disconnect() + if self.uri: + self.ws = self.client.ws_connect( + self.uri+"&user_id="+str(bot.me.id), autoclose=False, autoping=False, timeout=5 + ) + self.connection = await self.ws._coro + + async def disconnect(self): + if self.connection: + with contextlib.suppress(Exception): + await self.connection.close() + self.ws = None + self.connection = None + + + async def keep_alive(self): + while True: + try: + if not self.uri: + await sleep(1) + continue + await self.connect() + except Exception: + await sleep(1) + continue + await self.get() + await sleep(1) + if self.need_stop: + await self.disconnect() + self.need_stop = False + break + + async def get(self): + ws_ = self.connection + if not ws_: + return + while True: + msg = await ws_.receive() + if msg.type == aiohttp.WSMsgType.TEXT: + bot.loop.create_task(self.process_message(msg.data)) + elif msg.type == aiohttp.WSMsgType.PING: + await ws_.pong() + elif msg.type == aiohttp.WSMsgType.BINARY: + pass + elif msg.type != aiohttp.WSMsgType.PONG: + if msg.type == aiohttp.WSMsgType.CLOSE: + await ws_.close() + elif msg.type == aiohttp.WSMsgType.ERROR: + print(f"Error during receive {ws_.exception()}") + break + self.ws = None + self.connection = None + + async def push(self, msg): + if self.is_connected(): + await self.connection.send_str(msg) + + @staticmethod + async def process_message(text: str): + try: + data = json.loads(text) + except Exception: + return + echo = data.get("echo", "") + action = data.get("action", None) + action_data = data.get("data", None) + bot_action = getattr(bot, action) + if bot_action and action_data: + message = await bot_action(**action_data) + message = str(message.__str__()) + message = message.replace("{", '{"echo": "'+str(echo)+'",', 1) + await ws.push(message) + + +ws = WebSocket() + + +@Hook.on_startup() +async def connect_ws(): + try: + # await ws.connect() + bot.loop.create_task(ws.keep_alive()) + except Exception as e: + await log(f"[ws] Connection failed: {e}") + + +@listener(incoming=True, outgoing=True) +async def websocket_push(message: Message): + with contextlib.suppress(Exception): + await ws.push(message.__str__()) + + +@listener(command="sillyplus", description="sillyplus Connect") +async def websocket_to_connect(message: Message): + if ws.is_connected(): + return await message.edit("傻+ 已连接") + else: + return await message.edit("傻+ 已离线") + diff --git a/core/carry.go b/core/carry.go index 9e337d0..4f61f66 100644 --- a/core/carry.go +++ b/core/carry.go @@ -298,8 +298,8 @@ func initCarry() { } if ncg.In { if ncg.Enable { - AddListenOnGroup(ncg.ID, fmt.Sprintf("已为采集群(%s)开启监听模式", name)) - AddNoReplyGroups(ncg.ID, fmt.Sprintf("已为采集群(%s)开启禁言模式", name)) + AddListenOnGroup(ncg.ID, fmt.Sprintf("已为采集群(%s)开启监听模式", name), ncg.Platform) + AddNoReplyGroups(ncg.ID, fmt.Sprintf("已为采集群(%s)开启禁言模式", name), ncg.Platform) } else { RemListenOnGroup(ncg.ID, fmt.Sprintf("已为采集群(%s)关闭监听模式", name)) } @@ -321,8 +321,8 @@ func initCarry() { if name == "" { name = ncg.ID } - AddListenOnGroup(ncg.ID, fmt.Sprintf("已为采集群(%s)开启监听模式", name)) - AddNoReplyGroups(ncg.ID, fmt.Sprintf("已为采集群(%s)开启禁言模式", name)) + AddListenOnGroup(ncg.ID, fmt.Sprintf("已为采集群(%s)开启监听模式", name), ncg.Platform) + AddNoReplyGroups(ncg.ID, fmt.Sprintf("已为采集群(%s)开启禁言模式", name), ncg.Platform) } } else { return nil @@ -349,8 +349,8 @@ func setCgs() { if name == "" { name = cg.ID } - AddListenOnGroup(cg.ID, fmt.Sprintf("已为采集群(%s)开启监听模式", name)) - AddNoReplyGroups(cg.ID, fmt.Sprintf("已为采集群(%s)开启禁言模式", name)) + AddListenOnGroup(cg.ID, fmt.Sprintf("已为采集群(%s)开启监听模式", name), cg.Platform) + AddNoReplyGroups(cg.ID, fmt.Sprintf("已为采集群(%s)开启禁言模式", name), cg.Platform) } cgs = append(cgs, cg) return nil diff --git a/core/function.go b/core/function.go index 2fb392d..6db669d 100644 --- a/core/function.go +++ b/core/function.go @@ -39,15 +39,15 @@ type GroupInfo struct { Enable bool `json:"enable"` } -var AddNoReplyGroups = func(code string, desc string) { - _, loaded := NoReplyGroups.LoadOrStore(code, true) +var AddNoReplyGroups = func(code string, desc string, plt string) { + _, loaded := NoReplyGroups.LoadOrStore(code, plt) if !loaded { logs.Info(desc) } } -var AddListenOnGroup = func(code string, desc string) { - _, loaded := ListenOnGroups.LoadOrStore(code, true) +var AddListenOnGroup = func(code string, desc string, plt string) { + _, loaded := ListenOnGroups.LoadOrStore(code, plt) if !loaded { logs.Info(desc) } diff --git a/core/plugin_utils.go b/core/plugin_utils.go index db1489d..d687996 100644 --- a/core/plugin_utils.go +++ b/core/plugin_utils.go @@ -365,6 +365,23 @@ func SetPluginMethod(vm *goja.Runtime, uuid string, on_start bool, running func( vm: vm, }) vm.Set("createNickName", CreateNickName) + vm.Set("getListenOnGroups", func(plt string) []string { + groups := []string{} + ListenOnGroups.Range(func(key, value any) bool { + if value == plt { + groups = append(groups, key.(string)) + } + return true + }) + StaticListenOnGroups.Range(func(key, value any) bool { + if value == plt { + groups = append(groups, key.(string)) + } + return true + }) + groups = utils.Unique(groups) + return groups + }) } func EncryptPlugin(script string) string { diff --git a/core/plugin_web_response.go b/core/plugin_web_response.go index 051a0dc..e12dbbc 100644 --- a/core/plugin_web_response.go +++ b/core/plugin_web_response.go @@ -27,6 +27,7 @@ type Response struct { status int isRedirect bool conn *WsConn + vm *goja.Runtime } func (r *Response) Send(gv goja.Value) *Response { @@ -81,7 +82,10 @@ func (r *Response) Json(ps ...interface{}) interface{} { r.content += fmt.Sprint(data) } if r.conn != nil { - _, result := r.conn.WriteMessage(1, d, pattern) + err, result := r.conn.WriteMessage(1, d, pattern) + if err != nil { + panic(Error(r.vm, err)) + } r.content = "" return result } diff --git a/core/websocket.go b/core/websocket.go index 025548b..3fdf879 100644 --- a/core/websocket.go +++ b/core/websocket.go @@ -50,7 +50,7 @@ func (wc *WsConn) WriteMessage(messageType int, data []byte, pattern map[string] if v, ok := pattern["$timeout"]; ok { timeout = utils.Int(v) delete(pattern, "$timeout") - } // + } wp.Value = pattern key := atomic.AddInt64(&wc.Key, 1) wp.Chan = make(chan map[string]interface{}, 1) @@ -94,9 +94,7 @@ func handleWebsocket(c *gin.Context) { c: c, // uuid: uuid, } - var res = &Response{ - c: c, - } + var upGrader = websocket.Upgrader{ CheckOrigin: func(r *http.Request) bool { return true @@ -108,16 +106,20 @@ func handleWebsocket(c *gin.Context) { return } wc := &WsConn{} - res.conn = wc + req._event = "connect" wc.conn = ws go function.Handle(&Faker{ Type: "websocket", }, func(vm *goja.Runtime) { - vm.Set("res", res) + vm.Set("res", &Response{ + c: c, + conn: wc, + vm: vm, + }) vm.Set("req", req) }) - time.Sleep(time.Millisecond*500) + time.Sleep(time.Millisecond * 500) for { _, data, err := ws.ReadMessage() wc.patterns.Range(func(key, value any) bool { @@ -174,7 +176,11 @@ func handleWebsocket(c *gin.Context) { function.Handle(&Faker{ Type: "websocket", }, func(vm *goja.Runtime) { - vm.Set("res", res) + vm.Set("res", &Response{ + c: c, + conn: wc, + vm: vm, + }) vm.Set("req", req) }) ws.Close() @@ -190,7 +196,11 @@ func handleWebsocket(c *gin.Context) { go function.Handle(&Faker{ Type: "websocket", }, func(vm *goja.Runtime) { - vm.Set("res", res) + vm.Set("res", &Response{ + c: c, + conn: wc, + vm: vm, + }) vm.Set("req", req) }) }