import dataclasses
import json
import os
import stat
import subprocess
from typing import (
    AbstractSet,
    List,
    Tuple,
    Optional,
    Dict,
    Any,
    Union,
    TypedDict,
    NotRequired,
)
from collections.abc import Mapping, Iterable, Sequence, Iterator, Container

from debputy.analysis import REFERENCE_DATA_TABLE
from debputy.analysis.analysis_util import flatten_ppfs
from debputy.dh.dh_assistant import (
    resolve_active_and_inactive_dh_commands,
    read_dh_addon_sequences,
    extract_dh_compat_level,
)
from debputy.integration_detection import determine_debputy_integration_mode
from debputy.packager_provided_files import (
    PackagerProvidedFile,
    detect_all_packager_provided_files,
)
from debputy.packages import BinaryPackage, SourcePackage
from debputy.plugin.api import (
    VirtualPath,
    packager_provided_file_reference_documentation,
)
from debputy.plugin.api.feature_set import PluginProvidedFeatureSet
from debputy.plugin.api.impl import plugin_metadata_for_debputys_own_plugin
from debputy.plugin.api.impl_types import (
    PluginProvidedKnownPackagingFile,
    DebputyPluginMetadata,
    KnownPackagingFileInfo,
    DHCompatibilityBasedRule,
    PackagerProvidedFileClassSpec,
    expand_known_packaging_config_features,
)
from debputy.plugin.api.spec import DebputyIntegrationMode
from debputy.util import (
    assume_not_none,
    escape_shell,
    _trace_log,
    _is_trace_log_enabled,
    render_command,
)

PackagingFileInfo = TypedDict(
    "PackagingFileInfo",
    {
        "path": str,
        "binary-package": NotRequired[str],
        "install-path": NotRequired[str],
        "install-pattern": NotRequired[str],
        "file-categories": NotRequired[list[str]],
        "config-features": NotRequired[list[str]],
        "pkgfile-is-active-in-build": NotRequired[bool],
        "pkgfile-stem": NotRequired[str],
        "pkgfile-explicit-package-name": NotRequired[bool],
        "pkgfile-name-segment": NotRequired[str],
        "pkgfile-architecture-restriction": NotRequired[str],
        "likely-typo-of": NotRequired[str],
        "likely-generated-from": NotRequired[list[str]],
        "related-tools": NotRequired[list[str]],
        "documentation-uris": NotRequired[list[str]],
        "debputy-cmd-templates": NotRequired[list[list[str]]],
        "generates": NotRequired[str],
        "generated-from": NotRequired[str],
    },
)


def _perl_json_bool(base: Any | None) -> Any | None:
    if base is None:
        return None

    if isinstance(base, bool):
        return base

    # Account for different ways that Perl will or could return a JSON bool,
    # when not using `JSON::PP::{true,false}`.
    #
    # https://salsa.debian.org/debian/debputy/-/issues/119#note_627057
    if isinstance(base, int):
        if base == 0:
            return False
        return True if base == 1 else base
    if isinstance(base, str):
        return False if base == "" else base
    return base


