from collections.abc import Callable
import glob
import hashlib
import itertools
import json
import logging
import os
from pathlib import Path
import re
import tempfile
from typing import TypeVar
from urllib.parse import urlparse, urlsplit, urlunsplit

from esphome import git, yaml_util
from esphome.core import CORE, Library
from esphome.espidf.framework import archive_extract_all, download_from_mirrors, rmdir
from esphome.helpers import write_file_if_changed

_LOGGER = logging.getLogger(__name__)

PathType = str | os.PathLike

#
# Constants from platformio
#

FILTER_REGEX = re.compile(r"([+-])<([^>]+)>")
DEFAULT_BUILD_SRC_FILTER = (
    "+<*> -<.git/> -<.svn/> -<example/> -<examples/> -<test/> -<tests/>"
)
DEFAULT_BUILD_SRC_DIRS = "src"
DEFAULT_BUILD_INCLUDE_DIR = "include"
DEFAULT_BUILD_FLAGS = []
SRC_FILE_EXTENSIONS = [
    ".c",
    ".cpp",
    ".cc",
    ".cxx",
    ".c++",
    ".S",
    ".spp",
    ".SPP",
    ".sx",
    ".s",
    ".asm",
    ".ASM",
]

ESP32_PLATFORM = "espressif32"
DOMAIN = "pio_components"

ESPHOME_DATA_KEY = "ESPHOME"
ESPHOME_DATA_EXTRA_CMAKE_KEY = "EXTRA_CMAKE"


class Source:
    def download(self, dir_suffix: str, force: bool = False) -> Path:
        raise NotImplementedError()


class URLSource(Source):
    def __init__(self, url: str):
        self.url = url

    def download(self, dir_suffix: str, force: bool = False) -> Path:
        base_dir = Path(CORE.data_dir) / DOMAIN
        h = hashlib.new("sha256")
        h.update(self.url.encode())
        path = base_dir / h.hexdigest()[:8] / dir_suffix
        # Marker file written last to signal a complete extraction. Using a
        # marker (instead of just `path.is_dir()`) means an interrupted
        # extraction is correctly detected and re-run on the next invocation,
        # and lets us extract directly into ``path`` — avoiding a
        # post-extraction rename that races with antivirus on Windows.
        extracted_marker = path / ".esphome_extracted"
        if not extracted_marker.is_file() or force:
            rmdir(path, msg=f"Clean up library directory {path}")

            # Download in temporary file
            with tempfile.NamedTemporaryFile() as tmp:
                _LOGGER.info("Downloading %s ...", self.url)
                _LOGGER.debug("Location: %s", path)

                download_from_mirrors([self.url], {}, tmp.file)

                _LOGGER.debug("Extracting archive to %s ...", path)
                archive_extract_all(tmp.file, path)
                extracted_marker.touch()
        return path

    def __str__(self):
        return self.url


class GitSource(Source):
    def __init__(self, url: str, ref: str):
        self.url = url
        self.ref = ref

    def download(self, dir_suffix: str, force: bool = False) -> Path:
        path, _ = git.clone_or_update(
            url=self.url,
            ref=self.ref,
            refresh=git.NEVER_REFRESH if not force else None,
            domain=DOMAIN,
            submodules=[],
            subpath=Path(dir_suffix),
        )
        return path

    def __str__(self):
        return f"{self.url}#{self.ref}"


class InvalidIDFComponent(Exception):
    pass


class IDFComponent:
    def __init__(self, name: str, version: str, source: Source | None):
        self.name = name
        self.version = version
        self.source = source
        self.data = {}
        self.dependencies: list[IDFComponent] = []
        self._path: Path | None = None

    def __str__(self):
        return f"{self.name}@{self.version}={self.source}"

    @property
    def path(self) -> Path:
        if self._path is None:
            raise RuntimeError(f"path not set for component {self}")
        return self._path

    @path.setter
    def path(self, value: Path) -> None:
        self._path = value

    def get_sanitized_name(self):
        return re.sub(r"[^a-zA-Z0-9_.\-/]", "_", self.name)

    def get_require_name(self):
        return self.get_sanitized_name().replace("/", "__")

    def download(self, force: bool = False):
        """
        The dependency name should match the directory name at the end of the override path.
        The ESP-IDF build system uses the directory name as the component name, so the directory of the override_path should match the component name.
        If you want to specify the full name of the component with the namespace, replace / in the component name with __.
        @see https://docs.espressif.com/projects/idf-component-manager/en/latest/reference/manifest_file.html
        """
        self.path = self.source.download(self.get_sanitized_name(), force=force)


