from collections.abc import Callable, Iterable
from contextlib import AbstractContextManager
from dataclasses import dataclass
import importlib
import importlib.abc
import importlib.resources
import importlib.util
import logging
from pathlib import Path
import sys
from types import ModuleType
from typing import TYPE_CHECKING, Any

from esphome.const import SOURCE_FILE_EXTENSIONS
from esphome.types import ConfigType

if TYPE_CHECKING:
    from esphome.cpp_generator import MockObjClass
    from esphome.external_files import RemoteFile

# `esphome.core.config` is imported lazily in `_lookup_module` when the
# "esphome" pseudo-component is first resolved. It pulls in
# `esphome.automation` and `esphome.config_validation`, which together
# dominate `esphome.__main__` startup cost when loaded eagerly.
# `esphome.cpp_generator` is similarly avoided at module scope; it pulls
# in `esphome.yaml_util` and is only needed for the `MockObjClass` type
# annotation, which is resolved lazily via `TYPE_CHECKING`.

_LOGGER = logging.getLogger(__name__)


@dataclass(frozen=True, order=True)
class FileResource:
    package: str
    resource: str

    def path(self) -> AbstractContextManager[Path]:
        return importlib.resources.as_file(
            importlib.resources.files(self.package) / self.resource
        )


class ComponentManifest:
    def __init__(self, module: ModuleType, recursive_sources: bool = False):
        self.module = module
        self.recursive_sources = recursive_sources

    @property
    def package(self) -> str:
        """Return the package name the module is contained in.

        Examples:
        - esphome/components/gpio/__init__.py -> esphome.components.gpio
        - esphome/components/gpio/switch/__init__.py -> esphome.components.gpio.switch
        - esphome/components/a4988/stepper.py -> esphome.components.a4988
        """
        return self.module.__package__

    @property
    def is_platform(self) -> bool:
        return len(self.module.__name__.split(".")) == 4

    @property
    def is_platform_component(self) -> bool:
        return getattr(self.module, "IS_PLATFORM_COMPONENT", False)

    @property
    def is_target_platform(self) -> bool:
        return getattr(self.module, "IS_TARGET_PLATFORM", False)

    @property
    def config_schema(self) -> Any | None:
        return getattr(self.module, "CONFIG_SCHEMA", None)

    @property
    def multi_conf(self) -> bool:
        return getattr(self.module, "MULTI_CONF", False)

    @property
    def multi_conf_no_default(self) -> bool:
        return getattr(self.module, "MULTI_CONF_NO_DEFAULT", False)

    @property
    def to_code(self) -> Callable[[Any], None] | None:
        return getattr(self.module, "to_code", None)

    @property
    def dependencies(self) -> list[str]:
        return getattr(self.module, "DEPENDENCIES", [])

    @property
    def conflicts_with(self) -> list[str]:
        return getattr(self.module, "CONFLICTS_WITH", [])

    @property
    def auto_load(
        self,
    ) -> list[str] | Callable[[], list[str]] | Callable[[ConfigType], list[str]]:
        return getattr(self.module, "AUTO_LOAD", [])

    @property
    def codeowners(self) -> list[str]:
        return getattr(self.module, "CODEOWNERS", [])

    @property
    def aliases(self) -> list[str]:
        """Legacy names that should transparently route to this component.

        See the :func:`_build_alias_map` documentation for how aliases are
        discovered (AST scan, no execution) and registered both for the YAML
        loader (top-level key rename in :mod:`esphome.config`) and for
        Python imports (``sys.meta_path`` finder, below).
        """
        return getattr(self.module, "ALIASES", [])

    @property
    def alias_removal_version(self) -> str | None:
        """Optional ESPHome version when the alias warning becomes a hard error.

        Surfaced in the deprecation warning emitted by the YAML pre-pass so
        users know how long they have to migrate. ``None`` means the warning
        does not mention a specific version.
        """
        return getattr(self.module, "ALIAS_REMOVAL_VERSION", None)

    @property
    def instance_type(self) -> "MockObjClass | None":
        return getattr(self.module, "INSTANCE_TYPE", None)

    @property
    def final_validate_schema(self) -> Callable[[ConfigType], None] | None:
        """Components can declare a `FINAL_VALIDATE_SCHEMA` cv.Schema that gets called
        after the main validation. In that function checks across components can be made.

        Note that the function can't mutate the configuration - no changes are saved
        """
        return getattr(self.module, "FINAL_VALIDATE_SCHEMA", None)

    @property
    def prefetch_files(
        self,
    ) -> Callable[[list[ConfigType]], Iterable[list["RemoteFile"]]] | None:
        """Optional `PREFETCH_FILES` hook for batched remote file downloads.

        A generator called once per run with the component's raw, pre-schema
        config entries; each yield is a stage of ``RemoteFile`` downloaded in
        one parallel pass before schema validation, so a later stage may
        derive URLs from earlier files' content. Best effort: skip anything
        unrecognized. On platform components, place it on the platform
        sub-module; a domain-module hook receives every entry.
        """
        return getattr(self.module, "PREFETCH_FILES", None)

    @property
    def legacy_config_migrate(self) -> Callable[[ConfigType], ConfigType | None] | None:
        """Optional `LEGACY_CONFIG_MIGRATE` callable on a platform component module.

        Called once, before platform entries are processed, with the raw top-level
        config for this domain. It may transform a pre-platform-format config (e.g.
        a bare list or legacy dict form) into the normalized list of `platform:`
        tagged entries and return it. Returning ``None`` means "already in the new
        format, leave untouched". This is an intentionally removable deprecation
        shim hook.
        """
        return getattr(self.module, "LEGACY_CONFIG_MIGRATE", None)

    @property
    def expand_platform_config(
        self,
    ) -> Callable[[list[ConfigType]], list[ConfigType]] | None:
        """Optional `EXPAND_PLATFORM_CONFIG` callable; runs on the normalized `platform:`-tagged
        entry list before per-entry CONFIG_SCHEMA. Must return a list (raise `cv.Invalid` for user errors)."""
        return getattr(self.module, "EXPAND_PLATFORM_CONFIG", None)

    @property
    def resources(self) -> list[FileResource]:
        """Return a list of all file resources defined in the package of this component.

        By default only files directly in the package directory are returned. Manifests
        constructed with ``recursive_sources=True`` also descend into non-subpackage
        subdirectories (subdirectories without an ``__init__.py``), so core code can
        live under ``esphome/core/<group>/`` without every component paying the cost.
        """
        ret: list[FileResource] = []

        # Get filter function for source files
        filter_source_files_func = getattr(self.module, "FILTER_SOURCE_FILES", None)

        # Get list of files to exclude
        excluded_files = (
            set(filter_source_files_func()) if filter_source_files_func else set()
        )

        root = importlib.resources.files(self.package)

        for child in root.iterdir():
            name = child.name
            if child.is_file():
                if Path(name).suffix not in SOURCE_FILE_EXTENSIONS:
                    continue
                if name in excluded_files:
                    continue
                ret.append(FileResource(self.package, name))
            elif self.recursive_sources and child.is_dir() and name != "__pycache__":
                # Skip Python subpackages — they load as their own components.
                if child.joinpath("__init__.py").is_file():
                    continue
                for sub in child.iterdir():
                    if not sub.is_file():
                        continue
                    if Path(sub.name).suffix not in SOURCE_FILE_EXTENSIONS:
                        continue
                    resource = f"{name}/{sub.name}"
                    if resource in excluded_files:
                        continue
                    ret.append(FileResource(self.package, resource))

        return ret