def scan_debian_dir(
    feature_set: PluginProvidedFeatureSet,
    source_package: SourcePackage,
    binary_packages: Mapping[str, BinaryPackage],
    debian_dir: VirtualPath,
    *,
    uses_dh_sequencer: bool = True,
    dh_sequences: AbstractSet[str] | None = None,
) -> tuple[list[PackagingFileInfo], list[str], int, object | None]:
    known_packaging_files = feature_set.known_packaging_files
    debputy_plugin_metadata = plugin_metadata_for_debputys_own_plugin()
    reference_data_set_names = [
        "config-features",
        "file-categories",
    ]
    for n in reference_data_set_names:
        assert n in REFERENCE_DATA_TABLE

    annotated: list[PackagingFileInfo] = []
    seen_paths: dict[str, PackagingFileInfo] = {}

    if dh_sequences is None:
        r = read_dh_addon_sequences(debian_dir)
        if r is not None:
            bd_sequences, dr_sequences, uses_dh_sequencer = r
            dh_sequences = bd_sequences | dr_sequences
        else:
            dh_sequences = set()
            uses_dh_sequencer = False
    debputy_integration_mode = determine_debputy_integration_mode(
        source_package.fields,
        dh_sequences,
    )
    is_debputy_package = debputy_integration_mode is not None
    dh_compat_level, dh_assistant_exit_code = extract_dh_compat_level()
    dh_issues = []

    static_packaging_files = {
        kpf.detection_value: kpf
        for kpf in known_packaging_files.values()
        if kpf.detection_method == "path"
    }
    dh_pkgfile_docs = {
        kpf.detection_value: kpf
        for kpf in known_packaging_files.values()
        if kpf.detection_method == "dh.pkgfile"
    }
    if is_debputy_package:
        all_debputy_ppfs = list(
            flatten_ppfs(
                detect_all_packager_provided_files(
                    feature_set,
                    debian_dir,
                    binary_packages,
                    allow_fuzzy_matches=True,
                    detect_typos=True,
                    ignore_paths=static_packaging_files,
                )
            )
        )
    else:
        all_debputy_ppfs = []

    if dh_compat_level is not None:
        (
            all_dh_ppfs,
            _,
            dh_issues,
            dh_assistant_exit_code,
        ) = resolve_debhelper_config_files(
            debian_dir,
            binary_packages,
            debputy_plugin_metadata,
            feature_set,
            dh_sequences,
            dh_compat_level,
            uses_dh_sequencer,
            debputy_integration_mode=debputy_integration_mode,
            ignore_paths=static_packaging_files,
        )

    else:
        all_dh_ppfs = []

    for ppf in all_debputy_ppfs:
        key = ppf.path.path
        ref_doc = ppf.definition.reference_documentation
        documentation_uris = (
            ref_doc.format_documentation_uris if ref_doc is not None else None
        )
        details: PackagingFileInfo = {
            "path": key,
            "binary-package": ppf.package_name,
            "pkgfile-stem": ppf.definition.stem,
            "pkgfile-explicit-package-name": ppf.uses_explicit_package_name,
            "pkgfile-is-active-in-build": ppf.definition.has_active_command,
            "debputy-cmd-templates": [
                ["debputy", "plugin", "show", "p-p-f", ppf.definition.stem]
            ],
        }
        if ppf.fuzzy_match and key.endswith(".in"):
            _merge_list(details, "file-categories", ["generic-template"])
            details["generates"] = key[:-3]
        elif assume_not_none(ppf.path.parent_dir).get(ppf.path.name + ".in"):
            _merge_list(details, "file-categories", ["generated"])
            details["generated-from"] = key + ".in"
        name_segment = ppf.name_segment
        arch_restriction = ppf.architecture_restriction
        if name_segment is not None:
            details["pkgfile-name-segment"] = name_segment
        if arch_restriction:
            details["pkgfile-architecture-restriction"] = arch_restriction
        seen_paths[key] = details
        annotated.append(details)
        static_details = static_packaging_files.get(key)
        if static_details is not None:
            # debhelper compat rules does not apply to debputy files
            _add_known_packaging_data(details, static_details, None)
        if documentation_uris:
            details["documentation-uris"] = list(documentation_uris)

    _merge_ppfs(annotated, seen_paths, all_dh_ppfs, dh_pkgfile_docs, dh_compat_level)

    for virtual_path in _scan_debian_dir(debian_dir):
        key = virtual_path.path
        if key in seen_paths or _skip_path(virtual_path):
            continue

        static_match = static_packaging_files.get(virtual_path.path)
        if static_match is not None:
            details: PackagingFileInfo = {
                "path": key,
            }
            annotated.append(details)
            if assume_not_none(virtual_path.parent_dir).get(virtual_path.name + ".in"):
                details["generated-from"] = key + ".in"
                _merge_list(details, "file-categories", ["generated"])
            _add_known_packaging_data(details, static_match, dh_compat_level)

    return annotated, reference_data_set_names, dh_assistant_exit_code, dh_issues


