#!/usr/bin/python3
import cgi, sys
sys.path.insert(0, '/opt/ngon/apps')
from managers.auth_manager import AuthManager, generate_login_page_html
_form = cgi.FieldStorage()
_auth = AuthManager('inventory_check')
_auth_required, _should_exit, _headers = _auth.require_auth(_form)
if _should_exit:
print("Content-Type: application/json")
if _headers:
print(_headers)
print("")
if _auth_required:
print('{"success": false, "error": "Authentication required"}')
else:
print('{"success": true}')
sys.exit(0)
if _auth_required:
print("Content-Type: text/html")
print("")
print(generate_login_page_html("Inventory Comparison"))
sys.exit(0)
import csv
import os
import sys
import json
from collections import defaultdict
from datetime import datetime
# Import universal navigation
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from links import generate_dropdown_html, generate_dropdown_css, generate_dropdown_js
# --- Configuration ---
# Define the absolute paths to your data files on the server.
MINERS_CSV_PATH = '/opt/ngon/data/live/miner_status.csv'
INVENTORY_CSV_PATH = '/opt/ngon/data/live/miner_inventory.csv'
def read_csv_file(file_path):
"""Reads a CSV file and returns its content as a list of dictionaries."""
if not os.path.exists(file_path):
return None, f"Error: File not found at {file_path}"
try:
with open(file_path, mode='r', encoding='utf-8') as infile:
reader = csv.DictReader(infile)
return list(reader), None
except Exception as e:
return None, f"Error reading or parsing {file_path}: {e}"
def get_partial_serial(serial):
"""
Creates a partial serial key from the first 6 and last 6 characters.
Returns None if the serial is too short.
"""
if serial and len(serial) >= 12:
return serial[:6] + serial[-6:]
return None
def generate_results_table(title, data, headers, include_download_button=False):
"""Generates an HTML table string for a given set of results."""
# Clean headers for display
display_headers = [h for h in headers if not h.startswith('__')]
table_html = f"""
{title}
"""
# Add download button for the "Found Online Only" section
if include_download_button and data:
# Extract IP addresses and create CSV content directly
ip_list = [item.get('ip', '') for item in data if item.get('ip', '').strip()]
csv_content = "IP Address\n" + "\n".join(ip_list)
# Create data URL for direct download
import base64
csv_bytes = csv_content.encode('utf-8')
csv_b64 = base64.b64encode(csv_bytes).decode('utf-8')
data_url = f"data:text/csv;base64,{csv_b64}"
table_html += f'''
Download IP List ({len(ip_list)} IPs)
'''
table_html += """
"""
if not data:
table_html += '
No discrepancies found in this category.
'
else:
table_html += '
'
for header in display_headers:
table_html += f'
{header.replace("_", " ").title()}
'
table_html += '
'
for item in data:
table_html += '
'
for header in display_headers:
value = item.get(header, '')
table_html += f'
{value}
'
table_html += '
'
table_html += '
'
table_html += "
"
return table_html
def get_inventory_file_timestamp():
"""Get the inventory.csv file timestamp information"""
try:
if not os.path.exists(INVENTORY_CSV_PATH):
return "N/A", "File not found"
# Get file modification time
file_mtime = os.path.getmtime(INVENTORY_CSV_PATH)
file_time = datetime.fromtimestamp(file_mtime)
time_ago = datetime.now() - file_time
# Format timestamp
formatted_time = file_time.strftime('%m/%d %H:%M:%S')
# Format time ago nicely
if time_ago.total_seconds() < 60:
time_ago_str = f"{int(time_ago.total_seconds())}s ago"
elif time_ago.total_seconds() < 3600:
time_ago_str = f"{int(time_ago.total_seconds()/60)}m ago"
else:
time_ago_str = f"{int(time_ago.total_seconds()/3600)}h ago"
return formatted_time, time_ago_str
except Exception as e:
return "Error", str(e)
def generate_error_page(message):
"""Generates an HTML page to display a critical error."""
return f"""
A critical error occurred:{message}
"""
def main():
"""Main function to perform comparison and generate HTML."""
results_html = ""
# Get inventory file timestamp
inventory_timestamp, inventory_age = get_inventory_file_timestamp()
miners_data, miners_error = read_csv_file(MINERS_CSV_PATH)
inventory_data, inventory_error = read_csv_file(INVENTORY_CSV_PATH)
if miners_error: results_html += generate_error_page(miners_error)
if inventory_error: results_html += generate_error_page(inventory_error)
if not results_html:
# --- 1. Create efficient lookups for inventory data ---
# Now using FULL serial numbers for primary matching
inventory_by_full_serial = {item.get('Serial', ''): item for item in inventory_data if item.get('Serial')}
inventory_by_mac = {item.get('MAC', '').upper(): item for item in inventory_data if item.get('MAC')}
# --- 2. Perform the one-way check: Online Miners vs. Inventory ---
mac_mismatches = []
serial_mismatches = []
unmatched_miners = []
incomplete_miners = []
used_inventory_items = set()
for miner in miners_data:
online_full_serial = miner.get('serial', '').strip()
online_mac = miner.get('mac', '').strip().upper()
# Check for miners with missing serial or MAC address
if not online_full_serial or not online_mac:
incomplete_miners.append({
'serial': miner.get('serial', ''),
'mac': miner.get('mac', ''),
'ip': miner.get('ip', ''),
'pod': miner.get('pod', ''),
'miner_type': miner.get('miner_type', ''),
'missing_data': ('Serial' if not online_full_serial else '') +
(' & MAC' if not online_full_serial and not online_mac else
'MAC' if not online_mac else '')
})
continue
matched = False
# Priority 1: Check for match by FULL serial number
if online_full_serial and online_full_serial in inventory_by_full_serial:
inventory_item = inventory_by_full_serial[online_full_serial]
inventory_mac = inventory_item.get('MAC', '').upper()
matched = True
if online_mac != inventory_mac:
mac_mismatches.append({
'Full Serial (Matched)': online_full_serial,
'Online MAC': online_mac,
'Inventory MAC': inventory_mac,
})
used_inventory_items.add(inventory_item.get('Serial'))
used_inventory_items.add(inventory_item.get('MAC'))
# Priority 2: If no full serial match, check for match by MAC
if not matched and online_mac in inventory_by_mac:
inventory_item = inventory_by_mac[online_mac]
if inventory_item.get('Serial') not in used_inventory_items and inventory_item.get('MAC') not in used_inventory_items:
inventory_full_serial = inventory_item.get('Serial', '')
matched = True
if online_full_serial != inventory_full_serial:
serial_mismatches.append({
'MAC Address (Matched)': online_mac,
'Online Serial': online_full_serial,
'Inventory Serial': inventory_full_serial,
})
used_inventory_items.add(inventory_item.get('Serial'))
used_inventory_items.add(inventory_item.get('MAC'))
# If still no match by either method, it's truly unmatched
if not matched:
unmatched_miners.append(miner)
# --- 3. Generate HTML Sections for All Categories ---
results_html += generate_results_table(
'MAC Mismatch (Full Serial Match Found)', mac_mismatches,
['Full Serial (Matched)', 'Online MAC', 'Inventory MAC']
)
results_html += generate_results_table(
'Full Serial Mismatch (MAC Match Found)', serial_mismatches,
['MAC Address (Matched)', 'Online Serial', 'Inventory Serial']
)
results_html += generate_results_table(
'Miners Found Online Only (No Match Found)', unmatched_miners,
['serial', 'mac', 'ip', 'pod', 'miner_type'],
include_download_button=True # Enable download button for this section
)
results_html += generate_results_table(
'Miners with Incomplete Data (Missing Serial/MAC)', incomplete_miners,
['serial', 'mac', 'ip', 'pod', 'miner_type', 'missing_data']
)
print("Content-type: text/html\n")
print(f"""
Inventory Comparison Tool