@ -1,3 +1,4 @@
# this got big...
from flask import Flask , jsonify
from flask_apscheduler import APScheduler
import psutil
@ -6,12 +7,20 @@ import requests
from requests import RequestException , Response
import json
from subprocess import check_output
import win32api
from datetime import datetime , timedelta
import sys
from pathlib import Path
import wmi
import pythoncom
import ctypes
from ctypes import wintypes
app = Flask ( __name__ )
scheduler = APScheduler ( )
app . config [ ' JSONIFY_PRETTYPRINT_REGULAR ' ] = True
# Bits to Bytes etc
# human readable bytes
def bytes_to_human_readable ( bytes ) :
for unit in [ ' B ' , ' KB ' , ' MB ' , ' GB ' , ' TB ' ] :
if bytes < 1024.0 :
@ -111,8 +120,10 @@ def get_disk_info():
partitions = psutil . disk_partitions ( )
for partition in partitions :
usage = psutil . disk_usage ( partition . mountpoint )
drive_letter = partition . device . replace ( ' \\ \\ ' , ' \\ ' ) . rstrip ( ' \\ ' )
disk_info . append ( {
' device ' : partition . device . replace ( ' \\ \\ ' , ' \\ ' ) . rstrip ( ' \\ ' ) ,
' device ' : drive_letter ,
' label ' : get_drive_label ( drive_letter ) ,
#'mountpoint': partition.mountpoint,
' fstype ' : partition . fstype ,
' total ' : bytes_to_human_readable ( usage . total ) ,
@ -122,6 +133,101 @@ def get_disk_info():
} )
return disk_info
# drive label function
def get_drive_label ( drive_letter : str ) - > str :
result = " none "
root = drive_letter . strip ( )
if not root . endswith ( ' : ' ) :
root + = ' : '
if not root . endswith ( ' \\ ' ) :
root + = ' \\ '
# Make sure the drive actually exists
if not os . path . exists ( root ) :
# Not a valid drive letter, return None so the caller can decide what to do.
print ( f " [DEBUG] Drive ' { root } ' does not exist. " )
result = " drive does not exist 0_o "
# Prepare buffers for the Win32 API call
volume_name_buf = ctypes . create_unicode_buffer ( 260 ) # MAX_PATH
fs_name_buf = ctypes . create_unicode_buffer ( 260 )
serial_number = wintypes . DWORD ( )
max_component_len = wintypes . DWORD ( )
file_system_flags = wintypes . DWORD ( )
# Call GetVolumeInformationW
res = ctypes . windll . kernel32 . GetVolumeInformationW (
ctypes . c_wchar_p ( root ) , # lpRootPathName
volume_name_buf , # lpVolumeNameBuffer
ctypes . sizeof ( volume_name_buf ) , # nVolumeNameSize
ctypes . byref ( serial_number ) , # lpVolumeSerialNumber
ctypes . byref ( max_component_len ) , # lpMaximumComponentLength
ctypes . byref ( file_system_flags ) , # lpFileSystemFlags
fs_name_buf , # lpFileSystemNameBuffer
ctypes . sizeof ( fs_name_buf ) # nFileSystemNameSize
)
if res == 0 : # The call failed
err = ctypes . get_last_error ( )
print ( f " [ERROR] GetVolumeInformationW failed for ' { root } ' . "
f " Win32 error code: { err } " )
result = " label error "
else :
result = volume_name_buf . value
if volume_name_buf . value == ' ' :
result = " no_label "
return result
# os info function
def get_os_info ( ) - > str :
result = " windows "
pythoncom . CoInitialize ( )
#try:
wmi_data = wmi . WMI ( )
os_info = wmi_data . Win32_OperatingSystem ( ) [ 0 ]
#return {
# "Name" : os_info.Name,
# "Version" : os_info.Version,
# "BuildNumber": os_info.BuildNumber,
# "InstallDate": os_info.InstallDate,
# "ProductType": int(os_info.ProductType)
#}
# 1. Major version + edition (e.g. "10 Pro")
# os_info.Caption → "Microsoft Windows 10 Pro"
parts = os_info . Caption . split ( )
major = parts [ 2 ] # "10"
edition = parts [ 3 ] # "Pro" (for server: "2019", etc.)
major_edition = f " { major } { edition } "
# 2. Build number
build = os_info . BuildNumber
# 3. Install date (WMI gives an ISO‑
# e.g. "20210930142300.000000+000"
install_ts = os_info . InstallDate [ : 14 ] # "20210930142300"
dt = datetime . strptime ( install_ts , " % Y % m %d % H % M % S " )
install_date = f " { dt . month } - { dt . day } - { dt . year } "
result = f " Windows { major_edition } - Build { build } - Installed { install_date } "
#except Exception as e:
# print(e)
# result = "wmi_error"
return result
# server reporter info
def get_server_info ( ) - > dict :
result = { }
drives_dict = get_crystal_disk_info ( )
data_dict = {
" hostname " : " {{ hostname_output.stdout_lines[0] }} " ,
" os_string " : get_os_info ( ) ,
" drives " : drives_dict [ " drives " ] ,
" API_KEY " : " deadbeef " ,
" storage_summary " : get_disk_info ( )
}
result = data_dict
return result
# Flask endpoints
@app.route ( ' /disk ' , methods = [ ' GET ' ] )
@ -132,18 +238,23 @@ def disk():
def drive_health ( ) :
return jsonify ( get_crystal_disk_info ( ) )
@app.route ( ' /full ' , methods = [ ' GET ' ] )
def full_summary ( ) :
return jsonify ( get_server_info ( ) )
@app.route ( ' /test ' , methods = [ ' GET ' ] )
def test_flask ( ) :
return jsonify ( {
" message " : " hello world " ,
" os_string " : get_os_info ( )
} )
def server_reporter ( ) :
#base_url="https://cosmostat.matt-cloud.com"
base_url = " http://10.200.27.20:5001 "
url = f " { base_url } /storage_client_update "
drives _dict = get_crystal_disk _info ( )
data_dict = {
" hostname " : " {{ hostname_output.stdout_lines[0] }} " ,
" drives " : drives_dict [ " drives " ] ,
" API_KEY " : " deadbeef " ,
" storage_summary " : get_disk_info ( )
}
data _dict = get_server _info ( )
result = [ ]
try :
response = requests . post ( url , json = data_dict )
@ -170,6 +281,13 @@ if __name__ == '__main__':
return result
# gonna try something wild
run_date = datetime . now ( ) + timedelta ( seconds = 15 )
scheduler . add_job ( id = ' init_disk_info ' ,
func = update_disk_info ,
trigger = ' date ' ,
run_date = run_date )
scheduler . add_job ( id = ' update_disk_info ' ,
func = update_disk_info ,
trigger = ' interval ' ,
@ -183,6 +301,6 @@ if __name__ == '__main__':
scheduler . init_app ( app )
scheduler . start ( )
(
# update_disk_info( )
app . run ( host = ' 0.0.0.0 ' , port = { { api_service_port } } )
