# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.

"""DolphinScheduler Task and TaskRelation object."""

from __future__ import annotations

import copy
import types
import warnings
from collections.abc import Sequence
from datetime import timedelta
from logging import getLogger

from pydolphinscheduler import configuration
from pydolphinscheduler.constants import (
    Delimiter,
    IsCache,
    ResourceKey,
    Symbol,
    TaskFlag,
    TaskPriority,
    TaskTimeoutFlag,
)
from pydolphinscheduler.core.parameter import BaseDataType, Direction, ParameterHelper
from pydolphinscheduler.core.resource import Resource
from pydolphinscheduler.core.resource_plugin import ResourcePlugin
from pydolphinscheduler.core.workflow import Workflow, WorkflowContext
from pydolphinscheduler.exceptions import PyDSParamException, PyResPluginException
from pydolphinscheduler.java_gateway import gateway
from pydolphinscheduler.models import Base
from pydolphinscheduler.utils.date import timedelta2timeout

logger = getLogger(__name__)


class TaskRelation(Base):
    """TaskRelation object, describe the relation of exactly two tasks."""

    # Add attr `_KEY_ATTR` to overwrite :func:`__eq__`, it is make set
    # `Task.workflow._task_relations` work correctly.
    _KEY_ATTR = {
        "pre_task_code",
        "post_task_code",
    }

    _DEFINE_ATTR = {
        "pre_task_code",
        "post_task_code",
    }

    _DEFAULT_ATTR = {
        "name": "",
        "preTaskVersion": 1,
        "postTaskVersion": 1,
        "conditionType": 0,
        "conditionParams": {},
    }

    def __init__(
        self,
        pre_task_code: int,
        post_task_code: int,
        name: str | None = None,
    ):
        super().__init__(name)
        self.pre_task_code = pre_task_code
        self.post_task_code = post_task_code

    def __hash__(self):
        return hash(f"{self.pre_task_code} {Delimiter.DIRECTION} {self.post_task_code}")


