From b11a6f26d82e9e365a2492ba5f80ab0db85efe76 Mon Sep 17 00:00:00 2001 From: cproudlock Date: Mon, 13 Jul 2026 11:20:58 -0400 Subject: [PATCH] WJ legacy-import reference loader: harness + reference/employees stages Layer 2 of the import design (see the memory + scratchpad/IMPORT-PLAN.md): a SITE-SPECIFIC reference loader that maps WJ's classic-ASP schema onto the maintained, schema-agnostic IMPORT-API contract. Other sites copy the pattern against their own source DB; nobody runs this loader as-is. Harness (scripts/site_imports/wjf/harness.py): builds the app against the current DATABASE_URL (point it at a throwaway import DB), mints an unscoped admin PAT in-process, and drives the real import endpoints through the app test client with Authorization: Bearer + X-Import-Mode - exercising the same routes/authz/validation an HTTP client would, no running server needed. Read-only pymysql access to the three scratch source DBs; legacy-id -> new-id crosswalks persist to JSON so a crashed run resumes and later stages resolve FKs. Stages implemented + verified idempotent against a fresh scratch target (shopdb_flask_import): reference (vendors 46, businessunits 13, operatingsystems 11) and employees (directory 415, re-run updated-not-duplicated). Remaining stages (models, applications, assets hub + crosswalk, dependents, network, usb, verify) are stubbed with the same shape; README documents the adoption playbook. idmap.json is generated state (gitignored). Co-Authored-By: Claude Opus 4.8 --- .gitignore | 1 + scripts/site_imports/__init__.py | 0 scripts/site_imports/wjf/README.md | 45 ++++++++ scripts/site_imports/wjf/__init__.py | 0 scripts/site_imports/wjf/harness.py | 157 +++++++++++++++++++++++++++ scripts/site_imports/wjf/run.py | 134 +++++++++++++++++++++++ 6 files changed, 337 insertions(+) create mode 100644 scripts/site_imports/__init__.py create mode 100644 scripts/site_imports/wjf/README.md create mode 100644 scripts/site_imports/wjf/__init__.py create mode 100644 scripts/site_imports/wjf/harness.py create mode 100644 scripts/site_imports/wjf/run.py diff --git a/.gitignore b/.gitignore index c946bec..9592a83 100644 --- a/.gitignore +++ b/.gitignore @@ -75,3 +75,4 @@ secrets.yml *_secret *_secrets credentials.json +scripts/site_imports/wjf/idmap.json diff --git a/scripts/site_imports/__init__.py b/scripts/site_imports/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/scripts/site_imports/wjf/README.md b/scripts/site_imports/wjf/README.md new file mode 100644 index 0000000..9ec2ad0 --- /dev/null +++ b/scripts/site_imports/wjf/README.md @@ -0,0 +1,45 @@ +# WJ classic-ASP import loader (reference implementation) + +This is **site-specific reference glue**, not product code. It maps West +Jefferson's classic-ASP `shopdb` / `cmmc_usb` / `wjf_employees` schema onto the +maintained, schema-agnostic import contract in `docs/IMPORT-API.md`. + +**Every adopting site has its own source database.** Nobody else runs this +loader. Instead, copy the pattern: + +1. Point the harness at your source DB(s) (edit `harness.Source`). +2. Write per-entity stages that read your tables and POST to the same + `docs/IMPORT-API.md` endpoints with `Authorization: Bearer ` and + `X-Import-Mode: true`. +3. Persist legacy-id -> new-id crosswalks (see `harness.IdMap`) so later stages + resolve foreign keys and a crashed run resumes. + +The import API is the stable contract; loaders are per-site. The mapping itself +can be produced with the agent-assisted workflow (point Fable/Opus agents at a +source DB + this contract -> they emit the mapping + a loader skeleton). + +## Running (against a THROWAWAY import database) + +```bash +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 +``` + +Prereqs: a fresh target DB built with `flask db upgrade` + `flask plugin +upgrade-all` + `flask seed permissions/settings/reference-data`, and the three +source dumps loaded into scratch DBs (`shopdb_src`, `cmmc_usb_src`, +`wjf_employees_src`). See `scratchpad/IMPORT-PLAN.md` for the full mapping, +resolved decisions, and remaining stages. + +## Status + +- **Implemented + verified idempotent:** `reference` (vendors, businessunits, + operatingsystems), `employees` (directory bulk upsert; photos deferred). +- **TODO stages:** `models`, `applications`, `assets` (the hub - fan machines + out by type, persist the machineid->assetid crosswalk), `dependents` + (installs, warranties, notifications, KB), `network` (+ subnets/VLANs), `usb` + (cmmc device + checkinout pairing), `verify`. + +The harness (PAT auth, import-mode, id-map persistence, endpoint error capture) +is proven; the remaining stages are additional `stage_*` functions in `run.py` +following the same shape. diff --git a/scripts/site_imports/wjf/__init__.py b/scripts/site_imports/wjf/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/scripts/site_imports/wjf/harness.py b/scripts/site_imports/wjf/harness.py new file mode 100644 index 0000000..760e360 --- /dev/null +++ b/scripts/site_imports/wjf/harness.py @@ -0,0 +1,157 @@ +"""Import harness for the WJ classic-ASP -> flask reference loader. + +This is SITE-SPECIFIC reference glue, NOT product code. It maps West Jefferson's +classic schema onto the maintained, schema-agnostic import contract in +docs/IMPORT-API.md. Other sites copy this pattern against their own source DB; +they do not run this loader. + +The harness builds the flask app against the current DATABASE_URL (point it at a +throwaway import database), mints an unscoped admin PAT in-process, and drives +the REAL import endpoints through the app test client with the X-Import-Mode +header - so it exercises the same routes/authz/validation an HTTP client would, +without needing a running server. Legacy-id -> new-id maps persist to JSON so a +crashed run resumes and later stages resolve foreign keys. +""" + +import json +import os +import pymysql + +from shopdb import create_app +from shopdb.extensions import db + + +def _source_password(): + """Root password for the scratch source DBs: reuse the dev DATABASE_URL in + .env unless SOURCE_DB_PASSWORD overrides.""" + if os.environ.get('SOURCE_DB_PASSWORD'): + return os.environ['SOURCE_DB_PASSWORD'] + env = os.path.join(os.path.dirname(__file__), '..', '..', '..', '.env') + with open(os.path.abspath(env)) as handle: + for line in handle: + if line.startswith('DATABASE_URL='): + after = line.split('root:', 1)[1] + return after.split('@', 1)[0] + raise RuntimeError('could not resolve source DB password') + + +class Source: + """Read-only access to the three scratch source databases.""" + + def __init__(self, host='127.0.0.1', port=3306, user='root', password=None): + self.password = password or _source_password() + self.host, self.port, self.user = host, port, user + self._conns = {} + + def db(self, name): + if name not in self._conns: + self._conns[name] = pymysql.connect( + host=self.host, port=self.port, user=self.user, + password=self.password, database=name, charset='utf8mb4', + cursorclass=pymysql.cursors.DictCursor) + return self._conns[name] + + def rows(self, dbname, sql, args=None): + with self.db(dbname).cursor() as cur: + cur.execute(sql, args or ()) + return cur.fetchall() + + +class IdMap: + """Persisted legacy-id -> new-id crosswalks, one namespace per entity.""" + + def __init__(self, path): + self.path = path + self._data = {} + if os.path.exists(path): + with open(path) as handle: + self._data = json.load(handle) + + def put(self, namespace, legacy, new): + self._data.setdefault(namespace, {})[str(legacy)] = new + self.save() + + def get(self, namespace, legacy, default=None): + return self._data.get(namespace, {}).get(str(legacy), default) + + def has(self, namespace, legacy): + return str(legacy) in self._data.get(namespace, {}) + + def count(self, namespace): + return len(self._data.get(namespace, {})) + + def save(self): + with open(self.path, 'w') as handle: + json.dump(self._data, handle, indent=1) + + +class Harness: + """App + admin PAT + import-mode test client + id maps.""" + + def __init__(self, idmap_path=None): + self.app = create_app() + self.appcontext = self.app.app_context() + self.appcontext.push() + self._silence_sql_logging() + self.client = self.app.test_client() + self.secret = self._mint_admin_pat() + default = os.path.join(os.path.dirname(__file__), 'idmap.json') + self.ids = IdMap(idmap_path or default) + self.source = Source() + self.errors = [] + + def _silence_sql_logging(self): + import logging + logging.getLogger('sqlalchemy.engine').setLevel(logging.WARNING) + + 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 + (docs/IMPORT-API.md). Returns the clear secret.""" + from werkzeug.security import generate_password_hash + from shopdb.core.models import User, Role, ApiToken + admin = User.query.join(User.roles).filter(Role.rolename == 'admin').first() + if not admin: + role = Role.query.filter_by(rolename='admin').first() \ + or Role(rolename='admin', description='Administrator') + db.session.add(role) + admin = User(username='importer', email='importer@localhost', + passwordhash=generate_password_hash('x'), isactive=True) + admin.roles.append(role) + db.session.add(admin) + db.session.commit() + secret = ApiToken.generate_secret() + token = ApiToken(userid=admin.userid, name='wjf legacy import', + tokenprefix=ApiToken.prefix_of(secret), + tokenhash=ApiToken.hash_secret(secret), scopes=None) + db.session.add(token) + db.session.commit() + return secret + + def _headers(self, import_mode): + # A PAT authenticates as Authorization: Bearer shopdb_pat_... (the + # before_request shim resolves it into a JWT identity). X-API-Key is for + # collector/service tokens, not personal tokens. + headers = {'Authorization': f'Bearer {self.secret}'} + if import_mode: + headers['X-Import-Mode'] = 'true' + return headers + + def post(self, path, payload, import_mode=True): + """POST to an import endpoint. Returns (status, data). 409s are returned + (idempotency handled by the caller), other 4xx/5xx are recorded.""" + resp = self.client.post(path, json=payload, headers=self._headers(import_mode)) + body = resp.get_json() or {} + data = body.get('data', body) + if resp.status_code >= 400 and resp.status_code != 409: + self.errors.append((path, resp.status_code, payload, data)) + return resp.status_code, data + + def get(self, path): + resp = self.client.get(path, headers=self._headers(False)) + body = resp.get_json() or {} + return resp.status_code, body.get('data', body) + + def close(self): + self.ids.save() + self.appcontext.pop() diff --git a/scripts/site_imports/wjf/run.py b/scripts/site_imports/wjf/run.py new file mode 100644 index 0000000..016c728 --- /dev/null +++ b/scripts/site_imports/wjf/run.py @@ -0,0 +1,134 @@ +"""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 _upsert(h, path, payload, unique_field, 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.""" + status, data = h.post(path, payload) + if status in (200, 201): + return _id_of(data) + if status == 409: + _, rows = h.get(list_path or path) + 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 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 + + +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') + 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') + 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') + 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} + + +STAGES = { + 'reference': stage_reference, + 'employees': stage_employees, + # TODO: 'models', 'applications', 'assets', 'dependents', 'network', 'usb' +} + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument('--stages', default='reference,employees', + 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()