Files

560 lines
20 KiB
Python

#!/usr/bin/env python3
"""Clean up asset names by stripping data that's now in structured fields.
Strips from asset names:
- Floor info (now in `floor` column) — only when the floor number matches
- Building number (now in `building_number` column) — with BLDG/Building prefix
- Address fragments (now in `address` column) — number+street pairs that overlap
- "G-" Cantaloupe separator artifacts
- Disney park emoji+display suffixes (now in `disney_park`)
- Park prefixes like "Epcot-" (redundant with disney_park)
- Duplicated building name segments at start of name
- Cantaloupe truncated-name artifacts
Rule: NEVER remove content that isn't also in a structured column.
Usage:
python scripts/clean_asset_names.py # dry-run (report only)
python scripts/clean_asset_names.py --apply # update DB for real
"""
import argparse
import re
from difflib import SequenceMatcher
from pathlib import Path
from typing import Optional
import sqlite3
DB_PATH = str(Path(__file__).resolve().parent.parent / "assets.db")
# ─── Park emoji+display suffix patterns ─────────────────────────────────────
# Maps disney_park code → emoji+display name suffix patterns at end of names.
PARK_EMOJI_PATTERNS = [
(r'[ \t]*\U0001f3e8\s+\S.*$', 'park_resort'), # 🏨 Resort
(r'[ \t]*\U0001f30d\s+\S.*$', 'park_epcot'), # 🌍 Epcot
(r'[ \t]*\U0001f3f0\s+\S.*$', 'park_mk'), # 🏰 Magic Kingdom
(r'[ \t]*\U0001f33f\s+\S.*$', 'park_ak'), # 🌿 Animal Kingdom
(r'[ \t]*\U0001f3ac\s+\S.*$', 'park_hs'), # 🎬 Hollywood Studios
(r'[ \t]*\U0001f6cd\ufe0f?\s+\S.*$', 'park_ds'), # 🛍️ Disney Springs
(r'[ \t]*\U0001f4cd\s+\S.*$', 'park_other'), # 📍 Other
(r'[ \t]*\U0001f3e2\s+\S.*$', 'park_office'), # 🏢 DRC
]
PARK_SUFFIX_RE = re.compile(
'|'.join(f'(?P<{code}>{pat})' for pat, code in PARK_EMOJI_PATTERNS),
re.UNICODE,
)
# ─── Floor regex patterns ──────────────────────────────────────────────────
# These match floor info in names like "4TH FLO", "1st Floor", "2nd Fl", "FL06"
# Only strip when the floor NUMBER in the name matches the floor column.
FLOOR_FULL_RE = re.compile(
r'\b(\d+)(?:ST|ND|RD|TH|st|nd|rd|th)\s+'
r'(?:FL(?:OOR)?|Floor|floor|FLO(?:OR)?|Floo|floo|FLR)\b',
re.IGNORECASE,
)
FLOOR_SHORT_RE = re.compile(r'\b(\d+)\s+fl\b', re.IGNORECASE)
# FL06, FL6WEST, FL 06 — digits after FL, 1-2 digits, word boundary or non-alpha
FLOOR_CODE_RE = re.compile(r'\bFL\s?0?(\d{1,2})(?!\d)', re.IGNORECASE)
# ─── Building number regex patterns ────────────────────────────────────────
BLDG_RE = re.compile(r'\b(?:BLDG|Bldg|Building)\s+(\d+)\b', re.IGNORECASE)
HASH_NUM_RE = re.compile(r'#\s*(\d{3,})\b')
# ─── Room/suite patterns ───────────────────────────────────────────────────
ROOM_RE = re.compile(r'\b(?:Room|room)\s+(\d+)\b')
SUITE_RE = re.compile(r'\b(?:SUITE|Suite|suite|Ste)\s+(\d+)\b')
# ─── Trailer patterns ──────────────────────────────────────────────────────
TRAILER_RE = re.compile(r'\b(?:Trailer|trailer)\s+(\d+)\b')
# ─── Known hotel/resort/venue names ─────────────────────────────────────────
# These appear in names but are NOT address fragments — they're venue names.
VENUE_NAMES = {
"Port Orleans", "French Quarter", "RIVERSIDE", "RIVER SIDE",
"Caribbean Beach", "CBR",
"Coronado Springs", "CSR",
"All Star Movies", "All Star Music", "All Star Sports",
"Art of Animation", "AoA",
"Pop Century",
"Animal Kingdom Lodge", "AKL", "Jambo", "Kidani",
"Wilderness Lodge", "WL", "Boulder Ridge", "Copper Creek",
"Contemporary", "Bay Lake",
"Polynesian", "Poly",
"Grand Floridian",
"Boardwalk", "BWV",
"Beach Club", "Yacht Club",
"Old Key West", "OKW",
"Saratoga Springs", "SSR",
"Riviera",
"Grand Destino",
"Shades of Green", "SOG",
"Four Seasons",
"Hilton",
"Wyndham",
"Marriott",
}
# Street suffix type words (don't strip these alone)
STREET_SUFFIXES = {
"Dr", "Drive", "Rd", "Road", "St", "Street",
"Blvd", "Boulevard", "Ave", "Avenue",
"Way", "Trl", "Trail", "Pkwy", "Parkway",
"Cir", "Circle", "Ct", "Court", "Ln", "Lane",
"Pl", "Place", "Hwy", "Highway",
}
# Directional prefixes
DIRECTIONALS = {"N", "S", "E", "W", "North", "South", "East", "West"}
def _normalise_street(name: str) -> str:
"""Normalise a street name for fuzzy comparison.
Strips directionals, standardises suffixes, lower-cases.
E.g. 'E Buena Vista Dr''buena vista'
'1251 Riverside Dr''riverside'
"""
parts = name.replace('.', '').split()
# Strip number prefix
while parts and not parts[0][0].isalpha():
parts = parts[1:]
# Strip directional prefix
if parts and parts[0] in DIRECTIONALS:
parts = parts[1:]
# Strip street suffix
if parts and parts[-1] in STREET_SUFFIXES:
parts = parts[:-1]
return ' '.join(p.lower() for p in parts if len(p) > 1 or p in ('N', 'S', 'E', 'W'))
def _ratio(a: str, b: str) -> float:
"""SequenceMatcher similarity ratio."""
return SequenceMatcher(None, a.lower(), b.lower()).ratio()
# Words that should NOT be part of an address street fragment
# (these are area/function words that follow numbers in names)
NON_STREET_WORDS = {
'BREAKROOM', 'BRKRM', 'BREAK', 'BREAK_ROOM', 'BREAK-ROOM',
'VENDING', 'CAFETERIA', 'CAFE', 'MAINTENANCE', 'WAREHOUSE',
'WARDROBE', 'LAUNDRY', 'LOBBY', 'GUEST', 'GUES',
'ADMIN', 'OFFICE', 'STOP', 'STATION', 'BUS',
'PARKING', 'SECURITY', 'GATE', 'ENTRY', 'ENTRANCE',
'LEFT', 'RIGHT', 'REAR', 'FRONT', 'BACK', 'BACKSTAGE',
'NORTH', 'SOUTH', 'EAST', 'WEST', 'WINTER', 'SUMMER', 'SPRING', 'FALL',
'TOWER', 'WING', 'ELEV', 'ELEVATOR', 'SPORTS', 'SPORT',
'NEAR', 'BY', 'AT', 'THE', 'AND', 'OF',
'BLDG', 'BUILDING', 'BLD',
}
def _find_number_street_pairs(name: str):
"""Find all (number, street_name) pairs in a name.
Yields (number, street_fragment, start, end) tuples.
Uses regex to capture number + street-name words (max 4) that stop
before a delimiter (hyphen, known separator, non-street word, or end).
E.g. 'Justin M - 1536 Buena Vista-Vending' → ('1536', '1536 Buena Vista', ...)
"""
# Build a lookahead that stops on: dash/separator, non-street word, uppercase (likely area name)
# We capture: number + up to 4 words that look like street names
# Stop conditions (in lookahead): dash | end | non-street word | single letter | 2+ uppercase
stop_words = '|'.join(sorted(NON_STREET_WORDS, key=len, reverse=True))
# Pattern: number followed by 1-4 street-name words, stopping at delimiter or non-street word
# A "street word" is a word that starts with uppercase letter and is not a known stop word
pat = re.compile(
r'\b(\d{3,})\s+'
r'((?:(?![A-Z]+(?:' + stop_words + r')\b)[A-Z][a-zA-Z.\']*\s+){1,4}?)'
r'(?=\s*[-–—]|\s*$|\s+(?:' + stop_words + r')\b)',
)
for m in pat.finditer(name):
num = m.group(1)
street = m.group(2).strip()
if street:
# Strip trailing hyphen fragments that got captured
street = re.sub(r'\s*[-–—]\s*.*$', '', street).strip()
if street:
yield num, street, m.start(), m.end()
def strip_park_suffix(name: str, disney_park: Optional[str]) -> str:
"""Strip the park emoji+display suffix from the end of a name."""
if not disney_park:
return name
m = PARK_SUFFIX_RE.search(name)
if m:
return name[:m.start()].rstrip()
return name
def strip_g_separator(name: str) -> str:
"""Strip the Cantaloupe 'G-' separator artifact.
Pattern: ' - G-Description' → the ' - G-' part is a Cantaloupe
convention where 'G-' separates contact from address from description.
Only matches when G- is preceded by dash/space (not mid-word like 'Brenda G').
"""
# Match: optional space/dash, then 'G-' at word start
name = re.sub(r'(?<![A-Za-z])\s*[-–—]?\s*G-\s*', ' ', name)
return name.strip()
def strip_floor(name: str, floor: str) -> str:
"""Strip floor info from name when the floor is in the floor column."""
if not floor:
return name
floor_num = floor.strip()
if not floor_num.isdigit():
return name
# Strip: "Nth FL" / "Nth FLOOR" patterns with matching number
# We only strip when the ordinal number matches floor_num
name = FLOOR_FULL_RE.sub(
lambda m: '' if m.group(1) == floor_num else m.group(0),
name,
)
# Strip: "N fl" pattern
def _short_repl(m):
return '' if m.group(1) == floor_num else m.group(0)
name = FLOOR_SHORT_RE.sub(_short_repl, name, count=1)
# Strip: FL0N or FLN code pattern
def _code_repl(m):
return '' if m.group(1) == floor_num else m.group(0)
name = FLOOR_CODE_RE.sub(_code_repl, name, count=1)
return name.strip()
def strip_building_number(name: str, building_number: str) -> str:
"""Strip building number from name when it's in building_number column."""
if not building_number:
return name
bn = building_number.strip()
if not bn:
return name
# Strip BLDG N, Building N, Bldg N
def _bldg_repl(m):
return '' if m.group(1) == bn else m.group(0)
name = BLDG_RE.sub(_bldg_repl, name)
# Strip #N patterns
def _hash_repl(m):
return '' if m.group(1) == bn else m.group(0)
name = HASH_NUM_RE.sub(_hash_repl, name)
return name.strip()
def strip_building_name_prefix(name: str, building_name: str) -> str:
"""Strip duplicated building name at the START of the asset name.
E.g. 'BLDG 215 4TH FLO - vending area''vending area'
when building_name is 'BLDG 215 4TH FLO'.
Only strips at the start when the name has additional content after the
building name with a separator.
"""
if not building_name or not name:
return name
bn_lower = building_name.lower().strip()
name_lower = name.lower().strip()
# Check if name STARTS WITH building_name followed by a separator
if name_lower.startswith(bn_lower):
remainder = name[len(building_name):].strip()
if remainder and remainder[0] in '-–—':
remainder = remainder[1:].strip()
if remainder:
return remainder
return name
def strip_address_from_name(name: str, address: str) -> str:
"""Strip address-like fragments from name when they overlap with address.
Uses fuzzy matching: extracts number+street pairs from the name and
compares them against the structured address. Only strips pairs where
the street-name components have significant overlap.
Rule #8: Only strip content that appears in a structured column.
We verify that the street name (not just the number) is in the address.
"""
if not address or not name:
return name
addr_normalised = _normalise_street(address)
if not addr_normalised:
return name
# Try exact match of address in name first (case-insensitive)
addr_lower = address.lower().strip()
name_lower = name.lower()
if addr_lower in name_lower:
idx = name_lower.index(addr_lower)
return (name[:idx] + name[idx + len(addr_lower):]).strip()
# Try fuzzy matching on number+street pairs
for num, street_frag, start, end in _find_number_street_pairs(name):
frag_normalised = _normalise_street(f'{num} {street_frag}')
# Check if the normalised street name overlaps with address
street_normalised = _normalise_street(street_frag)
if not street_normalised:
continue
# Match if the normalised street fragment has significant overlap with address
if _ratio(street_normalised, addr_normalised) >= 0.5:
# Also check: number must be in address or the street match is very strong
if num in address or _ratio(street_normalised, addr_normalised) >= 0.7:
name = (name[:start] + name[end:]).strip()
break
return name.strip()
def strip_epcot_prefix(name: str, disney_park: Optional[str]) -> str:
"""Strip 'Epcot-' prefix from asset names (redundant with disney_park)."""
if disney_park != 'epcot':
return name
name = re.sub(r'^Epcot\s*[-–—]\s*', '', name)
return name.strip()
def strip_room(name: str, room: str) -> str:
"""Strip room/suite number from name when it's in the room column."""
if not room:
return name
r = room.strip()
if not r:
return name
def _room_repl(m):
return '' if m.group(1) == r else m.group(0)
name = ROOM_RE.sub(_room_repl, name)
name = SUITE_RE.sub(_room_repl, name)
return name.strip()
def strip_trailer(name: str, trailer: str) -> str:
"""Strip trailer number from name when it's in the trailer_number column."""
if not trailer:
return name
t = trailer.strip()
if not t:
return name
def _trl_repl(m):
return '' if m.group(1) == t else m.group(0)
name = TRAILER_RE.sub(_trl_repl, name)
return name.strip()
def clean_name(name: str, row: dict) -> str:
"""Clean an asset name by stripping data now in structured fields.
Args:
name: Current asset name.
row: Dict with keys: address, floor, building_number, building_name,
room, trailer_number, disney_park.
Returns:
Cleaned name string.
"""
original = name
# Phase 1: Strip park emoji+display suffix (Disney assets)
name = strip_park_suffix(name, row.get('disney_park'))
# Phase 2: Strip Cantaloupe "G-" separator artifact
name = strip_g_separator(name)
# Phase 3: Strip 'Epcot-' prefix (redundant with disney_park)
name = strip_epcot_prefix(name, row.get('disney_park'))
# Phase 4: Strip duplicated building name prefix
# E.g. 'BLDG 215 4TH FLO - vending area' when building_name is 'BLDG 215 4TH FLO'
old_name = name
name = strip_building_name_prefix(name, row.get('building_name', ''))
building_prefix_stripped = (name != old_name)
# Phase 5: Strip address from name (only if street name matches address col)
name = strip_address_from_name(name, row.get('address', ''))
# Phase 6: Strip floor info (only if floor number matches)
name = strip_floor(name, row.get('floor', ''))
# Phase 7: Strip building number (only with BLDG/Building prefix match)
name = strip_building_number(name, row.get('building_number', ''))
# Phase 8: Strip room (only if room number matches)
name = strip_room(name, row.get('room', ''))
# Phase 9: Strip trailer (only if trailer number matches)
name = strip_trailer(name, row.get('trailer_number', ''))
# Phase 10: Final cleanup
# Collapse multiple spaces
name = re.sub(r'[ \t]+', ' ', name)
# Clean up "--", "- -" etc. (dedup separators)
name = re.sub(r'\s*[-–—]\s*[-–—]\s*', ' - ', name)
name = re.sub(r'\s*[-–—]\s+[-–—]\s*', ' - ', name)
# Remove leading/trailing hyphens and spaces
name = name.strip(' -–—')
name = name.strip()
# Collapse spaces again after separator cleanup
name = re.sub(r'[ \t]+', ' ', name)
# Phase 11: Handle degenerate names
# If the name is now empty or very short, fall back logically
if not name or len(name) < 2:
if building_prefix_stripped and row.get('building_name'):
# We stripped the building prefix, so the building_name IS the name
name = row['building_name']
elif row.get('building_name'):
name = row['building_name']
elif row.get('address'):
name = row['address']
else:
name = original.strip()
# If name is still just a building identifier (e.g. "Building 1180"),
# that's fine — it's readable and useful.
name = name.strip()
# Final length sanity: if name got too short but original was meaningful
if len(name) < 3 and len(original) > 5 and not building_prefix_stripped:
name = original.strip()
return name
# ─── Park display name mapping (for reporting only) ─────────────────────────
DISPLAY_NAMES = {
'resort': 'Resort',
'epcot': 'Epcot',
'magic-kingdom': 'Magic Kingdom',
'animal-kingdom': 'Animal Kingdom',
'hollywood-studios': 'Hollywood Studios',
'disney-springs': 'Disney Springs',
'office': 'DRC',
'other': 'Other',
}
def process_assets(conn: sqlite3.Connection, apply: bool = False):
"""Process all assets and clean their names."""
cur = conn.cursor()
cur.execute(
"SELECT id, name, address, floor, building_number, building_name, "
"room, trailer_number, disney_park FROM assets ORDER BY id"
)
rows = [dict(r) for r in cur.fetchall()]
stats = {
'total': len(rows),
'cleaned': 0,
'unchanged': 0,
'disney_stripped': 0,
'non_disney_stripped': 0,
}
updates = []
for row in rows:
name = row['name']
cleaned = clean_name(name, row)
if cleaned != name:
stats['cleaned'] += 1
if row.get('disney_park'):
stats['disney_stripped'] += 1
else:
stats['non_disney_stripped'] += 1
updates.append((cleaned, row['id']))
else:
stats['unchanged'] += 1
# Apply updates
if apply:
cur.executemany(
"UPDATE assets SET name = ? WHERE id = ?",
updates,
)
conn.commit()
return stats, updates
def print_sample_changes(updates: list, original_data: dict):
"""Print a sample of changes for verification."""
print(f"\n{'' * 72}")
print("Sample changes (first 25):")
print(f"{'' * 72}")
for i, (new_name, asset_id) in enumerate(updates[:25]):
old_name = original_data[asset_id]
print(f"\n [{asset_id}]")
print(f" Before: {old_name}")
print(f" After: {new_name}")
def main():
parser = argparse.ArgumentParser(
description="Clean up asset names by stripping data now in structured fields",
)
parser.add_argument(
"--apply", action="store_true",
help="Actually update the database (default: dry-run only)",
)
args = parser.parse_args()
print(f"DB: {DB_PATH}")
print(f"Mode: {'APPLY' if args.apply else 'DRY-RUN'}")
print()
conn = sqlite3.connect(DB_PATH)
conn.execute("PRAGMA journal_mode=WAL")
conn.row_factory = sqlite3.Row
# Capture original names BEFORE modifying anything
cur = conn.cursor()
cur.execute("SELECT id, name FROM assets")
original_data = {r['id']: r['name'] for r in cur.fetchall()}
stats, updates = process_assets(conn, apply=args.apply)
print(f"{'=' * 60}")
print(f" Results")
print(f"{'=' * 60}")
print(f" Total assets: {stats['total']}")
print(f" Names cleaned: {stats['cleaned']}")
print(f" Disney assets: {stats['disney_stripped']}")
print(f" Non-Disney assets: {stats['non_disney_stripped']}")
print(f" Unchanged: {stats['unchanged']}")
if updates:
print(f"\n Change rate: {len(updates) / stats['total'] * 100:.1f}%")
print_sample_changes(updates, original_data)
if args.apply:
print(f"\n{'=' * 60}")
print("✅ Changes applied to database!")
print(f"{'=' * 60}")
else:
print(f"\n{'' * 60}")
print("💡 Dry-run complete. Run with --apply to write changes.")
print(f"{'' * 60}")
conn.close()
if __name__ == '__main__':
main()