Segment API

Segment CDP Integration

Implements production-grade Segment (Customer Data Platform) integration using the analytics-python SDK. When loaded, this skill makes the model implement Segment Spec-compliant tracking: track() for events, identify() for user traits, group() for account-level properties, page()/screen() for views, alias() for identity linking. All implementations follow Segment best practices: use WRITE_KEY from environment, batch events with configurable flush, use anonymousId for logged-out users, userId for authenticated users, implement context fields for device/channel info, and always include required fields from the Segment Spec.

TL;DR Checklist

• Use analytics-python SDK with SEGMENT_WRITE_KEY from environment variable
• Initialize analytics.write_key = WRITE_KEY and configure analytics.debug
• Use anonymousId for logged-out users (UUID), userId for authenticated users
• Always include either anonymousId OR userId in every call
• Follow Segment Spec: track(), identify(), group(), page(), alias()
• Use properties dict for track() and page(), traits dict for identify() and group()
• Include context dict for device, channel, app, location info when available
• Use timestamp (ISO 8601 or datetime object) for historical backfill
• Set flush_at (batch size) and flush_interval (seconds) appropriately
• Call analytics.flush() explicitly before application shutdown

When to Use

Use this skill when:

• Implementing a Customer Data Platform (CDP) for unified tracking
• Routing user data to multiple tools (Amplitude, Mixpanel, HubSpot, Salesforce)
• Building server-side tracking for Segment or RudderStack
• Implementing Segment Spec-compliant event tracking
• Managing user identity across anonymous and authenticated states
• Tracking page/screen views with rich properties
• Implementing account-based (B2B) tracking with group()
• Importing historical data with proper timestamps
• Building a data layer that feeds multiple analytics/marketing tools

When NOT to Use

• For direct Amplitude tracking only — use coding-amplitude-api instead
• For direct Mixpanel tracking only — use coding-mixpanel-api instead
• For simple single-tool tracking (Segment adds complexity)
• When you need real-time event streaming (Segment has near-real-time)
• For error/exception tracking — use coding-sentry-api
• For server-side logging only — use logging libraries

Core Workflow

1. Initialize SDK — Set analytics.write_key = WRITE_KEY from SEGMENT_WRITE_KEY environment variable. Configure optional: analytics.debug = True/False, analytics.on_error = handler, analytics.flush_at, analytics.flush_interval. Checkpoint: Validate with a test identify() or track() call and verify in debugger.

2. Define Identity Strategy — Handle user identity lifecycle:

   • Anonymous: Generate UUID anonymousId on first visit
   • At signup/login: Keep anonymousId, add userId, call alias(userId, anonymousId)
   • Authenticated: Use userId as primary identifier
   • At logout: Generate new anonymousId

   Checkpoint: Every call has exactly one identifier: anonymousId OR userId (or both).

3. Implement Segment Spec Calls — Use the 5 core methods:

   • identify(userId, traits, context): Update user profile traits
   • track(userId/anonymousId, event, properties, context): Track events
   • group(userId/anonymousId, groupId, traits, context): Account-level properties
   • page(userId/anonymousId, category, name, properties, context): Web page views
   • screen(userId/anonymousId, category, name, properties, context): Mobile screens
   • alias(newId, previousId): Link two identities

   Checkpoint: Every method call follows Segment Spec argument order and field naming.

4. Add Context & Metadata — Include context dict when available:

   • context.device: Device info (id, manufacturer, model)
   • context.app: App info (name, version, build)
   • context.channel: "server", "browser", "mobile"
   • context.ip: IP address for geolocation
   • context.locale: User locale
   • context.userAgent: Browser user agent
   • context.page: Current page info (url, path, referrer)

   Checkpoint: Context fields follow Segment Spec naming conventions.

5. Handle Batching & Flush — Configure:

   • flush_at: Batch size (default 15, adjust for volume)
   • flush_interval: Seconds between flushes (default 10)
   • Call analytics.flush() explicitly at shutdown
   • Implement error handler with analytics.on_error = handler

   Checkpoint: Application explicitly flushes before exit; errors are caught and logged.

Implementation Patterns

Pattern 1: Segment SDK Initialization (BAD vs GOOD)

