"""Seed script - populate the database with demo data.
DO NOT run seed.py --force on production. Use migration scripts for schema changes.
Safety features:
--dry-run Show what would be affected (table row counts), touch nothing
--force --yes Required BOTH to actually wipe and re-seed (backup is taken first)
--backup-only Create a timestamped backup of instance/auth.db and exit
--restore <file> Restore instance/auth.db from a backup file and exit
"""
import sys
import os
import random
import shutil
import subprocess
from datetime import datetime, timezone, timedelta
sys.path.insert(0, os.path.dirname(__file__))
# Force instance/auth.db (matches production service) ā override shell env
db_path = os.path.join(os.path.abspath(os.path.dirname(__file__)), 'instance', 'auth.db')
os.environ['DATABASE_URL'] = 'sqlite:///' + db_path
from app import create_app
from app.models import (
db, User, UserCompany, Company, Project, KPIValue, Forecast,
Goal, GoalMetric, CoachingAssignment, CoachingScorecard,
)
def backup_database():
"""Copy instance/auth.db to a timestamped backup file. Returns backup path or None."""
if not os.path.exists(db_path):
print(f"No database found at {db_path} ā nothing to back up.")
return None
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
backup_path = f"{db_path}.backup.{timestamp}"
shutil.copy2(db_path, backup_path)
print(f"Backup created: {backup_path}")
return backup_path
def restore_database(backup_file):
"""Restore instance/auth.db from a backup file."""
if not os.path.exists(backup_file):
print(f"ERROR: Backup file not found: {backup_file}")
sys.exit(1)
if os.path.exists(db_path):
# Safety: back up the current DB before overwriting it with the restore
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
pre_restore = f"{db_path}.pre-restore.{timestamp}"
shutil.copy2(db_path, pre_restore)
print(f"Current database saved to: {pre_restore}")
shutil.copy2(backup_file, db_path)
print(f"Restored {db_path} from {backup_file}")
def is_production():
"""Return True if this looks like a production environment."""
if os.environ.get('FLASK_ENV') == 'production':
return True
# PID file check for the command-sovereignty service
pid_file = os.path.join(os.path.abspath(os.path.dirname(__file__)), 'command-sovereignty.pid')
if os.path.exists(pid_file):
return True
# systemd user service check
try:
result = subprocess.run(
['systemctl', '--user', 'is-active', 'command-sovereignty'],
capture_output=True, text=True, timeout=5,
)
if result.stdout.strip() == 'active':
return True
except (FileNotFoundError, subprocess.TimeoutExpired):
pass # systemctl unavailable ā fall through
return False
def dry_run():
"""Print row counts per table without modifying anything."""
app = create_app()
with app.app_context():
print("DRY RUN ā no data will be modified.\n")
print("Tables that would be wiped by --force --yes:")
total = 0
for table in reversed(db.metadata.sorted_tables):
count = db.session.execute(
db.select(db.func.count()).select_from(table)
).scalar()
total += count
print(f" {table.name:<24} {count} row(s)")
print(f"\nTotal rows that would be deleted: {total}")
def seed(force=False):
app = create_app()
with app.app_context():
# Safety check: count existing profiles
from app.models import User
existing_count = User.query.count()
if existing_count > 0 and not force:
print(f"\nā ļø WARNING: Database already has {existing_count} profile(s).")
print("Running seed.py will DELETE ALL EXISTING DATA including users you added.")
print("\nOptions:")
print(" python seed.py ā aborted (would delete data)")
print(" python seed.py --dry-run ā show what would be affected")
print(" python seed.py --force --yes ā wipe everything and re-seed (backup taken first)")
print("\nTo add users without wiping, use the app UI or create them directly.")
sys.exit(1)
# Backup before any destructive operation
if existing_count > 0 or os.path.exists(db_path):
backup_database()
# Clear existing data only if force=True or DB is empty
print("Clearing existing data...")
for table in reversed(db.metadata.sorted_tables):
db.session.execute(table.delete())
db.session.commit()
# Create Super Admin (grepples15@gmail.com)
print("Creating super admin user...")
super_admin = User(
email='grepples15@gmail.com',
full_name='Vincent Grepples',
company='Command Sovereignty',
phone='555-0123',
role='super_admin',
is_active=True,
)
super_admin.set_password(os.environ.get("SEED_SUPER_ADMIN_PASSWORD", "temp_pw_2026"))
db.session.add(super_admin)
# Create demo user ā basic demo experience, NOT super admin
print("Creating demo user...")
demo = User(
email='demo@commandcenter.com',
full_name='Demo User',
company='Demo Company',
role='user',
is_active=True,
)
demo.set_password(os.environ.get("SEED_DEMO_PASSWORD", "CmdSov-Demo-Xh1F_yVwAhs4"))
db.session.add(demo)
# Create sample users at different company-level roles
print("Creating sample company users...")
sample_users = [
User(
email='owner@greenfield.example.com',
full_name='Sarah Mitchell',
company='Greenfield Landscaping',
role='partner',
is_active=True,
),
User(
email='admin@greenfield.example.com',
full_name='James Carter',
company='Greenfield Landscaping',
role='user',
is_active=True,
),
User(
email='manager@greenfield.example.com',
full_name='Lisa Torres',
company='Greenfield Landscaping',
role='user',
is_active=True,
),
User(
email='rep@greenfield.example.com',
full_name='Mike Johnson',
company='Greenfield Landscaping',
role='user',
is_active=True,
),
User(
email='viewer@greenfield.example.com',
full_name='Emily Davis',
company='Greenfield Landscaping',
role='user',
is_active=True,
),
]
for u in sample_users:
u.set_password(os.environ.get("SEED_USER_PASSWORD", "password_2026")) # FIX: actual env var value, was a literal string
db.session.add(u)
db.session.commit()
# Create companies
print("Creating demo companies...")
companies = [
Company(
name='Greenfield Landscaping',
industry='Landscaping',
size='multi',
target_revenue=2500000,
annual_revenue=2100000,
website='https://greenfield.example.com',
),
Company(
name='Metro Roofing Solutions',
industry='Roofing',
size='single',
target_revenue=1800000,
annual_revenue=1500000,
website='https://metro-roofing.example.com',
),
Company(
name='Premier HVAC Services',
industry='HVAC',
size='enterprise',
target_revenue=3200000,
annual_revenue=2800000,
website='https://premier-hvac.example.com',
),
]
for company in companies:
db.session.add(company)
db.session.commit()
# Link super admin to all companies (as owner)
for company in companies:
uc = UserCompany(
user_id=super_admin.id,
company_id=company.id,
role='owner',
)
db.session.add(uc)
db.session.commit()
# Create a dedicated Demo Company for the demo account
demo_company = Company(
name='Demo Company',
industry='Home Improvement',
size='multi',
target_revenue=1000000,
annual_revenue=850000,
website='https://demo.example.com',
)
db.session.add(demo_company)
db.session.commit()
# Link demo user as OWNER of Demo Company
uc_demo = UserCompany(
user_id=demo.id,
company_id=demo_company.id,
role='owner',
)
db.session.add(uc_demo)
# Link sample users to Greenfield Landscaping with different company roles
if len(companies) > 0 and len(sample_users) == 5:
company_roles = ['owner', 'admin', 'manager', 'member', 'viewer']
for i, user in enumerate(sample_users):
uc = UserCompany(
user_id=user.id,
company_id=companies[0].id,
role=company_roles[i],
)
db.session.add(uc)
db.session.commit()
# Create forecasts and KPIs for each company
print("Creating forecast and KPI data...")
now = datetime.now(timezone.utc)
start_year = now.replace(month=1, day=1, hour=0, minute=0, second=0, microsecond=0)
for company in companies:
monthly_target = company.target_revenue / 12
for month in range(1, now.month + 1):
period_start = start_year.replace(month=month)
period_end = period_start + timedelta(days=30)
# Forecast data
actual = monthly_target * random.uniform(0.8, 1.15)
projected = monthly_target
forecast = Forecast(
company_id=company.id,
forecast_type='revenue',
period='monthly',
period_start=period_start,
period_end=period_end,
projected_value=projected,
actual_value=actual,
confidence=0.85,
)
db.session.add(forecast)
# KPI values
kpis = [
KPIValue(
company_id=company.id,
kpi_name='total_revenue',
category='revenue',
period='monthly',
period_start=period_start,
period_end=period_end,
value=round(actual, 2),
unit='USD',
),
KPIValue(
company_id=company.id,
kpi_name='leads_generated',
category='marketing',
period='monthly',
period_start=period_start,
period_end=period_end,
value=round(random.uniform(50, 120)),
unit='count',
),
KPIValue(
company_id=company.id,
kpi_name='conversion_rate',
category='sales',
period='monthly',
period_start=period_start,
period_end=period_end,
value=round(random.uniform(15, 35), 1),
unit='percent',
),
KPIValue(
company_id=company.id,
kpi_name='customer_satisfaction',
category='customer',
period='monthly',
period_start=period_start,
period_end=period_end,
value=round(random.uniform(4.0, 4.8), 1),
unit='score',
),
]
for kpi in kpis:
db.session.add(kpi)
db.session.commit()
# Create projects
print("Creating projects...")
projects = [
Project(
company_id=companies[0].id,
name='Website Redesign',
description='Complete website overhaul with new lead capture system',
status='active',
budget=25000,
start_date=now - timedelta(days=30),
end_date=now + timedelta(days=60),
priority='high',
),
Project(
company_id=companies[0].id,
name='CRM Integration',
description='Connect Salesforce with existing point-of-sale systems',
status='active',
budget=15000,
start_date=now - timedelta(days=10),
end_date=now + timedelta(days=30),
priority='medium',
),
Project(
company_id=companies[1].id,
name='Marketing Automation',
description='Setup email marketing and retargeting campaigns',
status='active',
budget=12000,
start_date=now - timedelta(days=20),
end_date=now + timedelta(days=45),
priority='high',
),
Project(
company_id=companies[2].id,
name='Fleet Expansion',
description='Add 5 new service vehicles with GPS tracking',
status='planning',
budget=180000,
start_date=now,
end_date=now + timedelta(days=90),
priority='low',
),
]
for project in projects:
db.session.add(project)
db.session.commit()
# Create goals
print("Creating goals...")
# --- Org-level goals (level='org') ---
org_goals = [
Goal(
company_id=companies[0].id,
name='Increase Q3 Revenue by 15%',
description='Grow revenue through new service offerings',
target_value=287500,
current_value=187000,
unit='$',
status='active',
start_date=now - timedelta(days=30),
end_date=now + timedelta(days=60),
level='org',
),
Goal(
company_id=companies[0].id,
name='Launch Commercial Division',
description='Expand into commercial landscaping market',
target_value=5,
current_value=2,
unit='#',
status='active',
start_date=now - timedelta(days=60),
end_date=now + timedelta(days=120),
level='org',
),
Goal(
company_id=companies[1].id,
name='Reduce Customer Acquisition Cost',
description='Lower CAC by 20% through organic growth',
target_value=250,
current_value=290,
unit='$',
status='active',
start_date=now - timedelta(days=30),
end_date=now + timedelta(days=30),
level='org',
),
Goal(
company_id=companies[2].id,
name='Expand to 6 Locations',
description='Open new service location in downtown area',
target_value=6,
current_value=5,
unit='#',
status='active',
start_date=now - timedelta(days=90),
end_date=now + timedelta(days=180),
level='org',
),
]
for goal in org_goals:
db.session.add(goal)
db.session.commit()
# --- Department-level goals under "Increase Q3 Revenue by 15%" ---
revenue_goal = org_goals[0] # Increase Q3 Revenue by 15%
dept_goals = [
Goal(
company_id=companies[0].id,
name='Sales Team Revenue Target',
description='Revenue target for the sales department',
target_value=150000,
current_value=98000,
unit='$',
status='active',
start_date=now - timedelta(days=30),
end_date=now + timedelta(days=60),
level='department',
parent_goal_id=revenue_goal.id,
),
Goal(
company_id=companies[0].id,
name='Marketing Lead Gen Target',
description='Lead generation target for marketing',
target_value=500,
current_value=320,
unit='#',
status='active',
start_date=now - timedelta(days=30),
end_date=now + timedelta(days=60),
level='department',
parent_goal_id=revenue_goal.id,
),
Goal(
company_id=companies[0].id,
name='Operations Efficiency',
description='Maintain service level agreements',
target_value=95,
current_value=88,
unit='%',
status='active',
start_date=now - timedelta(days=30),
end_date=now + timedelta(days=60),
level='department',
parent_goal_id=revenue_goal.id,
),
]
for goal in dept_goals:
db.session.add(goal)
db.session.commit()
# --- Team-level goals under "Sales Team Revenue Target" ---
sales_dept = dept_goals[0] # Sales Team Revenue Target
team_goals = [
Goal(
company_id=companies[0].id,
name='Residential Sales',
description='Residential landscaping revenue target',
target_value=100000,
current_value=65000,
unit='$',
status='active',
start_date=now - timedelta(days=30),
end_date=now + timedelta(days=60),
level='team',
parent_goal_id=sales_dept.id,
),
Goal(
company_id=companies[0].id,
name='Commercial Sales',
description='Commercial landscaping revenue target',
target_value=50000,
current_value=33000,
unit='$',
status='active',
start_date=now - timedelta(days=30),
end_date=now + timedelta(days=60),
level='team',
parent_goal_id=sales_dept.id,
),
]
for goal in team_goals:
db.session.add(goal)
db.session.commit()
# --- Rep-level goals ---
residential_team = team_goals[0] # Residential Sales
commercial_team = team_goals[1] # Commercial Sales
# sample_users: [0]=owner (Sarah Mitchell), [1]=admin (James Carter),
# [2]=manager (Lisa Torres), [3]=rep (Mike Johnson), [4]=viewer (Emily Davis)
rep_goals = [
Goal(
company_id=companies[0].id,
name='Mike Johnson - Residential',
description='Individual residential sales target',
target_value=60000,
current_value=40000,
unit='$',
status='active',
start_date=now - timedelta(days=30),
end_date=now + timedelta(days=60),
level='rep',
parent_goal_id=residential_team.id,
assigned_to=sample_users[3].id, # Mike Johnson (rep)
),
Goal(
company_id=companies[0].id,
name='Emily Davis - Residential',
description='Individual residential sales target',
target_value=40000,
current_value=25000,
unit='$',
status='active',
start_date=now - timedelta(days=30),
end_date=now + timedelta(days=60),
level='rep',
parent_goal_id=residential_team.id,
assigned_to=sample_users[4].id, # Emily Davis (viewer)
),
Goal(
company_id=companies[0].id,
name='James Carter - Commercial',
description='Individual commercial sales target',
target_value=30000,
current_value=20000,
unit='$',
status='active',
start_date=now - timedelta(days=30),
end_date=now + timedelta(days=60),
level='rep',
parent_goal_id=commercial_team.id,
assigned_to=sample_users[1].id, # James Carter (admin)
),
Goal(
company_id=companies[0].id,
name='Lisa Torres - Commercial',
description='Individual commercial sales target',
target_value=20000,
current_value=13000,
unit='$',
status='active',
start_date=now - timedelta(days=30),
end_date=now + timedelta(days=60),
level='rep',
parent_goal_id=commercial_team.id,
assigned_to=sample_users[2].id, # Lisa Torres (manager)
),
]
for goal in rep_goals:
db.session.add(goal)
db.session.commit()
# --- Department goals under "Launch Commercial Division" ---
launch_goal = org_goals[1] # Launch Commercial Division
launch_dept_goals = [
Goal(
company_id=companies[0].id,
name='Commercial Marketing',
description='Lead generation for commercial division',
target_value=50,
current_value=30,
unit='#',
status='active',
start_date=now - timedelta(days=60),
end_date=now + timedelta(days=120),
level='department',
parent_goal_id=launch_goal.id,
),
Goal(
company_id=companies[0].id,
name='Commercial Operations',
description='Set up new commercial crews',
target_value=3,
current_value=1,
unit='#',
status='active',
start_date=now - timedelta(days=60),
end_date=now + timedelta(days=120),
level='department',
parent_goal_id=launch_goal.id,
),
]
for goal in launch_dept_goals:
db.session.add(goal)
db.session.commit()
# Collect all goals for metric creation
all_goals = org_goals + dept_goals + team_goals + rep_goals + launch_dept_goals
# Add goal metrics
for goal in all_goals:
metric = GoalMetric(
goal_id=goal.id,
metric_name='progress',
target_value=goal.target_value,
actual_value=goal.current_value,
unit=goal.unit,
period='monthly',
period_start=goal.start_date,
period_end=goal.end_date,
)
db.session.add(metric)
db.session.commit()
# Create coaching assignments
print("Creating coaching assignments...")
sessions = [
CoachingAssignment(
company_id=companies[0].id,
coach_id=super_admin.id,
rep_id=super_admin.id,
focus_area='Revenue Review',
description='Analysis of Q2 performance and Q3 planning',
start_date=now - timedelta(days=7),
end_date=now - timedelta(days=7),
status='completed',
),
CoachingAssignment(
company_id=companies[1].id,
coach_id=super_admin.id,
rep_id=super_admin.id,
focus_area='Marketing Strategy',
description='Develop new marketing approach for summer season',
start_date=now - timedelta(days=14),
end_date=now - timedelta(days=14),
status='completed',
),
CoachingAssignment(
company_id=companies[0].id,
coach_id=super_admin.id,
rep_id=super_admin.id,
focus_area='Operations Efficiency',
description='Review operational bottlenecks and optimization opportunities',
start_date=now + timedelta(days=7),
end_date=now + timedelta(days=7),
status='active',
),
]
for session in sessions:
db.session.add(session)
db.session.commit()
print(f"\nSeed complete!")
print(f" Super Admin: grepples15@gmail.com / {os.environ.get('SEED_SUPER_ADMIN_PASSWORD', 'temp_pw_2026')}")
print(f" Demo: demo@commandcenter.com / {os.environ.get('SEED_DEMO_PASSWORD', 'CmdSov-Demo-Xh1F_yVwAhs4')}")
print(f" Sample company users (owner/admin/manager/rep/viewer):")
for u in sample_users:
print(f" {u.email} / {os.environ.get('SEED_USER_PASSWORD', 'password_2026')}")
print(f" Companies: {len(companies)}")
print(f" Projects: {len(projects)}")
print(f" Goals: {len(all_goals)}")
print(f" Coaching: {len(sessions)}")
print(f" Forecasts: ~{now.month * len(companies)} monthly records")
print(f" KPI Records: ~{now.month * len(companies) * 4} records")
if __name__ == '__main__':
args = sys.argv[1:]
# --restore <backup_file>: restore DB from a backup and exit
if '--restore' in args:
idx = args.index('--restore')
if idx + 1 >= len(args):
print("ERROR: --restore requires a backup file path. Usage: python seed.py --restore <backup_file>")
sys.exit(1)
restore_database(args[idx + 1])
sys.exit(0)
# --backup-only: just create a backup and exit
if '--backup-only' in args:
backup_database()
sys.exit(0)
# --dry-run: show what would be affected, touch nothing
if '--dry-run' in args:
dry_run()
sys.exit(0)
force = '--force' in args
yes = '--yes' in args
if force:
# Production guard: never wipe a production database
if is_production():
print("Refusing to wipe production database. Run migrations instead.")
sys.exit(1)
# Explicit confirmation: --force alone is not enough
if not yes:
print("This will DELETE ALL DATA including users, companies, forecasts, and goals. Use --force --yes to confirm.")
sys.exit(1)
seed(force=force)