import abc
import os
import stat
from collections import deque
from collections.abc import MutableMapping

from . import jsonlib
from .utils import git

# Cannot do `from ..gitignore import gitignore` because
# relative import beyond toplevel throws *ImportError*!
from gitignore import gitignore  # type: ignore


MYPY = False
if MYPY:
    # MYPY is set to True when run under Mypy.
    from typing import Dict, Optional, List, Set, Text, Iterable, Any, Tuple, Iterator
    from .manifest import Manifest  # cyclic import under MYPY guard
    stat_result = os.stat_result

    GitIgnoreCacheType = MutableMapping[bytes, bool]
else:
    GitIgnoreCacheType = MutableMapping


def get_tree(tests_root, manifest, manifest_path, cache_root,
             working_copy=True, rebuild=False):
    # type: (Text, Manifest, Optional[Text], Optional[Text], bool, bool) -> FileSystem
    tree = None
    if cache_root is None:
        cache_root = os.path.join(tests_root, u".wptcache")
    if not os.path.exists(cache_root):
        try:
            os.makedirs(cache_root)
        except IOError:
            cache_root = None

    if not working_copy:
        raise ValueError("working_copy=False unsupported")

    if tree is None:
        tree = FileSystem(tests_root,
                          manifest.url_base,
                          manifest_path=manifest_path,
                          cache_path=cache_root,
                          rebuild=rebuild)
    return tree


class GitHasher(object):
    def __init__(self, path):
        # type: (Text) -> None
        self.git = git(path)

    def _local_changes(self):
        # type: () -> Set[Text]
        """get a set of files which have changed between HEAD and working copy"""
        assert self.git is not None
        # note that git runs the command with tests_root as the cwd, which may
        # not be the root of the git repo (e.g., within a browser repo)
        cmd = ["diff-index", "--relative", "--no-renames", "--name-only", "-z", "HEAD"]
        data = self.git(*cmd)
        return set(data.split("\0"))

    def hash_cache(self):
        # type: () -> Dict[Text, Optional[Text]]
        """
        A dict of rel_path -> current git object id if the working tree matches HEAD else None
        """
        hash_cache = {}  # type: Dict[Text, Optional[Text]]

        if self.git is None:
            return hash_cache

        # note that git runs the command with tests_root as the cwd, which may
        # not be the root of the git repo (e.g., within a browser repo)
        cmd = ["ls-tree", "-r", "-z", "HEAD"]
        local_changes = self._local_changes()
        for result in self.git(*cmd).split("\0")[:-1]:  # type: Text
            data, rel_path = result.rsplit("\t", 1)
            hash_cache[rel_path] = None if rel_path in local_changes else data.split(" ", 3)[2]

        return hash_cache


class FileSystem(object):
    def __init__(self, tests_root, url_base, cache_path, manifest_path=None, rebuild=False):
        # type: (Text, Text, Optional[Text], Optional[Text], bool) -> None
        self.tests_root = tests_root
        self.url_base = url_base
        self.ignore_cache = None
        self.mtime_cache = None
        tests_root_bytes = tests_root.encode("utf8")
        if cache_path is not None:
            if manifest_path is not None:
                self.mtime_cache = MtimeCache(cache_path, tests_root, manifest_path, rebuild)
            if gitignore.has_ignore(tests_root_bytes):
                self.ignore_cache = GitIgnoreCache(cache_path, tests_root, rebuild)
        self.path_filter = gitignore.PathFilter(tests_root_bytes,
                                                extras=[b".git/"],
                                                cache=self.ignore_cache)
        git = GitHasher(tests_root)
        if git is not None:
            self.hash_cache = git.hash_cache()
        else:
            self.hash_cache = {}

    def __iter__(self):
        # type: () -> Iterator[Tuple[Text, Optional[Text], bool]]
        mtime_cache = self.mtime_cache
        for dirpath, dirnames, filenames in self.path_filter(
                walk(self.tests_root.encode("utf8"))):
            for filename, path_stat in filenames:
                path = os.path.join(dirpath, filename).decode("utf8")
                if mtime_cache is None or mtime_cache.updated(path, path_stat):
                    file_hash = self.hash_cache.get(path, None)
                    yield path, file_hash, True
                else:
                    yield path, None, False

    def dump_caches(self):
        # type: () -> None
        for cache in [self.mtime_cache, self.ignore_cache]:
            if cache is not None:
                cache.dump()


class CacheFile(metaclass=abc.ABCMeta):
    def __init__(self, cache_root, tests_root, rebuild=False):
        # type: (Text, Text, bool) -> None
        self.tests_root = tests_root
        if not os.path.exists(cache_root):
            os.makedirs(cache_root)
        self.path = os.path.join(cache_root, self.file_name)
        self.modified = False
        self.data = self.load(rebuild)

    @abc.abstractproperty
    def file_name(self):
        # type: () -> Text
        pass

    def dump(self):
        # type: () -> None
        if not self.modified:
            return
        with open(self.path, 'w') as f:
            jsonlib.dump_local(self.data, f)

    def load(self, rebuild=False):
        # type: (bool) -> Dict[Text, Any]
        data = {}  # type: Dict[Text, Any]
        try:
            if not rebuild:
                with open(self.path, 'r') as f:
                    try:
                        data = jsonlib.load(f)
                    except ValueError:
                        pass
                data = self.check_valid(data)
        except IOError:
            pass
        return data

    def check_valid(self, data):
        # type: (Dict[Text, Any]) -> Dict[Text, Any]
        """Check if the cached data is valid and return an updated copy of the
        cache containing only data that can be used."""
        return data


