sboulahtit/orion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
ba130d4171cbc9215dd55c39a72cbefcca6e1194
orion/scripts/seed/init_production.py
Samir Boulahtit e9253fbd84 refactor: rename Wizamart to Orion across entire codebase 
Replace all ~1,086 occurrences of Wizamart/wizamart/WIZAMART/WizaMart
with Orion/orion/ORION across 184 files. This includes database
identifiers, email addresses, domain references, R2 bucket names,
DNS prefixes, encryption salt, Celery app name, config defaults,
Docker configs, CI configs, documentation, seed data, and templates.

Renames homepage-wizamart.html template to homepage-orion.html.
Fixes duplicate file_pattern key in api.yaml architecture rule.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-14 16:46:56 +01:00

649 lines
20 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""
Production Database Initialization for Orion Platform
This script initializes ESSENTIAL data required for production:
- Platform admin user
- Default store roles and permissions
- Admin settings
- Platform configuration
This is SAFE to run in production and should be run after migrations.
Usage:
make init-prod
This script is idempotent - safe to run multiple times.
"""
import sys
from datetime import UTC, datetime
from pathlib import Path
# Add project root to path
project_root = Path(__file__).parent.parent
sys.path.insert(0, str(project_root))
import contextlib
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.core.config import (
print_environment_info,
settings,
validate_production_settings,
)
from app.core.database import SessionLocal
from app.core.environment import is_production
from app.modules.billing.models.subscription import SubscriptionTier
from app.modules.tenancy.models import AdminSetting, Platform, User
from app.modules.tenancy.services.permission_discovery_service import (
permission_discovery_service,
)
from middleware.auth import AuthManager
# Register all models with SQLAlchemy so string-based relationships resolve
for _mod in [
"app.modules.billing.models",
"app.modules.inventory.models",
"app.modules.cart.models",
"app.modules.messaging.models",
"app.modules.loyalty.models",
"app.modules.catalog.models",
"app.modules.customers.models",
"app.modules.orders.models",
"app.modules.marketplace.models",
"app.modules.cms.models",
]:
with contextlib.suppress(ImportError):
__import__(_mod)
# =============================================================================
# HELPER FUNCTIONS
# =============================================================================
def print_header(text: str):
"""Print formatted header."""
print("\n" + "=" * 70)
print(f" {text}")
print("=" * 70)
def print_step(step: int, text: str):
"""Print step indicator."""
print(f"\n[{step}] {text}")
def print_success(text: str):
"""Print success message."""
print(f" ✓ {text}")
def print_warning(text: str):
"""Print warning message."""
print(f" ⚠ {text}")
def print_error(text: str):
"""Print error message."""
print(f" ✗ {text}")
# =============================================================================
# INITIALIZATION FUNCTIONS
# =============================================================================
def create_admin_user(db: Session, auth_manager: AuthManager) -> User:
"""Create or get the platform admin user."""
# Check if admin already exists
admin = db.execute(
select(User).where(User.email == settings.admin_email)
).scalar_one_or_none()
if admin:
print_warning(f"Admin user already exists: {admin.email}")
return admin
# Create new admin
hashed_password = auth_manager.hash_password(settings.admin_password)
admin = User(
username=settings.admin_username,
email=settings.admin_email,
hashed_password=hashed_password,
role="admin",
is_super_admin=True,
first_name=settings.admin_first_name,
last_name=settings.admin_last_name,
is_active=True,
is_email_verified=True,
created_at=datetime.now(UTC),
updated_at=datetime.now(UTC),
)
db.add(admin)
db.flush()
print_success(f"Created admin user: {admin.email}")
return admin
def create_default_platforms(db: Session) -> list[Platform]:
"""Create all default platforms (OMS, Main, Loyalty+)."""
platform_defs = [
{
"code": "oms",
"name": "Orion OMS",
"description": "Order Management System for multi-store e-commerce",
"domain": settings.platform_domain,
"path_prefix": "oms",
"default_language": "fr",
"supported_languages": ["fr", "de", "en"],
"settings": {},
"theme_config": {},
},
{
"code": "main",
"name": "Orion",
"description": "Main marketing site showcasing all Orion platforms",
"domain": "orion.lu",
"path_prefix": None,
"default_language": "fr",
"supported_languages": ["fr", "de", "en"],
"settings": {"is_marketing_site": True},
"theme_config": {"primary_color": "#2563EB", "secondary_color": "#3B82F6"},
},
{
"code": "loyalty",
"name": "Loyalty+",
"description": "Customer loyalty program platform for Luxembourg businesses",
"domain": "loyalty.lu",
"path_prefix": "loyalty",
"default_language": "fr",
"supported_languages": ["fr", "de", "en"],
"settings": {"features": ["points", "rewards", "tiers", "analytics"]},
"theme_config": {"primary_color": "#8B5CF6", "secondary_color": "#A78BFA"},
},
]
platforms = []
for pdef in platform_defs:
existing = db.execute(
select(Platform).where(Platform.code == pdef["code"])
).scalar_one_or_none()
if existing:
print_warning(f"Platform already exists: {existing.name} ({existing.code})")
platforms.append(existing)
continue
platform = Platform(
code=pdef["code"],
name=pdef["name"],
description=pdef["description"],
domain=pdef["domain"],
path_prefix=pdef["path_prefix"],
default_language=pdef["default_language"],
supported_languages=pdef["supported_languages"],
settings=pdef["settings"],
theme_config=pdef["theme_config"],
is_active=True,
is_public=True,
created_at=datetime.now(UTC),
updated_at=datetime.now(UTC),
)
db.add(platform)
db.flush()
platforms.append(platform)
print_success(f"Created platform: {platform.name} ({platform.code})")
return platforms
def create_default_role_templates(db: Session) -> dict:
"""Create default role templates (not tied to any store).
These serve as templates that can be copied when creating store-specific roles.
Note: Actual roles are store-specific and created when stores are onboarded.
"""
print(" Available role presets:")
print(" - Manager: Nearly full access (except team management)")
print(" - Staff: Day-to-day operations")
print(" - Support: Customer service focused")
print(" - Viewer: Read-only access")
print(" - Marketing: Marketing and customer communication")
print_success("Role templates ready for store onboarding")
return {
"manager": list(permission_discovery_service.get_preset_permissions("manager")),
"staff": list(permission_discovery_service.get_preset_permissions("staff")),
"support": list(permission_discovery_service.get_preset_permissions("support")),
"viewer": list(permission_discovery_service.get_preset_permissions("viewer")),
"marketing": list(permission_discovery_service.get_preset_permissions("marketing")),
}
def create_admin_settings(db: Session) -> int:
"""Create essential admin settings."""
settings_created = 0
# Essential platform settings
default_settings = [
{
"key": "platform_name",
"value": settings.project_name,
"value_type": "string",
"description": "Platform name displayed in admin panel",
"is_public": True,
},
{
"key": "platform_url",
"value": f"https://{settings.platform_domain}",
"value_type": "string",
"description": "Main platform URL",
"is_public": True,
},
{
"key": "support_email",
"value": f"support@{settings.platform_domain}",
"value_type": "string",
"description": "Platform support email",
"is_public": True,
},
{
"key": "max_stores_per_user",
"value": str(settings.max_stores_per_user),
"value_type": "integer",
"description": "Maximum stores a user can own",
"is_public": False,
},
{
"key": "max_team_members_per_store",
"value": str(settings.max_team_members_per_store),
"value_type": "integer",
"description": "Maximum team members per store",
"is_public": False,
},
{
"key": "invitation_expiry_days",
"value": str(settings.invitation_expiry_days),
"value_type": "integer",
"description": "Days until team invitation expires",
"is_public": False,
},
{
"key": "require_store_verification",
"value": "true",
"value_type": "boolean",
"description": "Require admin verification for new stores",
"is_public": False,
},
{
"key": "allow_custom_domains",
"value": str(settings.allow_custom_domains).lower(),
"value_type": "boolean",
"description": "Allow stores to use custom domains",
"is_public": False,
},
{
"key": "maintenance_mode",
"value": "false",
"value_type": "boolean",
"description": "Enable maintenance mode",
"is_public": True,
},
# Logging settings
{
"key": "log_level",
"value": "INFO",
"value_type": "string",
"category": "logging",
"description": "Application log level (DEBUG, INFO, WARNING, ERROR, CRITICAL)",
"is_public": False,
},
{
"key": "log_file_max_size_mb",
"value": "10",
"value_type": "integer",
"category": "logging",
"description": "Maximum log file size in MB before rotation",
"is_public": False,
},
{
"key": "log_file_backup_count",
"value": "5",
"value_type": "integer",
"category": "logging",
"description": "Number of rotated log files to keep",
"is_public": False,
},
{
"key": "db_log_retention_days",
"value": "30",
"value_type": "integer",
"category": "logging",
"description": "Number of days to retain logs in database",
"is_public": False,
},
{
"key": "file_logging_enabled",
"value": "true",
"value_type": "boolean",
"category": "logging",
"description": "Enable file-based logging",
"is_public": False,
},
{
"key": "db_logging_enabled",
"value": "true",
"value_type": "boolean",
"category": "logging",
"description": "Enable database logging for critical events",
"is_public": False,
},
]
for setting_data in default_settings:
# Check if setting already exists
existing = db.execute(
select(AdminSetting).where(AdminSetting.key == setting_data["key"])
).scalar_one_or_none()
if not existing:
setting = AdminSetting(
key=setting_data["key"],
value=setting_data["value"],
value_type=setting_data["value_type"],
category=setting_data.get("category"),
description=setting_data.get("description"),
is_public=setting_data.get("is_public", False),
created_at=datetime.now(UTC),
updated_at=datetime.now(UTC),
)
db.add(setting)
settings_created += 1
db.flush()
if settings_created > 0:
print_success(f"Created {settings_created} admin settings")
else:
print_warning("Admin settings already exist")
return settings_created
def create_subscription_tiers(db: Session, platform: Platform) -> int:
"""Create default subscription tiers for the OMS platform."""
tier_defs = [
{
"code": "essential",
"name": "Essential",
"price_monthly_cents": 2900,
"price_annual_cents": 29000,
"is_public": True,
"display_order": 10,
},
{
"code": "professional",
"name": "Professional",
"price_monthly_cents": 7900,
"price_annual_cents": 79000,
"is_public": True,
"display_order": 20,
},
{
"code": "business",
"name": "Business",
"price_monthly_cents": 14900,
"price_annual_cents": 149000,
"is_public": True,
"display_order": 30,
},
{
"code": "enterprise",
"name": "Enterprise",
"price_monthly_cents": 29900,
"price_annual_cents": None,
"is_public": False,
"display_order": 40,
},
]
tiers_created = 0
for tdef in tier_defs:
existing = db.execute(
select(SubscriptionTier).where(SubscriptionTier.code == tdef["code"])
).scalar_one_or_none()
if existing:
print_warning(f"Tier already exists: {existing.name} ({existing.code})")
continue
tier = SubscriptionTier(
platform_id=platform.id,
code=tdef["code"],
name=tdef["name"],
price_monthly_cents=tdef["price_monthly_cents"],
price_annual_cents=tdef["price_annual_cents"],
is_public=tdef["is_public"],
display_order=tdef["display_order"],
is_active=True,
)
db.add(tier)
db.flush()
tiers_created += 1
print_success(f"Created tier: {tier.name} ({tier.code})")
return tiers_created
def verify_rbac_schema(db: Session) -> bool:
"""Verify that RBAC schema is in place."""
try:
from sqlalchemy import inspect
inspector = inspect(db.bind)
tables = inspector.get_table_names()
# Check users table has is_email_verified
if "users" in tables:
user_cols = {col["name"] for col in inspector.get_columns("users")}
if "is_email_verified" not in user_cols:
print_error("Missing 'is_email_verified' column in users table")
return False
# Check store_users has RBAC columns
if "store_users" in tables:
vu_cols = {col["name"] for col in inspector.get_columns("store_users")}
required_cols = {
"user_type",
"invitation_token",
"invitation_sent_at",
"invitation_accepted_at",
}
missing = required_cols - vu_cols
if missing:
print_error(f"Missing columns in store_users: {missing}")
return False
# Check roles table exists
if "roles" not in tables:
print_error("Missing 'roles' table")
return False
print_success("RBAC schema verified")
return True
except Exception as e:
print_error(f"Schema verification failed: {e}")
return False
# =============================================================================
# MAIN INITIALIZATION
# =============================================================================
def initialize_production(db: Session, auth_manager: AuthManager):
"""Initialize production database with essential data."""
print_header("PRODUCTION INITIALIZATION")
# Step 1: Verify RBAC schema
print_step(1, "Verifying RBAC schema...")
if not verify_rbac_schema(db):
print_error("RBAC schema not ready. Run migrations first:")
print(" make migrate-up")
sys.exit(1)
# Step 2: Create admin user
print_step(2, "Creating platform admin...")
create_admin_user(db, auth_manager)
# Step 3: Create default platforms
print_step(3, "Creating default platforms...")
platforms = create_default_platforms(db)
# Step 4: Set up default role templates
print_step(4, "Setting up role templates...")
create_default_role_templates(db)
# Step 5: Create admin settings
print_step(5, "Creating admin settings...")
create_admin_settings(db)
# Step 6: Seed subscription tiers
print_step(6, "Seeding subscription tiers...")
oms_platform = next((p for p in platforms if p.code == "oms"), None)
if oms_platform:
create_subscription_tiers(db, oms_platform)
else:
print_warning("OMS platform not found, skipping tier seeding")
# Commit all changes
db.commit()
print_success("All changes committed")
def print_summary(db: Session):
"""Print initialization summary."""
print_header("INITIALIZATION SUMMARY")
# Count records
user_count = db.query(User).filter(User.role == "admin").count()
setting_count = db.query(AdminSetting).count()
platform_count = db.query(Platform).count()
tier_count = db.query(SubscriptionTier).filter(SubscriptionTier.is_active.is_(True)).count()
print("\n📊 Database Status:")
print(f" Admin users: {user_count}")
print(f" Platforms: {platform_count}")
print(f" Admin settings: {setting_count}")
print(f" Sub. tiers: {tier_count}")
print("\n" + "─" * 70)
print("🔐 ADMIN CREDENTIALS")
print("─" * 70)
print(" URL: /admin/login")
print(f" Username: {settings.admin_username}")
print(f" Password: {settings.admin_password}")
print("─" * 70)
# Show security warnings if in production
if is_production():
warnings = validate_production_settings()
if warnings:
print("\n⚠️ SECURITY WARNINGS:")
for warning in warnings:
print(f" {warning}")
print("\n Please update your .env file with secure values!")
else:
print("\n⚠️ DEVELOPMENT MODE:")
print(" Default credentials are OK for development")
print(" Change them in production via .env file")
print("\n🚀 NEXT STEPS:")
print(" 1. Login to admin panel")
if is_production():
print(" 2. CHANGE DEFAULT PASSWORD IMMEDIATELY!")
print(" 3. Configure admin settings")
print(" 4. Create first store")
else:
print(" 2. Create demo data: make seed-demo")
print(" 3. Start development: make dev")
print("\n📝 FOR DEMO DATA (Development only):")
print(" make seed-demo")
# =============================================================================
# MAIN ENTRY POINT
# =============================================================================
def main():
"""Main entry point."""
print("\n" + "╔" + "═" * 68 + "╗")
print("║" + " " * 16 + "PRODUCTION INITIALIZATION" + " " * 27 + "║")
print("╚" + "═" * 68 + "╝")
# Show environment info
print_environment_info()
# Production safety check
if is_production():
warnings = validate_production_settings()
if warnings:
print("\n⚠️ PRODUCTION WARNINGS DETECTED:")
for warning in warnings:
print(f" {warning}")
print("\nUpdate your .env file before continuing!")
response = input("\nContinue anyway? (yes/no): ")
if response.lower() != "yes":
print("Initialization cancelled.")
sys.exit(0)
db = SessionLocal()
auth_manager = AuthManager()
try:
initialize_production(db, auth_manager)
print_summary(db)
print_header("✅ INITIALIZATION COMPLETED")
except KeyboardInterrupt:
db.rollback()
print("\n\n⚠️ Initialization interrupted")
sys.exit(1)
except Exception as e:
db.rollback()
print_header("❌ INITIALIZATION FAILED")
print(f"\nError: {e}\n")
import traceback
traceback.print_exc()
sys.exit(1)
finally:
db.close()
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink