CakeCN/hsa
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

code-server-liek

Browse Source
This commit is contained in:
Cai
2025-09-17 07:46:38 +00:00
parent 50a570eccf
commit 025c8c1a20
12 changed files with 797 additions and 1511 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1694
Html/.log
View File

File diff suppressed because it is too large Load Diff

6
Html/apps/services/backboard_service.py
View File

@@ -34,7 +34,7 @@ def realtime_response(config: dict, realtime_action: dict) -> None:
file_path = realtime_action.get("filePath")
assert isinstance(file_path, str)
with sudo_open(file_path, "r", encoding="utf-8") as f:
bb.active_file_content = f.read()
bb.active_file_content = f
bb.active_file_path = file_path
elif rtype == "paste":
@@ -44,14 +44,14 @@ def realtime_response(config: dict, realtime_action: dict) -> None:
bb.pasted_content = realtime_action.get("content")
bb.active_file_path = file_path
with sudo_open(file_path, "r", encoding="utf-8") as f:
bb.active_file_content = f.read()
bb.active_file_content = f
elif rtype == "fileEdit":
file_path = realtime_action.get("filePath")
assert isinstance(file_path, str)
bb.active_file_path = file_path
with sudo_open(file_path, "r", encoding="utf-8") as f:
bb.active_file_content = f.read()
bb.active_file_content = f
chatmanager.upload_bb()

121
Html/apps/static/算法效率-效率的重要性与实践验证.html Normal file
View File

