320 lines
9.7 KiB
Python
320 lines
9.7 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Seed the database with initial development data.
|
||
|
||
Creates:
|
||
- Admin user (admin/admin123)
|
||
- Test vendors (TESTVENDOR, WIZAMART)
|
||
|
||
Usage:
|
||
python scripts/seed_database.py
|
||
|
||
This script is idempotent - safe to run multiple times.
|
||
"""
|
||
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
# Add project root to path
|
||
project_root = Path(__file__).parent.parent
|
||
sys.path.insert(0, str(project_root))
|
||
|
||
from sqlalchemy.orm import Session
|
||
from sqlalchemy import select
|
||
|
||
from app.core.database import SessionLocal, engine
|
||
from models.database.user import User
|
||
from models.database.vendor import Vendor
|
||
from middleware.auth import AuthManager
|
||
from datetime import datetime, timezone
|
||
|
||
# Default credentials
|
||
DEFAULT_ADMIN_EMAIL = "admin@wizamart.com"
|
||
DEFAULT_ADMIN_USERNAME = "admin"
|
||
DEFAULT_ADMIN_PASSWORD = "admin123" # Change in production!
|
||
|
||
|
||
def verify_database_ready() -> bool:
|
||
"""
|
||
Verify that database tables exist.
|
||
|
||
Returns:
|
||
bool: True if database is ready, False otherwise
|
||
"""
|
||
try:
|
||
with engine.connect() as conn:
|
||
from sqlalchemy import text
|
||
# Check for users table
|
||
result = conn.execute(
|
||
text("SELECT name FROM sqlite_master WHERE type='table' AND name='users'")
|
||
)
|
||
users_table = result.fetchall()
|
||
|
||
# Check for vendors table
|
||
result = conn.execute(
|
||
text("SELECT name FROM sqlite_master WHERE type='table' AND name='vendors'")
|
||
)
|
||
vendors_table = result.fetchall()
|
||
|
||
return len(users_table) > 0 and len(vendors_table) > 0
|
||
|
||
except Exception as e:
|
||
print(f"❌ Error checking database: {e}")
|
||
return False
|
||
|
||
|
||
def create_admin_user(db: Session, auth_manager: AuthManager) -> tuple[User, bool]:
|
||
"""
|
||
Create admin user if it doesn't exist.
|
||
|
||
Args:
|
||
db: Database session
|
||
auth_manager: AuthManager instance for password hashing
|
||
|
||
Returns:
|
||
tuple: (User object, was_created boolean)
|
||
"""
|
||
# Check if admin already exists
|
||
existing_admin = db.execute(
|
||
select(User).where(User.username == DEFAULT_ADMIN_USERNAME)
|
||
).scalar_one_or_none()
|
||
|
||
if existing_admin:
|
||
return existing_admin, False
|
||
|
||
# Create new admin user
|
||
admin = User(
|
||
email=DEFAULT_ADMIN_EMAIL,
|
||
username=DEFAULT_ADMIN_USERNAME,
|
||
hashed_password=auth_manager.hash_password(DEFAULT_ADMIN_PASSWORD),
|
||
role="admin",
|
||
is_active=True,
|
||
created_at=datetime.now(timezone.utc),
|
||
updated_at=datetime.now(timezone.utc)
|
||
)
|
||
|
||
db.add(admin)
|
||
db.flush() # Get the ID
|
||
|
||
return admin, True
|
||
|
||
|
||
def create_vendor(
|
||
db: Session,
|
||
vendor_code: str,
|
||
name: str,
|
||
subdomain: str,
|
||
owner_user_id: int,
|
||
description: str = None
|
||
) -> tuple[Vendor, bool]:
|
||
"""
|
||
Create vendor if it doesn't exist.
|
||
|
||
Args:
|
||
db: Database session
|
||
vendor_code: Unique vendor code
|
||
name: Vendor name
|
||
subdomain: Subdomain for the vendor
|
||
owner_user_id: ID of the owner user
|
||
description: Optional description
|
||
|
||
Returns:
|
||
tuple: (Vendor object, was_created boolean)
|
||
"""
|
||
# Check if vendor already exists
|
||
existing_vendor = db.execute(
|
||
select(Vendor).where(Vendor.vendor_code == vendor_code)
|
||
).scalar_one_or_none()
|
||
|
||
if existing_vendor:
|
||
return existing_vendor, False
|
||
|
||
# Create new vendor
|
||
vendor = Vendor(
|
||
vendor_code=vendor_code,
|
||
subdomain=subdomain,
|
||
name=name,
|
||
description=description or f"{name} - Development vendor",
|
||
owner_user_id=owner_user_id,
|
||
contact_email=f"contact@{subdomain}.com",
|
||
contact_phone="+352 123 456 789",
|
||
website=f"https://{subdomain}.com",
|
||
is_active=True,
|
||
is_verified=True,
|
||
created_at=datetime.now(timezone.utc),
|
||
updated_at=datetime.now(timezone.utc)
|
||
)
|
||
|
||
db.add(vendor)
|
||
db.flush()
|
||
|
||
return vendor, True
|
||
|
||
|
||
def seed_database():
|
||
"""Main seeding function."""
|
||
|
||
print("\n" + "╔" + "═" * 68 + "╗")
|
||
print("║" + " " * 18 + "DATABASE SEEDING SCRIPT" + " " * 27 + "║")
|
||
print("╚" + "═" * 68 + "╝\n")
|
||
|
||
# ========================================================================
|
||
# STEP 1: VERIFY DATABASE
|
||
# ========================================================================
|
||
|
||
print("STEP 1: Verifying database...")
|
||
|
||
if not verify_database_ready():
|
||
print("\n❌ ERROR: Database not ready!")
|
||
print("\n Required tables don't exist yet.")
|
||
print(" Please run database migrations first:\n")
|
||
print(" alembic upgrade head\n")
|
||
sys.exit(1)
|
||
|
||
print(" ✓ Database tables verified\n")
|
||
|
||
# ========================================================================
|
||
# STEP 2: CREATE ADMIN USER
|
||
# ========================================================================
|
||
|
||
db = SessionLocal()
|
||
auth_manager = AuthManager()
|
||
|
||
try:
|
||
print("STEP 2: Creating admin user...")
|
||
|
||
admin, admin_created = create_admin_user(db, auth_manager)
|
||
|
||
if admin_created:
|
||
print(f" ✓ Admin user created (ID: {admin.id})")
|
||
print(f" Username: {admin.username}")
|
||
print(f" Email: {admin.email}")
|
||
else:
|
||
print(f" ℹ️ Admin user already exists (ID: {admin.id})")
|
||
print(f" Username: {admin.username}")
|
||
print(f" Email: {admin.email}")
|
||
|
||
# ====================================================================
|
||
# STEP 3: CREATE VENDORS
|
||
# ====================================================================
|
||
|
||
print("\nSTEP 3: Creating vendors...")
|
||
|
||
# Create TESTVENDOR
|
||
test_vendor, test_created = create_vendor(
|
||
db,
|
||
vendor_code="TESTVENDOR",
|
||
name="Test Vendor",
|
||
subdomain="testvendor",
|
||
owner_user_id=admin.id,
|
||
description="Development test vendor for general testing"
|
||
)
|
||
|
||
if test_created:
|
||
print(f" ✓ TESTVENDOR created (ID: {test_vendor.id})")
|
||
else:
|
||
print(f" ℹ️ TESTVENDOR already exists (ID: {test_vendor.id})")
|
||
|
||
# Create WIZAMART
|
||
wizamart, wizamart_created = create_vendor(
|
||
db,
|
||
vendor_code="WIZAMART",
|
||
name="WizaMart",
|
||
subdomain="wizamart",
|
||
owner_user_id=admin.id,
|
||
description="Primary development vendor for theme customization"
|
||
)
|
||
|
||
if wizamart_created:
|
||
print(f" ✓ WIZAMART created (ID: {wizamart.id})")
|
||
else:
|
||
print(f" ℹ️ WIZAMART already exists (ID: {wizamart.id})")
|
||
|
||
# Commit all changes
|
||
db.commit()
|
||
|
||
# ====================================================================
|
||
# SUMMARY
|
||
# ====================================================================
|
||
|
||
print("\n" + "╔" + "═" * 68 + "╗")
|
||
print("║" + " " * 22 + "SEEDING COMPLETE!" + " " * 28 + "║")
|
||
print("╚" + "═" * 68 + "╝\n")
|
||
|
||
if admin_created or test_created or wizamart_created:
|
||
print("✅ New items created:")
|
||
if admin_created:
|
||
print(" • Admin user")
|
||
if test_created:
|
||
print(" • TESTVENDOR")
|
||
if wizamart_created:
|
||
print(" • WIZAMART")
|
||
else:
|
||
print("ℹ️ All items already existed - no changes made")
|
||
|
||
print("\n" + "─" * 70)
|
||
print("📝 ADMIN LOGIN CREDENTIALS")
|
||
print("─" * 70)
|
||
print(f" URL: http://localhost:8000/admin/login")
|
||
print(f" Username: {DEFAULT_ADMIN_USERNAME}")
|
||
print(f" Password: {DEFAULT_ADMIN_PASSWORD}")
|
||
print("─" * 70)
|
||
|
||
print("\n" + "─" * 70)
|
||
print("🏪 VENDORS")
|
||
print("─" * 70)
|
||
|
||
vendors = db.execute(select(Vendor)).scalars().all()
|
||
for v in vendors:
|
||
print(f"\n {v.vendor_code}")
|
||
print(f" Name: {v.name}")
|
||
print(f" Subdomain: {v.subdomain}.wizamart.com")
|
||
print(f" Theme URL: http://localhost:8000/admin/vendors/{v.vendor_code}/theme")
|
||
print(f" Verified: {'✓' if v.is_verified else '✗'}")
|
||
print(f" Active: {'✓' if v.is_active else '✗'}")
|
||
|
||
print("\n" + "─" * 70)
|
||
print("🚀 NEXT STEPS")
|
||
print("─" * 70)
|
||
print(" 1. Start the server:")
|
||
print(" python -m uvicorn app.main:app --reload\n")
|
||
print(" 2. Login to admin panel:")
|
||
print(" http://localhost:8000/admin/login\n")
|
||
print(" 3. Test theme editor:")
|
||
print(" http://localhost:8000/admin/vendors/WIZAMART/theme\n")
|
||
|
||
if admin_created:
|
||
print("⚠️ SECURITY: Change the default admin password after first login!")
|
||
|
||
print()
|
||
|
||
except Exception as e:
|
||
db.rollback()
|
||
print("\n" + "╔" + "═" * 68 + "╗")
|
||
print("║" + " " * 28 + "ERROR" + " " * 35 + "║")
|
||
print("╚" + "═" * 68 + "╝\n")
|
||
print(f"❌ Failed to seed database: {e}")
|
||
print(f"\nError type: {type(e).__name__}")
|
||
print("\nCommon issues:")
|
||
print(" - Database migrations not run (run: alembic upgrade head)")
|
||
print(" - Database connection issues")
|
||
print(" - Foreign key constraint violations")
|
||
print()
|
||
import traceback
|
||
traceback.print_exc()
|
||
sys.exit(1)
|
||
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
seed_database()
|
||
except KeyboardInterrupt:
|
||
print("\n\n⚠️ Seeding interrupted by user")
|
||
sys.exit(1)
|
||
except Exception as e:
|
||
print(f"\n❌ Fatal error: {e}")
|
||
sys.exit(1)
|