import asyncio
import collections
from typing import (
    Optional,
    Any,
    TYPE_CHECKING,
    cast,
    Union,
)
from collections.abc import Mapping, Callable, Sequence, Awaitable

import pytest

from debputy.filesystem_scan import VirtualPathBase
from debputy.linting.lint_util import (
    LinterPositionCodec,
    LintStateImpl,
    LintState,
    AsyncLinterImpl,
    te_range_to_lsp,
)
from debputy.lsp.config.config_options import (
    ALL_DEBPUTY_CONFIG_OPTIONS,
    DebputyConfigOption,
)
from debputy.lsp.config.debputy_config import DebputyConfig
from debputy.lsp.diagnostics import DiagnosticData
from debputy.lsp.maint_prefs import (
    MaintainerPreferenceTable,
    EffectiveFormattingPreference,
    determine_effective_preference,
)
from debputy.lsp.quickfixes import CODE_ACTION_HANDLERS, CodeActionName
from debputy.lsp.text_edit import apply_text_edits
from debputy.packages import DctrlParser
from debputy.plugin.api.feature_set import PluginProvidedFeatureSet
from debputy.util import T
from debputy.lsp.vendoring._deb822_repro.locatable import (
    Range as TERange,
)

if TYPE_CHECKING:
    from lsprotocol import types
else:
    from debputy.lsprotocol import types


try:
    from Levenshtein import distance

    HAS_LEVENSHTEIN = True
except ImportError:
    HAS_LEVENSHTEIN = False

LINTER_POSITION_CODEC = LinterPositionCodec()


class TestDebputyConfig(DebputyConfig):

    def set_config_value(
        self, config_option: DebputyConfigOption[T], value: T | None
    ) -> None:
        self._config[config_option] = value

    def reset_config(self, config_option: DebputyConfigOption[T]) -> None:
        self.set_config_value(config_option, config_option.default_value)


class BaseLintStateWrapper:
    def __init__(
        self,
        path: str,
        debputy_plugin_feature_set: PluginProvidedFeatureSet,
        dctrl_parser: DctrlParser,
    ) -> None:
        self._debputy_plugin_feature_set = debputy_plugin_feature_set
        self.dctrl_lines: list[str] | None = None
        self.path = path
        self._dctrl_parser = dctrl_parser
        self.source_root: VirtualPathBase | None = None
        self.maint_preference_table = MaintainerPreferenceTable({}, {})
        self.effective_preference: EffectiveFormattingPreference | None = None
        self.debputy_config = TestDebputyConfig(
            {c: c.default_value for c in ALL_DEBPUTY_CONFIG_OPTIONS}
        )

    def _setup(
        self,
        lines: list[str],
        *,
        lint_handler: Optional["AsyncLinterImpl"] = None,
    ) -> LintStateImpl:
        source_package = None
        binary_packages = None
        dctrl_lines = self.dctrl_lines
        if dctrl_lines is not None:
            _, source_package, binary_packages = (
                self._dctrl_parser.parse_source_debian_control(
                    dctrl_lines, ignore_errors=True
                )
            )
        source_root = self.source_root
        debian_dir = source_root.get("debian") if source_root is not None else None
        return LintStateImpl(
            self._debputy_plugin_feature_set,
            self.maint_preference_table,
            source_root,
            debian_dir,
            self.path,
            "".join(lines) if lines is not None else "",
            lines,
            self.debputy_config,
            source_package,
            binary_packages,
            self.effective_preference,
            lint_handler,
        )


class LintWrapper(BaseLintStateWrapper):

    def __init__(
        self,
        path: str,
        handler: Callable[[LintState], Awaitable[None]],
        debputy_plugin_feature_set: PluginProvidedFeatureSet,
        dctrl_parser: DctrlParser,
    ) -> None:
        super().__init__(path, debputy_plugin_feature_set, dctrl_parser)
        self._handler = handler
        self._lint_state: LintStateImpl | None = None

    @property
    def lint_state(self) -> LintStateImpl:
        lint_state = self._lint_state
        assert (
            lint_state is not None
        ), "lint_state must be fetched after the diagnostics have been run"
        return lint_state

    def __call__(self, lines: list[str]) -> list["types.Diagnostic"]:
        self._lint_state = lint_state = self._setup(lines, lint_handler=self._handler)
        return check_diagnostics(asyncio.run(lint_state.gather_diagnostics()))


class ReformatWrapper(BaseLintStateWrapper):

    def __init__(
        self,
        path: str,
        handler: Callable[[LintState], Sequence[types.TextEdit] | None],
        debputy_plugin_feature_set: PluginProvidedFeatureSet,
        dctrl_parser: DctrlParser,
        maint_preference_table: MaintainerPreferenceTable,
    ) -> None:
        super().__init__(path, debputy_plugin_feature_set, dctrl_parser)
        self._handler = handler
        self.maint_preference_table = maint_preference_table

    def reformat(self, lines: list[str]) -> list["types.TextEdit"]:
        def _style(lint_state: LintState) -> EffectiveFormattingPreference | None:
            pref_result = determine_effective_preference(
                lint_state.maint_preference_table, lint_state.source_package, None
            )
            return pref_result[0] if pref_result else None

        return self._reformat(lines, _style)

    def reformat_with_named_style(
        self, named_style: str, lines: list[str]
    ) -> list["types.TextEdit"]:
        return self._reformat(
            lines, lambda s: s.maint_preference_table.named_styles[named_style]
        )

    def _reformat(
        self,
        lines: list[str],
        style_selector: Callable[[LintState], EffectiveFormattingPreference | None],
    ) -> list["types.TextEdit"]:
        lint_state = self._setup(lines, lint_handler=None)
        lint_state.effective_preference = style_selector(lint_state)
        reformat_result = self._handler(lint_state)
        return reformat_result if reformat_result is not None else []