@@ -0,0 +1,121 @@
<html>
<head>
<style>
img {
max-width: 100%;
height: auto;
}
</style>
<link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/github-markdown-css/5.2.0/github-markdown.min.css">
</head>
<body class="markdown-body">
<h1>效率的重要性与实践验证</h1>
<blockquote>
<p>Auto-generated at 2025-09-16 05:03</p>
</blockquote>
<h3>理论热身：算法选择的智慧</h3>
<h4>任务：分析与决策</h4>
<p>项目组目前有两套备选的交通信号灯同步算法，它们将在不同性能的服务器上运行：</p>
<ul>
<li>
<p><strong>方案A</strong>：部署在超级服务器上（10^9 次运算/秒），采用的是一种较为简单的算法，处理n个路口需要 $2n^2$ 次计算。</p>
</li>
<li>
<p><strong>方案B</strong>：部署在普通服务器上（10^7 次运算/秒），但采用的是一种更优化的算法，处理n个路口需要 $50nlog_2n$ 次计算。</p>
</li>
</ul>
<p>你的项目经理（右侧的Agent）希望你通过分析，来判断哪个方案更具前景。请与TA对话，逐一回答以下问题。</p>
<h4>思考题</h4>
<p>与右侧的Agent对话，回答以下问题：</p>
<ol>
<li>
<p><strong>概念回顾</strong>：首先，请向你的“经理”解释，根据课程所学，一个合格的“算法”应具备哪些基本特征？ </p>
</li>
<li>
<p><strong>小规模测试</strong>：对于一个包含100个路口的小型城区（n=100），计算并说明方案A和方案B分别需要多长时间完成计算？在这种情况下，你会推荐哪个方案？</p>
</li>
<li>
<p><strong>大规模应用</strong>：现在，我们需要为一座拥有100万个路口的大都市（n=1,000,000）进行规划。再次计算并说明两个方案的耗时。你的推荐会改变吗？为什么？</p>
</li>
<li>
<p><strong>总结陈词</strong>：综合以上分析，向你的“经理”总结：为什么一个更优的算法设计，其重要性远超硬件性能的提升？ 这验证了课程中提到的哪个核心观点？</p>
</li>
</ol>
<h3>编程实践：验证算法的真实性能</h3>
<h4>任务：编码与分析</h4>
<p>理论分析让你认识到了算法效率的重要性。现在，你需要通过编程来亲身感受不同交通状况对同一算法性能的巨大影响。我们将以“插入排序”为例，来处理三种典型的交通流量数据，这分别对应算法分析中的<strong>最好</strong>，<strong>最坏</strong>和<strong>平均</strong>情况。</p>
<h5>题目：模拟交通流量排序</h5>
<p>实现插入排序算法，并验证其在处理“畅通无阻”（数据有序）、“交通大堵塞”（数据逆序）和“随机车流”（数据随机）三种模式时的效率差异。</p>
<h5>代码框架</h5>
<p>在代码编辑区，完成 <code>insertion_sort(arr)</code> 函数的实现后，运行完整代码，并与Agent讨论结果。
<strong>请创建任意文件，将下面代码写入到编辑器中</strong>
```python
import random
import time</p>
<p>def insertion_sort(arr):
"""
实现插入排序算法
参数arr: 待排序的交通数据数组（整数代表车辆通行次序）
返回: 排序后的数组
"""
# TODO: 请在此处实现你的插入排序逻辑</p>
<p>def generate_traffic_data(n):
"""
生成模拟交通数据
参数n: 数据规模（路口数量或监控点数量）
返回: 三种不同交通状况的数据
"""
random_data = [random.randint(0, 10**6) for _ in range(n)]
# 模拟“畅通无阻”：交通流按次序进入，数据基本有序 (Best Case)
best_case_data = sorted(random_data)
# 模拟“交通大堵塞”：疏散时情况完全反转，数据逆序 (Worst Case)
worst_case_data = sorted(random_data, reverse=True)
# 模拟“随机车流”：正常但无规律的交通状况 (Average Case)
average_case_data = random_data
return best_case_data, worst_case_data, average_case_data</p>
<p>def measure_performance(func, data):
"""
测量算法性能
参数func: 排序函数
参数data: 交通数据
返回: 执行时间(毫秒)
"""
start_time = time.perf_counter_ns()
func(data.copy()) # 使用副本避免影响其他测试
end_time = time.perf_counter_ns()
return (end_time - start_time) / 10**6 # 转换为毫秒</p>
<h1>测试不同规模的路口网络</h1>
<p>network_sizes = [1000, 5000, 10000]
print("交通数据处理算法性能测试:")
for size in network_sizes:
best, worst, avg = generate_traffic_data(size)</p>
<pre><code>time_best = measure_performance(insertion_sort, best)
time_worst = measure_performance(insertion_sort, worst)
time_avg = measure_performance(insertion_sort, avg)
print(f"网络规模 n={size}:")
print(f" - 畅通无阻 (Best Case): {time_best:.2f} ms")
print(f" - 交通大堵塞 (Worst Case): {time_worst:.2f} ms")
print(f" - 随机车流 (Average Case): {time_avg:.2f} ms")
</code></pre>
<p>```</p>
<h4>分析与讨论</h4>
<p>完成编程并得到输出后，请与右侧Agent讨论以下问题，以检验你的理解：</p>
<ol>
<li>
<p><strong>结果分析</strong>：当网络规模从1000增加到10000时，“交通大堵塞”（最坏情况）的处理时间增长了大约多少倍？这更符合O(n)（线性）还是O(n2)（二次）的增长模式？</p>
</li>
<li>
<p><strong>原因探究</strong>：为什么“畅通无阻”（最好情况）的处理速度如此之快？它的时间复杂度是什么？请结合你的代码逻辑来解释。</p>
</li>
<li>
<p><strong>实践应用</strong>：根据你的实验结果，你认为插入排序是否适合用于需要快速响应大规模交通拥堵的实时预警系统？为什么？</p>
</li>
<li>
<p><strong>融会贯通</strong>：结合第一关的理论分析和第二关的编程实验，你对“算法是解决问题的核心”这句话有了怎样更深的理解？</p>
</li>
</ol>
</body>
</html>

2
Html/apps/utils/sudoUtil.py
View File

@@ -16,7 +16,7 @@ def sudo_open(file_path, mode='r', encoding='utf-8'):
if result.returncode != 0:
raise PermissionError(f"无法读取文件: {file_path}")
content = result.stdout
yield content
return content
elif mode == 'w':
raise PermissionError(f"不允许以sudo权限写入文件: {file_path}")
else:

25
Html/apps/views/vscode.py
View File

