MemoryBear/api/app/i18n/loader.py

"""
Translation file loader for i18n system.

This module handles loading translation files from multiple directories
(community edition + enterprise edition) and provides hot reload support.
"""

import json
import logging
from pathlib import Path
from typing import Any, Dict, List, Optional

logger = logging.getLogger(__name__)


class TranslationLoader:
    """
    Translation file loader that supports:
    - Loading from multiple directories (community + enterprise)
    - Hot reload of translation files
    - Automatic locale detection
    """

    def __init__(self, locales_dirs: Optional[List[str]] = None):
        """
        Initialize the translation loader.

        Args:
            locales_dirs: List of directories containing translation files.
                         If None, will auto-detect from settings.
        """
        if locales_dirs is None:
            locales_dirs = self._detect_locales_dirs()

        self.locales_dirs = [Path(d) for d in locales_dirs]
        logger.info(f"TranslationLoader initialized with directories: {self.locales_dirs}")

    def _detect_locales_dirs(self) -> List[str]:
        """
        Auto-detect translation directories from settings.

        Returns:
            List of translation directory paths
        """
        from app.core.config import settings

        dirs = []

        # 1. Core locales directory (community edition, required)
        core_dir = Path(settings.I18N_CORE_LOCALES_DIR)
        if core_dir.exists():
            dirs.append(str(core_dir))
            logger.debug(f"Found core locales directory: {core_dir}")
        else:
            logger.warning(f"Core locales directory not found: {core_dir}")

        # 2. Premium locales directory (enterprise edition, optional)
        if settings.I18N_PREMIUM_LOCALES_DIR:
            premium_dir = Path(settings.I18N_PREMIUM_LOCALES_DIR)
            if premium_dir.exists():
                dirs.append(str(premium_dir))
                logger.debug(f"Found premium locales directory: {premium_dir}")
        else:
            # Auto-detect premium directory
            premium_dir = Path("premium/locales")
            if premium_dir.exists():
                dirs.append(str(premium_dir))
                logger.debug(f"Auto-detected premium locales directory: {premium_dir}")

        if not dirs:
            logger.error("No translation directories found!")

        return dirs

    def get_available_locales(self) -> List[str]:
        """
        Get list of all available locales across all directories.

        Returns:
            List of locale codes (e.g., ['zh', 'en'])
        """
        locales = set()

        for locales_dir in self.locales_dirs:
            if not locales_dir.exists():
                continue

            for locale_dir in locales_dir.iterdir():
                if locale_dir.is_dir() and not locale_dir.name.startswith('.'):
                    locales.add(locale_dir.name)

        return sorted(list(locales))

    def load_locale(self, locale: str) -> Dict[str, Any]:
        """
        Load all translation files for a specific locale from all directories.

        Translation files are merged with priority:
        - Later directories override earlier directories
        - Enterprise translations override community translations

        Args:
            locale: Locale code (e.g., 'zh', 'en')

        Returns:
            Dictionary of translations organized by namespace
            Format: {namespace: {key: value, ...}, ...}
        """
        translations = {}

        # Load from each directory in order (later directories override earlier)
        for locales_dir in self.locales_dirs:
            locale_dir = locales_dir / locale
            if not locale_dir.exists():
                logger.debug(f"Locale directory not found: {locale_dir}")
                continue

            # Load all JSON files in this locale directory
            for json_file in locale_dir.glob("*.json"):
                namespace = json_file.stem

                try:
                    with open(json_file, "r", encoding="utf-8") as f:
                        new_translations = json.load(f)

                    # Merge translations (deep merge)
                    if namespace in translations:
                        translations[namespace] = self._deep_merge(
                            translations[namespace],
                            new_translations
                        )
                        logger.debug(
                            f"Merged translations: {locale}/{namespace} from {json_file}"
                        )
                    else:
                        translations[namespace] = new_translations
                        logger.debug(
                            f"Loaded translations: {locale}/{namespace} from {json_file}"
                        )

                except json.JSONDecodeError as e:
                    logger.error(
                        f"Failed to parse JSON file {json_file}: {e}"
                    )
                except Exception as e:
                    logger.error(
                        f"Failed to load translation file {json_file}: {e}"
                    )

        if not translations:
            logger.warning(f"No translations found for locale: {locale}")

        return translations

    def reload(self, locale: Optional[str] = None) -> Dict[str, Dict[str, Any]]:
        """
        Reload translation files.

        Args:
            locale: Specific locale to reload. If None, reloads all locales.

        Returns:
            Dictionary of reloaded translations
            Format: {locale: {namespace: {key: value}}}
        """
        if locale:
            logger.info(f"Reloading translations for locale: {locale}")
            return {locale: self.load_locale(locale)}
        else:
            logger.info("Reloading all translations")
            all_translations = {}
            for loc in self.get_available_locales():
                all_translations[loc] = self.load_locale(loc)
            return all_translations

    def _deep_merge(self, base: Dict, override: Dict) -> Dict:
        """
        Deep merge two dictionaries.

        Args:
            base: Base dictionary
            override: Dictionary with values to override

        Returns:
            Merged dictionary
        """
        result = base.copy()

        for key, value in override.items():
            if (
                key in result
                and isinstance(result[key], dict)
                and isinstance(value, dict)
            ):
                result[key] = self._deep_merge(result[key], value)
            else:
                result[key] = value

        return result