libinstall.py

import os
import re
import shutil
import subprocess
import sys
import tempfile

def run_silent(p):
  proc = subprocess.Popen(p, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
  out, err = proc.communicate()
  out, err = out.decode('utf8'), err.decode('utf8')
  return (proc.returncode == 0, out, err)

def run_verbose(p):
  return subprocess.run(p) == 0

class FileInstaller(object):
  @staticmethod
  def confirm_executable(program):
    if not FileInstaller.has_executable(program):
      raise RuntimeError('%s not installed, cannot proceed' % program)

  @staticmethod
  def has_executable(program):
    return shutil.which(program) is not None

  @staticmethod
  def check_exists(path):
    if os.path.exists(path):
      raise RuntimeError('Target file %s exists and is not a symlink.' % path)

  @staticmethod
  def create_dir(path):
    path = os.path.abspath(os.path.expanduser(path))
    FileInstaller.remove_symlink(path)
    if not os.path.exists(path):
      print('Creating directory %s...' % path)
      os.makedirs(path)

  @staticmethod
  def create_file(path):
    path = os.path.abspath(os.path.expanduser(path))
    dir = os.path.dirname(path)
    if not os.path.islink(dir):
      FileInstaller.create_dir(dir)
    if not os.path.exists(path):
      print('Creating file %s...' % path)
      with open(path, 'wb'):
        pass

  @staticmethod
  def copy_file(source, target):
    source = os.path.abspath(os.path.expanduser(source))
    target = os.path.expanduser(target)
    if target.endswith('/') or target.endswith('\\'):
      target = os.path.join(target, os.path.basename(source))
    FileInstaller.remove_symlink(target)
    print('Copying %s to %s...' % (source, target))
    FileInstaller.create_dir(os.path.dirname(target))
    shutil.copy(source, target)

  @staticmethod
  def merge_files(sources, target, comment_prefix = None):
    for i, _ in enumerate(sources):
      sources[i] = os.path.abspath(os.path.expanduser(sources[i]))
    target = os.path.expanduser(target)
    if target.endswith('/') or target.endswith('\\'):
      target = os.path.join(target, os.path.basename(sources[0]))
    FileInstaller.remove_symlink(target)
    FileInstaller.create_dir(os.path.dirname(target))
    print('Merging %s to %s...' % (sources, target))
    with open(target, 'w') as ft:
      if comment_prefix:
        ft.write('%s DO NOT EDIT THIS FILE!%s'
          % (comment_prefix, os.linesep))
        ft.write('%s File was generated automatically by dotfiles installer.%s'
          % (comment_prefix, os.linesep))
        ft.write('%s All changes in this file will be overwritten.%s%s'
          % (comment_prefix, os.linesep, os.linesep))
      for source in sources:
        with open(source) as fs:
            if comment_prefix:
              ft.write('%s Merge part %s of %s from: %s%s%s'
                % (comment_prefix, sources.index(source)+1, len(sources), source, os.linesep, os.linesep))
            ft.write(fs.read())
            ft.write(os.linesep)

  @staticmethod
  def create_symlink(source, target):
    source = os.path.abspath(os.path.expanduser(source))
    target = os.path.expanduser(target)
    if target.endswith('/') or target.endswith('\\'):
      target = os.path.join(target, os.path.basename(source))
    FileInstaller.remove_symlink(target)
    FileInstaller.check_exists(target)
    FileInstaller.create_dir(os.path.dirname(target))
    print('Linking %s to %s...' % (source, target))
    os.symlink(source, target)

  @staticmethod
  def remove_symlink(path):
    path = os.path.abspath(os.path.expanduser(path))
    if os.path.islink(path):
      print('Removing old symlink: %s ...' % path)
      os.unlink(path)

class CygwinPackageInstaller(object):
  name = 'cygwin'

  def supported(self):
    return FileInstaller.has_executable('apt-cyg')

  def is_installed(self, package):
    return len(run_silent(['apt-cyg', 'list', '^%s$' % package])[1]) > 0

  def is_available(self, package):
    return len(run_silent(['apt-cyg', 'listall', '^%s$' % package])[1]) > 0

  def install(self, package):
    return run_verbose(['apt-cyg', 'install', package])

class PacmanPackageInstaller(object):
  name = 'pacman'

  def supported(self):
    return FileInstaller.has_executable('pacman') and FileInstaller.has_executable('sudo')

  def is_installed(self, package):
    return run_silent(['pacman', '-Q', package])[0]

  def is_available(self, package):
    return run_silent(['pacman', '-Ss', package])[0]

  def install(self, package):
    return run_verbose(['sudo', 'pacman', '-S', package])

class AurPackageInstaller(object):
  name = 'yay'

  def supported(self):
    return FileInstaller.has_executable('yay') and FileInstaller.has_executable('sudo')

  def is_installed(self, package):
    return run_silent(['yay', '-Q', package])[0]

  def is_available(self, package):
    return run_silent(['yay', '-Ss', package])[0]

  def install(self, package):
    return run_verbose(['yay', '-S', package])

class PipPackageInstaller(object):
  name = 'pip'
  cache_dir = tempfile.gettempdir()

  def __init__(self):
    if 'cygwin' in sys.platform:
      self.executable = 'pip3'
      self.use_sudo = False
    else:
      self.executable = 'pip'
      self.use_sudo = True

  def supported(self):
    if self.use_sudo and not FileInstaller.has_executable('sudo'):
      return False
    return FileInstaller.has_executable(self.executable)

  def is_installed(self, package):
    return re.search(
      '^' + re.escape(package) + r'($|\s)',
      run_silent([self.executable, 'list'])[1],
      re.MULTILINE) is not None

  def is_available(self, package):
    return re.search(
      '^' + re.escape(package) + r'($|\s)',
      run_silent([self.executable, 'search', package, '--cache-dir', self.cache_dir])[1],
      re.MULTILINE) is not None

  def install(self, package):
    command = [self.executable, 'install', '--user' '--cache-dir', self.cache_dir, package]
    if self.use_sudo:
      command = ['sudo'] + command
    return run_verbose(command)

class PackageInstaller(object):
  INSTALLERS = [
    CygwinPackageInstaller(),
    PacmanPackageInstaller(),
    AurPackageInstaller(),
    PipPackageInstaller(),
  ]

  @staticmethod
  def try_install(package, method=None):
    try:
      PackageInstaller.install(package, method)
    except Exception as e:
      print('Error installing %s: %s' % (package, e))

  @staticmethod
  def install(package, method=None):
    if method is None:
      chosen_installers = PackageInstaller.INSTALLERS
    else:
      chosen_installers = [i for i in PackageInstaller.INSTALLERS if i.name == method]

    chosen_installers = [i for i in chosen_installers if i.supported()]
    if len(chosen_installers) == 0:
      if method is None:
        raise RuntimeError('No package manager is supported on this system!')
      else:
        raise RuntimeError('%s is not supported on this system!' % method)

    for installer in chosen_installers:
      if installer.is_installed(package):
        print('Package %s is already installed.' % package)
        return True
      elif installer.is_available(package):
        print('Package %s is available, installing with %s' % (package, installer.name))
        return installer.install(package)

    if method is None:
      raise RuntimeError('No package manager is capable of installing %s' % package)
    else:
      raise RuntimeError('%s is not capable of installing %s' % (method, package))