@@ -33,6 +33,29 @@ def desktop(user_uuid, course_id, chapter_name, lesson_name):
except subprocess.CalledProcessError as e:
print(f"Error creating directory {path_dir} details {str(e)}")
try:#使用sudo创建shared_group
subprocess.run(["sudo", "groupadd", f"shared_group_{user_id}"], check=True)
except subprocess.CalledProcessError as e:
print(f"Error creating shared_group: {e}")
try:#使用sudo将user_id加入shared_group
subprocess.run(["sudo", "usermod", "-a", "-G", f"shared_group_{user_id}", user_id], check=True)
# 自己也加入
subprocess.run(["sudo", "usermod", "-a", "-G", f"shared_group_{user_id}", 'flask'], check=True)
except subprocess.CalledProcessError as e:
print(f"Error adding user {user_id} to shared_group: {e}")
try:#将文件
subprocess.run(["sudo", "chown", "-R", f"flask:shared_group_{user_id}", user_home], check=True)
subprocess.run(["sudo", "chmod", "-R", "770", path_dir], check=True)
except subprocess.CalledProcessError as e:
print(f"Error changing directory {path_dir} to shared_group_{user_id}: {e}")
try:#继承目录
subprocess.run(["sudo", "chmod", "g+s", path_dir], check=True)
except subprocess.CalledProcessError as e:
print(f"Error changing directory {path_dir} to shared_group_{user_id}: {e}")
chatmanager = ChatManager()
current_app.extensions["user_uuid2chatmanager"][user_uuid] = chatmanager
code_server_port = 10000 + int(uuid.uuid4().int % 10000)
@@ -52,6 +75,8 @@ def desktop(user_uuid, course_id, chapter_name, lesson_name):
r'handle_path\s+/vsc' + re.escape(user_id) + r'/\* {.*?reverse_proxy.*?}',
re.DOTALL
)
for i in range(len(lines)):
if (lines[i].strip()==''): lines[i]=''
content = "".join(lines)
if pattern.search(content):
new_content = pattern.sub('', content)

282
code-server-like/app/sockets/terminal_service.py
View File