class ComponentMetaFinder(importlib.abc.MetaPathFinder):
    def __init__(
        self, components_path: Path, allowed_components: list[str] | None = None
    ) -> None:
        self._allowed_components = allowed_components
        self._finders = []
        for hook in sys.path_hooks:
            try:
                finder = hook(str(components_path))
            except ImportError:
                continue
            self._finders.append(finder)

    def find_spec(self, fullname: str, path: list[str] | None, target=None):
        if not fullname.startswith("esphome.components."):
            return None
        parts = fullname.split(".")
        if len(parts) != 3:
            # only handle direct components, not platforms
            # platforms are handled automatically when parent is imported
            return None
        component = parts[2]
        if (
            self._allowed_components is not None
            and component not in self._allowed_components
        ):
            return None

        for finder in self._finders:
            spec = finder.find_spec(fullname, target=target)
            if spec is not None:
                return spec
        return None


def clear_component_meta_finders():
    sys.meta_path = [x for x in sys.meta_path if not isinstance(x, ComponentMetaFinder)]


def install_meta_finder(
    components_path: Path, allowed_components: list[str] | None = None
):
    sys.meta_path.insert(0, ComponentMetaFinder(components_path, allowed_components))


def _lookup_module(domain: str, exception: bool) -> ComponentManifest | None:
    if domain in _COMPONENT_CACHE:
        return _COMPONENT_CACHE[domain]

    if domain == "esphome":
        import esphome.core.config

        manif = ComponentManifest(esphome.core.config, recursive_sources=True)
        _COMPONENT_CACHE[domain] = manif
        return manif

    # If `domain` is the legacy name of a renamed component, redirect to the
    # canonical module so the rest of the loader (and every caller of
    # `get_component(legacy)`) transparently sees the new component.
    alias_meta = get_alias_metadata().get(domain)
    if alias_meta is not None:
        manif = _lookup_module(alias_meta.canonical, exception)
        if manif is not None:
            _COMPONENT_CACHE[domain] = manif
        return manif

    try:
        module = importlib.import_module(f"esphome.components.{domain}")
    except ImportError as e:
        if exception:
            raise
        if "No module named" in str(e):
            _LOGGER.info(
                "Unable to import component %s: %s", domain, str(e), exc_info=False
            )
        else:
            _LOGGER.exception("Unable to import component %s:", domain)
        return None
    except Exception:  # pylint: disable=broad-except
        if exception:
            raise
        _LOGGER.exception("Unable to load component %s:", domain)
        return None

    manif = ComponentManifest(module)
    _COMPONENT_CACHE[domain] = manif
    return manif


