import json
import ntpath
import logging
import re
import posixpath
from xml.etree import ElementTree

from parallels.core import messages
from parallels.core import MigrationError
from parallels.core.utils.entity import Entity
from parallels.core.utils.plesk_cli_runner import PleskCLIRunnerCLI
from parallels.plesk.utils.xml_rpc.plesk import operator as plesk_ops
from parallels.core.utils.windows_utils import path_join as windows_path_join, get_from_registry, cmd_command
from parallels.core.utils import windows_utils
from parallels.core.utils.common import if_not_none, format_list, cached, is_empty

logger = logging.getLogger(__name__)


def get_windows_vhosts_dir(runner):
	vhosts_dir = get_from_registry(
		runner,
		['HKEY_LOCAL_MACHINE\SYSTEM\CurrentControlSet\Control\Session Manager\Environment'],
		'plesk_vhosts'
	)

	if vhosts_dir is None:
		raise Exception(messages.FAILED_DETECT_PLESK_VIRTUAL_HOSTS_DIRECTORY)

	vhosts_dir = vhosts_dir.strip('\\')
	logger.debug(messages.PLESK_VIRTUAL_HOSTS_DIRECTORY_IS_S % vhosts_dir)
	return vhosts_dir


def get_windows_plesk_dir(runner):
	plesk_dir = get_from_registry(
		runner,
		['HKEY_LOCAL_MACHINE\SYSTEM\CurrentControlSet\Control\Session Manager\Environment'],
		'plesk_dir'
	)

	if plesk_dir is None:
		raise Exception(messages.FAILED_DETECT_PLESK_BASE_DIRECTORY)

	plesk_dir = plesk_dir.strip('\\')
	logger.debug(messages.DEBUG_PLESK_BASE_DIRECTORY, plesk_dir)
	return plesk_dir


def get_windows_data_dir(runner):
	possible_registry_keys = [
		'HKLM\SOFTWARE\Wow6432Node\PLESK\PSA Config\Config',
		'HKLM\SOFTWARE\PLESK\PSA Config\Config'
	]
	data_dir = get_from_registry(runner, possible_registry_keys, 'PRODUCT_DATA_D')

	if data_dir is None:
		raise Exception(messages.FAILED_DETECT_PLESK_DATA_DIRECTORY)

	data_dir = data_dir.strip('\\')
	logger.debug(messages.DEBUG_PLESK_DATA_DIRECTORY, data_dir)
	return data_dir


def get_windows_dump_dir(runner):
	possible_registry_keys = [
		'HKLM\SOFTWARE\Wow6432Node\PLESK\PSA Config\Config',
		'HKLM\SOFTWARE\PLESK\PSA Config\Config'
	]
	dump_dir = get_from_registry(runner, possible_registry_keys, 'DUMP_D')

	dump_dir = dump_dir.strip('\\')
	logger.debug(messages.DEBUG_PLESK_DUMP_DIRECTORY, dump_dir)
	return dump_dir


def get_windows_vhost_dir(runner, vhost_name):
	vhosts_dir = get_windows_vhosts_dir(runner)

	vhost_name = vhost_name.encode('idna')
	vhost_dir = windows_path_join(vhosts_dir, vhost_name)
	logger.debug(messages.DEBUG_WINDOWS_VHOST_DIRECTORY, vhost_dir)

	return vhost_dir


def get_unix_vhost_dir(runner, vhost_name):
	vhosts_dir = get_unix_vhosts_dir(runner)

	vhost_name = vhost_name.encode('idna')
	vhost_dir = posixpath.join(vhosts_dir, vhost_name)
	logger.debug(messages.DEBUG_UNIX_VHOST_DIRECTORY, vhost_dir)

	return vhost_dir


def get_unix_vhost_system_dir(runner, vhost_name):
	vhosts_dir = get_unix_vhosts_dir(runner)

	vhost_name = vhost_name.encode('idna')
	vhost_dir = posixpath.join(vhosts_dir, 'system', vhost_name)
	logger.debug(messages.DEBUG_UNIX_VHOST_SYSTEM_DIRECTORY, vhost_dir)

	return vhost_dir


def get_unix_vhosts_dir(runner):
	"""Get directory with virtual hosts on Plesk for Unix server"""
	return get_unix_conf_var(
		runner, 'HTTPD_VHOSTS_D', messages.DEBUG_UNIX_VHOSTS_DIRECTORY
	)


