Initial commit: enterprise digital platform with portal SSO, DigiServer, IT Assets, NetworkView, Server Monitor
This commit is contained in:
ske087
@@ -0,0 +1,365 @@
|
||||
"""
|
||||
WMT management web routes – global settings, device registry, update requests.
|
||||
"""
|
||||
import csv
|
||||
import io
|
||||
from flask import Blueprint, render_template, request, redirect, url_for, flash, Response
|
||||
from datetime import datetime
|
||||
from app.models import WMTGlobalConfig, Device, WMTUpdateRequest
|
||||
from config.database_config import get_db
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
wmt_web_bp = Blueprint('wmt_web', __name__, url_prefix='/wmt')
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _get_or_create_global_config(session):
|
||||
cfg = session.query(WMTGlobalConfig).first()
|
||||
if cfg is None:
|
||||
cfg = WMTGlobalConfig()
|
||||
session.add(cfg)
|
||||
session.flush()
|
||||
return cfg
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Dashboard
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@wmt_web_bp.route('/')
|
||||
def index():
|
||||
"""WMT management dashboard."""
|
||||
try:
|
||||
with get_db().get_session() as session:
|
||||
global_cfg = _get_or_create_global_config(session)
|
||||
devices = session.query(Device).filter(Device.mac_address.isnot(None)).order_by(Device.nume_masa).all()
|
||||
pending_count = session.query(WMTUpdateRequest).filter_by(status='pending').count()
|
||||
recent_requests = (
|
||||
session.query(WMTUpdateRequest)
|
||||
.order_by(WMTUpdateRequest.submitted_at.desc())
|
||||
.limit(5)
|
||||
.all()
|
||||
)
|
||||
return render_template(
|
||||
'wmt/index.html',
|
||||
global_cfg=global_cfg,
|
||||
devices=devices,
|
||||
pending_count=pending_count,
|
||||
recent_requests=recent_requests,
|
||||
breadcrumbs=[{'url': url_for('wmt_web.index'), 'title': 'WMT Management'}],
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f'WMT dashboard error: {e}')
|
||||
flash(f'Error loading dashboard: {e}', 'error')
|
||||
return render_template('wmt/index.html', global_cfg=None, devices=[],
|
||||
pending_count=0, recent_requests=[])
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Global settings
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@wmt_web_bp.route('/settings', methods=['GET', 'POST'])
|
||||
def settings():
|
||||
"""View and edit global WMT configuration."""
|
||||
try:
|
||||
with get_db().get_session() as session:
|
||||
cfg = _get_or_create_global_config(session)
|
||||
|
||||
if request.method == 'POST':
|
||||
cfg.chrome_url = request.form.get('chrome_url', '').strip()
|
||||
cfg.chrome_local_url = request.form.get('chrome_local_url', '').strip() or None
|
||||
cfg.chrome_insecure_origin = request.form.get('chrome_insecure_origin', '').strip()
|
||||
cfg.card_api_base_url = request.form.get('card_api_base_url', '').strip()
|
||||
cfg.server_log_url = request.form.get('server_log_url', '').strip()
|
||||
cfg.internet_check_host = request.form.get('internet_check_host', '').strip()
|
||||
cfg.update_host = request.form.get('update_host', '').strip()
|
||||
cfg.update_user = request.form.get('update_user', '').strip()
|
||||
cfg.notes = request.form.get('notes', '').strip() or None
|
||||
cfg.updated_at = datetime.utcnow()
|
||||
cfg.updated_by = 'admin'
|
||||
flash('Global settings saved.', 'success')
|
||||
return redirect(url_for('wmt_web.settings'))
|
||||
|
||||
return render_template(
|
||||
'wmt/settings.html',
|
||||
cfg=cfg,
|
||||
breadcrumbs=[
|
||||
{'url': url_for('wmt_web.index'), 'title': 'WMT Management'},
|
||||
{'url': url_for('wmt_web.settings'), 'title': 'Global Settings'},
|
||||
],
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f'WMT settings error: {e}')
|
||||
flash(f'Error: {e}', 'error')
|
||||
return redirect(url_for('wmt_web.index'))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Update requests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@wmt_web_bp.route('/requests')
|
||||
def update_requests():
|
||||
"""List all device update requests."""
|
||||
status_filter = request.args.get('status', 'pending')
|
||||
try:
|
||||
with get_db().get_session() as session:
|
||||
query = session.query(WMTUpdateRequest)
|
||||
if status_filter != 'all':
|
||||
query = query.filter_by(status=status_filter)
|
||||
req_list = query.order_by(WMTUpdateRequest.submitted_at.desc()).all()
|
||||
pending_count = session.query(WMTUpdateRequest).filter_by(status='pending').count()
|
||||
|
||||
return render_template(
|
||||
'wmt/requests.html',
|
||||
requests=req_list,
|
||||
status_filter=status_filter,
|
||||
pending_count=pending_count,
|
||||
breadcrumbs=[
|
||||
{'url': url_for('wmt_web.index'), 'title': 'WMT Management'},
|
||||
{'url': url_for('wmt_web.update_requests'), 'title': 'Update Requests'},
|
||||
],
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f'WMT requests list error: {e}')
|
||||
flash(f'Error: {e}', 'error')
|
||||
return redirect(url_for('wmt_web.index'))
|
||||
|
||||
|
||||
@wmt_web_bp.route('/requests/<int:req_id>/accept', methods=['POST'])
|
||||
def accept_request(req_id):
|
||||
"""Accept an update request: apply proposed values to WMTDevice."""
|
||||
try:
|
||||
with get_db().get_session() as session:
|
||||
req = session.query(WMTUpdateRequest).filter_by(id=req_id).first()
|
||||
if not req:
|
||||
flash('Request not found.', 'error')
|
||||
return redirect(url_for('wmt_web.update_requests'))
|
||||
|
||||
# Find or create the Device
|
||||
device = session.query(Device).filter_by(mac_address=req.mac_address).first()
|
||||
if device is None:
|
||||
device = Device(
|
||||
mac_address=req.mac_address,
|
||||
hostname=req.proposed_hostname or '',
|
||||
device_ip=req.proposed_device_ip or '',
|
||||
nume_masa=req.proposed_device_name or '',
|
||||
)
|
||||
session.add(device)
|
||||
session.flush()
|
||||
req.device_id = device.id
|
||||
|
||||
# Apply proposed values
|
||||
if req.proposed_device_name is not None:
|
||||
device.nume_masa = req.proposed_device_name
|
||||
if req.proposed_hostname is not None:
|
||||
device.hostname = req.proposed_hostname
|
||||
if req.proposed_device_ip is not None:
|
||||
device.device_ip = req.proposed_device_ip
|
||||
device.config_updated_at = datetime.utcnow()
|
||||
device.info_reviewed_at = datetime.utcnow() # admin reviewed → push timestamp to devices
|
||||
|
||||
# Mark request as accepted
|
||||
req.status = 'accepted'
|
||||
req.admin_reviewed_at = datetime.utcnow()
|
||||
req.admin_notes = request.form.get('admin_notes', '').strip() or None
|
||||
|
||||
flash('Request accepted and device record updated.', 'success')
|
||||
except Exception as e:
|
||||
logger.error(f'WMT accept request error: {e}')
|
||||
flash(f'Error accepting request: {e}', 'error')
|
||||
return redirect(url_for('wmt_web.update_requests'))
|
||||
|
||||
|
||||
@wmt_web_bp.route('/requests/<int:req_id>/reject', methods=['POST'])
|
||||
def reject_request(req_id):
|
||||
"""Reject an update request (updates reviewed_at so WMT client won't re-submit)."""
|
||||
try:
|
||||
with get_db().get_session() as session:
|
||||
req = session.query(WMTUpdateRequest).filter_by(id=req_id).first()
|
||||
if not req:
|
||||
flash('Request not found.', 'error')
|
||||
return redirect(url_for('wmt_web.update_requests'))
|
||||
|
||||
req.status = 'rejected'
|
||||
req.admin_reviewed_at = datetime.utcnow()
|
||||
req.admin_notes = request.form.get('admin_notes', '').strip() or None
|
||||
|
||||
# Update device info_reviewed_at even though data didn't change –
|
||||
# this signals to the WMT client that the server has reviewed the state
|
||||
# so it won't keep re-submitting the same request.
|
||||
if req.device_id:
|
||||
device = session.query(Device).filter_by(id=req.device_id).first()
|
||||
if device:
|
||||
device.info_reviewed_at = datetime.utcnow()
|
||||
|
||||
flash('Request rejected.', 'warning')
|
||||
except Exception as e:
|
||||
logger.error(f'WMT reject request error: {e}')
|
||||
flash(f'Error rejecting request: {e}', 'error')
|
||||
return redirect(url_for('wmt_web.update_requests'))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Device registry
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@wmt_web_bp.route('/devices')
|
||||
def devices():
|
||||
"""List all WMT-registered devices."""
|
||||
try:
|
||||
with get_db().get_session() as session:
|
||||
device_list = (
|
||||
session.query(Device)
|
||||
.filter(Device.mac_address.isnot(None))
|
||||
.order_by(Device.nume_masa)
|
||||
.all()
|
||||
)
|
||||
return render_template(
|
||||
'wmt/devices.html',
|
||||
devices=device_list,
|
||||
breadcrumbs=[
|
||||
{'url': url_for('wmt_web.index'), 'title': 'WMT Management'},
|
||||
{'url': url_for('wmt_web.devices'), 'title': 'Devices'},
|
||||
],
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f'WMT devices list error: {e}')
|
||||
flash(f'Error: {e}', 'error')
|
||||
return redirect(url_for('wmt_web.index'))
|
||||
|
||||
|
||||
@wmt_web_bp.route('/devices/new', methods=['GET', 'POST'])
|
||||
def device_new():
|
||||
"""Register a new WMT device manually."""
|
||||
if request.method == 'POST':
|
||||
mac = (request.form.get('mac_address') or '').strip().lower()
|
||||
if not mac:
|
||||
flash('MAC address is required.', 'error')
|
||||
return render_template('wmt/device_form.html', device=None)
|
||||
try:
|
||||
with get_db().get_session() as session:
|
||||
existing = session.query(Device).filter_by(mac_address=mac).first()
|
||||
if existing:
|
||||
flash(f'Device with MAC {mac} already exists.', 'error')
|
||||
return render_template('wmt/device_form.html', device=None)
|
||||
device = Device(
|
||||
mac_address=mac,
|
||||
nume_masa=request.form.get('device_name', '').strip(),
|
||||
hostname=request.form.get('hostname', '').strip(),
|
||||
device_ip=request.form.get('device_ip', '').strip() or '127.0.0.1',
|
||||
location=request.form.get('location', '').strip() or None,
|
||||
card_presence=request.form.get('card_presence', 'enable'),
|
||||
description=request.form.get('notes', '').strip() or None,
|
||||
info_reviewed_at=datetime.utcnow(),
|
||||
)
|
||||
session.add(device)
|
||||
flash('Device registered.', 'success')
|
||||
return redirect(url_for('wmt_web.devices'))
|
||||
except Exception as e:
|
||||
logger.error(f'WMT device_new error: {e}')
|
||||
flash(f'Error: {e}', 'error')
|
||||
return render_template('wmt/device_form.html', device=None)
|
||||
|
||||
return render_template(
|
||||
'wmt/device_form.html',
|
||||
device=None,
|
||||
breadcrumbs=[
|
||||
{'url': url_for('wmt_web.index'), 'title': 'WMT Management'},
|
||||
{'url': url_for('wmt_web.devices'), 'title': 'Devices'},
|
||||
{'url': url_for('wmt_web.device_new'), 'title': 'New Device'},
|
||||
],
|
||||
)
|
||||
|
||||
|
||||
@wmt_web_bp.route('/devices/<int:device_id>/edit', methods=['GET', 'POST'])
|
||||
def device_edit(device_id):
|
||||
"""Edit an existing WMT device."""
|
||||
try:
|
||||
with get_db().get_session() as session:
|
||||
device = session.query(Device).filter_by(id=device_id).first()
|
||||
if not device:
|
||||
flash('Device not found.', 'error')
|
||||
return redirect(url_for('wmt_web.devices'))
|
||||
|
||||
if request.method == 'POST':
|
||||
device.nume_masa = request.form.get('device_name', '').strip()
|
||||
device.hostname = request.form.get('hostname', '').strip()
|
||||
device.device_ip = request.form.get('device_ip', '').strip() or device.device_ip
|
||||
device.location = request.form.get('location', '').strip() or None
|
||||
device.card_presence = request.form.get('card_presence', 'enable')
|
||||
device.description = request.form.get('notes', '').strip() or None
|
||||
device.config_updated_at = datetime.utcnow()
|
||||
device.info_reviewed_at = datetime.utcnow()
|
||||
flash('Device updated.', 'success')
|
||||
return redirect(url_for('wmt_web.devices'))
|
||||
|
||||
return render_template(
|
||||
'wmt/device_form.html',
|
||||
device=device,
|
||||
breadcrumbs=[
|
||||
{'url': url_for('wmt_web.index'), 'title': 'WMT Management'},
|
||||
{'url': url_for('wmt_web.devices'), 'title': 'Devices'},
|
||||
{'url': url_for('wmt_web.device_edit', device_id=device_id), 'title': 'Edit'},
|
||||
],
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f'WMT device_edit error: {e}')
|
||||
flash(f'Error: {e}', 'error')
|
||||
return redirect(url_for('wmt_web.devices'))
|
||||
|
||||
|
||||
@wmt_web_bp.route('/devices/export.csv')
|
||||
def devices_export_csv():
|
||||
"""Export all WMT devices as a CSV file."""
|
||||
try:
|
||||
with get_db().get_session() as session:
|
||||
device_list = (
|
||||
session.query(Device)
|
||||
.filter(Device.mac_address.isnot(None))
|
||||
.order_by(Device.nume_masa)
|
||||
.all()
|
||||
)
|
||||
output = io.StringIO()
|
||||
writer = csv.writer(output)
|
||||
writer.writerow(['hostname', 'mac_address', 'work_place', 'rfid_card_status'])
|
||||
for d in device_list:
|
||||
writer.writerow([
|
||||
d.hostname or '',
|
||||
d.mac_address or '',
|
||||
d.nume_masa or '',
|
||||
d.card_presence or 'enable',
|
||||
])
|
||||
csv_data = output.getvalue()
|
||||
return Response(
|
||||
csv_data,
|
||||
mimetype='text/csv',
|
||||
headers={'Content-Disposition': 'attachment; filename=wmt_devices.csv'},
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f'WMT devices_export_csv error: {e}')
|
||||
flash(f'Export error: {e}', 'error')
|
||||
return redirect(url_for('wmt_web.devices'))
|
||||
|
||||
|
||||
@wmt_web_bp.route('/devices/<int:device_id>/delete', methods=['POST'])
|
||||
def device_delete(device_id):
|
||||
"""Delete a WMT device."""
|
||||
try:
|
||||
with get_db().get_session() as session:
|
||||
device = session.query(Device).filter_by(id=device_id).first()
|
||||
if device:
|
||||
session.delete(device)
|
||||
flash('Device deleted.', 'success')
|
||||
else:
|
||||
flash('Device not found.', 'error')
|
||||
except Exception as e:
|
||||
logger.error(f'WMT device_delete error: {e}')
|
||||
flash(f'Error: {e}', 'error')
|
||||
return redirect(url_for('wmt_web.devices'))
|
||||
Reference in New Issue
Block a user