def get_component(domain: str, exception: bool = False) -> ComponentManifest | None:
    assert "." not in domain
    return _lookup_module(domain, exception)


def get_platform(domain: str, platform: str) -> ComponentManifest | None:
    full = f"{platform}.{domain}"
    return _lookup_module(full, False)


_COMPONENT_CACHE: dict[str, ComponentManifest] = {}
CORE_COMPONENTS_PATH = (Path(__file__).parent / "components").resolve()


def _replace_component_manifest(domain: str, manifest: ComponentManifest) -> None:
    """Replace the cached manifest for a component.

    This is an intentionally-supported hook for the C++ test infrastructure
    to install ``ComponentManifestOverride`` wrappers.  Normal application
    code should never call this.
    """
    _COMPONENT_CACHE[domain] = manifest


# ---------------------------------------------------------------------------
# Component aliases (renamed-platform back-compat)
# ---------------------------------------------------------------------------
#
# A component can declare ``ALIASES = ["legacy_name"]`` (and optionally
# ``ALIAS_REMOVAL_VERSION = "YYYY.M.0"``) in its ``__init__.py``, then run
# ``script/build_alias_registry.py`` to regenerate
# ``esphome/component_aliases.py`` (CI and a unit test fail if the registry
# is stale). Two integrations are then wired up automatically:
#
#   1. **Python imports** — a ``sys.meta_path`` finder (``_AliasFinder``)
#      intercepts ``esphome.components.<legacy>``/``...<legacy>.<sub>``
#      imports and resolves them against the canonical component so external
#      custom components that still import from the old path keep working.
#
#   2. **YAML loader** — ``_lookup_module`` consults the alias map so
#      ``get_component("legacy")`` returns the canonical manifest. The
#      ``esphome.config`` pre-pass uses the same map to rewrite legacy
#      top-level keys in the user's config (with a deprecation warning) so
#      dependency checks, schema validation and codegen all see only the
#      canonical name.
#
# Both lookups read the checked-in registry in ``esphome.component_aliases``
# (generated by ``script/build_alias_registry.py``, verified in CI), so no
# component-directory scan happens at runtime. ``_build_alias_map`` below is
# the generator's scan implementation; it **AST-parses** each component's
# ``__init__.py`` rather than importing it.


