USB external import, and filter chips on health-dot cards
This commit is contained in:
235
plugins/usb/cli.py
Normal file
235
plugins/usb/cli.py
Normal file
@@ -0,0 +1,235 @@
|
||||
"""CLI: move USB records off an external cmmc_usb database and into ShopDB.
|
||||
|
||||
flask usb import-external report what would happen, change nothing
|
||||
flask usb import-external --commit write it
|
||||
|
||||
A site running the external mode keeps its devices and its check-in/out history
|
||||
in a separate cmmc_usb database. This reads that database through the same
|
||||
connection settings the routes already use, converts it to the self-hosted
|
||||
tables, and leaves the source untouched. After a clean run, set
|
||||
usb_directory_mode to selfhosted and the external database can be retired.
|
||||
|
||||
TWO THINGS THIS HAS TO RECONCILE, and they are the reason it is a command
|
||||
rather than a few INSERTs:
|
||||
|
||||
The two schemas disagree about shape. External is an EVENT LOG - one row per
|
||||
action, check-out or check-in - while self-hosted keeps a PAIRED ROW, one
|
||||
checkout carrying both its times. Pairing is per device and in time order, and
|
||||
real logs are not tidy: a check-out with no return is an open checkout, but a
|
||||
return with no matching check-out, or two check-outs in a row, mean something
|
||||
was recorded wrong years ago. Those are reported rather than guessed at, and
|
||||
they never silently vanish.
|
||||
|
||||
Identity is the other one. usbdevices.serialnumber is unique and NOT NULL, and
|
||||
the external devices table has no serial column - only device_id. So device_id
|
||||
becomes the serial. A blank or duplicated device_id stops the import for that
|
||||
device instead of inventing a key.
|
||||
|
||||
Dry run by default. The point of the first run is the report.
|
||||
"""
|
||||
|
||||
from collections import defaultdict
|
||||
|
||||
import click
|
||||
from flask.cli import AppGroup, with_appcontext
|
||||
|
||||
from shopdb.api import db, cmmc_usb_connection
|
||||
|
||||
from .models import USBDevice, USBCheckout
|
||||
|
||||
|
||||
usb_cli = AppGroup('usb', help='USB device records.')
|
||||
|
||||
|
||||
# The external status vocabulary, mapped to the self-hosted boolean. Anything
|
||||
# unrecognised is treated as checked in and reported, rather than assumed.
|
||||
_CHECKED_OUT = {'checked-out', 'checked out', 'out', 'issued'}
|
||||
_RETIRED = {'retired', 'disposed', 'destroyed'}
|
||||
|
||||
|
||||
def _external_rows():
|
||||
"""(devices, log) from the external database, oldest log entry first."""
|
||||
conn = cmmc_usb_connection()
|
||||
try:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute('SELECT device_id, device_desc, device_owner, status, '
|
||||
'locker_location FROM devices')
|
||||
devices = cur.fetchall()
|
||||
cur.execute('SELECT log_id, badge_number, device_id, action, timestamp, '
|
||||
'scanned_viruses, locker_location, sanitized '
|
||||
'FROM checkinoutlog ORDER BY device_id, timestamp, log_id')
|
||||
log = cur.fetchall()
|
||||
finally:
|
||||
conn.close()
|
||||
return devices, log
|
||||
|
||||
|
||||
def _is_checkout(action):
|
||||
return str(action or '').strip().lower().replace('_', '-') in (
|
||||
'check-out', 'checkout', 'out')
|
||||
|
||||
|
||||
def _is_checkin(action):
|
||||
return str(action or '').strip().lower().replace('_', '-') in (
|
||||
'check-in', 'checkin', 'in', 'return')
|
||||
|
||||
|
||||
def _pair(entries, problems, deviceid):
|
||||
"""Walk one device's log in time order and fold it into checkout pairs."""
|
||||
pairs, open_row = [], None
|
||||
for row in entries:
|
||||
action = row.get('action')
|
||||
if _is_checkout(action):
|
||||
if open_row is not None:
|
||||
# Two check-outs with no return between them. The first is
|
||||
# unclosed and stays that way; say so rather than inventing a
|
||||
# return time that nobody recorded.
|
||||
problems.append(
|
||||
'device %s: check-out at %s while %s was still out - first '
|
||||
'left open' % (deviceid, row.get('timestamp'),
|
||||
open_row.get('badge_number')))
|
||||
pairs.append((open_row, None))
|
||||
open_row = row
|
||||
elif _is_checkin(action):
|
||||
if open_row is None:
|
||||
problems.append(
|
||||
'device %s: check-in at %s with no matching check-out - '
|
||||
'skipped' % (deviceid, row.get('timestamp')))
|
||||
continue
|
||||
pairs.append((open_row, row))
|
||||
open_row = None
|
||||
else:
|
||||
problems.append('device %s: unrecognised action %r at %s - skipped'
|
||||
% (deviceid, action, row.get('timestamp')))
|
||||
if open_row is not None:
|
||||
pairs.append((open_row, None))
|
||||
return pairs
|
||||
|
||||
|
||||
@usb_cli.command('import-external')
|
||||
@click.option('--commit', is_flag=True, default=False,
|
||||
help='Write the records. Without this the command only reports.')
|
||||
@with_appcontext
|
||||
def import_external(commit):
|
||||
"""Copy devices and check-in/out history from the external cmmc_usb DB."""
|
||||
try:
|
||||
devices, log = _external_rows()
|
||||
except Exception as error:
|
||||
raise click.ClickException(
|
||||
'Could not read the external database: %s\n'
|
||||
'Check the cmmc_usb_db_* settings, and that this host can reach it.'
|
||||
% error)
|
||||
|
||||
problems = []
|
||||
seen_serials = set()
|
||||
made_devices = 0
|
||||
made_checkouts = 0
|
||||
open_checkouts = 0
|
||||
|
||||
by_device = defaultdict(list)
|
||||
for row in log:
|
||||
by_device[row.get('device_id')].append(row)
|
||||
|
||||
for source in devices:
|
||||
serial = (source.get('device_id') or '').strip()
|
||||
if not serial:
|
||||
problems.append('device with no device_id - skipped, it cannot be '
|
||||
'given a serial number')
|
||||
continue
|
||||
if serial in seen_serials:
|
||||
problems.append('device_id %r appears twice - only the first is '
|
||||
'imported' % serial)
|
||||
continue
|
||||
seen_serials.add(serial)
|
||||
|
||||
status = str(source.get('status') or '').strip().lower()
|
||||
if status and status not in _CHECKED_OUT and status not in _RETIRED \
|
||||
and status not in ('checked-in', 'checked in', 'in', 'available'):
|
||||
problems.append('device %s: unrecognised status %r - treated as '
|
||||
'checked in' % (serial, source.get('status')))
|
||||
|
||||
device = USBDevice.query.filter_by(serialnumber=serial).first()
|
||||
if device is None:
|
||||
device = USBDevice(serialnumber=serial)
|
||||
db.session.add(device)
|
||||
made_devices += 1
|
||||
device.label = source.get('device_desc') or device.label
|
||||
device.storagelocation = source.get('locker_location') or device.storagelocation
|
||||
device.notes = _append_owner(device.notes, source.get('device_owner'))
|
||||
device.ischeckedout = status in _CHECKED_OUT
|
||||
|
||||
db.session.flush()
|
||||
|
||||
for out_row, in_row in _pair(by_device.get(source.get('device_id'), []),
|
||||
problems, serial):
|
||||
checkout = USBCheckout(
|
||||
usbdeviceid=device.usbdeviceid,
|
||||
sso=str(out_row.get('badge_number') or '')[:20],
|
||||
checkouttime=out_row.get('timestamp'),
|
||||
checkintime=in_row.get('timestamp') if in_row else None,
|
||||
)
|
||||
if in_row:
|
||||
checkout.waswiped = _tri(in_row.get('sanitized'))
|
||||
checkout.scannedviruses = _tri(in_row.get('scanned_viruses'))
|
||||
else:
|
||||
open_checkouts += 1
|
||||
db.session.add(checkout)
|
||||
made_checkouts += 1
|
||||
|
||||
# The device's own flag and the history have to agree, or the kiosk
|
||||
# will offer a drive that the log says somebody still holds.
|
||||
if device.ischeckedout and not _has_open(device):
|
||||
problems.append('device %s: marked checked out, but its log has no '
|
||||
'open check-out' % serial)
|
||||
|
||||
click.echo('devices read from external : %d' % len(devices))
|
||||
click.echo('log rows read : %d' % len(log))
|
||||
click.echo('devices to create : %d' % made_devices)
|
||||
click.echo('checkouts to create : %d' % made_checkouts)
|
||||
click.echo(' of which still open : %d' % open_checkouts)
|
||||
|
||||
if problems:
|
||||
click.echo('')
|
||||
click.echo('%d row(s) need a human decision:' % len(problems))
|
||||
for line in problems:
|
||||
click.echo(' - %s' % line)
|
||||
|
||||
if not commit:
|
||||
db.session.rollback()
|
||||
click.echo('')
|
||||
click.echo('DRY RUN - nothing was written. Re-run with --commit when the '
|
||||
'report above looks right.')
|
||||
return
|
||||
|
||||
db.session.commit()
|
||||
click.echo('')
|
||||
click.echo('Written. Next: set usb_directory_mode to selfhosted in Settings, '
|
||||
'confirm the kiosk and the device list against the external '
|
||||
'database, and only then retire it.')
|
||||
|
||||
|
||||
def _tri(value):
|
||||
"""External flags are 0/1/NULL. Keep NULL as None: unrecorded is not false."""
|
||||
if value is None:
|
||||
return None
|
||||
return bool(value)
|
||||
|
||||
|
||||
def _has_open(device):
|
||||
return USBCheckout.query.filter_by(usbdeviceid=device.usbdeviceid,
|
||||
checkintime=None).count() > 0
|
||||
|
||||
|
||||
def _append_owner(notes, owner):
|
||||
"""External device_owner has no column here, so keep it rather than lose it."""
|
||||
owner = (owner or '').strip()
|
||||
if not owner:
|
||||
return notes
|
||||
line = 'Imported owner: %s' % owner
|
||||
if notes and line in notes:
|
||||
return notes
|
||||
return (notes + '\n' + line) if notes else line
|
||||
|
||||
|
||||
def build_cli():
|
||||
return [usb_cli]
|
||||
Reference in New Issue
Block a user