def _skip_path(virtual_path: VirtualPath) -> bool:
    if virtual_path.is_symlink:
        try:
            st = os.stat(virtual_path.fs_path)
        except FileNotFoundError:
            return True
        else:
            if not stat.S_ISREG(st.st_mode):
                return True
    elif not virtual_path.is_file:
        return True
    return False


def _fake_PPFClassSpec(
    debputy_plugin_metadata: DebputyPluginMetadata,
    stem: str,
    doc_uris: Sequence[str] | None,
    install_pattern: str | None,
    *,
    default_priority: int | None = None,
    packageless_is_fallback_for_all_packages: bool = False,
    post_formatting_rewrite: str | None = None,
    bug_950723: bool = False,
    has_active_command: bool = False,
) -> PackagerProvidedFileClassSpec:
    if install_pattern is None:
        install_pattern = "not-a-real-ppf"
    if post_formatting_rewrite is not None:
        formatting_hook = _POST_FORMATTING_REWRITE[post_formatting_rewrite]
    else:
        formatting_hook = None
    return PackagerProvidedFileClassSpec(
        debputy_plugin_metadata,
        stem,
        install_pattern,
        allow_architecture_segment=True,
        allow_name_segment=True,
        default_priority=default_priority,
        default_mode=0o644,
        post_formatting_rewrite=formatting_hook,
        packageless_is_fallback_for_all_packages=packageless_is_fallback_for_all_packages,
        reservation_only=False,
        formatting_callback=None,
        bug_950723=bug_950723,
        has_active_command=has_active_command,
        reference_documentation=packager_provided_file_reference_documentation(
            format_documentation_uris=doc_uris,
        ),
    )


def _relevant_dh_compat_rules(
    compat_level: int | None,
    info: KnownPackagingFileInfo,
) -> Iterable[DHCompatibilityBasedRule]:
    if compat_level is None:
        return
    dh_compat_rules = info.get("dh_compat_rules")
    if not dh_compat_rules:
        return
    for dh_compat_rule in dh_compat_rules:
        rule_compat_level = dh_compat_rule.get("starting_with_compat_level")
        if rule_compat_level is not None and compat_level < rule_compat_level:
            continue
        yield dh_compat_rule


def _kpf_install_pattern(
    compat_level: int | None,
    ppkpf: PluginProvidedKnownPackagingFile,
) -> str | None:
    for compat_rule in _relevant_dh_compat_rules(compat_level, ppkpf.info):
        install_pattern = compat_rule.get("install_pattern")
        if install_pattern is not None:
            return install_pattern
    return ppkpf.info.get("install_pattern")


