# Copyright 2024 The Chromium Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

import os
import sys
import glob
import constants
import multiprocessing.dummy
import tempfile
from typing import Dict, Callable, List, Set
from pathlib import Path

from license_type import LicenseType
import license_utils

REPOSITORY_ROOT = os.path.abspath(
    os.path.join(os.path.dirname(__file__), os.pardir, os.pardir, os.pardir))

sys.path.insert(0, REPOSITORY_ROOT)
import components.cronet.gn2bp.targets as gn2bp_targets  # pylint: disable=wrong-import-position
import components.cronet.tools.utils as cronet_utils  # pylint: disable=wrong-import-position

_OUT_DIR = os.path.join(REPOSITORY_ROOT, "out")

# The license content of all transitive dependencies of the targets listed here will be added
# to the top-level LICENSE.
_TARGETS_TO_AGGREGATE_LICENSE_FOR_TRANSITIVELY = [
    "//components/cronet/android:cronet_non_test_package"
]
# The license generator will generate license file for all the targets found as a
# transitive dependency of this target.
# For each transitive dependency with a README.chromium, its license file will be
# processed.
_TARGETS_TO_PROCESS_LICENSE_FOR_TRANSITIVELY = (
    ["//components/cronet/android:cronet_package_android"] +
    _TARGETS_TO_AGGREGATE_LICENSE_FOR_TRANSITIVELY +
    gn2bp_targets.DEFAULT_TARGETS + gn2bp_targets.DEFAULT_TESTS)

METADATA_HEADER = """# This was automatically generated by {}
# This directory was imported from Chromium.""".format(
    os.path.basename(__file__))

_ROOT_CRONET = os.path.abspath(
    os.path.join(os.path.dirname(__file__), os.path.pardir, os.path.pardir,
                 os.path.pardir))


def _create_metadata_file(repo_path: str, directory_path: str, content: str,
                          verify_only: bool):
    """Creates a METADATA file with a header to ensure that this was generated
  through the script. If the header is not found then it is assumed that the
  METADATA file is created manually and will not be touched."""
    metadata = Path(os.path.join(directory_path, "METADATA"))
    if metadata.is_file() and METADATA_HEADER not in metadata.read_text():
        # This is a manually created file! Don't overwrite.
        return

    metadata_content = "\n".join([METADATA_HEADER, content])
    if verify_only:
        if not metadata.exists():
            raise Exception(
                f"Failed to find metadata file {metadata.relative_to(repo_path)}"
            )
        if not metadata.read_text() == metadata_content:
            raise Exception(
                f"Metadata content of {metadata.relative_to(repo_path)} does not match the expected."
                f"Please re-run create_android_metadata_license.py")
    else:
        metadata.write_text(metadata_content)


def _create_module_license_file(repo_path: str, directory_path: str,
                                licenses_names: List[str], verify_only: bool):
    """Creates a MODULE_LICENSE_XYZ files."""
    for license_name in licenses_names:
        license_file = Path(
            os.path.join(
                directory_path,
                f"MODULE_LICENSE_{license_utils.get_license_file_format(license_name)}"
            ))
        if verify_only:
            if not license_file.exists():
                raise Exception(
                    f"Failed to find module file {license_file.relative_to(repo_path)}"
                )
        else:
            license_file.touch()


def _maybe_create_license_file_symlink(directory_path: str,
                                       original_license_file: str,
                                       verify_only: bool):
    """Creates a LICENSE symbolic link only if it doesn't exist."""
    license_symlink_path = Path(os.path.join(directory_path, "LICENSE"))
    if license_symlink_path.exists():
        # The symlink is already there, skip.
        return

    if verify_only:
        if not license_symlink_path.exists():
            raise Exception(
                f"License symlink does not exist for {license_symlink_path}")
    else:
        # license_symlink_path.relative_to(.., walk_up=True) does not exist in
        # Python 3.10, this is the reason why os.path.relpath is used.
        os.symlink(
            os.path.relpath(original_license_file,
                            license_symlink_path.parent), license_symlink_path)


def _map_rust_license_path_to_directory(license_file_path: str) -> str:
    """ Returns the canonical path of the parent directory that includes
  the LICENSE file for rust crates.

  :param license_file_path: This is the filepath found in the README.chromium
  and the expected format is //some/path/license_file
  """
    if not license_file_path.startswith("//"):
        raise ValueError(
            f"Rust third-party crate's `License File` is expected to be absolute path "
            f"(Absolute GN labels are expected to start with //), "
            f"but found {license_file_path}")
    return license_file_path[2:license_file_path.rfind("/")]


def get_all_readme(repo_path: str):
    """Fetches all README.chromium files under |repo_path|."""
    return glob.glob("**/README.chromium", root_dir=repo_path, recursive=True)  # pylint: disable=unexpected-keyword-arg


def get_all_readme_through_gn(repo_path: str, targets: Set[str]):
    processed_build_files_path = set()
    for build_file_path in _get_all_build_files_path(repo_path, targets):
        processed_build_files_path.add(
            os.path.dirname(os.path.relpath(build_file_path, repo_path)))

    readme_files = set()
    for directory_path in processed_build_files_path:
        # The python version on the trybots does not support `root_dir` so we have
        # to work around this by appending it to the glob path then do a relpath.
        readme_files.update((os.path.relpath(
            readme_file_path, repo_path
        ) for readme_file_path in glob.glob(
            f"{os.path.join(repo_path, gn2bp_targets.README_MAPPING.get(directory_path, directory_path))}/**README.chromium",
            recursive=True)))

    return readme_files | constants.INCLUDED_README


