236 lines
9.1 KiB
Python
236 lines
9.1 KiB
Python
"""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]
|