<commit_before>#!/usr/bin/env python
#
# helper.py - Abstract provider of a non-Python helper program.
#
# February 2015, <NAME>
# Copyright (c) 2015 the COT project developers.
# See the COPYRIGHT.txt file at the top-level directory of this distribution
# and at https://github.com/glennmatthews/cot/blob/master/COPYRIGHT.txt.
#
# This file is part of the Common OVF Tool (COT) project.
# It is subject to the license terms in the LICENSE.txt file found in the
# top-level directory of this distribution and at
# https://github.com/glennmatthews/cot/blob/master/LICENSE.txt. No part
# of COT, including this file, may be copied, modified, propagated, or
# distributed except according to the terms contained in the LICENSE.txt file.

"""Interface for providers of non-Python helper programs.

Provides the ability to install the program if not already present,
and the ability to run the program as well.
"""

import contextlib
import logging
import os.path
import re
import requests
import shutil
import subprocess
import tarfile
from distutils.spawn import find_executable
from distutils.version import StrictVersion

try:
    # Python 3.x
    from tempfile import TemporaryDirectory
except ImportError:
    # Python 2.x
    import tempfile

    @contextlib.contextmanager
    def TemporaryDirectory(suffix='', prefix='tmp', dir=None):
        """Create a temporary directory and make sure it's deleted later.

        Reimplementation of Python 3's ``tempfile.TemporaryDirectory``.
        """
        tempdir = tempfile.mkdtemp(suffix, prefix, dir)
        try:
            yield tempdir
        finally:
            shutil.rmtree(tempdir)

from verboselogs import VerboseLogger
logging.setLoggerClass(VerboseLogger)
logger = logging.getLogger(__name__)


# Default method to be used when COT.helpers is run by itself without
# the broader COT package.
# The COT package will override this with e.g. the CLI.confirm method.
def confirm(prompt, force=False):
    """Prompt user to confirm the requested operation.

    Auto-accepts if :attr:`force` is set to ``True``.

    :param str prompt: Message to prompt the user with
    :return: ``True`` (user confirms acceptance) or ``False``
      (user declines)
    """
    return True


class HelperNotFoundError(OSError):

    """A helper program cannot be located."""


class HelperError(EnvironmentError):

    """A helper program exited with non-zero return code."""


class Helper(object):

    """A provider of a non-Python helper program.

    **Class Properties**

    .. autosummary::
      :nosignatures:

      PACKAGE_MANAGERS

    **Class Methods**

    .. autosummary::
      :nosignatures:

      apt_install
      port_install
      yum_install
      download_and_expand
      find_executable

    **Instance Properties**

    .. autosummary::
      name
      path
      version

    **Instance Methods**

    .. autosummary::
      :nosignatures:

      call_helper
      install_helper
    """

    PACKAGE_MANAGERS = {
        "port":    find_executable('port'),
        "apt-get": find_executable('apt-get'),
        "yum":     find_executable('yum'),
    }
    """Class-level lookup for package manager executables."""

    @classmethod
    def find_executable(cls, name):
        """Wrapper for :func:`distutils.spawn.find_executable`."""
        return find_executable(name)

    @classmethod
    @contextlib.contextmanager
    def download_and_expand(cls, url):
        """Context manager for downloading and expanding a .tar.gz file.

        Creates a temporary directory, downloads the specified URL into
        the directory, unzips and untars the file into this directory,
        then yields to the given block. When the block exits, the temporary
        directory and its contents are deleted.

        ::

          with self.download_and_expand("http://example.com/foo.tgz") as d:
            # archive contents have been extracted to 'd'
            ...
          # d is automatically cleaned up.

        :param str url: URL of a .tgz or .tar.gz file to download.
        """
        with TemporaryDirectory(prefix="cot_helper") as d:
            logger.debug("Temporary directory is {0}".format(d))
            logger.verbose("Downloading and extracting {0}".format(url))
            response = requests.get(url, stream=True)
            tgz = os.path.join(d, 'helper.tgz')
            with open(tgz, 'wb') as f:
                shutil.copyfileobj(response.raw, f)
            del response
            logger.debug("Extracting {0}".format(tgz))
            # the "with tarfile.open()..." construct isn't supported in 2.6
            tarf = tarfile.open(tgz, "r:gz")
            try:
                tarf.extractall(path=d)
            finally:
                tarf.close()
            try:
                yield d
            finally:
                logger.debug("Cleaning up temporary directory {0}".format(d))

    _apt_updated = False
    """Whether we have run 'apt-get update' yet."""

    @classmethod
    def apt_install(cls, package):
        """Try to use ``apt-get`` to install a package."""
        if not cls.PACKAGE_MANAGERS['apt-get']:
            return False
        if not cls._apt_updated:
            cls._check_call(['sudo', 'apt-get', '-q', 'update'])
            cls._apt_updated = True
        cls._check_call(['sudo', 'apt-get', '-q', 'install', package])
        return True

    _port_updated = False
    """Whether we have run 'port selfupdate' yet."""

    @classmethod
    def port_install(cls, package):
        """Try to use ``port`` to install a package."""
        if not cls.PACKAGE_MANAGERS['port']:
            return False
        if not cls._port_updated:
            cls._check_call(['sudo', 'port', 'selfupdate'])
            cls._port_updated = True<commit_msg>Check dpkg before calling apt-get<commit_after>#!/usr/bin/env python
