390 lines
16 KiB
Python
390 lines
16 KiB
Python
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])
|