import logging
from typing import (
    Any,
    NamedTuple,
    Optional,
    TYPE_CHECKING,
    Union,
)
from uuid import UUID

from sqlalchemy import (
    exists,
    false,
    select,
    sql,
    true,
    update,
)

from galaxy import (
    exceptions,
    model,
)
from galaxy.managers.context import ProvidesUserContext
from galaxy.model import (
    DynamicTool,
    User,
    UserDynamicToolAssociation,
)
from galaxy.tool_util.cwl import tool_proxy
from galaxy.tool_util.parser.yaml import YamlToolSource
from galaxy.tool_util.toolbox import AbstractToolBox
from galaxy.tool_util_models.dynamic_tool_models import (
    DynamicToolPayload,
    DynamicUnprivilegedToolCreatePayload,
)
from galaxy.tools import (
    create_tool_from_source,
    Tool,
)
from .base import (
    ModelManager,
    raise_filter_err,
)
from .executables import artifact_class

log = logging.getLogger(__name__)

if TYPE_CHECKING:
    from galaxy.managers.base import OrmFilterParsersType


def tool_payload_to_tool(app, tool_dict: dict[str, Any]) -> Optional[Tool]:
    tool_source = YamlToolSource(tool_dict)
    tool = create_tool_from_source(app, tool_source=tool_source, tool_dir=None)
    return tool


class ToolRunReference(NamedTuple):
    tool_id: Optional[str]
    tool_uuid: Optional[str]
    tool_version: Optional[str]


def get_tool_from_trans(trans: ProvidesUserContext, tool_ref: ToolRunReference) -> Tool:
    return get_tool_from_toolbox(trans.app.toolbox, tool_ref, trans.user)


def get_tool_from_toolbox(toolbox: AbstractToolBox, tool_ref: ToolRunReference, user: Optional[User]) -> Tool:
    tool = toolbox.get_tool(
        tool_id=tool_ref.tool_id, tool_uuid=tool_ref.tool_uuid, tool_version=tool_ref.tool_version, user=user
    )
    if not tool:
        log.debug(f"Not found tool with kwds [{tool_ref}]")
        raise exceptions.ToolMissingException("Tool not found.")
    return tool