#
# helper.py - Abstract provider of a non-Python helper program.
#
# February 2015, <NAME>
# Copyright (c) 2015 the COT project developers.
# See the COPYRIGHT.txt file at the top-level directory of this distribution
# and at https://github.com/glennmatthews/cot/blob/master/COPYRIGHT.txt.
#
# This file is part of the Common OVF Tool (COT) project.
# It is subject to the license terms in the LICENSE.txt file found in the
# top-level directory of this distribution and at
# https://github.com/glennmatthews/cot/blob/master/LICENSE.txt. No part
# of COT, including this file, may be copied, modified, propagated, or
# distributed except according to the terms contained in the LICENSE.txt file.

"""Interface for providers of non-Python helper programs.

Provides the ability to install the program if not already present,
and the ability to run the program as well.
"""

import contextlib
import logging
import os.path
import re
import requests
import shutil
import subprocess
import tarfile
from distutils.spawn import find_executable
from distutils.version import StrictVersion

try:
    # Python 3.x
    from tempfile import TemporaryDirectory
except ImportError:
    # Python 2.x
    import tempfile

    @contextlib.contextmanager
    def TemporaryDirectory(suffix='', prefix='tmp', dir=None):
        """Create a temporary directory and make sure it's deleted later.

        Reimplementation of Python 3's ``tempfile.TemporaryDirectory``.
        """
        tempdir = tempfile.mkdtemp(suffix, prefix, dir)
        try:
            yield tempdir
        finally:
            shutil.rmtree(tempdir)

from verboselogs import VerboseLogger
logging.setLoggerClass(VerboseLogger)
logger = logging.getLogger(__name__)


# Default method to be used when COT.helpers is run by itself without
# the broader COT package.
# The COT package will override this with e.g. the CLI.confirm method.
def confirm(prompt, force=False):
    """Prompt user to confirm the requested operation.

    Auto-accepts if :attr:`force` is set to ``True``.

    :param str prompt: Message to prompt the user with
    :return: ``True`` (user confirms acceptance) or ``False``
      (user declines)
    """
    return True


class HelperNotFoundError(OSError):

    """A helper program cannot be located."""


class HelperError(EnvironmentError):

    """A helper program exited with non-zero return code."""


class Helper(object):

    """A provider of a non-Python helper program.

    **Class Properties**

    .. autosummary::
      :nosignatures:

      PACKAGE_MANAGERS

    **Class Methods**

    .. autosummary::
      :nosignatures:

      apt_install
      port_install
      yum_install
      download_and_expand
      find_executable

    **Instance Properties**

    .. autosummary::
      name
      path
      version

    **Instance Methods**

    .. autosummary::
      :nosignatures:

      call_helper
      install_helper
    """

    PACKAGE_MANAGERS = {
        "port":    find_executable('port'),
        "apt-get": find_executable('apt-get'),
        "yum":     find_executable('yum'),
    }
    """Class-level lookup for package manager executables."""

    @classmethod
    def find_executable(cls, name):
        """Wrapper for :func:`distutils.spawn.find_executable`."""
        return find_executable(name)

    @classmethod
    @contextlib.contextmanager
    def download_and_expand(cls, url):
        """Context manager for downloading and expanding a .tar.gz file.

        Creates a temporary directory, downloads the specified URL into
        the directory, unzips and untars the file into this directory,
        then yields to the given block. When the block exits, the temporary
        directory and its contents are deleted.

        ::

          with self.download_and_expand("http://example.com/foo.tgz") as d:
            # archive contents have been extracted to 'd'
            ...
          # d is automatically cleaned up.

        :param str url: URL of a .tgz or .tar.gz file to download.
        """
        with TemporaryDirectory(prefix="cot_helper") as d:
            logger.debug("Temporary directory is {0}".format(d))
            logger.verbose("Downloading and extracting {0}".format(url))
            response = requests.get(url, stream=True)
            tgz = os.path.join(d, 'helper.tgz')
            with open(tgz, 'wb') as f:
                shutil.copyfileobj(response.raw, f)
            del response
            logger.debug("Extracting {0}".format(tgz))
            # the "with tarfile.open()..." construct isn't supported in 2.6
            tarf = tarfile.open(tgz, "r:gz")
            try:
                tarf.extractall(path=d)
            finally:
                tarf.close()
            try:
                yield d
            finally:
                logger.debug("Cleaning up temporary directory {0}".format(d))

    _apt_updated = False
    """Whether we have run 'apt-get update' yet."""

    @classmethod
    def apt_install(cls, package):
        """Try to use ``apt-get`` to install a package."""
        if not cls.PACKAGE_MANAGERS['apt-get']:
            return False
        # check if it's already installed
        msg = cls._check_output(['dpkg', '-s', package], require_success=False)
        if re.search('install ok installed', msg):
            return True
        if not cls._apt_updated:
            cls._check_call(['sudo', 'apt-get', '-q', 'update'])
            cls._apt_updated = True
        cls._check_call(['sudo', 'apt-get', '-q', 'install', package])
        return True

    _port_updated = False
    """Whether we have run 'port selfupdate' yet."""

    @classmethod
    def port_install(cls, package):
        """Try to use ``port`` to install a package."""
        if not cls.PACKAGE_MANAGERS['port']:
            return False
        if not cls._port_updated:
            cls._check_call(['sudo', 'port', 'selfupdate'])
            cls._port_updated = True
{'fraction_non_alphanumeric': 0.1012282930961457, 'fraction_numerical': 0.0028801355357899196, 'mean_word_length': 2.5646135265700485, 'pattern_counts': {'":': 6, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 4, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 14, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6304178', 'n_tokens_mistral': 3403, 'n_tokens_neox': 3238, 'n_words': 1098}