_ALIAS_META_CACHE: dict[str, "AliasMeta"] | None = None


@dataclass(frozen=True)
class AliasMeta:
    """Metadata for a single deprecated alias entry.

    Used by the YAML pre-pass in :mod:`esphome.config` to produce a
    deprecation warning citing the canonical name and (optionally) the
    removal version declared by the canonical component.
    """

    canonical: str
    removal_version: str | None


def get_alias_metadata() -> dict[str, AliasMeta]:
    """Return the legacy-name → :class:`AliasMeta` map, built lazily from
    the generated registry."""
    global _ALIAS_META_CACHE  # noqa: PLW0603
    if _ALIAS_META_CACHE is None:
        from esphome.component_aliases import COMPONENT_ALIASES

        _ALIAS_META_CACHE = {
            alias: AliasMeta(canonical=canonical, removal_version=removal_version)
            for alias, (canonical, removal_version) in COMPONENT_ALIASES.items()
        }
    return _ALIAS_META_CACHE


def _build_alias_map() -> tuple[dict[str, str], dict[str, AliasMeta]]:
    """Scan every core component dir for ``ALIASES`` declarations.

    Uses :mod:`ast` to read each component's ``__init__.py`` without
    executing it — component import side-effects (logger setup,
    namespace registration, etc.) shouldn't run just because we're
    enumerating aliases.

    Raises if the same alias is claimed by two canonical components, since
    silently picking one would cause non-deterministic routing depending on
    directory-iteration order. Also raises if an alias shadows an existing
    component package: that would hijack a live component domain and, in the
    self-alias case (alias == canonical), send ``_lookup_module`` into
    infinite recursion redirecting a domain to itself.
    """
    import ast

    alias_to_canonical: dict[str, str] = {}
    alias_to_meta: dict[str, AliasMeta] = {}

    if not CORE_COMPONENTS_PATH.is_dir():
        return alias_to_canonical, alias_to_meta

    for child in sorted(CORE_COMPONENTS_PATH.iterdir()):
        if not child.is_dir():
            continue
        init = child / "__init__.py"
        if not init.is_file():
            continue
        aliases, removal_version = _read_aliases(init, ast)
        if not aliases:
            continue
        canonical = child.name
        for alias in aliases:
            if (CORE_COMPONENTS_PATH / alias / "__init__.py").is_file():
                from esphome.core import EsphomeError

                raise EsphomeError(
                    f"Component alias '{alias}' (declared by '{canonical}') "
                    "shadows an existing component package of the same name. "
                    "An alias may only name a component that no longer exists."
                )
            if alias in alias_to_canonical:
                from esphome.core import EsphomeError

                raise EsphomeError(
                    f"Component alias '{alias}' is declared by both "
                    f"'{alias_to_canonical[alias]}' and '{canonical}'. "
                    "Each alias must map to exactly one canonical component."
                )
            alias_to_canonical[alias] = canonical
            alias_to_meta[alias] = AliasMeta(
                canonical=canonical, removal_version=removal_version
            )
    return alias_to_canonical, alias_to_meta


