Initial commit: add compliance_checks table, per-check metadata on assets, and compliance audit trail
This commit is contained in:
389
app/routes/assets.py
Normal file
389
app/routes/assets.py
Normal file
@@ -0,0 +1,389 @@
|
||||
import json
|
||||
from datetime import date, datetime
|
||||
from flask import (Blueprint, render_template, redirect, url_for,
|
||||
flash, request, current_app, jsonify)
|
||||
from flask_login import login_required, current_user
|
||||
from app.extensions import db
|
||||
from app.models.asset import Asset, ASSET_TYPES, ASSET_STATUSES
|
||||
from app.models.audit_log import AuditLog
|
||||
from app.models.compliance_check import ComplianceCheck
|
||||
from app.services.dell_service import lookup_service_tag
|
||||
|
||||
bp = Blueprint('assets', __name__, url_prefix='/assets')
|
||||
|
||||
|
||||
def _parse_date(value):
|
||||
"""Convert a 'YYYY-MM-DD' string to a date object; return None if blank."""
|
||||
if not value:
|
||||
return None
|
||||
if isinstance(value, date):
|
||||
return value
|
||||
try:
|
||||
return date.fromisoformat(value.strip())
|
||||
except (ValueError, AttributeError):
|
||||
return None
|
||||
|
||||
|
||||
def _log(action, record_id, description, old=None, new=None):
|
||||
entry = AuditLog(
|
||||
table_name='assets',
|
||||
record_id=record_id,
|
||||
action=action,
|
||||
old_values=json.dumps(old) if old else None,
|
||||
new_values=json.dumps(new) if new else None,
|
||||
performed_by_id=current_user.id,
|
||||
ip_address=request.remote_addr,
|
||||
description=description,
|
||||
)
|
||||
db.session.add(entry)
|
||||
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# List
|
||||
# ------------------------------------------------------------------
|
||||
@bp.route('/')
|
||||
@login_required
|
||||
def index():
|
||||
page = request.args.get('page', 1, type=int)
|
||||
q = request.args.get('q', '').strip()
|
||||
status_filter = request.args.get('status', '')
|
||||
type_filter = request.args.get('asset_type', '')
|
||||
|
||||
query = Asset.query
|
||||
if q:
|
||||
like = f'%{q}%'
|
||||
query = query.filter(
|
||||
db.or_(
|
||||
Asset.serial_number.like(like),
|
||||
Asset.service_tag.like(like),
|
||||
Asset.asset_tag.like(like),
|
||||
Asset.brand.like(like),
|
||||
Asset.model.like(like),
|
||||
)
|
||||
)
|
||||
if status_filter:
|
||||
query = query.filter_by(status=status_filter)
|
||||
if type_filter:
|
||||
query = query.filter_by(asset_type=type_filter)
|
||||
|
||||
pagination = query.order_by(Asset.created_at.desc()).paginate(
|
||||
page=page, per_page=current_app.config['ITEMS_PER_PAGE'], error_out=False
|
||||
)
|
||||
return render_template(
|
||||
'assets/index.html',
|
||||
pagination=pagination, q=q,
|
||||
status_filter=status_filter, type_filter=type_filter,
|
||||
asset_types=ASSET_TYPES, asset_statuses=ASSET_STATUSES,
|
||||
)
|
||||
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Create
|
||||
# ------------------------------------------------------------------
|
||||
@bp.route('/new', methods=['GET', 'POST'])
|
||||
@login_required
|
||||
def create():
|
||||
if request.method == 'POST':
|
||||
sn = request.form.get('serial_number', '').strip()
|
||||
if not sn:
|
||||
flash('Serial Number is required.', 'danger')
|
||||
return render_template('assets/form.html', asset=None,
|
||||
asset_types=ASSET_TYPES, asset_statuses=ASSET_STATUSES)
|
||||
|
||||
if Asset.query.filter_by(serial_number=sn).first():
|
||||
flash(f'An asset with serial number {sn} already exists.', 'danger')
|
||||
return render_template('assets/form.html', asset=None,
|
||||
asset_types=ASSET_TYPES, asset_statuses=ASSET_STATUSES)
|
||||
|
||||
service_tag = request.form.get('service_tag', '').strip() or None
|
||||
if service_tag and Asset.query.filter_by(service_tag=service_tag).first():
|
||||
flash(f'An asset with service tag {service_tag} already exists.', 'danger')
|
||||
return render_template('assets/form.html', asset=None,
|
||||
asset_types=ASSET_TYPES, asset_statuses=ASSET_STATUSES)
|
||||
|
||||
asset = Asset(
|
||||
serial_number=sn,
|
||||
service_tag=service_tag,
|
||||
asset_tag=request.form.get('asset_tag', '').strip() or None,
|
||||
asset_type=request.form.get('asset_type', 'Laptop'),
|
||||
brand=request.form.get('brand', '').strip() or None,
|
||||
model=request.form.get('model', '').strip() or None,
|
||||
processor=request.form.get('processor', '').strip() or None,
|
||||
ram_gb=request.form.get('ram_gb', type=int),
|
||||
storage_gb=request.form.get('storage_gb', type=int),
|
||||
operating_system=request.form.get('operating_system', '').strip() or None,
|
||||
mac_address=request.form.get('mac_address', '').strip() or None,
|
||||
purchase_date=_parse_date(request.form.get('purchase_date')),
|
||||
warranty_expiry=_parse_date(request.form.get('warranty_expiry')),
|
||||
purchase_price=request.form.get('purchase_price', type=float),
|
||||
supplier=request.form.get('supplier', '').strip() or None,
|
||||
po_number=request.form.get('po_number', '').strip() or None,
|
||||
status=request.form.get('status', 'available'),
|
||||
location=request.form.get('location', '').strip() or None,
|
||||
notes=request.form.get('notes', '').strip() or None,
|
||||
created_by_id=current_user.id,
|
||||
)
|
||||
db.session.add(asset)
|
||||
db.session.flush()
|
||||
_log('create', asset.id, f'Created asset SN={sn}',
|
||||
new={'serial_number': sn, 'asset_type': asset.asset_type})
|
||||
db.session.commit()
|
||||
flash(f'Asset {sn} created.', 'success')
|
||||
return redirect(url_for('assets.detail', asset_id=asset.id))
|
||||
|
||||
# Pre-fill values from Dell lookup (passed as query string params)
|
||||
prefill = {
|
||||
'service_tag': request.args.get('service_tag', ''),
|
||||
'serial_number': request.args.get('serial_number', ''),
|
||||
'brand': request.args.get('brand', ''),
|
||||
'model': request.args.get('model', ''),
|
||||
'asset_type': request.args.get('asset_type', ''),
|
||||
'operating_system': request.args.get('operating_system', ''),
|
||||
'warranty_expiry': request.args.get('warranty_expiry', ''),
|
||||
'purchase_date': request.args.get('purchase_date', ''),
|
||||
}
|
||||
return render_template('assets/form.html', asset=None,
|
||||
asset_types=ASSET_TYPES, asset_statuses=ASSET_STATUSES,
|
||||
prefill=prefill)
|
||||
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Dell service-tag lookup (AJAX)
|
||||
# ------------------------------------------------------------------
|
||||
@bp.route('/dell-lookup')
|
||||
@login_required
|
||||
def dell_lookup():
|
||||
tag = request.args.get('tag', '').strip()
|
||||
if not tag:
|
||||
return jsonify({'error': 'Service tag is required.'}), 400
|
||||
|
||||
# Check for duplicates first
|
||||
existing = Asset.query.filter_by(service_tag=tag.upper()).first()
|
||||
if existing:
|
||||
return jsonify({
|
||||
'error': f'An asset with service tag {tag.upper()} already exists.',
|
||||
'existing_id': existing.id,
|
||||
}), 409
|
||||
|
||||
try:
|
||||
data = lookup_service_tag(tag)
|
||||
return jsonify(data)
|
||||
except RuntimeError as exc:
|
||||
return jsonify({'error': str(exc)}), 502
|
||||
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Detail
|
||||
# ------------------------------------------------------------------
|
||||
@bp.route('/<int:asset_id>')
|
||||
@login_required
|
||||
def detail(asset_id):
|
||||
asset = Asset.query.get_or_404(asset_id)
|
||||
history = asset.assignments.order_by(db.text('assigned_date DESC')).all()
|
||||
docs = asset.paperwork_docs.order_by(db.text('created_at DESC')).all()
|
||||
compliance_log = (
|
||||
AuditLog.query
|
||||
.filter_by(table_name='assets', record_id=asset_id, action='compliance_update')
|
||||
.order_by(AuditLog.performed_at.desc())
|
||||
.all()
|
||||
)
|
||||
check_history = (
|
||||
ComplianceCheck.query
|
||||
.filter_by(asset_id=asset_id)
|
||||
.order_by(ComplianceCheck.performed_at.desc())
|
||||
.all()
|
||||
)
|
||||
return render_template('assets/detail.html', asset=asset, history=history,
|
||||
docs=docs, compliance_log=compliance_log,
|
||||
check_history=check_history)
|
||||
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Edit
|
||||
# ------------------------------------------------------------------
|
||||
@bp.route('/<int:asset_id>/edit', methods=['GET', 'POST'])
|
||||
@login_required
|
||||
def edit(asset_id):
|
||||
asset = Asset.query.get_or_404(asset_id)
|
||||
|
||||
if request.method == 'POST':
|
||||
old = {'serial_number': asset.serial_number, 'status': asset.status}
|
||||
|
||||
new_sn = request.form.get('serial_number', '').strip()
|
||||
if not new_sn:
|
||||
flash('Serial Number is required.', 'danger')
|
||||
return render_template('assets/form.html', asset=asset,
|
||||
asset_types=ASSET_TYPES, asset_statuses=ASSET_STATUSES)
|
||||
|
||||
# Check uniqueness only if SN changed
|
||||
if new_sn != asset.serial_number:
|
||||
if Asset.query.filter(Asset.serial_number == new_sn, Asset.id != asset_id).first():
|
||||
flash(f'Serial number {new_sn} is already used by another asset.', 'danger')
|
||||
return render_template('assets/form.html', asset=asset,
|
||||
asset_types=ASSET_TYPES, asset_statuses=ASSET_STATUSES)
|
||||
|
||||
asset.serial_number = new_sn
|
||||
asset.service_tag = request.form.get('service_tag', '').strip() or None
|
||||
asset.asset_tag = request.form.get('asset_tag', '').strip() or None
|
||||
asset.asset_type = request.form.get('asset_type', asset.asset_type)
|
||||
asset.brand = request.form.get('brand', '').strip() or None
|
||||
asset.model = request.form.get('model', '').strip() or None
|
||||
asset.processor = request.form.get('processor', '').strip() or None
|
||||
asset.ram_gb = request.form.get('ram_gb', type=int)
|
||||
asset.storage_gb = request.form.get('storage_gb', type=int)
|
||||
asset.operating_system = request.form.get('operating_system', '').strip() or None
|
||||
asset.mac_address = request.form.get('mac_address', '').strip() or None
|
||||
asset.purchase_date = _parse_date(request.form.get('purchase_date'))
|
||||
asset.warranty_expiry = _parse_date(request.form.get('warranty_expiry'))
|
||||
asset.purchase_price = request.form.get('purchase_price', type=float)
|
||||
asset.supplier = request.form.get('supplier', '').strip() or None
|
||||
asset.po_number = request.form.get('po_number', '').strip() or None
|
||||
asset.status = request.form.get('status', asset.status)
|
||||
asset.location = request.form.get('location', '').strip() or None
|
||||
asset.notes = request.form.get('notes', '').strip() or None
|
||||
|
||||
_log('update', asset.id, f'Updated asset SN={asset.serial_number}',
|
||||
old=old, new={'serial_number': asset.serial_number, 'status': asset.status})
|
||||
db.session.commit()
|
||||
flash('Asset updated.', 'success')
|
||||
return redirect(url_for('assets.detail', asset_id=asset_id))
|
||||
|
||||
return render_template('assets/form.html', asset=asset,
|
||||
asset_types=ASSET_TYPES, asset_statuses=ASSET_STATUSES)
|
||||
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Compliance update (Desktop / Laptop specific fields)
|
||||
# ------------------------------------------------------------------
|
||||
_COMPLIANCE_FIELDS = {
|
||||
'inventory_number': 'Inventory Number',
|
||||
'ad_device_name': 'AD Device Name',
|
||||
'location_note': 'Location Note',
|
||||
'encryption_checked': 'Encryption Checked',
|
||||
'backup_checked': 'Backup Checked',
|
||||
'hr_notified': 'HR Notified',
|
||||
}
|
||||
|
||||
@bp.route('/<int:asset_id>/compliance', methods=['POST'])
|
||||
@login_required
|
||||
def update_compliance(asset_id):
|
||||
asset = Asset.query.get_or_404(asset_id)
|
||||
|
||||
old = {
|
||||
'inventory_number': asset.inventory_number,
|
||||
'ad_device_name': asset.ad_device_name,
|
||||
'location_note': asset.location_note,
|
||||
'encryption_checked': asset.encryption_checked,
|
||||
'backup_checked': asset.backup_checked,
|
||||
'hr_notified': asset.hr_notified,
|
||||
}
|
||||
|
||||
asset.inventory_number = request.form.get('inventory_number', '').strip() or None
|
||||
asset.ad_device_name = request.form.get('ad_device_name', '').strip() or None
|
||||
asset.location_note = request.form.get('location_note', '').strip() or None
|
||||
|
||||
new_encryption = bool(request.form.get('encryption_checked'))
|
||||
new_backup = bool(request.form.get('backup_checked'))
|
||||
new_hr = bool(request.form.get('hr_notified'))
|
||||
|
||||
notes = request.form.get('compliance_notes', '').strip() or None
|
||||
now = datetime.utcnow()
|
||||
|
||||
# Record a ComplianceCheck event for each boolean that changed
|
||||
_check_map = [
|
||||
('encryption', 'encryption_checked', new_encryption,
|
||||
'encryption_checked_by_id', 'encryption_checked_at'),
|
||||
('backup', 'backup_checked', new_backup,
|
||||
'backup_checked_by_id', 'backup_checked_at'),
|
||||
('hr', 'hr_notified', new_hr,
|
||||
'hr_notified_by_id', 'hr_notified_at'),
|
||||
]
|
||||
for check_type, field, new_val, by_field, at_field in _check_map:
|
||||
if old[field] != new_val:
|
||||
setattr(asset, field, new_val)
|
||||
setattr(asset, by_field, current_user.id)
|
||||
setattr(asset, at_field, now)
|
||||
db.session.add(ComplianceCheck(
|
||||
asset_id=asset_id,
|
||||
check_type=check_type,
|
||||
checked=new_val,
|
||||
performed_by_id=current_user.id,
|
||||
performed_at=now,
|
||||
notes=notes,
|
||||
))
|
||||
|
||||
new = {
|
||||
'inventory_number': asset.inventory_number,
|
||||
'ad_device_name': asset.ad_device_name,
|
||||
'location_note': asset.location_note,
|
||||
'encryption_checked': asset.encryption_checked,
|
||||
'backup_checked': asset.backup_checked,
|
||||
'hr_notified': asset.hr_notified,
|
||||
}
|
||||
|
||||
# Build human-readable description of what changed
|
||||
changed = [
|
||||
f'{_COMPLIANCE_FIELDS[k]}: {repr(old[k])} → {repr(new[k])}'
|
||||
for k in _COMPLIANCE_FIELDS if old[k] != new[k]
|
||||
]
|
||||
if changed:
|
||||
_log('compliance_update', asset.id,
|
||||
f'Compliance updated: {"; ".join(changed)}',
|
||||
old=old, new=new)
|
||||
db.session.commit()
|
||||
flash('Compliance fields updated.', 'success')
|
||||
else:
|
||||
flash('No changes detected.', 'info')
|
||||
|
||||
return redirect(url_for('assets.detail', asset_id=asset_id))
|
||||
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Quick lookup by SN or service tag (AJAX)
|
||||
# ------------------------------------------------------------------
|
||||
@bp.route('/lookup')
|
||||
@login_required
|
||||
def lookup():
|
||||
q = request.args.get('q', '').strip()
|
||||
if not q:
|
||||
return jsonify(None)
|
||||
asset = Asset.query.filter(
|
||||
db.or_(Asset.serial_number == q, Asset.service_tag == q)
|
||||
).first()
|
||||
if not asset:
|
||||
return jsonify(None)
|
||||
return jsonify({
|
||||
'id': asset.id,
|
||||
'serial_number': asset.serial_number,
|
||||
'service_tag': asset.service_tag,
|
||||
'brand': asset.brand,
|
||||
'model': asset.model,
|
||||
'asset_type': asset.asset_type,
|
||||
'status': asset.status,
|
||||
})
|
||||
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Search (AJAX dropdown)
|
||||
# ------------------------------------------------------------------
|
||||
@bp.route('/search')
|
||||
@login_required
|
||||
def search():
|
||||
q = request.args.get('q', '').strip()
|
||||
if len(q) < 2:
|
||||
return jsonify([])
|
||||
like = f'%{q}%'
|
||||
assets = Asset.query.filter(
|
||||
db.or_(
|
||||
Asset.serial_number.like(like),
|
||||
Asset.service_tag.like(like),
|
||||
Asset.asset_tag.like(like),
|
||||
)
|
||||
).limit(15).all()
|
||||
return jsonify([{
|
||||
'id': a.id,
|
||||
'text': f'{a.brand or ""} {a.model or ""} — SN: {a.serial_number}'.strip(' —'),
|
||||
'serial_number': a.serial_number,
|
||||
'service_tag': a.service_tag,
|
||||
'status': a.status,
|
||||
} for a in assets])
|
||||
Reference in New Issue
Block a user