search_personalization/user_preferences.py

"""
User Preferences Module

Stores and retrieves per-user preference weights derived from UBI data.
Preferences are stored in a 'user_preferences' OpenSearch index and used
to pre-apply filters or sort results for returning users.
"""

import json
import logging
from typing import Dict, Any, Optional
from datetime import datetime
from search_personalization.data_loader import get_opensearch_client

logger = logging.getLogger(__name__)

INDEX_NAME = "user_preferences"

INDEX_MAPPING = {
    "mappings": {
        "properties": {
            "user_id": {"type": "keyword"},
            "preferred_max_price": {"type": "float"},
            "preferred_min_price": {"type": "float"},
            "preferred_categories": {"type": "keyword"},
            "price_sensitivity": {"type": "float"},
            "total_filter_events": {"type": "integer"},
            "updated_at": {"type": "date"},
        }
    }
}


def ensure_index_exists() -> None:
    """Create the user_preferences index if it doesn't exist."""
    client = get_opensearch_client()
    if not client.indices.exists(index=INDEX_NAME):
        client.indices.create(index=INDEX_NAME, body=INDEX_MAPPING)
        logger.info(f"Created index: {INDEX_NAME}")


def save_user_preferences(user_id: str, preferences: Dict[str, Any]) -> bool:
    """Save or update preferences for a user (upsert by user_id)."""
    client = get_opensearch_client()
    ensure_index_exists()
    preferences["user_id"] = user_id
    preferences["updated_at"] = datetime.utcnow().isoformat() + "Z"
    try:
        client.index(index=INDEX_NAME, id=user_id, body=preferences, refresh="wait_for")
        return True
    except Exception as e:
        logger.error(f"Failed to save preferences for {user_id}: {e}")
        return False


def get_user_preferences(user_id: str) -> Optional[Dict[str, Any]]:
    """Load preferences for a user. Returns None if not found."""
    client = get_opensearch_client()
    try:
        if not client.indices.exists(index=INDEX_NAME):
            return None
        resp = client.get(index=INDEX_NAME, id=user_id)
        return resp["_source"]
    except Exception:
        return None