WJ import loader: catalog + assets-hub stages (the machineid->assetid crosswalk)

Adds the keystone stages to the reference loader.

catalog: seeds modeltypes (from classic machinetypes + category), the per-plugin
asset subtypes routed by machinetype (machines/network/measuringtool types),
computer subtypes (from pctype), the 5-row controllertypes vendor/model split,
and the models catalog - each with a persisted legacy->new crosswalk. Verified:
modeltypes 31, computertypes 12, models 118.

assets (the hub): fans classic machines out to the right endpoint by
machinetypeid (+ the pctype metrology override), applying the resolved
decisions - assetnumber = machinenumber else hostname, skip 9999, skip duplicate
machinenumbers, LocationOnly/printer/USB routed out. Persists the machineid ->
assetid crosswalk every downstream stage needs. Verified against a fresh scratch
target: 884 assets (computer 623, machine 68, network 58, measuringtool 135),
zero endpoint errors, idempotent re-run (stays 884). Skips: location 158, dup 69,
other 53, 9999 1.

Harness now runs each plugin's idempotent on_install so the AssetType rows exist
(a DB built with plugin upgrade-all instead of a fresh install lacks them, and
the create routes 500 without them). 409-resolve lookups page through per_page.

Remaining stages: communications (primary IP fold - source is the communications
table, not machines.ipaddress1), applications/installs, warranties, notifications,
KB, subnets/VLANs, usb, verify.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

scripts/site_imports/wjf/harness.py

@@ -94,6 +94,7 @@ class Harness:
         self.appcontext.push()
         self._silence_sql_logging()
         self.client = self.app.test_client()
+        self._ensure_asset_types()
         self.secret = self._mint_admin_pat()
         default = os.path.join(os.path.dirname(__file__), 'idmap.json')
         self.ids = IdMap(idmap_path or default)
@@ -104,6 +105,21 @@ class Harness:
         import logging
         logging.getLogger('sqlalchemy.engine').setLevel(logging.WARNING)
 
+    def _ensure_asset_types(self):
+        """Register each plugin's AssetType + seeded subtypes. on_install (which
+        does this) is skipped when a DB is built with `flask plugin upgrade-all`
+        rather than a fresh `install`, so run it idempotently here - the create
+        routes 500 without the AssetType row."""
+        manager = self.app.extensions.get('plugin_manager')
+        if not manager:
+            return
+        for _name, plugin in manager.get_all_plugins().items():
+            try:
+                plugin.on_install(self.app)
+            except Exception:
+                pass
+        db.session.commit()
+
     def _mint_admin_pat(self):
         """Create (or reuse) an admin user and an unscoped admin PAT. Unscoped
         because import mode + role gates require full owner authority

scripts/site_imports/wjf/run.py

@@ -19,27 +19,72 @@ import argparse
 from .harness import Harness
 
 
-def _upsert(h, path, payload, unique_field, list_path=None):
+def _id_of(row, idfield):
+    if not isinstance(row, dict):
+        return None
+    if row.get(idfield) is not None:
+        return row[idfield]
+    # asset-create responses nest the id at the top level (assetid) too
+    for key in (idfield, 'id', 'assetid'):
+        if row.get(key) is not None:
+            return row[key]
+    return None
+
+
+def _upsert(h, path, payload, unique_field, idfield, list_path=None):
     """POST payload; on 409 resolve the existing row by its unique field.
-    Returns the new/existing id (best-effort key 'id' variants), or None."""
+    Returns the new/existing id, or None."""
     status, data = h.post(path, payload)
     if status in (200, 201):
-        return _id_of(data)
+        return _id_of(data, idfield)
     if status == 409:
-        _, rows = h.get(list_path or path)
+        lookup = list_path or path
+        lookup += ('&' if '?' in lookup else '?') + 'per_page=10000'
+        _, rows = h.get(lookup)
         items = rows if isinstance(rows, list) else rows.get('items', rows)
         for row in (items or []):
             if str(row.get(unique_field, '')).strip().lower() == \
                     str(payload[unique_field]).strip().lower():
