import os
import pipes
import re
import sys
from collections import namedtuple
from contextlib import contextmanager
from functools import wraps

from parallels.core import MigrationError, MigrationNoContextError
from parallels.core import messages
from parallels.core.actions.migration_list.utils import get_migration_list_file_location
from parallels.core.logging import get_logger
from parallels.core.logging import log_context
from parallels.core.registry import Registry
from parallels.core.runners.base import BaseRunner
from parallels.core.runners.entities import ExecutionOptions
from parallels.core.utils import unix_utils, get_thirdparties_base_path
from parallels.core.utils.cagefs import is_cagefs_enabled_for_user, disable_cagefs, enable_cagefs
from parallels.core.utils.common import cached, is_ascii_string, safe_format, open_no_inherit, default, mkdir_p
from parallels.core.utils.common_constants import SSH_DEFAULT_PORT
from parallels.core.utils.download_utils import download_zip
from parallels.core.utils.file_utils import files_have_equal_contents
from parallels.core.utils.locks.file_lock_exception import FileLockException
from parallels.core.utils.steps_profiler import get_default_steps_profiler
from parallels.core.utils.unix_utils import format_command

logger = get_logger(__name__)


def copy_directory_content_unix(
    source_ip, source_user, source_server, target_server, source_path, target_path,
    key_pathname, exclude=None, skip_if_source_dir_not_exists=False,
    rsync_additional_args=None,
    user_mapping=None,
    group_mapping=None,
    source_rsync_bin=None, source_port=SSH_DEFAULT_PORT,
    target_user=None,
    preserve_owner=False,
    preserve_group=False,
    preserve_perms=False
):
    """
    :type source_ip: str | unicode
    :type source_user: str | unicode
    :type source_server: parallels.core.connections.server.Server
    :type target_server: parallels.core.connections.server.Server
    :type source_path: str | unicode
    :type target_path: str| unicode
    :type key_pathname: str | unicode
    :type exclude: list[str | unicode] | None
    :type skip_if_source_dir_not_exists: bool
    :type rsync_additional_args: list[str | unicode] | None
    :type user_mapping: dict[str | unicode, str | unicode] | None
    :type group_mapping: dict[str | unicode, str | unicode] | None
    :type source_rsync_bin: str | unicode
    :type source_port: int | str | unicode
    :type target_user: str | unicode | None
    :type preserve_owner: bool
    :type preserve_group: bool
    :type preserve_perms: bool
    :rtype: None
    """
    exclude = default(exclude, [])
    user_mapping = default(user_mapping, {})
    group_mapping = default(group_mapping, {})

    with source_server.runner() as runner_source:
        if runner_source.file_exists(source_path):
            _copy_directory_content_unix_plain(
                source_ip, source_user, source_server, target_server, source_path, target_path,
                key_pathname, exclude=exclude,
                rsync_additional_args=rsync_additional_args,
                user_mapping=user_mapping,
                group_mapping=group_mapping,
                source_rsync_bin=source_rsync_bin,
                source_port=source_port,
                target_user=target_user,
                preserve_owner=preserve_owner,
                preserve_group=preserve_group,
                preserve_perms=preserve_perms
            )
        else:
            if skip_if_source_dir_not_exists:
                # source directory does not exists, so there is nothing to copy
                return
            else:
                raise SourceDirectoryDoesNotExistError(
                    safe_format(messages.SOURCE_DIRECTORY_DOES_NOT_EXIST, path=source_path, server_ip=source_ip)
                )


class SourceDirectoryDoesNotExistError(MigrationError):
    pass


DEFAULT_RSYNC_PATH = '/usr/bin/rsync'


def detect_rsync_path(runner, node_info_string):
    if runner.file_exists(DEFAULT_RSYNC_PATH):
        return DEFAULT_RSYNC_PATH
    else:
        raise MigrationError(
            messages.UNABLE_TO_FIND_RSYNC_BINARY % (
                node_info_string, DEFAULT_RSYNC_PATH
            )
        )


