|
| 1 | +#!/usr/bin/env python3 |
| 2 | +# /// script |
| 3 | +# requires-python = ">=3.11" |
| 4 | +# dependencies = [ |
| 5 | +# "packaging", |
| 6 | +# "ruamel.yaml==0.18.6", |
| 7 | +# "urllib3", |
| 8 | +# ] |
| 9 | +# /// |
| 10 | + |
| 11 | +import argparse |
| 12 | +import dataclasses |
| 13 | +import json |
| 14 | +import enum |
| 15 | +import re |
| 16 | +from pathlib import Path |
| 17 | +from typing import Any, NoReturn, Self |
| 18 | + |
| 19 | +import urllib3 |
| 20 | +from packaging.specifiers import SpecifierSet |
| 21 | +from ruamel.yaml import YAML |
| 22 | + |
| 23 | + |
| 24 | +http = urllib3.PoolManager() |
| 25 | +yaml = YAML(typ="safe") |
| 26 | +_LAVALINK_VERSION_PATTERN = re.compile( |
| 27 | + rb""" |
| 28 | + ^ |
| 29 | + (?P<version> |
| 30 | + (?P<major>0|[1-9]\d*)\.(?P<minor>0|[1-9]\d*)\.(?P<patch>0|[1-9]\d*) |
| 31 | + (?:-rc\.(?P<rc>0|[1-9]\d*))? |
| 32 | + # additional build metadata, can be used by our downstream Lavalink |
| 33 | + # if we need to alter an upstream release |
| 34 | + (?:\+red\.(?P<red>[1-9]\d*))? |
| 35 | + ) |
| 36 | + $ |
| 37 | + """, |
| 38 | + re.MULTILINE | re.VERBOSE, |
| 39 | +) |
| 40 | +_RELEASES_YAML = Path(__file__).absolute().parent / "releases.yaml" |
| 41 | +_RED_JARS_REPO = "https://github.com/Cog-Creators/Lavalink-Jars" |
| 42 | +_DEFAULT_PLUGIN_REPOSITORY = "https://maven.lavalink.dev/releases" |
| 43 | + |
| 44 | + |
| 45 | +@dataclasses.dataclass() |
| 46 | +class Plugin: |
| 47 | + group: str |
| 48 | + name: str |
| 49 | + version: str |
| 50 | + repository: str |
| 51 | + |
| 52 | + @property |
| 53 | + def url(self) -> str: |
| 54 | + return ( |
| 55 | + f"{self.repository}/{self.group.replace('.', '/')}" |
| 56 | + f"/{self.name}/{self.version}/{self.name}-{self.version}.jar" |
| 57 | + ) |
| 58 | + |
| 59 | + |
| 60 | +def _raise_type_error(release_name: str, msg: str) -> NoReturn: |
| 61 | + raise TypeError(f"For {release_name!r} release: {msg}") |
| 62 | + |
| 63 | + |
| 64 | +class ReleaseStream(enum.Enum): |
| 65 | + STABLE = "stable" |
| 66 | + PREVIEW = "preview" |
| 67 | + |
| 68 | + |
| 69 | +@dataclasses.dataclass() |
| 70 | +class ReleaseInfo: |
| 71 | + release_name: str |
| 72 | + jar_version: str |
| 73 | + jar_url: str |
| 74 | + yt_plugin: Plugin |
| 75 | + java_versions: tuple[int, ...] |
| 76 | + release_stream: ReleaseStream |
| 77 | + # inclusive |
| 78 | + min_red_version: str |
| 79 | + # exclusive |
| 80 | + max_red_version: str = "" |
| 81 | + application_yml_overrides: dict[str, Any] = dataclasses.field(default_factory=dict) |
| 82 | + |
| 83 | + @classmethod |
| 84 | + def parse(cls, release_name: Any, data: Any) -> Self: |
| 85 | + if not isinstance(release_name, str): |
| 86 | + raise TypeError(f"expected release name to be a string, got {release_name!r} instead") |
| 87 | + if not isinstance(data, dict): |
| 88 | + _raise_type_error(release_name, "expected release info to be a dictionary") |
| 89 | + |
| 90 | + jar_version = cls._parse_jar_version(release_name, data) |
| 91 | + jar_url = cls._get_jar_url(release_name, jar_version) |
| 92 | + yt_plugin = cls._parse_yt_plugin(release_name, data) |
| 93 | + java_versions = cls._parse_java_versions(release_name, data) |
| 94 | + min_red_version = cls._parse_min_red_version(release_name, data) |
| 95 | + release_stream = cls._parse_release_stream(release_name, data) |
| 96 | + application_yml_overrides = cls._parse_application_yml_overrides(release_name, data) |
| 97 | + |
| 98 | + return cls( |
| 99 | + release_name=release_name, |
| 100 | + jar_version=jar_version, |
| 101 | + jar_url=jar_url, |
| 102 | + yt_plugin=yt_plugin, |
| 103 | + java_versions=java_versions, |
| 104 | + release_stream=release_stream, |
| 105 | + min_red_version=min_red_version, |
| 106 | + application_yml_overrides=application_yml_overrides, |
| 107 | + ) |
| 108 | + |
| 109 | + @property |
| 110 | + def red_version(self) -> SpecifierSet: |
| 111 | + specifiers = SpecifierSet(f">={self.min_red_version}") |
| 112 | + if self.max_red_version: |
| 113 | + specifiers &= SpecifierSet(f"<{self.max_red_version}") |
| 114 | + return specifiers |
| 115 | + |
| 116 | + def as_json_dict(self) -> dict[str, Any]: |
| 117 | + return { |
| 118 | + "release_name": self.release_name, |
| 119 | + "jar_version": self.jar_version, |
| 120 | + "jar_url": self.jar_url, |
| 121 | + "yt_plugin_version": self.yt_plugin.version, |
| 122 | + "java_versions": list(self.java_versions), |
| 123 | + "red_version": str(self.red_version), |
| 124 | + "release_stream": self.release_stream.value, |
| 125 | + "application_yml_overrides": self.application_yml_overrides, |
| 126 | + } |
| 127 | + |
| 128 | + @staticmethod |
| 129 | + def _parse_jar_version(release_name: str, data: dict[Any, Any]) -> str: |
| 130 | + try: |
| 131 | + jar_version = data["jar_version"] |
| 132 | + except KeyError: |
| 133 | + _raise_type_error(release_name, "expected jar_version to be set") |
| 134 | + if not isinstance(jar_version, str): |
| 135 | + _raise_type_error(release_name, "expected jar_version to be a string") |
| 136 | + return jar_version |
| 137 | + |
| 138 | + @staticmethod |
| 139 | + def _get_jar_url(release_name: str, jar_version: str) -> str: |
| 140 | + jar_url = f"{_RED_JARS_REPO}/releases/download/{jar_version}/Lavalink.jar" |
| 141 | + resp = http.request("HEAD", jar_url) |
| 142 | + if resp.status >= 400: |
| 143 | + _raise_type_error( |
| 144 | + release_name, f"expected Lavalink.jar to be available at: {jar_url}" |
| 145 | + ) |
| 146 | + return jar_url |
| 147 | + |
| 148 | + @staticmethod |
| 149 | + def _parse_yt_plugin(release_name: str, data: dict[Any, Any]) -> Plugin: |
| 150 | + try: |
| 151 | + yt_plugin_version = data["yt_plugin_version"] |
| 152 | + except KeyError: |
| 153 | + _raise_type_error(release_name, "expected yt_plugin_version to be set") |
| 154 | + if not isinstance(yt_plugin_version, str): |
| 155 | + _raise_type_error(release_name, "expected yt_plugin_version to be a string") |
| 156 | + yt_plugin = Plugin( |
| 157 | + group="dev.lavalink.youtube", |
| 158 | + name="youtube-plugin", |
| 159 | + version=yt_plugin_version, |
| 160 | + repository=_DEFAULT_PLUGIN_REPOSITORY, |
| 161 | + ) |
| 162 | + resp = http.request("HEAD", yt_plugin.url) |
| 163 | + if resp.status >= 400: |
| 164 | + _raise_type_error( |
| 165 | + release_name, f"expected YT plugin to be available at: {yt_plugin.url}" |
| 166 | + ) |
| 167 | + return yt_plugin |
| 168 | + |
| 169 | + @staticmethod |
| 170 | + def _parse_java_versions(release_name: str, data: dict[Any, Any]) -> tuple[int, ...]: |
| 171 | + try: |
| 172 | + java_versions = data["java_versions"] |
| 173 | + except KeyError: |
| 174 | + _raise_type_error(release_name, "expected java_versions to be set") |
| 175 | + if not ( |
| 176 | + isinstance(java_versions, list) |
| 177 | + and all(isinstance(x, int) for x in java_versions) |
| 178 | + ): |
| 179 | + _raise_type_error( |
| 180 | + release_name, "expected java_versions to be a list of version numbers (integers)" |
| 181 | + ) |
| 182 | + return tuple(java_versions) |
| 183 | + |
| 184 | + @staticmethod |
| 185 | + def _parse_min_red_version(release_name: str, data: dict[Any, Any]) -> str: |
| 186 | + try: |
| 187 | + min_red_version = data["min_red_version"] |
| 188 | + except KeyError: |
| 189 | + _raise_type_error(release_name, "expected min_red_version to be set") |
| 190 | + if not isinstance(min_red_version, str): |
| 191 | + _raise_type_error(release_name, "expected min_red_version to be a string") |
| 192 | + return min_red_version |
| 193 | + |
| 194 | + @staticmethod |
| 195 | + def _parse_release_stream(release_name: str, data: dict[Any, Any]) -> ReleaseStream: |
| 196 | + try: |
| 197 | + raw_release_stream = data["release_stream"] |
| 198 | + except KeyError: |
| 199 | + _raise_type_error(release_name, "expected release_stream to be set") |
| 200 | + if not isinstance(raw_release_stream, str): |
| 201 | + _raise_type_error(release_name, "expected release_stream to be a string") |
| 202 | + try: |
| 203 | + release_stream = ReleaseStream(raw_release_stream) |
| 204 | + except ValueError: |
| 205 | + _raise_type_error( |
| 206 | + release_name, |
| 207 | + ( |
| 208 | + "expected release_stream to be one of: " |
| 209 | + + ", ".join(member.value for member in ReleaseStream) |
| 210 | + ), |
| 211 | + ) |
| 212 | + return release_stream |
| 213 | + |
| 214 | + @staticmethod |
| 215 | + def _parse_application_yml_overrides( |
| 216 | + release_name: str, data: dict[Any, Any] |
| 217 | + ) -> dict[str, Any]: |
| 218 | + overrides = data.get("application_yml_overrides", {}) |
| 219 | + if not isinstance(overrides, dict): |
| 220 | + _raise_type_error( |
| 221 | + release_name, "expected application_yml_overrides to be a dictionary" |
| 222 | + ) |
| 223 | + return overrides |
| 224 | + |
| 225 | + |
| 226 | +def parse_releases() -> list[ReleaseInfo]: |
| 227 | + with open(_RELEASES_YAML, encoding="utf-8") as fp: |
| 228 | + data = yaml.load(fp) |
| 229 | + |
| 230 | + if not isinstance(data, dict): |
| 231 | + raise TypeError("expected top-level object in the YAML file to be a dictionary") |
| 232 | + try: |
| 233 | + releases = data["releases"] |
| 234 | + except KeyError: |
| 235 | + raise TypeError("expected releases to be set") |
| 236 | + |
| 237 | + errors = [] |
| 238 | + parsed_releases = [] |
| 239 | + previous_release: ReleaseInfo | None = None |
| 240 | + |
| 241 | + for name, release_data in releases.items(): |
| 242 | + print(f"Processing {name!r}...") |
| 243 | + try: |
| 244 | + info = ReleaseInfo.parse(name, release_data) |
| 245 | + except ValueError as exc: |
| 246 | + errors.append(str(exc)) |
| 247 | + continue |
| 248 | + if previous_release is not None: |
| 249 | + if previous_release.min_red_version != info.min_red_version: |
| 250 | + info.max_red_version = previous_release.min_red_version |
| 251 | + else: |
| 252 | + info.max_red_version = previous_release.max_red_version |
| 253 | + parsed_releases.append(info) |
| 254 | + previous_release = info |
| 255 | + |
| 256 | + if errors: |
| 257 | + raise ValueError("\n".join(f"- {err}" for err in errors)) |
| 258 | + |
| 259 | + return parsed_releases |
| 260 | + |
| 261 | + |
| 262 | +def generate_index(releases: list[ReleaseInfo], output_dir: Path) -> None: |
| 263 | + output = [release.as_json_dict() for release in releases] |
| 264 | + output_dir.mkdir(parents=True, exist_ok=True) |
| 265 | + with open(output_dir / "index.0.json", "w", encoding="utf-8") as fp: |
| 266 | + json.dump(output, fp, indent=4) |
| 267 | + with open(output_dir / "index.0-min.json", "w", encoding="utf-8") as fp: |
| 268 | + json.dump(output, fp, separators=(",", ":")) |
| 269 | + |
| 270 | + |
| 271 | +def generate_index_cmd(args: argparse.Namespace) -> None: |
| 272 | + releases = parse_releases() |
| 273 | + generate_index(releases, Path(args.output_dir)) |
| 274 | + |
| 275 | + |
| 276 | +def main() -> None: |
| 277 | + parser = argparse.ArgumentParser() |
| 278 | + subparsers = parser.add_subparsers(title="available commands") |
| 279 | + |
| 280 | + generate_index = subparsers.add_parser("generate-index") |
| 281 | + generate_index.add_argument("output_dir", help="The directory to output the index files to.") |
| 282 | + generate_index.set_defaults(func=generate_index_cmd) |
| 283 | + |
| 284 | + args = parser.parse_args() |
| 285 | + args.func(args) |
| 286 | + |
| 287 | + |
| 288 | +if __name__ == "__main__": |
| 289 | + main() |
0 commit comments