def get_unix_mailnames_dir(runner):
	"""Get directory with mail messages on Plesk for Unix server"""
	return get_unix_conf_var(
		runner, '(PLESK|QMAIL)_MAILNAMES_D', messages.DEBUG_MAIL_MESSAGES_DIRECTORY
	)


def get_unix_dump_dir(runner):
	"""Get directory where Plesk stores backups on Plesk for Unix server"""
	return get_unix_conf_var(
		runner, 'DUMP_D', messages.DEBUG_BACKUP_DUMPS_DIRECTORY
	)


def get_unix_product_root_dir(runner):
	"""Get directory where Plesk for Unix is installed"""
	return get_unix_conf_var(
		runner, 'PRODUCT_ROOT_D', messages.DEBUG_UNIX_PRODUCT_ROOT_DIRECTORY
	)


def get_unix_conf_var(runner, var_name, description=None):
	"""Read variable from psa.conf config of Plesk for Unix

	Arguments:
	- runner - runner for Plesk for Unix server (from run_command module)
	- var_name - regular expression to match variable name
	- description - description of variable, just for debug logging

	Returns variable value.
	"""
	stdout = runner.run(
		'/bin/grep', ['-m1', '-E', '^\s*%s' % var_name, '/etc/psa/psa.conf']
	)
	var_value = stdout.strip().replace('\t', ' ').partition(' ')[2].strip().rstrip('/')
	if description is not None:
		logger.debug('%s: %s', description, var_value)
	else:
		logger.debug(
			messages.PLESK_FOR_UNIX_CONFIGURATION_VARIABLE_S,
			var_name, var_value
		)
	return var_value


def get_windows_db_server_credentials(server, db_type, db_host):
	def _get_credentials(_server, _db_type, _db_host):
		exit_code, stdout, _ = _call_php_cli_unchecked(_server, 'cu\database-registrar.php', [
			'--get-credentials', _db_host, '-type', _db_type
		])
		if exit_code != 0:
			return None
		return stdout.strip()

	output = None
	if db_type == 'mssql' and db_host.startswith('.\\') and ':' not in db_host:
		# workaround for bug PPP-15779: port 0 specified in Plesk database for local MSSQL server,
		# installer automatically, so first try to check this case
		output = _get_credentials(server, db_type, '%s:0' % db_host)
	if output is None:
		output = _get_credentials(server, db_type, db_host)

	if output is None:
		raise Exception(messages.UNABLE_RETRIVE_CREDENTIALS_FOR_S_DATABASE % (
			db_type, db_host, server.description()
		))

	xml = ElementTree.XML(output)
	return xml.findtext('username'), xml.findtext('password')


def get_unix_db_server_credentials(server, db_type, db_host):
	output = _call_php_cli(server, 'api-cli/database-registrar.php', [
		'--get-credentials', db_host, '-type', db_type
	]).strip()
	xml = ElementTree.XML(output)
	return xml.findtext('username'), xml.findtext('password')


def check_capability(plesk_server, input_path, output_path):
	"""
	Run check that given Plesk server is capable to migrate listed objects and provide report with detected issues
	:type plesk_server: parallels.plesk.connections.target_server.PleskTargetServer
	:param str input_path: path to capability dump
	:param output_path: path to report generated by Plesk capability checker
	"""
	_call_php_cli(plesk_server, 'backup/Conflicts/Runner.php', [
		'--check-capability',
		'--session-path="%s"' % plesk_server.session_dir(),
		'--log=conflict-resolve.log',
		'--capability-info="%s"' % input_path,
		'--capability-info-out="%s"' % output_path,
		'--used-space-coefficient=0.7',
		'--max-transfer-download-time=86400',
		'--min-transfer-download-speed=1.25'
	])


def convert_wildcard_to_path(domain_name):
	if domain_name.startswith('*'):
		return '_%s' % (domain_name[1:],)
	else:
		return domain_name


def get_plesk_for_unix_version(runner, plesk_dir):
	"""Get Plesk version of Plesk for Unix

	Returns two first numbers of the version as tuple of integers, for example (12, 5) or (9, 0).

	:rtype: tuple[int]
	"""
	version_str = runner.get_file_contents(posixpath.join(plesk_dir, 'version'))
	return tuple(map(int, re.split('[\. ]', version_str)[:2]))


