Files
canteen-asset-tracker/scripts/import_cantaloupe.py
T
shawn faacf0bce9 Rename 'Add Asset' tab to 'Add / Find Asset' across nav + tabs
- Drawer nav: 'Add Asset' → 'Add / Find Asset'
- Bottom tab bar: 'Add Asset' → 'Add / Find Asset'
- Updated e2e_browser_test.py assertions to match
2026-05-25 23:14:52 -04:00

224 lines
7.2 KiB
Python

#!/usr/bin/env python3
"""
Import Cantaloupe export (Machine List xlsx) into canteen assets DB.
Populates: company, location_area, place, floor, building_number,
building_name, room, trailer_number
Preserves: lat, lng, photo_path, walking_directions, and any GPS data
"""
import re
import sqlite3
import sys
from pathlib import Path
try:
import openpyxl
except ImportError:
print("Need openpyxl: pip install openpyxl")
sys.exit(1)
DB = str(Path(__file__).parent.parent / "assets.db")
# Patterns for person names at start of place column
PERSON_RE = re.compile(
r'^(?:\w+\s+[A-Z]\.?\s*-\s*|' # "Carrianne C - ", "Donna B - "
r'\w+(?:\s+\w+)?\s+-\s*)?' # "Helena M - " optional
)
# Address-like patterns (number + street suffix)
ADDR_RE = re.compile(
r'\b\d+\s+(?:\w+\s+){0,3}'
r'(?:ST|DR|RD|LN|BLVD|AVE|PKWY|CIR|CT|PL|WAY|TRL|TERR|HWY|LOOP|PATH|RUN|ROW|DRIVE|STREET|ROAD|LANE|BOULEVARD|AVENUE|PARKWAY|CIRCLE|COURT|PLACE|WAY|TRAIL|TERRACE|HIGHWAY)\b',
re.IGNORECASE
)
# Floor patterns to extract
FLOOR_RE1 = re.compile(r'(\d+)\s*(ST|ND|RD|TH)\s*(?:FL(?:OOR|R)?|FLOOR|FLR|FLO)\b', re.IGNORECASE)
FLOOR_RE2 = re.compile(r'(?:^|[-\s])(?:FL|FLOOR|FLR)\s*(\d{1,2})(?:\s|-|$)', re.IGNORECASE)
FLOOR_RE3 = re.compile(r'(\d+)\s*(st|nd|rd|th)\s*floor\b', re.IGNORECASE)
# Building patterns
BLDG_RE = re.compile(r'(?:BLDG|BUILDING|BLD)\s*#?\s*([A-Za-z0-9][-A-Za-z0-9 .]{0,20}?)(?:\s|$|-|,)', re.IGNORECASE)
# Trailer patterns
TRAILER_RE = re.compile(r'(?:TRAILER|TRLR)\s*#?\s*(\d+)', re.IGNORECASE)
# Room/Suite patterns
ROOM_RE = re.compile(r'(?:ROOM|SUITE|STE)\s*#?\s*([A-Za-z0-9][-A-Za-z0-9 .]{0,15}?)(?:\s|-|$|,)', re.IGNORECASE)
# Person name pattern (capitalized first name followed by capitalized single letter + dash)
PERSON_NAME = re.compile(r'^[A-Z][a-z]+(?:\s+[A-Z][a-z]+)*\s+[A-Z]\.?\s*-\s*')
# Words that are not venue names
NON_VENUE = {
'breakroom', 'break room', 'break rm', 'hallway', 'vending area',
'warehouse', 'lobby', 'lounge', 'cafeteria', 'kitchen',
'employee', 'employee break', 'staff', 'office', 'restroom',
'upstairs', 'downstairs', 'laundry', 'covered ar', 'covered area',
'outside', 'outdoor', 'patio', 'balcony', 'deck',
'storage', 'closet', 'server', 'maint', 'maintenance',
'1st fl', '2nd fl', '3rd fl', '4th fl', '5th fl', '1st floor',
'2nd floor', '3rd floor', '4th floor', '5th floor',
'1st', '2nd', '3rd', '4th', '5th',
}
def parse_place(raw_place):
"""
Parse the Cantaloupe Place column to extract structured info.
Returns (clean_place, floor, building_number, building_name, room, trailer_number)
"""
if not raw_place:
return '', '', '', '', '', ''
text = raw_place.strip()
floor = ''
building_number = ''
building_name = ''
room = ''
trailer_number = ''
# --- Extract floor info first ---
m = FLOOR_RE1.search(text)
if m:
floor = f"{m.group(1)}{m.group(2).lower()} Floor"
text = text.replace(m.group(0), ' ').strip()
if not floor:
m = FLOOR_RE3.search(text)
if m:
floor = f"{m.group(1)}{m.group(2).lower()} Floor"
text = text.replace(m.group(0), ' ').strip()
if not floor:
m = FLOOR_RE2.search(text)
if m and m.group(1).isdigit() and int(m.group(1)) <= 50:
floor = f"Floor {m.group(1)}"
text = text.replace(m.group(0), ' ').strip()
# --- Extract building ---
m = BLDG_RE.search(text)
if m:
building_number = m.group(1).strip().rstrip('-., ')
text = text.replace(m.group(0), ' ').strip()
# --- Extract trailer ---
m = TRAILER_RE.search(text)
if m:
trailer_number = m.group(1)
text = text.replace(m.group(0), ' ').strip()
# --- Extract room/suite ---
m = ROOM_RE.search(text)
if m:
room = m.group(1).strip().rstrip('-., ')
text = text.replace(m.group(0), ' ').strip()
# --- Now clean up the place name ---
# Split by dashes to analyze segments
segments = [s.strip() for s in re.split(r'\s*-\s*', text) if s.strip()]
clean_place = ''
for seg in segments:
# Skip person names (e.g., "Carrianne C", "Todd J", "Donna B")
if PERSON_NAME.match(seg + ' - '):
continue
# Skip pure addresses (number + name + suffix)
if ADDR_RE.match(seg) or re.match(r'^\d+\s+\w+', seg):
continue
# Skip single letters (like "G" - Disney building zone)
if len(seg) == 1 and seg.isalpha():
continue
# Skip segments that look like the customer/company name (will come from col 22)
clean_place = seg # Take the last non-address, non-person segment
# If we found nothing useful, take the last segment
if not clean_place and segments:
clean_place = segments[-1]
# Clean up the place value
clean_place = re.sub(r'\s+', ' ', clean_place).strip(' ,-.')
# If place is too generic or empty, leave it blank
if not clean_place or clean_place.lower() in NON_VENUE:
clean_place = ''
return clean_place, floor, building_number, building_name, room, trailer_number
def import_export(filepath):
"""Main import function"""
wb = openpyxl.load_workbook(filepath)
ws = wb.active
conn = sqlite3.connect(DB)
cur = conn.cursor()
updated = 0
errors = []
for row in range(2, ws.max_row + 1):
asset_id = str(ws.cell(row, 3).value or '').strip()
if not asset_id:
continue
customer = str(ws.cell(row, 22).value or '').strip()
city = str(ws.cell(row, 6).value or '').strip()
raw_place = str(ws.cell(row, 4).value or '').strip()
address = str(ws.cell(row, 7).value or '').strip()
# Parse the Place column
cleaned_place, floor, bldg_num, bldg_name, room, trailer = parse_place(raw_place)
# Build update fields - only set if non-empty
updates = {}
if customer:
updates['company'] = customer
if city:
updates['location_area'] = city
if cleaned_place:
updates['place'] = cleaned_place
if floor:
updates['floor'] = floor
if bldg_num:
updates['building_number'] = bldg_num
if room:
updates['room'] = room
if trailer:
updates['trailer_number'] = trailer
if not updates:
continue
# Build SQL
set_clauses = [f"{k} = ?" for k in updates]
values = list(updates.values()) + [asset_id]
sql = f"UPDATE assets SET {', '.join(set_clauses)} WHERE machine_id = ?"
try:
cur.execute(sql, values)
if cur.rowcount:
updated += 1
else:
errors.append(f"MID {asset_id} not found")
except Exception as e:
errors.append(f"MID {asset_id}: {e}")
conn.commit()
conn.close()
wb.close()
print(f"Rows updated: {updated}")
if errors:
print(f"Errors ({len(errors)}):")
for e in errors[:10]:
print(f" {e}")
return updated
if __name__ == '__main__':
if len(sys.argv) < 2:
print("Usage: import_cantaloupe.py <xlsx_file>")
sys.exit(1)
import_export(sys.argv[1])