canteen-asset-tracker/scripts/extract_asset_location_data.py

#!/usr/bin/env python3
"""
Extract floor, building_number, room, trailer_number, and building_name
from asset names in assets.db.  Also deduplicate duplicated name segments.

Usage:
    python3 scripts/extract_asset_location_data.py       # dry-run (report only)
    python3 scripts/extract_asset_location_data.py --apply  # update DB
"""

import re
import sqlite3
import sys
from pathlib import Path
from typing import Optional

DB_PATH = str(Path(__file__).resolve().parent.parent / "assets.db")


# ─── Regex patterns ──────────────────────────────────────────────────────────

# Floor: "1ST FL", "2nd Floor", "3rd Floor", "4TH FLO", "5th floor", "6TH FL",
#        "SOG MAIN 3RD FLOOR", "7th floor", "G-REAR 6TH FL", etc.
# We capture the floor number as the first group.
FLOOR_PATTERNS = [
    re.compile(r'\b(\d+)(?:ST|ND|RD|TH|st|nd|rd|th)\s+(?:FL(?:OOR)?|Floor|floor|FLO(?:OR)?|Floo|floo)\b', re.IGNORECASE),
    # "2ND FLOOR", "3RD FLOOR" etc.
    re.compile(r'\b(\d+)(?:TH|ST|ND|RD)\s+FLOOR\b', re.IGNORECASE),
    # "3rd FLR" (FLR abbreviation)
    re.compile(r'\b(\d+)(?:ST|ND|RD|TH|st|nd|rd|th)\s+FLR\b', re.IGNORECASE),
]

# Floor: "FL02" in "FL02NEWBLD" -> floor "2"
# Must be FL + 1-2 digits + 2+ uppercase letters or end of field (via word boundary).
# Excludes "FL4950", "FL2500", "FL8108", "FL13323" (4+ digits after FL).
# Also matches "FL10W" (floor 10 west), "FL24TWR" (floor 24 tower).
FLOOR_CODE = re.compile(r'\bFL(\d{1,2})(?:(?:[A-Z]{2,})|\b)')

# Building-number patterns
# "BLDG 215", "Bldg 210", "Building 1180", "BLDG 220", "Bldg 215-3rd Floor"
BUILDING_PATTERNS = [
    re.compile(r'\b(?:BLDG|Bldg|Building|Bldg)\s+(\d+)\b'),
    # "#6328", "#811", "#9842", "#5288", "#1001", "# 6328"  (store/chain numbers)
    re.compile(r'#\s*(\d{3,})\b'),
]

# Room patterns
# "Room 1290", "Room 1323", "SUITE 100", "SUITE 214", "Ste 400", "Ste 108"
ROOM_PATTERNS = [
    re.compile(r'\b(?:Room|room)\s+(\d+)\b'),
    re.compile(r'\b(?:SUITE|Suite|suite|Ste)\s+(\d+)\b'),
    # "Breakroom suite 700" (suite at end)
    re.compile(r'\bsuite\s+(\d+)\b'),
]

# Trailer patterns
# "Trailer 1051", "Trailer 1135", "Trailer 1055", "Trailer 1011", "Trailer 1145", "Trailer 1155"
# Also "Trailer 342" (3-digit trailer)
TRAILER_PATTERNS = [
    re.compile(r'\b(?:Trailer|trailer)\s+(\d+)\b'),
]


# ─── Known building/venue names (prefixes to check) ──────────────────────────
# These appear as prefix segments in asset names and are valid building names.
BUILDING_NAME_PREFIXES = [
    "Buena Vista Palace",
    "Hilton Worldwide",
    "Four Seasons Resort",
    "Gran Destino",
    "Shades of Green",
    "Coronado Springs",
    "Caribbean Beach",
    "All Star Movies",
    "Disney Springs",
    "Epcot",
    "Celebration Suites",
]

