WJ import loader: tail stages (locations, relationships, subnets, usb, verify)
Some checks failed
CI / backend (push) Successful in 1m38s
CI / naming (push) Successful in 1s
CI / frontend (push) Successful in 7s
CI / migrations-mysql (push) Failing after 7s

Completes the loader end to end. Verified on a fresh scratch target, zero
endpoint errors:

- locations: the 24 islocationonly rows -> core Locations (crosswalk machineid
  -> locationid).
- relationships: 93 active edges imported (206 of 299 dropped because an
  endpoint became a Location / was skipped / dedup-lost); types folded onto the
  seeded canonical set; dedup on (source,target,type).
- subnets: 37 (full CIDR reconstructed as INET_NTOA(ipstart)+suffix; VLANs
  lookup-or-create by number; 3 duplicate CIDRs first-wins-skipped).
- usb: 18 cmmc devices + 232 check-in/out events, paired with per-device open
  state so unpaired log rows do not 400. Needs the usb plugin enabled + usb
  directory mode selfhosted.
- verify: source-vs-target row-count audit (assets 1167->933 by the skip rules,
  applications 121=121, employees 415=415, KB 342->341).

Full pipeline default runs all 14 stages in order. NOTE: the import target needs
every bundled plugin enabled (usb ships disabled in this dev registry - enable
it before importing).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
cproudlock
2026-07-13 12:05:33 -04:00
parent 40a89b360a
commit 99ad6c0ebb

View File

