from parallels.common import messages
from contextlib import contextmanager
import logging
import threading
import Queue
from parallels.common import MigrationError, MigrationNoContextError

from parallels.common.actions.base.action_pointer import ActionPointer
from parallels.common.actions.base.compound_action import CompoundAction
from parallels.common.actions.base.condition_action_pointer import ConditionActionPointer
from parallels.common.actions.base.subscription_action import SubscriptionAction
from parallels.common.logging_context import subscription_context
from parallels.common.migrator_config import MultithreadingStatus
from parallels.common.utils.migration_progress import SubscriptionMigrationStatus
from parallels.common.utils.steps_profiler import get_default_steps_profiler
from parallels.utils import ilen, all_equal

logger = logging.getLogger(__name__)


class ActionRunnerBySubscription(object):
	"""Executes workflow actions and entry points by subscription

	"By subscription" means that:
	for the 1st subscription all actions are executed
	for the 2nd subscription all actions are executed
	and so on.

	As we have actions that are executed for all subscriptions at bulk,
	we can't have true "by subscription" runner, so this runner works by subscriptions
	only when possible, for such bulk actions it works by layers.
	"""
	def __init__(self, global_context, multithreading):
		"""Class constructor

		:type global_context: parallels.common.global_context.GlobalMigrationContext
		:type multithreading: parallels.common.migrator_config.MultithreadingParams
		"""
		self._global_context = global_context
		self._multithreading = multithreading
		self._executed_common_paths = list()
		self._keyboard_interrupt = False

	def run_entry_point(self, entry_point):
		"""Run specified entry point. Execute shutdown actions.

		:type entry_point: parallels.common.actions.base.entry_point_action.EntryPointAction
		:rtype: None
		"""
		try:
			self.run(entry_point)
		finally:
			self._perform_shutdown(entry_point)

	def run(self, action, action_id=None):
		"""Run specified action

		:type action: parallels.common.actions.base.base_action.BaseAction
		:type action_id: basestring
		:rtype: None
		"""
		# 1) Make flat actions list out of actions tree
		flat_actions = self._make_flat_action_list(action_id, action)
		# 2) Group actions in a list into 2 kind of blocks: subscriptions blocks and common blocks
		blocks = self._split_actions_by_blocks(flat_actions)

		for i, block in enumerate(blocks):
			# 3) For each block create a tree, consisting only of block items
			actions_tree = self._build_tree(block)

			# 4) Run block with corresponding run function
			if isinstance(block, ByLayerActionFlatBlock):
				logger.debug("Enter common action block")
				self._run_common_actions_tree(actions_tree)
				logger.debug("Exit common action block")
			elif isinstance(block, BySubscriptionActionFlatBlock):
				logger.debug(messages.ENTER_SUBSCRIPTION_ACTION_BLOCK)
				self._run_subscription_actions(actions_tree, None)
				logger.debug(messages.EXIT_SUBSCRIPTION_ACTION_BLOCK)
			else:
				assert False

	def _run_common_actions_tree(self, actions_tree, path=None):
		"""
		:type actions_tree: parallels.common.actions.base.compound_action.CompoundAction
		:type path: Iterable[basestring] | None
		:rtype: None
		"""
		if path is None:
			path = []

		for action_id, action in actions_tree.get_all_actions():
			with self._profile_action(action_id, action):
				with get_default_steps_profiler().measure_time_gap('Common', action.get_description()):
					action_path = path + [action_id]
					if isinstance(action, CompoundAction):
						self._log_action_start(action)
						self._run_common_actions_tree(action, action_path)
						self._log_action_end(action)
					elif isinstance(action, SubscriptionAction):
						subscriptions = list(self._global_context.iter_all_subscriptions())
						for num, subscription in enumerate(subscriptions, start=1):
							with subscription_context(subscription.name):
								self._run_subscription_action(subscription, action)
					else:
						self._run_common_action_plain(action)

					self._mark_common_path_as_executed(action_path)

	def _run_common_action_plain(self, action):
		"""
		:type action: parallels.common.actions.base.common_action.CommonAction
		:rtype: None
		"""

		def run():
			logger.debug(messages.CHECKING_WHETHER_IT_IS_REQUIRED_EXECUTE)
			if not action.filter_action(self._global_context):
				logger.debug(messages.NO_NEED_EXECUTE_ACTION_CONTINUE_NEXT)
				return
			self._log_action_start(action)
			action.run(self._global_context)
			self._log_action_end(action)

		action_failure_message = action.get_failure_message(self._global_context)

		if action.is_critical():
			try:
				run()
			except MigrationNoContextError:
				# "no context" errors should be displayed "as is", with not information
				# about action, stopped migration, etc
				raise
			except Exception as e:
				logger.debug("Exception: ", exc_info=True)
				full_failure_message = messages.S_CAUSE_S_THAT_IS_CRITICAL % (
					action_failure_message, e
				)
				raise MigrationError(full_failure_message)
		else:
			with self._global_context.safe.try_general(action_failure_message, ''):
				run()

	def _mark_common_path_as_executed(self, path):
		"""Mark action as executed, for correct detection of necessary shutdown actions

		:type path: Iterable[basestring]
		:rtype: None
		"""
		self._executed_common_paths.append("/".join(path))

	def _log_action(self, action, level=0):
		"""
		:type action: parallels.common.actions.base.base_action.BaseAction
		:type level: int
		:rtype: None
		"""
		if action.get_description() is not None:
			description = action.get_description()
			logging_function = self._get_log_function(action)
			logging_function(("    " * level) + description)

	def _log_action_end(self, action):
		"""
		:type action: parallels.common.actions.base.base_action.BaseAction
		:rtype: None
		"""
		if action.get_description() is None:
			return

		if action.get_logging_properties().compound:
			log_function = self._get_log_function(action)
			log_function(u"FINISH: %s", action.get_description())

	def _log_action_start(self, action):
		"""
		:type action: parallels.common.actions.base.base_action.BaseAction
		:rtype: None
		"""
		if action.get_description() is None:
			return

		log_function = self._get_log_function(action)
		if action.get_logging_properties().compound:
			log_function(u"")
			log_function(u"START: %s", action.get_description())
		else:
			log_function(action.get_description())

	@staticmethod
	def _get_log_function(action):
		"""
		:type action: parallels.common.actions.base.base_action.BaseAction
		:rtype: (basestring, *list[basestring]) -> None
		"""
		if action.get_logging_properties().info_log:
			log_function = logger.info
		else:
			log_function = logger.debug
		return log_function

	def _flat_subscription_actions(self, action):
		"""
		:type action: parallels.common.actions.base.base_action.BaseAction
		:rtype: list[parallels.common.actions.base.subscription_action.SubscriptionAction]
		"""
		if isinstance(action, CompoundAction):
			result = []
			for _, child_action in action.get_all_actions():
				result += self._flat_subscription_actions(child_action)
			return result
		else:
			return [action]

	def _run_subscription_actions(self, action, action_id):
		"""
		:type action: parallels.common.actions.base.base_action.BaseAction
		:rtype: None
		"""
		leaf_actions = self._flat_subscription_actions(action)

		locks = {}

		for leaf_action in leaf_actions:
			if not leaf_action.get_multithreading_properties().can_use_threads:
				locks[leaf_action] = threading.Lock()

		subscription_queue = Queue.Queue()

		subscription_count = ilen(self._global_context.iter_all_subscriptions())

		for num, subscription in enumerate(self._global_context.iter_all_subscriptions(), start=1):
			self._global_context.migration_progress.get_subscription(subscription.name)
			subscription_queue.put((num, subscription))

		full_multithreading = self._multithreading.status == MultithreadingStatus.FULL
		if full_multithreading or self._is_multithreading_enabled_for_actions(leaf_actions):
			num_workers = self._multithreading.num_workers
		else:
			num_workers = 1

		logger.debug(messages.RUN_GROUP_ACTIONS_S_WORKERS % num_workers)
		plain_log = num_workers > 1

		if plain_log:
			common_parents = self._list_common_parents(action, action_id)
		else:
			common_parents = []

		threads = []

		with self._profile_action_item_list(common_parents):
			for worker_num in range(0, num_workers):
				thread = threading.Thread(
					target=lambda: self._run_subscription_queue_worker(
						subscription_queue, action, action_id, locks, subscription_count, plain_log,
					),
					name='SubscriptionThread%s' % (worker_num + 1)
				)
				# It is not good practice to have daemon threads, but that is required to force
				# interruption of a thread, as there is no safe way to do that (actually there is no way at all).
				thread.daemon = True
				thread.start()
				threads.append(thread)

		def wait_for_threads_to_finish():
			while any(t.is_alive() for t in threads):
				for t in threads:
					if t.is_alive():
						# wait not more than for 1 second, if no timeout is specified
						# then you can't get KeyboardInterrupt (Ctrl+C)
						t.join(1)

		try:
			wait_for_threads_to_finish()
		except KeyboardInterrupt:
			self._keyboard_interrupt = True

			logger.info(
				messages.USER_INTERRUPT_REQUESTED_TRYING_STOP_MIGRATION)

			try:
				# wait for workers to finish once more
				# they should process '_keyboard_interrupt' flag - stop on next action
				wait_for_threads_to_finish()
			except KeyboardInterrupt:
				logger.info(messages.USER_INTERRUPT_REQUESTED_FORCING_STOP_MIGRATION)
				raise

			raise

	def _run_subscription_queue_worker(
		self, subscription_queue, action, action_id, locks, subscription_count, plain_log=False
	):
		"""
		:type subscription_queue: Queue.Queue
		:type action: parallels.common.actions.base.base_action.BaseAction
		:type locks: dict[parallels.common.actions.base.base_action.BaseAction, threading.Lock]
		:type subscription_count: int
		:type plain_log: bool
		:rtype: None
		"""
		while True:
			try:
				subscription_num, subscription = subscription_queue.get(False)
			except Queue.Empty:
				return  # no more subscriptions in a queue

			with subscription_context(subscription.name):
				try:
					subscription_progress = self._global_context.migration_progress.get_subscription(
						subscription.name
					)
					subscription_progress.status = SubscriptionMigrationStatus.IN_PROGRESS
					leaf_actions = self._flat_subscription_actions(action)
					has_info_log_count = len([
						leaf_action for leaf_action in leaf_actions if action.get_logging_properties().info_log
					])
					if has_info_log_count:
						overall_logging_function = logger.info
					else:
						overall_logging_function = logger.debug

					if not plain_log:
						overall_logging_function("")
						overall_logging_function(
							"START Processing subscription '%s' (#%s out of %s)",
							subscription.name, subscription_num, subscription_count
						)

					self._run_action_tree(
						subscription, action, action_id, locks, plain_log=plain_log,
						subscription_num=subscription_num, subscription_count=subscription_count
					)

					if not plain_log:
						overall_logging_function(messages.END_PROCESSING_SUBSCRIPTION_S, subscription.name)
				except Exception as e:
					logger.error(
						messages.INTERNAL_ERROR_IN_SUBSCRIPTION_THREAD_S,
						subscription.name, unicode(e)
					)
					logger.debug(messages.EXCEPTION_IN_SUBSCRIPTION_THREAD, exc_info=True)
				finally:
					subscription_queue.task_done()

	def _run_subscription_action(
		self, subscription, action, lock=None, level=0, plain_log=False,
	):
		"""
		:type subscription: parallels.common.migrated_subscription.MigratedSubscription
		:type action: parallels.common.actions.base.subscription_action.SubscriptionAction
		:type lock: threading.Lock
		:type level: int
		:type plain_log: bool
		:rtype: None
		"""
		if self._global_context.safe.is_failed_subscription(subscription.name):
			# subscription is completely failed, do not execute any actions
			return

		if action.filter_subscription(self._global_context, subscription):
			self._log_action(action, level if not plain_log else 0)
			subscription_progress = self._global_context.migration_progress.get_subscription(
				subscription.name
			)
			subscription_progress.action = action.get_description()
			subscription_progress.status = SubscriptionMigrationStatus.IN_PROGRESS

			if lock is not None:
				lock.acquire()
			try:
				safe = self._global_context.safe
				action_failure_message = action.get_failure_message(
					self._global_context, subscription
				)
				full_failure_message = action_failure_message
				if action.is_critical():
					full_failure_message += (
						messages.MIGRATION_FOR_THAT_SUBSCRIPTION_IS_CONSIDERED)

				with (
					get_default_steps_profiler().measure_time_gap(
						"Subscription '%s'" % subscription.name,
						action.get_description()
					)
				):
					safe.try_subscription_with_rerun(
						lambda: action.run(self._global_context, subscription),
						subscription.name,
						error_message=full_failure_message,
						repeat_error=messages.S_TRY_REPEAT_OPERATION_ONCE_MORE % action_failure_message,
						repeat_count=action.get_repeat_count(),
						is_critical=action.is_critical()
					)
			finally:
				subscription_progress.action = None
				if subscription_progress.status not in SubscriptionMigrationStatus.finished():
					subscription_progress.status = SubscriptionMigrationStatus.ON_HOLD
				if lock is not None:
					lock.release()

	def _run_action_tree(
		self, subscription, action_tree, action_id, locks, level=0, plain_log=False,
		subscription_num=None, subscription_count=None
	):
		"""
		:type subscription: parallels.common.migrated_subscription.MigratedSubscription
		:type locks: dict[parallels.common.actions.base.base_action.BaseAction, threading.Lock]
		:type level: int
		:type plain_log: bool
		:type subscription_num: int
		:type subscription_count: int
		:rtype: None
		"""
		if self._keyboard_interrupt:
			# graceful subscription migration interrupt by user (Ctrl+C) request
			return

		if self._global_context.safe.is_failed_subscription(subscription.name):
			# subscription is completely failed, do not execute any actions
			return

		with self._profile_action(action_id, action_tree, not plain_log):
			if isinstance(action_tree, CompoundAction):
				if not plain_log:
					self._log_action(action_tree, level)
				for child_action_id, child_action in action_tree.get_all_actions():
					self._run_action_tree(
						subscription, child_action, child_action_id, locks, level+1, plain_log=plain_log,
						subscription_num=subscription_num, subscription_count=subscription_count
					)
			elif isinstance(action_tree, SubscriptionAction):
				lock = locks.get(action_tree)
				self._run_subscription_action(
					subscription, action_tree, lock, level, plain_log=plain_log,
				)

	def _make_flat_action_list(self, action_id, action, parent_actions=None):
		"""Make list of "flat" actions out of actions tree.

		By "flat" we mean plain list of leaf actions that should be executed
		(comparing to tree with compound actions).

		Each item of returned list contains:
		- Leaf action ID.
		- Leaf action object.
		- Ordered list of all parent actions, each item has
			- Parent action ID.
			- Parent action object.

		:type action_id: basestring
		:type action: parallels.common.actions.base.base_action.BaseAction
		:type parent_actions: list[parallels.common.workflow.runner.by_subscription.ActionItem] | None
		:rtype: list[parallels.common.workflow.runner.by_subscription.FlatAction]
		"""
		if parent_actions is None:
			parent_actions = []

		action = self._resolve_action_pointers(action)

		if isinstance(action, CompoundAction):
			# compound action with child actions
			return sum([
				self._make_flat_action_list(
					child_action_id, child_action, parent_actions + [ActionItem(action_id, action)]
				)
				for child_action_id, child_action in action.get_all_actions()
			], [])
		else:
			# leaf action
			return [FlatAction(ActionItem(action_id, action), parent_actions)]

	@staticmethod
	def _split_actions_by_blocks(flat_actions):
		"""Split flat actions list by 'subscription' and 'common' action groups

		'subscription' action group consists of actions that are executed per-subscription
		(parallels.common.actions.base.subscription_action.SubscriptionAction)
		'common' action group consists of actions that are executed overall, per-migration
		(parallels.common.actions.base.common_action.CommonAction)

		Returns ordered (according to workflow execution sequence) list of action groups.

		:type flat_actions: list[parallels.common.workflow.runner.by_subscription.FlatAction]
		:rtype: list[
			parallels.common.workflow.runner.by_subscription.SubscriptionActionFlatBlock|
			parallels.common.workflow.runner.by_subscription.CommonActionFlatBlock
		]
		"""
		# either SubscriptionActionFlatBlock.BLOCK_TYPE or CommonActionFlatBlock.BLOCK_TYPE
		previous_action_type = None

		blocks = []
		current_block = []

		for flat_action in flat_actions:
			is_by_subscription_action = (
				isinstance(flat_action.action_item.action, SubscriptionAction)
				and
				all([parent.action.run_by_subscription for parent in flat_action.parent_action_items])
			)
			if is_by_subscription_action:
				action_type = BySubscriptionActionFlatBlock.BLOCK_TYPE
			else:
				action_type = ByLayerActionFlatBlock.BLOCK_TYPE

			if action_type == previous_action_type:
				# type is the same, add to already created block
				current_block.flat_actions.append(flat_action)
			else:
				# start new block
				if is_by_subscription_action:
					current_block = BySubscriptionActionFlatBlock()
				else:
					current_block = ByLayerActionFlatBlock()
				current_block.flat_actions.append(flat_action)
				blocks.append(current_block)

			previous_action_type = action_type

		return blocks

	@staticmethod
	def _build_tree(flat_block):
		"""Build action tree out of flat actions block

		:type flat_block: parallels.common.workflow.runner.by_subscription.BaseActionFlatBlock
		:rtype parallels.common.actions.base.compound_action.CompoundAction
		"""
		root_action = CompoundAction()

		all_parent_actions = set([
			parent_action_item.action
			for flat_action in flat_block.flat_actions
			for parent_action_item in flat_action.parent_action_items
		])
		old_action_to_new_action = dict()

		# foreach compound action in parents list - create new compound action with same description and properties,
		# but empty list of child actions
		for parent_action in all_parent_actions:
			new_action = CompoundAction(
				parent_action.get_description(),
				parent_action.get_logging_properties(),
				parent_action.run_by_subscription
			)
			old_action_to_new_action[parent_action] = new_action

		restored_relations = set([])
		for flat_action in flat_block.flat_actions:
			current_new_parent_action = root_action
			current_parent_action = None

			# 1) Restore tree structure of compound actions
			for parent_action_item in flat_action.parent_action_items[1:]:
				old_action = parent_action_item.action
				new_action = old_action_to_new_action[old_action]

				if (current_parent_action, old_action) not in restored_relations:
					current_new_parent_action.insert_action(parent_action_item.action_id, new_action)
					restored_relations.add((current_parent_action, old_action))

				current_parent_action = old_action
				current_new_parent_action = new_action

			# 2) Insert leaf actions into the new tree
			current_new_parent_action.insert_action(
				flat_action.action_item.action_id, flat_action.action_item.action
			)

		return root_action

	def _list_common_parents(self, action, action_id):
		"""Take an action tree and find out the maximum common parents list

		Return list of parent action items (each action item is action id plus action object)
		from the most outer to the inner ones. Each of an item is a parent of
		each leaf action in the tree.

		:type action: parallels.common.actions.base.base_action.BaseAction
		:type action_id: basestring
		:rtype: list[parallels.common.workflow.runner.by_subscription.ActionItem]
		"""
		common_parents = []

		flat_actions_list = self._make_flat_action_list(action_id, action)
		if len(flat_actions_list) > 0:
			min_parents_list_length = min([
				len(flat_action.parent_action_items)
				for flat_action in flat_actions_list
			])
			for i in range(1, min_parents_list_length):
				parent_action_items = [
					flat_action.parent_action_items[i]
					for flat_action in flat_actions_list
				]
				if all_equal(parent_action_items):
					common_parents.append(parent_action_items[0])
				else:
					break

		return common_parents

	def _resolve_action_pointers(self, action):
		"""If specified action is action pointer - resolve it, otherwise return "as-is"

		:type action: parallels.common.actions.base.base_action.BaseAction
		:rtype: parallels.common.actions.base.base_action.BaseAction
		"""
		if isinstance(action, ActionPointer):
			action = action.resolve()
		elif isinstance(action, ConditionActionPointer):
			action = action.resolve(self._global_context)
		return action

	def _is_multithreading_enabled_for_actions(self, actions):
		"""Check if multithreading could be used for a group of subscription actions

		Multithreading is enabled for a group, if it is enabled for all actions in a group

		:type actions: list[parallels.common.actions.base.subscription_action.SubscriptionAction]
		:rtype: bool
		"""
		for action in actions:
			if not self._is_multithreading_enabled(action):
				return False

		return True

	def _is_multithreading_enabled(self, action):
		"""Check if multithreading could be used for a single subscription action

		:type action: parallels.common.actions.base.subscription_action.SubscriptionAction
		:rtype: bool
		"""
		action_threading_props = action.get_multithreading_properties()

		if self._multithreading.status == MultithreadingStatus.DISABLED:
			multithreading_enabled = False
		elif self._multithreading.status == MultithreadingStatus.DEFAULT:
			multithreading_enabled = action_threading_props.use_threads_by_default
		elif self._multithreading.status == MultithreadingStatus.FULL:
			multithreading_enabled = action_threading_props.can_use_threads
		else:
			assert False

		return multithreading_enabled

	@staticmethod
	@contextmanager
	def _profile_action(action_id, action, profiling_enabled=True):
		"""Measure time of an action passed

		This function considers that for all parent actions this function was already called, and we're
		already in that context.

		:type action: parallels.common.actions.base.base_action.BaseAction
		:type action_id: basestring
		:type profiling_enabled: boolean
		:return:
		"""
		if action.get_description() is not None and profiling_enabled:
			with get_default_steps_profiler().measure_time(
				action_id, action.get_description()
			):
				yield
		else:
			# 1) Compound actions that have no descriptions are profiled by all child actions.
			# 2) Other actions that have no descriptions are measured as "Other" action
			# 3) Explicitly disabled profiling
			yield

	@contextmanager
	def _profile_action_item_list(self, action_item_list):
		"""Measure time of an action passed as list parent actions

		Action item list is a list of all parent action, plus an action
		for which we measure time itself. The list is ordered - the outer actions are at the beginning,
		the inner are at the end

		:type action_item_list: list[parallels.common.workflow.runner.by_subscription.ActionItem]
		"""
		if len(action_item_list) == 0:
			yield
		else:
			first_action_item = action_item_list[0]
			action_items_list_tail = action_item_list[1:]
			action_description = first_action_item.action.get_description()
			if action_description is not None:
				with get_default_steps_profiler().measure_time(
					first_action_item.action_id,
				):
					with self._profile_action_item_list(action_items_list_tail):
						yield
			else:
				with self._profile_action_item_list(action_items_list_tail):
					yield

	def _perform_shutdown(self, entry_point):
		"""Run shutdown actions for specified entry point, considering already executed actions

		:type entry_point: parallels.common.actions.base.entry_point_action.EntryPointAction
		:rtype: None
		"""
		# keep shutdown actions ordered, don't use unordered structures like built-in dict() or set()
		shutdown_paths = []

		def add_shutdown_path(new_shutdown_path):
			if (
				# schedule run each shutdown action only once
				new_shutdown_path not in shutdown_paths
				and
				# run it only if it was not executed before
				new_shutdown_path not in self._executed_common_paths
			):
				shutdown_paths.append(new_shutdown_path)

		# add shutdown actions for each set-up action
		all_shutdown_actions = entry_point.get_shutdown_actions()
		for executed_path in self._executed_common_paths:
			for shutdown_path in all_shutdown_actions.get(executed_path, []):
				add_shutdown_path(shutdown_path)

		# add overall shutdown actions
		for shutdown_path in all_shutdown_actions.get(None, []):
			add_shutdown_path(shutdown_path)

		# execute selected shutdown actions
		for shutdown_path in shutdown_paths:
			logger.debug("Execute shutdown action '%s'", shutdown_path)
			self.run(entry_point.get_path(shutdown_path), shutdown_path)


class BaseActionFlatBlock(object):
	def __init__(self, flat_actions=None):
		if flat_actions is None:
			flat_actions = []
		self.flat_actions = flat_actions

	def __eq__(self, other):
		return other.flat_actions == self.flat_actions


class BySubscriptionActionFlatBlock(BaseActionFlatBlock):
	BLOCK_TYPE = 'subscription'


class ByLayerActionFlatBlock(BaseActionFlatBlock):
	BLOCK_TYPE = 'common'


class FlatAction(object):
	def __init__(self, action_item, parent_action_items):
		self.action_item = action_item
		self.parent_action_items = parent_action_items


class ActionItem(object):
	def __init__(self, action_id, action):
		self.action_id = action_id
		self.action = action

	def __eq__(self, other):
		return other.action == self.action and other.action_id == self.action_id
