diff --git a/ahserver/processorResource.py b/ahserver/processorResource.py index f009387..9724e71 100644 --- a/ahserver/processorResource.py +++ b/ahserver/processorResource.py @@ -47,7 +47,8 @@ from .baseProcessor import PythonScriptProcessor, MarkdownProcessor from .xlsxdsProcessor import XLSXDataSourceProcessor from .llmProcessor import LlmProcessor, LlmSProcessor, LlmAProcessor -from .websocketProcessor import WebsocketProcessor, XtermProcessor +from .websocketProcessor import WebsocketProcessor +from .xtermProcessor import XtermProcessor from .sqldsProcessor import SQLDataSourceProcessor from .functionProcessor import FunctionProcessor from .proxyProcessor import ProxyProcessor diff --git a/ahserver/websocketProcessor.py b/ahserver/websocketProcessor.py index ea30ff4..d64d8e7 100644 --- a/ahserver/websocketProcessor.py +++ b/ahserver/websocketProcessor.py @@ -11,91 +11,6 @@ from appPublic.dictObject import DictObject from appPublic.log import info, debug, warning, error, exception, critical from .baseProcessor import BaseProcessor, PythonScriptProcessor -class XtermProcessor(PythonScriptProcessor): - @classmethod - def isMe(self,name): - return name=='xterm' - - async def ws_2_process(self, ws): - async for msg in ws: - if msg.type == aiohttp.WSMsgType.TEXT: - resize_pattern = '_#_resize_#_' - heartbeat_pattern = '_#_heartbeat_#_' - if msg.data.startswith(resize_pattern): - row, col = [ int(i) for i in msg.data[len(resize_pattern):].split(',')] - await self.p_obj.set_terminal_size(row, col) - continue - if msg.data == heartbeat_pattern: - await ws_send(ws, heartbeat_pattern) - continue - self.p_obj.stdin.write(msg.data) - elif msg.type == aiohttp.WSMsgType.ERROR: - # print('ws connection closed with exception %s' % ws.exception()) - return - - async def process_2_ws(self, ws): - while self.running: - x = await self.p_obj.stdout.read(1024) - await self.ws_sendstr(ws, x) - - async def datahandle(self,request): - await self.path_call(request) - - async def path_call(self, request, params={}): - # - # xterm file is a python script as dspy file - # it must return a DictObject with sshnode information - # parameters: nodeid - # - await self.set_run_env(request, params=params) - login_info = await super().path_call(request, params=params) - if login_info is None: - raise Exception('data error') - - debug(f'{login_info=}') - ws = web.WebSocketResponse() - await ws.prepare(request) - await self.create_process(login_info) - r1 = self.ws_2_process(ws) - r2 = self.process_2_ws(ws) - await asyncio.gather(r1,r2) - self.retResponse = ws - return ws - - async def create_process(self, login_info): - # id = lenv['params_kw'].get('termid') - host = login_info['host'] - port = login_info.get('port', 22) - commandline = login_info.get('commandline', 'bash') - username = login_info.get('username', 'root') - password = login_info.get('password',None) - client_key = login_info.get('client_key', None) - passphrase = login_info.get('passphrase', None) - jumpers = login_info.get('jumpers', []) - self.sshnode = SSHNode(host, username=username, - password=password, - port=port, - client_keys=[] if client_key is None else [client_key], - passphrase=passphrase, - jumpers=jumpers) - await self.sshnode.connect() - self.p_obj = await self.sshnode._process(commandline, - term_type='xterm-256color', - term_size=(80, 24), - encoding='utf-8') - self.running = True - - async def ws_sendstr(self, ws:web.WebSocketResponse, s:str): - data = { - "type":1, - "data":s - } - await ws.send_str(json.dumps(data, indent=4, ensure_ascii=False)) - - def close_process(self): - self.sshnode.close() - self.p_obj.close() - async def ws_send(ws:web.WebSocketResponse, data): info(f'data={data} {ws=}') d = { diff --git a/ahserver/xtermProcessor.py b/ahserver/xtermProcessor.py new file mode 100644 index 0000000..9962690 --- /dev/null +++ b/ahserver/xtermProcessor.py @@ -0,0 +1,85 @@ +import asyncio +import aiohttp +import aiofiles +import json +import codecs +from aiohttp import web +import aiohttp_cors +from traceback import print_exc +from appPublic.sshx import SSHServer +from appPublic.dictObject import DictObject +from appPublic.log import info, debug, warning, error, exception, critical +from .baseProcessor import BaseProcessor, PythonScriptProcessor + +class XtermProcessor(PythonScriptProcessor): + @classmethod + def isMe(self,name): + return name=='xterm' + + async def ws_2_process(self, ws): + async for msg in ws: + if msg.type == aiohttp.WSMsgType.TEXT: + resize_pattern = '_#_resize_#_' + heartbeat_pattern = '_#_heartbeat_#_' + if msg.data.startswith(resize_pattern): + row, col = [ int(i) for i in msg.data[len(resize_pattern):].split(',')] + await self.p_obj.set_terminal_size(row, col) + continue + if msg.data == heartbeat_pattern: + await ws_send(ws, heartbeat_pattern) + continue + self.p_obj.stdin.write(msg.data) + elif msg.type == aiohttp.WSMsgType.ERROR: + # print('ws connection closed with exception %s' % ws.exception()) + return + + async def process_2_ws(self, ws): + while self.running: + x = await self.p_obj.stdout.read(1024) + await self.ws_sendstr(ws, x) + + async def datahandle(self,request): + await self.path_call(request) + + async def path_call(self, request, params={}): + # + # xterm file is a python script as dspy file + # it must return a DictObject with sshnode information + # parameters: nodeid + # + await self.set_run_env(request, params=params) + login_info = await super().path_call(request, params=params) + if login_info is None: + raise Exception('data error') + + debug(f'{login_info=}') + ws = web.WebSocketResponse() + await ws.prepare(request) + await self.run_xterm(ws, login_info) + self.retResponse = ws + return ws + + async def run_xterm(self, login_info): + # id = lenv['params_kw'].get('termid') + self.sshnode = SSHServer(login_info) + async with self.sshnode.get_connector() as conn: + self.p_obj = await conn.create_process(term_type='xterm', term_size=(24, 80)) + stdin_task = asyncio.create_task(self.ws_2_process(ws)) + try: + while self.running: + x = await self.p_obj.stdout.read(1024) + await self.ws_sendstr(ws, x) + except (asyncio.CancelledError, EOFError): + pass + finally: + self.p_obj.close() + stdin_task.cancel() + + async def ws_sendstr(self, ws:web.WebSocketResponse, s:str): + data = { + "type":1, + "data":s + } + await ws.send_str(json.dumps(data, indent=4, ensure_ascii=False)) + +