"""Segment SDK initialization patterns.

Key concepts:
- WRITE_KEY: Project write key (public, not secret)
- Batch events by default (15 events per batch)
- Async flushing by default
- On_error handler for failed deliveries

Important: Segment SDK for Python is called 'analytics-python' on PyPI,
and the import is 'analytics' (not 'segment').
"""

from __future__ import annotations

import os
import json
import logging
from typing import Any, Optional, Dict, List, Callable
from datetime import datetime, timezone
from uuid import uuid4
from dataclasses import dataclass, field

import analytics  # from 'analytics-python' package

logger = logging.getLogger(__name__)


# ===================================================================
# ❌ BAD — hardcoded key, no error handler, no shutdown flush
# ===================================================================

def bad_segment_init_bad() -> None:
    """❌ BAD: Don't do any of these things."""
    
    # ❌ Hardcoded write key! Never commit this!
    analytics.write_key = "abcdef1234567890abcdef1234567890"
    
    # ❌ No error handler - failed deliveries are silent
    # ❌ No explicit flush at shutdown
    # ❌ Using generic event names
    analytics.track(
        user_id="user-123",
        event="click",  # ❌ Too generic
        properties={
            "button_name": "signup",  # ❌ snake_case instead of camelCase
        },
        # ❌ Missing context
    )


# ===================================================================
# ✅ GOOD — env-based config, error handling, graceful shutdown
# ===================================================================


class SegmentError(Exception):
    """Base exception for Segment integration errors."""
    pass


class SegmentConfigError(SegmentError):
    """Configuration is invalid or missing."""
    pass


@dataclass
class SegmentConfig:
    """Segment configuration from environment variables.
    
    Environment variables:
        SEGMENT_WRITE_KEY: Segment project write key (required)
        SEGMENT_DEBUG: Enable debug mode (1/0)
        SEGMENT_FLUSH_AT: Batch size (events per flush)
        SEGMENT_FLUSH_INTERVAL: Flush interval in seconds
    """
    
    write_key: Optional[str] = None
    debug: bool = False
    flush_at: int = 15
    flush_interval: float = 10.0
    gzip: bool = True
    sync_mode: bool = False
    
    @classmethod
    def from_env(cls) -> "SegmentConfig":
        """Load configuration from environment variables."""
        
        def parse_bool(env_var: str, default: bool) -> bool:
            val = os.environ.get(env_var)
            if val is None:
                return default
            return val.lower() in ("1", "true", "yes", "on")
        
        def parse_int(env_var: str, default: int) -> int:
            val = os.environ.get(env_var)
            if val is None:
                return default
            try:
                return int(val)
            except ValueError:
                return default
        
        def parse_float(env_var: str, default: float) -> float:
            val = os.environ.get(env_var)
            if val is None:
                return default
            try:
                return float(val)
            except ValueError:
                return default
        
        return cls(
            write_key=os.environ.get("SEGMENT_WRITE_KEY"),
            debug=parse_bool("SEGMENT_DEBUG", False),
            flush_at=parse_int("SEGMENT_FLUSH_AT", 15),
            flush_interval=parse_float("SEGMENT_FLUSH_INTERVAL", 10.0),
            gzip=parse_bool("SEGMENT_GZIP", True),
            sync_mode=parse_bool("SEGMENT_SYNC_MODE", False),
        )
    
    def is_enabled(self) -> bool:
        """Check if Segment should be enabled."""
        if not self.write_key:
            return False
        
        # Check for explicit disable
        if os.environ.get("SEGMENT_DISABLED") == "1":
            return False
        
        # Disable in test environment unless explicitly enabled
        env = os.environ.get("ENV", "").lower()
        if env in ("test", "testing", "local"):
            if os.environ.get("SEGMENT_FORCE_ENABLE") != "1":
                return False
        
        return True
    
    def validate(self) -> bool:
        """Validate configuration.
        
        Returns:
            True if valid
            
        Raises:
            SegmentConfigError: If invalid when enabled
        """
        if not self.is_enabled():
            logger.info("Segment disabled by configuration")
            return True
        
        if not self.write_key:
            raise SegmentConfigError("SEGMENT_WRITE_KEY is required when enabled")
        
        if self.flush_at < 1 or self.flush_at > 1000:
            raise SegmentConfigError(f"flush_at must be 1-1000, got {self.flush_at}")
        
        if self.flush_interval < 0.1:
            raise SegmentConfigError(f"flush_interval must be >= 0.1s, got {self.flush_interval}")
        
        return True


