This commit is contained in:
yumoqing 2025-07-15 17:29:39 +08:00
parent e4d1b86e5e
commit c7c9a3e446
3 changed files with 87 additions and 86 deletions

View File

@ -47,7 +47,8 @@ from .baseProcessor import PythonScriptProcessor, MarkdownProcessor
from .xlsxdsProcessor import XLSXDataSourceProcessor from .xlsxdsProcessor import XLSXDataSourceProcessor
from .llmProcessor import LlmProcessor, LlmSProcessor, LlmAProcessor from .llmProcessor import LlmProcessor, LlmSProcessor, LlmAProcessor
from .websocketProcessor import WebsocketProcessor, XtermProcessor from .websocketProcessor import WebsocketProcessor
from .xtermProcessor import XtermProcessor
from .sqldsProcessor import SQLDataSourceProcessor from .sqldsProcessor import SQLDataSourceProcessor
from .functionProcessor import FunctionProcessor from .functionProcessor import FunctionProcessor
from .proxyProcessor import ProxyProcessor from .proxyProcessor import ProxyProcessor

View File

@ -11,91 +11,6 @@ from appPublic.dictObject import DictObject
from appPublic.log import info, debug, warning, error, exception, critical from appPublic.log import info, debug, warning, error, exception, critical
from .baseProcessor import BaseProcessor, PythonScriptProcessor 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): async def ws_send(ws:web.WebSocketResponse, data):
info(f'data={data} {ws=}') info(f'data={data} {ws=}')
d = { d = {

View File

@ -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))