-                return _id_of(row)
+                return _id_of(row, idfield)
     return None
 
 
-def _id_of(row):
-    for key in ('vendorid', 'businessunitid', 'osid', 'id'):
-        if isinstance(row, dict) and row.get(key) is not None:
-            return row[key]
-    return None
+# --- classic machinetype routing (per the resolved import decisions) ---------
+MEASURING_MTYPES = {3, 5, 8, 23, 47, 48}      # CMM, Wax, Eddy, Measuring, Inspection, Spline
+MEASURING_PCTYPES = {5, 6, 7, 8}              # PC subtypes: CMM, Wax/Trace, Keyence, Genspect
+COMPUTER_MTYPES = {33, 20}                     # PC, Server
+NETWORK_MTYPES = {16, 17, 18, 19, 46}          # Access, IDF, Camera, Switch, Firewall
+LOCATION_MTYPES = {1}                          # LocationOnly -> core Locations
+SKIP_MTYPES = {15, 44}                          # Printer (printers table), USB (cmmc source)
+# classic controllertypeid -> (vendor, model) hand split; 1=TBD dropped
+CONTROLLER_SPLIT = {2: ('Fanuc', '31i-MB'), 6: ('Fanuc', None),
+                    7: ('Okuma', None), 8: ('Makino', None)}
+
+
+def _route(machinetypeid, pctypeid):
+    if machinetypeid in LOCATION_MTYPES:
+        return 'location'
+    if machinetypeid in SKIP_MTYPES:
+        return 'skip'
+    if machinetypeid in NETWORK_MTYPES:
+        return 'network'
+    if machinetypeid in MEASURING_MTYPES:
+        return 'measuringtool'
+    if machinetypeid in COMPUTER_MTYPES:
+        if machinetypeid == 33 and pctypeid in MEASURING_PCTYPES:
+            return 'measuringtool'
+        return 'computer'
+    return 'machine'
+
+
+def _modeltype_category(machinetypeid):
+    if machinetypeid in COMPUTER_MTYPES:
+        return 'PC'
+    if machinetypeid in NETWORK_MTYPES:
+        return 'Network'
+    if machinetypeid == 15:
+        return 'Printer'
+    return 'Equipment'
 
 
 def stage_reference(h):
@@ -50,7 +95,7 @@ def stage_reference(h):
     vendors = h.source.rows('shopdb_src',
                             'SELECT vendorid, vendor FROM vendors WHERE isactive=1')
     for v in vendors:
-        newid = _upsert(h, '/api/vendors', {'vendor': v['vendor']}, 'vendor')
+        newid = _upsert(h, '/api/vendors', {'vendor': v['vendor']}, 'vendor', 'vendorid')
         if newid:
             h.ids.put('vendor', v['vendorid'], newid)
     counts['vendors'] = h.ids.count('vendor')
@@ -59,7 +104,7 @@ def stage_reference(h):
                         'SELECT businessunitid, businessunit FROM businessunits WHERE isactive=1')
     for b in bus:
         newid = _upsert(h, '/api/businessunits', {'businessunit': b['businessunit']},
-                        'businessunit')
+                        'businessunit', 'businessunitid')
         if newid:
             h.ids.put('businessunit', b['businessunitid'], newid)
     counts['businessunits'] = h.ids.count('businessunit')
@@ -70,7 +115,7 @@ def stage_reference(h):
         name = (o['operatingsystem'] or '').strip()
         if not name:
             continue
-        newid = _upsert(h, '/api/operatingsystems', {'osname': name}, 'osname')
+        newid = _upsert(h, '/api/operatingsystems', {'osname': name}, 'osname', 'osid')
         if newid:
             h.ids.put('os', o['osid'], newid)
     counts['operatingsystems'] = h.ids.count('os')
@@ -99,16 +144,176 @@ def stage_employees(h):
     return {'employees_posted': len(rows), 'result': data}
 
 
