ge-aerospace/shopdb-flask
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
2d09fa32019153df0ef66666a90ee3a8dbb01d28
shopdb-flask/scripts/site_imports/wjf/run.py
cproudlock 2d09fa3201
Some checks failed
CI / backend (push) Failing after 7m15s
Details
CI / naming (push) Failing after 7m22s
Details
CI / frontend (push) Failing after 7m14s
Details
CI / migrations-mysql (push) Failing after 7m14s
Details
Collect what bays actually have, separately from what they are told to have 
ShopDB knew what a bay SHOULD have and nothing about what it DOES. Adding the
observed half makes a rollout a review instead of a typing exercise: the floor
reports itself in, you look, and you adopt.

The collection uses the mechanism that already exists rather than a new one.
POST /api/collector/printers dispatches to the printers plugin's
apply_collector_payload, the same ADR-006 hook the computers and backups plugins
implement. New client script, new plugin-owned table, no new transport and no new
credential.

OBSERVED AND ASSIGNED STAY APART, and that is the point rather than a detail. A
collector report can never write an assignment row: _reconcile_edges is the only
function that writes usesprinter/defaultprinter, it has two call sites, and both
are authenticated routes a human calls. If a drifted bay's own state were allowed
to become what it is told to install, every configuration error would become
permanent the next time that PC checked in.

Seeding an assignment from observed state is explicit -
POST /assignments/seed-from-observed - because a rollout adopts many machines at
once. It routes through the same _reconcile_edges as the editor, so there is one
write path with two doors, and a queue matching no known printer is REFUSED
rather than guessed into an assignment. That last rule is the lesson from the
measuring tools: adopting on a weak key produced 43 duplicate instruments.

Two fixes on top of what the agents built. The replace deleted a host's previous
rows by exact case-folded name while the read path treats a short name and its
FQDN as one machine, so a PC that changed spelling appeared to hold every queue
twice - which reads as drift that is not there. And the client sent 'reportedat'
where the declared schema said 'observedat'.

Also here: the legacy loader now imports machines.printerid, the classic system's
record of each machine's default printer, which it silently dropped - the
production import would have lost every one. And Set-ShopdbPrinters.ps1 finally
registers the per-user logon task, staging Apply-ShopdbDefaultPrinter.ps1 to
C:\ProgramData first because the share it lives on is mounted only during the
enforcement cycle and the task runs at logon when it is gone.

VALIDATED ON WINDOWS 11 (build 26200), not just on Linux pwsh, which parses these
scripts happily and executes none of the spooler branches.

The reporter: posts a correct payload with the X-API-Key header; resolves BaseUrl
and CollectorKey from HKLM when given no arguments; suppresses the virtual queues
by port; resolves port addresses; and reads the CONSOLE USER's default out of
HKU rather than SYSTEM's own, which is a different and usually wrong answer.

Two results matter more than the rest. With the spooler stopped, both the cmdlet
and the CIM path fail and the script posts NOTHING - verified against a capture
server that recorded zero requests, where an empty list would instead have
erased that host's observed rows and read as a bay that lost its printers. A
genuinely empty host still posts [], because that is a real and different fact.

The logon task registers as the Users group at Limited, and falls back to the
well-known SID S-1-5-32-545 when the group name will not resolve, as it will not
on localised Windows. It was then run with the source directory RENAMED AWAY, to
stand in for the share being unmounted, and it still moved the user's default -
which is the whole reason the script is staged to C:\ProgramData rather than run
from where it lives.

The guarantees against damage were re-checked rather than assumed: an empty
assignment changes nothing, an unreachable server changes nothing, -WhatIfOnly
leaves no queue, no task, no staged file and no registry value behind, and a
drifted queue is repointed IN PLACE with Set-Printer so whoever has it as their
default keeps it.

Not covered by any of this: the driver-staging path, which needs a real vendor
package rather than the class drivers a VM ships with.
2026-08-19 15:32:18 -04:00

931 lines
42 KiB
Python
Raw Blame History