class SegmentClient:
    """Production-grade Segment client with lifecycle management.
    
    Features:
    - Configurable batching
    - Error handling
    - Graceful shutdown with flush
    - Identity management helpers
    - Context building
    """
    
    def __init__(self, config: SegmentConfig) -> None:
        self._config = config
        self._is_initialized = False
        self._shutdown_hooks: List[Callable[[], None]] = []
    
    def initialize(self) -> "SegmentClient":
        """Initialize the Segment SDK.
        
        Call this once at application startup.
        """
        if self._is_initialized:
            return self
        
        if not self._config.is_enabled():
            logger.info("Segment disabled - using no-op mode")
            self._is_initialized = True
            return self
        
        # Validate config
        self._config.validate()
        
        # Configure analytics SDK
        analytics.write_key = self._config.write_key
        analytics.debug = self._config.debug
        analytics.flush_at = self._config.flush_at
        analytics.flush_interval = self._config.flush_interval
        analytics.gzip = self._config.gzip
        analytics.sync_mode = self._config.sync_mode
        
        # Set error handler
        analytics.on_error = self._handle_error
        
        self._is_initialized = True
        
        logger.info(
            "Segment initialized: flush_at=%d, flush_interval=%.1fs, gzip=%s",
            self._config.flush_at,
            self._config.flush_interval,
            self._config.gzip,
        )
        
        return self
    
    def shutdown(self, flush: bool = True) -> None:
        """Shutdown the client and optionally flush remaining events.
        
        Call this at application shutdown.
        
        Args:
            flush: If True, flush all buffered events
        """
        if not self._is_initialized:
            return
        
        # Run shutdown hooks
        for hook in self._shutdown_hooks:
            try:
                hook()
            except Exception as e:
                logger.warning("Shutdown hook failed: %s", e)
        
        # Flush remaining events
        if flush and self._config.is_enabled():
            logger.info("Flushing remaining Segment events...")
            analytics.flush()
            logger.info("Segment flush complete")
        
        self._is_initialized = False
        logger.info("Segment client shut down")
    
    def _handle_error(self, exception: Exception, batch: List[Dict[str, Any]]) -> None:
        """Handle failed delivery.
        
        Args:
            exception: The exception that occurred
            batch: The batch of events that failed
        """
        logger.error(
            "Segment delivery failed: %s, batch_size=%d",
            exception,
            len(batch),
        )
    
    def add_shutdown_hook(self, hook: Callable[[], None]) -> None:
        """Add a hook to be called at shutdown.
        
        Args:
            hook: Callable to execute
        """
        self._shutdown_hooks.append(hook)
    
    # ===================================================================
    # Core Segment Spec Methods
    # ===================================================================
    
    def identify(
        self,
        user_id: Optional[str] = None,
        anonymous_id: Optional[str] = None,
        traits: Optional[Dict[str, Any]] = None,
        context: Optional[Dict[str, Any]] = None,
        timestamp: Optional[datetime] = None,
    ) -> None:
        """Identify a user and set their traits.
        
        Call when:
        - User signs up
        - User logs in
        - User updates their profile
        - Periodically to sync user traits
        
        Args:
            user_id: Your internal user identifier
            anonymous_id: Anonymous identifier (if no user_id)
            traits: User properties (name, email, plan, etc.)
            context: Context about the call
            timestamp: Historical timestamp (for backfill)
        """
        if not self._is_initialized:
            return
        
        if not self._config.is_enabled():
            return
        
        if not user_id and not anonymous_id:
            logger.warning("Segment identify called without user_id or anonymous_id")
            return
        
        # Build kwargs
        kwargs: Dict[str, Any] = {}
        
        if user_id:
            kwargs["user_id"] = str(user_id)
        if anonymous_id:
            kwargs["anonymous_id"] = str(anonymous_id)
        if traits:
            kwargs["traits"] = self._sanitize_traits(traits)
        if context:
            kwargs["context"] = context
        if timestamp:
            kwargs["timestamp"] = self._format_timestamp(timestamp)
        
        analytics.identify(**kwargs)
        logger.debug("Segment identify: user_id=%s, traits_keys=%s",
                    user_id, list(traits.keys()) if traits else [])
    
    def track(
        self,
        event: str,
        user_id: Optional[str] = None,
        anonymous_id: Optional[str] = None,
        properties: Optional[Dict[str, Any]] = None,
        context: Optional[Dict[str, Any]] = None,
        timestamp: Optional[datetime] = None,
    ) -> None:
        """Track an event.
        
        Call when:
        - User performs an action
        - User views something important
        - System state changes
        
        Args:
            event: Event name (PascalCase: "User Signed Up", "Checkout Completed")
            user_id: User identifier
            anonymous_id: Anonymous identifier
            properties: Event-specific properties
            context: Context about the call
            timestamp: Historical timestamp
        """
        if not self._is_initialized:
            return
        
        if not self._config.is_enabled():
            return
        
        if not event:
            logger.warning("Segment track called without event name")
            return
        
        if not user_id and not anonymous_id:
            logger.warning("Segment track called without user_id or anonymous_id")
            return
        
        kwargs: Dict[str, Any] = {"event": event}
        
        if user_id:
            kwargs["user_id"] = str(user_id)
        if anonymous_id:
            kwargs["anonymous_id"] = str(anonymous_id)
        if properties:
            kwargs["properties"] = self._sanitize_properties(properties)
        if context:
            kwargs["context"] = context
        if timestamp:
            kwargs["timestamp"] = self._format_timestamp(timestamp)
        
        analytics.track(**kwargs)
        logger.debug("Segment track: event=%s, user_id=%s", event, user_id)
    
    def group(
        self,
        group_id: str,
        user_id: Optional[str] = None,
        anonymous_id: Optional[str] = None,
        traits: Optional[Dict[str, Any]] = None,
        context: Optional[Dict[str, Any]] = None,
        timestamp: Optional[datetime] = None,
    ) -> None:
        """Associate a user with a group (account, company, team).
        
        Use for B2B analytics:
        - Link user to their company
        - Set company-level properties
        - Build account-based funnels
        
        Args:
            group_id: Account/company/team identifier
            user_id: User identifier
            anonymous_id: Anonymous identifier
            traits: Group properties (name, plan, employeeCount, etc.)
            context: Context about the call
            timestamp: Historical timestamp
        """
        if not self._is_initialized:
            return
        
        if not self._config.is_enabled():
            return
        
        if not group_id:
            logger.warning("Segment group called without group_id")
            return
        
        if not user_id and not anonymous_id:
            logger.warning("Segment group called without user_id or anonymous_id")
            return
        
        kwargs: Dict[str, Any] = {"group_id": str(group_id)}
        
        if user_id:
            kwargs["user_id"] = str(user_id)
        if anonymous_id:
            kwargs["anonymous_id"] = str(anonymous_id)
        if traits:
            kwargs["traits"] = self._sanitize_traits(traits)
        if context:
            kwargs["context"] = context
        if timestamp:
            kwargs["timestamp"] = self._format_timestamp(timestamp)
        
        analytics.group(**kwargs)
        logger.debug("Segment group: group_id=%s, user_id=%s", group_id, user_id)
    
    def page(
        self,
        name: Optional[str] = None,
        category: Optional[str] = None,
        user_id: Optional[str] = None,
        anonymous_id: Optional[str] = None,
        properties: Optional[Dict[str, Any]] = None,
        context: Optional[Dict[str, Any]] = None,
        timestamp: Optional[datetime] = None,
    ) -> None:
        """Track a page view (web).
        
        Call when:
        - User navigates to a page
        - Single-page app route changes
        
        Args:
            name: Page name (e.g., "Checkout", "Dashboard")
            category: Page category (e.g., "Billing", "Marketing")
            user_id: User identifier
            anonymous_id: Anonymous identifier
            properties: Page properties (url, path, title, referrer)
            context: Context about the call
            timestamp: Historical timestamp
        """
        if not self._is_initialized:
            return
        
        if not self._config.is_enabled():
            return
        
        if not user_id and not anonymous_id:
            logger.warning("Segment page called without user_id or anonymous_id")
            return
        
        kwargs: Dict[str, Any] = {}
        
        if user_id:
            kwargs["user_id"] = str(user_id)
        if anonymous_id:
            kwargs["anonymous_id"] = str(anonymous_id)
        if category:
            kwargs["category"] = category
        if name:
            kwargs["name"] = name
        if properties:
            kwargs["properties"] = self._sanitize_properties(properties)
        if context:
            kwargs["context"] = context
        if timestamp:
            kwargs["timestamp"] = self._format_timestamp(timestamp)
        
        analytics.page(**kwargs)
        logger.debug("Segment page: name=%s, category=%s, user_id=%s", name, category, user_id)
    
    def screen(
        self,
        name: Optional[str] = None,
        category: Optional[str] = None,
        user_id: Optional[str] = None,
        anonymous_id: Optional[str] = None,
        properties: Optional[Dict[str, Any]] = None,
        context: Optional[Dict[str, Any]] = None,
        timestamp: Optional[datetime] = None,
    ) -> None:
        """Track a screen view (mobile).
        
        Mobile equivalent of page().
        
        Args:
            name: Screen name
            category: Screen category
            user_id: User identifier
            anonymous_id: Anonymous identifier
            properties: Screen properties
            context: Context about the call
            timestamp: Historical timestamp
        """
        if not self._is_initialized:
            return
        
        if not self._config.is_enabled():
            return
        
        if not user_id and not anonymous_id:
            logger.warning("Segment screen called without user_id or anonymous_id")
            return
        
        kwargs: Dict[str, Any] = {}
        
        if user_id:
            kwargs["user_id"] = str(user_id)
        if anonymous_id:
            kwargs["anonymous_id"] = str(anonymous_id)
        if category:
            kwargs["category"] = category
        if name:
            kwargs["name"] = name
        if properties:
            kwargs["properties"] = self._sanitize_properties(properties)
        if context:
            kwargs["context"] = context
        if timestamp:
            kwargs["timestamp"] = self._format_timestamp(timestamp)
        
        analytics.screen(**kwargs)
        logger.debug("Segment screen: name=%s, user_id=%s", name, user_id)
    
    def alias(
        self,
        user_id: str,
        previous_id: Optional[str] = None,
        context: Optional[Dict[str, Any]] = None,
        timestamp: Optional[datetime] = None,
    ) -> None:
        """Link two user identities.
        
        Call when:
        - Anonymous user signs up (link anonymousId to userId)
        - User merges accounts
        
        Important:
        - In most SDKs, alias() takes (new_id, previous_id)
        - The new_id becomes the canonical identifier
        
        Args:
            user_id: The new canonical user identifier
            previous_id: The previous identifier (anonymousId or old userId)
            context: Context about the call
            timestamp: Historical timestamp
        """
        if not self._is_initialized:
            return
        
        if not self._config.is_enabled():
            return
        
        if not user_id:
            logger.warning("Segment alias called without user_id")
            return
        
        kwargs: Dict[str, Any] = {"user_id": str(user_id)}
        
        if previous_id:
            kwargs["previous_id"] = str(previous_id)
        if context:
            kwargs["context"] = context
        if timestamp:
            kwargs["timestamp"] = self._format_timestamp(timestamp)
        
        analytics.alias(**kwargs)
        logger.info("Segment alias: user_id=%s, previous_id=%s", user_id, previous_id)
    
    def flush(self) -> None:
        """Flush all buffered events."""
        if self._config.is_enabled():
            analytics.flush()
    
    # ===================================================================
    # Helpers
    # ===================================================================
    
    @staticmethod
    def generate_anonymous_id() -> str:
        """Generate a new anonymous ID (UUID).
        
        Call at first visit or after logout.
        
        Returns:
            UUID string for anonymous tracking
        """
        return str(uuid4())
    
    @staticmethod
    def build_context(
        channel: Optional[str] = "server",
        app_name: Optional[str] = None,
        app_version: Optional[str] = None,
        ip_address: Optional[str] = None,
        locale: Optional[str] = None,
        user_agent: Optional[str] = None,
        page_url: Optional[str] = None,
        page_path: Optional[str] = None,
        page_referrer: Optional[str] = None,
        device_id: Optional[str] = None,
        device_model: Optional[str] = None,
        device_manufacturer: Optional[str] = None,
        **additional: Any,
    ) -> Dict[str, Any]:
        """Build a Segment context dict.
        
        Context provides additional metadata about the call.
        
        Args:
            channel: "server", "browser", "mobile"
            app_name: Application name
            app_version: Application version
            ip_address: IP address for geolocation
            locale: User locale (e.g., "en-US")
            user_agent: Browser user agent
            page_url: Full page URL
            page_path: Page path
            page_referrer: Referrer URL
            device_id: Device identifier
            device_model: Device model
            device_manufacturer: Device manufacturer
            **additional: Additional context fields
        
        Returns:
            Context dict for Segment calls
        """
        context: Dict[str, Any] = {"channel": channel or "server"}
        
        if app_name or app_version:
            context["app"] = {}
            if app_name:
                context["app"]["name"] = app_name
            if app_version:
                context["app"]["version"] = app_version
        
        if ip_address:
            context["ip"] = ip_address
        
        if locale:
            context["locale"] = locale
        
        if user_agent:
            context["userAgent"] = user_agent
        
        if page_url or page_path or page_referrer:
            context["page"] = {}
            if page_url:
                context["page"]["url"] = page_url
            if page_path:
                context["page"]["path"] = page_path
            if page_referrer:
                context["page"]["referrer"] = page_referrer
        
        if device_id or device_model or device_manufacturer:
            context["device"] = {}
            if device_id:
                context["device"]["id"] = device_id
            if device_model:
                context["device"]["model"] = device_model
            if device_manufacturer:
                context["device"]["manufacturer"] = device_manufacturer
        
        # Add additional fields
        context.update(additional)
        
        return context
    
    def _sanitize_properties(self, properties: Dict[str, Any]) -> Dict[str, Any]:
        """Sanitize properties for Segment.
        
        Segment supports: string, number, boolean, list, None
        Nested objects are OK but flatten when possible.
        """
        result: Dict[str, Any] = {}
        
        for key, value in properties.items():
            if isinstance(value, (str, int, float, bool, type(None))):
                result[key] = value
            elif isinstance(value, list):
                result[key] = [
                    self._sanitize_value(v) if isinstance(v, dict) else v
                    for v in value
                ]
            elif isinstance(value, dict):
                # Nested objects are OK, but stringify if too large
                json_str = json.dumps(value)
                if len(json_str) < 1000:
                    result[key] = value
                else:
                    result[key] = "[object]"  # Too large
            else:
                result[key] = str(value)
        
        return result
    
    def _sanitize_traits(self, traits: Dict[str, Any]) -> Dict[str, Any]:
        """Sanitize user/group traits.
        
        Same as properties but with specific trait handling.
        """
        return self._sanitize_properties(traits)
    
    def _sanitize_value(self, value: Any) -> Any:
        """Sanitize a single value."""
        if isinstance(value, dict):
            return self._sanitize_properties(value)
        return value
    
    def _format_timestamp(self, dt: datetime) -> datetime:
        """Format timestamp for Segment.
        
        Segment SDK accepts datetime objects directly.
        Ensure timezone-aware.
        """
        if dt.tzinfo is None:
            # Naive datetime - assume UTC
            return dt.replace(tzinfo=timezone.utc)
        return dt