class Task(Base):
    """Task object, parent class for all exactly task type.

    :param name: The name of the task. Node names within the same workflow must be unique.
    :param task_type:
    :param description: default None
    :param flag: default TaskFlag.YES,
    :param task_priority: default TaskPriority.MEDIUM
    :param worker_group: default configuration.WORKFLOW_WORKER_GROUP
    :param environment_name: default None
    :param task_group_id: Identify of task group to restrict the parallelism of tasks instance run, default 0.
    :param task_group_priority: Priority for same task group to, the higher the value, the higher the
        priority, default 0.
    :param delay_time: deault 0
    :param fail_retry_times: default 0
    :param fail_retry_interval: default 1
    :param timeout_notify_strategy: default, None
    :param timeout: Timeout attribute for task, in minutes. Task is consider as  timed out task when the
        running time of a task exceeds than this value. when data type is :class:`datetime.timedelta` will
        be converted to int(in minutes). default ``None``
    :param resource_list: default None
    :param wait_start_timeout: default None
    :param condition_result: default None,
    :param resource_plugin: default None
    :param is_cache: default False
    :param input_params: default None, input parameters, {param_name: param_value}
    :param output_params: default None, input parameters, {param_name: param_value}
    """

    _DEFINE_ATTR = {
        "name",
        "code",
        "version",
        "task_type",
        "task_params",
        "description",
        "flag",
        "task_priority",
        "worker_group",
        "environment_code",
        "delay_time",
        "fail_retry_times",
        "fail_retry_interval",
        "task_group_id",
        "task_group_priority",
        "timeout_flag",
        "timeout_notify_strategy",
        "timeout",
        "is_cache",
    }

    # task default attribute will into `task_params` property
    _task_default_attr = {
        "local_params",
        "resource_list",
        "dependence",
        "wait_start_timeout",
        "condition_result",
    }
    # task attribute ignore from _task_default_attr and will not into `task_params` property
    _task_ignore_attr: set = set()
    # task custom attribute define in sub class and will append to `task_params` property
    _task_custom_attr: set = set()

    ext: set = None
    ext_attr: str | types.FunctionType = None

    DEFAULT_CONDITION_RESULT = {"successNode": [""], "failedNode": [""]}

    def __init__(
        self,
        name: str,
        task_type: str,
        description: str | None = None,
        flag: str | None = TaskFlag.YES,
        task_priority: str | None = TaskPriority.MEDIUM,
        worker_group: str | None = configuration.WORKFLOW_WORKER_GROUP,
        environment_name: str | None = None,
        task_group_id: int | None = 0,
        task_group_priority: int | None = 0,
        delay_time: int | None = 0,
        fail_retry_times: int | None = 0,
        fail_retry_interval: int | None = 1,
        timeout_notify_strategy: str | None = None,
        timeout: timedelta | int | None = None,
        workflow: Workflow | None = None,
        resource_list: list | None = None,
        dependence: dict | None = None,
        wait_start_timeout: dict | None = None,
        condition_result: dict | None = None,
        resource_plugin: ResourcePlugin | None = None,
        is_cache: bool | None = False,
        input_params: dict | None = None,
        output_params: dict | None = None,
        *args,
        **kwargs,
    ):
        super().__init__(name, description)
        self.task_type = task_type
        self.flag = flag
        self._is_cache = is_cache
        self.task_priority = task_priority
        self.worker_group = worker_group
        self._environment_name = environment_name
        self.task_group_id = task_group_id
        self.task_group_priority = task_group_priority
        self.fail_retry_times = fail_retry_times
        self.fail_retry_interval = fail_retry_interval
        self.delay_time = delay_time
        self.timeout_notify_strategy = timeout_notify_strategy
        self._timeout: timedelta | int = timeout
        self._workflow = None
        self._input_params = input_params or {}
        self._output_params = output_params or {}
        if "process_definition" in kwargs:
            warnings.warn(
                "The `process_definition` parameter is deprecated, please use `workflow` instead.",
                DeprecationWarning,
            )
            self.workflow = kwargs.pop("process_definition")
        else:
            self.workflow: Workflow = workflow or WorkflowContext.get()

        if "local_params" in kwargs:
            warnings.warn(
                """The `local_params` parameter is deprecated,
                please use `input_params` and `output_params` instead.
                """,
                DeprecationWarning,
            )
            self._local_params = kwargs.get("local_params")

        self._upstream_task_codes: set[int] = set()
        self._downstream_task_codes: set[int] = set()
        self._task_relation: set[TaskRelation] = set()
        # move attribute code and version after _workflow and workflow declare
        self.code, self.version = self.gen_code_and_version()
        # Add task to workflow, maybe we could put into property workflow latter

        if self.workflow is not None and self.code not in self.workflow.tasks:
            self.workflow.add_task(self)
        else:
            logger.warning(
                "Task code %d already in workflow, prohibit re-add task.",
                self.code,
            )

        # Attribute for task param
        self._resource_list = resource_list or []
        self.dependence = dependence or {}
        self.wait_start_timeout = wait_start_timeout or {}
        self._condition_result = condition_result or self.DEFAULT_CONDITION_RESULT
        self.resource_plugin = resource_plugin
        self.get_content()

    @property
    def workflow(self) -> Workflow | None:
        """Get attribute workflow."""
        return self._workflow

    @workflow.setter
    def workflow(self, workflow: Workflow | None):
        """Set attribute workflow."""
        self._workflow = workflow

    @property
    def timeout(self) -> int:
        """Get attribute timeout."""
        if isinstance(self._timeout, int):
            if self._timeout < 0:
                raise PyDSParamException("The timeout value must be greater than 0")
            return self._timeout
        return timedelta2timeout(self._timeout) if self._timeout else 0

    @property
    def timeout_flag(self) -> str:
        """Whether the timeout attribute is being set or not."""
        return TaskTimeoutFlag.ON if self._timeout else TaskTimeoutFlag.OFF

    @property
    def is_cache(self) -> str:
        """Whether the cache is being set or not."""
        if isinstance(self._is_cache, bool):
            return IsCache.YES if self._is_cache else IsCache.NO
        else:
            raise PyDSParamException("is_cache must be a bool")

    @property
    def resource_list(self) -> list[dict[str, Resource]]:
        """Get task define attribute `resource_list`."""
        resources = set()
        for res in self._resource_list:
            if isinstance(res, str):
                resources.add(
                    Resource(
                        name=res, user_name=self.user_name
                    ).get_fullname_from_database()
                )
            elif isinstance(res, dict) and ResourceKey.NAME in res:
                warnings.warn(
                    """`resource_list` should be defined using List[str] with resource paths,
                       the use of ids to define resources will be remove in version 3.2.0.
                    """,
                    DeprecationWarning,
                    stacklevel=2,
                )
                resources.add(res.get(ResourceKey.NAME))
        return [{ResourceKey.NAME: r} for r in resources]

    @property
    def user_name(self) -> str | None:
        """Return username of workflow."""
        if self.workflow:
            return self.workflow.user.name
        else:
            raise PyDSParamException("`user_name` cannot be empty.")

    @property
    def condition_result(self) -> dict:
        """Get attribute condition_result."""
        return self._condition_result

    @condition_result.setter
    def condition_result(self, condition_result: dict | None):
        """Set attribute condition_result."""
        self._condition_result = condition_result

    def _get_attr(self) -> set[str]:
        """Get final task task_params attribute.

        Base on `_task_default_attr`, append attribute from `_task_custom_attr` and subtract attribute from
        `_task_ignore_attr`.
        """
        attr = copy.deepcopy(self._task_default_attr)
        attr -= self._task_ignore_attr
        attr |= self._task_custom_attr
        return attr

    @property
    def task_params(self) -> dict | None:
        """Get task parameter object.

        Will get result to combine _task_custom_attr and custom_attr.
        """
        custom_attr = self._get_attr()
        return self.get_define_custom(custom_attr=custom_attr)

    def get_plugin(self):
        """Return the resource plug-in.

        according to parameter resource_plugin and parameter
        workflow.resource_plugin.
        """
        if self.resource_plugin is None:
            if self.workflow.resource_plugin is not None:
                return self.workflow.resource_plugin
            else:
                raise PyResPluginException(
                    "The execution command of this task is a file, but the resource plugin is empty"
                )
        else:
            return self.resource_plugin

    def get_content(self):
        """Get the file content according to the resource plugin."""
        if self.ext_attr is None and self.ext is None:
            return
        _ext_attr = getattr(self, self.ext_attr)
        if _ext_attr is not None:
            if isinstance(_ext_attr, str) and _ext_attr.endswith(tuple(self.ext)):
                res = self.get_plugin()
                content = res.read_file(_ext_attr)
                setattr(self, self.ext_attr.lstrip(Symbol.UNDERLINE), content)
            else:
                if self.resource_plugin is not None or (
                    self.workflow is not None
                    and self.workflow.resource_plugin is not None
                ):
                    index = _ext_attr.rfind(Symbol.POINT)
                    if index != -1:
                        raise ValueError(
                            f"This task does not support files with suffix {_ext_attr[index:]},"
                            f"only supports {Symbol.COMMA.join(str(suf) for suf in self.ext)}"
                        )
                setattr(self, self.ext_attr.lstrip(Symbol.UNDERLINE), _ext_attr)

    def __hash__(self):
        return hash(self.code)

    def __lshift__(self, other: Task | Sequence[Task]):
        """Implement Task << Task."""
        self.set_upstream(other)
        return other

    def __rshift__(self, other: Task | Sequence[Task]):
        """Implement Task >> Task."""
        self.set_downstream(other)
        return other

    def __rrshift__(self, other: Task | Sequence[Task]):
        """Call for Task >> [Task] because list don't have __rshift__ operators."""
        self.__lshift__(other)
        return self

    def __rlshift__(self, other: Task | Sequence[Task]):
        """Call for Task << [Task] because list don't have __lshift__ operators."""
        self.__rshift__(other)
        return self

    def _set_deps(self, tasks: Task | Sequence[Task], upstream: bool = True) -> None:
        """Set parameter tasks dependent to current task.

        it is a wrapper for :func:`set_upstream` and :func:`set_downstream`.
        """
        if not isinstance(tasks, Sequence):
            tasks = [tasks]

        for task in tasks:
            if upstream:
                self._upstream_task_codes.add(task.code)
                task._downstream_task_codes.add(self.code)

                if self._workflow:
                    task_relation = TaskRelation(
                        pre_task_code=task.code,
                        post_task_code=self.code,
                        name=f"{task.name} {Delimiter.DIRECTION} {self.name}",
                    )
                    self.workflow._task_relations.add(task_relation)
            else:
                self._downstream_task_codes.add(task.code)
                task._upstream_task_codes.add(self.code)

                if self._workflow:
                    task_relation = TaskRelation(
                        pre_task_code=self.code,
                        post_task_code=task.code,
                        name=f"{self.name} {Delimiter.DIRECTION} {task.name}",
                    )
                    self.workflow._task_relations.add(task_relation)

    def set_upstream(self, tasks: Task | Sequence[Task]) -> None:
        """Set parameter tasks as upstream to current task."""
        self._set_deps(tasks, upstream=True)

    def set_downstream(self, tasks: Task | Sequence[Task]) -> None:
        """Set parameter tasks as downstream to current task."""
        self._set_deps(tasks, upstream=False)

    # TODO code should better generate in bulk mode when :ref: workflow run submit or start
    def gen_code_and_version(self) -> tuple:
        """Generate task code and version from java gateway.

        If task name do not exists in workflow before, if will generate new code and version id
        equal to 0 by java gateway, otherwise if will return the exists code and version.
        """
        # TODO get code from specific project workflow and task name
        result = gateway.get_code_and_version(
            self.workflow._project, self.workflow.name, self.name
        )
        # result = gateway.entry_point.genTaskCodeList(DefaultTaskCodeNum.DEFAULT)
        # gateway_result_checker(result)
        return result.get("code"), result.get("version")

    @property
    def environment_code(self) -> str:
        """Convert environment name to code."""
        if self._environment_name is None:
            return None
        return gateway.query_environment_info(self._environment_name)

    @property
    def local_params(self):
        """Convert local params."""
        local_params = (
            copy.deepcopy(self._local_params) if hasattr(self, "_local_params") else []
        )
        local_params.extend(
            ParameterHelper.convert_params(self._input_params, Direction.IN)
        )
        local_params.extend(
            ParameterHelper.convert_params(self._output_params, Direction.OUT)
        )
        return local_params

    def add_in(
        self,
        name: str,
        value: int | str | float | bool | BaseDataType | None = None,
    ):
        """Add input parameters.

        :param name: name of the input parameter.
        :param value: value of the input parameter.

        It could be simply command::

            task.add_in("a")
            task.add_in("b", 123)
            task.add_in("c", bool)
            task.add_in("d", ParameterType.LONG(123))

        """
        self._input_params[name] = value

    def add_out(
        self,
        name: str,
        value: int | str | float | bool | BaseDataType | None = None,
    ):
        """Add output parameters.

        :param name: name of the output parameter.
        :param value: value of the output parameter.

        It could be simply command::

            task.add_out("a")
            task.add_out("b", 123)
            task.add_out("c", bool)
            task.add_out("d", ParameterType.LONG(123))

        """
        self._output_params[name] = value