def _get_build_files_paths(repo_path, targets, arch):
    with tempfile.TemporaryDirectory(dir=_OUT_DIR) as gn_out_dir:
        cronet_utils.gn(gn_out_dir,
                        ' '.join(cronet_utils.get_gn_args_for_aosp(arch)))
        return cronet_utils.get_transitive_deps_build_files(
            repo_path, gn_out_dir, targets)


def _get_all_build_files_path(repo_path: str, targets: Set[str]) -> Set[str]:
    all_build_files_path = set()
    with multiprocessing.dummy.Pool(len(cronet_utils.ARCHS)) as pool:
        results = [
            pool.apply_async(_get_build_files_paths,
                             (repo_path, targets, arch))
            for arch in cronet_utils.ARCHS
        ]
        for result in results:
            all_build_files_path.update(result.get())
    return all_build_files_path


def should_skip_readme_file(readme_path: str) -> bool:
    return readme_path in constants.IGNORED_README


def update_license(repo_path: str = _ROOT_CRONET,
                   post_process_dict: Dict[str, Callable] = None,
                   verify_only: bool = False,
                   reachable_through_dependencies: bool = True):
    """
  Updates the licensing files for the entire repository of external/cronet.

  Running this will generate the following files for each README.chromium

  * LICENSE, this is a symbolic link and only created if there is no LICENSE
  file already.
  * METADATA
  * MODULE_LICENSE_XYZ, XYZ represents the license found in README.chromium.

  Running in verify-only mode will ensure that everything is up to date, an
  exception will be thrown if there needs to be any changes.
  :param repo_path: Absolute path to Cronet's AOSP repository
  :param post_process_dict: A dictionary that includes post-processing, this
  post processing is not done on the README.chromium file but on the Metadata
  structure that is extracted from them.
  :param verify_only: Ensures that everything is up to date or throws.
  :param reachable_through_dependencies: Ensures that the license generator
  will only process README.chromium files that are reachable through cronet
  as a transitive dependency.
  """
    if post_process_dict is None:
        post_process_dict = constants.POST_PROCESS_OPERATION
    readme_files_to_process = None
    if reachable_through_dependencies:
        readme_files_to_process = get_all_readme_through_gn(
            repo_path, _TARGETS_TO_PROCESS_LICENSE_FOR_TRANSITIVELY)
    else:
        readme_files_to_process = get_all_readme(repo_path)

    if readme_files_to_process == 0:
        raise Exception(
            f"Failed to find any README.chromium files under {repo_path}")

    readme_to_license_content = {}
    for readme_file in readme_files_to_process:
        if should_skip_readme_file(readme_file):
            continue
        readme_directory = os.path.dirname(
            os.path.abspath(os.path.join(repo_path, readme_file)))

        metadata = license_utils.parse_chromium_readme_file(
            os.path.abspath(os.path.join(repo_path, readme_file)),
            post_process_dict.get(readme_file, lambda _metadata: _metadata))

        license_file_path = metadata.get_license_file_path()
        if license_file_path is None:
            # If there is no license file, create one with just the license name. This
            # is necessary for e.g.
            #   //third_party/boringssl/src/pki/testdata/nist-pkits/README.chromium
            license_file_path = "LICENSE"
            with open(os.path.join(readme_directory, license_file_path),
                      "w") as license_file:
                license_file.write("\n".join(metadata.get_licenses()))

        license_directory = readme_directory
        if (os.path.relpath(readme_directory,
                            repo_path).startswith("third_party/rust/")):
            # We must do a mapping as Chromium stores the README.chromium
            # in a different directory than where the src/LICENSE is stored.
            license_directory = os.path.join(
                repo_path,
                _map_rust_license_path_to_directory(license_file_path))

        resolved_license_file_path = os.path.join(
            repo_path,
            license_utils.resolve_license_path(readme_directory,
                                               license_file_path))
        if not os.path.exists(resolved_license_file_path):
            raise Exception(
                f"License file {resolved_license_file_path} does not exist for README.chromium: {readme_file}"
            )
        _maybe_create_license_file_symlink(license_directory,
                                           resolved_license_file_path,
                                           verify_only)
        _create_module_license_file(repo_path, license_directory,
                                    metadata.get_licenses(), verify_only)
        _create_metadata_file(repo_path, license_directory,
                              metadata.to_android_metadata(), verify_only)
        readme_to_license_content[readme_file] = cronet_utils.read_file(
            resolved_license_file_path)

    readme_for_top_level = None
    if reachable_through_dependencies:
        readme_for_top_level = get_all_readme_through_gn(
            repo_path, _TARGETS_TO_AGGREGATE_LICENSE_FOR_TRANSITIVELY)
    else:
        readme_for_top_level = get_all_readme(repo_path)

    top_level_license_file = Path(os.path.join(repo_path, "LICENSE"))
    license_aggregated = []
    for readme in sorted(readme_for_top_level):
        if should_skip_readme_file(readme):
            continue

        license_aggregated += [
            "============================\n", f"Path = {readme}\n",
            "============================\n", readme_to_license_content[readme]
        ]
    top_level_license_file.write_text("\n".join(license_aggregated))


if __name__ == '__main__':
    sys.exit(
        update_license(post_process_dict=constants.POST_PROCESS_OPERATION,
                       reachable_through_dependencies=True))