music-assistant-server

1"""
2Controller that owns translation strings for server-provided objects.
3
4Covers config entries, config option titles, provider manifests and localizable media item names.
5English source strings are authored in ``strings.json`` files (a shared
6``music_assistant/strings.json`` plus a per-provider/per-controller ``strings.json``);
7``build_translations.py`` compiles them into ``translations/en.json``, which Lokalise syncs against,
8and translated languages are downloaded to ``translations/<lang>.json``.
9
10At runtime ``translations/en.json`` is loaded eagerly (the final fallback) and each translated
11locale lazily on first use, so a lookup is always an in-memory dict get.
12"""
13
14from __future__ import annotations
15
16import asyncio
17import os
18from typing import TYPE_CHECKING, Any
19
20from music_assistant_models.helpers import create_safe_string
21
22from music_assistant.helpers.api import api_command
23from music_assistant.helpers.json import load_json_dict
24from music_assistant.models.core_controller import CoreController
25
26if TYPE_CHECKING:
27    from music_assistant_models.config_entries import CoreConfig
28
29# package paths (this file lives at music_assistant/controllers/translations/__init__.py)
30PACKAGE_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
31# translations/ holds the flat locale files: the generated en.json (English source, pushed to
32# Lokalise) and the downloaded per-language <lang>.json files. The hand-authored sources live in
33# strings.json files (music_assistant/strings.json + per-provider/per-controller strings.json).
34TRANSLATIONS_PATH = os.path.join(PACKAGE_ROOT, "translations")
35
36# the language the in-repo source strings are authored in
37SOURCE_LANGUAGE = "en"
38SOURCE_FILE = os.path.join(TRANSLATIONS_PATH, f"{SOURCE_LANGUAGE}.json")
39
40
41class TranslationController(CoreController):
42    """Loads and resolves translation strings for server-provided objects."""
43
44    domain: str = "translations"
45
46    def __init__(self, mass: Any) -> None:
47        """Initialize the translations controller."""
48        super().__init__(mass)
49        self.manifest.name = "Translations"
50        self.manifest.description = "Translation strings for server-provided objects."
51        # FQ key -> English source string (eager, always loaded)
52        self._source: dict[str, str] = {}
53        # locale -> {FQ key -> translated string} (lazy, populated on first use)
54        self._locales: dict[str, dict[str, str]] = {}
55        # locale -> file path, discovered at startup (no parsing)
56        self._locale_files: dict[str, str] = {}
57        # de-duplicate concurrent cold loads of the same locale
58        self._locale_locks: dict[str, asyncio.Lock] = {}
59        self._available_locales: set[str] = {SOURCE_LANGUAGE}
60
61    @property
62    def available_locales(self) -> set[str]:
63        """Return the set of locales the server can serve."""
64        return self._available_locales
65
66    @api_command("translations/locales")
67    async def get_available_locales(self) -> list[str]:
68        """Return the list of available UI locales (sorted)."""
69        return sorted(self._available_locales)
70
71    async def setup(self, config: CoreConfig) -> None:
72        """Load the translation strings."""
73        self.config = config
74        self._locale_files = await asyncio.to_thread(_discover_locale_files)
75        self._available_locales = {SOURCE_LANGUAGE, *self._locale_files}
76        self._source = await self._load_flat(SOURCE_FILE)
77        self.logger.debug(
78            "Loaded %s source strings across %s locale(s)",
79            len(self._source),
80            len(self._available_locales),
81        )
82
83    def get_translation(
84        self,
85        key: str,
86        locale: str | None = None,
87        owner: str | None = None,
88        params: list[str] | None = None,
89    ) -> str | None:
90        """
91        Resolve a translation key for the given locale.
92
93        Accepts either a fully-qualified key (e.g. "provider.ytmusic.manifest.name") or a
94        relative key (e.g. "settings.cookie.label") plus an optional ``owner`` hint
95        (provider domain/instance) used to build the owner-namespaced candidate. Returns the
96        resolved string, or None when nothing matches so the caller keeps its existing value.
97        Never raises.
98
99        :param key: Fully-qualified or relative translation key.
100        :param locale: Requested locale (e.g. "nl" or "de_DE"); None falls back to the source.
101        :param owner: Optional owner hint (provider domain/instance) for relative keys.
102        :param params: Optional positional arguments for ``{0}``/``{1}`` placeholders.
103        """
104        owner_prefix = _owner_prefix(owner) if owner else None
105        # Candidates are ordered owner-specific -> common -> bare (see _candidate_keys); each
106        # is probed in the requested locale, its base language and finally the English source.
107        for candidate in _candidate_keys(key, owner_prefix):
108            value = self._lookup(candidate, locale)
109            if value is not None:
110                return _format(value, params)
111        return None
112
113    async def ensure_locale_loaded(self, locale: str | None) -> None:
114        """
115        Load (and cache) the bundle for a locale and its base language if not already loaded.
116
117        Call this when a connection declares/changes its locale so subsequent lookups never
118        have to read from disk.
119
120        :param locale: The locale to warm up (e.g. "nl" or "de_DE").
121        """
122        if not locale:
123            return
124        for candidate in _locale_candidates(locale):
125            if (
126                candidate == SOURCE_LANGUAGE
127                or candidate in self._locales
128                or candidate not in self._locale_files
129            ):
130                continue
131            lock = self._locale_locks.setdefault(candidate, asyncio.Lock())
132            async with lock:
133                if candidate in self._locales:
134                    continue
135                self._locales[candidate] = await self._load_flat(self._locale_files[candidate])
136
137    async def reverse_lookup_media_names(self, query: str) -> set[str]:
138        """
139        Return the canonical (English) media names whose localized value matches ``query``.
140
141        Lets localized item names be found by the name the user sees: a text search that returns
142        nothing literally can be retried against these canonical names (which equal the items'
143        stored ``search_name``). Only genre and playlist names (``*.media.genre.*`` /
144        ``*.media.playlist.*`` under any owner — ``common.`` or a provider, e.g.
145        ``provider.builtin.media.playlist.*``) are considered — the searchable library media
146        types; browse and recommendation folder titles are display-only and never library items.
147
148        The reverse-translation always uses the metadata controller's configured language
149        (``CONF_LANGUAGE``), which doubles as the fallback search locale; an English, unknown or
150        untranslatable language yields an empty set (the literal search already covers English).
151
152        :param query: The (possibly localized) search query.
153        """
154        locale = self.mass.metadata.locale
155        normalized = create_safe_string(query, True, True)
156        if not normalized or not locale or locale.split("_")[0] == SOURCE_LANGUAGE:
157            return set()
158        await self.ensure_locale_loaded(locale)
159        bundle = self._locales.get(locale) or self._locales.get(locale.split("_")[0])
160        if not bundle:
161            return set()
162        matches: set[str] = set()
163        for key, value in bundle.items():
164            if not key.endswith(".name"):
165                continue
166            if ".media.genre." not in key and ".media.playlist." not in key:
167                continue
168            if normalized in create_safe_string(value, True, True):
169                if english := self._source.get(key):
170                    matches.add(english)
171        return matches
172
173    def _lookup(self, key: str, locale: str | None) -> str | None:
174        """Look up a single candidate key, locale bundle first then English source."""
175        if locale:
176            for candidate in _locale_candidates(locale):
177                if candidate == SOURCE_LANGUAGE:
178                    break
179                if (bundle := self._locales.get(candidate)) and key in bundle:
180                    return bundle[key]
181        return self._source.get(key)
182
183    async def _load_flat(self, path: str) -> dict[str, str]:
184        """Load a flat {fq_key: str} translations file, tolerating a missing file or errors."""
185        if not os.path.isfile(path):
186            return {}
187        try:
188            data = await load_json_dict(path)
189        except Exception as err:
190            self.logger.warning("Failed to load translations file %s: %s", path, err)
191            return {}
192        return {key: value for key, value in data.items() if isinstance(value, str)}
193
194
195def _discover_locale_files() -> dict[str, str]:
196    """
197    Discover the locale files in translations/ (blocking).
198
199    Returns a map of language -> file path for every ``translations/<lang>.json`` except the
200    English source ``en.json``. Each file is a flat, fully-qualified key->string map.
201    """
202    locale_files: dict[str, str] = {}
203    if not os.path.isdir(TRANSLATIONS_PATH):
204        return locale_files
205    for filename in os.listdir(TRANSLATIONS_PATH):  # noqa: PTH208, RUF100
206        if not filename.endswith(".json"):
207            continue
208        lang = filename[: -len(".json")]
209        if lang == SOURCE_LANGUAGE:
210            continue
211        locale_files[lang] = os.path.join(TRANSLATIONS_PATH, filename)
212    return locale_files
213
214
215def _owner_prefix(owner: str) -> str:
216    """Map an owner hint (provider domain/instance, or an already-rooted prefix) to a key prefix."""
217    if owner.startswith(("provider.", "core.", "common.")):
218        return owner
219    return f"provider.{owner}"
220
221
222def _candidate_keys(key: str, owner_prefix: str | None = None) -> list[str]:
223    """
224    Build the ordered list of translation keys to try for a requested key.
225
226    A fully-qualified key (starting with ``provider.``/``core.``/``common.``) is tried
227    as-is plus a ``common.`` rewrite that drops the owner segment. A relative key is tried
228    under the owner prefix (if any), then ``common.``, then bare. Multi-instance providers carry
229    an ``<domain>--<id>`` instance id, so the domain-only prefix is also tried before ``common.``.
230    Any candidate ending in ``.name`` also gets a bare fallback (dropping ``.name``).
231    """
232    roots = ("provider.", "core.", "common.")
233    base_candidates: list[str] = []
234    if key.startswith(roots):
235        base_candidates.append(key)
236        for prefix in ("provider.", "core."):
237            if key.startswith(prefix):
238                rest = key[len(prefix) :]
239                if "." in rest:
240                    base_candidates.append(f"common.{rest.split('.', 1)[1]}")
241                break
242    else:
243        if owner_prefix:
244            base_candidates.append(f"{owner_prefix}.{key}")
245            # a multi-instance owner is "provider.<domain>--<id>"; also try the bare domain
246            domain_prefix = owner_prefix.split("--", 1)[0]
247            if domain_prefix != owner_prefix:
248                base_candidates.append(f"{domain_prefix}.{key}")
249        base_candidates.append(f"common.{key}")
250        base_candidates.append(key)
251    candidates: list[str] = []
252    seen: set[str] = set()
253    for candidate in base_candidates:
254        variants = (
255            (candidate, candidate[: -len(".name")]) if candidate.endswith(".name") else (candidate,)
256        )
257        for variant in variants:
258            if variant not in seen:
259                seen.add(variant)
260                candidates.append(variant)
261    return candidates
262
263
264def _locale_candidates(locale: str) -> list[str]:
265    """Return [normalized locale, base language] (deduplicated, order preserved)."""
266    normalized = locale.replace("-", "_")
267    base = normalized.split("_", 1)[0]
268    return [normalized] if normalized == base else [normalized, base]
269
270
271def _format(template: str, params: list[str] | None) -> str:
272    """Substitute positional ``{0}``/``{1}`` placeholders, leaving the template intact on error."""
273    if not params:
274        return template
275    try:
276        return template.format(*params)
277    except IndexError, KeyError, ValueError:
278        return template
279