diff --git a/src/python/jw/build/lib/util.py b/src/python/jw/build/lib/util.py new file mode 100644 index 00000000..df30d9e7 --- /dev/null +++ b/src/python/jw/build/lib/util.py @@ -0,0 +1,89 @@ +# -*- coding: utf-8 -*- + +import os, sys, subprocess, json, time +from argparse import Namespace +from urllib.parse import urlparse + +def run_cmd(cmd: list[str], wd=None, throw=True, verbose=False) -> str|None: # export + + tokens = [cmd[0]] + for token in cmd[1:]: + if token.find(' ') != -1: + token = '"' + token + '"' + tokens.append(token) + subject = ' '.join(tokens) + if wd is not None: + subject += f' in {wd}' + if verbose: + delim_len = 120 + delim = f'---- running {subject} -' + delim = delim + '-' * (delim_len - len(delim)) + print(',' + delim + ' >') + + cwd: str|None = None + if wd is not None: + cwd = os.getcwd() + os.chdir(wd) + + ret = '' + try: + p = subprocess.Popen(cmd, shell=False, stdout=subprocess.PIPE, stderr=None, close_fds=True) + for line in iter(p.stdout.readline, b''): + line = line.decode(sys.stdout.encoding) + ret += line + p.wait() + if verbose: + print('`' + delim + ' <') + if p.returncode: + if verbose: + print(' '.join(cmd) + ' failed') + raise Exception(time.strftime('%Y-%m-%d %H:%M') + f': failed to run "{subject}"') + finally: + if cwd: + os.chdir(cwd) + return ret + +def run_curl(args: list[str], parse_json: bool=True, wd=None, throw=None, verbose=False) -> dict|str: # export + cmd = ['curl'] + if not verbose: + cmd.append('-s') + cmd.extend(args) + ret = run_cmd(cmd, wd=wd, throw=throw, verbose=verbose) + if parse_json: + return json.loads(ret) + return ret + +def get_username(args: Namespace|None=None, url: str|None=None) -> str: # export + url_user = None if url is None else urlparse(url).username + if args is not None: + if args.username is not None: + if url_user is not None and url_user != args.username: + raise Exception(f'Username mismatch: called with --username="{args.username}", URL has user name "{url_user}"') + return args.username + if url is not None: + return url_user + raise Exception(f'Neither URL nor command-line arguments available, can\'t get user name') + +def get_password(args: Namespace|None=None, url: str|None=None, askpass_env: list[str]=[]) -> str: # export + if args is None and url is None and not askpass_env: + raise Exception(f'Neither URL nor command-line arguments nor askpass environment variable available, can\'t get password') + if args is not None and hasattr(args, 'password'): # use getattr(), because we don't necessarily want to have insecure --password among options + ret = getattr(args, 'password') + if ret is not None: + return ret + if url is not None: + parsed = urlparse(url) + if parsed.password is not None: + return parsed.password + exes = [] + if args is not None: + exes.append(getattr(args, 'askpass')) + for var in askpass_env: + exes.append(os.getenv(var)) + for exe in exes: + if exe is None: + continue + ret = run_cmd(exe, throw=False) + if ret is not None: + return ret + return None