import json
import os
import re
import subprocess
import sys
from itertools import chain
from typing import (
    Optional,
    List,
    Set,
    FrozenSet,
    Tuple,
)
from collections.abc import Callable, Mapping, Iterable, Container

from debian.deb822 import Deb822

from debputy.commands.debputy_cmd.context import CommandContext
from debputy.commands.debputy_cmd.output import IOBasedOutputStyling
from debputy.dh.debhelper_emulation import CannotEmulateExecutableDHConfigFile
from debputy.dh.dh_assistant import read_dh_addon_sequences
from debputy.dh_migration.migrators import Migrator, MigrationTarget
from debputy.dh_migration.migrators_impl import (
    INTEGRATION_MODE_DH_DEBPUTY,
    INTEGRATION_MODE_DH_DEBPUTY_RRR,
)
from debputy.dh_migration.models import (
    FeatureMigration,
    AcceptableMigrationIssues,
    UnsupportedFeature,
    ConflictingChange,
    MigrationRequest,
)
from debputy.highlevel_manifest import HighLevelManifest
from debputy.integration_detection import determine_debputy_integration_mode
from debputy.manifest_parser.exceptions import ManifestParseException
from debputy.plugin.api import VirtualPath
from debputy.plugin.api.spec import DebputyIntegrationMode, INTEGRATION_MODE_FULL
from debputy.util import _error, _warn, _info, escape_shell, assume_not_none

SUPPORTED_MIGRATIONS: Mapping[MigrationTarget, frozenset[MigrationTarget]] = {
    INTEGRATION_MODE_FULL: frozenset([INTEGRATION_MODE_FULL]),
    INTEGRATION_MODE_DH_DEBPUTY: frozenset(
        [
            "dh-single-to-multi-binary",
            "dh-package-prefixed-config-files",
            INTEGRATION_MODE_DH_DEBPUTY,
            INTEGRATION_MODE_FULL,
        ]
    ),
    INTEGRATION_MODE_DH_DEBPUTY_RRR: frozenset(
        [
            "dh-single-to-multi-binary",
            "dh-package-prefixed-config-files",
            INTEGRATION_MODE_DH_DEBPUTY_RRR,
            INTEGRATION_MODE_DH_DEBPUTY,
            INTEGRATION_MODE_FULL,
        ]
    ),
}


def _print_migration_summary(
    fo: IOBasedOutputStyling,
    migrations: list[FeatureMigration],
    compat: int,
    min_compat_level: int,
    required_plugins: set[str],
    requested_plugins: set[str] | None,
) -> None:
    warning_count = 0

    for migration in migrations:
        if not migration.anything_to_do:
            continue
        underline = "-" * len(migration.tagline)
        if migration.warnings:
            if warning_count:
                _warn("")
            _warn(f"Summary for migration: {migration.tagline}")
            if not fo.optimize_for_screen_reader:
                _warn(f"-----------------------{underline}")
            warning_count += len(migration.warnings)
            for warning in migration.warnings:
                _warn(f"    * {warning}")

    if compat < min_compat_level:
        if warning_count:
            _warn("")
        _warn("Supported debhelper compat check")
        if not fo.optimize_for_screen_reader:
            _warn("--------------------------------")
        warning_count += 1
        _warn(
            f"The migration tool assumes debhelper compat {min_compat_level}+ semantics, but this package"
            f" is using compat {compat}.  Consider upgrading the package to compat {min_compat_level}"
            " first."
        )

    if required_plugins:
        if requested_plugins is None:
            warning_count += 1
            needed_plugins = ", ".join(f"debputy-plugin-{n}" for n in required_plugins)
            if warning_count:
                _warn("")
            _warn("Missing debputy plugin check")
            if not fo.optimize_for_screen_reader:
                _warn("----------------------------")
            _warn(
                f"The migration tool could not read d/control and therefore cannot tell if all the required"
                f" plugins have been requested.  Please ensure that the package Build-Depends on: {needed_plugins}"
            )
        else:
            missing_plugins = required_plugins - requested_plugins
            if missing_plugins:
                warning_count += 1
                needed_plugins = ", ".join(
                    f"debputy-plugin-{n}" for n in missing_plugins
                )
                if warning_count:
                    _warn("")
                _warn("Missing debputy plugin check")
                if not fo.optimize_for_screen_reader:
                    _warn("----------------------------")
                _warn(
                    f"The migration tool asserted that the following `debputy` plugins would be required, which"
                    f" are not explicitly requested.  Please add the following to Build-Depends: {needed_plugins}"
                )

    if warning_count:
        _warn("")
        _warn(
            f"/!\\ Total number of warnings or manual migrations required: {warning_count}"
        )