def create_rsync_command(
    key_pathname, source_runner, source_user, source_ip, 
    source_filepath, target_filepath, exclude=None,
    rsync_additional_args=None,
    source_rsync_bin=None,
    source_port=SSH_DEFAULT_PORT, target_user=None,
    preserve_owner=False, preserve_group=False, preserve_perms=False
):
    """
    :type key_pathname: str | unicode
    :type source_runner: parallels.core.runners.base.BaseRunner
    :type source_user: str | unicode
    :type source_ip: str | unicode
    :type source_filepath: str | unicode
    :type target_filepath: str | unicode
    :type exclude: list[str | unicode] | None
    :type rsync_additional_args: list[str | unicode] | None
    :type source_rsync_bin: str | unicode | None
    :type source_port: int | str | unicode
    :type target_user: str | unicode | None
    :type preserve_owner: bool
    :type preserve_group: bool
    :type preserve_perms: bool
    :rtype: tuple
    """
    exclude = default(exclude, [])
    rsync_additional_args = default(rsync_additional_args, [])

    umask_command = u'umask 022'
    # Compose base command
    rsync_command = (
        u'/usr/bin/rsync -e {ssh_command} --timeout=30 --recursive --links --times --devices --specials'
    )
    rsync_command = umask_command + u'; ' + rsync_command
    rsync_args = dict(
        ssh_command=format_command(
            u'ssh -i {key} -p {port} '
            u'-o PasswordAuthentication=no -o StrictHostKeyChecking=no -o GSSAPIAuthentication=no',
            key=key_pathname,
            port=source_port
        )
    )

    # Add source rsync path argument
    if source_rsync_bin is None:
        # detect rsync binary location on the source node
        source_rsync_bin = detect_rsync_path(source_runner, source_ip)
        if source_rsync_bin != DEFAULT_RSYNC_PATH:  # non-default path
            rsync_command += u' --rsync-path={source_rsync_path}'
            rsync_args.update(dict(
                source_rsync_path=source_rsync_bin
            ))

    # Add preserve owner, group, permissions options
    if preserve_owner:
        rsync_command += u' --owner'
    if preserve_group:
        rsync_command += u' --group'
    if preserve_perms:
        rsync_command += u' --perms'
    else:
        rsync_command += u' --chmod=Fu=rw,Fgo=r,Du=rwx,Dgo=rx'

    # Add additional arguments
    for num, rsync_additional_arg in enumerate(rsync_additional_args):
        key = u'rsync_additional_arg_%s' % num
        rsync_command += u' {%s}' % key
        rsync_args[key] = rsync_additional_arg

    # Add excludes
    for num, exclude_value in enumerate(exclude):
        key = u'exclude_%s' % num
        rsync_command += u' --exclude={%s}' % key
        rsync_args[key] = exclude_value

    # Add source and destination paths
    rsync_command += u' {source_path} {target_path}'
    rsync_args.update(dict(
        source_path=u'{source_user}@{source_ip}:{source_filepath}'.format(
            source_user=source_user,
            source_ip=source_ip,
            source_filepath=pipes.quote(source_filepath)
        ),
        target_path=target_filepath
    ))

    if target_user is None:
        return rsync_command, rsync_args
    else:
        su_command = u'su {target_user} --shell /bin/sh -c {rsync_command}'
        su_args = dict(
            target_user=target_user,
            rsync_command=format_command(rsync_command, **rsync_args)
        )
        return su_command, su_args


