# myapp/views/vscode.py import re import os, uuid, json from flask import Blueprint, current_app, session, redirect, url_for, render_template, request, jsonify from ..services.backboard_service import realtime_response from ..extension_ase.ase_client.manager import ChatManager import subprocess bp = Blueprint("vscode", __name__) @bp.route("/desktop////") def desktop(user_uuid, course_id, chapter_name, lesson_name): username2uuid = current_app.extensions["username2uuid"] uuid2username = current_app.extensions["uuid2username"] userid_recorder = current_app.extensions["userid_recorder"] if "user_uuid" not in session: return redirect(url_for("auth.login")) # 全局映射 user_id = uuid2username[user_uuid] userid_recorder[f"{user_id}&{course_id}"] = session["user_uuid"] user_home = os.path.join("/home", user_id) try: subprocess.run(["sudo", "useradd", "-m", "-d", user_home, user_id]) # 创建用户 subprocess.run(["sudo", "chown", "-R", user_id, user_home]) # 设置目录权限为新用户 except subprocess.CalledProcessError as e: print(f"Error creating user {user_id}: {e}") path_dir = os.path.join("/home", user_id, course_id, chapter_name, lesson_name) try: # 使用 sudo 执行 mkdir 命令来创建目录 subprocess.run(["sudo", "-u", user_id, "mkdir", "-p", path_dir], check=True) print(f"Directory {path_dir} created successfully for user {user_id}") except subprocess.CalledProcessError as e: print(f"Error creating directory {path_dir} details {str(e)}") chatmanager = ChatManager() current_app.extensions["user_uuid2chatmanager"][user_uuid] = chatmanager code_server_port = 10000 + int(uuid.uuid4().int % 10000) chatmanager.start_code_server(user_uuid, user_id, path_dir, code_server_port) caddy_conf = f""" handle_path /vsc{user_uuid}/* {{ reverse_proxy localhost:{code_server_port} }} """ caddy_config_path = "/etc/caddy/Caddyfile" with open(caddy_config_path, "r") as f: lines = f.readlines() pattern = re.compile( r'handle_path\s+/vsc' + re.escape(user_uuid) + r'/\* {.*?reverse_proxy.*?}', re.DOTALL ) content = "".join(lines) if pattern.search(content): new_content = pattern.sub('', content) lines = new_content.splitlines(keepends=True) if len(lines) >= 2: insert_position = len(lines) - 2 else: insert_position = len(lines) if caddy_conf[-1] not in ['\n', '\r']: caddy_conf += '\n' lines.insert(insert_position, caddy_conf) with open(caddy_config_path, "w") as f: f.writelines(lines) try: subprocess.run(["sudo", "caddy", "reload", "--config", "/etc/caddy/Caddyfile"]) except subprocess.CalledProcessError as e: print(f"Error reload caddy {e}") cfg = current_app.config["VSCODE_WEB_PATH"] path_for_vscode = path_dir config_path = os.path.join(path_dir, ".config") tmpd = { "user_uuid": session["user_uuid"], "course_id": course_id, "chapter_name": chapter_name, "lesson_name": lesson_name, "path": path_for_vscode } config_json = json.dumps(tmpd, ensure_ascii=False) # 使用 sudo 和 tee 命令将 JSON 数据写入文件 subprocess.run( ["sudo", "-u", user_id, "bash", "-c", f"echo '{config_json}' | tee {config_path} > /dev/null"], check=True ) # 确保文件的所有权归 user_id subprocess.run( ["sudo", "chown", f"{user_id}:{user_id}", config_path], check=True ) return render_template( "desktop.html", vscode_web_url=current_app.config["VSCODE_WEB_URL"]+f"/vsc{user_uuid}", user_uuid=session["user_uuid"], course_id=course_id, chapter_name=chapter_name, lesson_name=lesson_name, workspace_path=path_for_vscode, ) @bp.route("/desktop_nouser///") def desktop_nouser(course_id, chapter_name, lesson_name): if "user_uuid" not in session: return redirect(url_for("auth.login")) # 如果你有 auth 蓝图 user_uuid = session["user_uuid"] return redirect(url_for("vscode.desktop", user_uuid=user_uuid, course_id=course_id, chapter_name=chapter_name, lesson_name=lesson_name)) @bp.route("/vscode_data", methods=["POST"]) def vscode_data(): # data = request.get_json(force=True) # config = data.get("config") or {} # realtime_response(config, data) # 调用服务层逻辑 # return jsonify({"status": "success", "received": data}) return jsonify({"status": "success", "received": "success"})