def _get_package_from_pio_registry(
    username: str | None, pkgname: str, requirements: str
) -> tuple[str, str, str | None, str | None]:
    """
    Fetch package information from PlatformIO registry.

    This function queries the PlatformIO registry to find a library package
    that matches the given criteria and returns its metadata including version
    and download URL.

    Args:
        username: The owner/username of the package (can be None)
        pkgname: The name of the package
        requirements: Version requirements (e.g., "^1.0.0")

    Returns:
        tuple[str, str, str | None, str | None]:
        A tuple containing (owner, name, version, download_url)
        where version and download_url can be None if not found
    """

    from platformio.package.manager._registry import PackageManagerRegistryMixin
    from platformio.package.meta import PackageSpec

    # Create a minimal PackageManagerRegistry class
    class PackageManagerRegistry(PackageManagerRegistryMixin):
        def __init__(self):
            self._registry_client = None
            self.pkg_type = "library"

        @staticmethod
        def is_system_compatible(value, custom_system=None):
            return True

    pio_registry = PackageManagerRegistry()

    # Fetch package metadata from registry
    package = pio_registry.fetch_registry_package(
        PackageSpec(
            owner=username,
            name=pkgname,
        )
    )
    owner = package["owner"]["username"]
    name = package["name"]

    # Find the best matching version based on requirements
    version = pio_registry.pick_best_registry_version(
        package.get("versions"),
        PackageSpec(owner=username, name=pkgname, requirements=requirements),
    )

    #  If no version found, return with None for version and URL
    if not version:
        return owner, name, None, None

    # Find the compatible package file for this version
    pkgfile = pio_registry.pick_compatible_pkg_file(version["files"])

    #  If no package file found, return with None for URL but valid version
    if not pkgfile:
        return owner, name, version["name"], None

    return owner, name, version["name"], pkgfile["download_url"]


def _apply_extra_script(component: IDFComponent) -> None:
    """Run a PIO ``extraScript`` and fold its captured env vars into
    ``component.data["build"]["flags"]`` so the existing -L/-l/-D
    extraction in ``generate_cmakelists_txt`` picks them up."""
    extra_script = component.data.get("build", {}).get("extraScript")
    if not extra_script:
        return
    # Resolve and confine to the component dir so a malicious library.json
    # can't escape (e.g. ``"extraScript": "../../etc/passwd"``).
    library_root = component.path.resolve()
    script_path = (component.path / extra_script).resolve()
    if not script_path.is_relative_to(library_root) or not script_path.is_file():
        return
    from esphome.components.esp32 import get_esp32_variant
    from esphome.espidf.extra_script import captured_as_build_flags, run_extra_script

    idf_target = get_esp32_variant().lower().replace("-", "")
    result = run_extra_script(
        script_path, library_dir=component.path, idf_target=idf_target
    )
    extra_flags = captured_as_build_flags(result, library_dir=component.path)
    if not extra_flags:
        return
    flags = component.data.setdefault("build", {}).setdefault("flags", [])
    if isinstance(flags, str):
        flags = [flags]
    flags.extend(extra_flags)
    component.data["build"]["flags"] = flags


T = TypeVar("T")


def _ensure_list(obj: T | list[T]) -> list[T]:
    """
    Convert an object to a list if it isn't already a list.

    Args:
        obj: Object that may or may not already be a list.

    Returns:
        list[T]: The original list if ``obj`` is a list, otherwise a single-item
        list containing ``obj``.
    """
    return [obj] if not isinstance(obj, list) else obj


def _owner_pkgname_to_name(owner: str | None, pkgname: str) -> str:
    """
    Convert owner and package name to a standardized component name.

    This function combines owner and package name with a forward slash when
    both are provided, otherwise returns just the package name.

    Args:
        owner: The owner/username of the package (can be None)
        pkgname: The name of the package

    Returns:
        str: The standardized component name in "owner/pkgname" format or just "pkgname"
    """
    return f"{owner}/{pkgname}" if owner else pkgname


