feat(control): 系统控制

test
jshixiong 1 year ago
parent c85e4e5149
commit 4e827f2513

@ -26,8 +26,12 @@ def create_app():
log_file_path = '/data/ww/py_sys_monitor/info/access_log'
def log_request_response(response):
response_data = response.get_json(silent=True)
if "route.get_resource" == request.endpoint:
return response
elif "route.get_login_info" == request.endpoint:
response_data = {}
log_entry = {
'time': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
@ -41,7 +45,7 @@ def create_app():
"json": request.get_json(silent=True)
}, indent=4),
'response_status': response.status,
'response_data': response.get_json(silent=True)
'response_data': json.dumps(response_data, indent=4)
}
os.makedirs(os.path.dirname(log_file_path), exist_ok=True)

@ -8,6 +8,7 @@ import app.login_info as login_info
import app.request_info as request_info
import app.imitate_thread as imitate_thread
import app.sys_clean as sys_clean
import app.sys_control as sys_control
route_bp = Blueprint('route', __name__)
@ -81,3 +82,13 @@ def get_junk_file():
def clean_junk_file():
filenames = request.json.get('file_list', [])
return sys_clean.clean_junk_file(filenames)
# cpu跑满
@route_bp.route('/sys/cpu_full', methods=['POST'])
def execute_bling():
return sys_control.execute_cpu_full_script()
# 内存跑满
@route_bp.route('/sys/cpu_full', methods=['POST'])
def execute_bling():
return sys_control.execute_memory_full_script()

@ -0,0 +1,24 @@
import subprocess
import logging
from flask import jsonify
def execute_cpu_full_script():
script_path = '/data/ww/py_sys_monitor/control/cpu_full.py'
try:
result = subprocess.run(['python3', script_path], capture_output=True, text=True, check=True)
output = result.stdout
return jsonify({"message": "successfully", "output": output})
except subprocess.CalledProcessError as e:
logging.info("")
return jsonify({"message": "failed", "error": str(e.output)})
def execute_memory_full_script():
script_path = '/data/ww/py_sys_monitor/control/memory_full.py'
try:
result = subprocess.run(['python3', script_path], capture_output=True, text=True, check=True)
output = result.stdout
return jsonify({"message": "successfully", "output": output})
except subprocess.CalledProcessError as e:
logging.info("")
return jsonify({"message": "failed", "error": str(e.output)})
Loading…
Cancel
Save