# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project

from __future__ import annotations

from typing import TYPE_CHECKING

from vllm.logger import init_logger
from vllm.tool_parsers.tool_strict_level import ToolStrictLevel

if TYPE_CHECKING:
    from vllm.parser.abstract_parser import Parser
    from vllm.reasoning import ReasoningParser
    from vllm.tokenizers import TokenizerLike
    from vllm.tool_parsers import ToolParser

logger = init_logger(__name__)


class ParserManager:
    """Provides a unified Parser by composing reasoning and tool parser adapters."""

    @classmethod
    def get_tool_parser(
        cls,
        tool_parser_name: str | None = None,
        enable_auto_tools: bool = False,
        model_name: str | None = None,
    ) -> type[ToolParser] | None:
        """Get the tool parser based on the name."""
        from vllm.tool_parsers import ToolParserManager

        parser: type[ToolParser] | None = None
        if not enable_auto_tools or tool_parser_name is None:
            return parser
        logger.info_once('"auto" tool choice has been enabled.')

        try:
            if (
                tool_parser_name == "pythonic"
                and model_name
                and model_name.startswith("meta-llama/Llama-3.2")
            ):
                logger.warning(
                    "Llama3.2 models may struggle to emit valid pythonic tool calls"
                )
            parser = ToolParserManager.get_tool_parser(tool_parser_name)
        except Exception as e:
            raise TypeError(
                "Error: --enable-auto-tool-choice requires "
                f"tool_parser:'{tool_parser_name}' which has not "
                "been registered"
            ) from e
        return parser

    @classmethod
    def get_reasoning_parser(
        cls,
        reasoning_parser_name: str | None,
    ) -> type[ReasoningParser] | None:
        """Get the reasoning parser based on the name."""
        from vllm.reasoning import ReasoningParserManager

        parser: type[ReasoningParser] | None = None
        if not reasoning_parser_name:
            return None
        try:
            parser = ReasoningParserManager.get_reasoning_parser(reasoning_parser_name)
            assert parser is not None
        except Exception as e:
            raise TypeError(f"{reasoning_parser_name=} has not been registered") from e
        return parser

    @classmethod
    def validate_parser_tokenizer(
        cls,
        parser_cls: type[Parser],
        tokenizer: TokenizerLike,
        tool_parser_name: str | None = None,
        reasoning_parser_name: str | None = None,
        model_name: str | None = None,
    ) -> None:
        """Fail at startup if the tokenizer lacks what the parsers need."""
        try:
            parser_cls(tokenizer)
        except Exception as e:
            flags = []
            if parser_cls.tool_parser_cls is not None:
                flags.append(f"--tool-call-parser {tool_parser_name}")
            if parser_cls.reasoning_parser_cls is not None:
                flags.append(f"--reasoning-parser {reasoning_parser_name}")
            raise TypeError(
                f"{' and '.join(flags)} cannot be used with the tokenizer of "
                f"{model_name!r}: {e}"
            ) from e

    @classmethod
    def get_parser(
        cls,
        tool_parser_name: str | None = None,
        reasoning_parser_name: str | None = None,
        enable_auto_tools: bool = False,
        model_name: str | None = None,
        is_harmony: bool = False,
        tool_strict_level: str | None = None,
        tokenizer: TokenizerLike | None = None,
    ) -> type[Parser] | None:
        """Get a Parser that handles both reasoning and tool parsing.

        Composes the individual parsers into a ``DelegatingParser`` subclass.

        Args:
            tool_parser_name: The name of the tool parser.
            reasoning_parser_name: The name of the reasoning parser.
            enable_auto_tools: Whether auto tool choice is enabled.
            model_name: The model name for parser-specific warnings.
            is_harmony: Whether the selected model uses the Harmony format.
                        If True, HarmonyParser is always returned.
            tool_strict_level: Server-side floor for tool-call structural
                tags (``--tool-strict-level``), or ``None`` for the tool
                parser's default level.
            tokenizer: Tokenizer the composed parser is validated against
                at startup (see :meth:`validate_parser_tokenizer`), including
                its `response_template` metadata when a response-template
                parser is selected. ``None`` skips the check (e.g.
                ``skip_tokenizer_init``).

        Returns:
            A Parser class, or None if neither parser is specified.

        """
        parser_cls = cls._compose_parser(
            tool_parser_name=tool_parser_name,
            reasoning_parser_name=reasoning_parser_name,
            enable_auto_tools=enable_auto_tools,
            model_name=model_name,
            is_harmony=is_harmony,
            tool_strict_level=tool_strict_level,
            tokenizer=tokenizer,
        )
        if parser_cls is not None and tokenizer is not None:
            cls.validate_parser_tokenizer(
                parser_cls,
                tokenizer,
                tool_parser_name=tool_parser_name,
                reasoning_parser_name=reasoning_parser_name,
                model_name=model_name,
            )
        return parser_cls

    @classmethod
    def _compose_parser(
        cls,
        tool_parser_name: str | None,
        reasoning_parser_name: str | None,
        enable_auto_tools: bool,
        model_name: str | None,
        is_harmony: bool,
        tool_strict_level: str | None,
        tokenizer: TokenizerLike | None,
    ) -> type[Parser] | None:
        if not tool_parser_name and not reasoning_parser_name:
            return None

        reasoning_parser_cls = cls.get_reasoning_parser(reasoning_parser_name)
        tool_parser_cls = cls.get_tool_parser(
            tool_parser_name, enable_auto_tools, model_name
        )

        if reasoning_parser_cls is None and tool_parser_cls is None:
            return None

        strict_level = (
            None
            if tool_strict_level is None
            else ToolStrictLevel.from_name(tool_strict_level)
        )

        if is_harmony:
            from vllm.parser.harmony import HarmonyParser

            HarmonyParser.reasoning_parser_cls = reasoning_parser_cls
            HarmonyParser.tool_parser_cls = tool_parser_cls
            HarmonyParser.tool_strict_level = strict_level
            return HarmonyParser

        from vllm.parser.response_template import (
            response_template_parser_cls,
            validate_tokenizer_response_template,
        )

        template_parsers = {
            response_template_parser_cls(parser_cls)
            for parser_cls in (reasoning_parser_cls, tool_parser_cls)
            if parser_cls is not None
        }
        template_parser_cls = next(iter(template_parsers - {None}), None)
        if template_parser_cls is not None:
            if len(template_parsers) > 1:
                raise TypeError(
                    f"--reasoning-parser {reasoning_parser_name} and "
                    f"--tool-call-parser {tool_parser_name} cannot be combined"
                )
            if tokenizer is not None:
                validate_tokenizer_response_template(
                    tokenizer,
                    reasoning_parser=reasoning_parser_name,
                    # Without auto tool choice, no tool parser is selected.
                    tool_parser=tool_parser_name if tool_parser_cls else None,
                )
            return type(
                template_parser_cls.__name__,
                (template_parser_cls,),
                {
                    "reasoning_parser_cls": reasoning_parser_cls,
                    "tool_parser_cls": tool_parser_cls,
                    "tool_strict_level": strict_level,
                    "_enable_auto_tools": enable_auto_tools,
                },
            )

        if reasoning_parser_name == "kimi_k3" or tool_parser_name == "kimi_k3":
            from vllm.parser.kimi_k3 import KimiK3Parser

            r_cls = reasoning_parser_cls
            t_cls = tool_parser_cls

            class _KimiK3Parser(KimiK3Parser):
                reasoning_parser_cls = r_cls
                tool_parser_cls = t_cls
                tool_strict_level = strict_level

            return _KimiK3Parser

        if {reasoning_parser_name, tool_parser_name} & {
            "cohere_command3",
            "cohere_command4",
        }:
            from vllm.parser.cohere_command import CohereCommandParser

            r_cls = reasoning_parser_cls
            t_cls = tool_parser_cls

            class _CohereCommandParser(CohereCommandParser):
                reasoning_parser_cls = r_cls
                tool_parser_cls = t_cls
                tool_strict_level = strict_level

            return _CohereCommandParser

        if "minicpmv" in {reasoning_parser_name, tool_parser_name}:
            from vllm.parser.minicpmv import MiniCPMVUnifiedParser

            r_cls = reasoning_parser_cls
            t_cls = tool_parser_cls

            class _MiniCPMVParser(MiniCPMVUnifiedParser):
                reasoning_parser_cls = r_cls
                tool_parser_cls = t_cls
                tool_strict_level = strict_level

            return _MiniCPMVParser

        from vllm.parser.abstract_parser import DelegatingParser

        r_cls = reasoning_parser_cls
        t_cls = tool_parser_cls

        class _Parser(DelegatingParser):
            reasoning_parser_cls = r_cls
            tool_parser_cls = t_cls
            tool_strict_level = strict_level

        return _Parser