shawn/canteen-seed-import
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
0205866e7ae789527caa45bf80ba82d3bb8faf2b
canteen-seed-import/parser.py
T
shawn 0205866e7a feat: add disney_area column and resort area mapping 
- Add DISNEY_AREA_MAP in parser.py mapping 40+ Disney properties to 8 resort areas
- Add disney_area to parse_customer() output, resolver after disney_park
- Update db_writer.py schema: ORDERED_COLUMNS, AUTO_COLUMNS, COLUMN_TYPES
- Add /api/status endpoint query for by_disney_area distribution
- Add area row to web dashboard UI
- Auto-migration adds column to existing databases
2026-05-24 23:36:41 -04:00

1709 lines
58 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
"""
🌱 Canteen Seed Data Import — CSV Parser & Header Mapper
Reads the raw Cantaloupe Excel export (Machine_List.xlsx, 1,848 machines × 62 columns)
and parses it into structured machine records mapping to DB columns.
Usage:
from parser import parse_excel, parse_excel_to_json
machines = parse_excel('seed-data/Machine_List.xlsx')
for m in machines[:3]:
print(json.dumps(m, indent=2, default=str))
"""
import re
import json
from datetime import datetime
import openpyxl
# ─── Disney Property Mapping (Section 1) ──────────────────────────────────────
DISNEY_PROPERTY_MAP = {
"D-ART OF ANIMATION": "Art of Animation Resort",
"D-All Star Music": "All-Star Music Resort",
"D-All Star Sports": "All-Star Sports Resort",
"D-BAY LAKE TOWER": "Bay Lake Tower at Contemporary",
"D-BLIZZARD BEACH": "Blizzard Beach Water Park",
"D-Boardwalk": "Disney's BoardWalk Inn",
"D-Celebration": "Celebration (Admin)",
"D-ContemporaryHotel": "Disney's Contemporary Resort",
"D-DISNEY VENDING": "Disney Support Services",
"D-DISNEY WORLD SS": "Disney World Support Services",
"D-DRC": "Disney Regional Center",
"D-Disney Springs": "Disney Springs",
"D-FORT WILDERNESS": "Fort Wilderness Resort & Campground",
"D-GRAND FLORIDIAN": "Disney's Grand Floridian Resort",
"D-POLYNESIAN RESORT": "Disney's Polynesian Village Resort",
"D-POP CENTURY": "Disney's Pop Century Resort",
"D-PORT ORLEANS FrQt": "Port Orleans French Quarter",
"D-Port Orleans Rvsd": "Port Orleans Riverside",
"D-RIVIERA RESORT": "Disney's Riviera Resort",
"D-Saratoga Springs": "Saratoga Springs Resort & Spa",
"D-Treehouse Villas": "Treehouse Villas (Saratoga Springs)",
"D-WIDE WORLD SPORTS": "ESPN Wide World of Sports",
"D-WILDERNESS LODGE": "Disney's Wilderness Lodge",
"D-Winter Summerland": "Winter Summerland Mini Golf",
"D-YACHT AND BEACH": "Disney's Yacht & Beach Club Resorts",
}
# Additional Disney customer prefix patterns (from data)
DISNEY_CUSTOMER_PREFIXES = [
"D-All Star Movie",
"D-Animal Kngdm Lodge",
"D-Animal Kingdom",
"D-ART OF ANIMATION",
"D-All Star Music",
"D-All Star Sports",
"D-BAY LAKE TOWER",
"D-BLIZZARD BEACH",
"D-Boardwalk",
"D-Celebration",
"D-ContemporaryHotel",
"D-Coronado Springs",
"D-DISNEY VENDING",
"D-DISNEY WORLD SS",
"D-DRC",
"D-Disney Springs",
"D-Epcot",
"D-FORT WILDERNESS",
"D-GRAND FLORIDIAN",
"D-Gran Destino",
"D-Hollywood Studios",
"D-Magic Kingdom",
"D-POLYNESIAN RESORT",
"D-POP CENTURY",
"D-PORT ORLEANS FrQt",
"D-Port Orleans Rvsd",
"D-RIVIERA RESORT",
"D-Saratoga Springs",
"D-Treehouse Villas",
"D-WIDE WORLD SPORTS",
"D-WILDERNESS LODGE",
"D-Winter Summerland",
"D-YACHT AND BEACH",
]
# ─── Disney Resort Area Mapping (Section 2) ────────────────────────────────────
#
# Maps cleaned disney_park values to their Walt Disney World resort area.
# Based on official WDW geography: resorts grouped by proximity to parks
# and transportation access (monorail, Skyliner, boat, bus).
#
# Areas:
# Magic Kingdom Resort Area — Monorail loop & Seven Seas Lagoon
# Epcot Resort Area — Crescent Lake, Skyliner to Epcot/HS
# Animal Kingdom Resort Area — Near Animal Kingdom park
# Disney Springs Resort Area — Near shopping/dining district
# Wide World of Sports Area — All-Star/Pop Century corridor (SE)
# Parks & Attractions — Theme parks, ESPN, cast areas
# Water Parks — Blizzard Beach, Typhoon Lagoon, golf
# Admin & Support — Corporate/admin locations
DISNEY_AREA_MAP = {
# ──── Magic Kingdom Resort Area ─────────────────────────────────────────────
# Hotels along the monorail loop and Seven Seas Lagoon / Bay Lake
"Disney's Contemporary Resort": "Magic Kingdom Resort Area",
"Disney's Polynesian Village Resort": "Magic Kingdom Resort Area",
"Disney's Grand Floridian Resort": "Magic Kingdom Resort Area",
"Bay Lake Tower at Contemporary": "Magic Kingdom Resort Area",
"Disney's Wilderness Lodge": "Magic Kingdom Resort Area",
"Fort Wilderness Resort & Campground": "Magic Kingdom Resort Area",
"D-Island Tower Polynesian": "Magic Kingdom Resort Area",
# ──── Epcot Resort Area ─────────────────────────────────────────────────────
# Hotels along Crescent Lake — walk/Skyliner/boat to Epcot & Hollywood Studios
"Disney's BoardWalk Inn": "Epcot Resort Area",
"Disney's Yacht & Beach Club Resorts": "Epcot Resort Area",
"Disney's Riviera Resort": "Epcot Resort Area",
# ──── Animal Kingdom Resort Area ────────────────────────────────────────────
# Near Animal Kingdom park (Jambo House, Kidani Village)
"D-Animal Kngdm Lodge": "Animal Kingdom Resort Area",
"D-Kidani Village GUEST": "Animal Kingdom Resort Area",
"D-Coronado Springs": "Animal Kingdom Resort Area",
"D-Gran Destino": "Animal Kingdom Resort Area", # At Coronado Springs
# ──── Disney Springs Resort Area ────────────────────────────────────────────
# Near the shopping/dining district
"Saratoga Springs Resort & Spa": "Disney Springs Resort Area",
"Treehouse Villas (Saratoga Springs)": "Disney Springs Resort Area",
"Disney Springs": "Disney Springs Resort Area",
"Port Orleans Riverside": "Disney Springs Resort Area",
"Port Orleans French Quarter": "Disney Springs Resort Area",
"D-Caribbean Beach Guest": "Disney Springs Resort Area",
"D-CARIBBEAN BEACH CAST": "Disney Springs Resort Area",
"D-OLD KEY WEST GUEST": "Disney Springs Resort Area",
"D-OLD KEY WEST CAST": "Disney Springs Resort Area",
# ──── Wide World of Sports Resort Area ──────────────────────────────────────
# All-Star Resorts & Pop Century / Art of Animation corridor (southeastern)
"Art of Animation Resort": "Wide World of Sports Resort Area",
"Disney's Pop Century Resort": "Wide World of Sports Resort Area",
"All-Star Sports Resort": "Wide World of Sports Resort Area",
"All-Star Music Resort": "Wide World of Sports Resort Area",
"D-All Star Movie": "Wide World of Sports Resort Area",
# ──── Parks & Attractions ───────────────────────────────────────────────────
# Theme parks, cast-member areas, sports complex
"D-Magic Kingdom": "Parks & Attractions",
"D-Hollywood Studios": "Parks & Attractions",
"D-Animal Kingdom": "Parks & Attractions",
"D-Epcot": "Parks & Attractions",
"ESPN Wide World of Sports": "Parks & Attractions",
"D-ESPN 2 CAST": "Parks & Attractions",
# ──── Water Parks ───────────────────────────────────────────────────────────
"Blizzard Beach Water Park": "Water Parks",
"D-TYPHOON LAGOON CAST": "Water Parks",
"Winter Summerland Mini Golf": "Water Parks",
"D-FANTASIA GOLF Guest": "Water Parks",
# ──── Admin & Support ───────────────────────────────────────────────────────
"Celebration (Admin)": "Admin & Support",
"Disney Regional Center": "Admin & Support",
"Disney World Support Services": "Admin & Support",
"Disney Support Services": "Admin & Support",
}
# ─── Make Normalization Table (Section 5) ─────────────────────────────────────
MAKE_NORMALIZATION = {
"Crane Co": "Crane",
"Crane Nat": "Crane",
"NATIONAL": "Crane",
"Crane National": "Crane",
"Dixie Narco": "DN",
"Royal Vendors": "Royal",
"Vendo Co": "Vendo",
"U Select It": "USI",
"Automatic Merchandising": "AMS",
"Automatic Products": "AP",
}
# ─── Known suffixes to strip from Place (Pattern A) ───────────────────────────
PLACE_SUFFIXES = [
"Breakroom", "Br", "Break Room", "BREAKROOM",
"BREAK AREA", "Break Area",
"Vending Area", "Vending",
"Break",
"Breakroom ", # trailing space
]
# Floor extraction patterns
FLOOR_PATTERNS = [
(re.compile(r'(\d+)\s*(?:ST|ND|RD|TH)\s*(?:FL|FLOOR)', re.IGNORECASE), lambda m: m.group(1)),
(re.compile(r'(?:FL|FLOOR)\s*(\d+)', re.IGNORECASE), lambda m: m.group(1)),
(re.compile(r'(\d+)\s*[Tt]'), lambda m: str(int(m.group(1)))), # 5T, 6T (Contemporary Tower)
(re.compile(r'(\d+)\s*[Ff][Ll]'), lambda m: m.group(1)), # 2 fl
(re.compile(r'(\d+)[Nn]'), lambda m: str(int(m.group(1)))), # 2N
]
# Building extraction patterns
BUILDING_PATTERNS = [
re.compile(r'BLDG\s*#?\s*(\d+(?:\s*\w+)?)', re.IGNORECASE),
re.compile(r'Building\s+(\d+(?:\s*\w+)?)', re.IGNORECASE),
re.compile(r'Bldg\s+(\d+(?:\s*\w+)?)', re.IGNORECASE),
re.compile(r'DAAR\s+(\d+(?:-\d+)?)', re.IGNORECASE),
re.compile(r'TOWER\s+(\d+)', re.IGNORECASE),
re.compile(r'Loop\s+(\d+)', re.IGNORECASE),
re.compile(r'(\w+\s+(?:VILLAS?|BLDG|BUILDING))', re.IGNORECASE),
]
# Suite extraction patterns
SUITE_PATTERNS = [
(re.compile(r'(?:Suite|SUITE|suite|Ste|STE|ste)\s*#?\s*(\d+)'), lambda m: m.group(1)),
(re.compile(r'(\d+)\s*(?:Suite|SUITE|suite|Ste|STE|ste)'), lambda m: m.group(1)),
]
# Room extraction patterns
ROOM_PATTERNS = [
(re.compile(r'(?:Room|ROOM|room)\s*#?\s*(\d+)'), lambda m: f"Room {m.group(1)}"),
(re.compile(r'(?:RM|Rm)\s*#?\s*(\d+)'), lambda m: f"Room {m.group(1)}"),
(re.compile(r'Hotel by (Room \d+)', re.IGNORECASE), lambda m: m.group(1)),
# Room-type locations (not breakrooms)
(re.compile(r'(?<!Break)(?:Mailroom|MAILROOM|Showroom|SHOWROOM|Game Room|Game Room|Green Room|Emergency Room|Waiting Room|WAITINGROOM|WaitRoom|ICU WAITING ROOM|CARDIO WAITINGROOM|Conf Room|Driver.s Room|Laundry Room)'), lambda m: m.group(0).strip()),
]
# Trailer extraction patterns
TRAILER_PATTERNS = [
(re.compile(r'(?:Trailer|TRAILER|TRLR)\s*#?\s*(\d+)'), lambda m: f"Trailer {m.group(1)}"),
(re.compile(r'(?:Trailer|TRAILER)\s+(\w+)'), lambda m: f"Trailer {m.group(1)}"),
(re.compile(r'MASH Trailer', re.IGNORECASE), lambda m: "Trailer MASH"),
(re.compile(r'Safari Trailer', re.IGNORECASE), lambda m: "Trailer Safari"),
(re.compile(r'^Trailer$', re.IGNORECASE), lambda m: "Trailer"),
]
# ─── Helper: safe string ──────────────────────────────────────────────────────
def _s(val):
"""Convert value to string, handling None."""
if val is None:
return ""
return str(val).strip()
def _num(val):
"""Convert to float or None."""
if val is None:
return None
try:
return float(val)
except (ValueError, TypeError):
return None
def _int_or_none(val):
"""Convert to int or None."""
if val is None:
return None
try:
return int(val)
except (ValueError, TypeError):
return None
def _bool_from_yesno(val):
"""Convert Yes/No to boolean."""
if val is None:
return None
s = str(val).strip().lower()
if s in ("yes", "true", "1", "y"):
return True
if s in ("no", "false", "0", "n"):
return False
return None
# ─── 1. Device / Telemetry Parser (Section 7) ─────────────────────────────────
def parse_device(device_val):
"""
Parse Device (Telemetry ID) column.
Returns dict with telemetry_provider and card_reader_brand.
"""
raw = _s(device_val)
result = {
"telemetry_provider": None,
"card_reader_brand": None,
}
if not raw:
return result
# Extract suffix in parentheses
m = re.search(r'\(([^)]+)\)', raw)
if m:
suffix = m.group(1).strip()
suffix_upper = suffix.upper()
if suffix_upper == "EPORT":
result["telemetry_provider"] = "ePort"
result["card_reader_brand"] = "Cantaloupe"
elif suffix_upper == "NAYAX":
result["telemetry_provider"] = "NAYAX"
result["card_reader_brand"] = "Nayax"
elif suffix_upper == "CMS":
result["telemetry_provider"] = "CMS"
result["card_reader_brand"] = "CMS"
else:
# Unknown suffix
result["telemetry_provider"] = suffix
result["card_reader_brand"] = "Unknown"
else:
# No parentheses - check for known prefixes
if raw.startswith("VJ") or raw.startswith("K3CT") or raw.startswith("VK"):
result["telemetry_provider"] = "ePort"
result["card_reader_brand"] = "Cantaloupe"
elif raw.isdigit():
result["telemetry_provider"] = "Unknown"
result["card_reader_brand"] = "Unknown"
else:
result["telemetry_provider"] = "Unknown"
result["card_reader_brand"] = "Unknown"
return result
# ─── 2. Type / Class / Make / Model Parser (Section 5) ────────────────────────
def normalize_make(make_raw):
"""Normalize make to canonical form."""
if not make_raw:
return None
make = make_raw.strip()
return MAKE_NORMALIZATION.get(make, make)
def parse_type(type_val, class_fallback=None, make_fallback=None, model_fallback=None):
"""
Parse Type column into class, make, model, is_glass_front.
Format: CLASS (MAKE MODEL)
Special: MAKE - MODEL (NATIONAL - 168 SERIES, VENDO - 721, etc.)
Falls back to standalone Class/Make/Model columns if Type doesn't parse.
"""
raw = _s(type_val)
is_gf = False
cls = None
make = None
model = None
if not raw or raw == "Unknown" or raw == "":
# Fallback to standalone columns
cls = _s(class_fallback) if class_fallback else "Unknown"
make = _s(make_fallback) if make_fallback else "Unknown"
model = _s(model_fallback) if model_fallback else "Unknown"
# Check standalone class for GF prefix
if cls and cls.upper().startswith("GF "):
is_gf = True
cls = cls[3:].strip()
return {
"class": cls or "Unknown",
"make": normalize_make(make) or "Unknown",
"model": model or "Unknown",
"is_glass_front": is_gf,
}
# ── Pattern 1: CLASS (MAKE MODEL) ──
m = re.match(r'^(GF\s+)?(\w+(?:/\w+)?)\s*\(([^)]+)\)$', raw)
if m:
gf_prefix = m.group(1) is not None
cls = m.group(2).strip()
inside = m.group(3).strip()
# Parse inside: MAKE MODEL (first word = make, rest = model)
parts = inside.split(None, 1)
make = parts[0] if parts else "Unknown"
model = parts[1] if len(parts) > 1 else "Unknown"
# Handle special cases inside parens
if not parts:
make = "Unknown"
model = "Unknown"
elif len(parts) == 1:
make = parts[0]
model = "Unknown"
else:
make = parts[0]
model = parts[1]
if gf_prefix:
is_gf = True
# Check the raw class for GF prefix too
if raw.upper().startswith("GF "):
is_gf = True
cls = raw.split(None, 1)[1].split("(")[0].strip()
else:
# ── Pattern 2: MAKE - MODEL (dash-separated) ──
m2 = re.match(r'^(GF\s+)?(\w+(?:\s+\w+)?)\s*-\s*(.+)$', raw)
if m2:
gf_prefix = m2.group(1) is not None
prefix = m2.group(2).strip().upper()
model_raw = m2.group(3).strip()
# Map prefix to make + class
prefix_info = _map_dash_prefix(prefix, model_raw, raw)
if prefix_info:
cls = prefix_info["class"]
make = prefix_info["make"]
model = prefix_info["model"]
else:
# Unknown prefix - try standalone columns
cls = _s(class_fallback) if class_fallback else "Unknown"
make = _s(make_fallback) if make_fallback else prefix
model = model_raw
else:
# ── Pattern 3: No parens, no dash ──
# Try standalone columns
cls = _s(class_fallback) if class_fallback else "Unknown"
make = _s(make_fallback) if make_fallback else "Unknown"
model = _s(model_fallback) if model_fallback else "Unknown"
# Check raw for GF prefix
if raw.upper().startswith("GF "):
is_gf = True
cls = raw[3:].strip()
# Handle special Type-only values like "Bev", "Snack", "Food"
simple_class_match = re.match(r'^(GF\s+)?(Snack|Bev|Food|Snack/Bev|Unknown)$', raw, re.IGNORECASE)
if simple_class_match:
if simple_class_match.group(1):
is_gf = True
cls = simple_class_match.group(2)
# Keep standalone make/model
if not make_fallback:
make = "Unknown"
if not model_fallback:
model = "Unknown"
# Normalize class
if cls:
cls = cls.strip()
# Strip GF if still present in class name
if cls.upper().startswith("GF "):
is_gf = True
cls = cls[3:].strip()
# Class normalization
cls_normalized = {
"Snack/Food": "Snack",
}.get(cls, cls)
cls = cls_normalized
# Fallback gaps from standalone columns
if not cls or cls == "Unknown" or cls == "":
cls = _s(class_fallback) if class_fallback else "Unknown"
if not make or make == "Unknown" or make == "":
make = _s(make_fallback) if make_fallback else "Unknown"
if not model or model == "Unknown" or model == "":
model = _s(model_fallback) if model_fallback else "Unknown"
return {
"class": cls or "Unknown",
"make": normalize_make(make) or "Unknown",
"model": model or "Unknown",
"is_glass_front": is_gf,
}
def _map_dash_prefix(prefix, model_raw, original_raw):
"""
Map dash-separated type prefixes to class/make/model.
Based on actual data analysis of Type column patterns.
"""
prefix = prefix.strip()
model_raw = model_raw.strip()
# Class mapping for dash patterns (from data observations)
CLASS_FOR_PREFIX = {
"NATIONAL": None, # varies
"VENDO": "Bev",
"DIXIE NARCO": None, # varies
"ROYAL": "Bev",
"CRANE": None, # varies
"USI": "Snack",
"AUTOMATIC PRODUCTS": "Snack",
"WITTERN": "Snack",
"PEPSI": "Bev",
"UNKNOWN": None,
}
# Make mapping for dash patterns
MAKE_FOR_PREFIX = {
"NATIONAL": "Crane",
"VENDO": "Vendo",
"DIXIE NARCO": "DN",
"ROYAL": "Royal",
"CRANE": "Crane",
"USI": "USI",
"AUTOMATIC PRODUCTS": "VE", # based on actual data
"WITTERN": "USI", # based on actual data
"PEPSI": "DN", # based on actual data
"UNKNOWN": "Unknown",
}
# Model normalization for specific patterns
MODEL_MAP = {
"168 SERIES": "15x/16x",
"167 SERIES": "15x/16x",
"158 SERIES": "15x/16x",
"181 SERIES": "Merchant",
"3000 SERIES": "Evoke",
"121TC SERIES": "Unknown",
"LCM2 SERIES": "Unknown",
"SNACK AP MISC": "Unknown",
"MISC SNACK": "Misc Snack",
"PEPSI LOANER COMBO": "Loaner Combo",
"CAN BEV MISC": "Can Bev Misc",
"501 SERIES": "501E",
"5800 SERIES": "BevMax 4",
"276 SERIES": "276E",
"720 HVV SERIES": "HVV (720)",
"BEV MAX 5800": "BevMax 5800",
"CAN/B DN 501E SERIES": "501E",
"472 SERIES": "472",
'BEV MAX 4 W/7" SCREE': "BevMax 4",
}
# Check if original has GF prefix
is_gf = original_raw.upper().startswith("GF ")
# Determine class based on prefix + model
cls = None
if prefix in CLASS_FOR_PREFIX:
cls = CLASS_FOR_PREFIX[prefix]
# For NATIONAL, class depends on model
if prefix == "NATIONAL":
if model_raw in ("471", "472", "472 SERIES"):
cls = "Food" if is_gf else "Snack"
elif model_raw in ("186", "187", "168 SERIES", "167 SERIES", "158 SERIES", "181 SERIES"):
cls = "Snack"
else:
cls = "Snack"
# For DIXIE NARCO, class depends on model
if prefix == "DIXIE NARCO":
if is_gf or model_raw in ("3800", "5800 SERIES", "BEV MAX 5800"):
cls = "Bev"
elif model_raw in ("501 SERIES", "276 SERIES", "720 HVV SERIES", "CAN/B DN 501E SERIES"):
cls = "Bev"
else:
cls = "Bev"
# For CRANE
if prefix == "CRANE":
if model_raw in ("180-36",):
cls = "Snack"
elif model_raw in ("429D GPL",):
cls = "Food"
elif 'BEV' in model_raw.upper():
cls = "Snack" # BevMax models used as snack machines sometimes
else:
cls = "Snack"
# For WITTERN
if prefix == "WITTERN":
cls = "Snack"
# Normalize model
model = MODEL_MAP.get(model_raw, model_raw)
# For ROYAL, normalize model
if prefix == "ROYAL":
if model_raw in ("550", "660"):
model = "GIII"
# For VENDO, normalize model
if prefix == "VENDO":
if model_raw in ("621", "721", "821"):
model = f"621/721/821"
make = MAKE_FOR_PREFIX.get(prefix, prefix)
return {
"class": cls or "Unknown",
"make": make,
"model": model,
}
# ─── 3. Place Parser (Section 2) ──────────────────────────────────────────────
# Person prefix + address pattern for Pattern B (Disney only)
# Requires: FirstName LastInitial - Number Street...
PATTERN_B_RE = re.compile(
r'^([A-Z][a-z]+)\s+([A-Z])\.?\s*-\s+(\d)'
)
# Address pattern for Pattern B
ADDRESS_PATTERN_RE = re.compile(
r'^\d+\s+\w+(?:\s+\w+)?\s*(?:\w+)?'
)
# Common breakroom/vending suffixes to strip
KNOWN_PLACE_SUFFIXES = [
"Breakroom", "Br", "Break Room", "BREAKROOM", "BREAK AREA",
"Vending Area", "Vending", "Break", "BREAK",
"Breakroom ", "breakroom",
]
def parse_place(place_val, customer_val=None):
"""
Parse Place column into location fields.
Pattern A (simple, 75%): Venue-VenueBreakroom → place
Pattern B (Disney, 25%): PersonName - Address-G/C-Zone FLOOR
Returns dict with: place, building, floor, zone, zone_type
"""
raw = _s(place_val)
result = {
"place": raw or None,
"building": None,
"floor": None,
"zone": None,
"zone_type": None,
"suite": None,
"room": None,
"trailer": None,
}
if not raw:
return result
cust = _s(customer_val) if customer_val else ""
is_disney = cust.upper().startswith("D-")
# ── Try Pattern B first (has person name + address structure) ──
# Only for Disney customers
if is_disney:
pattern_b_result = _try_pattern_b(raw)
if pattern_b_result:
return pattern_b_result
# ── Try simplified Pattern B (G/C- prefix without person name) ──
# E.g., "Gran Destino-G-3rd Floor", "Cast Services-G-AK LODGE..."
simplified_b_result = _try_simplified_pattern_b(raw, is_disney)
if simplified_b_result:
return simplified_b_result
# ── Pattern A (dash-separated, last segment = place) ──
return _parse_pattern_a(raw)
def _try_pattern_b(raw):
"""
Try Pattern B: PersonName - Address-G/C-Zone FLOOR
Returns parsed dict or None.
"""
# Check for person prefix + address pattern
m = PATTERN_B_RE.match(raw)
if not m:
return None
# Strip person prefix (FirstName + Initial + space + dash + space)
first_name = m.group(1)
last_init = m.group(2)
prefix_end = raw.index(m.group(0)) + len(m.group(0)) - 1 # Keep the first digit
remainder = raw[prefix_end:].strip()
# Strip address (starts with digits, up to a dash or G- or end)
addr_match = re.match(r'^(\d+\s+[\w\s]+?)(?:\s*-\s*G-|,\s*G-|\s*-|$)', remainder)
if addr_match:
remainder = remainder[addr_match.end():].strip()
else:
# Try simpler: just find the first -G- or - after the address
g_idx = remainder.find('-G-')
dash_idx = remainder.find(' - ')
if g_idx >= 0:
remainder = remainder[g_idx:].strip()
elif dash_idx >= 0:
remainder = remainder[dash_idx + 3:].strip()
# Now parse the remainder for G/C- prefix, zone, floor, building
return _parse_remainder(remainder)
def _try_simplified_pattern_b(raw, is_disney):
"""
Try simplified Pattern B: No person prefix but has G/C- segments.
E.g., "Gran Destino-G-3rd Floor", "Cast Services-G-AK LODGE ZEBRA 5TH REAR"
"""
# Check if it has G- or C- after a dash
if "-G-" not in raw and " -G-" not in raw and not raw.strip().startswith("C-"):
return None
# If it starts with C- (Cast area), handle specifically
if raw.strip().startswith("C-"):
return _parse_cast_place(raw)
# Check for "xxx -G- yyy" or "xxx-G-yyy"
m = re.search(r'-G-(.+)$', raw)
if not m:
return None
# Split at the -G-
prefix_part = raw[:m.start()].strip()
g_part = m.group(1).strip()
# If prefix has address pattern, try to strip it
prefix = _strip_address(prefix_part)
# Parse G-part for floor, building, zone
return _parse_g_part(g_part, prefix)
def _parse_cast_place(raw):
"""
Parse C- prefixed places like:
"C-MK EASTGATE SECURITY-Be Our Guest"
"C-MK EASTGATE SECURITY-Space Mountain"
"C- RIDE AND SHOW-C- RIDE AND SHOW"
"""
# Strip leading C-
content = re.sub(r'^C-\s*', '', raw)
parts = content.split('-')
parts = [p.strip() for p in parts if p.strip()]
if not parts:
return {
"place": raw,
"building": None,
"floor": None,
"zone": None,
"zone_type": "cast",
"suite": None,
"room": None,
"trailer": None,
}
# Last part is the zone/area
last = parts[-1]
return {
"place": last,
"building": parts[0] if len(parts) > 1 else None,
"floor": None,
"zone": last,
"zone_type": "cast",
"suite": None,
"room": None,
"trailer": None,
}
def _parse_g_part(g_part, prefix):
"""
Parse the G-prefixed portion for floor, building, zone.
E.g., "REAR 6TH FL", "BUILDING 9 LOBBY", "COZY CONE POOL", "3rd Floor"
"""
result = {
"place": prefix or g_part,
"building": None,
"floor": None,
"zone": None,
"zone_type": "guest",
"suite": None,
"room": None,
"trailer": None,
}
# Extract floor
floor_info = _extract_floor(g_part)
# Extract building
building_info = _extract_building(g_part)
# The remainder after removing floor and building is the zone
remainder = g_part
if floor_info["raw"]:
remainder = remainder.replace(floor_info["raw"], "", 1).strip()
if building_info["raw"]:
remainder = remainder.replace(building_info["raw"], "", 1).strip()
# Clean multiple spaces
remainder = re.sub(r'\s+', ' ', remainder).strip()
result["floor"] = floor_info["floor"]
result["building"] = building_info["building"]
result["zone"] = remainder if remainder else None
# If we have a prefix (venue name), use it as place
if prefix:
result["place"] = prefix
return result
def _parse_remainder(remainder):
"""
Parse the remainder after stripping person and address.
Contains G/C- prefix, zone, floor, building.
"""
result = {
"place": remainder,
"building": None,
"floor": None,
"zone": None,
"zone_type": None,
"suite": None,
"room": None,
"trailer": None,
}
if not remainder:
return result
# Check for G- or C- prefix
zone_type = None
text = remainder
if text.upper().startswith("G-"):
zone_type = "guest"
text = text[2:].strip()
elif text.upper().startswith("C-"):
zone_type = "cast"
text = text[2:].strip()
result["zone_type"] = zone_type
# Extract floor from text
floor_info = _extract_floor(text)
# Extract building from text (after floor removal)
text_no_floor = text
if floor_info["raw"]:
text_no_floor = text_no_floor.replace(floor_info["raw"], "", 1).strip()
building_info = _extract_building(text_no_floor)
# After removing floor and building, remaining is zone
zone_remainder = text_no_floor
if building_info["raw"]:
zone_remainder = zone_remainder.replace(building_info["raw"], "", 1).strip()
# Clean up
zone_remainder = re.sub(r'\s+', ' ', zone_remainder).strip()
result["floor"] = floor_info["floor"]
result["building"] = building_info["building"]
result["zone"] = zone_remainder if zone_remainder else None
# Set place to the original G/C- stripped content, or use zone
if result["zone"]:
result["place"] = result["zone"]
elif result["building"]:
result["place"] = result["building"]
else:
result["place"] = text
return result
def _strip_address(text):
"""
Try to strip address pattern from start of text.
E.g., "Gran Destino" from "Gran Destino-G-3rd Floor"
"""
# If text ends with -G- (stripped), the prefix is the venue
return text
def _parse_pattern_a(raw):
"""
Parse Pattern A: dash-separated, last segment = place.
Per the handbook:
1. Split on `-`
2. Take the last non-empty segment
3. Strip known suffixes (Breakroom, Br, Break Room, BREAKROOM, Vending, Break, etc.)
4. Extract floor info if present
5. Extract building info if present
6. Remaining text is the place name
"""
result = {
"place": None,
"building": None,
"floor": None,
"zone": None,
"zone_type": None,
}
if not raw:
return result
text = raw.strip()
# Check for C- or G- prefix on the overall string (area marker)
zone_type = None
if text.upper().startswith("C-"):
zone_type = "cast"
text = text[2:].strip()
elif text.upper().startswith("G-"):
zone_type = "guest"
text = text[2:].strip()
result["zone_type"] = zone_type
# Split by dash
parts = [p.strip() for p in text.split('-') if p.strip()]
if not parts:
result["place"] = text or raw
return result
# Take the last segment
last = parts[-1]
# Strip known suffixes
stripped = _strip_place_suffix(last)
# If stripping left empty, the last segment was entirely a suffix.
# Use the second-to-last segment as the place.
if not stripped:
if len(parts) >= 2:
stripped = _strip_place_suffix(parts[-2])
last = parts[-2]
else:
stripped = last
# Extract floor info
floor_info = _extract_floor(stripped)
text_no_floor = stripped
if floor_info["raw"]:
text_no_floor = stripped.replace(floor_info["raw"], "", 1).strip()
# Extract building info
building_info = _extract_building(text_no_floor)
text_no_building = text_no_floor
if building_info["raw"]:
text_no_building = text_no_floor.replace(building_info["raw"], "", 1).strip()
# Extract suite info
suite_info = _extract_suite(text_no_building)
text_no_extra = text_no_building
if suite_info["raw"]:
text_no_extra = text_no_building.replace(suite_info["raw"], "", 1).strip()
# Extract room info
room_info = _extract_room(text_no_extra)
text_no_extra2 = text_no_extra
if room_info["raw"]:
text_no_extra2 = text_no_extra.replace(room_info["raw"], "", 1).strip()
# Extract trailer info
trailer_info = _extract_trailer(text_no_extra2)
text_no_extra3 = text_no_extra2
if trailer_info["raw"]:
text_no_extra3 = text_no_extra2.replace(trailer_info["raw"], "", 1).strip()
# Clean up
remainder = re.sub(r'\s+', ' ', text_no_extra3).strip()
# If everything was consumed by extractions (last segment was
# purely floor/building/suite/trailer info), fall back to
# the second-to-last segment as the place
if not remainder and len(parts) >= 2:
second_last = _strip_place_suffix(parts[-2])
second_last = re.sub(r'\s+', ' ', second_last).strip() if second_last else None
if second_last:
remainder = second_last
result["floor"] = floor_info["floor"]
result["building"] = building_info["building"]
result["suite"] = suite_info["suite"]
result["room"] = room_info["room"]
result["trailer"] = trailer_info["trailer"]
result["zone"] = remainder if remainder else None
result["place"] = remainder if remainder else last
return result
def _strip_place_suffix(text):
"""Strip known breakroom/vending suffixes from place text.
Only strips when the suffix is at the END of the text (last word(s)),
so 'Breakroom Main' is NOT stripped but 'Turano Breakroom' IS stripped
to 'Turano'.
"""
if not text:
return text
text_lower = text.lower().strip()
# Suffix phrases to strip from the END (ordered longest-first to avoid partial matches)
suffix_phrases = [
"breakroom", "break room", "break area",
"vending area",
"breakro", # truncated form
]
# Single-word suffix tokens (strip as last word)
suffix_words = [
"breakroom", "break", "vending", "br",
"breakro", # truncated
]
# Try stripping as the entire remaining text first (phrase-level)
for suffix in suffix_phrases:
sl = suffix.lower()
if text_lower.endswith(sl):
# Make sure the suffix is the last thing (not part of a compound name)
# Check that the suffix isn't preceded by another word without a space
before = text_lower[:-len(sl)].strip()
if before and not before[-1].isalnum():
# Good, there's a space/separator before the suffix
return text[:-(len(suffix) + (1 if text_lower[-len(sl)-1:-len(sl)] == ' ' else 0))].strip()
elif not before:
# Text IS entirely the suffix
return ""
# Try stripping just the last word if it's a suffix
words = text.split()
if len(words) >= 1:
last_word = words[-1].lower().strip('.,;:!?')
if last_word in suffix_words:
return ' '.join(words[:-1])
# Try stripping last two words: "Breakroom Main" won't match but "Vending Area" will
if len(words) >= 2:
last_two = ' '.join(w.lower().strip('.,;:!?') for w in words[-2:])
if last_two in suffix_phrases:
return ' '.join(words[:-2])
return text
def _extract_floor(text):
"""Extract floor number from text. Returns dict with floor and raw match."""
for pattern, extractor in FLOOR_PATTERNS:
m = pattern.search(text)
if m:
try:
floor_num = int(extractor(m))
return {"floor": floor_num, "raw": m.group(0)}
except (ValueError, IndexError):
continue
return {"floor": None, "raw": None}
def _extract_building(text):
"""Extract building info from text. Returns dict with building and raw match."""
for pattern in BUILDING_PATTERNS:
m = pattern.search(text)
if m:
return {"building": m.group(0).strip(), "raw": m.group(0)}
return {"building": None, "raw": None}
def _extract_suite(text):
"""Extract suite number from text. Returns dict with suite and raw match."""
for pattern, extractor in SUITE_PATTERNS:
m = pattern.search(text)
if m:
return {"suite": extractor(m), "raw": m.group(0)}
return {"suite": None, "raw": None}
def _extract_room(text):
"""Extract room info from text. Returns dict with room and raw match."""
for pattern, extractor in ROOM_PATTERNS:
m = pattern.search(text)
if m:
return {"room": extractor(m), "raw": m.group(0)}
return {"room": None, "raw": None}
def _extract_trailer(text):
"""Extract trailer info from text. Returns dict with trailer and raw match."""
for pattern, extractor in TRAILER_PATTERNS:
m = pattern.search(text)
if m:
return {"trailer": extractor(m), "raw": m.group(0)}
return {"trailer": None, "raw": None}
# ─── 4. Customer Parser (Section 1) ───────────────────────────────────────────
def parse_customer(customer_val):
"""
Parse Customer column into company and disney_park.
Disney customers (D- prefix) get mapped to a Disney property.
Non-Disney customers map directly to company.
"""
raw = _s(customer_val)
result = {
"company": None,
"disney_park": None,
"disney_area": None,
}
if not raw:
return result
raw_upper = raw.upper()
# Check for Disney prefix
if raw_upper.startswith("D-"):
result["company"] = "Disney"
# Find the best matching Disney property
# Try exact match first
if raw in DISNEY_PROPERTY_MAP:
result["disney_park"] = DISNEY_PROPERTY_MAP[raw]
else:
# Try case-insensitive prefix matching
matched = False
for prefix, park_name in DISNEY_PROPERTY_MAP.items():
if raw_upper.startswith(prefix.upper()):
result["disney_park"] = park_name
matched = True
break
if not matched:
# Try prefix matching against the raw customer
for prefix in DISNEY_CUSTOMER_PREFIXES:
if raw_upper.startswith(prefix.upper()):
result["disney_park"] = prefix # use as-is
break
if not result["disney_park"]:
result["disney_park"] = raw
else:
result["company"] = raw
# Resolve Disney resort area from disney_park
if result["disney_park"] and result["disney_park"] in DISNEY_AREA_MAP:
result["disney_area"] = DISNEY_AREA_MAP[result["disney_park"]]
return result
# ─── 5. Serial Number Validator (Section 4) ───────────────────────────────────
# OCR character confusion table
OCR_CORRECTIONS = {
'S': '5',
's': '5',
'I': '1',
'l': '1', # lowercase L also often confused with 1
'O': '0',
'o': '0',
'B': '8',
'b': '8',
'G': '6',
'g': '6',
}
def validate_serial(serial_val):
"""
Validate and OCR-correct serial number.
Returns dict with: serial_number (corrected), is_valid, is_placeholder
"""
raw = _s(serial_val)
result = {
"serial_number": None,
"is_valid": False,
"is_placeholder": False,
}
if not raw:
return result
# Step 1: Strip non-alphanumeric characters (dashes, spaces, dots)
cleaned = re.sub(r'[^a-zA-Z0-9]', '', raw)
if not cleaned:
return result
# Step 2: Check for placeholder values
# Placeholder: < 5 chars, or obvious placeholders
if len(cleaned) < 5:
result["is_placeholder"] = True
return result
# Check for known placeholder values
if cleaned.lower() in ("pending", "n/a", "none", "null", "unknown"):
result["is_placeholder"] = True
return result
if cleaned == '*' or cleaned == 'S' or cleaned == 'a':
result["is_placeholder"] = True
return result
# Step 3: Apply OCR correction
corrected = []
for ch in cleaned:
corrected.append(OCR_CORRECTIONS.get(ch, ch))
corrected_str = ''.join(corrected)
# Step 4: Check if all chars are the same (invalid)
if len(set(corrected_str)) == 1:
result["is_valid"] = False
result["is_placeholder"] = True
return result
# Step 5: Length validation (valid serials are 9-14 chars after cleaning)
is_valid = 9 <= len(corrected_str) <= 14
result["serial_number"] = corrected_str
result["is_valid"] = is_valid
return result
# ─── 6. Alert Parser (Section 8) ──────────────────────────────────────────────
def parse_alerts(alert_val, coil_alerts_val=None, product_alerts_val=None):
"""
Parse Alerts column into structured alert data.
Returns dict with:
- has_alert: bool
- severity: info | critical | none
- service_scheduled: bool
- out_of_touch: bool
- out_of_touch_hours: int | None
- has_not_dexed: bool
- not_dexed_duration: str | None
- coil_alert_count: int
- product_alert_count: int
- raw_text: str | None
"""
result = {
"has_alert": False,
"severity": "none",
"service_scheduled": False,
"service_is_today": False,
"out_of_touch": False,
"out_of_touch_hours": None,
"has_not_dexed": False,
"not_dexed_duration": None,
"coil_alert_count": 0,
"product_alert_count": 0,
"raw_text": None,
}
# Process coil and product alerts
coil = _int_or_none(coil_alerts_val)
product = _int_or_none(product_alerts_val)
if coil and coil > 0:
result["coil_alert_count"] = coil
if product and product > 0:
result["product_alert_count"] = product
# Process Alert text
raw = _s(alert_val)
if not raw:
# Check if there are numeric alerts
if result["coil_alert_count"] > 0 or result["product_alert_count"] > 0:
result["has_alert"] = True
result["severity"] = "info"
return result
result["raw_text"] = raw
result["has_alert"] = True
# Check for service scheduled
if "scheduled for service today" in raw.lower():
result["service_scheduled"] = True
result["service_is_today"] = True
result["severity"] = "info"
if "scheduled for service on" in raw.lower():
result["service_scheduled"] = True
result["severity"] = "info"
# Check for out of touch
oot_match = re.search(r'Out of touch for ([\d.]+)\s*(hours?|days?)', raw, re.IGNORECASE)
if oot_match:
result["out_of_touch"] = True
value = float(oot_match.group(1))
unit = oot_match.group(2).lower()
if "hour" in unit:
result["out_of_touch_hours"] = value
elif "day" in unit:
result["out_of_touch_hours"] = value * 24
result["severity"] = "critical"
# Check for Has not DEXed
dex_match = re.search(r'Has not DEXed in ([\d.]+)\s*(hours?|days?)', raw, re.IGNORECASE)
if dex_match:
result["has_not_dexed"] = True
result["not_dexed_duration"] = dex_match.group(0)
if result["severity"] == "none":
result["severity"] = "info"
return result
# ─── 7. Sales Parser ──────────────────────────────────────────────────────────
def parse_sales(raw):
"""
Parse sales-related columns from the raw dict.
Returns dict with yearly_sales, monthly_sales, weekly_sales,
days_since_restock, daily_avg_sales, today_sales, yesterday_sales,
sales_since_restock.
"""
return {
"yearly_sales": _num(raw.get("Yearly Sales")),
"monthly_sales": _num(raw.get("Monthly Sales")),
"weekly_sales": _num(raw.get("Weekly Sales")),
"daily_avg_sales": _num(raw.get("Daily Average Sales")),
"today_sales": _num(raw.get("Today Sales")),
"yesterday_sales": _num(raw.get("Yesterday Sales")),
"sales_since_restock": _num(raw.get("Sales (Restock)")),
"days_since_restock": _int_or_none(raw.get("Days Since Restock")),
}
# ─── 8. Priority Scorer (Section 9) ──────────────────────────────────────────
def compute_priority(yearly_sales):
"""
Compute priority tier from yearly sales.
Low: ≤ $2,000
Mid: $2,001 - $15,000
High: $15,001+
"""
if yearly_sales is None:
return "Unknown"
if yearly_sales <= 2000:
return "Low"
elif yearly_sales <= 15000:
return "Mid"
else:
return "High"
# ─── 9. Machine ID → Class Correction ───────────────────────────────────────
def correct_class_by_machine_id(machine):
"""
Post-processing: correct class based on machine_id prefix patterns.
Rules from data analysis:
- 90xxx-99xxx → Bev (Beverage machine ID range)
- 60xxx-73xxx → Snack or Food (Snack/Food machine ID range)
Only corrects when the model is clearly a beverage machine (BevMax, Media)
or the class is Unknown and the ID range gives a strong signal.
"""
mid = str(machine.get("machine_id", ""))
cls = machine.get("class", "Unknown")
model = machine.get("model", "")
make = machine.get("make", "")
# Determine ID range
is_bev_range = mid.startswith(("90", "91", "92", "93", "94", "95", "96", "97", "98", "99"))
is_snack_range = mid.startswith(("60", "61", "62", "63", "64", "65", "66", "67", "68", "69", "70", "71", "72", "73"))
# Rule 1: In Bev range but classed as Snack with BevMax/Media model → fix to Bev
if is_bev_range and cls in ("Snack", "Snack/Bev") and model and ("BevMax" in model or "Media" in model):
machine["class"] = "Bev"
machine["class_corrected"] = True
machine["class_correction_reason"] = f"Model '{model}' in Bev ID range"
# Rule 2: Unknown class in Bev range → set to Bev
elif is_bev_range and cls == "Unknown":
machine["class"] = "Bev"
machine["class_corrected"] = True
machine["class_correction_reason"] = "Unknown in Bev ID range"
# Rule 3: Unknown class in Snack/Food range → set to Snack
elif is_snack_range and cls == "Unknown":
machine["class"] = "Snack"
machine["class_corrected"] = True
machine["class_correction_reason"] = "Unknown in Snack ID range"
return machine
# ─── 10. Main Parser ───────────────────────────────────────────────────────────
# Full column mapping: Excel column index (1-based) → DB field name
COLUMN_MAPPING = {
1: "Device",
2: "Location",
3: "Asset ID",
4: "Place",
5: "Type",
6: "City",
7: "Address",
8: "Last Contact Time",
9: "Last Dex Report Time",
10: "Last Restock",
11: "Coil Alerts",
12: "Product Alerts",
13: "Sales (Restock)",
14: "Daily Average Sales",
15: "Today Sales",
16: "Yesterday Sales",
17: "Weekly Sales",
18: "Monthly Sales",
19: "Yearly Sales",
20: "Days Since Restock",
21: "Prepick Group",
22: "Customer",
23: "Management Company",
24: "Management Account",
25: "Machine Management Code",
26: "Route",
27: "Subroute",
28: "Changer Par",
29: "Acquired From",
30: "Purchase Date",
31: "Purchase Price",
32: "Depreciation Years",
33: "Post-Depreciation Monthly Cost",
34: "State",
35: "Postal Code",
36: "Deployed",
37: "Pulled Date",
38: "Serial Number",
39: "Class",
40: "Make",
41: "Model",
42: "Cash Discount",
43: "Tax Jurisdiction",
44: "Commission Plan",
45: "Barcode",
46: "Non-Revenue",
47: "Added Date",
48: "Phone",
49: "Fax",
50: "Email",
51: "Has Cashless",
52: "Branch",
53: "Location Code",
54: "Customer Code",
55: "Last Inventory",
56: "Asset Family",
57: "Status",
58: "Alerts",
59: "Valid Address",
60: "Business Type",
61: "Primary Consumer Type",
62: "Machine Branding",
}
def parse_excel(filepath):
"""
Parse the Excel file and return a list of machine record dicts.
Each record maps raw columns to DB fields, with parsing applied
to Device, Type, Place, Customer, Serial Number, Status, and Alerts.
"""
wb = openpyxl.load_workbook(filepath, read_only=False, data_only=True)
ws = wb.active
# Compute dimensions
max_col = ws.max_column or 62
max_row = ws.max_row or 1849
# Build column index map from header row
headers = [ws.cell(1, c).value for c in range(1, max_col + 1)]
col_map = {}
for i, h in enumerate(headers, 1):
col_map[h] = i
machines = []
for row_idx in range(2, max_row + 1):
row = [ws.cell(row_idx, c).value for c in range(1, ws.max_column + 1)]
# Build raw data dict with field names as keys
raw = {}
for col_idx, field_name in COLUMN_MAPPING.items():
raw[field_name] = row[col_idx - 1] if col_idx - 1 < len(row) else None
# ── Parse each field ──
# Device → telemetry
device_info = parse_device(raw["Device"])
# Type → class, make, model, is_glass_front
type_info = parse_type(
raw["Type"],
class_fallback=raw.get("Class"),
make_fallback=raw.get("Make"),
model_fallback=raw.get("Model"),
)
# Place → location fields
place_info = parse_place(raw["Place"], raw["Customer"])
# Customer → company + disney_park
customer_info = parse_customer(raw["Customer"])
# Serial Number → validate + OCR correct
serial_info = validate_serial(raw["Serial Number"])
# Status → remote_pricing_status
status_raw = _s(raw.get("Status"))
remote_pricing_status = status_raw if status_raw else None
# Alerts → parsed structure
alerts_info = parse_alerts(
raw.get("Alerts"),
coil_alerts_val=raw.get("Coil Alerts"),
product_alerts_val=raw.get("Product Alerts"),
)
# Sales data
sales_info = parse_sales(raw)
# Compute priority from yearly sales
priority = compute_priority(sales_info["yearly_sales"])
# Has Cashless
has_cashless = _bool_from_yesno(raw.get("Has Cashless"))
# Deployed
deployed = _bool_from_yesno(raw.get("Deployed"))
# Build machine record
machine = {
# Identity
"machine_id": _s(raw.get("Asset ID")),
"serial_number": serial_info["serial_number"],
"serial_is_valid": serial_info["is_valid"],
"serial_is_placeholder": serial_info["is_placeholder"],
# Type info
"class": type_info["class"],
"make": type_info["make"],
"model": type_info["model"],
"is_glass_front": type_info["is_glass_front"],
# Location
"place": place_info["place"],
"building_name": place_info["building"],
"floor": place_info["floor"],
"suite": place_info["suite"],
"room": place_info["room"],
"trailer": place_info["trailer"],
"zone": place_info["zone"],
"zone_type": place_info["zone_type"],
"address": _s(raw.get("Address")),
"location_area": _s(raw.get("City")),
"state": _s(raw.get("State")),
"postal_code": _s(raw.get("Postal Code")),
# Customer
"company": customer_info["company"],
"disney_park": customer_info["disney_park"],
"disney_area": customer_info["disney_area"],
# Telemetry
"telemetry_provider": device_info["telemetry_provider"],
"card_reader_brand": device_info["card_reader_brand"],
# Pricing
"remote_pricing_status": remote_pricing_status,
# Alerts
"alerts": alerts_info,
# Sales & Priority
"yearly_sales": sales_info["yearly_sales"],
"monthly_sales": sales_info["monthly_sales"],
"weekly_sales": sales_info["weekly_sales"],
"daily_avg_sales": sales_info["daily_avg_sales"],
"today_sales": sales_info["today_sales"],
"yesterday_sales": sales_info["yesterday_sales"],
"sales_since_restock": sales_info["sales_since_restock"],
"days_since_restock": sales_info["days_since_restock"],
"priority": priority,
# Other
"prepick_group": _s(raw.get("Prepick Group")),
"has_cashless": has_cashless,
"deployed": deployed,
# Dates
"install_date": raw.get("Added Date"),
"dex_report_date": raw.get("Last Dex Report Time"),
"pulled_date": raw.get("Pulled Date"),
"last_contact_time": raw.get("Last Contact Time"),
"last_restock": raw.get("Last Restock"),
# Raw reference (for debugging)
"coil_alert_count": _int_or_none(raw.get("Coil Alerts")),
"product_alert_count": _int_or_none(raw.get("Product Alerts")),
"route": _s(raw.get("Route")),
"subroute": _s(raw.get("Subroute")),
"branch": _s(raw.get("Branch")),
"asset_family": _s(raw.get("Asset Family")),
"valid_address": _s(raw.get("Valid Address")),
}
# Apply machine ID → class correction
machine = correct_class_by_machine_id(machine)
machines.append(machine)
wb.close()
return machines
def parse_excel_to_json(filepath, pretty=True):
"""Parse Excel and return JSON string of all machines."""
machines = parse_excel(filepath)
indent = 2 if pretty else None
return json.dumps(machines, indent=indent, default=str, ensure_ascii=False)
# ─── CLI / Test ──────────────────────────────────────────────────────────────
if __name__ == "__main__":
import sys
filepath = sys.argv[1] if len(sys.argv) > 1 else "seed-data/Machine_List.xlsx"
print(f"📄 Parsing {filepath}...")
machines = parse_excel(filepath)
print(f"✅ Parsed {len(machines)} machines\n")
print("=" * 60)
print("FIRST 3 MACHINES (JSON)")
print("=" * 60)
for i, m in enumerate(machines[:3]):
print(f"\n--- Machine {i + 1} ---")
print(json.dumps(m, indent=2, default=str, ensure_ascii=False))
print("\n" + "=" * 60)
print("SUMMARY STATISTICS")
print("=" * 60)
# Count non-null values
fields = [
"telemetry_provider", "company", "disney_park",
"class", "is_glass_front", "remote_pricing_status",
"priority", "yearly_sales",
]
for field in fields:
count = sum(1 for m in machines if m.get(field) is not None and m.get(field) != "")
print(f" {field:30s}: {count:5d} / {len(machines)}")
# Telemetry provider distribution
from collections import Counter
telem_dist = Counter(m.get("telemetry_provider") or "None" for m in machines)
print("\n Telemetry Provider Distribution:")
for provider, count in telem_dist.most_common():
print(f" {provider:15s}: {count}")
# Disney count
disney_count = sum(1 for m in machines if m.get("disney_park"))
print(f"\n Disney machines: {disney_count}")
# Serial validation stats
valid_serials = sum(1 for m in machines if m.get("serial_is_valid"))
placeholder_serials = sum(1 for m in machines if m.get("serial_is_placeholder"))
print(f" Valid serials: {valid_serials}")
print(f" Placeholder serials: {placeholder_serials}")
# Priority distribution
priority_dist = Counter(m.get("priority") or "Unknown" for m in machines)
print("\n Priority Distribution:")
for pri, count in priority_dist.most_common():
print(f" {pri:10s}: {count}")
# Glass front count
gf_count = sum(1 for m in machines if m.get("is_glass_front"))
print(f"\n Glass Front machines: {gf_count}")
# Remote pricing status distribution
rp_dist = Counter(m.get("remote_pricing_status") or "None" for m in machines)
print("\n Remote Pricing Status:")
for status, count in rp_dist.most_common():
print(f" {status:20s}: {count}")
print("\n✅ Done!")
Reference in New Issue View Git Blame Copy Permalink