388 lines
15 KiB
Python
388 lines
15 KiB
Python
"""Dry-run analyzer for the WINC (West Manufacturing) machine adoption.
|
|
|
|
READ ONLY. This script opens the database, reads the May asset export, decides
|
|
what an import WOULD change, and prints a report. It never writes. Nothing here
|
|
commits, updates, or inserts - that is a separate tool, written only once this
|
|
report reads clean and a human has signed off on it.
|
|
|
|
What it does, in order:
|
|
|
|
1. Load every existing machine's asset number and build the crosswalk key
|
|
(the workstation number, e.g. 0083A) the export will be matched on.
|
|
2. Read the export with a real CSV parser (quoted commas, BOM) and, for each
|
|
row that carries a workstation number, decide:
|
|
- does it match an existing machine? (in scope vs skipped)
|
|
- what name would we set? (export description)
|
|
- what location would we set? (WM-<bay> -> locations, via
|
|
aliases.LOCATION_OVERRIDES for
|
|
the exceptions)
|
|
- what business unit would we set? (eq1 -> aliases.EQ1_TO_
|
|
BUSINESSUNIT -> businessunits)
|
|
3. Classify each field resolution as RESOLVED, BLANK (nothing to set, leave
|
|
as-is), NEEDS-CREATE (target value is valid but not yet in the database),
|
|
or UNRESOLVED (no alias entry - a human must add one to aliases.py).
|
|
4. Print a summary and, with --rows, a per-row table. With --csv PATH it also
|
|
writes the full plan so the numbers can be diffed between runs.
|
|
|
|
Run it against the real database read-only:
|
|
|
|
python -m scripts.site_imports.winc.dry_run --export "exports/ASSET_current May26.csv"
|
|
python -m scripts.site_imports.winc.dry_run --export EXPORT.csv --rows
|
|
python -m scripts.site_imports.winc.dry_run --export EXPORT.csv --csv plan.csv
|
|
|
|
The design goal is reliability over cleverness: one match rule, one location
|
|
rule, one business-unit rule, every exception in aliases.py, and a report that
|
|
tells a human exactly what is unresolved so the alias table can be completed
|
|
before anyone runs the writer.
|
|
"""
|
|
|
|
import argparse
|
|
import csv
|
|
import os
|
|
import re
|
|
import sys
|
|
from collections import OrderedDict
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(
|
|
os.path.dirname(os.path.abspath(__file__))))))
|
|
|
|
from dotenv import load_dotenv
|
|
|
|
# Same reasoning as scripts/export_catalog.py: a plain `python -m` run gets
|
|
# neither wsgi.py's nor the flask CLI's .env load, so DATABASE_URL would be
|
|
# unset and config would silently fall back to a hardcoded localhost URL. Load
|
|
# it here before anything reads config.
|
|
load_dotenv(Path(__file__).resolve().parents[3] / '.env')
|
|
|
|
from scripts.site_imports.approval import build_packet, format_packet
|
|
from scripts.site_imports.winc import writer
|
|
from scripts.site_imports.suggest import suggest
|
|
from scripts.site_imports.resolver import (
|
|
BLANK,
|
|
NEEDS_CREATE,
|
|
RESOLVED,
|
|
Resolution,
|
|
UNRESOLVED,
|
|
load_lookup,
|
|
norm as _norm,
|
|
resolve_alias,
|
|
resolve_name,
|
|
)
|
|
from scripts.site_imports.winc.aliases import (
|
|
EQ1_TO_BUSINESSUNIT,
|
|
LOCATION_OVERRIDES,
|
|
)
|
|
|
|
# WM- is the only bay prefix in the WINC export; a location code resolves by
|
|
# dropping it and matching a location by name. Everything else is an override.
|
|
_WM_PREFIX = re.compile(r'^WM-?', re.IGNORECASE)
|
|
|
|
# A workstation number is a LEADING run of digits with an optional single letter
|
|
# suffix. The export writes it clean (W0083A); the machines table stores the
|
|
# same number but often with a trailing descriptor (0083 Debur, 0083A Inspect).
|
|
# So the key is the leading token on both sides, matched unanchored - anchoring
|
|
# with $ would drop every existing machine whose name carries a descriptor and
|
|
# silently shrink the in-scope set. The leading W is the single biggest gotcha
|
|
# in this data, so it is stripped here in exactly one place.
|
|
_WSNUM = re.compile(r'^(\d{3,5}[A-Z]?)')
|
|
|
|
|
|
def workstation_key(assetnum):
|
|
"""The crosswalk key (e.g. '0083A') for an asset number, or None.
|
|
|
|
Strips a leading W and keeps the leading run of digits plus an optional
|
|
single trailing letter, ignoring any descriptor that follows. Returns None
|
|
for anything that is not a workstation number (blank rows, facilities codes,
|
|
named test machines) so non-machine rows fall out of scope on their own
|
|
rather than being force-fit.
|
|
"""
|
|
text = (assetnum or '').strip().upper()
|
|
if text.startswith('W'):
|
|
text = text[1:]
|
|
match = _WSNUM.match(text)
|
|
return match.group(1) if match else None
|
|
|
|
|
|
def load_existing_machine_keys(connection):
|
|
"""Map workstation key -> list of existing machine asset numbers.
|
|
|
|
A list, not a single value, because the machines table has a few numbers
|
|
that collide once reduced to their workstation key (0083 exists as
|
|
'0083 Debur' and '0083 Inspection'). The analyzer reports those as ambiguous
|
|
rather than guessing which one the export row means.
|
|
"""
|
|
from sqlalchemy import text
|
|
|
|
keys = OrderedDict()
|
|
rows = connection.execute(text(
|
|
"SELECT a.assetnumber "
|
|
"FROM assets a JOIN machines m ON m.assetid = a.assetid"))
|
|
for (assetnumber,) in rows:
|
|
key = workstation_key(assetnumber)
|
|
if key:
|
|
keys.setdefault(key, []).append(assetnumber)
|
|
return keys
|
|
|
|
|
|
def load_locations(connection, display=None):
|
|
"""locationname (normalized) -> locationid, from the live locations table."""
|
|
return load_lookup(connection, 'locations', 'locationid', 'locationname',
|
|
display=display)
|
|
|
|
|
|
def load_businessunits(connection):
|
|
"""businessunit (normalized) -> businessunitid, from the live table."""
|
|
return load_lookup(connection, 'businessunits', 'businessunitid',
|
|
'businessunit')
|
|
|
|
|
|
def resolve_location(raw_location, locations_by_name):
|
|
"""Resolution for a raw WM bay code.
|
|
|
|
Rule: an override in aliases wins; otherwise drop the WM- prefix and match a
|
|
location by name. A blank export cell means "leave the asset's location as
|
|
is" and is reported as BLANK, not UNRESOLVED, because there is nothing to
|
|
set and nothing for a human to fix. Overrides are checked first because they
|
|
carry the exceptions the plain prefix rule cannot handle.
|
|
"""
|
|
raw = (raw_location or '').strip()
|
|
if not raw:
|
|
return Resolution(BLANK, None, None)
|
|
|
|
if _norm(raw) in LOCATION_OVERRIDES:
|
|
return resolve_alias(raw, LOCATION_OVERRIDES, locations_by_name)
|
|
|
|
name = _WM_PREFIX.sub('', raw).strip()
|
|
return resolve_name(name, locations_by_name)
|
|
|
|
|
|
def resolve_businessunit(raw_eq1, businessunits_by_name):
|
|
"""Resolution for an eq1 value stream.
|
|
|
|
A blank eq1 is BLANK (leave the asset alone). A non-blank value must have an
|
|
alias entry; without one it is UNRESOLVED so a human adds it to aliases.py. A
|
|
mapped canonical name that does not yet exist in businessunits is
|
|
NEEDS-CREATE - flagged, never created here.
|
|
"""
|
|
return resolve_alias(raw_eq1, EQ1_TO_BUSINESSUNIT, businessunits_by_name)
|
|
|
|
|
|
def read_export(path):
|
|
"""Rows of the export as dicts. utf-8-sig handles the BOM; the csv module
|
|
handles the quoted commas embedded in some descriptions."""
|
|
with open(path, newline='', encoding='utf-8-sig') as handle:
|
|
return list(csv.DictReader(handle))
|
|
|
|
|
|
def parse_export_text(text):
|
|
"""Rows of an export given its raw CSV text (same shape as read_export).
|
|
|
|
Split out so an API route can hand the analyzer an uploaded file's bytes
|
|
without writing them to disk, while the CLI keeps reading from a path.
|
|
"""
|
|
import io
|
|
|
|
return list(csv.DictReader(io.StringIO(text)))
|
|
|
|
|
|
def analyze(connection, export_path):
|
|
"""Build the per-row plan from a file path. Returns (plan, summary, lookups)."""
|
|
return analyze_rows(connection, read_export(export_path))
|
|
|
|
|
|
def analyze_rows(connection, export_rows):
|
|
"""Build the per-row plan. Returns (plan_rows, summary_dict, lookups).
|
|
|
|
Takes already-parsed export rows so the same analysis serves both the CLI
|
|
(which reads a path) and the API route (which reads uploaded bytes).
|
|
"""
|
|
existing = load_existing_machine_keys(connection)
|
|
location_names = {}
|
|
locations = load_locations(connection, display=location_names)
|
|
businessunits = load_businessunits(connection)
|
|
|
|
plan = []
|
|
summary = OrderedDict([
|
|
('export_rows', len(export_rows)),
|
|
('non_workstation_rows', 0),
|
|
('in_scope', 0),
|
|
('skipped_no_existing_machine', 0),
|
|
('ambiguous_match', 0),
|
|
('location_' + RESOLVED, 0),
|
|
('location_' + BLANK, 0),
|
|
('location_' + NEEDS_CREATE, 0),
|
|
('location_' + UNRESOLVED, 0),
|
|
('businessunit_' + RESOLVED, 0),
|
|
('businessunit_' + BLANK, 0),
|
|
('businessunit_' + NEEDS_CREATE, 0),
|
|
('businessunit_' + UNRESOLVED, 0),
|
|
])
|
|
|
|
for row in export_rows:
|
|
assetnum = (row.get('assetnum') or '').strip()
|
|
key = workstation_key(assetnum)
|
|
if key is None:
|
|
summary['non_workstation_rows'] += 1
|
|
continue
|
|
|
|
matches = existing.get(key, [])
|
|
if not matches:
|
|
summary['skipped_no_existing_machine'] += 1
|
|
continue
|
|
|
|
summary['in_scope'] += 1
|
|
ambiguous = len(matches) > 1
|
|
if ambiguous:
|
|
summary['ambiguous_match'] += 1
|
|
|
|
loc = resolve_location(row.get('location'), locations)
|
|
bu = resolve_businessunit(row.get('eq1'), businessunits)
|
|
summary['location_' + loc.status] += 1
|
|
summary['businessunit_' + bu.status] += 1
|
|
|
|
plan.append(OrderedDict([
|
|
('assetnum', assetnum),
|
|
('workstation_key', key),
|
|
('matched_asset', '|'.join(matches)),
|
|
('ambiguous', 'yes' if ambiguous else ''),
|
|
('name', (row.get('description') or '').strip()),
|
|
('raw_location', (row.get('location') or '').strip()),
|
|
('location_status', loc.status),
|
|
('locationid', loc.id if loc.id is not None else ''),
|
|
('location_name', loc.name or ''),
|
|
('raw_eq1', (row.get('eq1') or '').strip()),
|
|
('businessunit_status', bu.status),
|
|
('businessunitid', bu.id if bu.id is not None else ''),
|
|
('businessunit_name', bu.name or ''),
|
|
]))
|
|
|
|
return plan, summary, {'locations': locations,
|
|
'location_names': location_names,
|
|
'businessunits': businessunits}
|
|
|
|
|
|
def print_summary(summary):
|
|
print('=' * 60)
|
|
print('WINC machine adoption - DRY RUN (no changes written)')
|
|
print('=' * 60)
|
|
for label, value in summary.items():
|
|
print(' %-32s %6d' % (label.replace('_', ' '), value))
|
|
|
|
|
|
def print_unresolved(plan, lookups=None):
|
|
loc = OrderedDict()
|
|
bu = OrderedDict()
|
|
for row in plan:
|
|
if row['location_status'] == UNRESOLVED:
|
|
loc[row['raw_location']] = loc.get(row['raw_location'], 0) + 1
|
|
if row['businessunit_status'] == UNRESOLVED:
|
|
bu[row['raw_eq1']] = bu.get(row['raw_eq1'], 0) + 1
|
|
needs_loc = OrderedDict()
|
|
needs_bu = OrderedDict()
|
|
for row in plan:
|
|
if row['location_status'] == NEEDS_CREATE:
|
|
needs_loc[row['location_name']] = needs_loc.get(row['location_name'], 0) + 1
|
|
if row['businessunit_status'] == NEEDS_CREATE:
|
|
needs_bu[row['businessunit_name']] = needs_bu.get(row['businessunit_name'], 0) + 1
|
|
|
|
if loc:
|
|
print('\nUNRESOLVED locations (add to aliases.LOCATION_OVERRIDES):')
|
|
location_lookup = (lookups or {}).get('locations') or {}
|
|
location_names = (lookups or {}).get('location_names') or {}
|
|
for value, count in sorted(loc.items(), key=lambda kv: -kv[1]):
|
|
print(' %5d %s' % (count, value))
|
|
stripped = _WM_PREFIX.sub('', value).strip()
|
|
for hint in suggest(value, stripped, location_lookup,
|
|
display=location_names):
|
|
print(' -> suggest "%s" (%s, confidence %s)'
|
|
% (hint.candidate, hint.reason, hint.confidence))
|
|
if bu:
|
|
print('\nUNRESOLVED eq1 (add to aliases.EQ1_TO_BUSINESSUNIT):')
|
|
for value, count in sorted(bu.items(), key=lambda kv: -kv[1]):
|
|
print(' %5d %s' % (count, value))
|
|
if needs_loc:
|
|
print('\nNEEDS-CREATE locations (mapped, but not yet in locations):')
|
|
for value, count in sorted(needs_loc.items(), key=lambda kv: -kv[1]):
|
|
print(' %5d %s' % (count, value))
|
|
if needs_bu:
|
|
print('\nNEEDS-CREATE business units (mapped, but not yet in businessunits):')
|
|
for value, count in sorted(needs_bu.items(), key=lambda kv: -kv[1]):
|
|
print(' %5d %s' % (count, value))
|
|
|
|
|
|
def print_rows(plan):
|
|
print('\nPER-ROW PLAN:')
|
|
header = ('assetnum', 'matched_asset', 'name', 'raw_location',
|
|
'location_status', 'raw_eq1', 'businessunit_status')
|
|
print(' ' + ' | '.join(header))
|
|
for row in plan:
|
|
print(' ' + ' | '.join(str(row[h]) for h in header))
|
|
|
|
|
|
def write_csv(plan, path):
|
|
if not plan:
|
|
return
|
|
with open(path, 'w', newline='', encoding='utf-8') as handle:
|
|
writer = csv.DictWriter(handle, fieldnames=list(plan[0].keys()))
|
|
writer.writeheader()
|
|
writer.writerows(plan)
|
|
|
|
|
|
def build_app_connection():
|
|
"""Yield a live SQLAlchemy connection inside an app context.
|
|
|
|
Returns (app, connection). Caller is responsible for closing nothing - the
|
|
connection lives for the app context and the process exits right after.
|
|
"""
|
|
from shopdb import create_app
|
|
from shopdb.api import db
|
|
|
|
app = create_app()
|
|
context = app.app_context()
|
|
context.push()
|
|
return app, db.session.connection()
|
|
|
|
|
|
def main(argv=None):
|
|
parser = argparse.ArgumentParser(
|
|
description='Dry-run analyzer for WINC machine adoption. Read only.')
|
|
parser.add_argument('--export', required=True,
|
|
help='Path to the ASSET export CSV.')
|
|
parser.add_argument('--rows', action='store_true',
|
|
help='Print the full per-row plan.')
|
|
parser.add_argument('--csv', dest='csv_out',
|
|
help='Write the full per-row plan to this CSV path.')
|
|
parser.add_argument('--packet', action='store_true',
|
|
help='Print the approval packet (hash, needs-create '
|
|
'and blocking summary) a writer would enforce.')
|
|
args = parser.parse_args(argv)
|
|
|
|
if not os.path.exists(args.export):
|
|
parser.error('export file not found: %s' % args.export)
|
|
|
|
_app, connection = build_app_connection()
|
|
plan, summary, lookups = analyze(connection, args.export)
|
|
|
|
print_summary(summary)
|
|
print_unresolved(plan, lookups)
|
|
if args.rows:
|
|
print_rows(plan)
|
|
if args.csv_out:
|
|
write_csv(plan, args.csv_out)
|
|
print('\nWrote per-row plan: %s (%d rows)' % (args.csv_out, len(plan)))
|
|
if args.packet:
|
|
packet = build_packet(
|
|
'winc-machine-adoption', plan, summary,
|
|
[('businessunit', 'businessunit_status', 'raw_eq1',
|
|
'businessunit_name'),
|
|
('location', 'location_status', 'raw_location', 'location_name')],
|
|
creatable=writer._CREATABLE_GROUPS)
|
|
print('\n' + format_packet(packet))
|
|
|
|
return 0
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|