# Global client (lazy-loaded)
_global_client: Optional[SegmentClient] = None


def get_segment_client() -> SegmentClient:
    """Get or create global SegmentClient."""
    global _global_client
    if _global_client is None:
        config = SegmentConfig.from_env()
        _global_client = SegmentClient(config)
        if config.is_enabled():
            _global_client.initialize()
    return _global_client

Pattern 2: Identity Management & Standard Events

"""Identity management lifecycle and standard Segment events.

Identity lifecycle:
1. Anonymous visit → generate anonymousId
2. User signs up → alias(userId, anonymousId)
3. User logs in → use userId, sync traits with identify()
4. User logs out → generate new anonymousId

Standard event patterns follow Segment Spec conventions.
"""

from __future__ import annotations

import logging
from typing import Any, Optional, Dict
from dataclasses import dataclass
from datetime import datetime, timezone
from uuid import uuid4

logger = logging.getLogger(__name__)


class IdentityManager:
    """Manages user identity for Segment tracking.
    
    Handles the complete identity lifecycle:
    - Anonymous device ID generation
    - Alias at signup/login
    - Logout with new anonymous ID
    - Traits synchronization
    """
    
    def __init__(self, client: Any) -> None:
        self._client = client
        self._user_id: Optional[str] = None
        self._anonymous_id: str = str(uuid4())
        self._is_aliased: bool = False
    
    @property
    def user_id(self) -> Optional[str]:
        """Get current user ID (authenticated users)."""
        return self._user_id
    
    @property
    def anonymous_id(self) -> str:
        """Get current anonymous ID."""
        return self._anonymous_id
    
    @property
    def current_identifier(self) -> str:
        """Get the best identifier to use for tracking.
        
        Returns:
            user_id if authenticated, otherwise anonymous_id
        """
        return self._user_id if self._user_id else self._anonymous_id
    
    def set_anonymous_id(self, anonymous_id: str) -> None:
        """Set existing anonymous ID (from cookie/storage)."""
        self._anonymous_id = anonymous_id
        if not self._user_id:
            logger.info("Restored anonymous_id: %s", anonymous_id)
    
    def generate_new_anonymous_id(self) -> str:
        """Generate a new anonymous ID.
        
        Call at first visit or after logout.
        
        Returns:
            New UUID
        """
        self._anonymous_id = str(uuid4())
        self._user_id = None
        self._is_aliased = False
        logger.info("Generated new anonymous_id: %s", self._anonymous_id)
        return self._anonymous_id
    
    def signup(
        self,
        user_id: str,
        traits: Optional[Dict[str, Any]] = None,
        context: Optional[Dict[str, Any]] = None,
    ) -> None:
        """Handle user signup.
        
        1. Call alias() to link anonymousId to userId
        2. Call identify() to set user traits
        3. Track "User Signed Up" event
        
        Args:
            user_id: Your internal user identifier
            traits: User properties
            context: Context dict
        """
        previous_id = self._anonymous_id
        
        # Only alias if we haven't already aliased this pair
        if not self._is_aliased:
            self._client.alias(
                user_id=user_id,
                previous_id=previous_id,
                context=context,
            )
            self._is_aliased = True
            logger.info("Aliased anonymous_id=%s to user_id=%s", previous_id, user_id)
        
        # Set current user
        self._user_id = user_id
        
        # Identify the user with traits
        if traits:
            # First-touch traits via set_once pattern
            # In Segment, identify() sets traits that can be updated
            # For first-touch only, use traits that you only set at signup
            self._client.identify(
                user_id=user_id,
                traits=traits,
                context=context,
            )
        
        # Track signup event
        self._client.track(
            event="User Signed Up",
            user_id=user_id,
            properties={
                "signupMethod": traits.get("signupMethod") if traits else None,
                "referralCode": traits.get("referralCode") if traits else None,
            },
            context=context,
        )
    
    def login(
        self,
        user_id: str,
        traits: Optional[Dict[str, Any]] = None,
        context: Optional[Dict[str, Any]] = None,
        perform_alias: bool = False,
    ) -> None:
        """Handle user login.
        
        Args:
            user_id: Your internal user identifier
            traits: User properties to update
            context: Context dict
            perform_alias: Whether to call alias() (use False if already done at signup)
        """
        previous_id = self._user_id or self._anonymous_id
        
        if perform_alias and not self._is_aliased:
            self._client.alias(
                user_id=user_id,
                previous_id=previous_id,
                context=context,
            )
            self._is_aliased = True
        
        self._user_id = user_id
        
        # Update traits
        login_traits = dict(traits or {})
        login_traits["lastLoginDate"] = datetime.now(timezone.utc).isoformat()
        
        # Increment login count (if tracking)
        # Note: Segment doesn't have increment via identify - use your own tracking
        
        self._client.identify(
            user_id=user_id,
            traits=login_traits,
            context=context,
        )
        
        # Track login event
        self._client.track(
            event="User Logged In",
            user_id=user_id,
            properties={
                "loginMethod": login_traits.get("loginMethod"),
                "isNewDevice": login_traits.get("isNewDevice"),
            },
            context=context,
        )
    
    def logout(
        self,
        context: Optional[Dict[str, Any]] = None,
    ) -> None:
        """Handle user logout.
        
        1. Track "User Logged Out" event
        2. Update user traits (isLoggedIn = false)
        3. Generate new anonymous ID
        """
        if self._user_id:
            # Track logout event
            self._client.track(
                event="User Logged Out",
                user_id=self._user_id,
                properties={
                    "logoutDate": datetime.now(timezone.utc).isoformat(),
                },
                context=context,
            )
            
            # Update traits
            self._client.identify(
                user_id=self._user_id,
                traits={
                    "isLoggedIn": False,
                    "lastLogoutDate": datetime.now(timezone.utc).isoformat(),
                },
                context=context,
            )
        
        # Generate new anonymous identity
        self.generate_new_anonymous_id()


