366 lines
15 KiB
Python
366 lines
15 KiB
Python
"""
|
||
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'))
|