sboulahtit/orion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
33c5875bc8589f9dbfcb54a485a511e43854c8ec
orion/scripts/seed_demo.py
Samir Boulahtit 65f296e883 fix: make db-reset work in non-interactive mode 
- Add FORCE_RESET environment variable to skip confirmation prompt
- Update Makefile db-reset target to use FORCE_RESET=true
- Handle EOFError gracefully with helpful message
- Fix duplicate translation creation in seed script
- Check for existing translations before inserting

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-12 22:36:49 +01:00

815 lines
27 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""
Demo Database Seeder for Wizamart Platform
Creates DEMO/TEST data for development and testing:
- Demo vendors with realistic data
- Test customers and addresses
- Sample products
- Demo orders
- Vendor 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 vendor 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 argparse
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 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 middleware.auth import AuthManager
from models.database.admin import PlatformAlert
from models.database.company import Company
from models.database.customer import Customer, CustomerAddress
from models.database.marketplace_import_job import MarketplaceImportJob
from models.database.marketplace_product import MarketplaceProduct
from models.database.marketplace_product_translation import MarketplaceProductTranslation
from models.database.order import Order, OrderItem
from models.database.product import Product
from models.database.user import User
from models.database.vendor import Role, Vendor, VendorUser
from models.database.vendor_domain import VendorDomain
from models.database.vendor_theme import VendorTheme
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 company configurations (NEW: Company-based architecture)
DEMO_COMPANIES = [
{
"name": "WizaCorp Ltd.",
"description": "Leading technology and electronics distributor",
"owner_email": "john.owner@wizacorp.com",
"owner_password": "password123",
"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",
"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",
"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 vendor configurations (linked to companies by index)
DEMO_VENDORS = [
{
"company_index": 0, # WizaCorp
"vendor_code": "WIZAMART",
"name": "WizaMart",
"subdomain": "wizamart",
"description": "Premium electronics and gadgets marketplace",
"theme_preset": "modern",
"custom_domain": "wizamart.shop",
},
{
"company_index": 1, # Fashion Group
"vendor_code": "FASHIONHUB",
"name": "Fashion Hub",
"subdomain": "fashionhub",
"description": "Trendy clothing and accessories",
"theme_preset": "vibrant",
"custom_domain": "fashionhub.store",
},
{
"company_index": 2, # BookWorld
"vendor_code": "BOOKSTORE",
"name": "The Book Store",
"subdomain": "bookstore",
"description": "Books, magazines, and educational materials",
"theme_preset": "classic",
"custom_domain": None,
},
]
# 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",
},
}
# =============================================================================
# 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 == "admin")).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 vendors, 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,
MarketplaceProduct,
Product,
VendorDomain,
VendorTheme,
Role,
VendorUser,
Vendor,
Company, # Delete companies (cascades to vendors)
PlatformAlert,
]
for table in tables_to_clear:
db.execute(delete(table))
# Delete non-admin users
db.execute(delete(User).where(User.role != "admin"))
db.commit()
print_success("All data deleted (admin preserved)")
# =============================================================================
# SEEDING FUNCTIONS
# =============================================================================
def create_demo_companies(db: Session, auth_manager: AuthManager) -> list[Company]:
"""Create demo companies with owner accounts."""
companies = []
# Determine how many companies to create based on mode
company_count = 1 if SEED_MODE == "minimal" else len(DEMO_COMPANIES)
companies_to_create = DEMO_COMPANIES[:company_count]
for company_data in companies_to_create:
# Check if company already exists
existing = db.execute(
select(Company).where(Company.name == company_data["name"])
).scalar_one_or_none()
if existing:
print_warning(f"Company already exists: {company_data['name']}")
companies.append(existing)
continue
# Check if owner user already exists
owner_user = db.execute(
select(User).where(User.email == company_data["owner_email"])
).scalar_one_or_none()
if not owner_user:
# Create owner user
owner_user = User(
username=company_data["owner_email"].split("@")[0],
email=company_data["owner_email"],
hashed_password=auth_manager.hash_password(
company_data["owner_password"]
),
role="vendor",
first_name=company_data["owner_first_name"],
last_name=company_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)
db.flush()
print_success(
f"Created owner user: {owner_user.email} (password: {company_data['owner_password']})"
)
else:
print_warning(f"Using existing user as owner: {owner_user.email}")
# Create company
company = Company(
name=company_data["name"],
description=company_data["description"],
owner_user_id=owner_user.id,
contact_email=company_data["contact_email"],
contact_phone=company_data.get("contact_phone"),
website=company_data.get("website"),
business_address=company_data.get("business_address"),
tax_number=company_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(company)
db.flush()
companies.append(company)
print_success(
f"Created company: {company.name} (Owner: {owner_user.email})"
)
db.flush()
return companies
def create_demo_vendors(
db: Session, companies: list[Company], auth_manager: AuthManager
) -> list[Vendor]:
"""Create demo vendors linked to companies."""
vendors = []
# Determine how many vendors to create based on mode
vendor_count = 1 if SEED_MODE == "minimal" else len(DEMO_VENDORS)
vendors_to_create = DEMO_VENDORS[:vendor_count]
for vendor_data in vendors_to_create:
# Check if vendor already exists
existing = db.execute(
select(Vendor).where(Vendor.vendor_code == vendor_data["vendor_code"])
).scalar_one_or_none()
if existing:
print_warning(f"Vendor already exists: {vendor_data['name']}")
vendors.append(existing)
continue
# Get company by index
company_index = vendor_data["company_index"]
if company_index >= len(companies):
print_error(
f"Invalid company_index {company_index} for vendor {vendor_data['name']}"
)
continue
company = companies[company_index]
# Create vendor linked to company (owner is inherited from company)
vendor = Vendor(
company_id=company.id, # Link to company
vendor_code=vendor_data["vendor_code"],
name=vendor_data["name"],
subdomain=vendor_data["subdomain"],
description=vendor_data["description"],
is_active=True,
is_verified=True,
created_at=datetime.now(UTC),
updated_at=datetime.now(UTC),
)
db.add(vendor)
db.flush()
# Link company owner to vendor as owner
vendor_user_link = VendorUser(
vendor_id=vendor.id,
user_id=company.owner_user_id,
user_type="owner",
is_active=True,
created_at=datetime.now(UTC),
)
db.add(vendor_user_link)
# Create vendor theme
theme_colors = THEME_PRESETS.get(
vendor_data["theme_preset"], THEME_PRESETS["modern"]
)
theme = VendorTheme(
vendor_id=vendor.id,
theme_name=vendor_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)
# Create custom domain if specified
if vendor_data.get("custom_domain"):
domain = VendorDomain(
vendor_id=vendor.id,
domain=vendor_data["custom_domain"], # ✅ Field is 'domain', not 'domain_name'
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)
vendors.append(vendor)
print_success(f"Created vendor: {vendor.name} ({vendor.vendor_code})")
db.flush()
return vendors
def create_demo_customers(
db: Session, vendor: Vendor, auth_manager: AuthManager, count: int
) -> list[Customer]:
"""Create demo customers for a vendor."""
customers = []
# Use a simple demo password for all customers
demo_password = "customer123"
for i in range(1, count + 1):
email = f"customer{i}@{vendor.subdomain}.example.com"
customer_number = f"CUST-{vendor.vendor_code}-{i:04d}"
# Check if customer already exists
existing_customer = (
db.query(Customer)
.filter(Customer.vendor_id == vendor.id, Customer.email == email)
.first()
)
if existing_customer:
customers.append(existing_customer)
continue # Skip creation, customer already exists
customer = Customer(
vendor_id=vendor.id,
email=email,
hashed_password=auth_manager.hash_password(demo_password),
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)
customers.append(customer)
db.flush()
new_count = len([c for c in customers if c.id is None or db.is_modified(c)])
if new_count > 0:
print_success(f"Created {new_count} customers for {vendor.name}")
else:
print_warning(f"Customers already exist for {vendor.name}")
return customers
def create_demo_products(db: Session, vendor: Vendor, count: int) -> list[Product]:
"""Create demo products for a vendor."""
products = []
for i in range(1, count + 1):
marketplace_product_id = f"{vendor.vendor_code}-MP-{i:04d}"
product_id = f"{vendor.vendor_code}-PROD-{i:03d}"
# Check if this product already exists (by vendor_sku)
existing_product = (
db.query(Product)
.filter(Product.vendor_id == vendor.id, Product.vendor_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://{vendor.subdomain}.example.com/products/sample-{i}",
image_link=f"https://{vendor.subdomain}.example.com/images/product-{i}.jpg",
price=str(Decimal(f"{(i * 10) % 500 + 9.99}")), # Store as string
brand=vendor.name,
gtin=f"TEST{vendor.id:02d}{i:010d}",
availability="in stock",
condition="new",
google_product_category="Electronics > Computers > Laptops",
marketplace="Wizamart",
vendor_name=vendor.name,
currency="EUR",
created_at=datetime.now(UTC),
updated_at=datetime.now(UTC),
)
db.add(marketplace_product)
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} - {vendor.name}",
description=f"This is a demo product for testing purposes in {vendor.name}. High quality and affordable.",
)
db.add(translation)
# Create the Product (vendor-specific entry)
product = Product(
vendor_id=vendor.id,
marketplace_product_id=marketplace_product.id,
vendor_sku=product_id, # Use vendor_sku for vendor'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)
products.append(product)
db.flush()
new_count = len([p for p in products if p.id is None or db.is_modified(p)])
if new_count > 0:
print_success(f"Created {new_count} products for {vendor.name}")
else:
print_warning(f"Products already exist for {vendor.name}")
return products
# =============================================================================
# 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 companies
print_step(4, "Creating demo companies...")
companies = create_demo_companies(db, auth_manager)
# Step 5: Create vendors
print_step(5, "Creating demo vendors...")
vendors = create_demo_vendors(db, companies, auth_manager)
# Step 6: Create customers
print_step(6, "Creating demo customers...")
for vendor in vendors:
create_demo_customers(
db, vendor, auth_manager, count=settings.seed_customers_per_vendor
)
# Step 7: Create products
print_step(7, "Creating demo products...")
for vendor in vendors:
create_demo_products(db, vendor, count=settings.seed_products_per_vendor)
# 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
company_count = db.query(Company).count()
vendor_count = db.query(Vendor).count()
user_count = db.query(User).count()
customer_count = db.query(Customer).count()
product_count = db.query(Product).count()
print("\n📊 Database Status:")
print(f" Companies: {company_count}")
print(f" Vendors: {vendor_count}")
print(f" Users: {user_count}")
print(f" Customers: {customer_count}")
print(f" Products: {product_count}")
# Show company details
companies = db.query(Company).all()
print("\n🏢 Demo Companies:")
for company in companies:
print(f"\n {company.name}")
print(f" Owner: {company.owner.email if company.owner else 'N/A'}")
print(f" Vendors: {len(company.vendors) if company.vendors else 0}")
print(f" Status: {'✓ Active' if company.is_active else '✗ Inactive'}")
if company.is_verified:
print(f" Verified: ✓")
# Show vendor details
vendors = db.query(Vendor).all()
print("\n🏪 Demo Vendors:")
for vendor in vendors:
print(f"\n {vendor.name} ({vendor.vendor_code})")
print(f" Subdomain: {vendor.subdomain}.{settings.platform_domain}")
# Query custom domains separately
custom_domain = (
db.query(VendorDomain)
.filter(VendorDomain.vendor_id == vendor.id, VendorDomain.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:
print(f" Custom: {domain_value}")
print(f" Status: {'✓ Active' if vendor.is_active else '✗ Inactive'}")
print("\n🔐 Demo Company Owner Credentials:")
print("─" * 70)
for i, company_data in enumerate(DEMO_COMPANIES[:company_count], 1):
company = companies[i - 1] if i <= len(companies) else None
print(f" Company {i}: {company_data['name']}")
print(f" Email: {company_data['owner_email']}")
print(f" Password: {company_data['owner_password']}")
if company and company.vendors:
for vendor in company.vendors:
print(
f" Vendor: http://localhost:8000/vendor/{vendor.vendor_code}/login"
)
print(
f" or http://{vendor.subdomain}.localhost:8000/vendor/login"
)
print()
print("\n🛒 Demo Customer Credentials:")
print("─" * 70)
print(" All customers:")
print(" Email: customer1@{subdomain}.example.com")
print(" Password: customer123")
print(" (Replace {subdomain} with vendor subdomain, e.g., wizamart)")
print()
print("\n🏪 Shop Access (Development):")
print("─" * 70)
for vendor in vendors:
print(f" {vendor.name}:")
print(
f" Path-based: http://localhost:8000/vendors/{vendor.vendor_code}/shop/"
)
print(f" Subdomain: http://{vendor.subdomain}.localhost:8000/")
print()
print("⚠️ ALL DEMO CREDENTIALS ARE INSECURE - For development only!")
print("\n🚀 NEXT STEPS:")
print(" 1. Start development: make dev")
print(" 2. Login as vendor:")
print(" • Path-based: http://localhost:8000/vendor/WIZAMART/login")
print(" • Subdomain: http://wizamart.localhost:8000/vendor/login")
print(" 3. Visit vendor shop: http://localhost:8000/vendors/WIZAMART/shop/")
print(" 4. Admin panel: http://localhost:8000/admin/login")
print(f" Username: {settings.admin_username}")
print(f" Password: {settings.admin_password}")
# =============================================================================
# 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()
Reference in New Issue View Git Blame Copy Permalink