def _dh_compat_level() -> int | None:
    try:
        res = subprocess.check_output(
            ["dh_assistant", "active-compat-level"], stderr=subprocess.DEVNULL
        )
    except subprocess.CalledProcessError:
        compat = None
    else:
        try:
            compat = json.loads(res)["declared-compat-level"]
        except RuntimeError:
            compat = None
        else:
            if not isinstance(compat, int):
                compat = None
    return compat


def _requested_debputy_plugins(debian_dir: VirtualPath) -> set[str] | None:
    ctrl_file = debian_dir.get("control")
    if not ctrl_file:
        return None

    dep_regex = re.compile("^([a-z0-9][-+.a-z0-9]+)", re.ASCII)
    plugins = set()

    with ctrl_file.open() as fd:
        ctrl = list(Deb822.iter_paragraphs(fd))
    source_paragraph = ctrl[0] if ctrl else {}

    for f in ("Build-Depends", "Build-Depends-Indep", "Build-Depends-Arch"):
        field = source_paragraph.get(f)
        if not field:
            continue

        for dep_clause in (d.strip() for d in field.split(",")):
            match = dep_regex.match(dep_clause.strip())
            if not match:
                continue
            dep = match.group(1)
            if not dep.startswith("debputy-plugin-"):
                continue
            plugins.add(dep[15:])
    return plugins


def _check_migration_target(
    context: CommandContext,
    migration_target: DebputyIntegrationMode | None,
) -> DebputyIntegrationMode:
    r = read_dh_addon_sequences(context.debian_dir)
    if r is not None:
        bd_sequences, dr_sequences, _ = r
        all_sequences = bd_sequences | dr_sequences
        detected_migration_target = determine_debputy_integration_mode(
            context.source_package().fields,
            all_sequences,
        )
    else:
        detected_migration_target = None

    if migration_target is not None and detected_migration_target is not None:
        supported_migrations = SUPPORTED_MIGRATIONS.get(
            detected_migration_target,
            frozenset([detected_migration_target]),
        )

        if (
            migration_target != detected_migration_target
            and migration_target not in supported_migrations
        ):
            _error(
                f"Cannot migrate apply migration {migration_target} as it conflicts with the current state (which is {detected_migration_target})"
            )

    if migration_target is not None:
        resolved_migration_target = migration_target
        _info(f'Using "{resolved_migration_target}" as migration (requested)')
    else:
        if detected_migration_target is not None:
            _info(
                f'Using "{detected_migration_target}" as migration (based on the packaging)'
            )
        else:
            detected_migration_target = INTEGRATION_MODE_DH_DEBPUTY
            _info(
                f'Using "{detected_migration_target}" as default migration target. Use --migration-target to choose!'
            )
        resolved_migration_target = detected_migration_target

    return resolved_migration_target


def _read_git_status_lines(
    lines: Iterable[bytes],
) -> Iterable[tuple[str, str, str | None]]:
    line_iter = iter(lines)
    while True:
        try:
            line = next(line_iter)
        except StopIteration:
            return
        if not line:
            # We get a final empty line
            continue
        if len(line) < 4:
            _error(
                f"Internal error: Got out of sync with the `git status --porcelain=v1 -z` output ({len(line)=})"
            )
        status_marker = line[0:2].decode("utf-8")
        if line[2] != 32:  # Space
            _error(
                "Internal error: Got out of sync with the `git status --porcelain=v1 -z` output (expected status + filename)"
            )
        if status_marker.startswith(("R", "C")) or status_marker.endswith(("R", "C")):
            try:
                second_filename = next(line_iter).decode("utf-8")
            except StopIteration:
                _error(
                    "Internal error: Expected one more line of output from `git status` but it was not there"
                )
        else:
            second_filename = None
        filename = line[3:].decode("utf-8")
        yield status_marker, filename, second_filename