class DynamicToolManager(ModelManager[DynamicTool]):
    """Manages dynamic tools stored in Galaxy's database."""

    model_class = DynamicTool

    def ensure_can_use_unprivileged_tool(self, user: model.User):
        stmt = select(
            exists().where(
                model.UserRoleAssociation.user_id == user.id,
                model.UserRoleAssociation.role_id == model.Role.id,
                model.Role.type == model.Role.types.USER_TOOL_EXECUTE,
                model.Role.deleted == false(),
            )
        )
        if not self.session().execute(stmt).scalar():
            raise exceptions.InsufficientPermissionsException("User is not allowed to run unprivileged tools")

    def get_tool_by_id_or_uuid(self, id_or_uuid: Union[int, str]) -> Union[DynamicTool, None]:
        if isinstance(id_or_uuid, int):
            return self.get_tool_by_id(id_or_uuid)
        else:
            return self.get_tool_by_uuid(id_or_uuid)

    def get_tool_by_uuid(self, uuid: Optional[Union[UUID, str]]):
        stmt = select(DynamicTool).where(DynamicTool.uuid == uuid, DynamicTool.public == true())
        return self.session().scalars(stmt).one_or_none()

    def get_tool_by_tool_id(self, tool_id):
        stmt = select(DynamicTool).where(DynamicTool.tool_id == tool_id, DynamicTool.public == true())
        return self.session().scalars(stmt).one_or_none()

    def get_unprivileged_tool_by_uuid(self, user: model.User, uuid: Union[UUID, str]):
        stmt = self.owned_unprivileged_statement(user).where(DynamicTool.uuid == uuid)
        return self.session().scalars(stmt).one_or_none()

    def get_unprivileged_tool_by_tool_id(self, user: model.User, tool_id: str):
        stmt = self.owned_unprivileged_statement(user).where(DynamicTool.tool_id == tool_id)
        return self.session().scalars(stmt).one_or_none()

    def get_tool_by_id(self, object_id):
        stmt = select(DynamicTool).where(DynamicTool.id == object_id, DynamicTool.public == true())
        return self.session().scalars(stmt).one_or_none()

    def create_tool(self, tool_payload: DynamicToolPayload):
        if not getattr(self.app.config, "enable_beta_tool_formats", False):
            raise exceptions.ConfigDoesNotAllowException(
                "Set 'enable_beta_tool_formats' in Galaxy config to create dynamic tools."
            )

        uuid = model.get_uuid()
        tool_directory: Optional[str] = None
        tool_path: Optional[str] = None
        if tool_payload.src == "from_path":
            tool_format, representation, _ = artifact_class(None, tool_payload.model_dump())
            tool_directory = tool_payload.tool_directory
            tool_path = tool_payload.path
        else:
            representation = tool_payload.representation.model_dump(by_alias=True, exclude_unset=True)
            if not representation:
                raise exceptions.ObjectAttributeMissingException("A tool 'representation' is required.")

            tool_format = representation.get("class")
            if not tool_format:
                raise exceptions.ObjectAttributeMissingException("Current tool representations require 'class'.")

        if tool_format in ("GalaxyTool", "GalaxyUserTool"):
            tool_id = representation.get("id")
            if not tool_id:
                tool_id = str(uuid)
        elif tool_format in ("CommandLineTool", "ExpressionTool"):
            # CWL tools
            if tool_path:
                proxy = tool_proxy(tool_path=tool_path, uuid=uuid)
            else:
                # Build a tool proxy so that we can convert to the persistable
                # hash.
                proxy = tool_proxy(
                    tool_object=representation["raw_process_reference"],
                    tool_directory=tool_directory,
                    uuid=uuid,
                )
            tool_id = proxy.galaxy_id()
        else:
            raise Exception(f"Unknown tool format [{tool_format}] encountered.")
        tool_version = representation.get("version")
        dynamic_tool = self.create(
            tool_format=tool_format,
            tool_id=tool_id,
            tool_version=tool_version,
            tool_path=tool_path,
            tool_directory=tool_directory,
            uuid=uuid,
            active=tool_payload.active,
            hidden=tool_payload.hidden,
            value=representation,
            public=True,
        )
        self.app.toolbox.load_dynamic_tool(dynamic_tool)
        return dynamic_tool

    def create_unprivileged_tool(
        self, user: model.User, tool_payload: DynamicUnprivilegedToolCreatePayload
    ) -> DynamicTool:
        if not getattr(self.app.config, "enable_beta_tool_formats", False):
            raise exceptions.ConfigDoesNotAllowException(
                "Set 'enable_beta_tool_formats' in Galaxy config to create dynamic tools."
            )
        self.ensure_can_use_unprivileged_tool(user)
        dynamic_tool = self.create(
            tool_format=tool_payload.representation.class_,
            tool_id=tool_payload.representation.id,
            tool_version=tool_payload.representation.version,
            active=tool_payload.active,
            hidden=tool_payload.hidden,
            value=tool_payload.representation.model_dump(by_alias=True),
            public=False,
            flush=True,
        )
        session = self.session()
        session.add(UserDynamicToolAssociation(user_id=user.id, dynamic_tool_id=dynamic_tool.id))
        session.commit()
        return dynamic_tool

    def list_tools(self, active=True):
        stmt = select(DynamicTool).where(DynamicTool.active == active)
        return self.session().scalars(stmt)

    def list_unprivileged_tools(self, user: model.User, active=True):
        self.ensure_can_use_unprivileged_tool(user)
        owned_statement = self.owned_unprivileged_statement(user=user)
        stmt = owned_statement.where(
            DynamicTool.active == active,
            UserDynamicToolAssociation.active == active,
        )
        return self.session().scalars(stmt)

    def owned_unprivileged_statement(self, user: model.User):
        self.ensure_can_use_unprivileged_tool(user)
        return (
            select(DynamicTool)
            .join(UserDynamicToolAssociation, DynamicTool.id == UserDynamicToolAssociation.dynamic_tool_id)
            .where(
                UserDynamicToolAssociation.user_id == user.id,
            )
            .order_by(UserDynamicToolAssociation.id.desc())
        )

    def deactivate_unprivileged_tool(self, user: model.User, dynamic_tool: DynamicTool):
        update_stmt = (
            update(UserDynamicToolAssociation)
            .where(
                UserDynamicToolAssociation.user_id == user.id,
                UserDynamicToolAssociation.dynamic_tool_id == dynamic_tool.id,
            )
            .values(active=False)
        )
        session = self.session()
        session.execute(update_stmt)
        session.commit()

    def deactivate(self, dynamic_tool: DynamicTool) -> DynamicTool:
        assert isinstance(dynamic_tool.uuid, UUID)
        del self.app.toolbox._tools_by_uuid[dynamic_tool.uuid]
        return self.update(dynamic_tool, {"active": False})


class ToolFilterMixin:
    orm_filter_parsers: "OrmFilterParsersType"

    def create_tool_filter(self, attr, op, val):
        def _create_tool_filter(model_class=None):
            if op == "eq":
                cond = model.Job.table.c.tool_id == val
            elif op == "contains":
                cond = model.Job.table.c.tool_id.contains(val, autoescape=True)
            else:
                raise_filter_err(attr, op, val, "bad op in filter")
            if model_class is model.HistoryDatasetAssociation:
                return sql.expression.and_(
                    model.Job.table.c.id == model.JobToOutputDatasetAssociation.table.c.job_id,
                    model.HistoryDatasetAssociation.table.c.id
                    == model.JobToOutputDatasetAssociation.table.c.dataset_id,
                    cond,
                )
            elif model_class is model.HistoryDatasetCollectionAssociation:
                return sql.expression.and_(
                    model.Job.id == model.JobToOutputDatasetAssociation.job_id,
                    model.JobToOutputDatasetAssociation.dataset_id == model.DatasetCollectionElement.hda_id,
                    model.DatasetCollectionElement.dataset_collection_id
                    == model.HistoryDatasetCollectionAssociation.collection_id,
                    cond,
                )
            else:
                return True

        return _create_tool_filter

    def _add_parsers(self):
        self.orm_filter_parsers.update(
            {
                "tool_id": self.create_tool_filter,
            }
        )