def get_plesk_for_windows_version(runner, plesk_dir):
	"""Get Plesk version for Plesk for Windows

	Returns two first numbers of the version as tuple of integers, for example (12, 5) or (9, 0).

	:rtype: tuple[int]
	"""
	version_str = runner.get_file_contents(ntpath.join(plesk_dir, 'version'))
	return tuple(map(int, re.split('[\. ]', version_str)[:2]))


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


def set_mime_types(server, vhost_name, mime_types, vdir_name=None):
	"""Set MIME types of virtual host on Windows Plesk server

	Arguments:
	- server - instance of Windows target server (should inherit 2 classes:
		- parallels.core.connections.target_servers.TargetServer
		- parallels.core.connections.plesk_server.PleskServer
	)
	- vhost_name - virtual host name (subscription or addon domain)
	- mime_types - dictionary of mime types (key - extension, value - mime type)
	- vdir_name - virtual directory to set mime type on; if None - mime types
		are set on whole server
	"""
	vhost_name = vhost_name.encode('idna')

	with server.runner() as runner:
		mimetypes_file = None
		if server.plesk_version >= (12, 0):
			mimetypes_file = server.get_session_file_path(
				'%s.mimetypes' % vhost_name
			)
			runner.upload_file_content(
				mimetypes_file,
				_mime_types_as_string(mime_types)
			)
			mimetypes_arg = mimetypes_file
		else:
			mimetypes_arg = _mime_types_as_string(mime_types)

		runner.sh(
			_get_webservermng_command(
				server.websrvmng_bin,
				'set-mime-types',
				args={
					'vhost-name': vhost_name,
					'vdir-name': vdir_name,
					'mime-types': mimetypes_arg
				}
			)
		)

		if mimetypes_file is not None:
			runner.remove_file(mimetypes_file)


def get_mime_types(server, vhost_name, vdir_name=None):
	"""Get MIME types of virtual host on Windows Plesk server

	Arguments:
	- server - instance of Windows target server (should inherit 2 classes:
		- parallels.core.connections.target_servers.TargetServer
		- parallels.core.connections.plesk_server.PleskServer
	)
	- vhost_name - virtual host name (subscription or addon domain)
	- vdir_name - virtual directory name
	"""
	vhost_name = vhost_name.encode('idna')
	with server.runner() as runner:
		mime_types_str = runner.sh(
			_get_webservermng_command(
				server.websrvmng_bin,
				'get-mime-types',
				args={	
					'vhost-name': vhost_name,
					'vdir-name': vdir_name,
				}
			)
		)
	return _parse_mime_types_string(mime_types_str)


def get_error_documents(server, vhost_name, vdir_name=None):
	"""Get error documents of virtual host on Windows Plesk server

	Arguments:
	- server - instance of Windows target server (should inherit 2 classes:
		- parallels.core.connections.target_servers.TargetServer
		- parallels.core.connections.plesk_server.PleskServer
	)
	- vhost_name - virtual host name (subscription or addon domain)
	- vdir_name - virtual directory name
	"""
	vhost_name = vhost_name.encode('idna')

	with server.runner() as runner:
		error_documents_str = runner.sh(
			_get_webservermng_command(
				server.websrvmng_bin,
				'get-error-docs',
				args={	
					'vhost-name': vhost_name,
					'vdir-name': vdir_name
				}
			)
		)
	return error_documents_str


def get_vdir_info(server, vhost_name):
	with server.runner() as runner:
		vdir_info = runner.sh('{websrvmng_path} --list-vdirs --vhost-name={vhost_name}', dict(
			websrvmng_path=server.websrvmng_bin, vhost_name=vhost_name.encode('idna')
		))
	return vdir_info


def _mime_types_as_string(mime_types):
	return u"".join([ 
		u"%s=%s;" % (ext, mime_type) 
		for ext, mime_type in mime_types.iteritems() 
	])


def _parse_mime_types_string(mime_types_string):
	mime_types = {}
	for mime_type_str in mime_types_string.split(';'):
		mime_type_str = mime_type_str.strip()
		if mime_type_str == '':
			continue
		ext, mime_type = mime_type_str.split('=')
		mime_types[ext] = mime_type
	return mime_types


