feat: add script to import machines from Cantaloupe Excel export
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
This commit is contained in:
@@ -0,0 +1,411 @@
|
||||
#!/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)
|
||||
Reference in New Issue
Block a user