shawn
07539b683e
Adds scripts/import_from_excel.py that reads Cantaloupe .xlsx exports and imports missing machines into the canteen-asset-tracker DB. - Dry-run by default, --write to commit, --prod for production DB - Maps all 62 spreadsheet columns to assets table schema - Stores full raw row as JSON in seed_data table - Backs up DB before import
412 lines
14 KiB
Python
412 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Import missing machines from Cantaloupe Excel export into canteen-asset-tracker.
|
|
|
|
Reads Machine List(8).xlsx, checks each Asset ID against assets.db,
|
|
and inserts any missing records. Stores full row JSON in seed_data table.
|
|
|
|
Usage:
|
|
python3 import_machines.py # Dry run (no writes)
|
|
python3 import_machines.py --write # Actually import
|
|
python3 import_machines.py --write --prod # Import to production DB
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import sqlite3
|
|
import sys
|
|
from datetime import datetime
|
|
|
|
import openpyxl
|
|
|
|
# ── Paths ──────────────────────────────────────────────────────────────────
|
|
EXCEL_PATH = os.path.expanduser(
|
|
"~/.hermes/profiles/coder/cache/documents/doc_4589f73a5e74_Machine List(8).xlsx"
|
|
)
|
|
|
|
DEV_DB = os.path.expanduser("~/projects/canteen-asset-tracker-dev/assets.dev.db")
|
|
PROD_DB = os.path.expanduser("~/projects/canteen-asset-tracker/assets.db")
|
|
|
|
# ── Column index mapping (0-based) ─────────────────────────────────────────
|
|
COL = {
|
|
"device": 0,
|
|
"location": 1,
|
|
"asset_id": 2,
|
|
"place": 3,
|
|
"type": 4,
|
|
"city": 5,
|
|
"address": 6,
|
|
"last_contact_time": 7,
|
|
"last_dex_report_time": 8,
|
|
"last_restock": 9,
|
|
"coil_alerts": 10,
|
|
"product_alerts": 11,
|
|
"sales_restock": 12,
|
|
"daily_avg_sales": 13,
|
|
"today_sales": 14,
|
|
"yesterday_sales": 15,
|
|
"weekly_sales": 16,
|
|
"monthly_sales": 17,
|
|
"yearly_sales": 18,
|
|
"days_since_restock": 19,
|
|
"prepick_group": 20,
|
|
"customer": 21,
|
|
"management_company": 22,
|
|
"management_account": 23,
|
|
"machine_management_code": 24,
|
|
"route": 25,
|
|
"subroute": 26,
|
|
"changer_par": 27,
|
|
"acquired_from": 28,
|
|
"purchase_date": 29,
|
|
"purchase_price": 30,
|
|
"depreciation_years": 31,
|
|
"post_depr_monthly_cost": 32,
|
|
"state": 33,
|
|
"postal_code": 34,
|
|
"deployed": 35,
|
|
"pulled_date": 36,
|
|
"serial_number": 37,
|
|
"class": 38,
|
|
"make": 39,
|
|
"model": 40,
|
|
"cash_discount": 41,
|
|
"tax_jurisdiction": 42,
|
|
"commission_plan": 43,
|
|
"barcode": 44,
|
|
"non_revenue": 45,
|
|
"added_date": 46,
|
|
"phone": 47,
|
|
"fax": 48,
|
|
"email": 49,
|
|
"has_cashless": 50,
|
|
"branch": 51,
|
|
"location_code": 52,
|
|
"customer_code": 53,
|
|
"last_inventory": 54,
|
|
"asset_family": 55,
|
|
"status": 56,
|
|
"alerts": 57,
|
|
"valid_address": 58,
|
|
"business_type": 59,
|
|
"primary_consumer_type": 60,
|
|
"machine_branding": 61,
|
|
}
|
|
|
|
# ── Helpers ────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def cell(row, key, default=""):
|
|
"""Get cell value by column key."""
|
|
idx = COL[key]
|
|
val = row[idx]
|
|
if val is None:
|
|
return default
|
|
if isinstance(val, datetime):
|
|
return val.isoformat()
|
|
return str(val).strip()
|
|
|
|
|
|
def cell_num(row, key, default=None):
|
|
"""Get numeric cell value."""
|
|
idx = COL[key]
|
|
val = row[idx]
|
|
if val is None:
|
|
return default
|
|
if isinstance(val, (int, float)):
|
|
return val
|
|
try:
|
|
return float(str(val).strip())
|
|
except (ValueError, TypeError):
|
|
return default
|
|
|
|
|
|
def cell_int(row, key, default=None):
|
|
"""Get integer cell value."""
|
|
n = cell_num(row, key, default)
|
|
if n is not None:
|
|
return int(n)
|
|
return None
|
|
|
|
|
|
def parse_type(raw_type):
|
|
"""Parse 'Snack (AMS Sensit 3)' -> (category='Snack', make='AMS', model='Sensit 3')"""
|
|
if not raw_type:
|
|
return "Other", "", ""
|
|
raw = raw_type.strip()
|
|
# Extract parenthetical part
|
|
paren_match = re.search(r"\((.+?)\)", raw)
|
|
paren = paren_match.group(1) if paren_match else ""
|
|
|
|
# Get the category (first word or before parentheses)
|
|
category = re.sub(r"\s*\(.*\).*", "", raw).strip()
|
|
if not category or category == "Unknown":
|
|
category = "Other"
|
|
|
|
# Parse make/model from parentheses
|
|
if paren:
|
|
parts = paren.split(None, 1)
|
|
make = parts[0] if len(parts) > 0 else ""
|
|
model = parts[1] if len(parts) > 1 else ""
|
|
else:
|
|
make = ""
|
|
model = ""
|
|
|
|
return category, make, model
|
|
|
|
|
|
def build_name(row):
|
|
"""Build asset name in format: 'Make @ Location' or 'AssetID / Address / Place'"""
|
|
loc = cell(row, "location")
|
|
place = cell(row, "place")
|
|
aid = cell(row, "asset_id")
|
|
_, make, model = parse_type(cell(row, "type"))
|
|
addr = cell(row, "address")
|
|
|
|
if make:
|
|
name = f"{make} @ {place or loc or addr or aid}"
|
|
else:
|
|
name = f"{aid} / {addr or loc} / {place}" if (addr or place) else aid
|
|
|
|
return name[:250] # DB column limit
|
|
|
|
|
|
def deployed_flag(val):
|
|
"""Convert deployed value to '1' or '0'."""
|
|
if isinstance(val, str) and val.lower() in ("yes", "y", "1", "true"):
|
|
return "1"
|
|
if isinstance(val, (int, float)):
|
|
return "1" if val else "0"
|
|
return "0"
|
|
|
|
|
|
# ── Import logic ──────────────────────────────────────────────────────────
|
|
|
|
|
|
def import_machines(target_db, dry_run=True):
|
|
"""Import missing machines from Excel into the target DB."""
|
|
|
|
if not os.path.exists(EXCEL_PATH):
|
|
print(f"ERROR: Excel file not found: {EXCEL_PATH}")
|
|
return
|
|
|
|
if not os.path.exists(target_db):
|
|
print(f"ERROR: Target DB not found: {target_db}")
|
|
return
|
|
|
|
# Load spreadsheet
|
|
wb = openpyxl.load_workbook(EXCEL_PATH)
|
|
ws = wb["Machine List"]
|
|
|
|
# Read all rows
|
|
all_rows = []
|
|
for r in ws.iter_rows(min_row=2, values_only=True):
|
|
aid = r[COL["asset_id"]]
|
|
if aid:
|
|
all_rows.append(r)
|
|
|
|
print(f"Total machines in spreadsheet: {len(all_rows)}")
|
|
|
|
# Connect to target DB
|
|
conn = sqlite3.connect(target_db)
|
|
conn.row_factory = sqlite3.Row
|
|
|
|
# Get existing machine_ids
|
|
existing = set()
|
|
for r in conn.execute("SELECT machine_id FROM assets WHERE machine_id IS NOT NULL"):
|
|
existing.add(str(r["machine_id"]).strip())
|
|
|
|
print(f"Existing machines in DB: {len(existing)}")
|
|
|
|
# Find missing
|
|
missing_rows = []
|
|
for row in all_rows:
|
|
aid = str(row[COL["asset_id"]]).strip()
|
|
if aid not in existing:
|
|
missing_rows.append((aid, row))
|
|
|
|
print(f"Missing (to import): {len(missing_rows)}")
|
|
missing_rows.sort(key=lambda x: int(x[0]))
|
|
|
|
if not missing_rows:
|
|
print("Nothing to import.")
|
|
conn.close()
|
|
return
|
|
|
|
# Show first 10
|
|
print(f"\nFirst 10 to import:")
|
|
for aid, row in missing_rows[:10]:
|
|
loc = cell(row, "location")
|
|
place = cell(row, "place")
|
|
typ = cell(row, "type")
|
|
print(f" {aid}: {loc[:60]} | {place} | {typ}")
|
|
|
|
if dry_run:
|
|
print(f"\n{'='*60}")
|
|
print(f"DRY RUN - no changes made. Re-run with --write to import.")
|
|
print(f"{'='*60}")
|
|
conn.close()
|
|
return
|
|
|
|
# ── Do the import ─────────────────────────────────────────────────
|
|
conn.execute("PRAGMA foreign_keys = OFF")
|
|
conn.execute("BEGIN TRANSACTION")
|
|
|
|
imported_count = 0
|
|
errors = []
|
|
seed_entries = []
|
|
|
|
for aid, row in missing_rows:
|
|
try:
|
|
cat, make, model = parse_type(cell(row, "type"))
|
|
# Use spreadsheet make/model if type parse didn't yield them
|
|
sp_make = cell(row, "make")
|
|
if sp_make:
|
|
make = sp_make
|
|
sp_model = cell(row, "model")
|
|
if sp_model:
|
|
model = sp_model
|
|
|
|
asset_data = {
|
|
"machine_id": aid,
|
|
"serial_number": cell(row, "serial_number"),
|
|
"name": build_name(row),
|
|
"category": cat,
|
|
"status": cell(row, "status") or "active",
|
|
"make": make,
|
|
"model": model,
|
|
"address": cell(row, "address"),
|
|
"customer_name": cell(row, "customer"),
|
|
"company": cell(row, "customer"),
|
|
"place": cell(row, "place"),
|
|
"seed_city": cell(row, "city"),
|
|
"state": cell(row, "state"),
|
|
"postal_code": cell(row, "postal_code"),
|
|
"route_name": cell(row, "route"),
|
|
"subroute_name": cell(row, "subroute"),
|
|
"branch": cell(row, "branch"),
|
|
"barcode": cell(row, "barcode"),
|
|
"seed_class": cell(row, "class"),
|
|
"device": cell(row, "device"),
|
|
"management_company": cell(row, "management_company"),
|
|
"machine_management_code": cell(row, "machine_management_code"),
|
|
"location_code": cell(row, "location_code"),
|
|
"customer_code": cell(row, "customer_code"),
|
|
"asset_family": cell(row, "asset_family"),
|
|
"business_type": cell(row, "business_type"),
|
|
"primary_consumer_type": cell(row, "primary_consumer_type"),
|
|
"machine_branding": cell(row, "machine_branding"),
|
|
"valid_address": cell(row, "valid_address"),
|
|
"phone": cell(row, "phone"),
|
|
"fax": cell(row, "fax"),
|
|
"email": cell(row, "email"),
|
|
"has_cashless": "1" if cell(row, "has_cashless", "").lower() in ("yes", "y", "1") else "0",
|
|
"non_revenue": cell(row, "non_revenue"),
|
|
"alerts": cell(row, "alerts"),
|
|
"coil_alerts": cell_int(row, "coil_alerts", 0),
|
|
"product_alerts": cell_int(row, "product_alerts", 0),
|
|
"daily_avg_sales": cell_num(row, "daily_avg_sales"),
|
|
"monthly_sales": cell_num(row, "monthly_sales"),
|
|
"yearly_sales": cell_num(row, "yearly_sales"),
|
|
"today_sales": cell_num(row, "today_sales"),
|
|
"yesterday_sales": cell_num(row, "yesterday_sales"),
|
|
"weekly_sales": cell_num(row, "weekly_sales"),
|
|
"sales_restock": cell_num(row, "sales_restock"),
|
|
"last_restock": cell(row, "last_restock"),
|
|
"days_since_restock": cell_int(row, "days_since_restock"),
|
|
"last_contact_time": cell(row, "last_contact_time"),
|
|
"last_dex_report_time": cell(row, "last_dex_report_time"),
|
|
"last_inventory": cell(row, "last_inventory"),
|
|
"prepick_group": cell(row, "prepick_group"),
|
|
"added_date": cell(row, "added_date"),
|
|
"install_date": cell(row, "added_date"), # seed has no install_date separate
|
|
"purchase_date": cell(row, "purchase_date"),
|
|
"purchase_price": cell_num(row, "purchase_price"),
|
|
"depreciation_years": cell_int(row, "depreciation_years"),
|
|
"cash_discount": cell_num(row, "cash_discount", 0),
|
|
"tax_jurisdiction": cell(row, "tax_jurisdiction"),
|
|
"commission_plan": cell(row, "commission_plan"),
|
|
"acquirer_from": cell(row, "acquired_from"),
|
|
"changer_par": cell(row, "changer_par"),
|
|
"deployed": deployed_flag(row[COL["deployed"]]) if row[COL["deployed"]] else "1",
|
|
"pulled_date": cell(row, "pulled_date"),
|
|
"geofence_radius_meters": 50,
|
|
}
|
|
|
|
# Build column names and placeholders
|
|
cols = list(asset_data.keys())
|
|
placeholders = ", ".join("?" for _ in cols)
|
|
col_names = ", ".join(cols)
|
|
vals = [asset_data[c] for c in cols]
|
|
|
|
conn.execute(
|
|
f"INSERT INTO assets ({col_names}) VALUES ({placeholders})",
|
|
vals,
|
|
)
|
|
|
|
# Store raw seed data
|
|
# Get the asset id we just inserted
|
|
new_id = conn.execute(
|
|
"SELECT id FROM assets WHERE machine_id = ?", (aid,)
|
|
).fetchone()["id"]
|
|
|
|
# Build raw_data dict from all spreadsheet cells
|
|
raw_data = {}
|
|
headers = [c.value for c in ws[1]]
|
|
for i, h in enumerate(headers):
|
|
v = row[i]
|
|
if isinstance(v, datetime):
|
|
v = v.isoformat()
|
|
raw_data[h] = v
|
|
|
|
conn.execute(
|
|
"INSERT INTO seed_data (asset_id, raw_data) VALUES (?, ?)",
|
|
(new_id, json.dumps(raw_data, default=str)),
|
|
)
|
|
|
|
imported_count += 1
|
|
|
|
except Exception as e:
|
|
errors.append(f" {aid}: {e}")
|
|
|
|
if errors:
|
|
conn.rollback()
|
|
print(f"\nERRORS - rolled back entire import:")
|
|
for e in errors:
|
|
print(e)
|
|
else:
|
|
conn.commit()
|
|
conn.execute("PRAGMA foreign_keys = ON")
|
|
|
|
conn.close()
|
|
|
|
print(f"\n{'='*60}")
|
|
print(f"Import complete!")
|
|
print(f" Imported: {imported_count} machines")
|
|
print(f" Errors: {len(errors)}")
|
|
if imported_count:
|
|
print(f" First: {missing_rows[0][0]}")
|
|
print(f" Last: {missing_rows[-1][0]}")
|
|
print(f"{'='*60}")
|
|
|
|
|
|
# ── Main ──────────────────────────────────────────────────────────────────
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser(description="Import machines from Excel into canteen asset tracker")
|
|
parser.add_argument("--write", action="store_true", help="Actually write to DB (default: dry run)")
|
|
parser.add_argument("--prod", action="store_true", help="Import to production DB (default: dev)")
|
|
args = parser.parse_args()
|
|
|
|
target = PROD_DB if args.prod else DEV_DB
|
|
mode = "LIVE IMPORT" if args.write else "DRY RUN"
|
|
label = "PRODUCTION" if args.prod else "DEV"
|
|
print(f"{'='*60}")
|
|
print(f" {mode} -> {label} DB")
|
|
print(f" Target: {target}")
|
|
print(f"{'='*60}\n")
|
|
|
|
import_machines(target, dry_run=not args.write)
|