# Suffixes that indicate what follows is a function/area, NOT part of building name.
AREA_SUFFIXES = [
    "Breakroom", "Break Room", "BRKRM", "Break", "BR",
    "Vending", "Vending Area",
    "Laundry", "Laundry Room",
    "Cafeteria", "Cafe",
    "Maintenance", "Warehouse",
    "Wardrobe", "Cast",
    "Lobby", "Guest",
    "Admin", "Office",
]


def extract_floor(name: str) -> Optional[str]:
    """Extract floor number from an asset name."""
    for pat in FLOOR_PATTERNS:
        m = pat.search(name)
        if m:
            return str(int(m.group(1)))  # Normalise "02" -> "2"
    m = FLOOR_CODE.search(name)
    if m:
        val = str(int(m.group(1)))  # Normalise "02" -> "2"
        # Ensure it's a reasonable floor number (1-100)
        try:
            num = int(val)
            if 1 <= num <= 100:
                return val
        except ValueError:
            pass
    # Edge: "Villas 5 fl by room" etc. — rare but present
    m = re.search(r'\b(\d+)\s+fl\b', name, re.IGNORECASE)
    if m:
        val = str(int(m.group(1)))
        try:
            if 1 <= int(val) <= 100:
                return val
        except ValueError:
            pass
    return None


def extract_building_number(name: str) -> Optional[str]:
    """Extract building number from an asset name."""
    for pat in BUILDING_PATTERNS:
        m = pat.search(name)
        if m:
            return m.group(1)
    return None


def extract_room(name: str) -> Optional[str]:
    """Extract room/suite number from an asset name."""
    for pat in ROOM_PATTERNS:
        m = pat.search(name)
        if m:
            return m.group(1)
    return None


def extract_trailer(name: str) -> Optional[str]:
    """Extract trailer number from an asset name."""
    for pat in TRAILER_PATTERNS:
        m = pat.search(name)
        if m:
            return m.group(1)
    return None


def extract_building_name(name: str) -> Optional[str]:
    """
    Attempt to derive a building/venue name from the asset name.
    
    Strategy: look for known building-name prefixes, or extract the
    first segment before a hyphen/separator that looks like a location name
    (not just a person name, address fragment, or functional descriptor).
    """
    name_upper = name.upper()
    
    # Check known building name prefixes
    for prefix in BUILDING_NAME_PREFIXES:
        if prefix.upper() in name_upper:
            return prefix
    
    # Try prefix extraction: the first significant segment
    # Split on common separators
    parts = re.split(r'\s*-\s*', name)
    first = parts[0].strip()
    
    # Skip if the first segment looks like a person name (has name pattern)
    # e.g. "Todd J - ...", "Brenda G - ...", "Kelli M - ..."
    person_pattern = re.match(r'^[A-Z][a-z]+ [A-Z]\.?$', first)
    if person_pattern:
        return None
    
    # Skip if it's an address fragment
    if re.match(r'^\d+\s+', first):
        # Could be a building name like "Building 1180" — but that's already
        # captured as building_number. Skip other addresses.
        if not re.match(r'^Building\s+\d+', first, re.IGNORECASE):
            return None
    
    # Skip if the first segment is clearly a person or Disney cast ID
    # "Jeremy B", "Randal W", "Susan G", "Brenda G", etc.
    if re.match(r'^[A-Z][a-z]+ [A-Z] -', first):
        return None
    
    # If first segment is long enough (>10 chars) and doesn't contain area keywords,
    # it's probably the building/venue name
    first_no_area = first
    for suffix in AREA_SUFFIXES:
        # Remove trailing area/function suffixes
        pat = re.compile(r'\s+' + re.escape(suffix) + r'$', re.IGNORECASE)
        first_no_area = pat.sub('', first_no_area)
    
    if len(first_no_area) > 8 and not re.search(r'\b(?:BREAK|VEND|LAUNDRY|CAFE|MAINT|WAREHOUSE|CAFETERIA|OFFICE|LOBBY|GUEST|ADMIN|SERVICES|BACKSTAGE|WARDROBE|CA$T|DINING|KITCHEN|STORAGE|WORKSHOP|SECURITY|RECEIVING|PARKING|OUTSIDE|OUTLET|HALLWAY)\b', first_no_area.upper()):
        return first_no_area.strip()
    
    return None