+def stage_catalog(h):
+    """Seed the type catalogs the asset routing + models depend on, capturing
+    crosswalks: modeltypes + per-plugin subtypes (from classic machinetypes),
+    computer subtypes (from pctype), controller vendor/model split, models."""
+    counts = {}
+    mtypes = h.source.rows('shopdb_src',
+                          'SELECT machinetypeid, machinetype FROM machinetypes')
+    for mt in mtypes:
+        name = (mt['machinetype'] or '').strip()
+        if not name:
+            continue
+        mtid = mt['machinetypeid']
+        # modeltype (types the models catalog)
+        modeltypeid = _upsert(h, '/api/modeltypes',
+                              {'modeltype': name, 'category': _modeltype_category(mtid)},
+                              'modeltype', 'modeltypeid')
+        if modeltypeid:
+            h.ids.put('modeltype', mtid, modeltypeid)
+        # per-plugin asset subtype, by route
+        route = _route(mtid, None)
+        if route == 'machine':
+            sid = _upsert(h, '/api/machines/types', {'machinetype': name},
+                          'machinetype', 'machinetypeid', '/api/machines/types')
+            if sid:
+                h.ids.put('machinesubtype', mtid, sid)
+        elif route == 'network':
+            sid = _upsert(h, '/api/network/types', {'networkdevicetype': name},
+                          'networkdevicetype', 'networkdevicetypeid', '/api/network/types')
+            if sid:
+                h.ids.put('networksubtype', mtid, sid)
+        elif route == 'measuringtool':
+            sid = _upsert(h, '/api/measuringtools/types', {'name': name},
+                          'name', 'measuringtooltypeid', '/api/measuringtools/types')
+            if sid:
+                h.ids.put('measuringsubtype', mtid, sid)
+    counts['modeltypes'] = h.ids.count('modeltype')
+
+    # computer subtypes from pctype
+    for pc in h.source.rows('shopdb_src', 'SELECT pctypeid, typename FROM pctype'):
+        name = (pc['typename'] or '').strip()
+        if not name:
+            continue
+        sid = _upsert(h, '/api/computers/types', {'computertype': name},
+                      'computertype', 'computertypeid', '/api/computers/types')
+        if sid:
+            h.ids.put('computertype', pc['pctypeid'], sid)
+    counts['computertypes'] = h.ids.count('computertype')
+
+    # controller split -> vendor + model, crosswalk controllertypeid -> both ids
+    for ctid, (vendor, model) in CONTROLLER_SPLIT.items():
+        vid = _upsert(h, '/api/vendors', {'vendor': vendor}, 'vendor', 'vendorid')
+        mid = None
+        if model and vid:
+            mid = _upsert(h, '/api/models', {'modelnumber': model, 'vendorid': vid},
+                          'modelnumber', 'modelnumberid')
+        h.ids.put('controllervendor', ctid, vid)
+        if mid:
+            h.ids.put('controllermodel', ctid, mid)
+
+    # models catalog (the only vendor source for machines)
+    for m in h.source.rows('shopdb_src',
+                          'SELECT modelnumberid, modelnumber, vendorid, machinetypeid '
+                          'FROM models WHERE isactive=1'):
+        number = (m['modelnumber'] or '').strip()
+        if not number:
+            continue
+        payload = {'modelnumber': number,
+                   'vendorid': h.ids.get('vendor', m['vendorid']),
+                   'modeltypeid': h.ids.get('modeltype', m['machinetypeid'])}
+        newid = _upsert(h, '/api/models', payload, 'modelnumber', 'modelnumberid')
+        if newid:
+            h.ids.put('model', m['modelnumberid'], newid)
+    counts['models'] = h.ids.count('model')
+    return counts
+
+
+def stage_assets(h):
+    """The hub: fan classic machines out to the right asset endpoint by type,
+    persisting the machineid -> assetid crosswalk everything downstream needs.
+    LocationOnly/printer/usb rows are skipped here (handled elsewhere / TODO)."""
+    machines = h.source.rows('shopdb_src', 'SELECT * FROM machines')
+    seen_assetnumbers = set()
+    counts = {'computer': 0, 'machine': 0, 'network': 0, 'measuringtool': 0,
+              'skip_location': 0, 'skip_other': 0, 'skip_dup': 0, 'skip_9999': 0}
+
+    # Pre-seed assetnumbers already in the target (idempotent re-run).
+    for route in ('computers', 'machines', 'network', 'measuringtools'):
+        _, rows = h.get(f'/api/{route}?per_page=5000')
+        for row in (rows.get('items', rows) if isinstance(rows, dict) else rows) or []:
+            if isinstance(row, dict) and row.get('assetnumber'):
+                seen_assetnumbers.add(row['assetnumber'].strip().lower())
+
+    for m in machines:
+        route = _route(m['machinetypeid'], m['pctypeid'])
+        if route == 'location':
+            counts['skip_location'] += 1
+            continue
+        if route == 'skip':
+            counts['skip_other'] += 1
+            continue
+
+        raw = (m['machinenumber'] or '').strip()
+        if raw == '9999':
+            counts['skip_9999'] += 1
+            continue
+        assetnumber = raw or (m['hostname'] or '').strip()
+        if not assetnumber:
+            counts['skip_other'] += 1
+            continue
+        key = assetnumber.lower()
+        if key in seen_assetnumbers:
+            counts['skip_dup'] += 1
+            continue
+        seen_assetnumbers.add(key)
+
+        base = {
+            'assetnumber': assetnumber,
+            'name': (m['alias'] or m['hostname'] or '').strip() or None,
+            'serialnumber': (m['serialnumber'] or '').strip() or None,
+            'businessunitid': h.ids.get('businessunit', m['businessunitid']),
+            'mapx': m['mapleft'], 'mapy': m['maptop'],
+            'notes': m['machinenotes'],
+            'dateadded': str(m['dateadded']) if m['dateadded'] else None,
+            'modifieddate': str(m['lastupdated']) if m['lastupdated'] else None,
+        }
+        if route == 'computer':
+            path = '/api/computers'
+            payload = {**base, 'hostname': (m['hostname'] or '').strip() or None,
+                       'computertypeid': h.ids.get('computertype', m['pctypeid']),
+                       'osid': h.ids.get('os', m['osid']),
+                       'ipaddress': (m['ipaddress1'] or '').strip() or None}
+        elif route == 'network':
+            path = '/api/network'
+            payload = {**base, 'hostname': (m['hostname'] or m['fqdn'] or '').strip() or None,
+                       'networkdevicetypeid': h.ids.get('networksubtype', m['machinetypeid']),
+                       'vendorid': None,
+                       'ipaddress': (m['ipaddress1'] or '').strip() or None}
+        elif route == 'measuringtool':
+            path = '/api/measuringtools'
+            payload = {**base,
+                       'measuringtooltypeid': h.ids.get('measuringsubtype', m['machinetypeid'])}
+        else:  # machine
+            path = '/api/machines'
+            payload = {**base, 'machinetypeid': h.ids.get('machinesubtype', m['machinetypeid']),
+                       'modelnumberid': h.ids.get('model', m['modelnumberid']),
+                       'vendorid': None,
+                       'controllervendorid': h.ids.get('controllervendor', m['controllertypeid']),
+                       'controllermodelid': h.ids.get('controllermodel', m['controllertypeid'])}
+
+        status, data = h.post(path, payload)
+        if status in (200, 201):
+            assetid = _id_of(data, 'assetid')
+            if assetid:
+                h.ids.put('asset', m['machineid'], assetid)
+                counts[route] += 1
+    return counts
+
+
 STAGES = {
     'reference': stage_reference,
     'employees': stage_employees,
-    # TODO: 'models', 'applications', 'assets', 'dependents', 'network', 'usb'
+    'catalog': stage_catalog,
+    'assets': stage_assets,
+    # TODO: 'applications', 'dependents', 'network-subnets', 'usb', 'verify'
 }
 
 
 def main():
     parser = argparse.ArgumentParser()
-    parser.add_argument('--stages', default='reference,employees',
+    parser.add_argument('--stages', default='reference,employees,catalog,assets',
                         help='comma list of stages to run')
     args = parser.parse_args()