99 lines
2.7 KiB
Python
99 lines
2.7 KiB
Python
from flask import Flask, jsonify, request
|
|
import sqlite3
|
|
import redis, json, time
|
|
from SSDObject import *
|
|
from flask_apscheduler import APScheduler
|
|
from HostRedis import *
|
|
|
|
app = Flask(__name__)
|
|
debug_output = False
|
|
secure_api = False
|
|
push_redis = True
|
|
|
|
####################################################
|
|
### Flask Routes
|
|
####################################################
|
|
|
|
# Route to check if a serial number exists in the database
|
|
@app.route('/check', methods=['GET'])
|
|
def check():
|
|
serial_lookup = request.args.get('serial_lookup')
|
|
if debug_output:
|
|
print(f"Serial to check: {serial_lookup}")
|
|
if not serial_lookup:
|
|
return jsonify({'error': 'No serial number provided'}), 400
|
|
|
|
exists = check_serial_exists(serial_lookup)
|
|
return jsonify({'serial_number_exists': exists, 'serial_lookup': serial_lookup})
|
|
|
|
# Route to get all drive records in JSON format
|
|
@app.route('/drives', methods=['GET'])
|
|
def index():
|
|
return get_all_drive_records()
|
|
|
|
# Route to return active drives
|
|
@app.route('/list_active_drives', methods=['GET'])
|
|
def list_active_drives():
|
|
return jsonify(list_disk_and_serial())
|
|
|
|
# host stats
|
|
@app.route('/host_stats', methods=['GET'])
|
|
def host_stats():
|
|
if push_redis:
|
|
update_stats_redis()
|
|
return jsonify(get_host_stats())
|
|
|
|
# test route
|
|
@app.route('/test', methods=['GET'])
|
|
def test():
|
|
db_check = "SELECT name FROM sqlite_master WHERE type='table';"
|
|
return query_db(db_check)
|
|
|
|
####################################################
|
|
### Flask Scheduler Handlers
|
|
####################################################
|
|
def ssd_scan():
|
|
disk_list_command = "ls -lo /dev/sd? | awk '{print $9}' | cut -d/ -f3"
|
|
disk_list = run_command(disk_list_command, zero_only = False)
|
|
disk_objects = []
|
|
for disk in disk_list:
|
|
if debug_output:
|
|
print(disk)
|
|
try:
|
|
disk_objects.append(SSDObject(dev_id = disk))
|
|
except (TypeError, ValueError, KeyError, OSError) as e:
|
|
print(f"Error - {e}")
|
|
if debug_output:
|
|
for disk in disk_objects:
|
|
print(disk)
|
|
print(list_disk_and_serial())
|
|
if push_redis:
|
|
update_stats_redis()
|
|
update_disk_redis()
|
|
time.sleep(0.2)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
# send immediate stats update to redis
|
|
if push_redis:
|
|
update_stats_redis()
|
|
|
|
# Flask scheduler for scanner
|
|
scheduler = APScheduler()
|
|
scheduler.add_job(id='ssd_check',
|
|
func=ssd_scan,
|
|
trigger='interval',
|
|
seconds=1)
|
|
scheduler.init_app(app)
|
|
scheduler.start()
|
|
|
|
if secure_api:
|
|
app.run(debug=True, host='172.17.0.1', port=5000)
|
|
else:
|
|
app.run(debug=True, host='0.0.0.0', port=5000)
|
|
|
|
|
|
|
|
|
|
|
|
|