def requires_levenshtein(func: Any) -> Any:
    return pytest.mark.skipif(
        not HAS_LEVENSHTEIN, reason="Missing python3-levenshtein"
    )(func)


def check_diagnostics(
    diagnostics: list["types.Diagnostic"] | None,
) -> list["types.Diagnostic"]:
    if diagnostics:
        for diagnostic in diagnostics:
            assert diagnostic.severity is not None
            assert diagnostic.source is not None
    elif diagnostics is None:
        diagnostics = []
    return diagnostics


def by_range_sort_key(diagnostic: types.Diagnostic) -> Any:
    start_pos = diagnostic.range.start
    end_pos = diagnostic.range.end
    return start_pos.line, start_pos.character, end_pos.line, end_pos.character


def group_diagnostics_by_severity(
    diagnostics: list["types.Diagnostic"] | None,
) -> Mapping["types.DiagnosticSeverity", list["types.Diagnostic"]]:
    if not diagnostics:
        return {}

    by_severity = collections.defaultdict(list)

    for diagnostic in sorted(diagnostics, key=by_range_sort_key):
        severity = diagnostic.severity
        assert severity is not None
        by_severity[severity].append(diagnostic)

    return by_severity


def diag_range_to_text(lines: Sequence[str], range_: "types.Range") -> str:
    parts = []
    for line_no in range(range_.start.line, range_.end.line + 1):
        line = lines[line_no]
        chunk = line
        if line_no == range_.start.line and line_no == range_.end.line:
            chunk = line[range_.start.character : range_.end.character]
        elif line_no == range_.start.line:
            chunk = line[range_.start.character :]
        elif line_no == range_.end.line:
            chunk = line[: range_.end.character]
        parts.append(chunk)
    return "".join(parts)


def te_range_to_text(lines: Sequence[str], range_: "TERange") -> str:
    return diag_range_to_text(lines, te_range_to_lsp(range_))


def standard_quick_fixes_for_diagnostic(
    lint_state: LintState,
    code_action_params: "types.CodeActionParams",
    diagnostic: "types.Diagnostic",
) -> Sequence[Union["types.CodeAction", "types.Command"]]:
    if not isinstance(diagnostic.data, dict):
        return []
    data: DiagnosticData = cast("DiagnosticData", diagnostic.data)
    quickfixes = data.get("quickfixes")
    if quickfixes is None:
        return []
    actions = []
    for action_suggestion in quickfixes:
        if (
            action_suggestion
            and isinstance(action_suggestion, Mapping)
            and "code_action" in action_suggestion
        ):
            action_name: CodeActionName = action_suggestion["code_action"]
            handler = CODE_ACTION_HANDLERS.get(action_name)
            if handler is not None:
                actions.extend(
                    handler(
                        lint_state,
                        cast("Mapping[str, str]", action_suggestion),
                        code_action_params,
                        diagnostic,
                    )
                )
            else:
                assert False, f"No codeAction handler for {action_name} !?"
    return actions


def apply_code_action_edits(
    doc_uri: str, lines: list[str], code_action: types.CodeAction
) -> str:
    """Apply the text edits for a given doc URI from a code action


    The code action must have at least one edit for the given document. It *may*
    have edits for other documents and these will be ignored (except when they
    use an unsupported feature).

    Unsupported features:
     * Any code action with a command
     * Any code action with non-text related edits (create file, rename file, etc.)

    :param doc_uri: The doc_uri for the document represented by the lines parameter
    :param lines: The lines of the document to apply the edit to
    :param code_action: The code action to apply
    :return: The file after edits.
    """
    workspace_edit = code_action.edit
    assert workspace_edit is not None
    # We cannot apply a command with this function
    assert code_action.command is None
    orig = "".join(lines)
    document_changes = workspace_edit.document_changes
    if document_changes:
        assert workspace_edit.changes is None
        # We cannot apply non-edits
        assert all(isinstance(e, types.TextDocumentEdit) for e in document_changes)
        changes = cast("Sequence[types.TextDocumentEdit", document_changes)
        doc_edits = [c.edits for c in changes if c.text_document.uri == doc_uri]
        assert len(doc_edits) == 1
        edits = doc_edits[0]
    else:
        changes = workspace_edit.changes
        assert changes is not None
        edits = changes.get(doc_uri)
        assert edits
    return apply_text_edits(orig, lines, edits)


def apply_formatting_edits(
    lines: list[str],
    edits: Sequence[types.TextEdit],
) -> str:
    orig = "".join(lines)
    return apply_text_edits(orig, lines, edits)


def code_actions_for(
    uri: str,
    text_range: "types.Range",
) -> "types.CodeActionsParams":
    return types.CodeActionParams(
        types.TextDocumentIdentifier(uri),
        text_range,
        # Ignore the actual list of diagnostics, since we do not use them.
        types.CodeActionContext([]),
    )