import os import re import platform import subprocess import psutil import winreg from flask import Flask, jsonify, request from flask_cors import CORS app = Flask(__name__) # <--- 2. 启用全局跨域支持 # 默认允许所有域名跨域访问所有接口。 # 如果想只允许特定前端域名，可以改成 CORS(app, origins=["http://localhost:8080", "http://你的域名.com"]) CORS(app, supports_credentials=True) # Windows 下分离进程的标志位，确保 exe 独立运行 DETACHED_PROCESS = 0x00000008 def get_cpu_model(): """ 通过读取 Windows 注册表获取准确的 CPU 型号 """ try: # 打开注册表路径 key = winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, r"HARDWARE\DESCRIPTION\System\CentralProcessor\0") # 读取 ProcessorNameString 的值 cpu_model, _ = winreg.QueryValueEx(key, "ProcessorNameString") return cpu_model.strip() except Exception: return "Unknown CPU" def is_valid_task_id(task_id): """ 校验 task_id，只允许字母、数字、下划线和连字符，防止目录穿越攻击 """ return re.match(r'^[\w\-]+$', task_id) is not None @app.route('/api/system_info', methods=['GET']) def get_system_info(): """ 接口 1：获取当前设备(Windows)的负载情况和硬件信息（附带字段说明） """ try: # 获取基础监控数据 cpu_usage = psutil.cpu_percent(interval=0.5) cpu_cores_physical = psutil.cpu_count(logical=False) cpu_cores_logical = psutil.cpu_count(logical=True) cpu_freq = psutil.cpu_freq() virtual_mem = psutil.virtual_memory() disk_usage = psutil.disk_usage('C:\\') # 构建返回数据，为每个返回值添加 desc 注释说明 info = { # "os": { # "system": platform.system(), # 操作系统名称 (如 Windows) # "release": platform.release(), # 操作系统主要版本号 (如 10, 11) # "version": platform.version(), # 操作系统的详细内部构建版本号 # "machine": platform.machine() # 系统架构 (如 AMD64 代表 64位) # }, "os": f"{platform.system()} {platform.version()} ({platform.machine()})", # 综合描述操作系统的字符串 "hardware": { "cpu": f"{get_cpu_model()} ({cpu_cores_physical}P/{cpu_cores_logical}L) {cpu_freq.max if cpu_freq else None}GHz", # CPU 型号和核心数的综合描述字符串 # "cpu_model": get_cpu_model(), # CPU 具体型号名称 # "cpu_cores_physical": cpu_cores_physical, # CPU 物理核心数 (真实核心) # "cpu_cores_logical": cpu_cores_logical, # CPU 逻辑核心数 (包含超线程) # "cpu_max_freq_mhz": cpu_freq.max if cpu_freq else None, # CPU 最大设计频率 (单位: MHz) "ram_total_gb": round(virtual_mem.total / (1024 ** 3), 2), # 物理内存 (RAM) 总容量 (单位: GB) "disk_c_total_gb": round(disk_usage.total / (1024 ** 3), 2) # 系统盘 (C盘) 总容量 (单位: GB) }, "load": { "cpu_usage_percent": cpu_usage, # 当前 CPU 总利用率 (单位: %) "ram_usage_percent": virtual_mem.percent, # 当前物理内存使用率 (单位: %) "ram_available_gb": round(virtual_mem.available / (1024 ** 3), 2), # 当前可用的物理内存大小 (单位: GB) "disk_c_usage_percent": disk_usage.percent, # 系统盘 (C盘) 存储空间使用率 (单位: %) "disk_c_free_gb": round(disk_usage.free / (1024 ** 3), 2) # 系统盘 (C盘) 剩余可用空间 (单位: GB) } } return jsonify({"code": 200, "status": "success", "data": info}), 200 except Exception as e: return jsonify({"code": 500, "status": "error", "message": str(e)}), 500 @app.route('/api/run_exe', methods=['POST']) def run_exe(): """ 接口 2：非阻塞地运行一个 exe 程序 """ data = request.get_json() if not data or 'exe_path' not in data: return jsonify({"status": "error", "message": "Missing 'exe_path' in request body."}), 400 exe_path = data['exe_path'] args = data.get('args', []) if not os.path.exists(exe_path): return jsonify({"status": "error", "message": f"File not found: {exe_path}"}), 404 try: command = [exe_path] + args process = subprocess.Popen( command, creationflags=DETACHED_PROCESS, close_fds=True, shell=False ) return jsonify({ "code": 200, "status": "success", "message": "Executable started successfully.", "pid": process.pid }), 200 except Exception as e: return jsonify({"code": 500, "status": "error", "message": str(e)}), 500 @app.route('/api/upload', methods=['POST']) def upload_files(): """ 接口 3：上传文件 接收 form-data 格式数据： - task_id: 任务ID (字符串) - file: 上传的文件 (可以同时上传多个文件) """ # 1. 获取并校验 task_id task_id = request.form.get('task_id') if not task_id: return jsonify({"status": "error", "message": "Missing 'task_id' in form data."}), 400 if not is_valid_task_id(task_id): return jsonify({"status": "error", "message": "Invalid 'task_id'. Only alphanumeric, dash, and underscore are allowed."}), 400 # 2. 准备目录 # os.path.join 会自动根据当前系统(Windows)处理路径分隔符 data_dir = os.path.join("data", task_id) lib_dir = "lib/module" # 自动创建不存在的目录 (exist_ok=True 避免目录已存在时报错) os.makedirs(data_dir, exist_ok=True) os.makedirs(lib_dir, exist_ok=True) saved_files = [] # 3. 遍历并保存文件 # request.files 是一个字典，通过 .getlist('file') 获取所有上传的文件（假设表单键名为 file） # 为了兼容不同的前端传法，我们遍历 request.files 中的所有文件对象 for key in request.files: for file in request.files.getlist(key): if file and file.filename: # 获取文件后缀，转换为小写 ext = os.path.splitext(file.filename)[1].lower() if ext == '.txt': # .txt 储存至 data/{task_id}/goods.txt save_path = os.path.join(data_dir, "goods.txt") file.save(save_path) saved_files.append({"original_name": file.filename, "saved_path": save_path}) elif ext == '.py': # .py 储存至 lib/{task_id}.py save_path = os.path.join(lib_dir, f"{task_id}.py") file.save(save_path) saved_files.append({"original_name": file.filename, "saved_path": save_path}) else: # 忽略不支持的文件格式 pass if not saved_files: return jsonify({"status": "error", "message": "No valid .txt or .py files were uploaded."}), 400 return jsonify({ "status": "success", "message": "Files uploaded successfully.", "data": saved_files }), 200 @app.route('/api/get_status', methods=['GET']) def get_status(): params = request.args task_id = params.get('task_id') progress = 0 status = 0 if os.path.exists(f"data/{task_id}/progress"): with open(f"data/{task_id}/progress", "r", encoding="utf-8") as f: progress = f.read().strip() if os.path.exists(f"data/{task_id}/status"): with open(f"data/{task_id}/status", "r", encoding="utf-8") as f: status = f.read().strip() return jsonify({ "status": "success", "message": f"Task '{task_id}' is ready.", "data": { "task_id": task_id, "progress": progress, "status": status } }), 200 if __name__ == '__main__': # 解决 jsonify 返回中文字符时变成 Unicode 编码的问题 app.config['JSON_AS_ASCII'] = False app.run(host='127.0.0.1', port=5000, debug=True)