#!/usr/bin/env python3
# Copyright 2022 The Fuchsia Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""Generate a content hash file from one or more source repository content.

By default, scan all files in the source paths (file or directory) and hash
their content.

- For symlinks, the link value is used as input (not the target file).

- For regular files, its sha1 digest is used as input.

- For git directories (which have a .git entry), use the HEAD commit as input
  to speed things dramatically. NOTE: This ignores changes to the index,
  during development.

- If --cipd-name=NAME is set, and <source_path>/.versions/NAME.cipd_version
  exists, its content will be used as input.

- Otherwise, for directories, all files in it are found recursively,
  and used as hash input independently.
"""

import argparse
import hashlib
import os
import sys
import typing as T
from pathlib import Path

_HASH = "sha1"

sys.path.insert(0, str(Path(__file__).parent))
import get_git_head_commit as gghc


def _depfile_quote(p: Path | str) -> str:
    """Quote a Path value for depfile output."""
    return str(p).replace("\\", "\\\\").replace(" ", "\\ ")


class FileState(object):
    """State object used to hash one or more source paths.

    Usage is:
       - Create instance
       - Call hash_source_path() as many times as possible.
       - Use content_hash property to get final result.
       - Use sorted_input_files to get list of input files.
    """

    def __init__(
        self,
        cipd_names: T.Sequence[str] = [],
        exclude_suffixes: T.Sequence[str] = [],
        git_binary: Path = Path("git"),
    ) -> None:
        """Create new instance.

        Args:
            cipd_names: A sequence of cipd names for prebuilt directories.
            exclude_suffixes: A sequence of filename suffixes to exclude from hashing.
            git_binary: Path to the git binary to use for .git repositories.
        """
        self._cipd_names = cipd_names
        self._exclude_suffixes = tuple(exclude_suffixes)
        self._git_binary = git_binary
        self._input_files: set[Path] = set()
        self._sorted_input_files: T.Optional[list[str]] = None
        self._hstate = hashlib.new(_HASH)

    def hash_source_path(self, source_path: Path) -> None:
        """Process and hash a given source file, updating internal state."""
        self._hstate.update(self.process_source_path(source_path).encode())

    def find_directory_files(self, source_path: Path) -> set[Path]:
        source_path.is_dir(), f"Input source path is not a directory: {source_path}"

        if self._cipd_names:
            for cipd_name in self._cipd_names:
                clang_version_file = (
                    source_path / ".versions" / f"{cipd_name}.cipd_version"
                )
                if clang_version_file.exists():
                    return set([clang_version_file])

        # Find all files in direcrory.
        dir_files: set[Path] = set()
        for dirpath, dirnames, filenames in os.walk(source_path):
            for filename in filenames:
                if filename.endswith(self._exclude_suffixes):
                    continue
                file_path = Path(os.path.join(dirpath, filename))
                dir_files.add(file_path)

        return dir_files

    def process_source_path(self, source_path: Path) -> str:
        """Process a given source file, and return a string descriptor for it.

        The first letter of the result corresponds to the type of the source path.
        This function is useful for unit-testing the implementation and verify
        that different types of source paths are handled correctly. Apart from
        that, consider this as an implementation detail.
        """
        if not source_path.exists():
            raise ValueError(f"Path does not exist: {source_path}")

        if source_path.is_dir() and (source_path / ".git").exists():
            head_commit = gghc.get_git_head_commit(
                source_path, self._git_binary
            )
            self._input_files.update(gghc.find_git_head_inputs(source_path))
            return "G" + head_commit

        if source_path.is_symlink():
            self._input_files.add(source_path)
            return "S" + str(source_path.readlink())

        if source_path.is_file():
            self._input_files.add(source_path)
            with source_path.open("rb") as f:
                digest = hashlib.file_digest(f, _HASH)
            return "F" + digest.hexdigest()

        assert source_path.is_dir(), f"Unexpected file type for {source_path}"

        if self._cipd_names:
            for cipd_name in self._cipd_names:
                clang_version_file = (
                    source_path / ".versions" / f"{cipd_name}.cipd_version"
                )
                if clang_version_file.exists():
                    return self.process_source_path(clang_version_file)

        # Get the list of files relative to the source directory.
        dir_files: list[str] = [
            os.path.relpath(f, source_path)
            for f in self.find_directory_files(source_path)
        ]

        # Process them recursively to build a directory description text
        # where each line looks like: <file> <type><digest>
        dir_content = "D\n"
        for dir_file in sorted(dir_files):
            file_hash = self.process_source_path(source_path / dir_file)
            dir_content += f" {dir_file} {file_hash}\n"
        return dir_content

    @property
    def content_hash(self) -> str:
        """Return final content hash as hexadecimal string."""
        return self._hstate.hexdigest()

    @property
    def sorted_input_files(self) -> list[str]:
        """Return the list of input files used by this instance."""
        if self._sorted_input_files is None:
            self._sorted_input_files = sorted(
                [str(p) for p in self._input_files]
            )
        return self._sorted_input_files

    def get_input_file_paths(self) -> set[Path]:
        """Return the set of input file Path values used by this instance."""
        return self._input_files


def main() -> int:
    parser = argparse.ArgumentParser(
        description=__doc__, formatter_class=argparse.RawTextHelpFormatter
    )
    parser.add_argument(
        "--cipd-name",
        action="append",
        default=[],
        help="Provide name for optional CIPD version file. Can be used multiple times.",
    )
    parser.add_argument(
        "--exclude-suffix",
        action="append",
        default=[],
        help='Exclude directory entries with given suffix (e.g. ".pyc").\n'
        + "Can be used multiple times.",
    )
    parser.add_argument(
        "--git-binary",
        type=Path,
        default=Path("git"),
        help="Specify git binary to use for git repositories.",
    )
    parser.add_argument(
        "source_path",
        type=Path,
        nargs="+",
        help="Source file or directory path.",
    )
    parser.add_argument(
        "--output", type=Path, help="Optional output file path."
    )
    parser.add_argument(
        "--depfile", type=Path, help="Optional Ninja depfile output file path."
    )
    parser.add_argument(
        "--inputs-list",
        type=Path,
        help="Write list of inputs to file, one path per line.",
    )
    args = parser.parse_args()

    if args.depfile and not args.output:
        parser.error("--depfile option requires --output.")

    fstate = FileState(args.cipd_name, args.exclude_suffix, args.git_binary)
    for source_path in args.source_path:
        fstate.hash_source_path(source_path)

    if args.output:
        # Do not modify existing output if it has the same content.
        current_content = "~~~"
        if args.output.exists():
            current_content = args.output.read_text()
        if current_content != fstate.content_hash:
            args.output.write_text(fstate.content_hash)
    else:
        print(fstate.content_hash)

    if args.inputs_list:
        args.inputs_list.write_text("\n".join(fstate.sorted_input_files) + "\n")

    if args.depfile:
        args.depfile.write_text(
            "%s: \\\n  %s\n"
            % (
                args.output,
                " \\\n  ".join(
                    _depfile_quote(f) for f in fstate.sorted_input_files
                ),
            )
        )

    return 0


if __name__ == "__main__":
    sys.exit(main())