def _copy_directory_content_unix_plain(
    source_ip, source_user, source_server, target_server,
    source_path, target_path, key_pathname,
    exclude=None, rsync_additional_args=None,
    user_mapping=None, group_mapping=None,
    source_rsync_bin=None, source_port=SSH_DEFAULT_PORT,
    target_user=None,
    preserve_owner=False, preserve_group=False, preserve_perms=False
):
    """
    :type source_ip: str | unicode
    :type source_user: str | unicode
    :type source_server: parallels.core.connections.server.Server
    :type target_server: parallels.core.connections.server.Server
    :type source_path: str | unicode
    :type target_path: str | unicode
    :type key_pathname: str | unicode
    :type exclude: list[str | unicode] | None
    :type rsync_additional_args: list[str | unicode] | None
    :type user_mapping: dict[str | unicode, str | unicode] | None
    :type group_mapping: dict[str | unicode, str | unicode] | None
    :type source_rsync_bin: str | unicode
    :type source_port: int | str | unicode
    :type target_user: str | unicode | None
    :type preserve_owner: bool
    :type preserve_group: bool
    :type preserve_perms: bool
    """
    exclude = default(exclude, [])
    user_mapping = default(user_mapping, {})
    group_mapping = default(group_mapping, {})

    if not source_path.endswith('/'):
        source_path += '/'
    if not target_path.endswith('/'):
        target_path += '/'

    # CageFS is a CloudLinux feature which allows to restrict access to files of other users.
    # When it is enabled, it is not possible to run "rsync" utility under user on the target server.
    # So, we temporary disable it for the user.
    disabled_cagefs = False

    if target_user is not None:
        is_cagefs_enabled = is_cagefs_enabled_for_user(target_server, target_user)
        if is_cagefs_enabled:
            try:
                disabled_cagefs = True
                disable_cagefs(target_server, target_user)
            except Exception as e:
                logger.exception()
                logger.fwarn(
                    messages.FAILED_TO_DISABLE_CAGEFS_FOR_TARGET_USER,
                    username=target_user, server=target_server.description(),
                    reason=e
                )

    try:
        with source_server.runner() as runner_source, target_server.runner() as runner_target:
            assert isinstance(runner_target, BaseRunner)
            cmd, args = create_rsync_command(
                key_pathname, runner_source, source_user,
                source_ip, source_path, target_path, exclude,
                rsync_additional_args,
                source_rsync_bin, source_port=source_port,
                target_user=target_user,
                preserve_owner=preserve_owner,
                preserve_group=preserve_group,
                preserve_perms=preserve_perms
            )
            # Define umask to overwrite the default mask, used by migrator.
            # For security reasons migrator can use special umask.
            # Calling rsync with the default umask can create files with unexpected permissions.
            runner_target.execute_command(cmd, args, ExecutionOptions(umask=000))

            for mapping_source_user, mapping_target_user in user_mapping.items():
                logger.debug(messages.CHANGE_FILE_PERMISSIONS_ACCORDING_USER_MAP)
                unix_utils.map_copy_owner_user(
                    runner_source, runner_target, source_path, target_path, mapping_source_user, mapping_target_user
                )
            for mapping_source_group, mapping_target_group in group_mapping.items():
                logger.debug(messages.CHANGE_FILE_PERMISSIONS_ACCORDING_GROUP_MAP)
                unix_utils.map_copy_owner_group(
                    runner_source, runner_target, source_path, target_path, mapping_source_group, mapping_target_group
                )
    finally:
        if disabled_cagefs:
            try:
                enable_cagefs(target_server, target_user)
            except Exception as e:
                logger.exception()
                logger.fwarn(
                    messages.FAILED_TO_ENABLE_CAGEFS_FOR_TARGET_USER,
                    username=target_user, server=target_server.description(),
                    reason=e
                )


def trace(op_name, message):
    def decorator(method):
        @wraps(method)
        def wrapper(self, *args, **kw):
            with trace_step(op_name, message, profile=True):
                return method(self, *args, **kw)
        return wrapper
    return decorator


@contextmanager
def trace_step(op_name, message, profile=False, log_level='info', compound=True):
    if log_level == 'info':
        log_function = logger.info
    elif log_level == 'debug':
        log_function = logger.debug
    else:
        raise Exception("Invalid log level '%s'" % log_level)

    if compound:
        # put an empty line before the block
        log_function(u"")
        log_function(messages.TRACE_START, message)
    else:
        log_function(message)
    try:
        with log_context(op_name):
            if profile:
                with get_default_steps_profiler().measure_time(
                    op_name, message
                ):
                    yield
            else:
                yield
        if compound:
            log_function(messages.TRACE_FINISH, message)
    except MigrationNoContextError:
        if compound:
            log_function(messages.TRACE_FINISH, message)
        raise
    except:
        logger.error(messages.TRACE_ABORT, message)
        raise