def _collect_filtered_files(src_dir: PathType, src_filters: list[str]) -> list[str]:
    """
    Recursively match files in a directory according to include/exclude patterns.

    This function processes a list of filter strings that indicate which files
    to include or exclude. Each filter is parsed into patterns with a sign:
    '+' for inclusion and '-' for exclusion. Directory patterns ending with '/'
    are normalized to include all their contents recursively.

    Args:
        src_dir (PathType): Root directory to search within.
        src_filters (list[str]): List of filter strings, which may contain multiple
            patterns. Each pattern can start with '+' or '-' to indicate inclusion
            or exclusion.

    Returns:
        list[str]: List of matched file paths as strings. Only files (not directories)
        are returned, even if a directory matches a pattern.
    """
    matches = list(
        itertools.chain.from_iterable(
            FILTER_REGEX.findall(src_filter) for src_filter in src_filters
        )
    )

    selected = set()

    for sign, pattern in matches:
        pattern = pattern.strip()

        if pattern.endswith("/"):
            pattern = pattern.rstrip("/") + "/**"

        full_pattern = os.path.join(glob.escape(str(src_dir)), pattern)

        matched = []
        for item in glob.glob(full_pattern, recursive=True):
            if not os.path.isdir(item):
                matched.append(item)
            else:
                # PlatformIO quirk: a directory matched with "*" should include all its
                # nested files and subdirectories, not just the directory itself.
                for root, _, files in os.walk(item):
                    matched.extend([os.path.join(root, f) for f in files])

        if sign == "+":
            selected.update(matched)
        elif sign == "-":
            selected.difference_update(matched)

    return [r for r in selected if os.path.isfile(r)]


def _convert_library_to_component(library: Library) -> IDFComponent:
    """
    Convert a Library object to an IDFComponent object by resolving its metadata.

    This function handles the conversion of library specifications to component
    objects, resolving versions through PlatformIO registry when needed or
    parsing direct repository URLs.

    Args:
        library: The Library object containing name, version, and/or repository information

    Returns:
        IDFComponent: The resolved component with name, version, and URL

    Raises:
        ValueError: If a repository URL is missing a reference (#)
        RuntimeError: If no artifact can be found for the library
    """
    name = None
    version = None
    source = None

    #  Repository is provided directly
    if library.repository:
        # Parse repository URL: path becomes the component name, fragment
        # becomes the git ref stored on GitSource.
        split_result = urlsplit(library.repository)
        if not split_result.fragment.strip():
            raise ValueError(f"Missing ref in URL {library.repository}")

        # Sanitize name
        name = str(split_result.path).strip("/")
        name = name.removesuffix(".git")

        # IDF Component Manager only accepts "*", a 40-char commit hash, or
        # semver here. The actual git ref is preserved in GitSource.ref;
        # override_path makes this field cosmetic at build time.
        version = "*"
        repository = urlunsplit(split_result._replace(fragment=""))

        source = GitSource(str(repository), split_result.fragment)

    # Version is provided - resolve using PlatformIO registry
    elif library.version:
        name = library.name
        if "/" not in name:
            owner, pkgname = None, name
        else:
            owner, pkgname = name.split("/", 1)

        owner, pkgname, version, url = _get_package_from_pio_registry(
            owner, pkgname, library.version
        )
        if url is None:
            raise RuntimeError(
                f"Can't find an pkg file from PlatformIO registry for library {library}"
            )

        name = _owner_pkgname_to_name(owner, pkgname)
        source = URLSource(url)

    if source is None:
        raise RuntimeError(f"Can't find an artifact associated to library {library}")

    assert name, "Missing library name"
    assert version, "Missing library version"

    return IDFComponent(name, version, source)


def _split_list_by_condition(
    items: list[str], match_fn: Callable[[str], str | None]
) -> tuple[list[str], list[str]]:
    """
    Splits a list into two lists based on a matching function.

    Args:
        items: List of items to split.
        match_fn: Function that returns a value for items that should go into the "matched" list.

    Returns:
        A tuple (matched, non_matched)
    """
    matched = []
    non_matched = []
    for item in items:
        result = match_fn(item)
        if result:
            matched.append(result)
        else:
            non_matched.append(item)
    return matched, non_matched


