pkgs/by-name/mu/music-assistant/update-providers.py

   1 #!/usr/bin/env nix-shell
   2 #!nix-shell -i python3 -p "python3.withPackages (ps: with ps; [ jinja2 mashumaro orjson aiofiles packaging ])" -p pyright ruff isort
   3 import asyncio
   4 import json
   5 import os.path
   6 import re
   7 import sys
   8 import tarfile
   9 import tempfile
  10 from dataclasses import dataclass, field
  11 from functools import cache
  12 from io import BytesIO
  13 from pathlib import Path
  14 from subprocess import check_output, run
  15 from typing import Dict, Final, List, Optional, Set, Union, cast
  16 from urllib.request import urlopen
  17
  18 from jinja2 import Environment
  19 from mashumaro.exceptions import MissingField
  20 from packaging.requirements import Requirement
  21
  22 TEMPLATE = """# Do not edit manually, run ./update-providers.py
  23
  24 {
  25   version = "{{ version }}";
  26   providers = {
  27 {%- for provider in providers | sort(attribute='domain') %}
  28     {{ provider.domain }} = {% if provider.available %}ps: with ps;{% else %}ps:{% endif %} [
  29 {%- for requirement in provider.available | sort %}
  30       {{ requirement }}
  31 {%- endfor %}
  32     ];{% if provider.missing %} # missing {{ ", ".join(provider.missing) }}{% endif %}
  33 {%- endfor %}
  34   };
  35 }
  36
  37 """
  38
  39
  40 ROOT: Final = (
  41     check_output(
  42         [
  43             "git",
  44             "rev-parse",
  45             "--show-toplevel",
  46         ]
  47     )
  48     .decode()
  49     .strip()
  50 )
  51
  52 PACKAGE_MAP = {
  53     "git+https://github.com/MarvinSchenkel/pytube.git": "pytube",
  54 }
  55
  56
  57 def run_sync(cmd: List[str]) -> None:
  58     print(f"$ {' '.join(cmd)}")
  59     process = run(cmd)
  60
  61     if process.returncode != 0:
  62         sys.exit(1)
  63
  64
  65 async def check_async(cmd: List[str]) -> str:
  66     print(f"$ {' '.join(cmd)}")
  67     process = await asyncio.create_subprocess_exec(
  68         *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
  69     )
  70     stdout, stderr = await process.communicate()
  71
  72     if process.returncode != 0:
  73         error = stderr.decode()
  74         raise RuntimeError(f"{cmd[0]} failed: {error}")
  75
  76     return stdout.decode().strip()
  77
  78
  79 class Nix:
  80     base_cmd: Final = [
  81         "nix",
  82         "--show-trace",
  83         "--extra-experimental-features",
  84         "nix-command",
  85     ]
  86
  87     @classmethod
  88     async def _run(cls, args: List[str]) -> Optional[str]:
  89         return await check_async(cls.base_cmd + args)
  90
  91     @classmethod
  92     async def eval(cls, expr: str) -> Union[List, Dict, int, float, str, bool]:
  93         response = await cls._run(["eval", "-f", f"{ROOT}/default.nix", "--json", expr])
  94         if response is None:
  95             raise RuntimeError("Nix eval expression returned no response")
  96         try:
  97             return json.loads(response)
  98         except (TypeError, ValueError):
  99             raise RuntimeError("Nix eval response could not be parsed from JSON")
 100
 101
 102 async def get_provider_manifests(version: str = "master") -> List:
 103     manifests = []
 104     with tempfile.TemporaryDirectory() as tmp:
 105         with urlopen(
 106             f"https://github.com/music-assistant/music-assistant/archive/refs/tags/{version}.tar.gz"
 107         ) as response:
 108             tarfile.open(fileobj=BytesIO(response.read())).extractall(
 109                 tmp, filter="data"
 110             )
 111
 112         basedir = Path(os.path.join(tmp, f"server-{version}"))
 113         sys.path.append(str(basedir))
 114         from music_assistant.common.models.provider import ProviderManifest  # type: ignore
 115
 116         for fn in basedir.glob("**/manifest.json"):
 117             try:
 118                 manifests.append(await ProviderManifest.parse(fn))
 119             except MissingField as ex:
 120                 print(f"Error parsing {fn}", ex)
 121
 122     return manifests
 123
 124
 125 @cache
 126 def packageset_attributes():
 127     output = check_output(
 128         [
 129             "nix-env",
 130             "-f",
 131             ROOT,
 132             "-qa",
 133             "-A",
 134             "music-assistant.python.pkgs",
 135             "--arg",
 136             "config",
 137             "{ allowAliases = false; }",
 138             "--json",
 139         ]
 140     )
 141     return json.loads(output)
 142
 143
 144 class TooManyMatches(Exception):
 145     pass
 146
 147
 148 class NoMatch(Exception):
 149     pass
 150
 151
 152 def resolve_package_attribute(package: str) -> str:
 153     pattern = re.compile(rf"^music-assistant\.python\.pkgs\.{package}$", re.I)
 154     packages = packageset_attributes()
 155     matches = []
 156     for attr in packages.keys():
 157         if pattern.match(attr):
 158             matches.append(attr.split(".")[-1])
 159
 160     if len(matches) > 1:
 161         raise TooManyMatches(
 162             f"Too many matching attributes for {package}: {' '.join(matches)}"
 163         )
 164     if not matches:
 165         raise NoMatch(f"No matching attribute for {package}")
 166
 167     return matches.pop()
 168
 169
 170 @dataclass
 171 class Provider:
 172     domain: str
 173     available: list[str] = field(default_factory=list)
 174     missing: list[str] = field(default_factory=list)
 175
 176     def __eq__(self, other):
 177         return self.domain == other.domain
 178
 179     def __hash__(self):
 180         return hash(self.domain)
 181
 182
 183 def resolve_providers(manifests) -> Set:
 184     providers = set()
 185     for manifest in manifests:
 186         provider = Provider(manifest.domain)
 187         for requirement in manifest.requirements:
 188             # allow substituting requirement specifications that packaging cannot parse
 189             if requirement in PACKAGE_MAP:
 190                 requirement = PACKAGE_MAP[requirement]
 191             requirement = Requirement(requirement)
 192             try:
 193                 provider.available.append(resolve_package_attribute(requirement.name))
 194             except TooManyMatches as ex:
 195                 print(ex, file=sys.stderr)
 196                 provider.missing.append(requirement.name)
 197             except NoMatch:
 198                 provider.missing.append(requirement.name)
 199         providers.add(provider)
 200     return providers
 201
 202
 203 def render(version: str, providers: Set):
 204     path = os.path.join(ROOT, "pkgs/by-name/mu/music-assistant/providers.nix")
 205     env = Environment()
 206     template = env.from_string(TEMPLATE)
 207     template.stream(version=version, providers=providers).dump(path)
 208
 209
 210 async def main():
 211     version: str = cast(str, await Nix.eval("music-assistant.version"))
 212     manifests = await get_provider_manifests(version)
 213     providers = resolve_providers(manifests)
 214     render(version, providers)
 215
 216
 217 if __name__ == "__main__":
 218     run_sync(["pyright", __file__])
 219     run_sync(["ruff", "check", "--ignore=E501", __file__])
 220     run_sync(["isort", __file__])
 221     run_sync(["ruff", "format", __file__])
 222     asyncio.run(main())