#!/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", }, ] # 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, }, ] # 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}, ] # 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"], }, ] # 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?

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

Technical Support

Need help with your gadgets? Our tech experts are here to help!

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

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, }, ], "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

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} 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 sp = StorePlatform( store_id=store.id, platform_id=platform_id, is_active=True, is_primary=(i == 0), custom_subdomain=custom_sub, ) db.add(sp) 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 # Get the OMS platform ID (stores are registered on OMS) from app.modules.tenancy.models import Platform oms_platform = db.execute( select(Platform).where(Platform.code == "oms") ).scalar_one_or_none() default_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 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=default_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.platform_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.is_primary.desc() ) ).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.is_primary.desc()) ).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()