class BatchTask(Task):
    """Task object, parent class for all exactly task type.

    :param name: The name of the task. Node names within the same workflow must be unique.
    :param task_type:
    :param description: default None
    :param flag: default TaskFlag.YES,
    :param task_priority: default TaskPriority.MEDIUM
    :param worker_group: default configuration.WORKFLOW_WORKER_GROUP
    :param environment_name: default None
    :param task_group_id: Identify of task group to restrict the parallelism of tasks instance run, default 0.
    :param task_group_priority: Priority for same task group to, the higher the value, the higher the
        priority, default 0.
    :param delay_time: deault 0
    :param fail_retry_times: default 0
    :param fail_retry_interval: default 1
    :param timeout_notify_strategy: default, None
    :param timeout: Timeout attribute for task, in minutes. Task is consider as  timed out task when the
        running time of a task exceeds than this value. when data type is :class:`datetime.timedelta` will
        be converted to int(in minutes). default ``None``
    :param resource_list: default None
    :param wait_start_timeout: default None
    :param condition_result: default None,
    :param resource_plugin: default None
    :param is_cache: default False
    :param input_params: default None, input parameters, {param_name: param_value}
    :param output_params: default None, input parameters, {param_name: param_value}
    """

    _DEFINE_ATTR = Task._DEFINE_ATTR | {"task_execute_type"}

    def __init__(
        self,
        name: str,
        task_type: str,
        description: str | None = None,
        flag: str | None = TaskFlag.YES,
        task_priority: str | None = TaskPriority.MEDIUM,
        worker_group: str | None = configuration.WORKFLOW_WORKER_GROUP,
        environment_name: str | None = None,
        task_group_id: int | None = 0,
        task_group_priority: int | None = 0,
        delay_time: int | None = 0,
        fail_retry_times: int | None = 0,
        fail_retry_interval: int | None = 1,
        timeout_notify_strategy: str | None = None,
        timeout: timedelta | int | None = None,
        workflow: Workflow | None = None,
        resource_list: list | None = None,
        dependence: dict | None = None,
        wait_start_timeout: dict | None = None,
        condition_result: dict | None = None,
        resource_plugin: ResourcePlugin | None = None,
        is_cache: bool | None = False,
        input_params: dict | None = None,
        output_params: dict | None = None,
        *args,
        **kwargs,
    ):
        super().__init__(
            name,
            task_type,
            description,
            flag,
            task_priority,
            worker_group,
            environment_name,
            task_group_id,
            task_group_priority,
            delay_time,
            fail_retry_times,
            fail_retry_interval,
            timeout_notify_strategy,
            timeout,
            workflow,
            resource_list,
            dependence,
            wait_start_timeout,
            condition_result,
            resource_plugin,
            is_cache,
            input_params,
            output_params,
            *args,
            **kwargs,
        )
        self.task_execute_type = "BATCH"


