/
/
/
1"""Download the pinned cliairplay binary for local development."""
2
3from __future__ import annotations
4
5import hashlib
6import platform
7import re
8import stat
9import subprocess
10import sys
11from pathlib import Path
12from tempfile import TemporaryDirectory
13from urllib.error import URLError
14from urllib.request import Request, urlopen
15
16# ruff: noqa: T201
17
18RELEASE_BASE_URL = "https://github.com/music-assistant/airplay-cli/releases/download"
19BINARY_DIR = Path("music_assistant/providers/airplay/bin")
20
21
22def fetch_airplay_cli(root: Path) -> Path | None:
23 """
24 Install the container-pinned cliairplay binary when missing or outdated.
25
26 :param root: Music Assistant server repository root.
27 :return: Local binary path, or None when the current platform is unsupported.
28 """
29 system = platform.system()
30 machine = platform.machine()
31 asset_name = _asset_name(system, machine)
32 if asset_name is None:
33 print(
34 f"No cliairplay release is available for {system.lower()}/{machine.lower()}; skipping."
35 )
36 return None
37
38 destination = root / BINARY_DIR / asset_name
39 if destination.exists():
40 if not destination.is_file():
41 msg = f"cliairplay destination is not a file: {destination}"
42 raise RuntimeError(msg)
43
44 dockerfile = root / "Dockerfile"
45 version = _read_docker_arg(dockerfile, "CLIAIRPLAY_VERSION")
46 if destination.exists():
47 if not _binary_is_outdated(destination, version):
48 print(f"Using existing AirPlay development binary: {destination}")
49 return destination
50 print(f"Updating AirPlay development binary to {version}: {destination}")
51
52 manifest_digest = _read_docker_arg(dockerfile, "CLIAIRPLAY_CHECKSUMS_SHA256")
53 release_url = f"{RELEASE_BASE_URL}/{version}"
54
55 manifest = _download(f"{release_url}/SHA256SUMS")
56 actual_manifest_digest = hashlib.sha256(manifest).hexdigest()
57 if actual_manifest_digest != manifest_digest:
58 msg = (
59 f"SHA256SUMS digest mismatch for cliairplay {version}: "
60 f"expected {manifest_digest}, got {actual_manifest_digest}"
61 )
62 raise RuntimeError(msg)
63 binary_digest = _binary_digest(manifest, asset_name)
64
65 binary = _download(f"{release_url}/{asset_name}")
66 actual_binary_digest = hashlib.sha256(binary).hexdigest()
67 if actual_binary_digest != binary_digest:
68 msg = (
69 f"Checksum mismatch for {asset_name}: "
70 f"expected {binary_digest}, got {actual_binary_digest}"
71 )
72 raise RuntimeError(msg)
73
74 destination.parent.mkdir(parents=True, exist_ok=True)
75 with TemporaryDirectory(prefix=".cliairplay-", dir=destination.parent) as temp_dir:
76 temp_binary = Path(temp_dir) / asset_name
77 temp_binary.write_bytes(binary)
78 temp_binary.chmod(temp_binary.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
79 temp_binary.replace(destination)
80
81 print(f"Installed AirPlay development binary: {destination}")
82 return destination
83
84
85def main() -> int:
86 """Download the local AirPlay development binary."""
87 root = Path(__file__).resolve().parent.parent
88 try:
89 fetch_airplay_cli(root)
90 except (OSError, RuntimeError) as err:
91 print(f"ERROR: {err}", file=sys.stderr)
92 return 1
93 return 0
94
95
96def _asset_name(system: str, machine: str) -> str | None:
97 """Return the release asset name for a platform."""
98 normalized_system = system.lower().replace("darwin", "macos")
99 normalized_machine = machine.lower()
100
101 if normalized_machine in ("amd64", "x86_64"):
102 architecture = "x86_64"
103 elif normalized_machine in ("aarch64", "arm64"):
104 architecture = "arm64" if normalized_system == "macos" else "aarch64"
105 else:
106 return None
107 if normalized_system not in ("linux", "macos"):
108 return None
109 return f"cliairplay-{normalized_system}-{architecture}"
110
111
112def _read_docker_arg(dockerfile: Path, name: str) -> str:
113 """Read a pinned build argument from the Dockerfile."""
114 try:
115 content = dockerfile.read_text(encoding="utf-8")
116 except OSError as err:
117 msg = f"Unable to read {dockerfile}: {err}"
118 raise RuntimeError(msg) from err
119 match = re.search(rf"^ARG\s+{re.escape(name)}=(\S+)\s*$", content, flags=re.MULTILINE)
120 if match is None:
121 msg = f"Unable to find {name} in {dockerfile}"
122 raise RuntimeError(msg)
123 return match.group(1)
124
125
126def _binary_is_outdated(binary_path: Path, pinned_version: str) -> bool:
127 """
128 Return whether an existing release binary predates the Docker pin.
129
130 Unknown versions are treated as local development builds and preserved.
131
132 :param binary_path: Existing cliairplay executable.
133 :param pinned_version: Release version configured in the Dockerfile.
134 """
135 current = _read_binary_version(binary_path)
136 pinned = _parse_release_version(pinned_version)
137 return current is not None and pinned is not None and current < pinned
138
139
140def _read_binary_version(binary_path: Path) -> tuple[int, int, int] | None:
141 """
142 Return the semantic version reported by a cliairplay executable.
143
144 :param binary_path: cliairplay executable to inspect.
145 """
146 try:
147 result = subprocess.run( # noqa: S603
148 [binary_path, "--check"],
149 capture_output=True,
150 check=False,
151 text=True,
152 timeout=5,
153 )
154 except OSError, subprocess.TimeoutExpired:
155 return None
156 if result.returncode != 0:
157 return None
158 match = re.search(r"\bcliairplay\s+(v?\d+\.\d+\.\d+)\s+check\b", result.stdout)
159 if match is None:
160 return None
161 return _parse_release_version(match.group(1))
162
163
164def _parse_release_version(value: str) -> tuple[int, int, int] | None:
165 """
166 Parse a three-part cliairplay release version.
167
168 :param value: Version with an optional ``v`` prefix.
169 """
170 match = re.fullmatch(r"v?(\d+)\.(\d+)\.(\d+)", value.strip())
171 if match is None:
172 return None
173 return (int(match.group(1)), int(match.group(2)), int(match.group(3)))
174
175
176def _binary_digest(manifest: bytes, asset_name: str) -> str:
177 """Return the single checksum for an asset in SHA256SUMS."""
178 try:
179 lines = manifest.decode("utf-8").splitlines()
180 except UnicodeDecodeError as err:
181 msg = "Unable to decode cliairplay SHA256SUMS"
182 raise RuntimeError(msg) from err
183
184 matches = []
185 for line in lines:
186 fields = line.split()
187 if len(fields) == 2 and fields[1].lstrip("*") == asset_name:
188 matches.append(fields[0].lower())
189 if len(matches) != 1 or re.fullmatch(r"[0-9a-f]{64}", matches[0]) is None:
190 msg = f"SHA256SUMS must contain exactly one valid checksum for {asset_name}"
191 raise RuntimeError(msg)
192 return matches[0]
193
194
195def _download(url: str) -> bytes:
196 """Download a release file."""
197 request = Request( # noqa: S310
198 url,
199 headers={"User-Agent": "music-assistant-dev-setup"},
200 )
201 try:
202 with urlopen(request, timeout=60) as response: # noqa: S310
203 return response.read()
204 except URLError as err:
205 msg = f"Unable to download {url}: {err}"
206 raise RuntimeError(msg) from err
207
208
209if __name__ == "__main__":
210 sys.exit(main())
211