def _get_webservermng_command(websrvmng_path, action, args):
	command = '{websrvmng_path} --{action}'.format(
		websrvmng_path=windows_utils.quote_arg(
			websrvmng_path
		),
		action=action
	)

	for key, value in args.iteritems():
		if value is not None:
			command += ' --{key}={value}'.format(
				key=key,
				value=windows_utils.quote_arg(value)
			)

	return command


def convert_wildcard_domain_path(domain_path):
	"""Convert wildcard domain path: in domain physical path '*' symbol replaced to '_'.

	Arguments:
	- domain_path - vhost path to domain
	"""
	return domain_path.replace('*', '_')


def get_apache_restart_interval(target_server):
	"""
	:type target_server: parallels.core.connections.target_servers.TargetServer
	"""
	with target_server.runner() as runner:
		return int(runner.sh('/usr/local/psa/bin/server_pref --get-apache-restart-interval'))


def set_apache_restart_interval_value(target_server, new_value):
	"""
	:type target_server: parallels.core.connections.target_servers.TargetServer
	"""
	with target_server.runner() as runner:
		runner.sh('/usr/local/psa/bin/server_pref -u -restart-apache {new_value}', dict(new_value=new_value))


def restart_plesk_apache(runner, plesk_dir=None):
	"""
	:type runner: parallels.core.run_command.BaseRunner
	"""
	if plesk_dir is None:
		plesk_dir = get_unix_product_root_dir(runner)
	runner.sh("%s/admin/bin/websrvmng -r" % plesk_dir)


def get_plesk_ips_with_cli(plesk_server):
	"""Get information about IP addresses on specified Plesk server with Plesk CLI

	:type plesk_server: parallels.core.connections.plesk_server.PleskServer
	:rtype: list[parallels.core.utils.plesk_utils.IPAddressInfo]
	"""
	runner = PleskCLIRunnerCLI(plesk_server)
	out = runner.run('ipmanage', ['--xml-info'])
	xml = ElementTree.XML(out)
	ip_addresses = []
	for ip_node in xml.findall('ip'):
		public_ip_address = ip_node.findtext('publicIp')
		if public_ip_address.strip() == '':
			public_ip_address = None
		ip_addresses.append(IPAddressInfo(
			ip_type=ip_node.findtext('type'),
			ip_address=ip_node.findtext('ip_address'),
			public_ip_address=public_ip_address,
			hostings=if_not_none(ip_node.findtext('hostings'), int),
			clients=if_not_none(ip_node.findtext('clients'), int),
			iface=ip_node.findtext('iface'),
			mask=ip_node.findtext('mask')
		))

	return ip_addresses


def get_plesk_ips_with_api(plesk_server):
	"""Get information about IP addresses on specified Plesk server with Plesk API

	:type plesk_server: parallels.core.connections.plesk_server.PleskServer
	:rtype: list[parallels.core.utils.plesk_utils.IPAddressInfo]
	"""
	ips = plesk_server.plesk_api().send(plesk_ops.IpOperator.Get()).data
	ip_addresses = []

	for ip_response in ips:
		ip_addresses.append(IPAddressInfo(
			ip_type=ip_response.ip_type,
			ip_address=ip_response.ip_address,
			public_ip_address=ip_response.public_ip_address,
			hostings=0  # can't determine with Plesk API, just set to zero
		))

	return ip_addresses


def refresh_node_components(node):
	"""
	:type node: parallels.core.connections.plesk_server.PleskServer
	"""
	with node.runner() as runner:
		runner.sh(ur'"%s\admin\bin\defpackagemng" --get --force' % node.plesk_dir)


def change_dedicated_iis_app_pool_state(plesk_server, subscription_name, state):
	"""
	:param parallels.core.connections.plesk_server.PleskServer plesk_server: Plesk panel server of subscription
	:param unicode subscription_name: subscription (webspace) name
	:param bool state: whether to enable or disaple IIS application poll
	"""
	PleskCLIRunnerCLI(plesk_server).run(
		'subscription',
		[
			'--set-iis-app-pool-settings',
			subscription_name.encode('idna'),
			'-iis-app-pool-turned-on',
			'true' if state else 'false'
		],
		env=dict(PLESK_RESTORE_MODE='1')
	)


