205 lines
8.1 KiB
Python
205 lines
8.1 KiB
Python
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) |