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