@@ -5,154 +5,166 @@ import uuid
import subprocess
import threading
from flask import Blueprint, request, current_app, session
from flask_socketio import Namespace, join_room, leave_room, emit, SocketIO
from flask_socketio import Namespace, join_room, leave_room, emit, SocketIO, rooms, disconnect
from threading import Thread
import sys
from ..extensions import socketio
import pty
import termios
import struct
import fcntl
import psutil
TERM_INIT_CONFIG = {
# instead of local server runnning this web terminal service
# "domain" is the target that you want to access through local server (with this web terminal)
# and before doing so - make sure you have username and port (on the "domain") to implement remote access
'domain': '127.0.0.1', # or ip address like 192.168.10.11
'client_path': {
'telnet': '/usr/bin/telnet', # confirmed location of your client binary (with cmd like 'which telnet')
'ssh': '/usr/bin/ssh'
}
}
terminal_bp = Blueprint('terminal', __name__, url_prefix='/vsc-like')
# 存储终端进程
command_dict = {}
terminal_busy = {}
terminals = {}
terminal_locks = {}
def check_process_running(command_name):
try:
# 使用 pgrep 查找与给定命令相关的进程
result = subprocess.run(
["pgrep", "-f", command_name],
stdout=subprocess.PIPE, stderr=subprocess.PIPE
)
return result.returncode == 0 # 如果返回值为 0，说明命令仍在运行
except Exception as e:
print(f"Error checking process: {e}")
return False
def monitor_process(socket_id, command_name, namespace):
while True:
is_running = check_process_running(command_name)
if is_running:
print(f"Process '{command_name}' is still running.")
else:
print(f"Process '{command_name}' is not running.")
break # 停止检查，如果进程不再运行
time.sleep(0.1) # 每 0.1 秒检查一次
command_dict[socket_id] = ''
terminal_busy[socket_id] = False
namespace.emit('terminal_output',{'data': '>'})
def start_terminal_process(socket_id, namespace):
def set_winsize(fd, row, col, xpix=0, ypix=0):
winsize = struct.pack("HHHH", row, col, xpix, ypix)
fcntl.ioctl(fd, termios.TIOCSWINSZ, winsize)
def read_and_forward_pty_output(fd=None, pid=None, room_id=None,namespace=None):
"""
监听多个子进程的输出，并通过 WebSocket 发给前端
使用 select 进行非阻塞的 I/O 操作，并实现动态流量控制
read data on pty master from the pty slave, and emit to the web terminal visitor
"""
process = terminals[socket_id]
# 获取子进程的 stdout 和 stderr
stdout = process.stdout
stdin = process.stdin
# 将 stdout 和 stderr 添加到 select 监听的文件描述符列表中
# 初始等待时间
timeout = 0.1
max_timeout = 2 # 设置最大超时时间为 2 秒
max_read_bytes = 1024 * 20
timeout=0.1
while True:
while True:
output = stdout.readline()
if not output:
break
timeout = 0.1
namespace.emit('terminal_output', {'data': output+'\r'}, room=socket_id)
timeout = min(timeout * 2, max_timeout)
time.sleep(timeout)
if process.poll() is not None:
break
socketio.sleep(timeout)
timeout=min(timeout*2, 0.4)
# using flask default web server, or uwsgi production web server
# when the child process is terminated, it will not disappear from linux process list
# and keep staying as a zombie process until the parent exits.
try:
child_process = psutil.Process(pid)
except psutil.NoSuchProcess as err:
return
if child_process.status() not in ('running', 'sleeping'):
return
if fd:
timeout_sec = 0
(data_ready, _, _) = select.select([fd], [], [], timeout_sec)
if data_ready:
# output = os.read(fd, max_read_bytes).decode('ascii')
timeout=0.1
try:
output = os.read(fd, max_read_bytes).decode()
except Exception as err:
output = """
***AQUI WEB TERM ERR***
{}
***********************
""".format(err)
# the key for different visitor to get different terminal (instead of mixing up)
# is to let the background task push pty response to each one's own (default) ROOM!
namespace.emit("pty_output", {"output": output}, room=room_id)
class VSCLikeNameSpace(Namespace):
def on_connect(self, data):
"""
处理 WebSocket 连接
- 验证会话
- 使用指定的用户启动 bash 终端
- 启动监听子进程输出的线程
"""
socket_id = session.get('uuid')
user_id = session.get('user_id') # 从 session 获取用户 ID
path = session.get('path')
def on_connect(self):
"""new client connected"""
if session.get('terminal_config', {}).get('child_pid', None):
print(session['terminal_config']['child_pid'])
# already started child process, don't start another
return
if not path or not user_id:
emit('error', {'message': 'Authentication failed, no path or user found for this session'})
return False # 拒绝连接
# 使用 sudo 启动终端进程，以指定的 user_id 用户身份执行
try:
process = subprocess.Popen(
['sudo', '-u', user_id, 'bash', '-il'], # 使用指定的 user_id 启动 bash
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True, # 确保输出是文本格式
)
terminals[socket_id] = process
terminal_busy[socket_id] = False
emit('connected', {'message': 'OK'})
join_room(socket_id, namespace='/vsc-like')
# create child process attached to a pty we can read from and write to
(child_pid, fd) = pty.fork()
lesson_path = session.get('path')
if child_pid == 0:
# this is the child process fork.
# anything printed here will show up in the pty, including the output
# of this subprocess
# subprocess.run('bash')
term_type = session.get('terminal_config').get('term_type')
path = TERM_INIT_CONFIG.get('client_path', {}).get(term_type, None)
if not path:
print("Can't locate {} binary, exit".format(term_type))
disconnect()
if term_type == 'telnet':
# switch to the right location of your telnet binary (example comes from OSX which got telnet from brew)
# or you can also make work like auto-detection, or manually but configurable
os.execl(path, 'telnet', '-l', session['terminal_config']['username'],
session['terminal_config']['domain'], '{}'.format(session['terminal_config']['port']))
elif term_type == 'ssh':
os.execl(path,'ssh', '-p','22',
#'{}'.format(session['terminal_config']['port']),
'{}@{}'.format(session['terminal_config']['username'], session['terminal_config']['domain']))
socketio.start_background_task(target=start_terminal_process, socket_id=socket_id, namespace=self)
except Exception as e:
emit('error', {'message': f'Error starting terminal process: {str(e)}'})
return False
def on_send_input(self, data):
socket_id = session.get('uuid')
input_data = data.get('input')
if socket_id not in command_dict:command_dict[socket_id]=''
# 检查当前 socket_id 是否有对应的终端
if socket_id in terminals:
# 获取该终端的锁进行线程同步
lock = terminal_locks.get(socket_id)
if not lock:
lock = threading.Lock()
terminal_locks[socket_id] = lock
with lock: # 在锁的保护下操作终端
try:
terminals[socket_id].stdin.write(input_data)
command_dict[socket_id]+=input_data
if input_data=='\r':
terminals[socket_id].stdin.flush()
emit('terminal_output', {'data': f'\r\n'}, room=socket_id)
if terminal_busy[socket_id]==False:
terminal_busy[socket_id] = True
socketio.start_background_task(monitor_process, socket_id, command_dict[socket_id], namespace=self)
elif input_data=='\x03':
terminals[socket_id].stdin.flush()
emit('terminal_output', {'data': f'^C\x03\r\n'}, room=socket_id)
else:
emit('terminal_output', {'data': input_data}, room=socket_id)
except Exception as e:
emit('error', {'message': f'Error sending input: {str(e)}'}, room=socket_id)
else:
current_app.logger.debug("wrong term type {}".format(term_type))
disconnect()
session['terminal_config'] = TERM_INIT_CONFIG
else:
emit('error', {'message': 'No terminal found for this session'}, room=socket_id)
session['terminal_config']['fd'] = fd
session['terminal_config']['child_pid'] = child_pid
session['terminal_config']['room_id'] = rooms()[0]
session.modified = True
set_winsize(fd, 50, 50)
current_app.logger.debug("child pid = {}".format(child_pid))
current_app.logger.debug("rooms of this session = {}".format(rooms()))
socketio.start_background_task(read_and_forward_pty_output, fd, child_pid, rooms()[0],self)
current_app.logger.debug("background task running")
# print(session)
def on_disconnect():
socket_id = session.get('uuid')
# 检查是否有终端进程需要终止
if socket_id in terminals:
try:
# 终止终端进程
terminals[socket_id].terminate()
del terminals[socket_id]
# 删除锁
if socket_id in terminal_locks:
del terminal_locks[socket_id]
emit('terminal_output', {'data': 'Terminal disconnected and terminated'}, room=socket_id)
except Exception as e:
emit('error', {'message': f'Error disconnecting terminal: {str(e)}'}, room=socket_id)
def on_pty_input(self, data):
"""write to the child pty, which now is the ssh process from this machine to the 'domain' configured
"""
print(f"get data {data}")
try:
child_process = psutil.Process(session.get('terminal_config').get('child_pid'))
except psutil.NoSuchProcess as err:
disconnect()
session['terminal_config'] = TERM_INIT_CONFIG
return
if child_process.status() not in ('running', 'sleeping'):
disconnect()
session['terminal_config'] = TERM_INIT_CONFIG
return
# print(session)
# print(data, 'from input')
fd = session.get('terminal_config').get('fd')
if fd:
# print("writing to ptd: %s" % data["input"])
# os.write(fd, data["input"].encode('ascii'))
os.write(fd, data["input"].encode())
def on_resize(self, data):
try:
child_process = psutil.Process(session.get('terminal_config').get('child_pid'))
except psutil.NoSuchProcess as err:
disconnect()
session['terminal_config'] = TERM_INIT_CONFIG
return
if child_process.status() not in ('running', 'sleeping'):
disconnect()
session['terminal_config'] = TERM_INIT_CONFIG
return
fd = session.get('terminal_config').get('fd')
if fd:
set_winsize(fd, data["rows"], data["cols"])
def on_disconnect(self):
try:
child_process = psutil.Process(session.get('terminal_config', {}).get('child_pid'))
except psutil.NoSuchProcess as err:
disconnect()
session['terminal_config'] = TERM_INIT_CONFIG
return
if child_process.status() in ('running', 'sleeping'):
# if visitor just close the browser tab then left alone the pty here
# it should be terminated by the parent process after
child_process.terminate()
current_app.logger.debug('user left the pty alone, terminated')
current_app.logger.debug('Client disconnected')

