poduck/smoothschedule
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
79b76bf2dcec70703e33f4d77aa1e14c2f15d288
smoothschedule/smoothschedule/smoothschedule/scheduling/schedule/management/commands/reseed_demo.py
poduck 79b76bf2dc Add demo tenant reseed, staff roles, and fix masquerade redirect 
Demo Tenant:
- Add block_emails field to Tenant model for demo accounts
- Add is_email_blocked() and wrapper functions in email_service
- Create reseed_demo management command with salon/spa theme
- Add Celery beat task for daily reseed at midnight UTC
- Create 100 appointments, 20 customers, 13 services, 12 resources

Staff Roles:
- Add StaffRole model with permission toggles
- Create default roles: Full Access, Front Desk, Limited Staff
- Add StaffRolesSettings page and hooks
- Integrate role assignment in Staff management

Bug Fixes:
- Fix masquerade redirect using wrong role names (tenant_owner vs owner)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-16 15:20:59 -05:00

703 lines
29 KiB
Python
Raw Blame History

"""
Daily Demo Tenant Reseed Command
Creates/reseeds a salon/spa themed demo tenant for sales demonstrations.
Designed to run daily at midnight UTC via Celery beat to keep appointments fresh.
Features:
- Salon/Spa themed business data (stylists, services, rooms)
- Pro subscription with all features enabled
- Email blocking enabled (no real emails sent)
- Appointments spanning 2 weeks past to 3 weeks future
- Sample automations installed
Usage:
python manage.py reseed_demo
python manage.py reseed_demo --quiet # Less output
python manage.py reseed_demo --appointments 150 # More appointments
"""
import random
from datetime import timedelta
from decimal import Decimal
from django.contrib.contenttypes.models import ContentType
from django.core.management.base import BaseCommand
from django.db import connection
from django.utils import timezone
from django_tenants.utils import schema_context, tenant_context
from smoothschedule.identity.core.models import Tenant, Domain
from smoothschedule.identity.users.models import User, StaffRole
from smoothschedule.scheduling.schedule.models import (
Event,
Participant,
Resource,
ResourceType,
Service,
ScheduledTask,
PluginTemplate,
PluginInstallation,
GlobalEventPlugin,
)
class Command(BaseCommand):
help = "Reseed demo tenant with fresh salon/spa data for sales demonstrations"
def add_arguments(self, parser):
parser.add_argument(
"--quiet",
action="store_true",
help="Reduce output verbosity",
)
parser.add_argument(
"--appointments",
type=int,
default=100,
help="Number of appointments to create (default: 100)",
)
def handle(self, *args, **options):
self.quiet = options.get("quiet", False)
self.appointment_count = options.get("appointments", 100)
if not self.quiet:
self.stdout.write("\n" + "=" * 70)
self.stdout.write(self.style.SUCCESS(" SERENITY SALON & SPA - DEMO RESEED"))
self.stdout.write("=" * 70 + "\n")
# Step 1: Get or create demo tenant
demo_tenant = self.setup_tenant()
# Step 2: Assign Pro subscription
self.assign_pro_subscription(demo_tenant)
# Step 3: Switch to tenant schema for tenant-specific data
with tenant_context(demo_tenant):
# Clear existing appointments
self.clear_appointments()
# Create or update tenant users
tenant_users = self.create_tenant_users(demo_tenant)
# Create or update resource types
resource_types = self.create_resource_types()
# Create or update services
services = self.create_services()
# Create or update resources
resources = self.create_resources(tenant_users, resource_types)
# Create or update customers
customers = self.create_customers(demo_tenant)
# Create fresh appointments
self.create_appointments(
resources=resources,
services=services,
customers=customers,
)
# Setup automations
self.setup_automations(tenant_users)
# Assign staff roles
self.assign_staff_roles(tenant_users)
if not self.quiet:
self.stdout.write("\n" + "=" * 70)
self.stdout.write(self.style.SUCCESS(" DEMO RESEED COMPLETE!"))
self.stdout.write("=" * 70)
self.stdout.write("\nAccess URL: http://demo.lvh.me:5173")
self.stdout.write("All passwords: test123\n")
def setup_tenant(self):
"""Get or create demo tenant with proper settings."""
if not self.quiet:
self.stdout.write("\n[1/9] Setting up Demo Tenant...")
# Get or create the demo tenant
# Note: Don't set branding colors yet - requires Pro subscription
try:
tenant = Tenant.objects.get(schema_name="demo")
# Update basic settings (not branding - that requires Pro)
tenant.name = "Serenity Salon & Spa"
tenant.timezone = "America/New_York"
tenant.block_emails = True
tenant.initial_setup_complete = True
tenant.save()
if not self.quiet:
self.stdout.write(f" {self.style.WARNING('UPDATED')} Demo tenant settings")
except Tenant.DoesNotExist:
tenant = Tenant.objects.create(
schema_name="demo",
name="Serenity Salon & Spa",
timezone="America/New_York",
block_emails=True,
initial_setup_complete=True,
)
if not self.quiet:
self.stdout.write(f" {self.style.SUCCESS('CREATED')} Demo tenant")
# Create domain
domain, created = Domain.objects.get_or_create(
domain="demo.lvh.me",
defaults={"tenant": tenant, "is_primary": True},
)
if created and not self.quiet:
self.stdout.write(f" {self.style.SUCCESS('CREATED')} Domain: demo.lvh.me")
return tenant
def assign_pro_subscription(self, tenant):
"""Assign Pro subscription to demo tenant."""
if not self.quiet:
self.stdout.write("\n[2/9] Assigning Pro Subscription...")
subscription_created = False
try:
from smoothschedule.billing.models import Subscription, Plan, PlanVersion
# Get Pro plan
try:
pro_plan = Plan.objects.get(code='pro')
pro_version = pro_plan.versions.filter(is_public=True).order_by('-created_at').first()
if pro_version:
subscription, created = Subscription.objects.update_or_create(
business=tenant,
defaults={
'plan_version': pro_version,
'status': 'active',
'current_period_start': timezone.now(),
'current_period_end': timezone.now() + timedelta(days=365),
}
)
subscription_created = True
status_str = self.style.SUCCESS('CREATED') if created else self.style.WARNING('UPDATED')
if not self.quiet:
self.stdout.write(f" {status_str} Pro subscription")
else:
if not self.quiet:
self.stdout.write(f" {self.style.WARNING('SKIPPED')} No Pro plan version found")
except Plan.DoesNotExist:
if not self.quiet:
self.stdout.write(f" {self.style.WARNING('SKIPPED')} Pro plan not found - run billing_seed_catalog first")
except ImportError:
if not self.quiet:
self.stdout.write(f" {self.style.WARNING('SKIPPED')} Billing models not available")
# Now that Pro subscription is assigned, we can set branding colors
# (requires white_label feature from Pro plan)
# Use direct update to bypass the save() permission check
# (The save check was designed for API updates, not management commands)
Tenant.objects.filter(pk=tenant.pk).update(
primary_color="#ec4899", # Pink
secondary_color="#f472b6", # Light pink
)
tenant.refresh_from_db()
if not self.quiet:
self.stdout.write(f" {self.style.SUCCESS('SET')} Branding colors (Pink theme)")
def create_tenant_users(self, tenant):
"""Create owner, manager, and staff users."""
if not self.quiet:
self.stdout.write("\n[3/9] Creating Users...")
users = {}
# Owner
owner_data = {
"username": "owner@demo.com",
"email": "owner@demo.com",
"first_name": "Victoria",
"last_name": "Stone",
"role": User.Role.TENANT_OWNER,
"tenant": tenant,
"phone": "555-100-0001",
}
owner, created = User.objects.get_or_create(
username=owner_data["username"],
defaults=owner_data,
)
if created:
owner.set_password("test123")
owner.save()
users["owner"] = owner
if not self.quiet:
status = self.style.SUCCESS("CREATED") if created else self.style.WARNING("EXISTS")
self.stdout.write(f" {status} {owner.email} (Owner)")
# Manager
manager_data = {
"username": "manager@demo.com",
"email": "manager@demo.com",
"first_name": "Marcus",
"last_name": "Chen",
"role": User.Role.TENANT_MANAGER,
"tenant": tenant,
"phone": "555-100-0002",
}
manager, created = User.objects.get_or_create(
username=manager_data["username"],
defaults=manager_data,
)
if created:
manager.set_password("test123")
manager.save()
users["manager"] = manager
if not self.quiet:
status = self.style.SUCCESS("CREATED") if created else self.style.WARNING("EXISTS")
self.stdout.write(f" {status} {manager.email} (Manager)")
# Staff members (stylists and spa therapists)
staff_data = [
{"first_name": "Sophia", "last_name": "Martinez", "title": "Senior Stylist"},
{"first_name": "Emma", "last_name": "Johnson", "title": "Stylist"},
{"first_name": "Olivia", "last_name": "Chen", "title": "Junior Stylist"},
{"first_name": "Isabella", "last_name": "Kim", "title": "Spa Therapist"},
{"first_name": "Mia", "last_name": "Taylor", "title": "Esthetician"},
]
staff_users = []
for staff in staff_data:
email = f"{staff['first_name'].lower()}.{staff['last_name'].lower()}@demo.com"
user_data = {
"username": email,
"email": email,
"first_name": staff["first_name"],
"last_name": staff["last_name"],
"role": User.Role.TENANT_STAFF,
"tenant": tenant,
"job_title": staff["title"],
}
user, created = User.objects.get_or_create(
username=email,
defaults=user_data,
)
if created:
user.set_password("test123")
user.save()
staff_users.append(user)
if not self.quiet:
status = self.style.SUCCESS("CREATED") if created else self.style.WARNING("EXISTS")
self.stdout.write(f" {status} {user.email} ({staff['title']})")
users["staff"] = staff_users
return users
def create_resource_types(self):
"""Create resource types for salon/spa."""
if not self.quiet:
self.stdout.write("\n[4/9] Creating Resource Types...")
types_data = [
{"name": "Stylist", "category": ResourceType.Category.STAFF, "description": "Hair stylists", "is_default": True},
{"name": "Spa Therapist", "category": ResourceType.Category.STAFF, "description": "Massage and spa specialists", "is_default": False},
{"name": "Station", "category": ResourceType.Category.OTHER, "description": "Hair styling stations", "is_default": False},
{"name": "Spa Room", "category": ResourceType.Category.OTHER, "description": "Private spa treatment rooms", "is_default": False},
{"name": "Equipment", "category": ResourceType.Category.OTHER, "description": "Shared equipment", "is_default": False},
]
resource_types = {}
for rt_data in types_data:
rt, created = ResourceType.objects.get_or_create(
name=rt_data["name"],
defaults=rt_data,
)
resource_types[rt_data["name"]] = rt
if not self.quiet:
status = self.style.SUCCESS("CREATED") if created else self.style.WARNING("EXISTS")
self.stdout.write(f" {status} {rt.name}")
return resource_types
def create_services(self):
"""Create salon/spa services."""
if not self.quiet:
self.stdout.write("\n[5/9] Creating Services...")
services_data = [
# Hair services
{"name": "Haircut & Style", "duration": 45, "price_cents": 6500, "description": "Precision cut with styling"},
{"name": "Hair Color - Full", "duration": 120, "price_cents": 15000, "description": "Full head color transformation"},
{"name": "Hair Color - Touch-up", "duration": 60, "price_cents": 8500, "description": "Root touch-up and refresh"},
{"name": "Blowout", "duration": 30, "price_cents": 4500, "description": "Professional blow dry and styling"},
{"name": "Deep Conditioning", "duration": 30, "price_cents": 3500, "description": "Intensive hair treatment"},
# Spa services
{"name": "Swedish Massage", "duration": 60, "price_cents": 9500, "description": "Relaxing full body massage"},
{"name": "Hot Stone Massage", "duration": 90, "price_cents": 12500, "description": "Therapeutic hot stone treatment"},
{"name": "Facial - Classic", "duration": 60, "price_cents": 8500, "description": "Deep cleansing facial"},
{"name": "Facial - Anti-Aging", "duration": 75, "price_cents": 11000, "description": "Advanced anti-aging treatment"},
# Nail services
{"name": "Manicure", "duration": 30, "price_cents": 3000, "description": "Classic nail care"},
{"name": "Pedicure", "duration": 45, "price_cents": 5000, "description": "Relaxing foot treatment"},
{"name": "Mani-Pedi Combo", "duration": 75, "price_cents": 7500, "description": "Complete hand and foot care"},
# Premium
{"name": "Bridal Package", "duration": 180, "price_cents": 35000, "description": "Complete bridal preparation", "variable_pricing": True, "deposit_amount_cents": 10000},
]
services = []
for i, svc_data in enumerate(services_data, 1):
svc_data["display_order"] = i
name = svc_data.pop("name")
service, created = Service.objects.get_or_create(
name=name,
defaults=svc_data,
)
services.append(service)
if not self.quiet:
price = (svc_data.get("price_cents", 0)) / 100
status = self.style.SUCCESS("CREATED") if created else self.style.WARNING("EXISTS")
self.stdout.write(f" {status} {name} ({svc_data.get('duration', 0)} min, ${price:.2f})")
return services
def create_resources(self, tenant_users, resource_types):
"""Create staff-linked and standalone resources."""
if not self.quiet:
self.stdout.write("\n[6/9] Creating Resources...")
resources = []
stylist_type = resource_types.get("Stylist")
spa_type = resource_types.get("Spa Therapist")
station_type = resource_types.get("Station")
spa_room_type = resource_types.get("Spa Room")
equipment_type = resource_types.get("Equipment")
# Staff-linked resources (stylists and spa therapists)
staff_users = tenant_users.get("staff", [])
for user in staff_users:
is_spa = "Spa" in (user.job_title or "") or "Esthetician" in (user.job_title or "")
resource_type = spa_type if is_spa else stylist_type
resource, created = Resource.objects.get_or_create(
user=user,
defaults={
"name": user.get_full_name(),
"description": user.job_title or "Staff member",
"resource_type": resource_type,
"type": Resource.Type.STAFF,
"max_concurrent_events": 1,
"user_can_edit_schedule": True,
},
)
resources.append(resource)
if not self.quiet:
status = self.style.SUCCESS("CREATED") if created else self.style.WARNING("EXISTS")
self.stdout.write(f" {status} {resource.name} (Staff)")
# Standalone resources - Hair Stations
for i in range(1, 4):
resource, created = Resource.objects.get_or_create(
name=f"Hair Station {i}",
defaults={
"description": f"Hair styling station #{i}",
"resource_type": station_type,
"type": Resource.Type.ROOM,
"max_concurrent_events": 1,
},
)
resources.append(resource)
if not self.quiet:
status = self.style.SUCCESS("CREATED") if created else self.style.WARNING("EXISTS")
self.stdout.write(f" {status} {resource.name} (Station)")
# Spa Rooms
for letter in ["A", "B"]:
resource, created = Resource.objects.get_or_create(
name=f"Spa Room {letter}",
defaults={
"description": f"Private spa treatment room {letter}",
"resource_type": spa_room_type,
"type": Resource.Type.ROOM,
"max_concurrent_events": 1,
},
)
resources.append(resource)
if not self.quiet:
status = self.style.SUCCESS("CREATED") if created else self.style.WARNING("EXISTS")
self.stdout.write(f" {status} {resource.name} (Spa Room)")
# Relaxation Lounge (multi-capacity)
resource, created = Resource.objects.get_or_create(
name="Relaxation Lounge",
defaults={
"description": "Shared relaxation area",
"resource_type": spa_room_type,
"type": Resource.Type.ROOM,
"max_concurrent_events": 5,
},
)
resources.append(resource)
if not self.quiet:
status = self.style.SUCCESS("CREATED") if created else self.style.WARNING("EXISTS")
self.stdout.write(f" {status} {resource.name} (Lounge)")
# Equipment
resource, created = Resource.objects.get_or_create(
name="Massage Chair",
defaults={
"description": "Portable massage chair",
"resource_type": equipment_type,
"type": Resource.Type.EQUIPMENT,
"max_concurrent_events": 1,
},
)
resources.append(resource)
if not self.quiet:
status = self.style.SUCCESS("CREATED") if created else self.style.WARNING("EXISTS")
self.stdout.write(f" {status} {resource.name} (Equipment)")
return resources
def create_customers(self, tenant):
"""Create customer users."""
if not self.quiet:
self.stdout.write("\n[7/9] Creating Customers...")
# Quick login customer
customer_demo, created = User.objects.get_or_create(
username="customer@demo.com",
defaults={
"email": "customer@demo.com",
"first_name": "Demo",
"last_name": "Customer",
"role": User.Role.CUSTOMER,
"tenant": tenant,
"phone": "555-200-0001",
},
)
if created:
customer_demo.set_password("test123")
customer_demo.save()
if not self.quiet:
status = self.style.SUCCESS("CREATED") if created else self.style.WARNING("EXISTS")
self.stdout.write(f" {status} {customer_demo.email} (Quick Login)")
customers = [customer_demo]
# Additional customers with salon-appropriate names
customer_data = [
("Jennifer", "Anderson", "jennifer.anderson@example.com"),
("Michelle", "Brooks", "michelle.brooks@example.com"),
("Amanda", "Clark", "amanda.clark@example.com"),
("Stephanie", "Davis", "stephanie.davis@example.com"),
("Nicole", "Evans", "nicole.evans@example.com"),
("Rachel", "Foster", "rachel.foster@example.com"),
("Lauren", "Garcia", "lauren.garcia@example.com"),
("Heather", "Hill", "heather.hill@example.com"),
("Kimberly", "Jackson", "kimberly.jackson@example.com"),
("Ashley", "King", "ashley.king@example.com"),
("Brittany", "Lewis", "brittany.lewis@example.com"),
("Tiffany", "Martin", "tiffany.martin@example.com"),
("Samantha", "Nelson", "samantha.nelson@example.com"),
("Christina", "Owens", "christina.owens@example.com"),
("Jessica", "Parker", "jessica.parker@example.com"),
("Elizabeth", "Quinn", "elizabeth.quinn@example.com"),
("Megan", "Roberts", "megan.roberts@example.com"),
("Sarah", "Smith", "sarah.smith@example.com"),
("Amber", "Thompson", "amber.thompson@example.com"),
]
for first_name, last_name, email in customer_data:
user, created = User.objects.get_or_create(
username=email,
defaults={
"email": email,
"first_name": first_name,
"last_name": last_name,
"role": User.Role.CUSTOMER,
"tenant": tenant,
},
)
if created:
user.set_password("test123")
user.save()
customers.append(user)
if not self.quiet:
self.stdout.write(f" {self.style.SUCCESS('READY')} {len(customers)} customers total")
return customers
def clear_appointments(self):
"""Clear existing appointments to prepare for fresh data."""
if not self.quiet:
self.stdout.write("\n Clearing existing appointments...")
deleted_participants = Participant.objects.all().delete()[0]
deleted_events = Event.objects.all().delete()[0]
if not self.quiet:
self.stdout.write(f" Deleted {deleted_events} events, {deleted_participants} participants")
def create_appointments(self, resources, services, customers):
"""Create fresh appointments spanning past and future dates."""
if not self.quiet:
self.stdout.write(f"\n[8/9] Creating {self.appointment_count} Appointments...")
# Filter to staff resources only for appointments
staff_resources = [r for r in resources if r.type == Resource.Type.STAFF]
if not staff_resources:
staff_resources = resources[:3]
resource_ct = ContentType.objects.get_for_model(Resource)
user_ct = ContentType.objects.get_for_model(User)
# Time range: 2 weeks ago to 3 weeks ahead
now = timezone.now()
start_date = now - timedelta(days=14)
end_date = now + timedelta(days=21)
days_range = (end_date - start_date).days
# Status weights: 60% scheduled, 25% completed, 10% canceled, 5% no-show
statuses = (
[Event.Status.SCHEDULED] * 60 +
[Event.Status.COMPLETED] * 25 +
[Event.Status.CANCELED] * 10 +
[Event.Status.NOSHOW] * 5
)
created_count = 0
for _ in range(self.appointment_count):
# Random date in range
random_day = random.randint(0, days_range - 1)
appointment_date = start_date + timedelta(days=random_day)
# Business hours: 9 AM - 7 PM
hour = random.randint(9, 18)
minute = random.choice([0, 15, 30, 45])
start_time = appointment_date.replace(hour=hour, minute=minute, second=0, microsecond=0)
# Pick random service, resource, customer
service = random.choice(services)
resource = random.choice(staff_resources)
customer = random.choice(customers)
# Determine status based on time
chosen_status = random.choice(statuses)
if start_time < now and chosen_status == Event.Status.SCHEDULED:
chosen_status = Event.Status.COMPLETED
elif start_time > now and chosen_status in [Event.Status.COMPLETED, Event.Status.NOSHOW]:
chosen_status = Event.Status.SCHEDULED
# Calculate end time
end_time = start_time + timedelta(minutes=service.duration)
# Create event
event = Event.objects.create(
title=f"{customer.get_full_name() or customer.email} - {service.name}",
start_time=start_time,
end_time=end_time,
status=chosen_status,
service=service,
notes=f"Service: {service.name}",
)
# Create resource participant
Participant.objects.create(
event=event,
role=Participant.Role.RESOURCE,
content_type=resource_ct,
object_id=resource.id,
)
# Create customer participant
Participant.objects.create(
event=event,
role=Participant.Role.CUSTOMER,
content_type=user_ct,
object_id=customer.id,
)
created_count += 1
if not self.quiet:
self.stdout.write(f" {self.style.SUCCESS('CREATED')} {created_count} appointments")
# Summary
scheduled = Event.objects.filter(status=Event.Status.SCHEDULED).count()
completed = Event.objects.filter(status=Event.Status.COMPLETED).count()
canceled = Event.objects.filter(status=Event.Status.CANCELED).count()
noshow = Event.objects.filter(status=Event.Status.NOSHOW).count()
self.stdout.write(f" Scheduled: {scheduled}")
self.stdout.write(f" Completed: {completed}")
self.stdout.write(f" Canceled: {canceled}")
self.stdout.write(f" No-show: {noshow}")
def setup_automations(self, tenant_users):
"""Setup sample automations and scheduled tasks."""
if not self.quiet:
self.stdout.write("\n[9/9] Setting up Automations...")
owner = tenant_users.get("owner")
try:
# Create scheduled tasks for demo (if the automation system is available)
# Daily Report Task
task, created = ScheduledTask.objects.get_or_create(
name="Daily Business Report",
defaults={
"description": "Send daily summary to owner",
"plugin_name": "daily_report",
"plugin_config": {"recipients": ["owner@demo.com"], "include_upcoming": True},
"schedule_type": ScheduledTask.ScheduleType.CRON,
"cron_expression": "0 8 * * *",
"status": ScheduledTask.Status.ACTIVE,
},
)
if not self.quiet:
status = self.style.SUCCESS("CREATED") if created else self.style.WARNING("EXISTS")
self.stdout.write(f" {status} Daily Report task")
# Weekly Cleanup Task
task, created = ScheduledTask.objects.get_or_create(
name="Weekly Cleanup",
defaults={
"description": "Clean up old completed appointments",
"plugin_name": "cleanup_old_events",
"plugin_config": {"days_old": 90, "dry_run": False},
"schedule_type": ScheduledTask.ScheduleType.CRON,
"cron_expression": "0 2 * * 0",
"status": ScheduledTask.Status.ACTIVE,
},
)
if not self.quiet:
status = self.style.SUCCESS("CREATED") if created else self.style.WARNING("EXISTS")
self.stdout.write(f" {status} Weekly Cleanup task")
except Exception as e:
if not self.quiet:
self.stdout.write(f" {self.style.WARNING('SKIPPED')} Automations setup: {e}")
def assign_staff_roles(self, tenant_users):
"""Assign staff roles to demo staff members."""
staff_users = tenant_users.get("staff", [])
# Role assignments: first gets Full Access, some get Front Desk, rest get Limited
role_assignments = {
0: "Full Access Staff", # Sophia
1: "Front Desk", # Emma
2: "Limited Staff", # Olivia
3: "Front Desk", # Isabella
4: "Limited Staff", # Mia
}
for i, user in enumerate(staff_users):
role_name = role_assignments.get(i, "Limited Staff")
try:
# Get tenant from user
if user.tenant:
role = StaffRole.objects.filter(
tenant=user.tenant,
name=role_name
).first()
if role:
user.staff_role = role
user.save(update_fields=["staff_role"])
except Exception:
pass # Staff roles may not be set up
Reference in New Issue View Git Blame Copy Permalink