class MtimeCache(CacheFile):
    file_name = u"mtime.json"

    def __init__(self, cache_root, tests_root, manifest_path, rebuild=False):
        # type: (Text, Text, Text, bool) -> None
        self.manifest_path = manifest_path
        super(MtimeCache, self).__init__(cache_root, tests_root, rebuild)

    def updated(self, rel_path, stat):
        # type: (Text, stat_result) -> bool
        """Return a boolean indicating whether the file changed since the cache was last updated.

        This implicitly updates the cache with the new mtime data."""
        mtime = stat.st_mtime
        if mtime != self.data.get(rel_path):
            self.modified = True
            self.data[rel_path] = mtime
            return True
        return False

    def check_valid(self, data):
        # type: (Dict[Any, Any]) -> Dict[Any, Any]
        if data.get(u"/tests_root") != self.tests_root:
            self.modified = True
        else:
            if self.manifest_path is not None and os.path.exists(self.manifest_path):
                mtime = os.path.getmtime(self.manifest_path)
                if data.get(u"/manifest_path") != [self.manifest_path, mtime]:
                    self.modified = True
            else:
                self.modified = True
        if self.modified:
            data = {}
            data["/tests_root"] = self.tests_root
        return data

    def dump(self):
        # type: () -> None
        if self.manifest_path is None:
            raise ValueError
        if not os.path.exists(self.manifest_path):
            return
        mtime = os.path.getmtime(self.manifest_path)
        self.data["/manifest_path"] = [self.manifest_path, mtime]
        self.data["/tests_root"] = self.tests_root
        super(MtimeCache, self).dump()


class GitIgnoreCache(CacheFile, GitIgnoreCacheType):
    file_name = "gitignore2.json"

    def check_valid(self, data):
        # type: (Dict[Any, Any]) -> Dict[Any, Any]
        ignore_path = os.path.join(self.tests_root, ".gitignore")
        mtime = os.path.getmtime(ignore_path)
        if data.get(u"/gitignore_file") != [ignore_path, mtime]:
            self.modified = True
            data = {}
            data[u"/gitignore_file"] = [ignore_path, mtime]
        return data

    def __contains__(self, key):
        # type: (Any) -> bool
        try:
            key = key.decode("utf-8")
        except Exception:
            return False

        return key in self.data

    def __getitem__(self, key):
        # type: (bytes) -> bool
        real_key = key.decode("utf-8")
        v = self.data[real_key]
        assert isinstance(v, bool)
        return v

    def __setitem__(self, key, value):
        # type: (bytes, bool) -> None
        real_key = key.decode("utf-8")
        if self.data.get(real_key) != value:
            self.modified = True
            self.data[real_key] = value

    def __delitem__(self, key):
        # type: (bytes) -> None
        real_key = key.decode("utf-8")
        del self.data[real_key]

    def __iter__(self):
        # type: () -> Iterator[bytes]
        return (key.encode("utf-8") for key in self.data)

    def __len__(self):
        # type: () -> int
        return len(self.data)


def walk(root):
    # type: (bytes) -> Iterable[Tuple[bytes, List[Tuple[bytes, stat_result]], List[Tuple[bytes, stat_result]]]]
    """Re-implementation of os.walk. Returns an iterator over
    (dirpath, dirnames, filenames), with some semantic differences
    to os.walk.

    This has a similar interface to os.walk, with the important difference
    that instead of lists of filenames and directory names, it yields
    lists of tuples of the form [(name, stat)] where stat is the result of
    os.stat for the file. That allows reusing the same stat data in the
    caller. It also always returns the dirpath relative to the root, with
    the root iself being returned as the empty string.

    Unlike os.walk the implementation is not recursive."""

    get_stat = os.stat
    is_dir = stat.S_ISDIR
    is_link = stat.S_ISLNK
    join = os.path.join
    listdir = os.listdir
    relpath = os.path.relpath

    root = os.path.abspath(root)
    stack = deque([(root, b"")])

    while stack:
        dir_path, rel_path = stack.popleft()
        try:
            # Note that listdir and error are globals in this module due
            # to earlier import-*.
            names = listdir(dir_path)
        except OSError:
            continue

        dirs, non_dirs = [], []
        for name in names:
            path = join(dir_path, name)
            try:
                path_stat = get_stat(path)
            except OSError:
                continue
            if is_dir(path_stat.st_mode):
                dirs.append((name, path_stat))
            else:
                non_dirs.append((name, path_stat))

        yield rel_path, dirs, non_dirs
        for name, path_stat in dirs:
            new_path = join(dir_path, name)
            if not is_link(path_stat.st_mode):
                stack.append((new_path, relpath(new_path, root)))