@@ -495,24 +495,187 @@ def stage_knowledgebase(h):
return {'kb_articles': made}
def stage_locations(h):
"""The 24 islocationonly rows -> core Locations. Crosswalk machineid ->
locationid (relationships can point at them)."""
made = 0
for m in h.source.rows('shopdb_src',
'SELECT machineid, machinenumber, alias FROM machines '
'WHERE machinetypeid=1 AND islocationonly=1'):
name = (m['alias'] or m['machinenumber'] or '').strip()
if not name:
continue
newid = _upsert(h, '/api/locations', {'locationname': name},
'locationname', 'locationid', '/api/locations')
if newid:
h.ids.put('location', m['machineid'], newid)
made += 1
return {'locations': made}
def stage_relationships(h):
"""Active machinerelationships -> asset relationships via the crosswalk.
Edges whose endpoints did not become assets (locations/skipped/dups) drop.
Dedups on the (source, target, type) triple."""
# Seed relationship types (case-insensitive collation folds Controls onto
# the seeded 'controls', etc), capturing the crosswalk.
for t in h.source.rows('shopdb_src',
'SELECT relationshiptypeid, relationshiptype FROM relationshiptypes WHERE isactive=1'):
name = (t['relationshiptype'] or '').strip()
if not name:
continue
newid = _upsert(h, '/api/assets/relationshiptypes', {'relationshiptype': name},
'relationshiptype', 'relationshiptypeid', '/api/assets/relationshiptypes')
if newid:
h.ids.put('relationshiptype', t['relationshiptypeid'], newid)
seen = set()
made = dropped = 0
for r in h.source.rows('shopdb_src',
'SELECT machineid, related_machineid, relationshiptypeid '
'FROM machinerelationships WHERE isactive=1'):
source = h.ids.get('asset', r['machineid'])
target = h.ids.get('asset', r['related_machineid'])
typeid = h.ids.get('relationshiptype', r['relationshiptypeid'])
if not (source and target and typeid) or source == target:
dropped += 1
continue
key = (source, target, typeid)
if key in seen:
continue
seen.add(key)
status, _ = h.post('/api/assets/relationships',
{'sourceassetid': source, 'targetassetid': target,
'relationshiptypeid': typeid})
if status in (200, 201):
made += 1
return {'relationships': made, 'dropped_unresolved': dropped}
def stage_subnets(h):
"""Subnets + VLANs. Classic cidr is the suffix only; full CIDR =
INET_NTOA(ipstart)+suffix. VLANs are lookup-or-create by number; duplicate
full CIDRs are first-wins (409 skip)."""
import socket
import struct
subtypes = {s['subnettypeid']: s['subnettype'] for s in
h.source.rows('shopdb_src', 'SELECT subnettypeid, subnettype FROM subnettypes')}
made = 0
for s in h.source.rows('shopdb_src',
'SELECT vlan, description, ipstart, cidr, subnettypeid FROM subnets WHERE isactive=1'):
if s['ipstart'] is None or not s['cidr']:
continue
network = socket.inet_ntoa(struct.pack('>I', s['ipstart'] & 0xffffffff))
suffix = str(s['cidr']).strip()
cidr = network + (suffix if suffix.startswith('/') else '/' + suffix)
vlanid = None
if s['vlan']:
vlanid = _upsert(h, '/api/network/vlans',
{'vlannumber': s['vlan'], 'name': f"VLAN {s['vlan']}"},
'vlannumber', 'vlanid', '/api/network/vlans')
name = (s['description'] or '').strip() or cidr
payload = {'name': name, 'cidr': cidr, 'networkaddress': network,
'vlanid': vlanid, 'subnettype': subtypes.get(s['subnettypeid'])}
status, _ = h.post('/api/network/subnets', payload)
if status in (200, 201):
made += 1
return {'subnets': made}
def stage_usb(h):
"""CMMC USB kiosk: devices + the check-in/out log paired into checkouts.
Requires selfhosted mode. cmmc users are directory data, not usb rows."""
from shopdb.core.models import Setting
from shopdb.extensions import db
Setting.set('usb_directory_mode', 'selfhosted', valuetype='string', category='usb')
db.session.commit()
devices = 0
for d in h.source.rows('cmmc_usb_src', 'SELECT device_id, device_desc, device_owner, locker_location FROM devices'):
did = (d['device_id'] or '').strip()
if not did or did.lower() in ('test', 'test123'):
continue
status, _ = h.post('/api/usb', {'device_id': did, 'device_desc': d['device_desc'],
'device_owner': d['device_owner'],
'locker_location': d['locker_location']})
if status in (200, 201, 409):
devices += 1
# Replay the log chronologically per device, tracking open state so we only
# check out an available device and check in an out one (the source log has
# unpaired entries - a checkin with no open checkout would 400).
events = 0
checked_out = {}
log = h.source.rows('cmmc_usb_src',
'SELECT device_id, badge_number, action, timestamp, locker_location '
'FROM checkinoutlog ORDER BY device_id, timestamp')
for e in log:
did = (e['device_id'] or '').strip()
badge = str(e['badge_number'] or '').strip()
action = (e['action'] or '').strip().lower()
if not (did and badge):
continue
when = str(e['timestamp']) if e['timestamp'] else None
if 'out' in action and not checked_out.get(did):
status, _ = h.post(f'/api/usb/{did}/checkout',
{'badge': badge, 'checkouttime': when,
'locker_location': e['locker_location']})
if status in (200, 201):
checked_out[did] = True
events += 1
elif 'in' in action and checked_out.get(did):
status, _ = h.post(f'/api/usb/{did}/checkin', {'badge': badge, 'checkintime': when})
if status in (200, 201):
checked_out[did] = False
events += 1
return {'usb_devices': devices, 'usb_events': events}
def stage_verify(h):
"""Row-count audit: source active count vs target count for each entity."""
checks = [
('vendors', 'shopdb_src', 'SELECT COUNT(*) c FROM vendors WHERE isactive=1', 'vendors'),
('assets', 'shopdb_src', 'SELECT COUNT(*) c FROM machines', 'assets'),
('applications', 'shopdb_src', 'SELECT COUNT(*) c FROM applications WHERE isactive=1', 'applications'),
('knowledgebase', 'shopdb_src', 'SELECT COUNT(*) c FROM knowledgebase WHERE isactive=1', 'knowledgebase'),
('employees', 'wjf_employees_src', 'SELECT COUNT(*) c FROM employees', 'directoryemployees'),
]
from shopdb.extensions import db
from sqlalchemy import text
report = {}
for label, sdb, sql, target_table in checks:
src = h.source.rows(sdb, sql)[0]['c']
tgt = db.session.execute(text(f'SELECT COUNT(*) FROM {target_table}')).scalar()
report[label] = f'source~{src} target={tgt}'
return report
STAGES = {
'reference': stage_reference,
'employees': stage_employees,
'catalog': stage_catalog,
'assets': stage_assets,
'locations': stage_locations,
'communications': stage_communications,
'applications': stage_applications,
'warranties': stage_warranties,
'notifications': stage_notifications,
'knowledgebase': stage_knowledgebase,
# TODO: 'locations', 'relationships', 'subnets', 'usb', 'verify'
'relationships': stage_relationships,
'subnets': stage_subnets,
'usb': stage_usb,
'verify': stage_verify,
}
def main():
parser = argparse.ArgumentParser()
parser.add_argument('--stages', default='reference,employees,catalog,assets',
help='comma list of stages to run')
parser.add_argument(
'--stages',
default='reference,employees,catalog,assets,locations,communications,'
'applications,warranties,notifications,knowledgebase,'
'relationships,subnets,usb,verify',
help='comma list of stages to run')
args = parser.parse_args()
h = Harness()