#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Demo Retail POS Configurations Seeding Script for POS Service Creates realistic POS configurations for child retail outlets This script runs as a Kubernetes init job inside the pos-service container. It populates child retail tenants with POS system configurations. Usage: python /app/scripts/demo/seed_demo_pos_retail.py Environment Variables Required: POS_DATABASE_URL - PostgreSQL connection string for POS database DEMO_MODE - Set to 'production' for production seeding LOG_LEVEL - Logging level (default: INFO) """ import asyncio import uuid import sys import os from datetime import datetime, timezone, timedelta from pathlib import Path # Add app to path sys.path.insert(0, str(Path(__file__).parent.parent.parent)) # Add shared to path for demo utilities sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent.parent)) from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine from sqlalchemy.orm import sessionmaker from sqlalchemy import select import structlog from shared.utils.demo_dates import BASE_REFERENCE_DATE from app.models.pos_config import POSConfiguration # Configure logging structlog.configure( processors=[ structlog.stdlib.add_log_level, structlog.processors.TimeStamper(fmt="iso"), structlog.dev.ConsoleRenderer() ] ) logger = structlog.get_logger() # Fixed Demo Tenant IDs (must match tenant service) DEMO_TENANT_CHILD_1 = uuid.UUID("d4e5f6a7-b8c9-40d1-e2f3-a4b5c6d7e8f9") # Madrid Centro DEMO_TENANT_CHILD_2 = uuid.UUID("e5f6a7b8-c9d0-41e2-f3a4-b5c6d7e8f9a0") # Barcelona Gràcia DEMO_TENANT_CHILD_3 = uuid.UUID("f6a7b8c9-d0e1-42f3-a4b5-c6d7e8f9a0b1") # Valencia Ruzafa # POS system configurations for retail outlets RETAIL_POS_CONFIGS = [ (DEMO_TENANT_CHILD_1, "Madrid Centro", "square", "Square"), (DEMO_TENANT_CHILD_2, "Barcelona Gràcia", "square", "Square"), (DEMO_TENANT_CHILD_3, "Valencia Ruzafa", "sumup", "SumUp") # Different POS system for variety ] async def seed_retail_pos_for_tenant( db: AsyncSession, tenant_id: uuid.UUID, tenant_name: str, pos_system: str, provider_name: str ) -> dict: """ Generate a demo POS configuration for a retail tenant Args: db: Database session tenant_id: UUID of the child tenant tenant_name: Name of the tenant (for logging) pos_system: POS system type (square, sumup, etc.) provider_name: Provider name for display Returns: Dict with seeding statistics """ logger.info("─" * 80) logger.info(f"Generating POS config for: {tenant_name}") logger.info(f"Tenant ID: {tenant_id}") logger.info(f"POS System: {pos_system}") logger.info("─" * 80) # Check if config already exists result = await db.execute( select(POSConfiguration).where( POSConfiguration.tenant_id == tenant_id, POSConfiguration.pos_system == pos_system ).limit(1) ) existing = result.scalar_one_or_none() if existing: logger.info(f"POS config already exists for {tenant_name}, skipping") return {"tenant_id": str(tenant_id), "configs_created": 0, "skipped": True} # Create demo POS configuration for retail outlet config = POSConfiguration( id=uuid.uuid4(), tenant_id=tenant_id, pos_system=pos_system, provider_name=provider_name, is_active=True, is_connected=True, encrypted_credentials="demo_retail_credentials_encrypted", environment="sandbox", location_id=f"LOC-{tenant_name.replace(' ', '-').upper()}-001", merchant_id=f"MERCH-RETAIL-{tenant_name.replace(' ', '-').upper()}", sync_enabled=True, sync_interval_minutes="5", # Sync every 5 minutes for retail auto_sync_products=True, auto_sync_transactions=True, last_sync_at=BASE_REFERENCE_DATE - timedelta(minutes=5), last_successful_sync_at=BASE_REFERENCE_DATE - timedelta(minutes=5), last_sync_status="success", last_sync_message="Retail POS sync completed successfully", provider_settings={ "api_key": f"demo_retail_{pos_system}_api_key_***", "location_id": f"LOC-{tenant_name.replace(' ', '-').upper()}-001", "environment": "sandbox", "device_id": f"DEVICE-RETAIL-{str(tenant_id).split('-')[0].upper()}", "receipt_footer": f"¡Gracias por visitar {tenant_name}!", "tax_enabled": True, "tax_rate": 10.0, # 10% IVA "currency": "EUR" }, last_health_check_at=BASE_REFERENCE_DATE - timedelta(minutes=1), health_status="healthy", health_message="Retail POS system operational - all services running", created_at=BASE_REFERENCE_DATE - timedelta(days=60), # Configured 60 days ago updated_at=BASE_REFERENCE_DATE - timedelta(minutes=5), notes=f"Demo POS configuration for {tenant_name} retail outlet" ) db.add(config) await db.commit() logger.info(f" ✅ Created POS config: {pos_system}") logger.info("") return { "tenant_id": str(tenant_id), "tenant_name": tenant_name, "configs_created": 1, "pos_system": pos_system, "skipped": False } async def seed_retail_pos(db: AsyncSession): """ Seed retail POS configurations for all child tenant templates Args: db: Database session Returns: Dict with overall seeding statistics """ logger.info("=" * 80) logger.info("💳 Starting Demo Retail POS Seeding") logger.info("=" * 80) logger.info("Creating POS system configurations for retail outlets") logger.info("") results = [] # Seed POS configs for each retail outlet for tenant_id, tenant_name, pos_system, provider_name in RETAIL_POS_CONFIGS: logger.info("") result = await seed_retail_pos_for_tenant( db, tenant_id, f"{tenant_name} (Retail Outlet)", pos_system, provider_name ) results.append(result) # Calculate totals total_configs = sum(r["configs_created"] for r in results if not r["skipped"]) logger.info("=" * 80) logger.info("✅ Demo Retail POS Seeding Completed") logger.info("=" * 80) return { "service": "pos_retail", "tenants_seeded": len(results), "total_configs_created": total_configs, "results": results } async def main(): """Main execution function""" logger.info("Demo Retail POS Seeding Script Starting") logger.info("Mode: %s", os.getenv("DEMO_MODE", "development")) logger.info("Log Level: %s", os.getenv("LOG_LEVEL", "INFO")) # Get database URL from environment database_url = os.getenv("POS_DATABASE_URL") or os.getenv("DATABASE_URL") if not database_url: logger.error("❌ POS_DATABASE_URL or DATABASE_URL environment variable must be set") return 1 # Convert to async URL if needed if database_url.startswith("postgresql://"): database_url = database_url.replace("postgresql://", "postgresql+asyncpg://", 1) logger.info("Connecting to POS database") # Create engine and session engine = create_async_engine( database_url, echo=False, pool_pre_ping=True, pool_size=5, max_overflow=10 ) async_session = sessionmaker( engine, class_=AsyncSession, expire_on_commit=False ) try: async with async_session() as session: result = await seed_retail_pos(session) logger.info("") logger.info("📊 Retail POS Seeding Summary:") logger.info(f" ✅ Retail outlets configured: {result['tenants_seeded']}") logger.info(f" ✅ Total POS configs: {result['total_configs_created']}") logger.info("") # Print per-tenant details for tenant_result in result['results']: if not tenant_result['skipped']: logger.info( f" {tenant_result['tenant_name']}: " f"{tenant_result['pos_system']} configured" ) logger.info("") logger.info("🎉 Success! Retail POS systems are ready for cloning.") logger.info("") logger.info("POS configuration details:") logger.info(" ✓ Auto-sync enabled (5-minute intervals)") logger.info(" ✓ Product and transaction sync configured") logger.info(" ✓ Tax settings: 10% IVA (Spain)") logger.info(" ✓ Multiple POS providers (Square, SumUp)") logger.info(" ✓ Sandbox environment for testing") logger.info("") logger.info("Next steps:") logger.info(" 1. Seed retail forecasting models") logger.info(" 2. Seed retail alerts") logger.info(" 3. Test POS transaction integration") logger.info("") return 0 except Exception as e: logger.error("=" * 80) logger.error("❌ Demo Retail POS Seeding Failed") logger.error("=" * 80) logger.error("Error: %s", str(e)) logger.error("", exc_info=True) return 1 finally: await engine.dispose() if __name__ == "__main__": exit_code = asyncio.run(main()) sys.exit(exit_code)