class StreamTask(Task):
    """Task object, parent class for all exactly task type.

    :param name: The name of the task. Node names within the same workflow must be unique.
    :param task_type:
    :param description: default None
    :param flag: default TaskFlag.YES,
    :param task_priority: default TaskPriority.MEDIUM
    :param worker_group: default configuration.WORKFLOW_WORKER_GROUP
    :param environment_name: default None
    :param task_group_id: Identify of task group to restrict the parallelism of tasks instance run, default 0.
    :param task_group_priority: Priority for same task group to, the higher the value, the higher the
        priority, default 0.
    :param delay_time: deault 0
    :param fail_retry_times: default 0
    :param fail_retry_interval: default 1
    :param timeout_notify_strategy: default, None
    :param timeout: Timeout attribute for task, in minutes. Task is consider as  timed out task when the
        running time of a task exceeds than this value. when data type is :class:`datetime.timedelta` will
        be converted to int(in minutes). default ``None``
    :param resource_list: default None
    :param wait_start_timeout: default None
    :param condition_result: default None,
    :param resource_plugin: default None
    :param is_cache: default False
    :param input_params: default None, input parameters, {param_name: param_value}
    :param output_params: default None, input parameters, {param_name: param_value}
    """

    _DEFINE_ATTR = Task._DEFINE_ATTR | {"task_execute_type"}

    def __init__(
        self,
        name: str,
        task_type: str,
        description: str | None = None,
        flag: str | None = TaskFlag.YES,
        task_priority: str | None = TaskPriority.MEDIUM,
        worker_group: str | None = configuration.WORKFLOW_WORKER_GROUP,
        environment_name: str | None = None,
        task_group_id: int | None = 0,
        task_group_priority: int | None = 0,
        delay_time: int | None = 0,
        fail_retry_times: int | None = 0,
        fail_retry_interval: int | None = 1,
        timeout_notify_strategy: str | None = None,
        timeout: timedelta | int | None = None,
        workflow: Workflow | None = None,
        resource_list: list | None = None,
        dependence: dict | None = None,
        wait_start_timeout: dict | None = None,
        condition_result: dict | None = None,
        resource_plugin: ResourcePlugin | None = None,
        is_cache: bool | None = False,
        input_params: dict | None = None,
        output_params: dict | None = None,
        *args,
        **kwargs,
    ):
        super().__init__(
            name,
            task_type,
            description,
            flag,
            task_priority,
            worker_group,
            environment_name,
            task_group_id,
            task_group_priority,
            delay_time,
            fail_retry_times,
            fail_retry_interval,
            timeout_notify_strategy,
            timeout,
            workflow,
            resource_list,
            dependence,
            wait_start_timeout,
            condition_result,
            resource_plugin,
            is_cache,
            input_params,
            output_params,
            *args,
            **kwargs,
        )
        self.task_execute_type = "STREAM"