def _read_aliases(
    init_path: Path, ast_module: ModuleType
) -> tuple[list[str], str | None]:
    """Extract ``ALIASES`` and ``ALIAS_REMOVAL_VERSION`` from a component
    ``__init__.py`` via AST parsing.

    Only handles the simple ``NAME = [str_literal, ...]`` / ``NAME = "..."``
    forms — anything more dynamic (function call, conditional, etc.) is
    silently ignored. Components should keep their alias declarations
    static so this scanner can see them.
    """
    try:
        source = init_path.read_text(encoding="utf-8")
    except OSError as err:
        _LOGGER.warning(
            "Could not read %s while scanning for component aliases: %s",
            init_path,
            err,
        )
        return [], None

    # Cheap substring pre-filter: almost no component declares ALIASES, and
    # parsing every component __init__.py with ast is comparatively expensive.
    # Skip the parse entirely unless the token appears in the file at all.
    if "ALIASES" not in source:
        return [], None

    try:
        tree = ast_module.parse(source)
    except SyntaxError as err:
        _LOGGER.warning(
            "Could not parse %s while scanning for component aliases: %s",
            init_path,
            err,
        )
        return [], None

    aliases: list[str] = []
    removal_version: str | None = None

    for node in tree.body:
        if not isinstance(node, ast_module.Assign):
            continue
        for target in node.targets:
            if not isinstance(target, ast_module.Name):
                continue
            if target.id == "ALIASES" and isinstance(node.value, ast_module.List):
                aliases.extend(
                    elt.value
                    for elt in node.value.elts
                    if isinstance(elt, ast_module.Constant)
                    and isinstance(elt.value, str)
                )
            elif (
                target.id == "ALIAS_REMOVAL_VERSION"
                and isinstance(node.value, ast_module.Constant)
                and isinstance(node.value.value, str)
            ):
                removal_version = node.value.value
    return aliases, removal_version


class _AliasFinder(importlib.abc.MetaPathFinder):
    """``sys.meta_path`` finder that resolves legacy-component imports.

    Routes ``esphome.components.<alias>[.<submod>]`` to the canonical
    component's module/submodule of the same name, so external code that
    still imports ``from esphome.components.rp2040 import boards`` keeps
    working without the canonical component having to maintain a shim
    package on disk.

    The finder caches the resolved module in ``sys.modules`` under the
    legacy name on first lookup, so subsequent imports hit the cache and
    skip this finder entirely.
    """

    _PREFIX = "esphome.components."

    def find_spec(self, fullname, path, target=None):  # noqa: ARG002
        if not fullname.startswith(self._PREFIX):
            return None
        # Anything matching the ``esphome.components.`` prefix splits into at
        # least three parts, so ``parts[2]`` (the domain) always exists.
        parts = fullname.split(".")
        domain = parts[2]
        alias_meta = get_alias_metadata().get(domain)
        if alias_meta is None:
            return None

        parts[2] = alias_meta.canonical
        canonical_fullname = ".".join(parts)
        try:
            canonical_module = importlib.import_module(canonical_fullname)
        except ModuleNotFoundError as err:
            # Only treat a missing *canonical target* as "no alias to
            # resolve" (let the normal import machinery report it). If some
            # other module is missing, the canonical exists but failed to
            # import one of its own dependencies — surface that real error
            # rather than masking it as an unresolved alias.
            if err.name == canonical_fullname:
                return None
            raise
        # Do NOT pre-populate ``sys.modules[fullname]`` here. Python's
        # ``_find_spec`` (in importlib._bootstrap) has an optimization that
        # detects ``name in sys.modules`` after a finder returns and prefers
        # ``sys.modules[name].__spec__`` over the finder's spec — for an
        # alias, that's the canonical module's own SourceFileLoader spec,
        # which Python then *re-loads*, defeating the aliasing. Letting
        # ``_load_unlocked`` populate sys.modules itself (via our
        # ``_AliasLoader.create_module``) sidesteps that branch.
        return importlib.util.spec_from_loader(fullname, _AliasLoader(canonical_module))


class _AliasLoader(importlib.abc.Loader):
    """No-op loader that returns the already-resolved canonical module.

    :class:`_AliasFinder` populates ``sys.modules`` itself; this loader
    just satisfies the :mod:`importlib` protocol so Python doesn't try to
    re-execute the module.
    """

    def __init__(self, module: ModuleType) -> None:
        self._module = module

    def create_module(self, spec):  # noqa: ARG002
        return self._module

    def exec_module(self, module):  # noqa: ARG002
        # Nothing to execute — the canonical module is already initialized.
        return None


# Register once at module load. Idempotent: re-installing the finder on
# repeated imports (e.g. by tests that reload `esphome.loader`) is a no-op
# because we check for an existing instance first.
def _install_alias_finder() -> None:
    for entry in sys.meta_path:
        if isinstance(entry, _AliasFinder):
            return
    sys.meta_path.append(_AliasFinder())


_install_alias_finder()