class DbServerInfo(namedtuple('DbServerInfo', ('dbtype', 'host', 'port', 'login', 'password'))):
    @staticmethod
    def from_plesk_api(db_server_info):
        """Convert from plesk_api.DbServerInfo to DbServerInfo"""
        return DbServerInfo(
            dbtype=db_server_info.dbtype,
            host=db_server_info.host,
            port=db_server_info.port,
            login=db_server_info.admin,
            password=db_server_info.password
        )

    @staticmethod
    def from_plesk_backup(db_server_info):
        """Convert plesks_migrator.data_model.DatabaseServer to DbServerInfo"""
        return DbServerInfo(
            dbtype=db_server_info.dbtype,
            host=db_server_info.host,
            port=db_server_info.port,
            login=db_server_info.login,
            password=db_server_info.password
        )


def get_package_root_path(package):
    """Get path to Python package root directory"""
    dirs = [p for p in package.__path__]
    assert all(d == dirs[0] for d in dirs)
    return dirs[0]


def get_package_extras_file_path(package, filename):
    """Get path to extras file of Python package

    :type filename: str | unicode
    :rtype: str | unicode
    """
    return os.path.join(
        get_package_root_path(package), 'extras', filename
    )


def get_package_scripts_file_path(package, filename):
    """Get path to scripts file of Python package

    :type filename: str | unicode
    :rtype: str | unicode
    """
    return os.path.join(
        get_package_root_path(package), 'scripts', filename
    )


def version_to_tuple(version):
    return tuple(map(int, (version.split(b'.'))))


def split_unix_path(path):
    """Split path by directory name and file name

    Returns tuple (directory name, file name)

    :rtype tuple
    """
    last_slash_pos = path.rfind('/')
    directory = path[:last_slash_pos]
    filename = path[last_slash_pos+1:]
    return directory, filename


def normalize_domain_name(domain_name):
    """Normalize domain name - convert to lowercase punycode, convert wildcard subdomains from "*" to "_" notation

    Use this function for comparison of domain names.

    :rtype str | unicode | None
    """
    if domain_name is None:
        return None

    return domain_name.encode('idna').lower().replace('*', '_')


@cached
def get_version():
    version_file_name = os.path.join(Registry.get_instance().get_base_dir(), 'version')
    if not os.path.exists(version_file_name):
        version = 'DEVELOPMENT'
    else:
        version = read_string_from_file(version_file_name)
    if sys.version_info.major >= 3:
        return version.encode('utf-8')
    else:
        return version


def read_string_from_file(file_name):
    with open_no_inherit(file_name, 'r') as fp:
        return fp.read().strip()


def format_owner_login(owner_login):
    """Format login variable of admin/reseller/client into a string to be used in messages

    Considers administrator (for which login is None, according to different models used in migrator)

    :type owner_login: str | unicode
    :rtype: unicode
    """
    if owner_login is None:
        return u'administrator'
    else:
        return u"'%s'" % owner_login


def get_option_value_safe(options, option_name, default_value):
    """Read command line option in a safe way

    If there is no command line option for a command, warning will be emitted and default value returned.
    If you will try to use option directly, migrator will completely fail.
    """
    if hasattr(options, option_name):
        return getattr(options, option_name)
    else:
        logger.warning(messages.NO_COMMAND_LINE_OPTION_WARNING.format(option_name=option_name))
        return default_value


def get_optional_option_value(options, option_name, default_value):
    """Read command line option, which could exist or not exist

    If there is no command line option for a command, default value returned.
    """
    if hasattr(options, option_name):
        return getattr(options, option_name)
    else:
        return default_value


def is_locked():
    result = True
    migration_lock = Registry.get_instance().get_migration_lock()
    try:
        migration_lock.acquire()
        migration_lock.release()
        result = False
    except FileLockException:
        # assume that command is still in progress
        pass
    return result