def _git_status() -> tuple[bool, Container[str]]:
    try:
        top_level = (
            subprocess.check_output(
                ["git", "rev-parse", "--show-toplevel"],
                stderr=subprocess.DEVNULL,
                cwd="debian",
            )
            .strip()
            .decode("utf-8")
        )
    except (subprocess.CalledProcessError, FileNotFoundError):
        return False, frozenset()

    if os.path.realpath(os.getcwd()) != os.path.realpath(top_level):
        # Patches welcome. The primary problem is to have the status output match the `_check_vcs_clashes` method.
        _error(
            f"Unsupported git repo: The `debputy` command only supports `git` when cwd is the git root (git root is: {top_level}). Please use --ignore-vcs to continue."
        )

    try:
        r = subprocess.check_output(
            ["git", "status", "--porcelain=v1", "-z"],
            stderr=subprocess.DEVNULL,
            cwd="debian",
        )
    except subprocess.CalledProcessError:
        _error(
            "The `git status --porcelain=v1 -z` command returned non-zero (note: the command outputs binary). Please use `--ignore-vcs` to continue anyway."
        )
    except FileNotFoundError:
        _error(
            "Could not run `git status` and there is a `.git` directory. Please use `--ignore-vcs` to continue anyway."
        )

    untracked_files = set()
    for status_marker, filename, second_filename in _read_git_status_lines(
        r.split(b"\0")
    ):
        if status_marker not in ("??", "!!"):
            _error(
                "The current git tree is not clean, please commit or stash any pending changes or use `--ignore-vcs` to continue anyway."
            )
        # There is never a second file for `??` or `!!`
        assert second_filename is None
        untracked_files.add(filename)
    return True, untracked_files


def _check_vcs_clashes(
    manifest: HighLevelManifest,
    migrations: list[FeatureMigration],
    untracked_files: Container[str],
) -> None:
    all_affected_files = []
    if any(m.successful_manifest_changes for m in migrations):
        all_affected_files.append(manifest.manifest_path)

    for previous_path, new_path in (
        p for m in migrations for p in m.rename_paths_on_success
    ):
        all_affected_files.append(previous_path)
        all_affected_files.append(new_path)

    all_affected_files.extend(p for m in migrations for p in m.remove_paths_on_success)

    clashing_paths = [p for p in all_affected_files if p in untracked_files]

    if clashing_paths:
        print(file=sys.stderr)
        _warn(
            "The following untracked or ignored paths would be affected by this migration"
        )
        _warn("")
        for path in clashing_paths:
            _warn(f" * {path}")
        _warn("")
        _warn("Since they are not tracked, the VCS cannot undo the changes to them")
        _warn("if the migration continued")
        print(file=sys.stderr)
        _error(
            "Please stash, commit or remove the above files before continuing. Alternatively, use --ignore-vcs"
        )