def resolve_debhelper_config_files(
    debian_dir: VirtualPath,
    binary_packages: Mapping[str, BinaryPackage],
    debputy_plugin_metadata: DebputyPluginMetadata,
    plugin_feature_set: PluginProvidedFeatureSet,
    dh_rules_addons: AbstractSet[str],
    dh_compat_level: int,
    saw_dh: bool,
    ignore_paths: Container[str] = frozenset(),
    *,
    debputy_integration_mode: DebputyIntegrationMode | None = None,
    cwd: str | None = None,
) -> tuple[list[PackagerProvidedFile], list[str], object | None, int]:

    dh_ppf_docs = {
        kpf.detection_value: kpf
        for kpf in plugin_feature_set.known_packaging_files.values()
        if kpf.detection_method == "dh.pkgfile"
    }

    dh_ppfs = {}
    commands, exit_code = _relevant_dh_commands(dh_rules_addons, cwd=cwd)

    cmd = ["dh_assistant", "list-guessed-dh-config-files"]
    if dh_rules_addons:
        addons = ",".join(dh_rules_addons)
        cmd.append(f"--with={addons}")
    try:
        output = subprocess.check_output(
            cmd,
            stderr=subprocess.DEVNULL,
            cwd=cwd,
        )
    except (subprocess.CalledProcessError, FileNotFoundError) as e:
        config_files = []
        issues = None
        missing_introspection: list[str] = []
        if isinstance(e, subprocess.CalledProcessError):
            exit_code = e.returncode
        else:
            exit_code = 127
        if _is_trace_log_enabled():
            _trace_log(
                f"Command {render_command(*cmd, cwd=cwd)} failed with {exit_code} ({cwd=})"
            )
    else:
        result = json.loads(output)
        config_files = result.get("config-files", [])
        missing_introspection: list[str] = result.get("commands-not-introspectable", [])
        issues = result.get("issues")
        if _is_trace_log_enabled():
            _trace_log(
                f"Command {render_command(*cmd, cwd=cwd)} returned successfully: {output}"
            )
    dh_commands = resolve_active_and_inactive_dh_commands(dh_rules_addons)
    for config_file in config_files:
        if not isinstance(config_file, dict):
            continue
        if config_file.get("file-type") != "pkgfile":
            continue
        stem = config_file.get("pkgfile")
        if stem is None:
            continue
        internal = config_file.get("internal")
        if isinstance(internal, dict):
            bug_950723 = internal.get("bug#950723", False) is True
        else:
            bug_950723 = False
        commands = config_file.get("commands")
        documentation_uris = []
        related_tools = []
        seen_commands = set()
        seen_docs = set()
        ppkpf = dh_ppf_docs.get(stem)

        if ppkpf:
            dh_cmds = ppkpf.info.get("debhelper_commands")
            doc_uris = ppkpf.info.get("documentation_uris")
            default_priority = ppkpf.info.get("default_priority")
            if doc_uris is not None:
                seen_docs.update(doc_uris)
                documentation_uris.extend(doc_uris)
            if dh_cmds is not None:
                seen_commands.update(dh_cmds)
                related_tools.extend(dh_cmds)
            install_pattern = _kpf_install_pattern(dh_compat_level, ppkpf)
            post_formatting_rewrite = ppkpf.info.get("post_formatting_rewrite")
            packageless_is_fallback_for_all_packages = ppkpf.info.get(
                "packageless_is_fallback_for_all_packages",
                False,
            )
            # If it is a debhelper PPF, then `has_active_command` is false by default.
            has_active_command = ppkpf.info.get("has_active_command", False)
        else:
            install_pattern = None
            default_priority = None
            post_formatting_rewrite = None
            packageless_is_fallback_for_all_packages = False
            has_active_command = False
        for command in commands:
            if isinstance(command, dict):
                command_name = command.get("command")
                if isinstance(command_name, str) and command_name:
                    if command_name not in seen_commands:
                        related_tools.append(command_name)
                        seen_commands.add(command_name)
                    manpage = f"man:{command_name}(1)"
                    if manpage not in seen_docs:
                        documentation_uris.append(manpage)
                        seen_docs.add(manpage)
                else:
                    continue
                is_active = _perl_json_bool(command.get("is-active", True))
                if is_active is None and command_name in dh_commands.active_commands:
                    is_active = True
                if not isinstance(is_active, bool):
                    continue
                if is_active:
                    has_active_command = True

        if debputy_integration_mode == "full":
            # dh commands are never active in full integration mode.
            has_active_command = False
        elif not saw_dh:
            # If we did not see `dh`, we assume classic `debhelper` where we have no way of knowing
            # which commands are active.
            has_active_command = True
        dh_ppfs[stem] = _fake_PPFClassSpec(
            debputy_plugin_metadata,
            stem,
            documentation_uris,
            install_pattern,
            default_priority=default_priority,
            post_formatting_rewrite=post_formatting_rewrite,
            packageless_is_fallback_for_all_packages=packageless_is_fallback_for_all_packages,
            bug_950723=bug_950723,
            has_active_command=has_active_command,
        )
    for ppkpf in dh_ppf_docs.values():
        stem = ppkpf.detection_value
        if stem in dh_ppfs:
            continue

        default_priority = ppkpf.info.get("default_priority")
        install_pattern = _kpf_install_pattern(dh_compat_level, ppkpf)
        post_formatting_rewrite = ppkpf.info.get("post_formatting_rewrite")
        packageless_is_fallback_for_all_packages = ppkpf.info.get(
            "packageless_is_fallback_for_all_packages",
            False,
        )
        has_active_command = (
            ppkpf.info.get("has_active_command", False) if saw_dh else False
        )
        if not has_active_command:
            dh_cmds = ppkpf.info.get("debhelper_commands")
            if dh_cmds:
                has_active_command = any(
                    c in dh_commands.active_commands for c in dh_cmds
                )
        dh_ppfs[stem] = _fake_PPFClassSpec(
            debputy_plugin_metadata,
            stem,
            ppkpf.info.get("documentation_uris"),
            install_pattern,
            default_priority=default_priority,
            post_formatting_rewrite=post_formatting_rewrite,
            packageless_is_fallback_for_all_packages=packageless_is_fallback_for_all_packages,
            has_active_command=has_active_command,
        )
    dh_ppf_feature_set = _fake_plugin_feature_set(plugin_feature_set, dh_ppfs)
    all_dh_ppfs = list(
        flatten_ppfs(
            detect_all_packager_provided_files(
                dh_ppf_feature_set,
                debian_dir,
                binary_packages,
                allow_fuzzy_matches=True,
                detect_typos=True,
                ignore_paths=ignore_paths,
            )
        )
    )
    return all_dh_ppfs, missing_introspection, issues, exit_code


