| 1 | from __future__ import annotations |
| 2 | |
| 3 | import base64 |
| 4 | import hashlib |
| 5 | import json |
| 6 | import os |
| 7 | import re |
| 8 | import shutil |
| 9 | import subprocess |
| 10 | import tempfile |
| 11 | import urllib.error |
| 12 | import urllib.request |
| 13 | import zipfile |
| 14 | from pathlib import Path |
| 15 | from typing import Any |
| 16 | |
| 17 | from helpers import files, plugins |
| 18 | from plugins._browser.helpers.config import PLUGIN_NAME, get_browser_config |
| 19 | from plugins._browser.helpers.runtime import close_all_runtimes_sync |
| 20 | |
| 21 | |
| 22 | EXTENSIONS_ROOT_DIR = ("usr", "_browser", "extensions") |
| 23 | EXTENSION_ID_RE = re.compile(r"^[a-p]{32}$") |
| 24 | WEB_STORE_ID_RE = re.compile(r"(?<![a-p])([a-p]{32})(?![a-p])") |
| 25 | CHROME_VERSION_RE = re.compile(r"(\d+(?:\.\d+){0,3})") |
| 26 | CHROME_I18N_MESSAGE_RE = re.compile(r"__MSG_([A-Za-z0-9_@.-]+)__") |
| 27 | DEFAULT_CHROME_PRODVERSION = "140.0.0.0" |
| 28 | EXTENSION_ID_ALPHABET = "abcdefghijklmnop" |
| 29 | CHROME_VERSION_COMMANDS = ( |
| 30 | ("google-chrome", "--version"), |
| 31 | ("chromium", "--version"), |
| 32 | ("chromium-browser", "--version"), |
| 33 | ) |
| 34 | WEB_STORE_DOWNLOAD_URL = ( |
| 35 | "https://clients2.google.com/service/update2/crx" |
| 36 | "?response=redirect" |
| 37 | "&prod=chromecrx" |
| 38 | "&prodversion={prodversion}" |
| 39 | "&acceptformat=crx2,crx3" |
| 40 | "&x=id%3D{extension_id}%26installsource%3Dondemand%26uc" |
| 41 | ) |
| 42 | |
| 43 | |
| 44 | def get_extensions_root() -> Path: |
| 45 | return Path(files.get_abs_path(*EXTENSIONS_ROOT_DIR)) |
| 46 | |
| 47 | |
| 48 | def parse_chrome_web_store_extension_id(value: str) -> str: |
| 49 | source = str(value or "").strip() |
| 50 | if EXTENSION_ID_RE.fullmatch(source): |
| 51 | return source |
| 52 | |
| 53 | match = WEB_STORE_ID_RE.search(source) |
| 54 | if match: |
| 55 | return match.group(1) |
| 56 | |
| 57 | raise ValueError("Enter a Chrome Web Store URL or a 32-character extension id.") |
| 58 | |
| 59 | |
| 60 | def list_browser_extensions() -> list[dict[str, Any]]: |
| 61 | config = get_browser_config() |
| 62 | enabled_paths = {str(Path(path).expanduser()) for path in config["extension_paths"]} |
| 63 | entries: list[dict[str, Any]] = [] |
| 64 | seen: set[str] = set() |
| 65 | |
| 66 | root = get_extensions_root() |
| 67 | if root.exists(): |
| 68 | for manifest_path in sorted(root.glob("**/manifest.json")): |
| 69 | if any(part.startswith(".") for part in manifest_path.relative_to(root).parts): |
| 70 | continue |
| 71 | entry = _extension_entry(manifest_path.parent, enabled_paths) |
| 72 | seen.add(entry["path"]) |
| 73 | entries.append(entry) |
| 74 | |
| 75 | for configured_path in config["extension_paths"]: |
| 76 | extension_dir = Path(configured_path).expanduser() |
| 77 | extension_path = str(extension_dir) |
| 78 | if extension_path in seen or not (extension_dir / "manifest.json").is_file(): |
| 79 | continue |
| 80 | entries.append(_extension_entry(extension_dir, enabled_paths)) |
| 81 | seen.add(extension_path) |
| 82 | |
| 83 | return entries |
| 84 | |
| 85 | |
| 86 | def install_chrome_web_store_extension(source: str) -> dict[str, Any]: |
| 87 | extension_id = parse_chrome_web_store_extension_id(source) |
| 88 | target = get_extensions_root() / "chrome-web-store" / extension_id |
| 89 | |
| 90 | with tempfile.TemporaryDirectory(prefix="browser-extension-control-") as tmp: |
| 91 | archive_path = Path(tmp) / f"{extension_id}.crx" |
| 92 | _download_crx(extension_id, archive_path) |
| 93 | payload_path = Path(tmp) / f"{extension_id}.zip" |
| 94 | payload_path.write_bytes(_crx_zip_payload(archive_path.read_bytes())) |
| 95 | target.parent.mkdir(parents=True, exist_ok=True) |
| 96 | try: |
| 97 | with zipfile.ZipFile(payload_path) as archive: |
| 98 | manifest = json.loads(archive.read("manifest.json")) |
| 99 | except KeyError as exc: |
| 100 | raise ValueError("Downloaded extension did not contain a manifest.json file.") from exc |
| 101 | except (json.JSONDecodeError, UnicodeDecodeError, zipfile.BadZipFile) as exc: |
| 102 | raise ValueError("Downloaded extension contained an invalid manifest.json file.") from exc |
| 103 | if not isinstance(manifest, dict): |
| 104 | raise ValueError("Downloaded extension contained an invalid manifest.json file.") |
| 105 | |
| 106 | current_manifest = _read_manifest(target) |
| 107 | if not ( |
| 108 | target.is_dir() |
| 109 | and manifest.get("version") |
| 110 | and manifest.get("version") == current_manifest.get("version") |
| 111 | ): |
| 112 | with tempfile.TemporaryDirectory( |
| 113 | prefix=f".{extension_id}-install-", |
| 114 | dir=target.parent, |
| 115 | ) as extracted: |
| 116 | extracted_path = Path(extracted) |
| 117 | _safe_extract_zip(payload_path, extracted_path) |
| 118 | config = get_browser_config() |
| 119 | if target.exists() and str(target) in config["extension_paths"]: |
| 120 | close_all_runtimes_sync() |
| 121 | _replace_extension_dir(extracted_path, target) |
| 122 | |
| 123 | config = _enable_extension_path(target) |
| 124 | manifest = _read_manifest(target) |
| 125 | return { |
| 126 | "ok": True, |
| 127 | "id": extension_id, |
| 128 | "name": _manifest_label(target, manifest, "name") or extension_id, |
| 129 | "version": manifest.get("version") or "", |
| 130 | "path": str(target), |
| 131 | "extension_paths": config["extension_paths"], |
| 132 | } |
| 133 | |
| 134 | |
| 135 | def _replace_extension_dir(source: Path, target: Path) -> None: |
| 136 | if not target.exists(): |
| 137 | source.rename(target) |
| 138 | return |
| 139 | |
| 140 | with tempfile.TemporaryDirectory( |
| 141 | prefix=f".{target.name}-previous-", |
| 142 | dir=target.parent, |
| 143 | ) as backup_dir: |
| 144 | backup = Path(backup_dir) / target.name |
| 145 | target.rename(backup) |
| 146 | try: |
| 147 | source.rename(target) |
| 148 | except BaseException: |
| 149 | backup.rename(target) |
| 150 | raise |
| 151 | |
| 152 | |
| 153 | def set_browser_extension_enabled(extension_path: str, enabled: bool) -> dict[str, Any]: |
| 154 | raw_path = str(extension_path or "").strip() |
| 155 | if not raw_path: |
| 156 | raise ValueError("Choose an extension first.") |
| 157 | |
| 158 | path = str(Path(raw_path).expanduser()) |
| 159 | directory = Path(path) |
| 160 | if enabled and not (directory / "manifest.json").is_file(): |
| 161 | raise ValueError("Extension folder must contain a manifest.json file.") |
| 162 | |
| 163 | config = get_browser_config() |
| 164 | paths = list(config["extension_paths"]) |
| 165 | if enabled: |
| 166 | if path not in paths: |
| 167 | paths.append(path) |
| 168 | else: |
| 169 | paths = [item for item in paths if str(Path(item).expanduser()) != path] |
| 170 | |
| 171 | config["extension_paths"] = paths |
| 172 | plugins.save_plugin_config(PLUGIN_NAME, "", "", config) |
| 173 | return config |
| 174 | |
| 175 | |
| 176 | def uninstall_browser_extension(extension_path: str) -> dict[str, Any]: |
| 177 | raw_path = str(extension_path or "").strip() |
| 178 | if not raw_path: |
| 179 | raise ValueError("Choose an extension first.") |
| 180 | |
| 181 | root = get_extensions_root().resolve() |
| 182 | extension_dir = Path(raw_path).expanduser().resolve() |
| 183 | if extension_dir == root or not extension_dir.is_relative_to(root): |
| 184 | raise ValueError("Only Browser-managed extension folders can be deleted.") |
| 185 | if not extension_dir.is_dir(): |
| 186 | raise ValueError("Extension folder was not found.") |
| 187 | |
| 188 | manifest = _read_manifest(extension_dir) |
| 189 | name = ( |
| 190 | _manifest_label(extension_dir, manifest, "name") |
| 191 | or _manifest_label(extension_dir, manifest, "short_name") |
| 192 | or extension_dir.name |
| 193 | ) |
| 194 | config = get_browser_config() |
| 195 | config["extension_paths"] = [ |
| 196 | path |
| 197 | for path in config["extension_paths"] |
| 198 | if Path(path).expanduser().resolve() != extension_dir |
| 199 | ] |
| 200 | |
| 201 | try: |
| 202 | shutil.rmtree(extension_dir) |
| 203 | except OSError as exc: |
| 204 | raise ValueError(f"Could not delete extension folder: {exc}") from exc |
| 205 | |
| 206 | plugins.save_plugin_config(PLUGIN_NAME, "", "", config) |
| 207 | return { |
| 208 | "ok": True, |
| 209 | "name": name, |
| 210 | "path": str(extension_dir), |
| 211 | "extension_paths": config["extension_paths"], |
| 212 | } |
| 213 | |
| 214 | |
| 215 | def _download_crx(extension_id: str, archive_path: Path) -> None: |
| 216 | prodversion = _detect_chrome_prodversion() |
| 217 | url = _build_web_store_download_url(extension_id, prodversion=prodversion) |
| 218 | request = urllib.request.Request( |
| 219 | url, |
| 220 | headers={ |
| 221 | "User-Agent": ( |
| 222 | "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 " |
| 223 | f"(KHTML, like Gecko) Chrome/{prodversion} Safari/537.36" |
| 224 | ) |
| 225 | }, |
| 226 | ) |
| 227 | try: |
| 228 | response = urllib.request.urlopen(request, timeout=120) |
| 229 | except urllib.error.HTTPError as exc: |
| 230 | raise ValueError( |
| 231 | f"Chrome Web Store download failed with HTTP {exc.code} for Chrome {prodversion}." |
| 232 | ) from exc |
| 233 | except urllib.error.URLError as exc: |
| 234 | reason = getattr(exc, "reason", exc) |
| 235 | raise ValueError(f"Chrome Web Store download failed: {reason}.") from exc |
| 236 | |
| 237 | with response: |
| 238 | status = response.getcode() |
| 239 | data = response.read() |
| 240 | if not data: |
| 241 | raise ValueError( |
| 242 | "Chrome Web Store did not return an extension package " |
| 243 | f"(HTTP {status}, Chrome {prodversion})." |
| 244 | ) |
| 245 | archive_path.write_bytes(data) |
| 246 | |
| 247 | |
| 248 | def _build_web_store_download_url(extension_id: str, *, prodversion: str | None = None) -> str: |
| 249 | return WEB_STORE_DOWNLOAD_URL.format( |
| 250 | extension_id=extension_id, |
| 251 | prodversion=_normalize_chrome_prodversion(prodversion or "") or DEFAULT_CHROME_PRODVERSION, |
| 252 | ) |
| 253 | |
| 254 | |
| 255 | def _detect_chrome_prodversion() -> str: |
| 256 | env_version = _normalize_chrome_prodversion(os.environ.get("A0_BROWSER_EXTENSION_PRODVERSION", "")) |
| 257 | if env_version: |
| 258 | return env_version |
| 259 | |
| 260 | for command in CHROME_VERSION_COMMANDS: |
| 261 | try: |
| 262 | completed = subprocess.run( |
| 263 | command, |
| 264 | check=False, |
| 265 | capture_output=True, |
| 266 | text=True, |
| 267 | timeout=5, |
| 268 | ) |
| 269 | except (OSError, subprocess.TimeoutExpired): |
| 270 | continue |
| 271 | |
| 272 | version = _normalize_chrome_prodversion( |
| 273 | " ".join(part for part in (completed.stdout, completed.stderr) if part) |
| 274 | ) |
| 275 | if version: |
| 276 | return version |
| 277 | |
| 278 | return DEFAULT_CHROME_PRODVERSION |
| 279 | |
| 280 | |
| 281 | def _normalize_chrome_prodversion(value: str) -> str: |
| 282 | match = CHROME_VERSION_RE.search(str(value or "")) |
| 283 | if not match: |
| 284 | return "" |
| 285 | parts = match.group(1).split(".") |
| 286 | return ".".join((parts + ["0", "0", "0", "0"])[:4]) |
| 287 | |
| 288 | |
| 289 | def _crx_zip_payload(data: bytes) -> bytes: |
| 290 | if data.startswith(b"PK"): |
| 291 | return data |
| 292 | if data[:4] != b"Cr24": |
| 293 | raise ValueError("Downloaded package is not a CRX or ZIP archive.") |
| 294 | |
| 295 | version = int.from_bytes(data[4:8], "little") |
| 296 | if version == 2: |
| 297 | public_key_len = int.from_bytes(data[8:12], "little") |
| 298 | signature_len = int.from_bytes(data[12:16], "little") |
| 299 | offset = 16 + public_key_len + signature_len |
| 300 | elif version == 3: |
| 301 | header_len = int.from_bytes(data[8:12], "little") |
| 302 | offset = 12 + header_len |
| 303 | else: |
| 304 | raise ValueError(f"Unsupported CRX version: {version}.") |
| 305 | |
| 306 | payload = data[offset:] |
| 307 | if not payload.startswith(b"PK"): |
| 308 | raise ValueError("CRX payload did not contain a ZIP archive.") |
| 309 | return payload |
| 310 | |
| 311 | |
| 312 | def _safe_extract_zip(archive_path: Path, target_dir: Path) -> None: |
| 313 | target_dir.mkdir(parents=True, exist_ok=True) |
| 314 | root = target_dir.resolve() |
| 315 | with zipfile.ZipFile(archive_path) as archive: |
| 316 | for member in archive.infolist(): |
| 317 | destination = (target_dir / member.filename).resolve() |
| 318 | if not destination.is_relative_to(root): |
| 319 | raise ValueError("Extension archive contains an unsafe path.") |
| 320 | if member.is_dir(): |
| 321 | destination.mkdir(parents=True, exist_ok=True) |
| 322 | continue |
| 323 | destination.parent.mkdir(parents=True, exist_ok=True) |
| 324 | with archive.open(member) as source, destination.open("wb") as output: |
| 325 | shutil.copyfileobj(source, output) |
| 326 | |
| 327 | |
| 328 | def _enable_extension_path(extension_path: Path) -> dict[str, Any]: |
| 329 | config = get_browser_config() |
| 330 | path = str(extension_path) |
| 331 | paths = list(config["extension_paths"]) |
| 332 | if path not in paths: |
| 333 | paths.append(path) |
| 334 | config["extension_paths"] = paths |
| 335 | plugins.save_plugin_config(PLUGIN_NAME, "", "", config) |
| 336 | return config |
| 337 | |
| 338 | |
| 339 | def _extension_entry(extension_dir: Path, enabled_paths: set[str]) -> dict[str, Any]: |
| 340 | manifest = _read_manifest(extension_dir) |
| 341 | extension_path = str(extension_dir) |
| 342 | can_delete = _is_managed_extension_dir(extension_dir) |
| 343 | extension_id = _extension_runtime_id(extension_dir, manifest) |
| 344 | source_id = _extension_source_id(extension_dir) |
| 345 | ui = _extension_ui(extension_id, manifest) |
| 346 | name = ( |
| 347 | _manifest_label(extension_dir, manifest, "name") |
| 348 | or _manifest_label(extension_dir, manifest, "short_name") |
| 349 | or extension_dir.name |
| 350 | ) |
| 351 | return { |
| 352 | "id": extension_id, |
| 353 | "source_id": source_id, |
| 354 | "name": name, |
| 355 | "raw_name": manifest.get("name") or "", |
| 356 | "description": _manifest_label(extension_dir, manifest, "description"), |
| 357 | "version": manifest.get("version") or "", |
| 358 | "path": extension_path, |
| 359 | "enabled": extension_path in enabled_paths, |
| 360 | "managed": can_delete, |
| 361 | "can_delete": can_delete, |
| 362 | "has_ui": bool(ui["open_url"]), |
| 363 | "open_url": ui["open_url"], |
| 364 | "open_label": ui["open_label"], |
| 365 | "ui": ui, |
| 366 | } |
| 367 | |
| 368 | |
| 369 | def _is_managed_extension_dir(extension_dir: Path) -> bool: |
| 370 | try: |
| 371 | root = get_extensions_root().resolve() |
| 372 | path = extension_dir.expanduser().resolve() |
| 373 | except OSError: |
| 374 | return False |
| 375 | return path != root and path.is_relative_to(root) |
| 376 | |
| 377 | |
| 378 | def _read_manifest(extension_path: Path) -> dict[str, Any]: |
| 379 | manifest_path = extension_path / "manifest.json" |
| 380 | try: |
| 381 | return json.loads(manifest_path.read_text(encoding="utf-8")) |
| 382 | except Exception: |
| 383 | return {} |
| 384 | |
| 385 | |
| 386 | def _extension_runtime_id(extension_dir: Path, manifest: dict[str, Any]) -> str: |
| 387 | key_id = _extension_id_from_manifest_key(str(manifest.get("key") or "")) |
| 388 | if key_id: |
| 389 | return key_id |
| 390 | return _extension_id_from_path(extension_dir) |
| 391 | |
| 392 | |
| 393 | def _extension_source_id(extension_dir: Path) -> str: |
| 394 | name = extension_dir.name |
| 395 | if not EXTENSION_ID_RE.fullmatch(name): |
| 396 | return "" |
| 397 | try: |
| 398 | if extension_dir.parent.name == "chrome-web-store": |
| 399 | return name |
| 400 | except OSError: |
| 401 | return "" |
| 402 | return "" |
| 403 | |
| 404 | |
| 405 | def _extension_id_from_manifest_key(key: str) -> str: |
| 406 | raw_key = str(key or "").strip() |
| 407 | if not raw_key: |
| 408 | return "" |
| 409 | try: |
| 410 | padding = "=" * (-len(raw_key) % 4) |
| 411 | public_key = base64.b64decode(raw_key + padding, validate=True) |
| 412 | except Exception: |
| 413 | return "" |
| 414 | return _extension_id_from_hash_input(public_key) |
| 415 | |
| 416 | |
| 417 | def _extension_id_from_path(extension_dir: Path) -> str: |
| 418 | return _extension_id_from_hash_input(str(extension_dir).encode("utf-8")) |
| 419 | |
| 420 | |
| 421 | def _extension_id_from_hash_input(value: bytes) -> str: |
| 422 | digest = hashlib.sha256(value).hexdigest()[:32] |
| 423 | return "".join(EXTENSION_ID_ALPHABET[int(char, 16)] for char in digest) |
| 424 | |
| 425 | |
| 426 | def _extension_ui(extension_id: str, manifest: dict[str, Any]) -> dict[str, Any]: |
| 427 | targets = _extension_ui_targets(extension_id, manifest) |
| 428 | primary = targets[0] if targets else {} |
| 429 | return { |
| 430 | "open_url": primary.get("url", ""), |
| 431 | "open_label": primary.get("label", ""), |
| 432 | "targets": targets, |
| 433 | } |
| 434 | |
| 435 | |
| 436 | def _extension_ui_targets(extension_id: str, manifest: dict[str, Any]) -> list[dict[str, str]]: |
| 437 | candidates: list[tuple[str, str, Any]] = [ |
| 438 | ("options", "Options", _manifest_nested_value(manifest, "options_ui", "page")), |
| 439 | ("options", "Options", manifest.get("options_page")), |
| 440 | ("popup", "Popup", _manifest_nested_value(manifest, "action", "default_popup")), |
| 441 | ("popup", "Popup", _manifest_nested_value(manifest, "browser_action", "default_popup")), |
| 442 | ("popup", "Popup", _manifest_nested_value(manifest, "page_action", "default_popup")), |
| 443 | ("side_panel", "Side panel", _manifest_nested_value(manifest, "side_panel", "default_path")), |
| 444 | ("devtools", "DevTools", manifest.get("devtools_page")), |
| 445 | ] |
| 446 | |
| 447 | chrome_url_overrides = manifest.get("chrome_url_overrides") |
| 448 | if isinstance(chrome_url_overrides, dict): |
| 449 | candidates.extend( |
| 450 | ( |
| 451 | (f"chrome_url_override_{name}", _extension_override_label(name), page) |
| 452 | for name, page in chrome_url_overrides.items() |
| 453 | ) |
| 454 | ) |
| 455 | |
| 456 | targets: list[dict[str, str]] = [] |
| 457 | seen_urls: set[str] = set() |
| 458 | for kind, label, page in candidates: |
| 459 | url = _extension_page_url(extension_id, page) |
| 460 | if not url or url in seen_urls: |
| 461 | continue |
| 462 | seen_urls.add(url) |
| 463 | targets.append( |
| 464 | { |
| 465 | "kind": kind, |
| 466 | "label": label, |
| 467 | "page": str(page or "").strip(), |
| 468 | "url": url, |
| 469 | } |
| 470 | ) |
| 471 | return targets |
| 472 | |
| 473 | |
| 474 | def _manifest_nested_value(manifest: dict[str, Any], key: str, nested_key: str) -> Any: |
| 475 | value = manifest.get(key) |
| 476 | if not isinstance(value, dict): |
| 477 | return "" |
| 478 | return value.get(nested_key) |
| 479 | |
| 480 | |
| 481 | def _extension_override_label(name: str) -> str: |
| 482 | normalized = str(name or "").replace("_", " ").strip() |
| 483 | return normalized[:1].upper() + normalized[1:] if normalized else "Extension page" |
| 484 | |
| 485 | |
| 486 | def _extension_page_url(extension_id: str, page: Any) -> str: |
| 487 | page_path = str(page or "").strip() |
| 488 | if not extension_id or not page_path: |
| 489 | return "" |
| 490 | if re.match(r"^[a-z][a-z0-9+.-]*:", page_path, flags=re.IGNORECASE): |
| 491 | return "" |
| 492 | page_path = page_path.lstrip("/") |
| 493 | if not page_path: |
| 494 | return "" |
| 495 | return f"chrome-extension://{extension_id}/{page_path}" |
| 496 | |
| 497 | |
| 498 | def _manifest_label(extension_dir: Path, manifest: dict[str, Any], key: str) -> str: |
| 499 | value = str(manifest.get(key) or "").strip() |
| 500 | if not value: |
| 501 | return "" |
| 502 | |
| 503 | messages = _load_locale_messages(extension_dir, str(manifest.get("default_locale") or "")) |
| 504 | if not messages: |
| 505 | return "" if CHROME_I18N_MESSAGE_RE.fullmatch(value) else value |
| 506 | |
| 507 | def replace_message(match: re.Match[str]) -> str: |
| 508 | message_key = match.group(1) |
| 509 | message = _resolve_locale_message(messages, message_key) |
| 510 | return message if message is not None else match.group(0) |
| 511 | |
| 512 | resolved = CHROME_I18N_MESSAGE_RE.sub(replace_message, value).strip() |
| 513 | if CHROME_I18N_MESSAGE_RE.fullmatch(resolved): |
| 514 | return "" |
| 515 | return resolved |
| 516 | |
| 517 | |
| 518 | def _load_locale_messages(extension_dir: Path, default_locale: str) -> dict[str, Any]: |
| 519 | locale_root = extension_dir / "_locales" |
| 520 | if not locale_root.is_dir(): |
| 521 | return {} |
| 522 | |
| 523 | preferred_locales = [ |
| 524 | default_locale, |
| 525 | default_locale.split("_", 1)[0] if default_locale else "", |
| 526 | "en_US", |
| 527 | "en", |
| 528 | ] |
| 529 | for locale in [item for item in preferred_locales if item]: |
| 530 | messages = _read_locale_file(locale_root / locale / "messages.json") |
| 531 | if messages: |
| 532 | return messages |
| 533 | |
| 534 | for messages_path in sorted(locale_root.glob("*/messages.json")): |
| 535 | messages = _read_locale_file(messages_path) |
| 536 | if messages: |
| 537 | return messages |
| 538 | return {} |
| 539 | |
| 540 | |
| 541 | def _read_locale_file(messages_path: Path) -> dict[str, Any]: |
| 542 | if not messages_path.is_file(): |
| 543 | return {} |
| 544 | try: |
| 545 | data = json.loads(messages_path.read_text(encoding="utf-8")) |
| 546 | except Exception: |
| 547 | return {} |
| 548 | return data if isinstance(data, dict) else {} |
| 549 | |
| 550 | |
| 551 | def _resolve_locale_message(messages: dict[str, Any], key: str) -> str | None: |
| 552 | entry = messages.get(key) |
| 553 | if not isinstance(entry, dict): |
| 554 | return None |
| 555 | message = str(entry.get("message") or "") |
| 556 | if not message: |
| 557 | return None |
| 558 | |
| 559 | placeholders = entry.get("placeholders") |
| 560 | if isinstance(placeholders, dict): |
| 561 | for name, placeholder in placeholders.items(): |
| 562 | if not isinstance(placeholder, dict): |
| 563 | continue |
| 564 | content = str(placeholder.get("content") or "") |
| 565 | if not content: |
| 566 | continue |
| 567 | message = re.sub( |
| 568 | rf"\${re.escape(str(name))}\$", |
| 569 | content, |
| 570 | message, |
| 571 | flags=re.IGNORECASE, |
| 572 | ) |
| 573 | return message |