def generate_cmakelists_txt(component: IDFComponent) -> str:
    """
    Generate a CMakeLists.txt file for an ESP-IDF component.

    This function creates the necessary CMake configuration to build a library
    with ESP-IDF, including source files, include directories, dependencies,
    and build flags.

    Args:
        component: The IDFComponent object containing library metadata and path

    Returns:
        str: The complete CMakeLists.txt content as a string
    """

    def escape_entry(p: PathType) -> str:
        # In CMakeLists.txt, backslashes need to be escaped
        return f'"{str(p)}"'.replace("\\", "\\\\")

    # Extract the values
    build_src_dir = component.data.get("build", {}).get("srcDir", None)
    if not build_src_dir:
        for d in ["src", "Src", "."]:
            if (component.path / Path(d)).is_dir():
                build_src_dir = d
                break

    build_include_dir = component.data.get("build", {}).get(
        "includeDir", DEFAULT_BUILD_INCLUDE_DIR
    )
    build_src_filter = _ensure_list(
        component.data.get("build", {}).get("srcFilter", DEFAULT_BUILD_SRC_FILTER)
    )
    build_flags = _ensure_list(
        component.data.get("build", {}).get("flags", DEFAULT_BUILD_FLAGS)
    )

    # List all sources files
    build_src_files = _collect_filtered_files(
        component.path / Path(build_src_dir), build_src_filter
    )

    # Only bake library.json-declared deps here. Project-managed and
    # built-in components come in via ${ESPHOME_PROJECT_MANAGED_COMPONENTS}
    # / ${ESPHOME_PROJECT_BUILTIN_COMPONENTS} set in the top-level
    # CMakeLists, so this file stays project-agnostic when shared from
    # the pio_components cache.
    requires: set[str] = {
        dependency.get_require_name() for dependency in component.dependencies
    }

    # Only keep sources
    build_src_files = [os.path.relpath(p, component.path) for p in build_src_files]
    build_src_files = [
        f for f in build_src_files if os.path.splitext(f)[1] in SRC_FILE_EXTENSIONS
    ]

    # Handle build flags
    include_dir_flags, build_flags = _split_list_by_condition(
        build_flags, lambda a: a[2:].strip() if a.startswith("-I") else None
    )
    link_directories, build_flags = _split_list_by_condition(
        build_flags, lambda a: a[2:].strip() if a.startswith("-L") else None
    )
    link_libraries, build_flags = _split_list_by_condition(
        build_flags, lambda a: a[2:].strip() if a.startswith("-l") else None
    )

    # Split include directories from build_flags
    # Only keep an include directory if it exists
    build_include_dirs = [build_include_dir, build_src_dir] + include_dir_flags
    build_include_dirs = [
        d for d in build_include_dirs if (component.path / Path(d)).is_dir()
    ]

    # Split build_flags list into private and public lists
    private_build_flags, public_build_flags = _split_list_by_condition(
        build_flags, lambda a: a if a.startswith("-W") else None
    )

    # Generate the component
    content = "idf_component_register(\n"
    if build_src_files:
        str_srcs = " ".join([escape_entry(p) for p in sorted(build_src_files)])
        content += f"  SRCS {str_srcs}\n"
    if build_include_dirs:
        str_include_dirs = " ".join([escape_entry(p) for p in build_include_dirs])
        content += f"  INCLUDE_DIRS {str_include_dirs}\n"
    # Project-managed and built-in component lists are set per-project
    # via idf_build_set_property in the top-level CMakeLists; expanded
    # here at configure time. Keeping them out of the per-lib REQUIRES
    # means this CMakeLists is project-agnostic and reusable from the
    # pio_components cache across builds.
    str_requires = " ".join(
        [
            *sorted(requires),
            "${ESPHOME_PROJECT_MANAGED_COMPONENTS}",
            "${ESPHOME_PROJECT_BUILTIN_COMPONENTS}",
        ]
    )
    content += f"  REQUIRES {str_requires}\n"
    content += ")\n"

    # Add public and private build flags
    if public_build_flags:
        content += "target_compile_options(${COMPONENT_LIB} PUBLIC\n"
        for build_flag in public_build_flags:
            str_build_flag = escape_entry(build_flag)
            content += f"  {str_build_flag}\n"
        content += ")\n"
    if private_build_flags:
        content += "target_compile_options(${COMPONENT_LIB} PRIVATE\n"
        for build_flag in private_build_flags:
            str_build_flag = escape_entry(build_flag)
            content += f"  {str_build_flag}\n"
        content += ")\n"

    # Add library paths and files
    if link_directories:
        content += "target_link_directories(${COMPONENT_LIB} INTERFACE\n"
        for link_directory in link_directories:
            str_build_flag = escape_entry(link_directory)
            content += f"  {str_build_flag}\n"
        content += ")\n"

    if link_libraries:
        content += "target_link_libraries(${COMPONENT_LIB} INTERFACE\n"
        for link_library in link_libraries:
            str_build_flag = escape_entry(link_library)
            content += f"  {str_build_flag}\n"
        content += ")\n"

    # Add custom CMake scripts
    content += "\n".join(
        component.data.get(ESPHOME_DATA_KEY, {}).get(ESPHOME_DATA_EXTRA_CMAKE_KEY, [])
    )

    return content


