"""Seed script — users, units, and categories. Called on first startup via lifespan hook. """ from __future__ import annotations import json import logging from pathlib import Path from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.core.security import hash_password from app.models.unit import Unit from app.models.user import User from app.models.category import Category logger = logging.getLogger(__name__) # ── Seed user data (dicts to avoid module-level model instantiation) ─ SEED_USERS_DATA = [ {"email": "jerome@denya.com", "full_name": "Jerome Tabiri", "phone": "+233000000001", "role": "Admin/Jerome"}, {"email": "wahab@denya.com", "full_name": "Wahab", "phone": "+233000000002", "role": "Admin/Wahab"}, {"email": "bella@denya.com", "full_name": "Bella", "phone": "+233000000003", "role": "CS Rep"}, {"email": "akua@denya.com", "full_name": "Akua", "phone": "+233000000004", "role": "CS Manager"}, {"email": "mercy@denya.com", "full_name": "Mercy Duah", "phone": "+233000000005", "role": "CS Manager"}, {"email": "ama@denya.com", "full_name": "Ama", "phone": "+233000000006", "role": "CS Rep"}, {"email": "nicholas@denya.com", "full_name": "Nicholas", "phone": "+233000000007", "role": "FM Dispatcher"}, {"email": "collins@denya.com", "full_name": "Collins", "phone": "+233000000008", "role": "FM Dispatcher"}, {"email": "prosper@denya.com", "full_name": "Prosper", "phone": "+233000000010", "role": "Tech"}, {"email": "sam@denya.com", "full_name": "Sam", "phone": "+233000000011", "role": "Tech"}, {"email": "steven@denya.com", "full_name": "Steven", "phone": "+233000000012", "role": "Tech"}, {"email": "junior@denya.com", "full_name": "Junior (Samuel)", "phone": "+233000000013", "role": "Tech"}, {"email": "francis@denya.com", "full_name": "Francis", "phone": "+233000000014", "role": "Tech"}, {"email": "desmond@denya.com", "full_name": "Desmond Afful", "phone": "+233000000015", "role": "Tech"}, {"email": "afful@denya.com", "full_name": "Afful", "phone": "+233000000016", "role": "Tech"}, {"email": "scott@denya.com", "full_name": "Scott Murray", "phone": "+233000000020", "role": "CEO"}, {"email": "director@denya.com", "full_name": "Director", "phone": "+233000000021", "role": "Director"}, ] async def seed_users(db: AsyncSession, default_password: str = "denya123") -> list[User]: """Insert seed users if they don't already exist.""" hashed = hash_password(default_password) created: list[User] = [] for data in SEED_USERS_DATA: result = await db.execute(select(User).where(User.email == data["email"])) if result.scalar_one_or_none() is None: user = User( email=data["email"], password_hash=hashed, full_name=data["full_name"], phone=data["phone"], role=data["role"], ) db.add(user) created.append(user) if created: await db.flush() for u in created: await db.refresh(u) logger.info("Seeded %d users", len(created)) else: logger.info("All seed users already exist") return created async def seed_units(db: AsyncSession, json_path: str | Path | None = None) -> list[Unit]: """Insert units from *apartment_mapping.json*. Falls back to a small built-in set if the file is not found. """ units_data: list[dict] = [] if json_path: p = Path(json_path) if p.exists(): with open(p) as f: raw = json.load(f) units_data = raw if isinstance(raw, list) else raw.get("units", []) logger.info("Loaded %d units from %s", len(units_data), p) if not units_data: # Built-in fallback for Pavilion Accra logger.warning("apartment_mapping.json not found; using built-in fallback") for wing in ("East", "West"): for floor in range(1, 11): for num in range(1, 7): code = f"{floor:02d}{num}{wing[0]}" units_data.append({ "apartment_code": code, "property": wing, "building": f"Pavilion {wing}", "floor": floor, }) created: list[Unit] = [] for entry in units_data: code = entry.get("apartment_code") or entry.get("unit") or "" if not code: continue result = await db.execute(select(Unit).where(Unit.apartment_code == code)) if result.scalar_one_or_none() is None: unit = Unit( property=entry.get("property", "East"), apartment_code=code, building=entry.get("building"), floor=entry.get("floor"), ) db.add(unit) created.append(unit) if created: await db.flush() for u in created: await db.refresh(u) logger.info("Seeded %d units", len(created)) else: logger.info("All units already exist") return created # ── Category taxonomy (PRD §7) ─────────────────────────────────────── # Top-level categories with sub-categories SEED_CATEGORIES_DATA: list[dict] = [ # ── Maintenance ────────────────────────────────────────────── {"type": "maintenance", "name": "Plumbing", "subs": ["WC not flushing", "Leaky tap", "Pipe burst", "Blocked drain", "Water heater"]}, {"type": "maintenance", "name": "Electrical", "subs": ["No power", "Tripped breaker", "Light fitting", "Socket repair", "Voltage issues"]}, {"type": "maintenance", "name": "HVAC", "subs": ["AC not cooling", "AC leaking", "Thermostat", "Fan not working"]}, {"type": "maintenance", "name": "Furniture", "subs": ["Broken bed", "Broken chair", "Broken table", "Wardrobe issue"]}, {"type": "maintenance", "name": "Appliances", "subs": ["Fridge", "Washing machine", "Microwave", "TV", "Kettle"]}, {"type": "maintenance", "name": "Cleaning", "subs": ["Deep clean requested", "Post-checkout turnover", "Common area cleaning"]}, {"type": "maintenance", "name": "Pest Control", "subs": ["Insects", "Rodents", "Fumigation"]}, {"type": "maintenance", "name": "Security", "subs": ["Lock broken", "Door not closing", "Window latch", "CCTV issue"]}, {"type": "maintenance", "name": "Internet", "subs": ["WiFi down", "Slow speed", "Router reset"]}, {"type": "maintenance", "name": "Structural", "subs": ["Wall crack", "Ceiling leak", "Floor tile", "Paint touch-up"]}, # ── Customer Service ───────────────────────────────────────── {"type": "cs", "name": "Check-in", "subs": ["Early check-in", "Key handover", "Welcome instructions"]}, {"type": "cs", "name": "Check-out", "subs": ["Late checkout", "Key return", "Inspection"]}, {"type": "cs", "name": "Housekeeping", "subs": ["Mid-stay cleaning", "Linen change", "Restocking"]}, {"type": "cs", "name": "Lost Property", "subs": ["Guest left items behind"]}, {"type": "cs", "name": "Billing", "subs": ["Invoice question", "Payment issue", "Deposit query"]}, {"type": "cs", "name": "Staff Behaviour", "subs": ["Staff conduct feedback"]}, # ── Emergency ──────────────────────────────────────────────── {"type": "emergency", "name": "Fire", "subs": ["Smoke detected", "Fire alarm", "Sprinkler issue"], "sla_urgency": "urgent"}, {"type": "emergency", "name": "Flood", "subs": ["Major water leak", "Burst pipe", "Overflowing"], "sla_urgency": "urgent"}, {"type": "emergency", "name": "Gas Leak", "subs": ["Gas smell", "Suspected leak"], "sla_urgency": "urgent"}, {"type": "emergency", "name": "Electrical Hazard", "subs": ["Sparking", "Exposed wires", "Power outage multiple units"], "sla_urgency": "urgent"}, ] async def seed_categories(db: AsyncSession) -> list[Category]: """Seed the categories table with the full PRD §7 taxonomy.""" created: list[Category] = [] for group in SEED_CATEGORIES_DATA: # Check if parent category exists result = await db.execute( select(Category).where( Category.type == group["type"], Category.name == group["name"], Category.parent_id.is_(None), ) ) parent = result.scalar_one_or_none() if parent is None: parent = Category( type=group["type"], name=group["name"], parent_id=None, sla_urgency=group.get("sla_urgency"), ) db.add(parent) await db.flush() created.append(parent) # Seed sub-categories for sub_name in group["subs"]: result = await db.execute( select(Category).where( Category.type == group["type"], Category.name == sub_name, Category.parent_id == parent.id, ) ) existing = result.scalar_one_or_none() if existing is None: sub = Category( type=group["type"], name=sub_name, parent_id=parent.id, sla_urgency=None, ) db.add(sub) created.append(sub) if created: await db.flush() for c in created: await db.refresh(c) logger.info("Seeded %d categories", len(created)) else: logger.info("All seed categories already exist") return created