24
code-server-like/app/views/file.py
View File

@@ -12,8 +12,7 @@ def create_folder():
filepath = request.json.get('path', '')
path = session.get('path')
parent = request.json.get('parent', '')
root_workspace_path = current_app.config['ROOT_WORKSPACE_PATH']
base_path = f'{root_workspace_path}/{path}'
base_path = f'{path}'
filepath = f'{base_path}/{parent}/{filepath}'
try:
os.makedirs(filepath)
@@ -27,8 +26,7 @@ def create_file():
filepath = request.json.get('path', '')
path = session.get('path')
parent = request.json.get('parent', '')
root_workspace_path = current_app.config['ROOT_WORKSPACE_PATH']
base_path = f'{root_workspace_path}/{path}'
base_path = f'{path}'
filepath = f'{base_path}/{parent}/{filepath}'
try:
open(filepath, 'w').close()
@@ -41,8 +39,7 @@ def rename_file():
user_root = session.get('path')
path = request.json.get('path', '')
new_name = request.json.get('newName', '')
root_workspace_path = current_app.config['ROOT_WORKSPACE_PATH']
base_path = f'{root_workspace_path}/{user_root}'
base_path = f'{user_root}'
filepath = f'{base_path}/{path}'
try:
path_list = filepath.split('/')
@@ -58,8 +55,7 @@ def rename_folder():
user_root = session.get('path')
path = request.json.get('path', '')
new_name = request.json.get('newName', '')
root_workspace_path = current_app.config['ROOT_WORKSPACE_PATH']
base_path = f'{root_workspace_path}/{user_root}'
base_path = f'{user_root}'
folderpath = f'{base_path}/{path}'
try:
path_list = folderpath.split('/')
@@ -74,9 +70,8 @@ def rename_folder():
def copy_files():
old_path = request.json.get('oldPath', '')
new_path = request.json.get('newPath', '')
root_workspace_path = current_app.config['ROOT_WORKSPACE_PATH']
user_root = session.get('path')
base_path = f'{root_workspace_path}/{user_root}'
base_path = f'{user_root}'
old_path = f'{base_path}/{old_path}'
new_path = f'{base_path}/{new_path}'
# oldpath的父目录就是newpath的话,对oldpath进行重命名
@@ -106,9 +101,8 @@ def copy_files():
@file_bp.route('/delete-file', methods=['POST'])
def delete_file():
path = request.json.get('path', '')
root_workspace_path = current_app.config['ROOT_WORKSPACE_PATH']
user_root = session.get('path')
base_path = f'{root_workspace_path}/{user_root}'
base_path = f'{user_root}'
filepath = f'{base_path}/{path}'
if os.path.isdir(filepath):
shutil.rmtree(filepath)
@@ -119,8 +113,7 @@ def delete_file():
@file_bp.route('/file-content', methods=['POST'])
def file_content():
path = session.get('path')
root_workspace_path = current_app.config['ROOT_WORKSPACE_PATH']
base_path = f'{root_workspace_path}/{path}'
base_path = f'{path}'
filepath = request.json.get('path', '')
filepath = f'{base_path}/{filepath}'
if not os.path.exists(filepath):
@@ -135,8 +128,7 @@ def file_content():
@file_bp.route('/save-file', methods=['POST'])
def save_file():
path = session.get('path')
root_workspace_path = current_app.config['ROOT_WORKSPACE_PATH']
base_path = f'{root_workspace_path}/{path}'
base_path = f'{path}'
filepath = request.json.get('path', '')
filepath = f'{base_path}/{filepath}'
content = request.json.get('content', '')

