pyrox.dev
1#!/usr/bin/env nix-shell
2#!nix-shell update-octave-shell.nix -i python3
3
4"""
5Update a Octave package expression by passing in the `.nix` file, or the directory containing it.
6You can pass in multiple files or paths.
7
8You'll likely want to use
9``
10 $ ./update-octave-libraries ../../pkgs/development/octave-modules/**/default.nix
11``
12to update all non-pinned libraries in that folder.
13"""
14
15import argparse
16import os
17import pathlib
18import re
19import requests
20import yaml
21from concurrent.futures import ThreadPoolExecutor as Pool
22from packaging.version import Version as _Version
23from packaging.version import InvalidVersion
24from packaging.specifiers import SpecifierSet
25import collections
26import subprocess
27import tempfile
28
29INDEX = "https://raw.githubusercontent.com/gnu-octave/packages/main/packages"
30"""url of Octave packages' source on GitHub"""
31
32EXTENSIONS = ['tar.gz', 'tar.bz2', 'tar', 'zip']
33"""Permitted file extensions. These are evaluated from left to right and the first occurance is returned."""
34
35PRERELEASES = False
36
37GIT = "git"
38
39NIXPGKS_ROOT = subprocess.check_output(["git", "rev-parse", "--show-toplevel"]).decode('utf-8').strip()
40
41import logging
42logging.basicConfig(level=logging.INFO)
43
44
45class Version(_Version, collections.abc.Sequence):
46
47 def __init__(self, version):
48 super().__init__(version)
49 # We cannot use `str(Version(0.04.21))` because that becomes `0.4.21`
50 # https://github.com/avian2/unidecode/issues/13#issuecomment-354538882
51 self.raw_version = version
52
53 def __getitem__(self, i):
54 return self._version.release[i]
55
56 def __len__(self):
57 return len(self._version.release)
58
59 def __iter__(self):
60 yield from self._version.release
61
62
63def _get_values(attribute, text):
64 """Match attribute in text and return all matches.
65
66 :returns: List of matches.
67 """
68 regex = '{}\s+=\s+"(.*)";'.format(attribute)
69 regex = re.compile(regex)
70 values = regex.findall(text)
71 return values
72
73def _get_unique_value(attribute, text):
74 """Match attribute in text and return unique match.
75
76 :returns: Single match.
77 """
78 values = _get_values(attribute, text)
79 n = len(values)
80 if n > 1:
81 raise ValueError("found too many values for {}".format(attribute))
82 elif n == 1:
83 return values[0]
84 else:
85 raise ValueError("no value found for {}".format(attribute))
86
87def _get_line_and_value(attribute, text):
88 """Match attribute in text. Return the line and the value of the attribute."""
89 regex = '({}\s+=\s+"(.*)";)'.format(attribute)
90 regex = re.compile(regex)
91 value = regex.findall(text)
92 n = len(value)
93 if n > 1:
94 raise ValueError("found too many values for {}".format(attribute))
95 elif n == 1:
96 return value[0]
97 else:
98 raise ValueError("no value found for {}".format(attribute))
99
100
101def _replace_value(attribute, value, text):
102 """Search and replace value of attribute in text."""
103 old_line, old_value = _get_line_and_value(attribute, text)
104 new_line = old_line.replace(old_value, value)
105 new_text = text.replace(old_line, new_line)
106 return new_text
107
108
109def _fetch_page(url):
110 r = requests.get(url)
111 if r.status_code == requests.codes.ok:
112 return list(yaml.safe_load_all(r.content))[0]
113 else:
114 raise ValueError("request for {} failed".format(url))
115
116
117def _fetch_github(url):
118 headers = {}
119 token = os.environ.get('GITHUB_API_TOKEN')
120 if token:
121 headers["Authorization"] = f"token {token}"
122 r = requests.get(url, headers=headers)
123
124 if r.status_code == requests.codes.ok:
125 return r.json()
126 else:
127 raise ValueError("request for {} failed".format(url))
128
129
130SEMVER = {
131 'major' : 0,
132 'minor' : 1,
133 'patch' : 2,
134}
135
136
137def _determine_latest_version(current_version, target, versions):
138 """Determine latest version, given `target`, returning the more recent version.
139 """
140 current_version = Version(current_version)
141
142 def _parse_versions(versions):
143 for v in versions:
144 try:
145 yield Version(v)
146 except InvalidVersion:
147 pass
148
149 versions = _parse_versions(versions)
150
151 index = SEMVER[target]
152
153 ceiling = list(current_version[0:index])
154 if len(ceiling) == 0:
155 ceiling = None
156 else:
157 ceiling[-1]+=1
158 ceiling = Version(".".join(map(str, ceiling)))
159
160 # We do not want prereleases
161 versions = SpecifierSet(prereleases=PRERELEASES).filter(versions)
162
163 if ceiling is not None:
164 versions = SpecifierSet(f"<{ceiling}").filter(versions)
165
166 return (max(sorted(versions))).raw_version
167
168
169def _get_latest_version_octave_packages(package, extension, current_version, target):
170 """Get latest version and hash from Octave Packages."""
171 url = "{}/{}.yaml".format(INDEX, package)
172 yaml = _fetch_page(url)
173
174 versions = list(map(lambda pv: pv['id'], yaml['versions']))
175 version = _determine_latest_version(current_version, target, versions)
176
177 try:
178 releases = [v if v['id'] == version else None for v in yaml['versions']]
179 except KeyError as e:
180 raise KeyError('Could not find version {} for {}'.format(version, package)) from e
181 for release in releases:
182 if release['url'].endswith(extension):
183 sha256 = release['sha256']
184 break
185 else:
186 sha256 = None
187 return version, sha256, None
188
189
190def _get_latest_version_github(package, extension, current_version, target):
191 def strip_prefix(tag):
192 return re.sub("^[^0-9]*", "", tag)
193
194 def get_prefix(string):
195 matches = re.findall(r"^([^0-9]*)", string)
196 return next(iter(matches), "")
197
198 # when invoked as an updateScript, UPDATE_NIX_ATTR_PATH will be set
199 # this allows us to work with packages which live outside of octave-modules
200 attr_path = os.environ.get("UPDATE_NIX_ATTR_PATH", f"octavePackages.{package}")
201 try:
202 homepage = subprocess.check_output(
203 ["nix", "eval", "-f", f"{NIXPGKS_ROOT}/default.nix", "--raw", f"{attr_path}.src.meta.homepage"])\
204 .decode('utf-8')
205 except Exception as e:
206 raise ValueError(f"Unable to determine homepage: {e}")
207 owner_repo = homepage[len("https://github.com/"):] # remove prefix
208 owner, repo = owner_repo.split("/")
209
210 url = f"https://api.github.com/repos/{owner}/{repo}/releases"
211 all_releases = _fetch_github(url)
212 releases = list(filter(lambda x: not x['prerelease'], all_releases))
213
214 if len(releases) == 0:
215 raise ValueError(f"{homepage} does not contain any stable releases")
216
217 versions = map(lambda x: strip_prefix(x['tag_name']), releases)
218 version = _determine_latest_version(current_version, target, versions)
219
220 release = next(filter(lambda x: strip_prefix(x['tag_name']) == version, releases))
221 prefix = get_prefix(release['tag_name'])
222 try:
223 sha256 = subprocess.check_output(["nix-prefetch-url", "--type", "sha256", "--unpack", f"{release['tarball_url']}"], stderr=subprocess.DEVNULL)\
224 .decode('utf-8').strip()
225 except:
226 # this may fail if they have both a branch and a tag of the same name, attempt tag name
227 tag_url = str(release['tarball_url']).replace("tarball","tarball/refs/tags")
228 sha256 = subprocess.check_output(["nix-prefetch-url", "--type", "sha256", "--unpack", tag_url], stderr=subprocess.DEVNULL)\
229 .decode('utf-8').strip()
230
231
232 return version, sha256, prefix
233
234def _get_latest_version_git(package, extension, current_version, target):
235 """NOTE: Unimplemented!"""
236 # attr_path = os.environ.get("UPDATE_NIX_ATTR_PATH", f"octavePackages.{package}")
237 # try:
238 # download_url = subprocess.check_output(
239 # ["nix", "--extra-experimental-features", "nix-command", "eval", "-f", f"{NIXPGKS_ROOT}/default.nix", "--raw", f"{attr_path}.src.url"])\
240 # .decode('utf-8')
241 # except Exception as e:
242 # raise ValueError(f"Unable to determine download link: {e}")
243
244 # with tempfile.TemporaryDirectory(prefix=attr_path) as new_clone_location:
245 # subprocess.run(["git", "clone", download_url, new_clone_location])
246 # newest_commit = subprocess.check_output(
247 # ["git" "rev-parse" "$(git branch -r)" "|" "tail" "-n" "1"]).decode('utf-8')
248 pass
249
250
251FETCHERS = {
252 'fetchFromGitHub' : _get_latest_version_github,
253 'fetchurl' : _get_latest_version_octave_packages,
254 'fetchgit' : _get_latest_version_git,
255}
256
257
258DEFAULT_SETUPTOOLS_EXTENSION = 'tar.gz'
259
260
261FORMATS = {
262 'setuptools' : DEFAULT_SETUPTOOLS_EXTENSION,
263}
264
265def _determine_fetcher(text):
266 # Count occurrences of fetchers.
267 nfetchers = sum(text.count('src = {}'.format(fetcher)) for fetcher in FETCHERS.keys())
268 if nfetchers == 0:
269 raise ValueError("no fetcher.")
270 elif nfetchers > 1:
271 raise ValueError("multiple fetchers.")
272 else:
273 # Then we check which fetcher to use.
274 for fetcher in FETCHERS.keys():
275 if 'src = {}'.format(fetcher) in text:
276 return fetcher
277
278
279def _determine_extension(text, fetcher):
280 """Determine what extension is used in the expression.
281
282 If we use:
283 - fetchPypi, we check if format is specified.
284 - fetchurl, we determine the extension from the url.
285 - fetchFromGitHub we simply use `.tar.gz`.
286 """
287 if fetcher == 'fetchurl':
288 url = _get_unique_value('url', text)
289 extension = os.path.splitext(url)[1]
290
291 elif fetcher == 'fetchFromGitHub' or fetcher == 'fetchgit':
292 if "fetchSubmodules" in text:
293 raise ValueError("fetchFromGitHub fetcher doesn't support submodules")
294 extension = "tar.gz"
295
296 return extension
297
298
299def _update_package(path, target):
300
301 # Read the expression
302 with open(path, 'r') as f:
303 text = f.read()
304
305 # Determine pname. Many files have more than one pname
306 pnames = _get_values('pname', text)
307
308 # Determine version.
309 version = _get_unique_value('version', text)
310
311 # First we check how many fetchers are mentioned.
312 fetcher = _determine_fetcher(text)
313
314 extension = _determine_extension(text, fetcher)
315
316 # Attempt a fetch using each pname, e.g. backports-zoneinfo vs backports.zoneinfo
317 successful_fetch = False
318 for pname in pnames:
319 if fetcher == "fetchgit":
320 logging.warning(f"You must update {pname} MANUALLY!")
321 return { 'path': path, 'target': target, 'pname': pname,
322 'old_version': version, 'new_version': version }
323 try:
324 new_version, new_sha256, prefix = FETCHERS[fetcher](pname, extension, version, target)
325 successful_fetch = True
326 break
327 except ValueError:
328 continue
329
330 if not successful_fetch:
331 raise ValueError(f"Unable to find correct package using these pnames: {pnames}")
332
333 if new_version == version:
334 logging.info("Path {}: no update available for {}.".format(path, pname))
335 return False
336 elif Version(new_version) <= Version(version):
337 raise ValueError("downgrade for {}.".format(pname))
338 if not new_sha256:
339 raise ValueError("no file available for {}.".format(pname))
340
341 text = _replace_value('version', new_version, text)
342 # hashes from pypi are 16-bit encoded sha256's, normalize it to sri to avoid merge conflicts
343 # sri hashes have been the default format since nix 2.4+
344 sri_hash = subprocess.check_output(["nix", "--extra-experimental-features", "nix-command", "hash", "to-sri", "--type", "sha256", new_sha256]).decode('utf-8').strip()
345
346
347 # fetchers can specify a sha256, or a sri hash
348 try:
349 text = _replace_value('sha256', sri_hash, text)
350 except ValueError:
351 text = _replace_value('hash', sri_hash, text)
352
353 if fetcher == 'fetchFromGitHub':
354 # in the case of fetchFromGitHub, it's common to see `rev = version;` or `rev = "v${version}";`
355 # in which no string value is meant to be substituted. However, we can just overwrite the previous value.
356 regex = '(rev\s+=\s+[^;]*;)'
357 regex = re.compile(regex)
358 matches = regex.findall(text)
359 n = len(matches)
360
361 if n == 0:
362 raise ValueError("Unable to find rev value for {}.".format(pname))
363 else:
364 # forcefully rewrite rev, incase tagging conventions changed for a release
365 match = matches[0]
366 text = text.replace(match, f'rev = "refs/tags/{prefix}${{version}}";')
367 # incase there's no prefix, just rewrite without interpolation
368 text = text.replace('"${version}";', 'version;')
369
370 with open(path, 'w') as f:
371 f.write(text)
372
373 logging.info("Path {}: updated {} from {} to {}".format(path, pname, version, new_version))
374
375 result = {
376 'path' : path,
377 'target': target,
378 'pname': pname,
379 'old_version' : version,
380 'new_version' : new_version,
381 #'fetcher' : fetcher,
382 }
383
384 return result
385
386
387def _update(path, target):
388
389 # We need to read and modify a Nix expression.
390 if os.path.isdir(path):
391 path = os.path.join(path, 'default.nix')
392
393 # If a default.nix does not exist, we quit.
394 if not os.path.isfile(path):
395 logging.info("Path {}: does not exist.".format(path))
396 return False
397
398 # If file is not a Nix expression, we quit.
399 if not path.endswith(".nix"):
400 logging.info("Path {}: does not end with `.nix`.".format(path))
401 return False
402
403 try:
404 return _update_package(path, target)
405 except ValueError as e:
406 logging.warning("Path {}: {}".format(path, e))
407 return False
408
409
410def _commit(path, pname, old_version, new_version, pkgs_prefix="octave: ", **kwargs):
411 """Commit result.
412 """
413
414 msg = f'{pkgs_prefix}{pname}: {old_version} -> {new_version}'
415
416 try:
417 subprocess.check_call([GIT, 'add', path])
418 subprocess.check_call([GIT, 'commit', '-m', msg])
419 except subprocess.CalledProcessError as e:
420 subprocess.check_call([GIT, 'checkout', path])
421 raise subprocess.CalledProcessError(f'Could not commit {path}') from e
422
423 return True
424
425
426def main():
427
428 epilog = """
429environment variables:
430 GITHUB_API_TOKEN\tGitHub API token used when updating github packages
431 """
432 parser = argparse.ArgumentParser(formatter_class=argparse.RawDescriptionHelpFormatter, epilog=epilog)
433 parser.add_argument('package', type=str, nargs='+')
434 parser.add_argument('--target', type=str, choices=SEMVER.keys(), default='major')
435 parser.add_argument('--commit', action='store_true', help='Create a commit for each package update')
436 parser.add_argument('--use-pkgs-prefix', action='store_true', help='Use octavePackages.${pname}: instead of octave: ${pname}: when making commits')
437
438 args = parser.parse_args()
439 target = args.target
440
441 packages = list(map(os.path.abspath, args.package))
442
443 logging.info("Updating packages...")
444
445 # Use threads to update packages concurrently
446 with Pool() as p:
447 results = list(filter(bool, p.map(lambda pkg: _update(pkg, target), packages)))
448
449 logging.info("Finished updating packages.")
450
451 commit_options = {}
452 if args.use_pkgs_prefix:
453 logging.info("Using octavePackages. prefix for commits")
454 commit_options["pkgs_prefix"] = "octavePackages."
455
456 # Commits are created sequentially.
457 if args.commit:
458 logging.info("Committing updates...")
459 # list forces evaluation
460 list(map(lambda x: _commit(**x, **commit_options), results))
461 logging.info("Finished committing updates")
462
463 count = len(results)
464 logging.info("{} package(s) updated".format(count))
465
466
467if __name__ == '__main__':
468 main()