def _fake_plugin_feature_set(
    plugin_feature_set: PluginProvidedFeatureSet,
    fake_defs: Mapping[str, PackagerProvidedFileClassSpec],
) -> PluginProvidedFeatureSet:
    return dataclasses.replace(plugin_feature_set, packager_provided_files=fake_defs)


def _merge_list(
    existing_table: dict[str, Any],
    key: str,
    new_data: Sequence[str] | None,
) -> None:
    if not new_data:
        return
    existing_values = existing_table.get(key, [])
    if isinstance(existing_values, tuple):
        existing_values = list(existing_values)
    assert isinstance(existing_values, list)
    seen = set(existing_values)
    existing_values.extend(x for x in new_data if x not in seen)
    existing_table[key] = existing_values


def _merge_ppfs(
    identified: list[PackagingFileInfo],
    seen_paths: dict[str, PackagingFileInfo],
    ppfs: list[PackagerProvidedFile],
    context: Mapping[str, PluginProvidedKnownPackagingFile],
    dh_compat_level: int | None,
) -> None:
    for ppf in ppfs:
        key = ppf.path.path
        ref_doc = ppf.definition.reference_documentation
        documentation_uris = (
            ref_doc.format_documentation_uris if ref_doc is not None else None
        )
        if not ppf.definition.installed_as_format.startswith("not-a-real-ppf"):
            try:
                parts = ppf.compute_dest()
            except RuntimeError:
                dest = None
            else:
                dest = "/".join(parts).lstrip(".")
        else:
            dest = None
        orig_details = seen_paths.get(key)
        if orig_details is None:
            details: PackagingFileInfo = {
                "path": key,
                "pkgfile-stem": ppf.definition.stem,
                "pkgfile-is-active-in-build": ppf.definition.has_active_command,
                "pkgfile-explicit-package-name": ppf.uses_explicit_package_name,
                "binary-package": ppf.package_name,
            }
            if ppf.expected_path is not None:
                details["likely-typo-of"] = ppf.expected_path
            identified.append(details)
        else:
            details = orig_details
            # We do not merge the "is typo" field; if the original
            for k, v in [
                ("pkgfile-stem", ppf.definition.stem),
                ("pkgfile-explicit-package-name", ppf.definition.has_active_command),
                ("binary-package", ppf.package_name),
            ]:
                if k not in details:
                    details[k] = v
            if ppf.definition.has_active_command and details.get(
                "pkgfile-is-active-in-build", False
            ):
                details["pkgfile-is-active-in-build"] = True
            if ppf.expected_path is None and "likely-typo-of" in details:
                del details["likely-typo-of"]

        name_segment = ppf.name_segment
        arch_restriction = ppf.architecture_restriction
        if name_segment is not None and "pkgfile-name-segment" not in details:
            details["pkgfile-name-segment"] = name_segment
        if (
            arch_restriction is not None
            and "pkgfile-architecture-restriction" not in details
        ):
            details["pkgfile-architecture-restriction"] = arch_restriction
        if ppf.fuzzy_match and key.endswith(".in"):
            _merge_list(details, "file-categories", ["generic-template"])
            details["generates"] = key[:-3]
        elif assume_not_none(ppf.path.parent_dir).get(ppf.path.name + ".in"):
            _merge_list(details, "file-categories", ["generated"])
            details["generated-from"] = key + ".in"
        if dest is not None and "install-path" not in details:
            details["install-path"] = dest

        extra_details = context.get(ppf.definition.stem)
        if extra_details is not None:
            _add_known_packaging_data(details, extra_details, dh_compat_level)

        _merge_list(details, "documentation-uris", documentation_uris)