# ─── Name deduplication ──────────────────────────────────────────────────────

def should_deduplicate(left: str, right: str) -> Optional[str]:
    """
    Check if `right` is a duplicate/continuation of `left`.
    Returns the merged name if deduplication is warranted, None otherwise.
    """
    l = left.strip()
    r = right.strip()
    if not l or not r:
        return None
    
    # Exact match: "X-X" -> keep "X"
    if l.lower() == r.lower():
        return l
    
    # Right starts with left: "X" + "-" + "X Y" -> "X Y"
    if r.lower().startswith(l.lower()):
        return r
    
    # Left starts with right: "X Y" + "-" + "X" -> "X Y"
    if l.lower().startswith(r.lower()):
        return l
    
    # Large common prefix (>= 65 % of the shorter string and >= 10 chars)
    min_len = min(len(l), len(r))
    if min_len >= 10:
        common = 0
        for a, b in zip(l.lower(), r.lower()):
            if a == b:
                common += 1
            else:
                break
        if common >= 0.65 * min_len:
            # Keep the longer variant
            return l if len(l) >= len(r) else r
    
    return None


def deduplicate_name(name: str) -> str:
    """
    Detect duplicated name segments and return a cleaned name.
    
    Strategy: for assets with multiple hyphen-separated segments, check
    if splitting at each hyphen position reveals duplication.
    """
    # Find all hyphen positions and try each
    hyphens = [m.start() for m in re.finditer('-', name)]
    
    best_name = name
    
    # Try splitting at first hyphen
    for idx in hyphens:
        left = name[:idx].strip()
        right = name[idx+1:].strip()
        
        result = should_deduplicate(left, right)
        if result is not None:
            candidate = result
            # If this is shorter or the same length but cleaner, use it
            if len(candidate) < len(best_name) or (len(candidate) <= len(best_name) and candidate != name):
                best_name = candidate
    
    # Iterative: run again on the result (for triple-nested duplicates)
    if best_name != name:
        best_name = deduplicate_name(best_name)
    
    return best_name


# ─── Main logic ──────────────────────────────────────────────────────────────

def process_assets(db: sqlite3.Connection, apply: bool = False) -> dict:
    """Process assets table, returning counts of extracted values."""
    cursor = db.cursor()
    
    # Fetch all assets with non-empty names
    cursor.execute("SELECT id, name, floor, building_number, room, trailer_number, building_name FROM assets WHERE name != '' ORDER BY id")
    rows = cursor.fetchall()
    
    counts = {
        'floor': 0,
        'building_number': 0,
        'room': 0,
        'trailer_number': 0,
        'building_name': 0,
        'name_cleaned': 0,
        'total': len(rows),
    }
    
    updates = []
    
    for row in rows:
        asset_id, name, cur_floor, cur_bldg_num, cur_room, cur_trailer, cur_bldg_name = row
        
        # 1. Deduplicate name
        cleaned = deduplicate_name(name)
        name_changed = (cleaned != name)
        
        # 2. Extract fields (only if currently empty)
        floor = cur_floor if cur_floor else (extract_floor(cleaned) or '')
        building_number = cur_bldg_num if cur_bldg_num else (extract_building_number(cleaned) or '')
        room = cur_room if cur_room else (extract_room(cleaned) or '')
        trailer = cur_trailer if cur_trailer else (extract_trailer(cleaned) or '')
        building_name = cur_bldg_name if cur_bldg_name else (extract_building_name(cleaned) or '')
        
        # Track counts
        if floor and not cur_floor:
            counts['floor'] += 1
        if building_number and not cur_bldg_num:
            counts['building_number'] += 1
        if room and not cur_room:
            counts['room'] += 1
        if trailer and not cur_trailer:
            counts['trailer_number'] += 1
        if building_name and not cur_bldg_name:
            counts['building_name'] += 1
        if name_changed:
            counts['name_cleaned'] += 1
        
        updates.append((floor, building_number, room, trailer, building_name, cleaned, asset_id))
    
    if apply:
        cursor.executemany(
            "UPDATE assets SET floor=?, building_number=?, room=?, trailer_number=?, building_name=?, name=? WHERE id=?",
            updates
        )
        db.commit()
    
    return counts


