Files
shopdb-flask/scripts/site_imports/wjf/run.py
cproudlock 9e2544a687 wjf loader: import active machines only (isactive=1)
Classic ASP keeps retired machines in the machines table as history (isactive=0);
every other stage already filters isactive=1, but the assets hub, metrology,
locations and the verify count read machines unfiltered, so ~240 retired units
(incl. all G-prefix hostnames and 61 dead metrology PCs that each synthesized a
phantom measuring tool) landed as live assets. Add the isactive=1 filter to
those four queries. Downstream stages resolve via the id crosswalk, so warranties
/comms/relationships/installs for retired machines now drop automatically.
2026-07-13 15:00:45 -04:00

818 lines
37 KiB
Python

"""WJ classic-ASP -> flask reference import loader (site-specific).
Run against a THROWAWAY import database:
DATABASE_URL='mysql+pymysql://root:PW@127.0.0.1:3306/shopdb_flask_import?charset=utf8mb4' \\
venv/bin/python -m scripts.site_imports.wjf.run --stages reference,employees
Stages are idempotent (re-running resolves existing rows by natural key), so a
crashed run resumes. See scripts/site_imports/wjf/harness.py for the contract
this drives, and scratchpad/IMPORT-PLAN.md for the full mapping + decisions.
Implemented: reference (vendors, businessunits, operatingsystems), employees.
TODO stages (assets hub + dependents + network + usb) are stubbed - they need
the machineid->assetid crosswalk this harness persists.
"""
import argparse
from .harness import Harness
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, or None."""
status, data = h.post(path, payload)
if status in (200, 201):
return _id_of(data, idfield)
if status == 409:
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, idfield)
return None
# --- classic machinetype routing (per the resolved import decisions) ---------
# measuringtools are the physical instruments (the CMM/gauge machine types). A
# PC that DRIVES one (pctype CMM/Genspect/Keyence/Wax) is still a computer, not a
# measuring tool - routing by pctype wrongly swept ~105 PCs into measuringtools.
MEASURING_MTYPES = {3, 5, 8, 23, 47, 48} # CMM, Wax, Eddy, Measuring, Inspection, Spline
COMPUTER_MTYPES = {33, 20} # PC, Server
NETWORK_MTYPES = {16, 17, 18, 19, 46} # Access, IDF, Camera, Switch, Firewall
SKIP_MTYPES = {15, 44} # Printer (printers table), USB (cmmc source)
# LocationOnly is the islocationonly BIT, not machinetypeid=1: of the 158 type-1
# rows only 24 carry the bit (real named areas). The other 134 are active,
# modelled shop machines just left untyped - they must become assets, not
# Locations.
def _truthy_bit(value):
"""MySQL bit(1) comes back from pymysql as bytes b'\\x00'/b'\\x01'."""
return value not in (0, None, False, b'\x00', b'', '0', '')
# 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, islocationonly=False):
if islocationonly:
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:
return 'computer' # a metrology PC is still a PC (it CONTROLS a tool)
return 'machine' # includes the 134 untyped (machinetypeid=1) shop machines
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):
"""Seed the reference tier that later FKs resolve against, capturing
legacy-id -> new-id crosswalks."""
counts = {}
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', 'vendorid')
if newid:
h.ids.put('vendor', v['vendorid'], newid)
counts['vendors'] = h.ids.count('vendor')
bus = h.source.rows('shopdb_src',
'SELECT businessunitid, businessunit FROM businessunits WHERE isactive=1')
for b in bus:
newid = _upsert(h, '/api/businessunits', {'businessunit': b['businessunit']},
'businessunit', 'businessunitid')
if newid:
h.ids.put('businessunit', b['businessunitid'], newid)
counts['businessunits'] = h.ids.count('businessunit')
oses = h.source.rows('shopdb_src',
'SELECT osid, operatingsystem FROM operatingsystems')
for o in oses:
name = (o['operatingsystem'] or '').strip()
if not name:
continue
newid = _upsert(h, '/api/operatingsystems', {'osname': name}, 'osname', 'osid')
if newid:
h.ids.put('os', o['osid'], newid)
counts['operatingsystems'] = h.ids.count('os')
return counts
def stage_employees(h):
"""Bulk-upsert the employee directory (selfhosted). Photos are deferred
(per the import decisions); occurrences are out of scope."""
from shopdb.core.models import Setting
Setting.set('employee_directory_mode', 'selfhosted', valuetype='string',
category='employees')
from shopdb.extensions import db
db.session.commit()
rows = h.source.rows('wjf_employees_src',
'SELECT SSO, First_Name, Last_Name, Team, Role FROM employees')
lines = ['SSO,First_Name,Last_Name,Team,Role']
for r in rows:
vals = [str(r['SSO']), (r['First_Name'] or '').strip(),
(r['Last_Name'] or '').strip(), (r['Team'] or '').strip(),
(r['Role'] or '').strip()]
lines.append(','.join(v.replace(',', ' ') for v in vals))
status, data = h.post('/api/employees/directory/import', {'csv': '\n'.join(lines)})
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()
mtid = mt['machinetypeid']
# machinetypeid=1 (LocationOnly) is not a real machine subtype - the
# untyped shop machines that carry it import with a null subtype.
if not name or mtid == 1:
continue
# 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)."""
# isactive=1 only: retired machines (classic keeps them as history) must not
# land as live assets. Every other stage already filters isactive=1; the
# assets hub is the one that leaked retired rows (incl. G-prefix hostnames).
machines = h.source.rows('shopdb_src', 'SELECT * FROM machines WHERE isactive=1')
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())
# On a duplicate machinenumber, first-wins - so process the richer asset
# types first (a PC carries installs/IP a bare untyped machine does not).
route_rank = {'computer': 0, 'measuringtool': 1, 'network': 2, 'machine': 3}
machines = sorted(machines, key=lambda m: route_rank.get(
_route(m['machinetypeid'], m['pctypeid'], _truthy_bit(m['islocationonly'])), 4))
for m in machines:
route = _route(m['machinetypeid'], m['pctypeid'],
_truthy_bit(m['islocationonly']))
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
def stage_printers(h):
"""Printers come from the printers TABLE (not the machines hub). assetnumber
is synthesized PRN-{printerid} (no machinenumber). The printer's IP folds
into a Communication via the create route. Location resolves from the host
machineid when it is one of the LocationOnly rows. Skips inactive."""
made = 0
for p in h.source.rows('shopdb_src',
'SELECT printerid, modelid, printerwindowsname, printercsfname, '
'serialnumber, fqdn, ipaddress, machineid, maptop, mapleft, iscsf, '
'installpath, printernotes, printerpin FROM printers WHERE isactive=1'):
payload = {
'assetnumber': f"PRN-{p['printerid']}",
'name': (p['printerwindowsname'] or '').strip() or None,
'serialnumber': (p['serialnumber'] or '').strip() or None,
'hostname': (p['fqdn'] or '').strip() or None,
'windowsname': (p['printerwindowsname'] or '').strip() or None,
'sharename': (p['printercsfname'] or '').strip() or None,
'ipaddress': (p['ipaddress'] or '').strip() or None,
'iscsf': _truthy_bit(p['iscsf']),
'installpath': p['installpath'], 'pin': p['printerpin'],
'notes': p['printernotes'], 'mapx': p['mapleft'], 'mapy': p['maptop'],
'modelnumberid': h.ids.get('model', p['modelid']),
'locationid': h.ids.get('location', p['machineid']),
}
status, _ = h.post('/api/printers', payload)
if status in (200, 201):
made += 1
return {'printers': made}
# classic pctype -> the measuring instrument that PC drives
METROLOGY_PCTYPE_TOOL = {5: 'CMM', 6: 'Form Tracer', 7: 'Vision System', 8: 'Genspect'}
def stage_metrology(h):
"""Metrology PCs (CMM/Genspect/Keyence/Wax) are computers that CONTROL an
instrument. Classic has no separate tool row, so synthesize a measuring-tool
asset per metrology PC and a Controls relationship (PC -> tool), mirroring
what the runtime collector does. Runs after the asset hub (needs the PC's
assetid from the crosswalk)."""
controls_id = _upsert(h, '/api/assets/relationshiptypes',
{'relationshiptype': 'Controls'}, 'relationshiptype',
'relationshiptypeid', '/api/assets/relationshiptypes')
made = linked = 0
for m in h.source.rows('shopdb_src',
'SELECT machineid, alias, hostname, machinenumber, businessunitid, '
'mapleft, maptop, pctypeid FROM machines '
'WHERE machinetypeid=33 AND pctypeid IN (5,6,7,8) '
'AND isactive=1 '
'AND (islocationonly IS NULL OR islocationonly=0)'):
pc_assetid = h.ids.get('asset', m['machineid'])
if not pc_assetid:
continue # PC not imported (dup-skipped etc.)
toolname = METROLOGY_PCTYPE_TOOL[m['pctypeid']]
typeid = _upsert(h, '/api/measuringtools/types', {'name': toolname},
'name', 'measuringtooltypeid', '/api/measuringtools/types')
pcname = (m['alias'] or m['hostname'] or m['machinenumber'] or '').strip()
status, data = h.post('/api/measuringtools', {
'assetnumber': f"MT-{m['machineid']}",
'name': (f"{pcname} {toolname}").strip() or toolname,
'measuringtooltypeid': typeid,
'businessunitid': h.ids.get('businessunit', m['businessunitid']),
'mapx': m['mapleft'], 'mapy': m['maptop']})
if status not in (200, 201):
continue
tool_assetid = _id_of(data, 'assetid')
made += 1
if tool_assetid and controls_id:
h.ids.put('metrologytool', m['machineid'], tool_assetid)
st, _ = h.post('/api/assets/relationships', {
'sourceassetid': pc_assetid, 'targetassetid': tool_assetid,
'relationshiptypeid': controls_id})
if st in (200, 201):
linked += 1
return {'measuringtools_synth': made, 'controls_links': linked}
def stage_communications(h):
"""Fold each asset's primary IP from the source communications table. There
is no bulk-communications endpoint, so this is one of the plan's documented
direct-ORM gaps (the API only takes a primary IP on asset create/update)."""
from shopdb.core.models import Communication, CommunicationType
from shopdb.extensions import db
ip_type = CommunicationType.query.filter_by(comtype='IP').first()
rows = h.source.rows('shopdb_src',
'SELECT machineid, address FROM communications '
'WHERE comstypeid=1 AND isprimary=1')
added = 0
for r in rows:
assetid = h.ids.get('asset', r['machineid'])
ip = (r['address'] or '').strip()
if not (assetid and ip and ip_type):
continue
if Communication.query.filter_by(assetid=assetid, isprimary=True).first():
continue
db.session.add(Communication(assetid=assetid, comtypeid=ip_type.comtypeid,
ipaddress=ip, isprimary=True))
added += 1
db.session.commit()
return {'primary_ips': added}
def stage_applications(h):
"""Support teams + applications catalog + versions + installs. Skips
inactive rows (decision); dedup of colliding app names is automatic via the
unique-appname 409-resolve."""
from plugins.computers.models import Computer
counts = {}
for t in h.source.rows('shopdb_src',
'SELECT supporteamid, teamname, teamurl FROM supportteams WHERE isactive=1'):
name = (t['teamname'] or '').strip()
if not name:
continue
newid = _upsert(h, '/api/supportteams', {'teamname': name, 'teamurl': t['teamurl']},
'teamname', 'supportteamid', '/api/supportteams')
if newid:
h.ids.put('supportteam', t['supporteamid'], newid)
counts['supportteams'] = h.ids.count('supportteam')
for a in h.source.rows('shopdb_src',
'SELECT appid, appname, appdescription, supportteamid, '
'isinstallable, applicationnotes, installpath, applicationlink, '
'documentationpath FROM applications WHERE isactive=1'):
name = (a['appname'] or '').strip()
if not name:
continue
payload = {'appname': name, 'appdescription': a['appdescription'],
'supportteamid': h.ids.get('supportteam', a['supportteamid']),
'isinstallable': bool(a['isinstallable']),
'applicationnotes': a['applicationnotes'], 'installpath': a['installpath'],
'applicationlink': a['applicationlink'],
'documentationpath': a['documentationpath']}
newid = _upsert(h, '/api/applications', payload, 'appname', 'appid',
'/api/applications')
if newid:
h.ids.put('app', a['appid'], newid)
counts['applications'] = h.ids.count('app')
for v in h.source.rows('shopdb_src',
'SELECT appversionid, appid, version, releasedate, notes, dateadded '
'FROM appversions WHERE isactive=1'):
appid = h.ids.get('app', v['appid'])
ver = (v['version'] or '').strip()
if not (appid and ver):
continue
status, data = h.post(f'/api/applications/{appid}/versions',
{'version': ver, 'releasedate': str(v['releasedate']) if v['releasedate'] else None,
'notes': v['notes'], 'dateadded': str(v['dateadded']) if v['dateadded'] else None})
if status in (200, 201):
h.ids.put('appversion', v['appversionid'], _id_of(data, 'appversionid'))
counts['appversions'] = h.ids.count('appversion')
installed = 0
for i in h.source.rows('shopdb_src',
'SELECT appid, appversionid, machineid FROM installedapps WHERE isactive=1'):
assetid = h.ids.get('asset', i['machineid'])
appid = h.ids.get('app', i['appid'])
if not (assetid and appid):
continue
computer = Computer.query.filter_by(assetid=assetid).first()
if not computer:
continue
payload = {'appid': appid}
vid = h.ids.get('appversion', i['appversionid'])
if vid:
payload['appversionid'] = vid
status, _ = h.post(f'/api/computers/{computer.computerid}/apps', payload)
if status in (200, 201):
installed += 1
counts['installs'] = installed
return counts
def stage_warranties(h):
"""Warranties -> POST /api/warranty, linked to the asset via the crosswalk.
The source has no vendor; hardcode Dell (decision)."""
linked = 0
for w in h.source.rows('shopdb_src',
'SELECT machineid, warrantyname, enddate, servicelevel, dateadded '
'FROM warranties'):
assetid = h.ids.get('asset', w['machineid'])
if not assetid:
continue
payload = {'vendor': 'Dell', 'provider': 'manual',
'servicelevel': (w['servicelevel'] or '').strip() or None,
'enddate': str(w['enddate']) if w['enddate'] else None,
'assetids': [assetid],
'dateadded': str(w['dateadded']) if w['dateadded'] else None}
status, _ = h.post('/api/warranty', payload)
if status in (200, 201):
linked += 1
return {'warranties': linked}
def stage_notifications(h):
"""Notification types + notifications."""
counts = {}
for t in h.source.rows('shopdb_src',
'SELECT notificationtypeid, typename, typecolor FROM notificationtypes WHERE isactive=1'):
name = (t['typename'] or '').strip()
if not name:
continue
newid = _upsert(h, '/api/notifications/types',
{'typename': name, 'typecolor': (t['typecolor'] or '').strip() or None},
'typename', 'notificationtypeid', '/api/notifications/types')
if newid:
h.ids.put('notificationtype', t['notificationtypeid'], newid)
counts['notificationtypes'] = h.ids.count('notificationtype')
# SSO -> "First Last" from the employee source, so recognition/training
# notifications display a name, not a bare SSO (the model shows employeename
# or falls back to employeesso).
ssoname = {}
for e in h.source.rows('wjf_employees_src',
'SELECT SSO, First_Name, Last_Name FROM employees'):
full = f"{(e['First_Name'] or '').strip()} {(e['Last_Name'] or '').strip()}".strip()
if full:
ssoname[str(e['SSO']).strip()] = full
def _names(employeesso):
if not employeesso:
return None
parts = [ssoname.get(s.strip()) for s in str(employeesso).split(',') if s.strip()]
parts = [p for p in parts if p]
return ', '.join(parts) or None
made = 0
for n in h.source.rows('shopdb_src',
'SELECT notificationtypeid, businessunitid, appid, notification, '
'starttime, endtime, ticketnumber, link, isshopfloor, employeesso '
'FROM notifications WHERE isactive=1'):
text = (n['notification'] or '').strip()
if not text:
continue
endtime = str(n['endtime']) if n['endtime'] else None
if endtime and endtime.startswith('2099'):
endtime = None
payload = {'notification': text,
'notificationtypeid': h.ids.get('notificationtype', n['notificationtypeid']),
'businessunitid': h.ids.get('businessunit', n['businessunitid']),
'appid': h.ids.get('app', n['appid']),
'starttime': str(n['starttime']) if n['starttime'] else None,
'endtime': endtime, 'ticketnumber': n['ticketnumber'], 'link': n['link'],
'isshopfloor': bool(n['isshopfloor']), 'employeesso': n['employeesso'],
'employeename': _names(n['employeesso'])}
status, _ = h.post('/api/notifications', payload)
if status in (200, 201):
made += 1
counts['notifications'] = made
return counts
def stage_knowledgebase(h):
"""KB articles. appid resolves through the applications name map (topics
folded into applications)."""
made = 0
for k in h.source.rows('shopdb_src',
'SELECT shortdescription, keywords, appid, linkurl, notes '
'FROM knowledgebase WHERE isactive=1'):
title = (k['shortdescription'] or '').strip()
url = (k['linkurl'] or '').strip()
if not (title and url):
continue
payload = {'shortdescription': title, 'linkurl': url,
'keywords': k['keywords'], 'notes': k['notes'],
'appid': h.ids.get('app', k['appid'])}
status, _ = h.post('/api/knowledgebase', payload)
if status in (200, 201):
made += 1
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 AND isactive=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. A "Controlled By"
# type is the reverse of "Controls" - flag it so edges get flipped to the
# forward direction instead of importing a redundant inverse type.
reverse_typeids = set()
controls_newid = None
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
lname = name.lower()
if 'controlled by' in lname:
reverse_typeids.add(t['relationshiptypeid'])
continue # do not create an inverse type; edges map to Controls
newid = _upsert(h, '/api/assets/relationshiptypes', {'relationshiptype': name},
'relationshiptype', 'relationshiptypeid', '/api/assets/relationshiptypes')
if newid:
h.ids.put('relationshiptype', t['relationshiptypeid'], newid)
if lname == 'controls':
controls_newid = 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'])
# "Controlled By" edges flip to the forward Controls direction.
if r['relationshiptypeid'] in reverse_typeids:
source, target = target, source
typeid = controls_newid
else:
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 WHERE isactive=1', '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,
'printers': stage_printers,
'metrology': stage_metrology,
'communications': stage_communications,
'applications': stage_applications,
'warranties': stage_warranties,
'notifications': stage_notifications,
'knowledgebase': stage_knowledgebase,
'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,locations,printers,'
'metrology,communications,applications,warranties,notifications,'
'knowledgebase,relationships,subnets,usb,verify',
help='comma list of stages to run')
args = parser.parse_args()
h = Harness()
print(f'== WJ import loader (target: {h.app.config["SQLALCHEMY_DATABASE_URI"].split("/")[-1]}) ==')
try:
for name in args.stages.split(','):
name = name.strip()
if name not in STAGES:
print(f' skip unknown stage: {name}')
continue
result = STAGES[name](h)
print(f' [{name}] {result}')
if h.errors:
print(f'\n {len(h.errors)} endpoint error(s):')
for path, code, _payload, data in h.errors[:10]:
print(f' {code} {path}: {data}')
finally:
h.close()
if __name__ == '__main__':
main()