42
code-server-like/app/views/routes.py
View File

@@ -4,9 +4,18 @@ from flask import Blueprint, request, jsonify, current_app, render_template, ses
from ..services.file_service import get_file_tree, load_config
main_bp = Blueprint('main', __name__, url_prefix='/vsc-like')
@main_bp.route('/<user_uuid>/<user_id>/<course_id>/<chapter_name>/<lesson_name>', methods=['GET'])
def vsc_like(user_uuid, user_id, course_id, chapter_name, lesson_name):
TERM_INIT_CONFIG = {
# instead of local server runnning this web terminal service
# "domain" is the target that you want to access through local server (with this web terminal)
# and before doing so - make sure you have username and port (on the "domain") to implement remote access
'domain': '15.168.14.27', # or ip address like 192.168.10.11
'client_path': {
'telnet': '/usr/bin/telnet', # confirmed location of your client binary (with cmd like 'which telnet')
'ssh': '/usr/bin/ssh'
}
}
@main_bp.route('/<user_uuid>/<user_id>/<course_id>/<chapter_name>/<lesson_name>/<port>', methods=['GET'])
def vsc_like(user_uuid, user_id, course_id, chapter_name, lesson_name, port):
uuid = user_uuid
base_path = current_app.config['ROOT_WORKSPACE_PATH']
path = os.path.join(base_path, user_id, course_id, chapter_name, lesson_name)
@@ -16,6 +25,11 @@ def vsc_like(user_uuid, user_id, course_id, chapter_name, lesson_name):
session['course_id'] = course_id
session['chapter_name'] = chapter_name
session['lesson_name'] = lesson_name
session['terminal_config'] = TERM_INIT_CONFIG
session['terminal_config']['term_type'] = 'ssh'
session['terminal_config']['username'] = user_id
session['terminal_config']['port'] = port
session.modified = True
config = load_config(path)
# if not config:
# return jsonify({"error": "Configuration file not found or invalid"}), 404
@@ -26,11 +40,17 @@ def vsc_like(user_uuid, user_id, course_id, chapter_name, lesson_name):
try:
subprocess.run(["sudo", "useradd", "-m", "-d", user_root, user_id]) # 创建用户
subprocess.run(["sudo", "chown", "-R", user_id, user_root]) # 设置目录权限为新用户
except subprocess.CalledProcessError as e:
print(f"Error creating user {user_id}: {e}")
try:
# 使用 sudo 执行 mkdir 命令来创建目录
subprocess.run(["sudo", "-u", user_id, "mkdir", "-p", path], check=True)
subprocess.run(["sudo", "-u", user_id, "mkdir", "-p", f'{user_root}/.ssh'], check=True)
flask_pub_key = "/home/flask/.ssh/id_ed25519.pub"
subprocess.run(["sudo", "cp", flask_pub_key, f"{user_root}/.ssh/authorized_keys"], check=True)
subprocess.run(["sudo", "chmod", "700", f"{user_root}/.ssh"], check=True)
subprocess.run(["sudo", "chmod", "600", f"{user_root}/.ssh/authorized_keys"], check=True)
print(f"Directory {path} created successfully for user {user_id}")
except subprocess.CalledProcessError as e:
print(f"Error creating directory {path} details {str(e)}")
@@ -44,17 +64,15 @@ def vsc_like(user_uuid, user_id, course_id, chapter_name, lesson_name):
subprocess.run(["sudo", "usermod", "-a", "-G", f"shared_group_{user_id}", 'flask'], check=True)
except subprocess.CalledProcessError as e:
print(f"Error adding user {user_id} to shared_group: {e}")
try:#将文件
subprocess.run(["sudo", "chown", "-R", f"flask:shared_group_{user_id}", user_root], check=True)
try:
subprocess.run(["sudo", "chown", "-R", f"flask:shared_group_{user_id}", path], check=True)
subprocess.run(["sudo", "chmod", "-R", "750", user_root], check=True)
subprocess.run(["sudo", "chmod", "-R", "770", path], check=True)
except subprocess.CalledProcessError as e:
print(f"Error changing directory {path} to shared_group_{user_id}: {e}")
try:#继承目录
subprocess.run(["sudo", "chmod", "g+s", path], check=True)
except subprocess.CalledProcessError as e:
print(f"Error changing directory {path} to shared_group_{user_id}: {e}")
return render_template('index.html')
return render_template('index.html', course_id=course_id, chapter_name=chapter_name, lesson_name=lesson_name)