"""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
import pymysql
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
def _put(h, path, payload):
"""PUT through the harness client. The harness wraps post/get only, so the
reconcile-style assignment route goes direct with the same import-mode
headers. Returns (status, data)."""
headers = {'Authorization': f'Bearer {h.secret}', 'X-Import-Mode': 'true'}
resp = h.client.put(path, json=payload, headers=headers)
body = resp.get_json() or {}
data = body.get('data', body)
if resp.status_code >= 400:
h.errors.append((path, resp.status_code, payload, data))
return resp.status_code, data
# --- 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'],
'levelid': h.defaultlevelid,
'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'],
'levelid': h.defaultlevelid,
'modelnumberid': h.ids.get('model', p['modelid']),
'locationid': h.ids.get('location', p['machineid']),
}
status, data = h.post('/api/printers', payload)
assetid = None
if status in (200, 201):
assetid = _id_of(data, 'assetid')
made += 1
elif status == 409:
# Re-run: PRN-<printerid> is already an asset. Resolve it anyway -
# without this branch a resumed run crosswalks nothing and the
# defaultprinters stage silently links nothing.
_, rows = h.get(f"/api/printers?assetnumber={payload['assetnumber']}&per_page=5")
items = rows.get('items', rows) if isinstance(rows, dict) else rows
for row in (items or []):
if str(row.get('assetnumber', '')).strip().lower() == \
payload['assetnumber'].lower():
assetid = _id_of(row, 'assetid')
break
if assetid:
h.ids.put('printer', p['printerid'], assetid)
return {'printers': made, 'crosswalked': h.ids.count('printer')}
# 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'],
'levelid': h.defaultlevelid})
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': _truthy_bit(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). Employee source is optional: without it
# (shopdb-only import) notifications keep the bare SSO.
ssoname = {}
try:
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
except pymysql.err.OperationalError:
print(' (no wjf_employees_src - notifications keep bare SSOs)')
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': _truthy_bit(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_defaultprinters(h):
"""classic machines.printerid -> a printer assignment on the imported asset.
Classic records exactly one default printer per machine row. It goes through
the assignment reconcile route rather than raw relationship posts: the
usesprinter + defaultprinter pairing, printer-type validation, the
default-must-be-in-the-set rule and the soft-delete/reactivate semantics all
live there, and reconciling converges on a re-run instead of duplicating.
Runs after assets and printers - it resolves both ends through their
crosswalks and drops the row when either end did not import. Needs the
usesprinter/defaultprinter relationship types seeded (flask seed
reference-data): the route writes nothing without them and every row fails.
"""
# printerid=0 means no printer recorded, not a dangling FK - excluded so it
# never lands in the drop count as something someone has to explain.
rows = h.source.rows('shopdb_src',
'SELECT machineid, printerid FROM machines '
'WHERE isactive=1 AND printerid IS NOT NULL AND printerid>0')
# Legacy names for the drop report. Most machines point at a retired
# placeholder printer, and a bare "dropped 560" reads like data loss.
printernames = {row['printerid']: (row['printerwindowsname'] or '').strip()
for row in h.source.rows(
'shopdb_src',
'SELECT printerid, printerwindowsname FROM printers')}
counts = {'source_rows': len(rows), 'linked': 0, 'already': 0,
'dropped_no_asset': 0, 'dropped_printer_not_imported': 0,
'failed': 0}
dropped_printers = {}
for r in rows:
assetid = h.ids.get('asset', r['machineid'])
if not assetid:
# Machine became a Location, was the 9999 placeholder, a duplicate
# machinenumber, or a skipped type.
counts['dropped_no_asset'] += 1
continue
printerassetid = h.ids.get('printer', r['printerid'])
if not printerassetid:
counts['dropped_printer_not_imported'] += 1
label = '{0} {1}'.format(r['printerid'],
printernames.get(r['printerid'], '?'))
dropped_printers[label] = dropped_printers.get(label, 0) + 1
continue
# Read first: the PUT reconciles the WHOLE set, so a blind write would
# unassign anything else already on this asset. Matching state is left
# alone so a second run is a no-op, not a rewrite.
_, current = h.get(f'/api/printers/assignments/for-asset/{assetid}')
current = current if isinstance(current, dict) else {}
assigned = list(current.get('printerassetids') or [])
if (current.get('defaultprinterassetid') == printerassetid
and printerassetid in assigned):
counts['already'] += 1
continue
if printerassetid not in assigned:
assigned.append(printerassetid)
status, _ = _put(h, f'/api/printers/assignments/for-asset/{assetid}',
{'printerassetids': assigned,
'defaultprinterassetid': printerassetid})
if status in (200, 201):
counts['linked'] += 1
else:
counts['failed'] += 1
if dropped_printers:
counts['dropped_printers'] = dict(
sorted(dropped_printers.items(), key=lambda item: -item[1])[:10])
return counts
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:
try:
src = h.source.rows(sdb, sql)[0]['c']
except pymysql.err.OperationalError:
report[label] = 'source db absent - skipped'
continue
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,
'defaultprinters': stage_defaultprinters,
'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,defaultprinters,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()
Reference in New Issue View Git Blame Copy Permalink