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