# 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 from ..auth.decorators import require_role from ..services.course_service import load_learning_progress bp = Blueprint("vscode", __name__) @bp.route("/desktop////") @require_role 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")) # 检查是否有 load_history 参数 load_history_str = request.args.get('load_history') if load_history_str is None: # 如果没有参数，跳转到确认页面 return redirect(url_for("vscode.desktop_nouser", course_id=course_id, chapter_name=chapter_name, lesson_name=lesson_name)) # 将字符串转换为布尔值 load_history = load_history_str.lower() == 'true' # 全局映射 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)}") 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(restart=not load_history) 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_id}/* {{ 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_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) 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"], "user_id": user_id, "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","-R", f"flask:shared_group_{user_id}", config_path], check=True ) subprocess.run( ["sudo", "chmod", "-R", "770", config_path], check=True ) return render_template( "code-like-desktop.html", # vscode_web_url=current_app.config["VSCODE_WEB_URL"]+f"/vsc{user_id}", vscode_web_url=current_app.config["CODE_LIKE_URL"], user_uuid=session["user_uuid"], course_id=course_id, chapter_name=chapter_name, lesson_name=lesson_name,user_id=user_id, workspace_path=path_for_vscode, ) @bp.route("/desktop_nouser///") @require_role 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"] uuid2username = current_app.extensions["uuid2username"] user_id = uuid2username[user_uuid] # 检查是否有学习记录 progress_result = load_learning_progress(user_id, course_id, chapter_name, lesson_name) if not progress_result["exists"]: # 没有学习记录，直接跳转到学习页面，不显示确认加载历史的页面 return redirect(url_for("vscode.desktop", user_uuid=user_uuid, course_id=course_id, chapter_name=chapter_name, lesson_name=lesson_name, load_history="false")) # 获取学习记录数据 progress_data = progress_result["data"] chapter_chain_now = progress_data.get("chapter_chain_now", -1) chapter_chain_length = progress_data.get("chapter_chain_length", 0) # 检查是否已经学完所有章节 is_course_finished = chapter_chain_now >= chapter_chain_length - 1 return render_template( "confirm_load_history.html", user_uuid=user_uuid, course_id=course_id, chapter_name=chapter_name, lesson_name=lesson_name, is_course_finished=is_course_finished ) @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"})