2022怡然: 后端接口
来自CloudWiki
(重定向自2022怡然: "主机列表"后端)
form_sumbit
def dingtalk_robot(webhook, secret,room_id,pc_id,reason): dogBOSS = DingtalkChatbot(webhook, secret) now_time = datetime.now().strftime('%Y.%m.%d %H:%M:%S') dogBOSS.send_markdown( title=f'来自梦无矶小仔的提醒', text= f'**机房id{room_id}**\n\n' f'**PC_id{pc_id}**\n\n' f'**故障原因{reason}**\n\n' f'**发送时间:** {now_time}\n\n', is_at_all=True) password = sha256_crypt.encrypt(str(form.password.data)) # 对密码进行加密 if request.method == 'POST' and form.validate(): # 如果提交表单,并字段验证通过 from flask import Flask, render_template, flash, redirect, url_for, session, request, logging #from mysql_util import MysqlUtil ` from passlib.hash import sha256_crypt from forms import RegisterForm,ArticleForm @app.route('/user_upload', methods=['post'], strict_slashes=False) ` from passlib.hash import sha256_crypt from forms import RegisterForm,ArticleForm @app.route('/user_upload', methods=['post'], strict_slashes=False) def user_upload(): form = RegisterForm(request.form) # 实例化表单类 room_id,pc_id,reason,time = null,null,null,null # 获取字段内容 room_id = form.room_id.data # username = form.username.data #user_info = request.values.to_dict() #room_id = user_info.get("room_id") #上传机房名称的变量名 #pc_id = user_info.get("pc_id") #reason = user_info.get("reason") #上传原因的变量名 #time = user_info.get("time") #上传时间的变量名 # cursor = conn.cursor() #cursor = conn1.cursor() #sql_insert = """insert into jieshou(room_id,reason,time) values(%s,%s,%s)""" #cursor.execute(sql_insert, (str(room_id),str(reason),str(time))) # content的内容是content=(movie,star,time)>,content的数据要与sql语句中的占位>符数量相等 #conn.commit() webhook = 'https://oapi.dingtalk.com/robot/send?access_token=47ec6dbe51f73e311e6dc4d2752d482d6a252c52690e7e5c46b67a026f2cd963' secrets = 'SEC4537ca10868ea4e0d5524499aac4dae0f3ea2aa0a9fe0b7185cfc1b51c3a50de' dingtalk_robot(webhook,secrets,room_id,pc_id,reason) return jsonify({"success": 0, "msg": "报告成功"})
远程开机关机
from flask import Flask, render_template, redirect, url_for, request, abort, jsonify from test_ansible2 import do_ansible import pymysql import json app = Flask("EmailDemo", static_folder="static", template_folder="templates") app.config["JSON_AS_ASCII"] = False conn1 = pymysql.connect(host='localhost', user='root', password='123456', port=3306, db='rooms') conn2 = pymysql.connect(host='localhost', user='root', password='123456', port=3306, db='Receive_mail') ####### 远程开机关机 ###### #查看全部机房列表 @app.route('/rooms', methods=['GET', 'POST']) def get_rooms_list(): conn1.ping(reconnect=True) cur = conn1.cursor() sql = "SELECT * from tb_rooms LIMIT 4" cur.execute(sql) u = cur.fetchall() jsondata = [] for row in u: result = {} result['id'] = row[0] result['title']="机房"+str(row[0]) result['total_num'] = row[1] result['running_num'] = row[2] result['close_num'] = int(row[1])-int(row[2]) jsondata.append(result) return jsonify(jsondata) abort(404) # 机房详细的机器列表 @app.route('/room/<int:id>', methods=['GET', 'POST']) def get_room(id): conn1.ping(reconnect=True) cur = conn1.cursor() sql = f"SELECT * from tb_rooms where room_id={id}" cur.execute(sql) u = cur.fetchall() jsondata = [] for row in u: result = {} result['pc_id'] = row[1] result['status'] = row[2] jsondata.append(result) return jsonify(jsondata) abort(404) #一键关闭某机房所有机器 close all machines of one room,GET操作 @app.route('/ansible/api/v1/machine_close/all/<string:host>', methods=['GET']) def shutdown_machine(host): host_list=[host] tasks_close=[ dict(action=dict(module='shell', args='shutdown -h now'), register='shell_out'), ] #do_ansible(host_list,tasks_close) return jsonify({'status': host+"shutdown order is sent"}) abort(404) #关闭某机房内某台机器 close one machines of one room,GET操作 @app.route('/ansible/machine_close/single/<string:host>/<string:pc_id>', methods=['GET']) def shutdown_one_machine(host,pc_id): host_list=[host] tasks_close=[ dict(action=dict(module='shell', args='shutdown -h now'), register='shell_out'), ] #do_ansible(host_list,tasks_close) return jsonify({'status': host+"_pc_"+pc_id+"shutdown order is sent"}) abort(404) #一键重启机房所有机器restart all machines of one room ,GET操作 @app.route('/ansible/api/v1/machine_restart/all/<string:host>', methods=['GET']) def restart_machine(host): host_list=[host] tasks=[ dict(action=dict(module='shell', args='reboot'), register='shell_out'), ] #do_ansible(host_list,tasks_close) return jsonify({'status': host+"restart order is sent"}) abort(404) #重启某机房内某台机器 restart one machine of one room ,GET操作 @app.route('/ansible/machine_restart/single/<string:host>/<string:pc_id>', methods=['GET']) def restart_one_machine(host,pc_id): host_list=[host] tasks=[ dict(action=dict(module='shell', args='reboot'), register='shell_out'), ] #do_ansible(host_list,tasks_close) return jsonify({'status': host+"_pc_"+pc_id+"restart order is sent"}) abort(404)
远程实时监控
def return_img_stream(img_local_path): """ 工具函数: 获取本地图片流 :param img_local_path:文件单张图片的本地绝对路径 :return: 图片流 """ import base64 img_stream = '' with open(img_local_path, 'rb') as img_f: img_stream = img_f.read() img_stream = base64.b64encode(img_stream).decode() return img_stream @app.route('/yuanchengjiankong') def yuanchengjiankong(): img_path = f'/home/esu/ansible/tu/fullscreen1.png' img_stream = return_img_stream(img_path) return img_stream