6
code-server-like/static/js/file.js
View File

@@ -296,7 +296,11 @@ function createNewFolder(event) {
function saveFile(event) {
event.stopPropagation();
const content = editor.getValue();
fetch(`/vsc-like/save-file`, { method: 'POST', body: JSON.stringify({ path: item.path+'/'+item.name, content: content }) , headers: { 'Content-Type': 'application/json' } })
console.log(selectItem)
if (selectedItem.path==undefined){
selectedItem.path='.';
}
fetch(`/vsc-like/save-file`, { method: 'POST', body: JSON.stringify({ path: selectedItem.path+'/'+selectedItem.name, content: content }) , headers: { 'Content-Type': 'application/json' } })
.then(() => loadFileTree());
}

86
code-server-like/static/js/terminal.js
View File

@@ -1,43 +1,63 @@
const socket = io('https://hsamooc.cn/vsc-like',{path:'/vsc-like-ws'});
let term;
document.addEventListener('DOMContentLoaded', () => {
// 创建一个新的 xterm 实例
term = new Terminal({
cursorBlink: true,
cols: 80,
rows: 24,
});
// 打开终端到 #terminal 元素
Terminal.applyAddon(fullscreen)
Terminal.applyAddon(fit)
Terminal.applyAddon(webLinks)
Terminal.applyAddon(search)
const term = new Terminal({
col:80,
row:20,
cursorBlink: 5,
macOptionIsMeta: true,
});
term.open(document.getElementById('terminal'));
// 启动连接并监听数据
socket.on('connected', function (data) {
term.write('>');
term.fit()
term.resize(term.cols, term.rows-5)
console.log(`size: ${term.cols} columns, ${term.rows} rows`)
// term.toggleFullScreen(true)
term.fit()
term.write("Welcome to AQ web remote-only terminal!\r\n")
term.on('key', (key, ev) => {
socket.emit("pty_input", {"input": key}, function(response) {
});
});
// 监听后端返回的终端输出并显示
socket.on('terminal_output', function (data) {
if (data.data=='\x7f'){
term.write('\b \b')
return
const socket = io('https://hsamooc.cn/vsc-like',{path:'/vsc-like-ws'});
const status = document.getElementById("status")
socket.on("pty_output", function(data){
console.log("new output", data)
term.write(data.output)
})
socket.on("connect", () => {
// fitToscreen()
status.innerHTML = '<span style="background-color: lightgreen;">connected</span>'
socket.emit("pty_input",{"input":"cd "+window.lesson_data.course_id+"/"+window.lesson_data.chapter_name+'/'+window.lesson_data.lesson_name+"\r"})
}
term.write(data.data); // 输出来自后端的数据
console.log("get",data.data)
});
)
// 监听用户在终端中输入的命令
term.onData(function (data) {
socket.emit('send_input', { input: data }); // 将用户输入通过 WebSocket 发送给后端
});
socket.on("disconnect", () => {
status.innerHTML = '<span style="background-color: #ff8383;">disconnected</span>'
})
function fitToscreen(){
term.fit()
socket.emit("resize", {"cols": term.cols, "rows": term.rows})
}
function debounce(func, wait_ms) {
let timeout
return function(...args) {
const context = this
clearTimeout(timeout)
timeout = setTimeout(() => func.apply(context, args), wait_ms)
}
}
const wait_ms = 500;
window.onresize = debounce(fitToscreen, wait_ms)
// 如果终端被断开，显示消息
socket.on('error', function (error) {
term.write(`\nError: ${error.message}`);
});
socket.on('disconnect', function () {
term.write('\nDisconnected from terminal');
});
});

17
code-server-like/templates/index.html
View File

@@ -7,11 +7,15 @@
<link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/monaco-editor/0.34.0/min/vs/editor/editor.main.css">
<script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/4.5.1/socket.io.min.js"></script>
<link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/font-awesome/5.15.4/css/all.min.css">
<link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/xterm/5.5.0/xterm.css" integrity="sha512-AbNrj/oSHJaILgcdnkYm+DQ08SqVbZ8jlkJbFyyS1WDcAaXAcAfxJnCH69el7oVgTwVwyA5u5T+RdFyUykrV3Q==" crossorigin="anonymous" referrerpolicy="no-referrer" />
<script src="https://cdnjs.cloudflare.com/ajax/libs/xterm/5.5.0/xterm.js" integrity="sha512-Gujw5GajF5is3nMoGv9X+tCMqePLL/60qvAv1LofUZTV9jK8ENbM9L+maGmOsNzuZaiuyc/fpph1KT9uR5w3CQ==" crossorigin="anonymous" referrerpolicy="no-referrer"></script>
<link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/xterm/5.5.0/xterm.css" />
<!-- <script src="https://cdnjs.cloudflare.com/ajax/libs/xterm/5.5.0/xterm.js"></script> -->
<script src="https://unpkg.com/xterm@3.6.0/dist/xterm.js"></script>
<script src="https://unpkg.com/xterm@3.6.0/dist/addons/fit/fit.js"></script>
<script src="https://unpkg.com/xterm@3.6.0/dist/addons/webLinks/webLinks.js"></script>
<script src="https://unpkg.com/xterm@3.6.0/dist/addons/fullscreen/fullscreen.js"></script>
<script src="https://unpkg.com/xterm@3.6.0/dist/addons/search/search.js"></script>
<link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/font-awesome/5.15.4/css/all.min.css">
<link rel="stylesheet" href="/vsc-like/static/css/index.css">
</head>
<body>
@@ -48,6 +52,7 @@
</div>
<!-- 终端部分 -->
<span style="font-size: small;">status: <span style="font-size: small;" id="status">connecting...</span></span>
<div id="terminal">
</div>
</div>
@@ -58,6 +63,12 @@
<script src="/vsc-like/static/js/file.js"></script>
<script src="/vsc-like/static/js/terminal.js"></script>
<script>
window.lesson_data = {
course_id:'{{course_id}}',
chapter_name:'{{chapter_name}}',
lesson_name:'{{lesson_name}}'
}
require.config({ paths: { vs: 'https://cdnjs.cloudflare.com/ajax/libs/monaco-editor/0.34.0/min/vs' } });
require(['vs/editor/editor.main'], function() {

3
requirements.txt
View File

@@ -7,4 +7,5 @@ websocket-client
eventlet
flask_sqlalchemy
flask-cors
pydantic
pydantic
psutil