# -*- 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