fetch assets from github release

ghrget.py

#!/usr/bin/env python3

import argparse
import logging
import os
import re
import requests
import shutil
import subprocess as sp
import tempfile

a = argparse.ArgumentParser(description='Fetch releases from GitHub')
a.add_argument('repo', metavar='REPOSITORY', help='github repository')
a.add_argument('--filter-regex', '-r', dest='regex', action='append', help='filter assets by regex')
a.add_argument('--dest', '-C', metavar='PATH', dest='dest', required=True, help='destination symlink pointing to unpacked assets')
a.add_argument('--clean', dest='clean', action='store_true', help='remove target that the symlink was pointing to')
a.add_argument('--unpacker', metavar='CMD', dest='unpacker', help='run "CMD <asset> <dest-name>" to unpack assets')
a.add_argument('--verbose', '-v', dest='verbose', action='count', default=0, help='show more messages')
a = a.parse_args()

logging.basicConfig(level=([logging.WARN, logging.INFO, logging.DEBUG][min(a.verbose, 2)]))

logger = logging.getLogger(__name__)

repo = a.repo.split('/')
if len(repo) != 2:
	raise ValueError(f'invalid repository {repo}')
repo = '/'.join(repo)

logger.debug('preparing filters')
filt = []
for fr in a.regex:
	fr = re.compile(fr)
	filt.append(lambda x: fr.search(x))

sess = requests.Session()
sess.headers['User-Agent'] = 'ghrfetch/0.1 ([email protected])'

logger.debug('finding latest release')
rel = sess.get(f'https://api.github.com/repos/{repo}/releases/latest', headers={
	'Accept': 'application/vnd.github.v3+json'
	}).json()

logger.debug('checking if local version is up to date')
if os.path.exists(a.dest):
	lastver = os.readlink(a.dest)[len(os.path.basename(a.dest)) + 1:]
	if lastver == rel['tag_name'].strip():
		logger.info('Package up to date')
		exit(0)

logger.debug('finding asset')
matchasset = None
for asset in rel['assets']:
	name = asset['name']
	match = True
	for f in filt:
		if not f(name):
			match = False
			break
	if match:
		matchasset = asset
		break

if not matchasset:
	logging.error('No assets matched all filters, failing')
	exit(1)

assetname = matchasset['name']
logger.info(f'Found matching asset {assetname}')

def unpack_asset(asset, dest, name):
	if a.unpacker:
		sp.run([a.unpacker, asset, dest, name], stdin=sp.DEVNULL, check=True)
		return

	if name.endswith('.xz'):
		with open(asset, 'rb') as rf, open(dest, 'wb') as wf:
			sp.run(['xz', '-cd'], stdin=rf, stdout=wf, check=True)
			os.fchmod(wf.fileno(), 0o755)
	else:
		raise RuntimeError('no suitable unpacker found')

with tempfile.NamedTemporaryFile(prefix=f'ghrget-{assetname}-') as f:
	logger.debug('downloading asset')
	for c in sess.get(matchasset['url'], headers={ 'Accept': 'application/octet-stream' }, stream=True).iter_content(chunk_size=(1024**2)*16):
		logger.debug(f'got chunk of {len(c)} bytes')
		f.write(c)
	f.flush()

	logger.debug('unpacking asset')
	realdest = f'{a.dest}-{rel["tag_name"].strip()}'
	if os.path.isdir(realdest):
		shutil.rmtree(realdest)
	elif os.path.exists(realdest):
		os.unlink(realdest)
	unpack_asset(f.name, realdest, assetname)

	logger.debug('updating symlink')
	cleanpath = None
	if a.clean and os.path.islink(a.dest):
		cleanpath = os.path.join(os.path.dirname(a.dest), os.readlink(a.dest))
	with tempfile.TemporaryDirectory(dir=os.path.dirname(a.dest), prefix=os.path.basename(a.dest) + '-') as td:
		p = os.path.join(td, 'newlink')
		os.symlink(os.path.basename(realdest), p)
		os.rename(p, a.dest)

	if cleanpath:
		logger.debug('cleaning old path')
		if os.path.isdir(cleanpath):
			shutil.rmtree(cleanpath)
		else:
			os.unlink(cleanpath)