def generate_idf_component_yml(component: IDFComponent) -> str:
    """
    Generate ESP-IDF component YAML configuration for a library.

    Args:
        component: IDFComponent object to generate YAML for

    Returns:
        YAML string representation of ESP-IDF component configuration
    """
    data = {}

    description = component.data.get("description")
    if description:
        data["description"] = description

    repository = component.data.get("repository", {}).get("url", None)
    if repository:
        data["repository"] = repository

    for dependency in component.dependencies:
        # Initialize dependencies section if needed
        if "dependencies" not in data:
            data["dependencies"] = {}

        # Every dependency goes through _generate_idf_component →
        # component.download() before this runs, so .path is always set.
        data["dependencies"][dependency.get_sanitized_name()] = {
            "override_path": str(dependency.path),
        }

    return yaml_util.dump(data)


def _check_library_data(data: dict):
    """
    Check if a library data is compatible with the ESP-IDF framework.

    Args:
        component: IDFComponent object being processed

    Raises:
        ValueError: If library has unsupported platforms or frameworks
    """
    platforms = data.get("platforms", "*")
    if isinstance(platforms, str):
        platforms = [a.strip() for a in platforms.split(",")]
    platforms = _ensure_list(platforms)

    # Check if library supports ESP-IDF platform
    valid_platforms = "*" in platforms or ESP32_PLATFORM in platforms

    if not valid_platforms:
        raise InvalidIDFComponent(f"Unsupported library platforms: {platforms}")

    frameworks = data.get("frameworks", "*")
    if isinstance(frameworks, str):
        frameworks = [a.strip() for a in frameworks.split(",")]
    frameworks = _ensure_list(frameworks)

    # Check if library supports ESP-IDF framework
    framework = "arduino" if CORE.using_arduino else "espidf"
    valid_framework = "*" in frameworks or framework in frameworks

    if not valid_framework:
        raise InvalidIDFComponent(f"Unsupported library frameworks: {frameworks}")


def _process_dependencies(component: IDFComponent):
    """
    Process library dependencies and generate ESP-IDF components.

    Args:
        component: IDFComponent object being processed

    Returns:
        None
    """

    name, version = component.name, component.version
    dependencies = component.data.get("dependencies")
    if not dependencies:
        return

    _LOGGER.info("Processing %s@%s component dependencies...", name, version)
    for dependency in dependencies:
        # Validate dependency structure
        if not all(k in dependency for k in ("name", "version")):
            _LOGGER.debug("Ignore invalid library: %s", dependency)
            continue

        try:
            _check_library_data(dependency)
        except InvalidIDFComponent as e:
            _LOGGER.debug(
                "Skip %s@%s: %s", dependency["name"], dependency["version"], str(e)
            )
            continue

        # The version field may actually contain a URL
        version = dependency["version"]
        url = None
        try:
            result = urlparse(version)
            if all([result.scheme, result.netloc]):
                url, version = version, None
        except (TypeError, ValueError):
            pass

        # Generate ESP-IDF component from PlatformIO library
        component.dependencies.append(
            _generate_idf_component(
                Library(
                    _owner_pkgname_to_name(
                        dependency.get("owner", None), dependency.get("name")
                    ),
                    version,
                    url,
                )
            )
        )