class IPAddressInfo(object):
	def __init__(self, ip_type, ip_address, public_ip_address, hostings=0, clients=0, iface=None, mask=None):
		self.ip_type = ip_type
		self.ip_address = ip_address
		self.public_ip_address = public_ip_address if public_ip_address is not None else ip_address
		self.iface = iface
		self.mask = mask

		if hostings is not None:
			self.hostings = hostings
		else:
			self.hostings = 0

		if clients is not None:
			self.clients = clients
		else:
			self.clients = 0

	@property
	def is_completely_free(self):
		"""Check if IP address is completely free - not used by anyone

		:rtype: bool
		"""
		return self.clients == 0 and self.hostings == 0

	def __repr__(self):
		return "IPAddressInfo(ip_type=%r, ip_address=%r, public_ip_address=%r, hostings=%r, clients=%r, iface=%r)" % (
			self.ip_type, self.ip_address, self.public_ip_address, self.hostings, self.clients, self.iface
		)


@cached
def get_database_subscription(panel_server, db_name, db_type, db_host):
	"""Get name of subscription which owns specified database. If database does not exist - return None

	:type panel_server: parallels.core.connections.plesk_server.PleskServer
	:type db_name: basestring
	:type db_type: basestring
	:type db_host: basestring
	:rtype: basestring | None
	"""
	query = """
		SELECT domains.name FROM data_bases
			JOIN DatabaseServers ON data_bases.db_server_id = DatabaseServers.id
			JOIN domains on data_bases.dom_id = domains.id
			WHERE
				data_bases.name = '{db_name}' AND
				DatabaseServers.host= '{db_host}' AND
				DatabaseServers.type = '{db_type}'
	""".format(db_name=db_name, db_type=db_type, db_host=db_host)
	results = _query_plesk_db(panel_server, query)
	if len(results) == 0:
		return None
	elif len(results) == 1:
		return results[0][0]
	else:
		raise MigrationError(
			messages.INCONSISTENCY_DATABASE_MULTIPLE_OWNERS.format(
				server=panel_server.description(), db_name=db_name, db_type=db_type, db_host=db_host,
				webspaces=format_list(results)
			)
		)

@cached
def get_database_user_subscription(panel_server, user_name, db_type, db_host):
	"""Get name of subscription which owns specified database user. If database user does not exist - return None.

	:type panel_server: parallels.core.connections.plesk_server.PleskServer
	:type user_name: basestring
	:type db_type: basestring
	:type db_host: basestring
	:rtype: basestring | None
	"""
	query = """
		SELECT domains.name FROM db_users
			JOIN DatabaseServers ON db_users.db_server_id = DatabaseServers.id
			JOIN domains on db_users.dom_id = domains.id
			WHERE
				db_users.login = '{user_name}' AND
				DatabaseServers.host= '{db_host}' AND
				DatabaseServers.type = '{db_type}'
	""".format(user_name=user_name, db_type=db_type, db_host=db_host)
	results = _query_plesk_db(panel_server, query)
	if len(results) == 0:
		return None
	elif len(results) == 1:
		return results[0][0]
	else:
		raise MigrationError(
			messages.INCONSISTENCY_DB_USER_MULTIPLE_OWNERS.format(
				server=panel_server.description(), db_user=user_name, db_type=db_type, db_host=db_host,
				webspaces=format_list(results)
			)
		)


def suspend_subscription(server, subscription_idn_name):
	"""
	:type server: parallels.core.connections.plesk_server.PleskServer |
	parallels.core.connections.source_server.SourceServer
	"""
	runner = PleskCLIRunnerCLI(server)
	runner.run('subscription', ['--webspace-off', subscription_idn_name])


