Plugin reports render; USB check-in/out audit report; kiosk drive lists
This commit is contained in:
@@ -22,9 +22,9 @@ cmmc_usb schema used here:
|
||||
|
||||
import logging
|
||||
import re
|
||||
from datetime import datetime
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
from flask import Blueprint, request
|
||||
from flask import Blueprint, request, Response
|
||||
from flask_jwt_extended import jwt_required
|
||||
|
||||
from shopdb.api import (
|
||||
@@ -1068,3 +1068,125 @@ def kiosk_checkin():
|
||||
|
||||
return success_response({'device_id': device_id},
|
||||
message='{} checked in'.format(device_id))
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Reports
|
||||
# =============================================================================
|
||||
# CMMC removable-media tracking is an audit obligation, and until now the only
|
||||
# way to answer "who had which drive, when" was to read the check-in/out log
|
||||
# page by page. This is the same question over a window: a day for "what
|
||||
# happened this shift", a month or six for a periodic review, a year or all time
|
||||
# for an audit.
|
||||
|
||||
# The windows the UI offers, in days. 0 means everything ever recorded.
|
||||
AUDITPERIODS = {'day': 1, 'month': 30, 'sixmonths': 182, 'year': 365, 'all': 0}
|
||||
|
||||
|
||||
def _auditdays():
|
||||
"""Days of history to report. Accepts ?period= or a raw ?days=."""
|
||||
period = (request.args.get('period') or '').strip().lower()
|
||||
if period in AUDITPERIODS:
|
||||
return AUDITPERIODS[period]
|
||||
days = request.args.get('days', type=int)
|
||||
return max(0, days) if days is not None else 30
|
||||
|
||||
|
||||
def _auditcsv(rows, columns, filename):
|
||||
"""CSV download; generate_csv is not on the contract surface."""
|
||||
import csv
|
||||
import io
|
||||
output = io.StringIO()
|
||||
writer = csv.writer(output)
|
||||
writer.writerow(columns)
|
||||
for row in rows:
|
||||
writer.writerow([row.get(column, '') for column in columns])
|
||||
return Response(output.getvalue(), mimetype='text/csv',
|
||||
headers={'Content-Disposition':
|
||||
'attachment; filename={}'.format(filename)})
|
||||
|
||||
|
||||
AUDITCOLUMNS = ['timestamp', 'action', 'device_id', 'device_desc', 'badge',
|
||||
'name', 'locker_location']
|
||||
|
||||
|
||||
@usb_bp.route('/reports/audit', methods=['GET'])
|
||||
@jwt_required()
|
||||
@require_permission('usb.view')
|
||||
def report_audit():
|
||||
"""Every check-out and check-in in the window, newest first.
|
||||
|
||||
Rows, not just totals: an audit asks who had a specific drive on a specific
|
||||
day, and a count cannot answer that. Summary figures ride along for the
|
||||
people who only want the shape of it.
|
||||
"""
|
||||
days = _auditdays()
|
||||
if _usb_selfhosted():
|
||||
rows = selfhosted.audit_rows(days)
|
||||
return _auditresponse(rows, days)
|
||||
|
||||
cutoff = None
|
||||
if days > 0:
|
||||
cutoff = (datetime.utcnow() - timedelta(days=days)).strftime(
|
||||
'%Y-%m-%d %H:%M:%S')
|
||||
|
||||
conn = cmmc_usb_connection()
|
||||
try:
|
||||
with conn.cursor() as cur:
|
||||
sql = ('SELECT l.timestamp, l.action, l.device_id, l.badge_number, '
|
||||
'l.locker_location, d.device_desc, '
|
||||
'u.first_name, u.last_name '
|
||||
'FROM checkinoutlog l '
|
||||
'LEFT JOIN devices d ON d.device_id = l.device_id '
|
||||
'LEFT JOIN users u ON u.badge_number = l.badge_number ')
|
||||
params = []
|
||||
if cutoff:
|
||||
sql += 'WHERE l.timestamp >= %s '
|
||||
params.append(cutoff)
|
||||
# A total order: timestamp alone ties for two events in the same
|
||||
# second and the same query then returns them in a different order
|
||||
# each run, which reads as the log being unstable.
|
||||
sql += 'ORDER BY l.timestamp DESC, l.log_id DESC'
|
||||
cur.execute(sql, tuple(params))
|
||||
rows = []
|
||||
for row in cur.fetchall() or []:
|
||||
name = '{} {}'.format(row.get('first_name') or '',
|
||||
row.get('last_name') or '').strip()
|
||||
stamp = row.get('timestamp')
|
||||
rows.append({
|
||||
'timestamp': stamp.isoformat() if hasattr(stamp, 'isoformat')
|
||||
else (stamp or ''),
|
||||
'action': row.get('action') or '',
|
||||
'device_id': row.get('device_id') or '',
|
||||
'device_desc': row.get('device_desc') or '',
|
||||
'badge': row.get('badge_number') or '',
|
||||
'name': name,
|
||||
'locker_location': row.get('locker_location') or '',
|
||||
})
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
return _auditresponse(rows, days)
|
||||
|
||||
|
||||
def _auditresponse(rows, days):
|
||||
"""One voice for both directory modes: same CSV, same JSON, same summary."""
|
||||
if request.args.get('format') == 'csv':
|
||||
return _auditcsv(rows, AUDITCOLUMNS, 'usb-audit.csv')
|
||||
return success_response({
|
||||
'columns': AUDITCOLUMNS,
|
||||
'rows': rows,
|
||||
'days': days,
|
||||
'summary': _auditsummary(rows),
|
||||
})
|
||||
|
||||
|
||||
def _auditsummary(rows):
|
||||
checkouts = sum(1 for row in rows if row['action'] == 'check-out')
|
||||
return {
|
||||
'events': len(rows),
|
||||
'checkouts': checkouts,
|
||||
'checkins': sum(1 for row in rows if row['action'] == 'check-in'),
|
||||
'devices': len({row['device_id'] for row in rows if row['device_id']}),
|
||||
'people': len({row['badge'] for row in rows if row['badge']}),
|
||||
}
|
||||
|
||||
@@ -16,7 +16,7 @@ History rows are synthesized from usbcheckouts (each row = a check-out event and
|
||||
if returned, a check-in event).
|
||||
"""
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
from shopdb.api import (
|
||||
db, success_response, error_response, ErrorCodes,
|
||||
@@ -452,3 +452,47 @@ def _kiosk_flag(key, default):
|
||||
if row is None or row.value is None or row.value == '':
|
||||
return default
|
||||
return str(row.value).strip().lower() in ('true', '1', 'yes', 'on')
|
||||
|
||||
|
||||
def audit_rows(days):
|
||||
"""Check-out / check-in events over a window (self-hosted mode).
|
||||
|
||||
One usbcheckouts row holds BOTH ends of a loan, so it expands into two
|
||||
events - the take and, when it has come back, the return. Reporting the row
|
||||
once would make a returned drive indistinguishable from one still out.
|
||||
|
||||
Rows only: the caller owns the CSV, the summary and the response shape, so
|
||||
both directory modes answer in exactly one voice.
|
||||
"""
|
||||
cutoff = None
|
||||
if days > 0:
|
||||
cutoff = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(days=days)
|
||||
|
||||
query = (db.session.query(USBCheckout, USBDevice)
|
||||
.join(USBDevice, USBDevice.usbdeviceid == USBCheckout.usbdeviceid))
|
||||
if cutoff is not None:
|
||||
# Either end of the loan can fall in the window: a drive taken before it
|
||||
# and returned inside it is a check-in the window has to show.
|
||||
query = query.filter(db.or_(USBCheckout.checkouttime >= cutoff,
|
||||
USBCheckout.checkintime >= cutoff))
|
||||
|
||||
rows = []
|
||||
for checkout, device in query.all():
|
||||
base = {
|
||||
'device_id': device.serialnumber,
|
||||
'device_desc': device.label or device.productname or '',
|
||||
'badge': checkout.sso or '',
|
||||
'name': checkout.checkoutname or '',
|
||||
'locker_location': device.storagelocation or '',
|
||||
}
|
||||
if checkout.checkouttime and (cutoff is None or checkout.checkouttime >= cutoff):
|
||||
rows.append(dict(base, timestamp=checkout.checkouttime.isoformat(),
|
||||
action='check-out'))
|
||||
if checkout.checkintime and (cutoff is None or checkout.checkintime >= cutoff):
|
||||
rows.append(dict(base, timestamp=checkout.checkintime.isoformat(),
|
||||
action='check-in'))
|
||||
|
||||
# Newest first, and a total order so two events in the same second do not
|
||||
# swap places between identical runs.
|
||||
rows.sort(key=lambda row: (row['timestamp'], row['device_id']), reverse=True)
|
||||
return rows
|
||||
|
||||
Reference in New Issue
Block a user