def perform_migration(
    fo: IOBasedOutputStyling,
    manifest: HighLevelManifest,
    acceptable_migration_issues: AcceptableMigrationIssues,
    permit_destructive_changes: bool | None,
    migration_target: DebputyIntegrationMode,
    manifest_parser_factory: Callable[[str], HighLevelManifest],
    migrators: list[Migrator],
    *,
    ignore_vcs: bool = False,
) -> None:
    migrations = []
    compat = _dh_compat_level()
    if compat is None:
        _error(
            'Cannot detect declared compat level (try running "dh_assistant active-compat-level")'
        )

    debian_dir = manifest.debian_dir
    mutable_manifest = assume_not_none(manifest.mutable_manifest)

    if not ignore_vcs:
        uses_git, untracked_files = _git_status()
    else:
        uses_git = False
        untracked_files = frozenset()

    if permit_destructive_changes is None and uses_git:
        permit_destructive_changes = True

    migration_request = MigrationRequest(
        debian_dir,
        manifest,
        acceptable_migration_issues,
        migration_target,
    )

    try:
        for migrator in migrators:
            feature_migration = FeatureMigration(migrator.__name__, fo)
            migrator(migration_request, feature_migration)
            migrations.append(feature_migration)
    except CannotEmulateExecutableDHConfigFile as e:
        _error(
            f"Unable to process the executable dh config file {e.config_file().fs_path}: {e.message()}"
        )
    except UnsupportedFeature as e:
        msg = (
            f"Unable to migrate automatically due to missing features in debputy. The feature is:"
            f"\n\n  * {e.message}"
        )
        keys = e.issue_keys
        if keys:
            primary_key = keys[0]
            alt_keys = ""
            if len(keys) > 1:
                alt_keys = (
                    f' Alternatively you can also use one of: {", ".join(keys[1:])}.  Please note that some'
                    " of these may cover more cases."
                )
            msg += (
                f"\n\nUse --acceptable-migration-issues={primary_key} to convert this into a warning and try again."
                " However, you should only do that if you believe you can replace the functionality manually"
                f" or the usage is obsolete / can be removed. {alt_keys}"
            )
        _error(msg)
    except ConflictingChange as e:
        _error(
            "The migration tool detected a conflict data being migrated and data already migrated / in the existing"
            "manifest."
            f"\n\n   * {e.message}"
            "\n\nPlease review the situation and resolve the conflict manually."
        )

    min_compat = max(
        (m.assumed_compat for m in migrations if m.assumed_compat is not None),
        default=0,
    )

    if compat < min_compat and "min-compat-level" not in acceptable_migration_issues:
        # The migration summary special-cases the compat mismatch and warns for us.
        _error(
            f"The migration tool assumes debhelper compat {min_compat} or later but the package is only on"
            f" compat {compat}.  This may lead to incorrect result."
            f"\n\nUse --acceptable-migration-issues=min-compat-level to convert this into a warning and"
            f" try again, if you want to continue regardless."
        )

    requested_plugins = _requested_debputy_plugins(debian_dir)
    required_plugins: set[str] = set()
    required_plugins.update(
        chain.from_iterable(
            m.required_plugins for m in migrations if m.required_plugins
        )
    )

    if uses_git:
        _check_vcs_clashes(manifest, migrations, untracked_files)

    _print_migration_summary(
        fo,
        migrations,
        compat,
        min_compat,
        required_plugins,
        requested_plugins,
    )
    migration_count = sum((m.performed_changes for m in migrations), 0)

    if not migration_count:
        _info(
            "debputy was not able to find any (supported) migrations that it could perform for you."
        )
        return

    if any(m.successful_manifest_changes for m in migrations):
        new_manifest_path = manifest.manifest_path + ".new"

        with open(new_manifest_path, "w") as fd:
            mutable_manifest.write_to(fd)

        try:
            _info("Verifying the new manifest")
            manifest_parser_factory(new_manifest_path)
        except ManifestParseException as e:
            raise AssertionError(
                "Could not parse the manifest generated from the migrator"
            ) from e

        if permit_destructive_changes:
            if os.path.isfile(manifest.manifest_path):
                os.rename(manifest.manifest_path, manifest.manifest_path + ".orig")
            os.rename(new_manifest_path, manifest.manifest_path)
            if uses_git:
                subprocess.check_call(["git", "add", manifest.manifest_path])
            _info(f"Updated manifest {manifest.manifest_path}")
        else:
            _info(
                f'Created draft manifest "{new_manifest_path}" (rename to "{manifest.manifest_path}"'
                " to activate it)"
            )
    else:
        _info("No manifest changes detected; skipping update of manifest.")

    removals: int = sum((len(m.remove_paths_on_success) for m in migrations), 0)
    renames: int = sum((len(m.rename_paths_on_success) for m in migrations), 0)

    if renames:
        if permit_destructive_changes:
            _info("Paths being renamed:")
        else:
            _info("Migration *would* rename the following paths:")
        prefix = "git " if uses_git else ""
        for previous_path, new_path in (
            p for m in migrations for p in m.rename_paths_on_success
        ):
            _info(f"   {prefix}mv {escape_shell(previous_path, new_path)}")

    if removals:
        if permit_destructive_changes:
            _info("Removals:")
        else:
            _info("Migration *would* remove the following files:")
        cmd = "git rm" if uses_git else "rm -f"
        for path in (p for m in migrations for p in m.remove_paths_on_success):
            _info(f"  {cmd} {escape_shell(path)}")

    if permit_destructive_changes is None:
        print()
        _info(
            "If you would like to perform the migration, please re-run with --apply-changes."
        )
    elif permit_destructive_changes:
        _apply_renames_and_removals(migrations, uses_git)

        print()
        _info("Migrations performed successfully")
        print()
        _info(
            "Remember to validate the resulting binary packages after rebuilding with debputy"
        )
        if uses_git:
            _info(
                "Use `git commit` to commit the changes or use `git reset` / `git restore` to undo them."
            )
    else:
        print()
        _info("No migrations performed as requested")


def _apply_renames_and_removals(
    migrations: list[FeatureMigration],
    uses_git: bool,
) -> None:
    for previous_path, new_path in (
        p for m in migrations for p in m.rename_paths_on_success
    ):
        if uses_git:
            subprocess.check_call(["git", "mv", previous_path, new_path])
        else:
            os.rename(previous_path, new_path)

    files_being_removed = [p for m in migrations for p in m.remove_paths_on_success]
    if uses_git and files_being_removed:
        command = ["git", "rm"]
        command.extend(files_being_removed)
        subprocess.check_call(command)
    elif files_being_removed:
        for path in (p for m in migrations for p in m.remove_paths_on_success):
            os.unlink(path)