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