class PHPHandler(Entity):
	def __init__(self, handler_id, version, full_version, handler_type):
		self._handler_id = handler_id
		self._version = version
		self._handler_type = handler_type
		self._full_version = full_version

	@property
	def handler_id(self):
		"""Plesk PHP handler ID. Examples: 'plesk-php53-fastcgi', 'plesk-php55-fpm', 'cgi', etc.
		:rtype: str
		"""
		return self._handler_id

	@property
	def version(self):
		"""Short version, as a string. Examples: '5', '5.3', '5.4'

		:rtype: str | None
		"""
		return self._version

	@property
	def full_version(self):
		"""Full version, as a string. Examples: '5.3.29', '5.4.44', '5.5.29'
		:rtype: str | None
		"""
		return self._full_version

	@property
	def handler_type(self):
		"""Plesk PHP handler type (mode in which Plesk PHP is running). Allowed value: 'cgi', 'fastcgi', 'module', 'fpm'

		:rtype: str
		"""
		return self._handler_type

	def pretty_str(self):
		"""User-friendly string representation of this object. Use in text messages provided to customer.

		:rtype: unicode
		"""
		if self.handler_id in ('cgi', 'fastcgi', 'module'):
			title = messages.PHP_HANLDER_TYPE_DEFAULT_OS
		elif self.handler_id.startswith('plesk-'):
			title = messages.PHP_HANDLER_TYPE_DEFAULT_PLESK
		else:
			title = messages.PHP_HANDLER_TYPE_CUSTOM % self.handler_id

		version = self.full_version
		if is_empty(version):
			version = self.version
		if is_empty(version):
			version = messages.PHP_UNKNOWN_VERSION

		mode = {
			'fastcgi': messages.PHP_MODE_FASTCGI,
			'cgi': messages.PHP_MODE_CGI,
			'fpm': messages.PHP_MODE_FPM,
			'module': messages.PHP_MODE_APACHE_MODULE
		}.get(self.handler_type)
		if mode is None:
			if self.handler_type is None:
				return messages.PHP_MODE_UNKNOWN
			else:
				return messages.PHP_MODE_ID_STRING % self.handler_type

		return messages.PHP_DESCRIPTION.format(
			title=title, version=version, mode=mode
		)


def get_php_handlers(server):
	"""List PHP handlers available on specified Plesk server

	:type server: parallels.core.connections.plesk_server.PleskServer
	:rtype: list[parallels.core.utils.plesk_utils.PHPHandler]
	"""
	runner = PleskCLIRunnerCLI(server)
	raw_handlers = json.loads(runner.run('php_handler', ['--list-json']))
	handlers = []
	for raw_handler in raw_handlers:
		handlers.append(PHPHandler(
			handler_id=raw_handler.get('id'),
			version=raw_handler.get('version'),
			full_version=raw_handler.get('fullVersion'),
			handler_type=raw_handler.get('type')
		))
	return handlers


def get_admin_guid(server):
	"""Get GUID of administrator in Plesk

	:type server: parallels.core.connections.plesk_server.PleskServer
	:rtype: str | unicode | None
	"""
	result_set = _query_plesk_db(server, "SELECT guid FROM clients WHERE login = 'admin'")
	if len(result_set) == 0 or len(result_set[0]) == 0:
		logger.warning(messages.FAILED_TO_GET_ADMIN_GUID)
		return None
	else:
		return result_set[0][0]



def _query_plesk_db(panel_server, query):
	"""
	:type panel_server: parallels.core.connections.plesk_server.PleskServer
	:type query: string
	"""
	if panel_server.is_windows():
		command = ur'%s\admin\bin\dbclient --direct-sql --sql="{query}"' % panel_server.plesk_dir
		query = " ".join([line.strip() for line in query.split()])
	else:
		command = u"mysql --silent --skip-column-names -u admin -p`cat /etc/psa/.psa.shadow` psa -e \"{query}\""

	with panel_server.runner() as runner:
		lines = runner.sh(command.format(query=query)).splitlines()

	if len(lines) == 0:
		return []

	if panel_server.is_windows():
		# the first line is a header line
		data_lines = lines[1:]
	else:
		# column names are skipped with an option, so data starts from the 1st line
		data_lines = lines

	results = []
	for line in data_lines:
		if line.strip() != '':
			line_data = line.strip().split()
			results.append([d.strip() for d in line_data])

	return results


def _call_php_cli(plesk_server, path, args):
	with plesk_server.runner() as runner:
		return runner.sh(_get_php_cli_command(plesk_server, path, args))


def _call_php_cli_unchecked(plesk_server, path, args):
	with plesk_server.runner() as runner:
		return runner.sh_unchecked(_get_php_cli_command(plesk_server, path, args))


def _get_php_cli_command(plesk_server, path, args):
	if plesk_server.is_windows():
		command_pattern = cmd_command(
			ur'"{plesk_dir}\admin\bin\php" -dauto_prepend_file="" '
			ur'"{plesk_dir}\admin\plib\{cli_path}" {args_string}'
		)
	else:
		command_pattern = ur'{plesk_dir}/bin/sw-engine-pleskrun {plesk_dir}/admin/plib/{cli_path} {args_string}'
	return command_pattern.format(
		plesk_dir=plesk_server.plesk_dir,
		cli_path=path,
		args_string=' '.join(args)
	)