def process_locations(db: sqlite3.Connection, apply: bool = False) -> dict:
    """Process locations table similarly."""
    cursor = db.cursor()
    cursor.execute("SELECT id, name, floor, building_number, trailer_number FROM locations WHERE name != '' ORDER BY id")
    rows = cursor.fetchall()
    
    counts = {
        'floor': 0,
        'building_number': 0,
        'trailer_number': 0,
        'name_cleaned': 0,
        'total': len(rows),
    }
    
    updates = []
    
    for row in rows:
        loc_id, name, cur_floor, cur_bldg_num, cur_trailer = row
        
        # Deduplicate (locations seem cleaner but process anyway)
        cleaned = deduplicate_name(name)
        name_changed = (cleaned != name)
        
        # Extract fields
        floor = cur_floor if cur_floor else (extract_floor(cleaned) or '')
        building_number = cur_bldg_num if cur_bldg_num else (extract_building_number(cleaned) or '')
        trailer = cur_trailer if cur_trailer else (extract_trailer(cleaned) or '')
        
        if floor and not cur_floor:
            counts['floor'] += 1
        if building_number and not cur_bldg_num:
            counts['building_number'] += 1
        if trailer and not cur_trailer:
            counts['trailer_number'] += 1
        if name_changed:
            counts['name_cleaned'] += 1
        
        updates.append((floor, building_number, trailer, cleaned, loc_id))
    
    if apply:
        cursor.executemany(
            "UPDATE locations SET floor=?, building_number=?, trailer_number=?, name=? WHERE id=?",
            updates
        )
        db.commit()
    
    return counts


def print_summary(label: str, counts: dict):
    """Print a human-readable summary of extraction results."""
    print(f"\n{'=' * 60}")
    print(f"  {label}")
    print(f"{'=' * 60}")
    print(f"  Total rows processed:  {counts['total']}")
    print(f"  Floors extracted:      {counts['floor']}")
    print(f"  Building numbers:      {counts['building_number']}")
    if 'room' in counts:
        print(f"  Rooms/Suites:          {counts['room']}")
    if 'trailer_number' in counts:
        print(f"  Trailer numbers:       {counts['trailer_number']}")
    if 'building_name' in counts:
        print(f"  Building names:        {counts['building_name']}")
    print(f"  Names cleaned:         {counts['name_cleaned']}")


def main():
    apply = '--apply' in sys.argv
    
    print(f"DB: {DB_PATH}")
    print(f"Mode: {'APPLY (will write to DB)' if apply else 'DRY RUN (no changes)'}")
    
    db = sqlite3.connect(DB_PATH)
    db.execute("PRAGMA journal_mode=WAL")
    
    print("\nProcessing assets...")
    asset_counts = process_assets(db, apply=apply)
    print_summary("ASSETS", asset_counts)
    
    print("\nProcessing locations...")
    loc_counts = process_locations(db, apply=apply)
    print_summary("LOCATIONS", loc_counts)
    
    db.close()
    
    if not apply:
        print("\n─── DRY RUN ───")
        print("Re-run with --apply to write changes to the database.")
    
    # Sample some cleaned names for verification
    print("\n─── Sample deduplications (first 20) ───")
    db2 = sqlite3.connect(DB_PATH)
    cursor = db2.cursor()
    cursor.execute("SELECT id, name FROM assets WHERE name LIKE '%-%' ORDER BY id LIMIT 20")
    for row in cursor.fetchall():
        original = row[1]
        cleaned = deduplicate_name(original)
        if cleaned != original:
            print(f"  [{row[0]}] '{original}'")
            print(f"       → '{cleaned}'")
    db2.close()


if __name__ == '__main__':
    main()