class StandardEvents:
    """Standard Segment Spec-compliant events.
    
    These follow Segment's Digital Analytics Specification and B2B SaaS spec.
    
    Event naming: "Title Case With Spaces"
    Properties: camelCase
    
    Reference: https://segment.com/docs/spec/
    """
    
    @staticmethod
    def track_user_signed_up(
        client: Any,
        user_id: str,
        signup_method: str,
        referral_code: Optional[str] = None,
        context: Optional[Dict[str, Any]] = None,
        **properties: Any,
    ) -> None:
        """Track "User Signed Up" event.
        
        Fires when a user completes the signup form and creates an account.
        
        Spec: https://segment.com/docs/spec/lifecycle/#user-signed-up
        """
        props: Dict[str, Any] = {
            "signupMethod": signup_method,
        }
        
        if referral_code:
            props["referralCode"] = referral_code
        
        props.update(properties)
        
        client.track(
            event="User Signed Up",
            user_id=user_id,
            properties=props,
            context=context,
        )
    
    @staticmethod
    def track_user_logged_in(
        client: Any,
        user_id: str,
        login_method: str,
        is_new_device: bool = False,
        context: Optional[Dict[str, Any]] = None,
        **properties: Any,
    ) -> None:
        """Track "User Logged In" event.
        
        Spec: https://segment.com/docs/spec/lifecycle/#user-logged-in
        """
        props: Dict[str, Any] = {
            "loginMethod": login_method,
            "isNewDevice": is_new_device,
        }
        props.update(properties)
        
        client.track(
            event="User Logged In",
            user_id=user_id,
            properties=props,
            context=context,
        )
    
    @staticmethod
    def track_user_logged_out(
        client: Any,
        user_id: str,
        context: Optional[Dict[str, Any]] = None,
        **properties: Any,
    ) -> None:
        """Track "User Logged Out" event."""
        client.track(
            event="User Logged Out",
            user_id=user_id,
            properties=properties,
            context=context,
        )
    
    @staticmethod
    def track_product_viewed(
        client: Any,
        user_id: str,
        product_id: str,
        product_name: str,
        product_sku: Optional[str] = None,
        category: Optional[str] = None,
        price: Optional[float] = None,
        currency: str = "USD",
        context: Optional[Dict[str, Any]] = None,
        **properties: Any,
    ) -> None:
        """Track "Product Viewed" event.
        
        Fires when a user views a product detail page.
        
        Spec: https://segment.com/docs/spec/ecommerce/#product-viewed
        """
        props: Dict[str, Any] = {
            "productId": product_id,
            "name": product_name,
        }
        
        if product_sku:
            props["sku"] = product_sku
        if category:
            props["category"] = category
        if price is not None:
            props["price"] = price
            props["currency"] = currency
        
        props.update(properties)
        
        client.track(
            event="Product Viewed",
            user_id=user_id,
            properties=props,
            context=context,
        )
    
    @staticmethod
    def track_checkout_started(
        client: Any,
        user_id: str,
        order_id: Optional[str] = None,
        total: Optional[float] = None,
        currency: str = "USD",
        item_count: Optional[int] = None,
        coupon: Optional[str] = None,
        context: Optional[Dict[str, Any]] = None,
        **properties: Any,
    ) -> None:
        """Track "Checkout Started" event.
        
        Fires when a user begins the checkout process.
        
        Spec: https://segment.com/docs/spec/ecommerce/#checkout-started
        """


…(truncated)