def _parse_library_json(library_json_path: PathType):
    """
    Load and parse a JSON file describing a library.

    Args:
        library_json_path (PathType): Path to the JSON file.

    Returns:
        dict: Parsed JSON content as a Python dictionary.
    """
    with open(library_json_path, encoding="utf8") as fp:
        return json.load(fp)


def _parse_library_properties(library_properties_path: PathType):
    """
    Parse a key-value platformio .properties style file into a dictionary.

    Args:
        library_properties_path (PathType): Path to the properties file.

    Returns:
        dict[str, str]: Mapping of parsed property keys to values.
    """
    with open(library_properties_path, encoding="utf8") as fp:
        data = {}
        for line in fp.read().splitlines():
            line = line.strip()
            if not line or "=" not in line:
                continue
            # skip comments
            if line.startswith("#"):
                continue
            key, value = line.split("=", 1)
            if not value.strip():
                continue
            data[key.strip()] = value.strip()
        return data


def _generate_idf_component(library: Library, force: bool = False) -> IDFComponent:
    """
    Generate an ESP-IDF component from a library specification.

    This function resolves the library, downloads it, processes metadata files,
    and generates necessary ESP-IDF build files (CMakeLists.txt, idf_component.yml).

    Args:
        library: The library specification containing name, version, and repository URL
        force: If True, forces re-download of the library even if it exists locally

    Returns:
        IDFComponent: The generated component object with resolved metadata
    """
    _LOGGER.info("Generate IDF component for %s library ...", library)

    # Resolve component name, version and url
    component = _convert_library_to_component(library)
    name, version = component.name, component.version

    # Download the library
    component.download(force)

    # Paths to component metadata and build files
    library_json_path = component.path / "library.json"
    library_properties_path = component.path / "library.properties"
    cmakelists_txt_path = component.path / "CMakeLists.txt"
    idf_component_yml_path = component.path / "idf_component.yml"

    # Bundled CMakeLists.txt / idf_component.yml are ignored -- library
    # authors' IDF support is frequently broken (bogus REQUIRES, hard-coded
    # arduino-esp32, etc.). We always regenerate.

    if library_json_path.is_file():
        component.data = _parse_library_json(library_json_path)
    elif library_properties_path.is_file():
        component.data = _parse_library_properties(library_properties_path)
    else:
        raise RuntimeError(
            "Invalid PIO library: missing library.json and/or library.properties"
        )

    # Check if the component is usable with ESP-IDF before executing any
    # third-party Python from the library (``_apply_extra_script`` below).
    _check_library_data(component.data)

    # If the library declares a PIO ``extraScript``, run it against a
    # fake SCons env so we can fold its captured LIBPATH/LIBS/etc into
    # the build-flag pipeline ``generate_cmakelists_txt`` consumes
    # below. Without this, libraries that wire per-MCU archive linking
    # via extraScript fail to link under native ESP-IDF.
    _apply_extra_script(component)

    # Handle the dependencies (convert PlatformIO library to ESP-IDF component if needed)
    _process_dependencies(component)

    _LOGGER.debug("Generating CMakeLists.txt for %s@%s  ...", name, version)
    write_file_if_changed(
        cmakelists_txt_path,
        generate_cmakelists_txt(component),
    )

    _LOGGER.debug("Generating idf_component.yml for %s@%s  ...", name, version)
    write_file_if_changed(
        idf_component_yml_path,
        generate_idf_component_yml(component),
    )

    return component


def generate_idf_component(
    library: Library, force: bool = False
) -> tuple[str, str, Path]:
    """
    Generate an ESP-IDF component and return its name, version, and path.

    This is a wrapper function that calls _generate_idf_component and returns
    the standardized tuple format (name, version, path).

    Args:
        library: The library specification containing name, version, and repository URL
        force: If True, forces re-download of the library even if it exists locally

    Returns:
        tuple[str, str, Path]: A tuple containing (component_name, component_version, component_path)
    """
    component = _generate_idf_component(library, force)
    return component.get_sanitized_name(), component.version, component.path
