#!/usr/bin/env python3
"""
Demo Database Seeder for Orion Platform
Creates DEMO/TEST data for development and testing:
- Demo stores with realistic data
- Test customers and addresses
- Sample products
- Demo orders
- Store themes and custom domains
- Test import jobs
⚠️ WARNING: This script creates FAKE DATA for development only!
⚠️ NEVER run this in production!
Prerequisites:
- Database migrations must be applied (make migrate-up)
- Production initialization must be run (make init-prod)
Usage:
make seed-demo # Normal demo seeding
make seed-demo-minimal # Minimal seeding (1 store only)
make seed-demo-reset # Delete all data and reseed (DANGEROUS!)
make db-reset # Full reset (migrate down/up + init + seed reset)
Environment Variables:
SEED_MODE=normal|minimal|reset - Seeding mode (default: normal)
FORCE_RESET=true - Skip confirmation in reset mode (for non-interactive use)
This script is idempotent when run normally.
"""
import sys
from datetime import UTC, datetime
from decimal import Decimal
from pathlib import Path
# Add project root to path
project_root = Path(__file__).parent.parent
sys.path.insert(0, str(project_root))
# =============================================================================
# MODE DETECTION (from environment variable set by Makefile)
# =============================================================================
import contextlib
import os
from sqlalchemy import delete, select
from sqlalchemy.orm import Session
from app.core.config import settings
from app.core.database import SessionLocal
from app.core.environment import get_environment, is_production
from app.modules.catalog.models import Product
from app.modules.cms.models import ContentPage, StoreTheme
from app.modules.customers.models.customer import Customer, CustomerAddress
from app.modules.marketplace.models import (
MarketplaceImportJob,
MarketplaceProduct,
MarketplaceProductTranslation,
)
from app.modules.orders.models import Order, OrderItem
# =============================================================================
# MODEL IMPORTS
# =============================================================================
# ALL models must be imported before any ORM query so SQLAlchemy can resolve
# cross-module string relationships (e.g. Store→StoreEmailTemplate,
# Platform→SubscriptionTier, Product→Inventory).
# Core modules
from app.modules.billing.models.merchant_subscription import MerchantSubscription
from app.modules.tenancy.models import (
Merchant,
PlatformAlert,
Role,
Store,
StoreDomain,
StorePlatform,
StoreUser,
User,
)
from middleware.auth import AuthManager
# Optional modules — import to register models with SQLAlchemy
for _mod in [
"app.modules.inventory.models",
"app.modules.cart.models",
"app.modules.billing.models",
"app.modules.messaging.models",
"app.modules.loyalty.models",
]:
with contextlib.suppress(ImportError):
__import__(_mod)
SEED_MODE = os.getenv("SEED_MODE", "normal") # normal, minimal, reset
FORCE_RESET = os.getenv("FORCE_RESET", "false").lower() in ("true", "1", "yes")
# =============================================================================
# DEMO DATA CONFIGURATION
# =============================================================================
# Demo merchant configurations (NEW: Merchant-based architecture)
DEMO_COMPANIES = [
{
"name": "WizaCorp Ltd.",
"description": "Leading technology and electronics distributor",
"owner_email": "john.owner@wizacorp.com",
"owner_password": "password123", # noqa: SEC001
"owner_first_name": "John",
"owner_last_name": "Smith",
"contact_email": "info@wizacorp.com",
"contact_phone": "+352 123 456 789",
"website": "https://www.wizacorp.com",
"business_address": "123 Tech Street, Luxembourg City, L-1234, Luxembourg",
"tax_number": "LU12345678",
},
{
"name": "Fashion Group S.A.",
"description": "International fashion and lifestyle retailer",
"owner_email": "jane.owner@fashiongroup.com",
"owner_password": "password123", # noqa: SEC001
"owner_first_name": "Jane",
"owner_last_name": "Merchant",
"contact_email": "contact@fashiongroup.com",
"contact_phone": "+352 234 567 890",
"website": "https://www.fashiongroup.com",
"business_address": "456 Fashion Avenue, Luxembourg, L-5678, Luxembourg",
"tax_number": "LU23456789",
},
{
"name": "BookWorld Publishing",
"description": "Books, education, and media content provider",
"owner_email": "bob.owner@bookworld.com",
"owner_password": "password123", # noqa: SEC001
"owner_first_name": "Bob",
"owner_last_name": "Seller",
"contact_email": "support@bookworld.com",
"contact_phone": "+352 345 678 901",
"website": "https://www.bookworld.com",
"business_address": "789 Library Lane, Esch-sur-Alzette, L-9012, Luxembourg",
"tax_number": "LU34567890",
},
{
"name": "LuxWeb Agency S.à r.l.",
"description": "Web design and hosting agency serving Luxembourg businesses",
"owner_email": "marc.owner@luxweb.lu",
"owner_password": "password123", # noqa: SEC001
"owner_first_name": "Marc",
"owner_last_name": "Weber",
"contact_email": "info@luxweb.lu",
"contact_phone": "+352 456 789 012",
"website": "https://www.luxweb.lu",
"business_address": "12 Rue du Web, Differdange, L-4501, Luxembourg",
"tax_number": "LU45678901",
},
{
"name": "WizaMart S.à r.l.",
"description": "Online marketplace for everyday essentials and home goods",
"owner_email": "alex.owner@wizamart.com",
"owner_password": "password123", # noqa: SEC001
"owner_first_name": "Alex",
"owner_last_name": "Dupont",
"contact_email": "info@wizamart.com",
"contact_phone": "+352 567 890 123",
"website": "https://www.wizamart.com",
"business_address": "88 Route d'Arlon, Strassen, L-8008, Luxembourg",
"tax_number": "LU56789012",
},
]
# Demo store configurations (linked to merchants by index)
DEMO_STORES = [
{
"merchant_index": 0, # WizaCorp
"store_code": "WIZATECH",
"name": "WizaTech",
"subdomain": "wizatech",
"description": "Premium electronics and gadgets marketplace",
"theme_preset": "modern",
"custom_domain": "wizatech.shop",
"custom_domain_platform": "oms", # Link domain to OMS platform
"platform_subdomains": { # Per-platform subdomain overrides
"loyalty": "wizatech-rewards", # wizatech-rewards.rewardflow.lu
},
},
{
"merchant_index": 0, # WizaCorp
"store_code": "WIZAGADGETS",
"name": "WizaGadgets",
"subdomain": "wizagadgets",
"description": "Smart home devices and IoT accessories",
"theme_preset": "modern",
"custom_domain": None,
},
{
"merchant_index": 0, # WizaCorp
"store_code": "WIZAHOME",
"name": "WizaHome",
"subdomain": "wizahome",
"description": "Home appliances and kitchen electronics",
"theme_preset": "classic",
"custom_domain": None,
},
{
"merchant_index": 1, # Fashion Group
"store_code": "FASHIONHUB",
"name": "Fashion Hub",
"subdomain": "fashionhub",
"description": "Trendy clothing and accessories",
"theme_preset": "vibrant",
"custom_domain": "fashionhub.store",
"custom_domain_platform": "loyalty", # Link domain to Loyalty platform
},
{
"merchant_index": 1, # Fashion Group
"store_code": "FASHIONOUTLET",
"name": "Fashion Outlet",
"subdomain": "fashionoutlet",
"description": "Discounted designer fashion and seasonal clearance",
"theme_preset": "vibrant",
"custom_domain": None,
},
{
"merchant_index": 2, # BookWorld
"store_code": "BOOKSTORE",
"name": "The Book Store",
"subdomain": "bookstore",
"description": "Books, magazines, and educational materials",
"theme_preset": "classic",
"custom_domain": None,
},
{
"merchant_index": 2, # BookWorld
"store_code": "BOOKDIGITAL",
"name": "BookWorld Digital",
"subdomain": "bookdigital",
"description": "E-books, audiobooks, and digital learning resources",
"theme_preset": "modern",
"custom_domain": None,
},
# LuxWeb Agency stores (hosting platform)
{
"merchant_index": 3, # LuxWeb Agency
"store_code": "LUXWEBSITES",
"name": "LuxWeb Sites",
"subdomain": "luxweb",
"description": "Professional websites for Luxembourg businesses",
"theme_preset": "modern",
"custom_domain": "luxweb.lu",
"custom_domain_platform": "hosting",
},
{
"merchant_index": 3, # LuxWeb Agency
"store_code": "LUXWEBHOSTING",
"name": "LuxWeb Hosting",
"subdomain": "luxwebhosting",
"description": "Web hosting, domains, and email services",
"theme_preset": "modern",
"custom_domain": None,
},
# WizaMart store (OMS platform, custom domain)
{
"merchant_index": 4, # WizaMart
"store_code": "WIZAMART",
"name": "WizaMart",
"subdomain": "wizamart",
"description": "Online marketplace for everyday essentials and home goods",
"theme_preset": "modern",
"custom_domain": "wizamart.com",
"custom_domain_platform": "oms",
},
]
# Demo subscriptions (linked to merchants by index)
DEMO_SUBSCRIPTIONS = [
# WizaCorp: OMS (professional, active) + Loyalty (essential, trial)
{"merchant_index": 0, "platform_code": "oms", "tier_code": "professional", "trial_days": 0},
{"merchant_index": 0, "platform_code": "loyalty", "tier_code": "essential", "trial_days": 14},
# Fashion Group: Loyalty only (essential, trial)
{"merchant_index": 1, "platform_code": "loyalty", "tier_code": "essential", "trial_days": 14},
# BookWorld: OMS (business, active)
{"merchant_index": 2, "platform_code": "oms", "tier_code": "business", "trial_days": 0},
# LuxWeb Agency: Hosting (professional, active)
{"merchant_index": 3, "platform_code": "hosting", "tier_code": "professional", "trial_days": 0},
# WizaMart: OMS (essential, trial)
{"merchant_index": 4, "platform_code": "oms", "tier_code": "essential", "trial_days": 30},
]
# Demo team members (linked to merchants by index, assigned to stores by store_code)
# Role must be one of: manager, staff, support, viewer, marketing (see ROLE_PRESETS)
DEMO_TEAM_MEMBERS = [
# WizaCorp team
{
"merchant_index": 0,
"email": "alice.manager@wizacorp.com",
"password": "password123", # noqa: SEC001
"first_name": "Alice",
"last_name": "Manager",
"role": "manager",
"store_codes": ["WIZATECH", "WIZAGADGETS"], # manages two stores
},
{
"merchant_index": 0,
"email": "charlie.staff@wizacorp.com",
"password": "password123", # noqa: SEC001
"first_name": "Charlie",
"last_name": "Staff",
"role": "staff",
"store_codes": ["WIZAHOME"],
},
# Fashion Group team
{
"merchant_index": 1,
"email": "diana.stylist@fashiongroup.com",
"password": "password123", # noqa: SEC001
"first_name": "Diana",
"last_name": "Stylist",
"role": "manager",
"store_codes": ["FASHIONHUB", "FASHIONOUTLET"],
},
{
"merchant_index": 1,
"email": "eric.sales@fashiongroup.com",
"password": "password123", # noqa: SEC001
"first_name": "Eric",
"last_name": "Sales",
"role": "staff",
"store_codes": ["FASHIONOUTLET"],
},
# BookWorld team
{
"merchant_index": 2,
"email": "fiona.editor@bookworld.com",
"password": "password123", # noqa: SEC001
"first_name": "Fiona",
"last_name": "Editor",
"role": "manager",
"store_codes": ["BOOKSTORE", "BOOKDIGITAL"],
},
# LuxWeb Agency team
{
"merchant_index": 3,
"email": "sophie.dev@luxweb.lu",
"password": "password123", # noqa: SEC001
"first_name": "Sophie",
"last_name": "Developer",
"role": "manager",
"store_codes": ["LUXWEBSITES", "LUXWEBHOSTING"],
},
{
"merchant_index": 3,
"email": "tom.support@luxweb.lu",
"password": "password123", # noqa: SEC001
"first_name": "Tom",
"last_name": "Support",
"role": "support",
"store_codes": ["LUXWEBHOSTING"],
},
]
# Theme presets
THEME_PRESETS = {
"modern": {
"primary": "#3b82f6",
"secondary": "#06b6d4",
"accent": "#f59e0b",
"background": "#f9fafb",
"text": "#111827",
},
"classic": {
"primary": "#1e40af",
"secondary": "#7c3aed",
"accent": "#dc2626",
"background": "#ffffff",
"text": "#374151",
},
"vibrant": {
"primary": "#ec4899",
"secondary": "#f59e0b",
"accent": "#8b5cf6",
"background": "#fef3c7",
"text": "#78350f",
},
}
# Store content page overrides (demonstrates CMS store override feature)
# Each store can override platform default pages with custom content
STORE_CONTENT_PAGES = {
"WIZATECH": [
{
"slug": "about",
"title": "About WizaTech",
"content": """
Welcome to WizaTech
Your premier destination for cutting-edge electronics and innovative gadgets.
Our Story
Founded by tech enthusiasts, WizaTech has been bringing the latest technology to customers since 2020.
We carefully curate our selection to ensure you get only the best products at competitive prices.
Why Choose WizaTech?
- Expert Selection: Our team tests and reviews every product
- Best Prices: We negotiate directly with manufacturers
- Fast Shipping: Same-day dispatch on orders before 2pm
- Tech Support: Free lifetime technical assistance
Visit Our Showroom
123 Tech Street, Luxembourg City
Open Monday-Saturday, 9am-7pm
""",
"meta_description": "WizaTech - Your trusted source for premium electronics and gadgets in Luxembourg",
"show_in_header": True,
"show_in_footer": True,
},
{
"slug": "contact",
"title": "Contact WizaTech",
"content": """
Get in Touch with WizaTech
Customer Support
- Email: support@wizatech.lu
- Phone: +352 123 456 789
- WhatsApp: +352 123 456 789
- Hours: Monday-Friday, 9am-6pm CET
Technical Support
Need help with your gadgets? Our tech experts are here to help!
- Email: tech@wizatech.lu
- Live Chat: Available on our website
Store Location
123 Tech Street
Luxembourg City, L-1234
Luxembourg
""",
"meta_description": "Contact WizaTech customer support for electronics and gadget inquiries",
"show_in_header": True,
"show_in_footer": True,
},
],
"FASHIONHUB": [
{
"slug": "about",
"title": "About Fashion Hub",
"content": """
Welcome to Fashion Hub
Where style meets affordability. Discover the latest trends in fashion and accessories.
Our Philosophy
At Fashion Hub, we believe everyone deserves to look and feel their best.
We curate collections from emerging designers and established brands to bring you
fashion-forward pieces at accessible prices.
What Makes Us Different
- Trend-Forward: New arrivals weekly from global fashion capitals
- Sustainable: 40% of our collection uses eco-friendly materials
- Inclusive: Sizes XS to 4XL available
- Personal Styling: Free virtual styling consultations
Join Our Community
Follow us on Instagram @FashionHubLux for styling tips and exclusive offers!
""",
"meta_description": "Fashion Hub - Trendy clothing and accessories for the style-conscious shopper",
"show_in_header": True,
"show_in_footer": True,
},
],
"LUXWEBSITES": [
{
"slug": "about",
"title": "About LuxWeb Sites",
"content": """
Welcome to LuxWeb Sites
Professional web design and development for Luxembourg businesses.
We create modern, responsive websites that help your business grow online.
Our Services
- Website Design: Custom designs tailored to your brand
- E-commerce: Online shops with secure payment integration
- Multilingual Sites: FR, DE, EN, and LB support built-in
- SEO Optimization: Get found on Google Luxembourg
- Maintenance: Ongoing updates and technical support
Why LuxWeb?
Based in Luxembourg, we understand the local market. All our sites are GDPR compliant
and optimized for the multilingual Luxembourg audience.
Contact Us
12 Rue du Web, Differdange
Email: info@luxweb.lu | Phone: +352 456 789 012
""",
"meta_description": "LuxWeb Sites - Professional web design for Luxembourg businesses",
"show_in_header": True,
"show_in_footer": True,
},
{
"slug": "contact",
"title": "Contact LuxWeb",
"content": """
Get in Touch
Request a Quote
Tell us about your project and we'll get back to you within 24 hours.
- Email: info@luxweb.lu
- Phone: +352 456 789 012
- WhatsApp: +352 456 789 012
Technical Support
Existing clients can reach our support team:
- Email: support@luxweb.lu
- Hours: Monday-Friday, 8am-6pm CET
Office
12 Rue du Web
Differdange, L-4501
Luxembourg
""",
"meta_description": "Contact LuxWeb for web design, hosting, and domain services in Luxembourg",
"show_in_header": True,
"show_in_footer": True,
},
],
"BOOKSTORE": [
{
"slug": "about",
"title": "About The Book Store",
"content": """
Welcome to The Book Store
Your literary haven in Luxembourg. From bestsellers to rare finds, we have something for every reader.
Our Heritage
Established by book lovers for book lovers, The Book Store has been serving
the Luxembourg community for over a decade. We pride ourselves on our carefully
curated selection and knowledgeable staff.
What We Offer
- Vast Selection: Over 50,000 titles across all genres
- Special Orders: Can't find what you're looking for? We'll get it for you
- Book Club: Monthly meetings and 15% member discount
- Author Events: Regular readings and book signings
- Children's Corner: Dedicated space for young readers
Visit Us
789 Library Lane, Esch-sur-Alzette
Open daily 10am-8pm, Sundays 12pm-6pm
""",
"meta_description": "The Book Store - Your independent bookshop in Luxembourg with a passion for literature",
"show_in_header": True,
"show_in_footer": True,
},
{
"slug": "faq",
"title": "Book Store FAQ",
"content": """
Frequently Asked Questions
Orders & Delivery
Do you ship internationally?
Yes! We ship to all EU countries. Non-EU shipping available on request.
How long does delivery take?
Luxembourg: 1-2 business days. EU: 3-7 business days.
Can I order books that aren't in stock?
Absolutely! We can order any book in print. Special orders usually arrive within 1-2 weeks.
Book Club
How do I join the book club?
Sign up at our store or email bookclub@bookstore.lu. Annual membership is €25.
What are the benefits?
15% discount on all purchases, early access to author events, and monthly reading recommendations.
Gift Cards
Do you sell gift cards?
Yes! Available in €10, €25, €50, and €100 denominations, or custom amounts.
""",
"meta_description": "Frequently asked questions about The Book Store - orders, delivery, and book club",
"show_in_header": False,
"show_in_footer": True,
},
],
}
# =============================================================================
# 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}")
# =============================================================================
# SAFETY CHECKS
# =============================================================================
def check_environment():
"""Prevent running demo seed in production."""
if is_production():
print_error("Cannot run demo seeding in production!")
print(" This script creates FAKE DATA for development only.")
print(f" Current environment: {get_environment()}")
print("\n To seed in production:")
print(" 1. Set ENVIRONMENT=development in .env (or ENV=development)")
print(" 2. Run: make seed-demo")
sys.exit(1)
print_success(f"Environment check passed: {get_environment()}")
def check_admin_exists(db: Session) -> bool:
"""Check if admin user exists."""
admin = db.execute(
select(User).where(User.role.in_(["super_admin", "platform_admin"])).limit(1)
).scalar_one_or_none()
if not admin:
print_error("No admin user found!")
print(" Run production initialization first:")
print(" make init-prod")
return False
print_success(f"Admin user exists: {admin.email}")
return True
# =============================================================================
# DATA DELETION (for reset mode)
# =============================================================================
def reset_all_data(db: Session):
"""Delete ALL data from database (except admin user)."""
print_warning("RESETTING ALL DATA...")
print(" This will delete all stores, customers, orders, etc.")
print(" Admin user will be preserved.")
# Skip confirmation if FORCE_RESET is set (for non-interactive use)
if FORCE_RESET:
print_warning("FORCE_RESET enabled - skipping confirmation")
else:
# Get confirmation interactively
try:
response = input("\n Type 'DELETE ALL DATA' to confirm: ")
if response != "DELETE ALL DATA":
print(" Reset cancelled.")
sys.exit(0)
except EOFError:
print_error("No interactive terminal available.")
print(
" Use FORCE_RESET=true to skip confirmation in non-interactive mode."
)
sys.exit(1)
# Delete in correct order (respecting foreign keys)
tables_to_clear = [
OrderItem,
Order,
CustomerAddress,
Customer,
MarketplaceImportJob,
Product, # Product references MarketplaceProduct
MarketplaceProductTranslation, # Translation references MarketplaceProduct
MarketplaceProduct,
ContentPage, # Delete store content pages (keep platform defaults)
StoreDomain,
StoreTheme,
Role,
StoreUser,
Store,
Merchant, # Delete merchants (cascades to stores)
PlatformAlert,
]
for table in tables_to_clear:
if table == ContentPage:
# Only delete store content pages, keep platform defaults
db.execute(delete(ContentPage).where(ContentPage.store_id.is_not(None)))
else:
db.execute(delete(table))
# Delete non-admin users
db.execute(delete(User).where(User.role.not_in(["super_admin", "platform_admin"])))
db.commit()
print_success("All data deleted (admin preserved)")
# =============================================================================
# SEEDING FUNCTIONS
# =============================================================================
def create_demo_merchants(db: Session, auth_manager: AuthManager) -> list[Merchant]:
"""Create demo merchants with owner accounts."""
merchants = []
# Determine how many merchants to create based on mode
merchant_count = 1 if SEED_MODE == "minimal" else len(DEMO_COMPANIES)
merchants_to_create = DEMO_COMPANIES[:merchant_count]
for merchant_data in merchants_to_create:
# Check if merchant already exists
existing = db.execute(
select(Merchant).where(Merchant.name == merchant_data["name"])
).scalar_one_or_none()
if existing:
print_warning(f"Merchant already exists: {merchant_data['name']}")
merchants.append(existing)
continue
# Check if owner user already exists
owner_user = db.execute(
select(User).where(User.email == merchant_data["owner_email"])
).scalar_one_or_none()
if not owner_user:
# Create owner user
owner_user = User(
username=merchant_data["owner_email"].split("@")[0],
email=merchant_data["owner_email"],
hashed_password=auth_manager.hash_password( # noqa: SEC001
merchant_data["owner_password"]
),
role="merchant_owner",
first_name=merchant_data["owner_first_name"],
last_name=merchant_data["owner_last_name"],
is_active=True,
is_email_verified=True,
created_at=datetime.now(UTC),
updated_at=datetime.now(UTC),
)
db.add(owner_user) # noqa: PERF006
db.flush()
print_success(
f"Created owner user: {owner_user.email} (password: {merchant_data['owner_password']})"
)
else:
print_warning(f"Using existing user as owner: {owner_user.email}")
# Create merchant
merchant = Merchant(
name=merchant_data["name"],
description=merchant_data["description"],
owner_user_id=owner_user.id,
contact_email=merchant_data["contact_email"],
contact_phone=merchant_data.get("contact_phone"),
website=merchant_data.get("website"),
business_address=merchant_data.get("business_address"),
tax_number=merchant_data.get("tax_number"),
is_active=True,
is_verified=True, # Auto-verified for demo
created_at=datetime.now(UTC),
updated_at=datetime.now(UTC),
)
db.add(merchant) # noqa: PERF006
db.flush()
merchants.append(merchant)
print_success(f"Created merchant: {merchant.name} (Owner: {owner_user.email})")
db.flush()
return merchants
def create_demo_subscriptions(db: Session, merchants: list[Merchant]) -> None:
"""Create demo merchant subscriptions."""
from app.modules.billing.models.subscription import SubscriptionTier
from app.modules.tenancy.models import Platform
sub_count = 1 if SEED_MODE == "minimal" else len(DEMO_SUBSCRIPTIONS)
subs_to_create = DEMO_SUBSCRIPTIONS[:sub_count]
for sub_data in subs_to_create:
merchant_index = sub_data["merchant_index"]
if merchant_index >= len(merchants):
print_error(f"Invalid merchant_index {merchant_index} for subscription")
continue
merchant = merchants[merchant_index]
# Look up platform by code
platform = db.execute(
select(Platform).where(Platform.code == sub_data["platform_code"])
).scalar_one_or_none()
if not platform:
print_warning(f"Platform '{sub_data['platform_code']}' not found, skipping")
continue
# Check if subscription already exists
existing = db.execute(
select(MerchantSubscription).where(
MerchantSubscription.merchant_id == merchant.id,
MerchantSubscription.platform_id == platform.id,
)
).scalar_one_or_none()
if existing:
print_warning(
f"Subscription already exists: {merchant.name} on {platform.name}"
)
continue
# Look up tier by code + platform
tier = db.execute(
select(SubscriptionTier).where(
SubscriptionTier.code == sub_data["tier_code"],
SubscriptionTier.platform_id == platform.id,
)
).scalar_one_or_none()
if not tier:
print_warning(
f"Tier '{sub_data['tier_code']}' not found for {platform.name}, skipping"
)
continue
from app.modules.billing.services.subscription_service import subscription_service
subscription = subscription_service.create_merchant_subscription(
db,
merchant_id=merchant.id,
platform_id=platform.id,
tier_code=sub_data["tier_code"],
trial_days=sub_data.get("trial_days", 14),
)
print_success(
f"Created subscription: {merchant.name} → {platform.name} "
f"({tier.name}, {subscription.status})"
)
db.flush()
def create_demo_stores(
db: Session, merchants: list[Merchant], auth_manager: AuthManager
) -> list[Store]:
"""Create demo stores linked to merchants."""
stores = []
# Determine how many stores to create based on mode
store_count = 1 if SEED_MODE == "minimal" else len(DEMO_STORES)
stores_to_create = DEMO_STORES[:store_count]
for store_data in stores_to_create:
# Check if store already exists
existing = db.execute(
select(Store).where(Store.store_code == store_data["store_code"])
).scalar_one_or_none()
if existing:
print_warning(f"Store already exists: {store_data['name']}")
stores.append(existing)
continue
# Get merchant by index
merchant_index = store_data["merchant_index"]
if merchant_index >= len(merchants):
print_error(
f"Invalid merchant_index {merchant_index} for store {store_data['name']}"
)
continue
merchant = merchants[merchant_index]
# Create store linked to merchant (owner is inherited from merchant)
store = Store(
merchant_id=merchant.id, # Link to merchant
store_code=store_data["store_code"],
name=store_data["name"],
subdomain=store_data["subdomain"],
description=store_data["description"],
is_active=True,
is_verified=True,
created_at=datetime.now(UTC),
updated_at=datetime.now(UTC),
)
db.add(store) # noqa: PERF006
db.flush()
# Link store to merchant's subscribed platforms
merchant_subs = db.execute(
select(MerchantSubscription.platform_id).where(
MerchantSubscription.merchant_id == merchant.id
)
).all()
# Build platform code→id lookup for this store's custom subdomain config
from app.modules.tenancy.models import Platform
platform_code_map = {}
if store_data.get("platform_subdomains") or store_data.get("custom_domain_platform"):
platform_rows = db.execute(select(Platform.id, Platform.code)).all()
platform_code_map = {code: pid for pid, code in platform_rows}
store_platforms = []
for i, (platform_id,) in enumerate(merchant_subs):
# Per-platform subdomain override for multi-platform stores
# Config uses platform codes; resolve to IDs
custom_sub = None
for pcode, subdomain_val in store_data.get("platform_subdomains", {}).items():
if platform_code_map.get(pcode) == platform_id:
custom_sub = subdomain_val
break
store_platforms.append(StorePlatform(
store_id=store.id,
platform_id=platform_id,
is_active=True,
custom_subdomain=custom_sub,
))
if store_platforms:
db.add_all(store_platforms)
if merchant_subs:
db.flush()
print_success(
f" Linked to {len(merchant_subs)} platform(s): "
f"{[pid for (pid,) in merchant_subs]}"
)
# Report custom subdomains if any
for pcode, subdomain_val in store_data.get("platform_subdomains", {}).items():
print_success(f" Custom subdomain on {pcode}: {subdomain_val}")
# Owner relationship is via Merchant.owner_user_id — no StoreUser needed
# Create store theme
theme_colors = THEME_PRESETS.get(
store_data["theme_preset"], THEME_PRESETS["modern"]
)
theme = StoreTheme(
store_id=store.id,
theme_name=store_data["theme_preset"],
colors={ # ✅ Use JSON format
"primary": theme_colors["primary"],
"secondary": theme_colors["secondary"],
"accent": theme_colors["accent"],
"background": theme_colors["background"],
"text": theme_colors["text"],
},
is_active=True,
created_at=datetime.now(UTC),
updated_at=datetime.now(UTC),
)
db.add(theme) # noqa: PERF006
# Create custom domain if specified
if store_data.get("custom_domain"):
# Resolve platform_id from platform code (if specified)
domain_platform_code = store_data.get("custom_domain_platform")
domain_platform_id = platform_code_map.get(domain_platform_code) if domain_platform_code else None
domain = StoreDomain(
store_id=store.id,
domain=store_data[
"custom_domain"
], # ✅ Field is 'domain', not 'domain_name'
platform_id=domain_platform_id,
is_verified=True, # Auto-verified for demo
is_primary=True,
verification_token=None,
created_at=datetime.now(UTC),
updated_at=datetime.now(UTC),
)
db.add(domain) # noqa: PERF006
stores.append(store)
print_success(f"Created store: {store.name} ({store.store_code})")
db.flush()
return stores
def _ensure_store_roles(db: Session, store: Store) -> dict[str, Role]:
"""Ensure default roles exist for a store, return name→Role lookup."""
from app.modules.tenancy.services.permission_discovery_service import (
permission_discovery_service,
)
existing = db.query(Role).filter(Role.store_id == store.id).all()
if existing:
return {r.name: r for r in existing}
role_names = ["manager", "staff", "support", "viewer", "marketing"]
roles = {}
for name in role_names:
permissions = list(permission_discovery_service.get_preset_permissions(name))
role = Role(
store_id=store.id,
name=name,
permissions=permissions,
)
db.add(role) # noqa: PERF006
roles[name] = role
db.flush()
print_success(f" Created default roles for {store.name}: {', '.join(role_names)}")
return roles
def create_demo_team_members(
db: Session, stores: list[Store], auth_manager: AuthManager
) -> list[User]:
"""Create demo team member users and assign them to stores with roles."""
if SEED_MODE == "minimal":
return []
team_users = []
# Build a store_code → Store lookup from the created stores
store_lookup = {s.store_code: s for s in stores}
# Pre-create default roles for all stores that team members will be assigned to
store_roles: dict[str, dict[str, Role]] = {}
for member_data in DEMO_TEAM_MEMBERS:
for store_code in member_data["store_codes"]:
if store_code not in store_roles:
store = store_lookup.get(store_code)
if store:
store_roles[store_code] = _ensure_store_roles(db, store)
for member_data in DEMO_TEAM_MEMBERS:
# Check if user already exists
user = db.execute(
select(User).where(User.email == member_data["email"])
).scalar_one_or_none()
if not user:
user = User(
username=member_data["email"].split("@")[0],
email=member_data["email"],
hashed_password=auth_manager.hash_password(member_data["password"]), # noqa: SEC001
role="store_member",
first_name=member_data["first_name"],
last_name=member_data["last_name"],
is_active=True,
is_email_verified=True,
created_at=datetime.now(UTC),
updated_at=datetime.now(UTC),
)
db.add(user) # noqa: PERF006
db.flush()
print_success(
f"Created team member: {user.email} (password: {member_data['password']})"
)
else:
print_warning(f"Team member already exists: {user.email}")
team_users.append(user)
# Assign user to stores with role
role_name = member_data["role"]
for store_code in member_data["store_codes"]:
store = store_lookup.get(store_code)
if not store:
print_warning(f"Store {store_code} not found, skipping assignment")
continue
# Check if StoreUser link already exists
existing_link = db.execute(
select(StoreUser).where(
StoreUser.store_id == store.id,
StoreUser.user_id == user.id,
)
).scalar_one_or_none()
if existing_link:
continue
# Look up role for this store
role = store_roles.get(store_code, {}).get(role_name)
store_user = StoreUser(
store_id=store.id,
user_id=user.id,
role_id=role.id if role else None,
is_active=True,
created_at=datetime.now(UTC),
)
db.add(store_user) # noqa: PERF006
print_success(
f" Assigned {user.first_name} to {store.name} as {role_name}"
)
db.flush()
return team_users
def create_demo_customers(
db: Session, store: Store, auth_manager: AuthManager, count: int
) -> list[Customer]:
"""Create demo customers for a store."""
customers = []
new_count = 0
# Use a simple demo password for all customers
demo_password = "customer123" # noqa: SEC001
for i in range(1, count + 1):
email = f"customer{i}@{store.subdomain}.example.com"
customer_number = f"CUST-{store.store_code}-{i:04d}"
# Check if customer already exists
existing_customer = (
db.query(Customer)
.filter(Customer.store_id == store.id, Customer.email == email)
.first()
)
if existing_customer:
customers.append(existing_customer)
continue # Skip creation, customer already exists
customer = Customer(
store_id=store.id,
email=email,
hashed_password=auth_manager.hash_password(demo_password), # noqa: SEC001
first_name=f"Customer{i}",
last_name="Test",
phone=f"+352123456{i:03d}",
customer_number=customer_number,
is_active=True,
created_at=datetime.now(UTC),
updated_at=datetime.now(UTC),
)
db.add(customer) # noqa: PERF006
customers.append(customer)
new_count += 1
db.flush()
if new_count > 0:
print_success(f"Created {new_count} customers for {store.name}")
else:
print_warning(f"Customers already exist for {store.name}")
return customers
def create_demo_products(db: Session, store: Store, count: int) -> list[Product]:
"""Create demo products for a store."""
products = []
new_count = 0
for i in range(1, count + 1):
marketplace_product_id = f"{store.store_code}-MP-{i:04d}"
product_id = f"{store.store_code}-PROD-{i:03d}"
# Check if this product already exists (by store_sku)
existing_product = (
db.query(Product)
.filter(Product.store_id == store.id, Product.store_sku == product_id)
.first()
)
if existing_product:
products.append(existing_product)
continue # Skip creation, product already exists
# Check if marketplace product already exists
existing_mp = (
db.query(MarketplaceProduct)
.filter(MarketplaceProduct.marketplace_product_id == marketplace_product_id)
.first()
)
if existing_mp:
marketplace_product = existing_mp
else:
# Create the MarketplaceProduct (base product data)
marketplace_product = MarketplaceProduct(
marketplace_product_id=marketplace_product_id,
source_url=f"https://{store.subdomain}.example.com/products/sample-{i}",
image_link=f"https://{store.subdomain}.example.com/images/product-{i}.jpg",
price=str(Decimal(f"{(i * 10) % 500 + 9.99}")), # Store as string
brand=store.name,
gtin=f"TEST{store.id:02d}{i:010d}",
availability="in stock",
condition="new",
google_product_category="Electronics > Computers > Laptops",
marketplace="OMS",
store_name=store.name,
currency="EUR",
created_at=datetime.now(UTC),
updated_at=datetime.now(UTC),
)
db.add(marketplace_product) # noqa: PERF006
db.flush() # Flush to get the marketplace_product.id
# Check if English translation already exists
existing_translation = (
db.query(MarketplaceProductTranslation)
.filter(
MarketplaceProductTranslation.marketplace_product_id
== marketplace_product.id,
MarketplaceProductTranslation.language == "en",
)
.first()
)
if not existing_translation:
# Create English translation
translation = MarketplaceProductTranslation(
marketplace_product_id=marketplace_product.id,
language="en",
title=f"Sample Product {i} - {store.name}",
description=f"This is a demo product for testing purposes in {store.name}. High quality and affordable.",
)
db.add(translation) # noqa: PERF006
# Create the Product (store-specific entry)
product = Product(
store_id=store.id,
marketplace_product_id=marketplace_product.id,
store_sku=product_id, # Use store_sku for store's internal product reference
price=float(Decimal(f"{(i * 10) % 500 + 9.99}")), # Store as float
is_active=True,
created_at=datetime.now(UTC),
updated_at=datetime.now(UTC),
)
db.add(product) # noqa: PERF006
products.append(product)
new_count += 1
db.flush()
if new_count > 0:
print_success(f"Created {new_count} products for {store.name}")
else:
print_warning(f"Products already exist for {store.name}")
return products
def create_demo_store_content_pages(db: Session, stores: list[Store]) -> int:
"""Create store-specific content page overrides.
These demonstrate the CMS store override feature where stores can
customize platform default pages with their own branding and content.
"""
created_count = 0
# Build store→primary platform lookup
from app.modules.tenancy.models import Platform
store_primary_platform: dict[int, int] = {}
sp_rows = db.execute(
select(StorePlatform.store_id, StorePlatform.platform_id)
.where(StorePlatform.is_active == True) # noqa: E712
.order_by(StorePlatform.joined_at)
).all()
for store_id, platform_id in sp_rows:
if store_id not in store_primary_platform:
store_primary_platform[store_id] = platform_id
# Fallback: OMS platform ID
oms_platform = db.execute(
select(Platform).where(Platform.code == "oms")
).scalar_one_or_none()
fallback_platform_id = oms_platform.id if oms_platform else 1
for store in stores:
store_pages = STORE_CONTENT_PAGES.get(store.store_code, [])
if not store_pages:
continue
platform_id = store_primary_platform.get(store.id, fallback_platform_id)
for page_data in store_pages:
# Check if this store page already exists
existing = db.execute(
select(ContentPage).where(
ContentPage.store_id == store.id,
ContentPage.slug == page_data["slug"],
)
).scalar_one_or_none()
if existing:
continue # Skip, already exists
# Create store content page override
page = ContentPage(
platform_id=platform_id,
store_id=store.id,
slug=page_data["slug"],
title=page_data["title"],
content=page_data["content"].strip(),
content_format="html",
meta_description=page_data.get("meta_description"),
is_published=True,
published_at=datetime.now(UTC),
show_in_header=page_data.get("show_in_header", False),
show_in_footer=page_data.get("show_in_footer", True),
show_in_legal=page_data.get("show_in_legal", False),
display_order=page_data.get("display_order", 0),
created_at=datetime.now(UTC),
updated_at=datetime.now(UTC),
)
db.add(page) # noqa: PERF006
created_count += 1
db.flush()
if created_count > 0:
print_success(f"Created {created_count} store content page overrides")
else:
print_warning("Store content pages already exist")
return created_count
# =============================================================================
# MAIN SEEDING
# =============================================================================
def seed_demo_data(db: Session, auth_manager: AuthManager):
"""Seed demo data for development."""
print_header("DEMO DATA SEEDING")
print(f" Mode: {SEED_MODE.upper()}")
# Step 1: Check environment
print_step(1, "Checking environment...")
check_environment()
# Step 2: Check admin exists
print_step(2, "Verifying admin user...")
if not check_admin_exists(db):
sys.exit(1)
# Step 3: Reset data if in reset mode
if SEED_MODE == "reset":
print_step(3, "Resetting data...")
reset_all_data(db)
# Step 4: Create merchants
print_step(4, "Creating demo merchants...")
merchants = create_demo_merchants(db, auth_manager)
# Step 5: Create merchant subscriptions (before stores, so StorePlatform linking works)
print_step(5, "Creating demo subscriptions...")
create_demo_subscriptions(db, merchants)
# Step 6: Create stores
print_step(6, "Creating demo stores...")
stores = create_demo_stores(db, merchants, auth_manager)
# Step 7: Create team members
print_step(7, "Creating demo team members...")
create_demo_team_members(db, stores, auth_manager)
# Step 8: Create customers
print_step(8, "Creating demo customers...")
for store in stores:
create_demo_customers(
db, store, auth_manager, count=settings.seed_customers_per_store
)
# Step 9: Create products
print_step(9, "Creating demo products...")
for store in stores:
create_demo_products(db, store, count=settings.seed_products_per_store)
# Step 10: Create store content pages
print_step(10, "Creating store content page overrides...")
create_demo_store_content_pages(db, stores)
# Commit all changes
db.commit()
print_success("All demo data committed")
def print_summary(db: Session):
"""Print seeding summary."""
print_header("SEEDING SUMMARY")
# Count records
merchant_count = db.query(Merchant).count()
store_count = db.query(Store).count()
user_count = db.query(User).count()
team_member_count = db.query(StoreUser).count()
customer_count = db.query(Customer).count()
product_count = db.query(Product).count()
platform_pages = db.query(ContentPage).filter(ContentPage.store_id.is_(None)).count()
store_pages = db.query(ContentPage).filter(ContentPage.store_id.is_not(None)).count()
print("\n📊 Database Status:")
print(f" Merchants: {merchant_count}")
print(f" Stores: {store_count}")
print(f" Users: {user_count}")
print(f" Team memberships: {team_member_count}")
print(f" Customers: {customer_count}")
print(f" Products: {product_count}")
print(f" Content Pages: {platform_pages} platform + {store_pages} store overrides")
# Show merchant details
merchants = db.query(Merchant).all()
print("\n🏢 Demo Merchants:")
for merchant in merchants:
print(f"\n {merchant.name}")
print(f" Owner: {merchant.owner.email if merchant.owner else 'N/A'}")
print(f" Stores: {len(merchant.stores) if merchant.stores else 0}")
print(f" Status: {'✓ Active' if merchant.is_active else '✗ Inactive'}")
if merchant.is_verified:
print(" Verified: ✓")
# Show store details with team members
stores = db.query(Store).all()
print("\n🏪 Demo Stores:")
for store in stores:
print(f"\n {store.name} ({store.store_code})")
print(f" Subdomain: {store.subdomain}.{settings.main_domain}")
# Show per-platform custom subdomains
from app.modules.tenancy.models import Platform
store_platforms = (
db.query(StorePlatform, Platform)
.join(Platform, Platform.id == StorePlatform.platform_id)
.filter(StorePlatform.store_id == store.id)
.all()
)
for sp, platform in store_platforms:
if sp.custom_subdomain:
pdomain = getattr(platform, "domain", platform.code)
print(f" [{platform.code}] Custom subdomain: {sp.custom_subdomain}.{pdomain}")
# Query custom domains separately
custom_domain = (
db.query(StoreDomain)
.filter(StoreDomain.store_id == store.id, StoreDomain.is_active == True)
.first()
)
if custom_domain:
# Try different possible field names (model field might vary)
domain_value = (
getattr(custom_domain, "domain", None)
or getattr(custom_domain, "domain_name", None)
or getattr(custom_domain, "name", None)
)
if domain_value:
platform_note = ""
if custom_domain.platform_id:
dp = db.query(Platform).filter(Platform.id == custom_domain.platform_id).first()
if dp:
platform_note = f" (linked to {dp.code})"
print(f" Custom: {domain_value}{platform_note}")
print(f" Status: {'✓ Active' if store.is_active else '✗ Inactive'}")
# Show team members for this store
store_users = (
db.query(StoreUser)
.filter(StoreUser.store_id == store.id)
.all()
)
if store_users:
for su in store_users:
user = db.query(User).filter(User.id == su.user_id).first()
if user:
role_name = su.role.name if su.role else "owner"
print(f" Team: {user.email} ({role_name})")
port = settings.api_port
base = f"http://localhost:{port}"
print("\n🔐 Demo Merchant Owner Credentials:")
print("─" * 70)
for i, merchant_data in enumerate(DEMO_COMPANIES[:merchant_count], 1):
merchant = merchants[i - 1] if i <= len(merchants) else None
print(f" Merchant {i}: {merchant_data['name']}")
print(f" Email: {merchant_data['owner_email']}")
print(f" Password: {merchant_data['owner_password']}") # noqa: SEC021
if merchant and merchant.stores:
for store in merchant.stores:
print(
f" Store panel: {base}/store/{store.store_code}/login"
)
print(
f" or http://{store.subdomain}.localhost:{port}/store/login" # noqa: SEC034
)
print()
print("\n👥 Demo Team Member Credentials:")
print("─" * 70)
for member_data in DEMO_TEAM_MEMBERS:
merchant_name = DEMO_COMPANIES[member_data["merchant_index"]]["name"]
store_codes = ", ".join(member_data["store_codes"])
print(f" {member_data['first_name']} {member_data['last_name']} ({merchant_name})")
print(f" Email: {member_data['email']}")
print(f" Password: {member_data['password']}") # noqa: SEC021
print(f" Role: {member_data['role']}")
print(f" Stores: {store_codes}")
print()
print("\n🛒 Demo Customer Credentials:")
print("─" * 70)
print(" All customers:")
print(" Email: customer1@{subdomain}.example.com")
print(" Password: customer123") # noqa: SEC021
print(" (Replace {subdomain} with store subdomain, e.g., wizatech)")
print()
# Build store → platform code mapping from store_platforms
from app.modules.tenancy.models import Platform
store_platform_rows = db.execute(
select(StorePlatform.store_id, Platform.code).join(
Platform, Platform.id == StorePlatform.platform_id
).where(StorePlatform.is_active == True).order_by( # noqa: E712
StorePlatform.store_id, StorePlatform.joined_at
)
).all()
store_platform_map: dict[int, list[str]] = {}
for store_id, platform_code in store_platform_rows:
store_platform_map.setdefault(store_id, []).append(platform_code)
# Build store→platform details (including custom subdomains) for production URLs
from app.modules.tenancy.models import Platform
store_platform_details: dict[int, list[dict]] = {}
sp_detail_rows = db.execute(
select(
StorePlatform.store_id,
Platform.code,
Platform.domain,
StorePlatform.custom_subdomain,
).join(Platform, Platform.id == StorePlatform.platform_id)
.where(StorePlatform.is_active == True) # noqa: E712
.order_by(StorePlatform.store_id, StorePlatform.joined_at)
).all()
for store_id, pcode, pdomain, custom_sub in sp_detail_rows:
store_platform_details.setdefault(store_id, []).append({
"code": pcode, "domain": pdomain, "custom_subdomain": custom_sub,
})
print("\n🏪 Store Access (Development):")
print("─" * 70)
for store in stores:
platform_codes = store_platform_map.get(store.id, [])
print(f" {store.name} ({store.store_code}):")
if platform_codes:
for pc in platform_codes:
print(f" [{pc}] Storefront: {base}/platforms/{pc}/storefront/{store.store_code}/")
print(f" [{pc}] Customer login: {base}/platforms/{pc}/storefront/{store.store_code}/account/login")
print(f" [{pc}] Store panel: {base}/platforms/{pc}/store/{store.store_code}/")
print(f" [{pc}] Store login: {base}/platforms/{pc}/store/{store.store_code}/login")
else:
print(" (!) No platform assigned")
print("\n🌐 Store Access (Production-style):")
print("─" * 70)
for store in stores:
details = store_platform_details.get(store.id, [])
print(f" {store.name} ({store.store_code}):")
for d in details:
subdomain = d["custom_subdomain"] or store.subdomain
pdomain = d["domain"] or f"{d['code']}.example.com"
suffix = " (custom subdomain)" if d["custom_subdomain"] else ""
print(f" [{d['code']}] Storefront: https://{subdomain}.{pdomain}/{suffix}")
print(f" [{d['code']}] Customer login: https://{subdomain}.{pdomain}/account/login")
print(f" [{d['code']}] Store panel: https://{subdomain}.{pdomain}/store/")
# Show custom domain if any
custom_domain = (
db.query(StoreDomain)
.filter(StoreDomain.store_id == store.id, StoreDomain.is_active == True)
.first()
)
if custom_domain:
dv = getattr(custom_domain, "domain", None)
if dv:
print(f" [custom] Storefront: https://{dv}/")
print()
print("⚠️ ALL DEMO CREDENTIALS ARE INSECURE - For development only!")
print("\n🚀 NEXT STEPS:")
print(" 1. Start development: make dev")
print(f" 2. Admin panel: {base}/admin/login")
print(f" 3. Merchant panel: {base}/merchants/login")
print(f" 4. Store login: {base}/platforms/oms/store/WIZATECH/login")
print(f" 5. Storefront: {base}/platforms/oms/storefront/WIZATECH/")
print(f" 6. Customer login: {base}/platforms/oms/storefront/WIZATECH/account/login")
# =============================================================================
# MAIN ENTRY POINT
# =============================================================================
def main():
"""Main entry point."""
print("\n" + "╔" + "═" * 68 + "╗")
print("║" + " " * 20 + "DEMO DATA SEEDING" + " " * 31 + "║")
print("╚" + "═" * 68 + "╝")
db = SessionLocal()
auth_manager = AuthManager()
try:
seed_demo_data(db, auth_manager)
print_summary(db)
print_header("✅ DEMO SEEDING COMPLETED")
except KeyboardInterrupt:
db.rollback()
print("\n\n⚠️ Seeding interrupted")
sys.exit(1)
except Exception as e:
db.rollback()
print_header("❌ SEEDING FAILED")
print(f"\nError: {e}\n")
import traceback
traceback.print_exc()
sys.exit(1)
finally:
db.close()
if __name__ == "__main__":
main()