def _relevant_dh_commands(
    dh_rules_addons: Iterable[str],
    cwd: str | None = None,
) -> tuple[list[str], int]:
    cmd = ["dh_assistant", "list-commands", "--output-format=json"]
    if dh_rules_addons:
        addons = ",".join(dh_rules_addons)
        cmd.append(f"--with={addons}")
    try:
        output = subprocess.check_output(
            cmd,
            stderr=subprocess.DEVNULL,
            cwd=cwd,
        )
    except (FileNotFoundError, subprocess.CalledProcessError) as e:
        exit_code = 127
        if isinstance(e, subprocess.CalledProcessError):
            exit_code = e.returncode
        if _is_trace_log_enabled():
            _trace_log(
                f"Command {render_command(*cmd, cwd=cwd)} failed with {exit_code} ({cwd=})"
            )
        return [], exit_code
    else:
        data = json.loads(output)
        commands_json = data.get("commands")
        commands = []
        if _is_trace_log_enabled():
            _trace_log(
                f"Command {render_command(*cmd, cwd=cwd)} returned successfully: {output}"
            )
        for command in commands_json:
            if isinstance(command, dict):
                command_name = command.get("command")
                if isinstance(command_name, str) and command_name:
                    commands.append(command_name)
        return commands, 0


def _add_known_packaging_data(
    details: PackagingFileInfo,
    plugin_data: PluginProvidedKnownPackagingFile,
    dh_compat_level: int | None,
):
    install_pattern = _kpf_install_pattern(
        dh_compat_level,
        plugin_data,
    )
    config_features = plugin_data.info.get("config_features")
    if config_features:
        config_features = expand_known_packaging_config_features(
            dh_compat_level or 0,
            config_features,
        )
        _merge_list(details, "config-features", config_features)

    if dh_compat_level is not None:
        extra_config_features = []
        for dh_compat_rule in _relevant_dh_compat_rules(
            dh_compat_level, plugin_data.info
        ):
            cf = dh_compat_rule.get("add_config_features")
            if cf:
                extra_config_features.extend(cf)
        if extra_config_features:
            extra_config_features = expand_known_packaging_config_features(
                dh_compat_level,
                extra_config_features,
            )
            _merge_list(details, "config-features", extra_config_features)
    if "install-pattern" not in details and install_pattern is not None:
        details["install-pattern"] = install_pattern
    for mk, ok in [
        ("file_categories", "file-categories"),
        ("documentation_uris", "documentation-uris"),
        ("debputy_cmd_templates", "debputy-cmd-templates"),
    ]:
        value = plugin_data.info.get(mk)
        if value and ok == "debputy-cmd-templates":
            value = [escape_shell(*c) for c in value]
        _merge_list(details, ok, value)


def _scan_debian_dir(debian_dir: VirtualPath) -> Iterator[VirtualPath]:
    for p in debian_dir.iterdir:
        yield p
        if p.is_dir and p.path in ("debian/source", "debian/tests"):
            yield from p.iterdir


_POST_FORMATTING_REWRITE = {
    "period-to-underscore": lambda n: n.replace(".", "_"),
}