def safe_idn_decode(domain_name):
    """Safely decode domain name from punycode to unicode

    If domain name is already in unicode - return it as is. If any error occurred during decoding - return domain
    name as is.

    :type domain_name: str | unicode
    :rtype: unicode
    """
    if not is_ascii_string(domain_name):
        return domain_name

    try:
        return domain_name.decode('idna')
    except:
        logger.exception()
        logger.warning(messages.UNABLE_TO_CONVERT_IDN_NAME.format(domain_name=domain_name))
        return domain_name


def is_unicode_domain(domain_name):
    """Whether specified domain is in unicode or in punycode (IDNA). For unicode form - return true.

    :rtype: bool
    """
    return domain_name.encode('idna') != domain_name


def safe_mail_idn_decode(email_address):
    """
    :type email_address: str | unicode
    :rtype: str | unicode
    """
    email_address_parts = email_address.split('@')
    if len(email_address_parts) != 2:
        return email_address
    mail_name, domain_name = email_address_parts
    return '@'.join([mail_name, safe_idn_decode(domain_name)])


def path_startswith(path1, path2):
    path1_items = [i for i in re.split('[/\\\\]', path1) if i != '']
    path2_items = [i for i in re.split('[/\\\\]', path2) if i != '']
    return path1_items == path2_items[:len(path1_items)]


def is_running_from_cli():
    """Check if migrator is running from command-line interface (comparing to GUI)

    :rtype: bool
    """
    messages_ui = os.getenv('MIGRATOR_MESSAGES_UI')
    return messages_ui != 'gui'


def ip_file_path(ip_address):
    """Get file path path which corresponds to specified IP address,

    Function replaces symbols which are valid in IP address, but not valid in filesystem path

    :type ip_address: str | unicode
    :rtype: str | unicode
    """
    return ip_address.replace(':', '_').replace('.', '_')


def download_thirdparty_zip(remote_name, local_name, log_message):
    """Download thirdparty ZIP archive and unpack to a directory

    If it was already downloaded before - skip.
    Returns path to a local directory with thirdparty.

    :type remote_name: str | unicode
    :type local_name: str | unicode
    :type log_message: str | unicode
    :rtype: str | unicode
    """
    dir_path = os.path.join(get_thirdparties_base_path(), local_name)

    if not os.path.exists(get_thirdparties_base_path()):
        mkdir_p(get_thirdparties_base_path())

    if not os.path.exists(dir_path):
        url = 'http://autoinstall.plesk.com/panel-migrator/thirdparties/%s' % remote_name
        logger.info(log_message.format(url=url))
        download_zip(url, dir_path)

    return dir_path


def is_cache_enabled_and_valid(global_context, cache_migration_list_file):
    """Whether caching is enabled and cache is valid for different files and models like converted dump, etc

    By default caching is disabled for 'transfer-accounts' and 'check' commands, but enabled for post-migration
    commands like 'copy-content', 'copy-web-content', 'test-sites', etc. It allows to run post-migration
    operations fast once migration was performed, but keep data up to date when running migration itself.
    Caching is performed by migration list: if migration list file changed (even by single whitespace),
    then cache is considered invalid.

    Cache migration list file is a file which contains contents of migration list file by the moment when cache
    was updated last time.

    :type global_context: parallels.core.global_context.GlobalMigrationContext
    :type cache_migration_list_file: str | unicode
    :rtype: bool
    """
    if not _is_caching_enabled(global_context):
        return False

    migration_list_file = get_migration_list_file_location(global_context)
    if migration_list_file is None:
        return False

    if not os.path.exists(cache_migration_list_file):
        return False

    if not os.path.exists(migration_list_file):
        return False

    if not files_have_equal_contents(migration_list_file, cache_migration_list_file):
        return False

    return True


def _is_caching_enabled(global_context):
    """Whether caching is enabled for different files and models like converted dump, existing objects model, etc

    By default caching is disabled for 'transfer-accounts' and 'check' commands, but enabled for post-migration
    commands like 'copy-content', 'copy-web-content', 'test-sites', etc.

    :type global_context: parallels.core.global_context.GlobalMigrationContext
    :rtype: bool
    """
    if hasattr(global_context.options, 'use_cache'):
        return global_context.options.use_cache
    else:
        return False
