HomeAssistantVS/custom_components/spook/repairs.py

"""Spook - Your homie."""

from __future__ import annotations

from abc import ABC, abstractmethod
import asyncio
from contextlib import suppress
from dataclasses import dataclass, field
from datetime import timedelta
import hashlib
import importlib
import json
from pathlib import Path
from typing import TYPE_CHECKING, Any, final

from homeassistant.components import blueprint, lovelace as lovelace_const
from homeassistant.components.automation import DOMAIN as AUTOMATION_DOMAIN
from homeassistant.components.energy.data import (
    async_get_manager as async_get_energy_manager,
)
from homeassistant.components.homeassistant import SERVICE_HOMEASSISTANT_RESTART
from homeassistant.components.repairs import ConfirmRepairFlow, RepairsFlow
from homeassistant.components.script import DOMAIN as SCRIPT_DOMAIN
from homeassistant.config_entries import (
    SIGNAL_CONFIG_ENTRY_CHANGED,
    ConfigEntry,
    ConfigEntryChange,
)
from homeassistant.const import CONF_ENTITIES, EVENT_STATE_CHANGED
from homeassistant.core import Event, HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import (
    area_registry as ar,
    collection,
    device_registry as dr,
    entity_registry as er,
    floor_registry as fr,
    issue_registry as ir,
    label_registry as lr,
)
from homeassistant.helpers.debounce import Debouncer
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_component import DATA_INSTANCES
from homeassistant.helpers.event import async_call_later, async_track_time_interval
from homeassistant.helpers.recorder import get_instance
from homeassistant.util.async_ import create_eager_task

from .attribute_checking import async_unknown_attributes
from .attribute_value_checking import as_yaml, async_unknown_attribute_values
from .const import DOMAIN, LOGGER
from .dashboard_resources import is_yaml_managed, redundant_item_ids
from .dismissals import async_get_dismissals
from .ectoplasms.spook.triggers.state_changed import async_watched_entity_ids
from .energy_preferences import (
    async_unknown_energy_entities,
    energy_preferences_without,
)
from .entity_filtering import (
    async_filter_known_entity_ids,
    async_get_all_entity_ids,
    async_name_helper_in_the_registry,
    is_not_an_entity_id,
)
from .entity_suggestions import (
    async_describe_warmed_unknown_entities,
    async_warm_rename_suggestions,
)
from .helper_sources import (
    MIN_MAX_ENTITY_IDS,
    async_helper_sources,
    async_unknown_helper_sources,
    async_unknown_min_max_members,
)
from .reference_extraction import (
    AttributeValue,
    NamedReferences,
    async_collect_mentioned_strings,
    async_referencing,
    extract_attribute_references_from_config,
    extract_state_references_from_config,
)
from .registry_usage import async_area_in_use, async_floor_in_use, async_label_in_use
from .repair_documentation import repair_documentation_url
from .state_checking import async_unknown_states
from .statistics_sources import async_settled_orphaned_statistic_ids

if TYPE_CHECKING:
    from collections.abc import Callable, Coroutine, Iterable, Mapping, Sized
    from datetime import datetime
    from types import ModuleType

    from homeassistant.data_entry_flow import FlowResult
    from homeassistant.util.event_type import EventType


# Yield to the event loop after every batch of this many inspected
# entities; inspections are CPU-bound and must not stall the loop on
# large installations.
INSPECTION_YIELD_INTERVAL = 50

# How long to give the recorder to say it has cleared what it was asked to.
# The same as Home Assistant allows its own Statistics page, which asks the
# recorder the same question the same way.
_CLEARING_TAKES_AT_MOST = 10

# Enough of a digest to tell two sets of findings apart. A collision would
# mean one dismissal covering a different finding in the same place, which is
# the thing this exists to prevent, and eight hex characters make that four
# billion to one.
_FINGERPRINT_LENGTH = 8


# A min/max helper needs at least this many members to function; Spook must
# not prune it below this.
_MIN_MAX_MINIMUM_MEMBERS = 2

# How long the first round of inspections after a start is spread over. All
# of them looking at once, on an instance that has only just started and
# often still has integrations adding devices, is a solid block of work at
# the worst moment for it. #1898.
FIRST_INSPECTIONS_SPREAD_OVER = timedelta(minutes=2)


def _plural(items: Sized) -> str:
    """Return the plural suffix for a sized collection."""
    return "" if len(items) == 1 else "s"


def _bulleted(items: Iterable[str]) -> str:
    """Return the markdown list a repair puts its findings in."""
    return "\n".join(f"- `{item}`" for item in items)


def format_config_entry_references(
    hass: HomeAssistant, entry_ids: Iterable[str]
) -> str:
    """Return the markdown list of config entries an action cannot use.

    One that exists was reported for being of another integration than the
    action asks for, so it says whose it is. That is usually the clue: the ID
    was copied over from an action of that other integration.
    """
    lines = []
    for entry_id in entry_ids:
        if (entry := hass.config_entries.async_get_entry(entry_id)) is None:
            lines.append(f"- `{entry_id}`")
        else:
            lines.append(
                f"- `{entry_id}` (belongs to {entry.title}, of `{entry.domain}`)"
            )
    return "\n".join(lines)


def _fingerprint(references: Iterable[str]) -> str:
    r"""Return a short digest of what a finding is about.

    Sorted first, because the same findings in a different order are the same
    findings, and an ID that moved would resurface an issue somebody had
    already dealt with.

    Each one is written down with its length in front of it, so that the
    digest reads two different sets two different ways. Joining them with a
    separator does not: a reference holding that separator borrows the one
    next to it, and `{"a\nb", "c"}` and `{"a", "b\nc"}` come out identical.
    Entity IDs cannot do that, but resource URLs, notifier names and
    customize keys are whatever somebody typed.
    """
    digest = hashlib.sha256()
    for reference in sorted(references):
        encoded = reference.encode()
        digest.update(f"{len(encoded)}:".encode())
        digest.update(encoded)

    return digest.hexdigest()[:_FINGERPRINT_LENGTH]


class AbstractSpookRepairBase(ABC):
    """Abstract base class to hold a Spook repairs."""

    domain: str
    repair: str

    hass: HomeAssistant
    issue_registry: ir.IssueRegistry
    area_registry: ar.AreaRegistry
    device_registry: dr.DeviceRegistry
    entity_registry: er.EntityRegistry

    issue_ids: set[str]

    def __init__(self, hass: HomeAssistant) -> None:
        """Initialize the service."""
        self.hass = hass
        self.issue_registry = ir.async_get(hass)
        self.area_registry = ar.async_get(hass)
        self.device_registry = dr.async_get(hass)
        self.entity_registry = er.async_get(hass)
        self.issue_ids = set()
        self._deactivated = False

    @final
    @callback
    # pylint: disable-next=too-many-arguments
    def async_create_issue(  # noqa: PLR0913
        self,
        *,
        breaks_in_ha_version: str | None = None,
        data: dict[str, str | int | float | None] | None = None,
        is_fixable: bool = False,
        is_persistent: bool = False,
        issue_domain: str | None = None,
        issue_id: str,
        learn_more_url: str | None = None,
        references: Iterable[str] | None = None,
        severity: ir.IssueSeverity = ir.IssueSeverity.WARNING,
        translation_key: str | None = None,
        translation_placeholders: dict[str, str] | None = None,
    ) -> None:
        """Create an issue.

        `translation_key` defaults to the repair's own name, which is what
        nearly every repair wants. Pass one when the same finding needs to be
        worded differently depending on what the house looks like, so that the
        alternative is a translated string of its own rather than a sentence
        smuggled in through a placeholder.

        `references` is what the issue is reporting, when that is a list of
        things rather than the one place they were found in. Pass it and the
        ID follows the findings, so that pressing "ignore" means "not these"
        rather than "nothing here, ever". Without it an issue keyed to a
        script keeps its dismissal while its text is quietly rewritten
        underneath, and the next genuinely broken thing in that script is
        hidden by a decision somebody made about something else. #1395.
        """
        if self._deactivated:
            return

        owner = issue_id
        findings = None if references is None else list(references)
        if findings is not None:
            issue_id = f"{issue_id}_{_fingerprint(findings)}"

        self.issue_ids.add(issue_id)
        ir.async_create_issue(
            self.hass,
            breaks_in_ha_version=breaks_in_ha_version,
            data=data,
            domain=DOMAIN,
            is_fixable=is_fixable,
            is_persistent=is_persistent,
            issue_domain=issue_domain or self.domain,
            issue_id=f"{self.repair}_{issue_id}",
            learn_more_url=learn_more_url or repair_documentation_url(self.repair),
            severity=severity,
            translation_key=translation_key or self.repair,
            translation_placeholders=translation_placeholders,
        )

        if findings is not None:
            self._async_carry_over_dismissal(
                f"{self.repair}_{issue_id}", owner, findings
            )

    @final
    @callback
    def _async_carry_over_dismissal(
        self,
        issue_id: str,
        owner: str,
        findings: list[str],
    ) -> None:
        """Keep an issue ignored when what it reports was ignored already.

        An issue filed under its findings is a new issue whenever they change,
        and Home Assistant only remembers an ignore on the issue it was
        pressed on. So a list that got shorter, or one that came back after
        an automation reloaded, arrived as something nobody had ever seen.
        Spook writes the decision down itself, and anything it reports that
        was all ignored before is ignored again from the start. Anything new
        on the list makes it news, and it comes up as usual.
        """
        dismissals = async_get_dismissals(self.hass)
        dismissals.async_offer(issue_id, self.repair, owner, findings)

        issue = self.issue_registry.async_get_issue(DOMAIN, issue_id)
        if issue is not None and issue.dismissed_version is not None:
            # Ignored, but perhaps from before Spook wrote these down, or by a
            # route it did not see. Either way it is a decision to keep.
            dismissals.async_dismiss_offered(issue_id)
            return

        if set(findings) <= dismissals.async_dismissed(self.repair, owner):
            ir.async_ignore_issue(self.hass, DOMAIN, issue_id, ignore=True)

    @final
    @callback
    def async_delete_issue(
        self,
        issue_id: str,
    ) -> None:
        """Remove an issue."""
        if self._deactivated:
            return

        self.issue_ids.discard(issue_id)
        ir.async_delete_issue(
            self.hass,
            domain=DOMAIN,
            issue_id=f"{self.repair}_{issue_id}",
        )

    @abstractmethod
    async def async_activate(self) -> None:
        """Handle the activating a repair."""
        raise NotImplementedError

    @abstractmethod
    async def async_inspect(self) -> None:
        """Trigger a repair check."""
        raise NotImplementedError

    async def async_deactivate(self) -> None:
        """Unregister the repair, and leave what it reported where it is.

        Deliberately takes nothing down, and that is the point of it. Somebody
        pressing "ignore" has that written on the issue itself, so deleting
        the issue takes the mark with it and the next inspection puts the same
        thing back as something nobody has ever seen. Home Assistant keeps an
        issue over a restart for exactly that reason, as an empty record
        holding only the mark, and reporting the same thing again finds it and
        leaves it alone.

        Which makes this a matter of not getting in the way. Every repair
        already compares what it left in the registry against what it finds
        when it next looks, and deletes what is no longer there, so the tidying
        this used to do happens anyway and happens later, when there is
        something to compare against. #1572.

        What it does stop is a look already under way. Shutting the debouncer
        down cancels the next one, not one waiting halfway through, and that
        one carried on afterwards: filing what it found for a Spook that was
        disabled, or after a reload clearing the fresh issues of the repair
        that replaced it. From here on it changes nothing.
        """
        self._deactivated = True


class AbstractSpookRepair(AbstractSpookRepairBase):
    """Abstract base class to hold a Spook repairs."""

    inspect_events: set[EventType[Any] | str] | None = None
    inspect_debouncer: Debouncer[Coroutine[Any, Any, None]]
    inspect_config_entry_changed: bool | str = False
    inspect_on_reload: bool | str = False

    #: Re-run the inspection on this fixed interval, on top of any event
    #: triggers. Needed by repairs whose findings change with the passage of
    #: time alone (e.g. something going stale), not in response to an event.
    inspect_interval: timedelta | None = None

    #: Wait this long after activating before looking at all, and ignore
    #: everything that would trigger a look until then. For repairs that ask
    #: the recorder, which has its hands full right after a start. After a
    #: start the repair manager hands every repair without one its own turn
    #: instead, so they do not all look at the same moment.
    first_inspection_delay: timedelta | None = None

    #: Re-run the inspection when an entity appears or goes. Needed by repairs
    #: about entity references, as plenty of entities never touch the entity
    #: registry: one set by a script or a template without a unique ID arrives
    #: as a state and nothing else, often well after Home Assistant started.
    inspect_on_entity_added_or_removed: bool = False

    automatically_clean_up_issues: bool = False
    possible_issue_ids: set[str]

    _event_subs: set[Callable[[], None]]

    def __init__(self, hass: HomeAssistant) -> None:
        """Initialize the repair."""
        super().__init__(hass)
        self._event_subs = set()
        self.possible_issue_ids = set()
        self._waiting_for_first_inspection = False

    async def _async_inspect_with_cleanup(self) -> None:
        """Run an inspection and clean up issues that are no longer valid."""
        # Don't inspect if we are stopping, or should not look yet
        if self.hass.is_stopping or self._waiting_for_first_inspection:
            return

        if not self.automatically_clean_up_issues:
            await self.async_inspect()
            return

        # Issues registered by earlier inspections. Anything not re-registered
        # during this inspection is no longer valid, including issues for
        # items that were removed entirely since the previous inspection.
        previous_issue_ids = self.issue_ids.copy()

        # Issues persisted in the issue registry for this repair. Covers
        # leftovers from before a restart, including those for items that
        # were removed while Home Assistant was down.
        prefix = f"{self.repair}_"
        registry_issue_ids = {
            issue_id.removeprefix(prefix)
            for domain, issue_id in self.issue_registry.issues
            if domain == DOMAIN and issue_id.startswith(prefix)
        }

        # Reset registered issues. If they are still valid, they will be
        # re-registered during the inspection.
        self.issue_ids.clear()

        try:
            await self.async_inspect()
        except Exception:
            # Restore the bookkeeping so the next successful inspection can
            # still clean up issues from before the failure.
            self.issue_ids.update(previous_issue_ids)
            raise

        # Remove issues that are no longer valid after the inspection:
        # - previous_issue_ids covers issues whose item was resolved or
        #   removed since the previous inspection.
        # - registry_issue_ids covers stale issues persisted from an earlier
        #   runtime.
        # - possible_issue_ids covers inspected items, as a safety net.
        stale_issue_ids = (
            previous_issue_ids | registry_issue_ids | self.possible_issue_ids
        ) - self.issue_ids
        for issue_id in stale_issue_ids:
            self.async_delete_issue(issue_id)

    async def async_activate(self) -> None:  # noqa: C901
        """Handle the activating a repair."""
        # Debouncer to prevent multiple inspections / inspections fired quickly
        # after each other.
        self.inspect_debouncer = Debouncer(
            self.hass,
            LOGGER,
            cooldown=3,
            immediate=False,
            function=self._async_inspect_with_cleanup,
        )

        if self.first_inspection_delay is None:
            # Spook says: Bounce!
            await self.inspect_debouncer.async_call()
        else:
            self._waiting_for_first_inspection = True

            @callback
            def _async_first_inspection(_: datetime) -> None:
                self._waiting_for_first_inspection = False
                self.inspect_debouncer.async_schedule_call()

            self._event_subs.add(
                async_call_later(
                    self.hass, self.first_inspection_delay, _async_first_inspection
                )
            )

        async def _async_call_inspect_debouncer(_: Event) -> None:
            # Trigger an inspection when an event is received from the event bus.
            await self.inspect_debouncer.async_call()

        if self.inspect_events is not None:
            for event in self.inspect_events:
                self._event_subs.add(
                    self.hass.bus.async_listen(event, _async_call_inspect_debouncer),
                )

        if self.inspect_interval is not None:

            async def _async_call_inspect_debouncer_interval(_: datetime) -> None:
                # Trigger an inspection when the interval timer fires.
                await self.inspect_debouncer.async_call()

            self._event_subs.add(
                async_track_time_interval(
                    self.hass,
                    _async_call_inspect_debouncer_interval,
                    self.inspect_interval,
                ),
            )

        if self.inspect_on_entity_added_or_removed:

            @callback
            def _entity_added_or_removed(event_data: Mapping[str, Any]) -> bool:
                """Return whether an entity appeared or went, not just changed."""
                return (
                    event_data.get("old_state") is None
                    or event_data.get("new_state") is None
                )

            @callback
            def _async_entity_added_or_removed(_: Event) -> None:
                self.inspect_debouncer.async_schedule_call()

            self._event_subs.add(
                self.hass.bus.async_listen(
                    EVENT_STATE_CHANGED,
                    _async_entity_added_or_removed,
                    event_filter=_entity_added_or_removed,
                ),
            )

        if self.inspect_on_reload:

            @callback
            def _filter_event(data: Mapping[str, Any] | Event) -> bool:
                """Filter for reload events."""
                event_data = data.data if isinstance(data, Event) else data
                service = event_data.get("service")
                if service is None:
                    return False
                if service == "reload_all":
                    return True
                if service != "reload":
                    return False
                if self.inspect_on_reload is True:
                    return True
                return self.inspect_on_reload == event_data.get("domain")

            self._event_subs.add(
                self.hass.bus.async_listen(
                    "call_service",
                    _async_call_inspect_debouncer,
                    event_filter=_filter_event,
                ),
            )

        if self.inspect_config_entry_changed:

            async def _async_config_entry_changed(  # pylint: disable=unused-argument
                change: ConfigEntryChange,  # noqa: ARG001
                entry: ConfigEntry,
            ) -> None:
                """Handle options update."""
                if (
                    self.inspect_config_entry_changed is not True
                    and entry.domain != self.inspect_config_entry_changed
                ):
                    return
                await self.inspect_debouncer.async_call()

            self._event_subs.add(
                async_dispatcher_connect(
                    self.hass,
                    SIGNAL_CONFIG_ENTRY_CHANGED,
                    _async_config_entry_changed,
                ),
            )

    async def async_deactivate(self) -> None:
        """Unregister the repair."""
        for sub in self._event_subs.copy():
            sub()
            self._event_subs.discard(sub)
        self.inspect_debouncer.async_shutdown()
        await super().async_deactivate()


class AbstractSpookEntityComponentUnknownReferencesRepair(AbstractSpookRepair, ABC):
    """Base class for repairs that find unknown references in component entities.

    Handles the shared boilerplate for inspecting entities loaded via
    `EntityComponent` (e.g. automations, scripts): iterating the component's
    entities, skipping unavailable ones, computing per-entity unknown references
    via a subclass hook, and creating an issue with the standard translation
    placeholders (``<reference_label>``, ``<entity_label>``, ``edit``,
    ``entity_id``).
    """

    automatically_clean_up_issues = True

    #: Entity class representing an unavailable/broken instance. Entities of
    #: this type are skipped during issue creation. ``None`` inspects every
    #: entity, including unavailable ones; repairs that diagnose *why* an
    #: entity is broken need exactly those.
    unavailable_entity_class: type | None = None

    #: Translation placeholder key holding the entity's display name (e.g.
    #: ``"automation"`` or ``"script"``).
    entity_label: str

    #: Translation placeholder key holding the bulleted list of unknown
    #: references (e.g. ``"areas"``, ``"floors"``, ``"entities"``).
    reference_label: str

    #: Format string used to build the ``edit`` placeholder. Must contain a
    #: ``{unique_id}`` field (e.g. ``"/config/automation/edit/{unique_id}"``).
    edit_url_pattern: str

    #: When the references are entities, enrich each with why it is unknown
    #: (deleted on/by, or a likely rename). Only set on entity repairs.
    references_are_entities: bool = False

    def _format_references(self, references: list[str]) -> str:
        """Return the bulleted reference list for the issue message."""
        if not self.references_are_entities:
            return "\n".join(f"- `{reference}`" for reference in references)

        # Something that is no entity ID at all comes after the entities that
        # do not exist, said to be what it is. No guess at what was meant:
        # `cover.blind.position` may have meant the cover, its position or
        # something else entirely, and a wrong suggestion is worse than none.
        lines = [
            # Warmed for the whole round in `async_inspect` already.
            async_describe_warmed_unknown_entities(
                self.hass,
                [entity for entity in references if not is_not_an_entity_id(entity)],
            ),
            *(
                f"- `{value}` (not an entity ID)"
                for value in references
                if is_not_an_entity_id(value)
            ),
        ]
        return "\n".join(line for line in lines if line)

    async def _async_setup_inspection(self) -> None:
        """Prepare per-inspection state (called once per inspection cycle).

        Override to cache lookups (e.g. known IDs from a registry) on ``self``
        for use during per-entity inspection.
        """

    # pylint: disable-next=unused-argument
    def _should_inspect_entity(self, entity: Any) -> bool:  # noqa: ARG002
        """Decide whether the given entity should be inspected.

        Defaults to inspecting every entity. Override (e.g. to skip disabled
        entities) when needed.
        """
        return True

    @abstractmethod
    async def _async_compute_unknown_references(self, entity: Any) -> set[str]:
        """Return the set of unknown referenced IDs for a single entity."""

    def _edit_url(self, entity: Any) -> str:
        """Return the URL to edit the given entity.

        Items created in YAML can lack a unique ID, in which case there is no
        editor to deep-link to; fall back to the domain's overview page.
        """
        if entity.unique_id is None:
            return f"/config/{self.domain}/dashboard"
        return self.edit_url_pattern.format(unique_id=entity.unique_id)

    async def async_inspect(self) -> None:
        """Trigger an inspection.

        Nothing goes into ``possible_issue_ids`` here. An issue raised by this
        repair is keyed to its findings rather than to the entity they were
        found in, so a list of inspected entities no longer names anything
        that could be in the registry. What this repair left behind is read
        back out of the registry instead, which finds all of it.
        """
        if self.domain not in (instances := self.hass.data.get(DATA_INSTANCES, {})):
            return

        entity_component = instances[self.domain]

        LOGGER.debug("Spook is inspecting: %s", self.repair)

        await self._async_setup_inspection()

        # Taken as a snapshot, because this loop gives the event loop a turn
        # every so often and an automation being added or removed during one of
        # those turns changes the collection underneath it. Home Assistant says
        # so itself: "callers that iterate over this asynchronously should make
        # a copy". Without it the whole inspection dies on a `RuntimeError` and
        # takes the round with it, since nothing above catches that. #1558.
        #
        # An entity that arrives while a round is running is missed until the
        # next one, which is minutes away and no worse than arriving a moment
        # after the round finished.
        entities = list(entity_component.entities)

        # Collected first and reported afterwards, rather than an issue raised
        # the moment one is found. Describing a broken entity reference means
        # working out what it was probably meant to say, which is the most
        # expensive thing in the whole inspection, and gathering the round's
        # findings first lets all of that happen in one go. #1667.
        findings: list[tuple[Any, list[str]]] = []

        for index, entity in enumerate(entities):
            if index and index % INSPECTION_YIELD_INTERVAL == 0:
                # Inspections are CPU-bound; periodically yield to the event
                # loop so large installations do not stall it.
                await asyncio.sleep(0)

            unavailable_class = self.unavailable_entity_class
            # pylint: disable-next=isinstance-second-argument-not-valid-type
            if unavailable_class is not None and isinstance(entity, unavailable_class):
                continue

            if not self._should_inspect_entity(entity):
                continue

            unknown = await self._async_compute_unknown_references(entity)
            if not unknown:
                continue

            findings.append((entity, sorted(unknown)))

        if self.references_are_entities and findings:
            await async_warm_rename_suggestions(
                self.hass,
                {
                    reference
                    for _, references in findings
                    for reference in references
                    if not is_not_an_entity_id(reference)
                },
            )

        for entity, sorted_unknown in findings:
            self.async_create_issue(
                issue_id=entity.entity_id,
                references=sorted_unknown,
                translation_placeholders={
                    self.reference_label: self._format_references(sorted_unknown),
                    self.entity_label: entity.name,
                    "edit": self._edit_url(entity),
                    "entity_id": entity.entity_id,
                },
            )
            LOGGER.debug(
                "Spook found unknown %s in %s and created an issue for it; %s: %s",
                self.reference_label,
                entity.entity_id,
                self.reference_label.capitalize(),
                ", ".join(sorted_unknown),
            )


def _finding_reference(entity_id: str, name: str | AttributeValue) -> str:
    """Return how an unknown attribute or state is written down in an issue.

    A value of an attribute keeps its type: `2` and `"2"` are not the same
    thing to an attribute. One compared as text, like Spook's own trigger
    does, is kept apart too: it can have another best guess than the same
    value compared the way core does, and both belong in the issue.
    """
    if isinstance(name, AttributeValue):
        reference = f"{entity_id}[{name.attribute}]:{json.dumps(name.value)}"
        return f"{reference}:as-text" if name.as_text else reference
    return f"{entity_id}:{name}"


@dataclass(frozen=True, slots=True)
class _UnknownName:
    """Something an entity never has, named in an automation or script."""

    entity_id: str
    name: str | AttributeValue
    suggestion: str | None


class AbstractSpookUnknownEntityNamesRepair(
    AbstractSpookEntityComponentUnknownReferencesRepair, ABC
):
    """Base for repairs about what an entity never has: an attribute, a state.

    A trigger or condition waiting on one loads fine and never fires, and
    says nothing about it. What an entity has is only known at runtime, and
    the last place to look is the recorder, so this does not look right at
    the start, when the recorder has its hands full. After that it looks
    whenever one of its own entities is added or removed, which is how a
    reload that changed something ends, on the usual events, and once a day
    for what changes with time alone: history being written.

    The whole round's references are worked out up front, so the recorder is
    asked once per round rather than once per automation.
    """

    first_inspection_delay = timedelta(minutes=10)
    inspect_interval = timedelta(days=1)

    #: How one finding reads in the issue, with ``{name}`` and ``{entity_id}``.
    finding_line: str

    _unknown_by_owner: dict[str, set[str]]
    _unknown_by_reference: dict[str, _UnknownName]
    _inspected_configs: dict[str, Any]
    _named: dict[str, tuple[Any, NamedReferences]]

    def __init__(self, hass: HomeAssistant) -> None:
        """Initialize the repair."""
        super().__init__(hass)
        self._named = {}
        self._inspected_configs = {}
        self._unknown_by_owner = {}
        self._unknown_by_reference = {}

    @abstractmethod
    def _references_in(self, config: dict[str, Any]) -> NamedReferences:
        """Return what a configuration names, and of which entities."""

    @abstractmethod
    async def _async_unknown(
        self, pairs: set[tuple[str, str | AttributeValue]]
    ) -> set[_UnknownName]:
        """Return the (entity ID, name) pairs the entity never has."""

    async def async_activate(self) -> None:
        """Look again once one of its own entities was added or removed.

        A reload removes what changed and adds it back once it is done, so
        that is when there is something new to look at. The reload service
        being called is too early: nothing has been reloaded yet.
        """
        await super().async_activate()

        prefix = f"{self.domain}."

        @callback
        def _own_entity_added_or_removed(event_data: Mapping[str, Any]) -> bool:
            """Return whether one of this domain's entities appeared or went."""
            return event_data["entity_id"].startswith(prefix) and (
                event_data.get("old_state") is None
                or event_data.get("new_state") is None
            )

        @callback
        def _look_again(_: Event) -> None:
            self.inspect_debouncer.async_schedule_call()

        self._event_subs.add(
            self.hass.bus.async_listen(
                EVENT_STATE_CHANGED,
                _look_again,
                event_filter=_own_entity_added_or_removed,
            )
        )

    def _named_in(
        self, entity: Any, config: dict[str, Any]
    ) -> set[tuple[str, str | AttributeValue]]:
        """Return the (entity ID, name) pairs a configuration names.

        Read once per configuration, remembered against the configuration
        itself: a reload hands over a new one.
        """
        named = self._named.get(entity.entity_id)
        if named is None or named[0] is not config:
            named = (config, self._references_in(config))
        self._named[entity.entity_id] = named
        references = named[1]

        # Resolved every round rather than remembered: what a registry ID
        # stands for moves when the entity is renamed, and what Spook's own
        # trigger watches moves with the registry. That one is asked of the
        # trigger itself, so it is exactly what the trigger would watch.
        pairs = {
            (entity_id, name)
            for reference, name in references.pairs
            if (
                entity_id := er.async_resolve_entity_id(self.entity_registry, reference)
            )
        }
        for trigger_config, name in references.followed:
            pairs.update(
                (entity_id, name)
                for entity_id in async_watched_entity_ids(self.hass, trigger_config)
            )
        return pairs

    def _is_inspected(self, entity: Any) -> bool:
        """Return whether the round looks at this entity, the way the base does."""
        unavailable_class = self.unavailable_entity_class
        # pylint: disable-next=isinstance-second-argument-not-valid-type
        if unavailable_class is not None and isinstance(entity, unavailable_class):
            return False
        return self._should_inspect_entity(entity)

    async def _async_setup_inspection(self) -> None:
        """Work out everything unknown of the round, in one go."""
        entities = list(self.hass.data[DATA_INSTANCES][self.domain].entities)

        configs: dict[str, Any] = {}
        named: dict[str, set[tuple[str, str | AttributeValue]]] = {}
        for entity in entities:
            # Reading a configuration is CPU-bound, and one can be big: the
            # event loop gets a turn after each.
            await asyncio.sleep(0)

            config = getattr(entity, "raw_config", None)
            if not isinstance(config, dict) or not self._is_inspected(entity):
                continue

            configs[entity.entity_id] = config
            named[entity.entity_id] = self._named_in(entity, config)

        # Whatever is gone or not looked at this round is forgotten.
        self._named = {
            entity_id: remembered
            for entity_id, remembered in self._named.items()
            if entity_id in named
        }

        unknown = await self._async_unknown(
            {pair for pairs in named.values() for pair in pairs}
        )
        by_pair = {(finding.entity_id, finding.name): finding for finding in unknown}

        self._inspected_configs = configs
        self._unknown_by_reference = {
            _finding_reference(*pair): finding for pair, finding in by_pair.items()
        }
        self._unknown_by_owner = {
            owner: {_finding_reference(*pair) for pair in pairs & by_pair.keys()}
            for owner, pairs in named.items()
        }

    async def _async_compute_unknown_references(self, entity: Any) -> set[str]:
        """Return what this entity names that its entities never have.

        Only for the very configuration it was worked out from, on the
        entity that is there now. One that was reloaded while the recorder
        was being asked, or while this round went by, may say something else
        entirely, and the next round looks at that.
        """
        current = self.hass.data[DATA_INSTANCES][self.domain].get_entity(
            entity.entity_id
        )
        if (
            current is not entity
            or self._inspected_configs.get(entity.entity_id) is None
            or getattr(entity, "raw_config", None)
            is not self._inspected_configs[entity.entity_id]
        ):
            return set()
        return self._unknown_by_owner.get(entity.entity_id, set())

    async def async_check_draft(self, draft: Any) -> list[str]:
        """Return what a draft names that is unknown, each with its best guess.

        A draft is in no round: it is not among the entities a round goes by,
        so it is asked about on its own, with the same readers.
        """
        unknown = await self._async_unknown(self._named_in(draft, draft.raw_config))

        found = []
        for finding in unknown:
            if isinstance(name := finding.name, AttributeValue):
                line = f"{name.attribute} of {finding.entity_id}: {as_yaml(name.value)}"
            else:
                line = _finding_reference(finding.entity_id, name)
            if finding.suggestion is not None:
                line += f" (did you mean {finding.suggestion}?)"
            found.append(line)
        return sorted(found)

    def _format_references(self, references: list[str]) -> str:
        """Return the list of findings, each with its entity and best guess."""
        lines = []
        for reference in references:
            finding = self._unknown_by_reference[reference]
            if isinstance(name := finding.name, AttributeValue):
                line = (
                    f"- `{name.attribute}` of `{finding.entity_id}`: "
                    f"`{as_yaml(name.value)}`"
                )
            else:
                line = self.finding_line.format(name=name, entity_id=finding.entity_id)
            if finding.suggestion is not None:
                line += f" (did you mean `{finding.suggestion}`?)"
            lines.append(line)
        return "\n".join(lines)


class AbstractSpookUnknownAttributesRepair(AbstractSpookUnknownEntityNamesRepair, ABC):
    """Base for repairs about attributes an entity never had."""

    reference_label = "attributes"
    finding_line = "- `{name}` of `{entity_id}`"

    def _references_in(self, config: dict[str, Any]) -> NamedReferences:
        """Return the attributes a configuration names, and of which entities."""
        return extract_attribute_references_from_config(config)

    async def _async_unknown(
        self, pairs: set[tuple[str, str | AttributeValue]]
    ) -> set[_UnknownName]:
        """Return the (entity ID, attribute) pairs the entity never had."""
        attributes = {
            (entity_id, name) for entity_id, name in pairs if isinstance(name, str)
        }
        return {
            _UnknownName(finding.entity_id, finding.attribute, finding.suggestion)
            for finding in await async_unknown_attributes(self.hass, attributes)
        }


class AbstractSpookUnknownStatesRepair(AbstractSpookUnknownEntityNamesRepair, ABC):
    """Base for repairs about states an entity is never in.

    Only for entities whose states are a fixed set; `state_checking` says
    which those are. The values of an attribute a trigger or condition waits
    for are in here too: the same mistake, made one level down, and fixed in
    the same place. `attribute_value_checking` says which attributes.
    """

    reference_label = "states"
    finding_line = "- `{name}` for `{entity_id}`"

    def _references_in(self, config: dict[str, Any]) -> NamedReferences:
        """Return the states a configuration names, and of which entities."""
        return extract_state_references_from_config(config)

    async def _async_unknown(
        self, pairs: set[tuple[str, str | AttributeValue]]
    ) -> set[_UnknownName]:
        """Return the (entity ID, state or attribute value) pairs never there."""
        states: set[tuple[str, str]] = set()
        values: set[tuple[str, AttributeValue]] = set()
        for entity_id, name in pairs:
            if isinstance(name, AttributeValue):
                values.add((entity_id, name))
            else:
                states.add((entity_id, name))

        unknown = {
            _UnknownName(finding.entity_id, finding.state, finding.suggestion)
            for finding in await async_unknown_states(self.hass, states)
        }
        unknown.update(
            _UnknownName(
                finding.entity_id,
                AttributeValue(
                    finding.attribute, finding.value, as_text=finding.as_text
                ),
                finding.suggestion,
            )
            for finding in await async_unknown_attribute_values(self.hass, values)
        )
        return unknown


class AbstractSpookSingleShotRepairs(AbstractSpookRepairBase, ABC):
    """Abstract class to hold repairs that are single a shot."""

    @final
    async def async_activate(self) -> None:
        """Actives the repairs."""
        await self.async_inspect()

    @final
    async def async_deactivate(self) -> None:
        """Unregister the repair."""
        await super().async_deactivate()


@dataclass
class SpookRepairManager:
    """Class to manage Spook repairs."""

    hass: HomeAssistant

    _repairs: set[AbstractSpookRepair] = field(default_factory=set)

    def __post_init__(self) -> None:
        """Post initialization."""
        self.issue_registry = ir.async_get(self.hass)
        LOGGER.debug("Spook repair manager initialized")

    async def async_setup(self, *, spread_first_inspections: bool = False) -> None:
        """Set up the Spook repairs.

        Right after a start, the first inspections are spread out rather than
        all run at once. Anything that happens before a repair's turn comes
        is not looked at separately, as that first look sees how things ended
        up anyway. On a reload or a first setup on a running instance, the
        house is quiet and everything looks straight away, like it always did.
        """
        LOGGER.debug("Setting up Spook repairs")

        modules: list[ModuleType] = []

        def _load_all_repair_modules() -> None:
            """Load all repair modules."""
            for module_file in Path(__file__).parent.rglob("ectoplasms/*/repairs/*.py"):
                if module_file.name == "__init__.py":
                    continue
                module_path = str(module_file.relative_to(Path(__file__).parent))[
                    :-3
                ].replace("/", ".")
                modules.append(importlib.import_module(f".{module_path}", __package__))

        await self.hass.async_add_import_executor_job(_load_all_repair_modules)

        first_inspection_delays = (
            _first_inspection_delays(modules) if spread_first_inspections else {}
        )

        await asyncio.gather(
            *(
                create_eager_task(
                    self._async_setup_repair_module(
                        module, first_inspection_delays.get(module.__name__)
                    )
                )
                for module in modules
            )
        )

    async def _async_setup_repair_module(
        self, module: ModuleType, first_inspection_delay: timedelta | None = None
    ) -> None:
        """Set up a single repair module, isolating failures.

        A repair that fails to set up must not prevent the rest of Spook
        from loading.
        """
        try:
            repair = module.SpookRepair(self.hass)

            if first_inspection_delay is not None:
                repair.first_inspection_delay = first_inspection_delay

            await self.async_activate(repair)
        # pylint: disable-next=broad-exception-caught
        except Exception:  # noqa: BLE001
            LOGGER.exception(
                "Spook repair %s failed to set up and has been skipped; "
                "please report this issue at "
                "https://github.com/frenck/spook/issues",
                module.__name__,
            )

    async def async_activate(self, repair: AbstractSpookRepair) -> None:
        """Register a Spook repair."""
        LOGGER.debug(
            "Registering Spook repairs: %s.%s",
            repair.domain,
            repair.repair,
        )
        await repair.async_activate()
        self._repairs.add(repair)

    async def async_on_unload(self) -> None:
        """Tear down the Spook repairs.

        Nothing is removed from the issue registry on the way out, on purpose.

        Somebody pressing "ignore" on a repair has that written down on the
        issue itself, and deleting the issue throws it away with everything
        else. Home Assistant is built for it to survive: an issue that is not
        marked persistent is still kept over a restart as an empty record
        holding only that, and creating the same issue again finds that record
        and leaves the mark alone. Which means the way to keep an ignored
        repair ignored is to not touch it.

        Spook used to clear them all out here, so every reload, every
        integration reload and every update through HACS quietly undid every
        ignore anybody had ever pressed.

        Leaving them behind costs nothing. Every repair already looks at what
        it left in the registry when it next inspects, and deletes whatever it
        does not find again, which covers the very things this was for: an
        issue about something that was resolved or removed while Spook was not
        running.
        """
        LOGGER.debug("Tearing down Spook repairs")
        for repair in self._repairs:
            LOGGER.debug(
                "Unregistering Spook repair: %s.%s",
                repair.domain,
                repair.repair,
            )
            await repair.async_deactivate()


def _takes_a_turn(repair_class: type | None) -> bool:
    """Return whether a repair waits for a turn at its first look."""
    return (
        isinstance(repair_class, type)
        and issubclass(repair_class, AbstractSpookRepair)
        and repair_class.first_inspection_delay is None
    )


def _first_inspection_delays(modules: Iterable[ModuleType]) -> dict[str, timedelta]:
    """Give every repair module its turn for a first look, evenly spread.

    In order of module name, so the same repair always looks at the same
    moment after a start, which keeps it predictable for people and tests.
    Repairs that set their own delay keep it: they wait for the recorder,
    which is a later moment than any of these turns.
    """
    taking_turns = sorted(
        module.__name__
        for module in modules
        if _takes_a_turn(getattr(module, "SpookRepair", None))
    )
    if not taking_turns:
        return {}

    turn = FIRST_INSPECTIONS_SPREAD_OVER / len(taking_turns)
    return {name: turn * index for index, name in enumerate(taking_turns)}


class RestartRequiredFixFlow(RepairsFlow):
    """Handler for a repairs issue flow that restarts Home Assistant."""

    issue_id = "restart_required"

    async def async_step_init(
        self,
        _: dict[str, str] | None = None,
    ) -> FlowResult:
        """Handle asking confirmation of restart."""
        return await self.async_step_confirm_restart()

    async def async_step_confirm_restart(
        self,
        user_input: dict[str, str] | None = None,
    ) -> FlowResult:
        """Handle the confirm of restart."""
        if user_input is not None:
            await self.hass.services.async_call(
                "homeassistant",
                SERVICE_HOMEASSISTANT_RESTART,
            )
            return self.async_create_entry(data={})

        return self.async_show_form(step_id="confirm_restart")


def _offered(data: Mapping[str, Any] | None, key: str) -> set[str]:
    """Return what an issue put in front of somebody, from its data.

    A fix acts on that and nothing else. What is broken right now can be more,
    and acting on that would change something nobody was shown.
    """
    return {item for item in str((data or {}).get(key, "")).split(",") if item}


class _RemoveOrIgnoreFixFlow(RepairsFlow):
    """Base for a leftover registry thing: remove it, or keep and stop nagging.

    Leftover registry things (empty areas, empty floors, unused labels) are
    tidiness, not breakage, so keeping one is a valid choice. A fixable issue
    cannot be dismissed from the repairs UI, so the flow offers an explicit
    "keep it" option that ignores the issue instead. Subclasses set ``_key``
    (the placeholder key), ``_id_key`` (the data key holding the id), and
    implement ``_remove``.
    """

    #: Placeholder key naming the thing (e.g. "area").
    _key: str
    #: Data key holding the thing's id (e.g. "empty_area_id").
    _id_key: str

    async def async_step_init(
        self,
        _: dict[str, str] | None = None,
    ) -> FlowResult:
        """Offer to remove the thing, fix it yourself, or keep and ignore it."""
        return self.async_show_menu(
            step_id="init",
            menu_options=["remove", "manage", "ignore"],
            description_placeholders=self._menu_placeholders(),
        )

    def _menu_placeholders(self) -> dict[str, str]:
        """Return the placeholders naming the thing in the menu step."""
        return {self._key: str((self.data or {}).get(self._key, ""))}

    async def async_step_manage(
        self,
        _: dict[str, str] | None = None,
    ) -> FlowResult:
        """Point the user at where to fix it themselves.

        Aborts rather than completing, so the issue stays until the user
        actually resolves it. The abort message links to the right page.
        """
        return self.async_abort(reason="manage")

    async def async_step_remove(
        self,
        _: dict[str, str] | None = None,
    ) -> FlowResult:
        """Remove the thing, if it is still what the issue said it was.

        An issue can sit there for days before somebody presses the button,
        and something may have moved in since. Removing it anyway would take
        that with it, so a thing that changed is left alone and the issue
        says so. Spook looks again on its own and updates the issue.
        """
        if not self._remove(str((self.data or {}).get(self._id_key, ""))):
            return self.async_abort(reason="changed")

        return self.async_create_entry(data={})

    async def async_step_ignore(
        self,
        _: dict[str, str] | None = None,
    ) -> FlowResult:
        """Keep the thing and ignore the issue.

        Aborting (rather than creating an entry) keeps the issue so the
        ignore sticks; a completed fix flow would delete it and it would
        just come back on the next inspection.

        The issue can be gone by the time somebody chooses: its findings
        changed while the menu was open, or it was cleared for a moment while
        something reloaded. Home Assistant raises on ignoring an issue that is
        not there, so the decision is written down directly instead, and it
        still holds when the issue comes back.
        """
        if ir.async_get(self.hass).async_get_issue(DOMAIN, self.issue_id) is None:
            async_get_dismissals(self.hass).async_dismiss_offered(self.issue_id)
        else:
            ir.async_ignore_issue(self.hass, DOMAIN, self.issue_id, ignore=True)

        return self.async_abort(reason="issue_ignored")

    @callback
    def _remove(self, thing_id: str) -> bool:
        """Remove the thing by id. Implemented by subclasses.

        Returns `False` when it changed since the issue was raised and was
        left alone. Already gone counts as done.
        """
        raise NotImplementedError


class EmptyAreaFixFlow(_RemoveOrIgnoreFixFlow):
    """Handler for an empty area: remove it, or keep it and stop nagging."""

    _key = "area"
    _id_key = "empty_area_id"

    @callback
    def _remove(self, thing_id: str) -> bool:
        """Remove the area, if it is still empty."""
        registry = ar.async_get(self.hass)
        # The area may already be gone if removed elsewhere meanwhile.
        if not registry.async_get_area(thing_id):
            return True

        # Deleting an area unassigns everything in it, so something assigned
        # since the issue was raised would quietly lose its area.
        if async_area_in_use(
            self.hass, thing_id, async_collect_mentioned_strings(self.hass)
        ):
            return False

        registry.async_delete(thing_id)
        return True


class AreaUnknownSensorsFixFlow(_RemoveOrIgnoreFixFlow):
    """Handler for an area whose temperature or humidity sensor is gone.

    Clears the settings that point at nothing, so the area can be given a
    new sensor. Only the settings the issue showed, and of those only the ones
    still pointing at nothing: a sensor that came back since is kept, and a
    setting that broke after the issue was raised is not the one somebody
    agreed to clear.
    """

    _key = "area"
    _id_key = "area_sensors_area_id"

    def _menu_placeholders(self) -> dict[str, str]:
        """Name the area and its missing sensors in the menu step."""
        data = self.data or {}
        return {key: str(data.get(key, "")) for key in ("area", "entities")}

    @callback
    def _remove(self, thing_id: str) -> bool:
        """Clear the area's sensor settings that still point at what was shown.

        A setting is cleared only while it holds the very entity the issue
        named and that entity is still unknown. One changed to another sensor
        since, which then went missing too, is not the setting somebody saw.
        """
        area_registry = ar.async_get(self.hass)
        if (area := area_registry.async_get_area(thing_id)) is None:
            return True

        offered = dict(
            reference.split(":", 1)
            for reference in str(
                (self.data or {}).get("area_sensors_references", "")
            ).split(",")
            if ":" in reference
        )
        known_entity_ids = async_get_all_entity_ids(self.hass)
        cleared = {
            field: None
            for field, entity_id in offered.items()
            if field in ("temperature_entity_id", "humidity_entity_id")
            and getattr(area, field) == entity_id
            and async_filter_known_entity_ids(
                self.hass, [entity_id], known_entity_ids=known_entity_ids
            )
        }
        if not cleared:
            return False

        area_registry.async_update(thing_id, **cleared)
        return True


class EmptyFloorFixFlow(_RemoveOrIgnoreFixFlow):
    """Handler for an empty floor: remove it, or keep it and stop nagging."""

    _key = "floor"
    _id_key = "empty_floor_id"

    @callback
    def _remove(self, thing_id: str) -> bool:
        """Remove the floor, if it is still empty."""
        registry = fr.async_get(self.hass)
        # The floor may already be gone if removed elsewhere meanwhile.
        if not registry.async_get_floor(thing_id):
            return True

        # Deleting a floor takes it off every area on it.
        if async_floor_in_use(
            self.hass, thing_id, async_collect_mentioned_strings(self.hass)
        ):
            return False

        registry.async_delete(thing_id)
        return True


class UnusedLabelFixFlow(_RemoveOrIgnoreFixFlow):
    """Handler for an unused label: remove it, or keep it and stop nagging."""

    _key = "label"
    _id_key = "unused_label_id"

    @callback
    def _remove(self, thing_id: str) -> bool:
        """Remove the label, if it is still unused."""
        registry = lr.async_get(self.hass)
        # The label may already be gone if removed elsewhere meanwhile.
        if not registry.async_get_label(thing_id):
            return True

        # Deleting a label strips it from everything carrying it.
        if async_label_in_use(
            self.hass, thing_id, async_collect_mentioned_strings(self.hass)
        ):
            return False

        registry.async_delete(thing_id)
        return True


class UnusedBlueprintFixFlow(_RemoveOrIgnoreFixFlow):
    """Handler for an unused blueprint: remove it, or keep it and stop nagging.

    Blueprint removal deletes a file, so it overrides ``async_step_remove``
    with the async removal instead of the synchronous ``_remove`` hook.
    """

    _key = "blueprint"
    _id_key = "unused_blueprint_path"

    def _menu_placeholders(self) -> dict[str, str]:
        """Name the blueprint and its domain in the menu step."""
        data = self.data or {}
        return {
            "blueprint": str(data.get("blueprint", "")),
            "domain": str(data.get("unused_blueprint_domain", "")),
        }

    async def async_step_remove(
        self,
        _: dict[str, str] | None = None,
    ) -> FlowResult:
        """Remove the blueprint file, if it still exists and is not in use."""
        data = self.data or {}
        domain = str(data.get("unused_blueprint_domain", ""))
        path = str(data.get("unused_blueprint_path", ""))
        domain_blueprints: dict[str, blueprint.DomainBlueprints] = self.hass.data.get(
            blueprint.DOMAIN, {}
        )
        if domain_blueprint := domain_blueprints.get(domain):
            # It may already be gone, or have gained a consumer meanwhile.
            with suppress(FileNotFoundError, blueprint.BlueprintInUse):
                await domain_blueprint.async_remove_blueprint(path)
        return self.async_create_entry(data={})


class DuplicateResourceFixFlow(_RemoveOrIgnoreFixFlow):
    """Handler for a dashboard resource listed more than once.

    Clearing the copies is async and needs the resource collection, so this
    overrides ``async_step_remove`` rather than the synchronous ``_remove``
    hook. One copy is kept: the most recently added, which for a card updated
    by adding a resource instead of editing one is the version wanted.
    """

    _key = "resource"
    _id_key = "duplicate_resource_url"

    def _menu_placeholders(self) -> dict[str, str]:
        """Name the resource, how many copies, and which URLs.

        The inherited version supplies only the name, and this dialog's text
        interpolates all three.
        """
        data = self.data or {}
        return {
            "resource": str(data.get("resource", "")),
            "resources": str(data.get("resources", "")),
            "count": str(data.get("count", "")),
        }

    async def async_step_init(
        self,
        _: dict[str, str] | None = None,
    ) -> FlowResult:
        """Offer the usual menu, unless there is nothing here that can fix it.

        Resources listed in YAML are static: nothing in Home Assistant can
        delete one, so offering to would be a button that quietly does
        nothing. Those get told where the file is instead.
        """
        lovelace = self.hass.data.get(lovelace_const.DOMAIN)
        resources = lovelace.resources if lovelace is not None else None

        if resources is not None and is_yaml_managed(resources):
            return self.async_abort(
                reason="yaml",
                description_placeholders=self._menu_placeholders(),
            )

        return await super().async_step_init()

    async def async_step_remove(
        self,
        _: dict[str, str] | None = None,
    ) -> FlowResult:
        """Clear the redundant copies, keeping the most recent."""
        key = str((self.data or {}).get(self._id_key, ""))

        if (lovelace := self.hass.data.get(lovelace_const.DOMAIN)) is None or (
            resources := lovelace.resources
        ) is None:
            return self.async_create_entry(data={})

        # A storage collection hands out nothing until it has been loaded, and
        # this flow cannot assume the inspection that raised the issue is what
        # loaded it. Reading it cold would clear nothing and still say it did.
        await resources.async_get_info()

        # Worked out again after every deletion rather than once up front.
        # Each delete awaits, and somebody editing the same resource in that
        # window can take away the copy this was going to keep. Against a
        # stale list that ends with every copy gone, which is not what anybody
        # asked for.
        tried: set[str] = set()
        while True:
            redundant = [
                item_id
                for item_id in redundant_item_ids(resources.async_items() or [], key)
                if item_id not in tried
            ]
            if not redundant:
                break

            item_id = redundant[0]
            # Remembered whatever happens, so a copy that cannot be deleted
            # cannot spin this loop forever either.
            tried.add(item_id)

            # Somebody clearing it themselves first is a fine ending too.
            with suppress(collection.ItemNotFound):
                await resources.async_delete_item(item_id)

        return self.async_create_entry(data={})


class PersonUnknownDeviceTrackerFixFlow(_RemoveOrIgnoreFixFlow):
    """Handler for a person's unknown device trackers.

    Offers to strip the unknown device trackers from the person, or to keep
    them and ignore the issue. Removal reuses Spook's own
    ``person.remove_device_tracker`` action, which only edits storage-backed
    persons; a YAML person cannot be edited, so that is reported instead.
    """

    _key = "person"
    _id_key = "person_entity_id"

    def _menu_placeholders(self) -> dict[str, str]:
        """Name the person and the offending device trackers in the menu."""
        data = self.data or {}
        return {
            "person": str(data.get("person", "")),
            "device_trackers": str(data.get("device_trackers", "")),
        }

    async def async_step_remove(
        self,
        _: dict[str, str] | None = None,
    ) -> FlowResult:
        """Remove the still-unknown device trackers from the person."""
        data = self.data or {}
        entity_id = str(data.get("person_entity_id", ""))
        candidates = [t for t in str(data.get("unknown_trackers", "")).split(",") if t]

        entity_registry = er.async_get(self.hass)
        unknown = [
            tracker
            for tracker in candidates
            if entity_registry.async_get(tracker) is None
            and self.hass.states.get(tracker) is None
        ]
        if unknown:
            try:
                await self.hass.services.async_call(
                    "person",
                    "remove_device_tracker",
                    {"entity_id": entity_id, "device_tracker": unknown},
                    blocking=True,
                )
            except HomeAssistantError:
                # YAML persons are not editable; nothing Spook can remove.
                return self.async_abort(reason="not_editable")
        return self.async_create_entry(data={})


class GroupUnknownMembersFixFlow(_RemoveOrIgnoreFixFlow):
    """Handler for a group's unknown members.

    Prunes the missing members from a UI-managed group's config entry. Only
    config-entry groups can be edited; a YAML group cannot, so that is
    reported instead.
    """

    _key = "group"
    _id_key = "group_entity_id"

    def _menu_placeholders(self) -> dict[str, str]:
        """Name the group and its missing members in the menu step."""
        data = self.data or {}
        return {
            "group": str(data.get("group", "")),
            "entity_id": str(data.get("group_entity_id", "")),
            "entities": str(data.get("entities", "")),
        }

    async def async_step_remove(
        self,
        _: dict[str, str] | None = None,
    ) -> FlowResult:
        """Remove the members that no longer exist from the group."""
        entity_id = str((self.data or {}).get("group_entity_id", ""))
        entity_registry = er.async_get(self.hass)

        entry_entity = entity_registry.async_get(entity_id)
        if entry_entity is None or entry_entity.config_entry_id is None:
            # A YAML group; its members live in configuration.yaml.
            return self.async_abort(reason="not_editable")

        # If the config entry itself is already gone, there is nothing left
        # to prune; the group's own entity is gone too, so the issue clears
        # on the next inspection. Completing the flow is the honest outcome.
        entry = self.hass.config_entries.async_get_entry(entry_entity.config_entry_id)
        if entry is not None:
            members = list(entry.options.get(CONF_ENTITIES) or [])
            # Only the members the issue named, and of those only the ones
            # still gone. Another member missing for a moment right now, an
            # integration reloading say, is not one somebody agreed to drop.
            # Asked the way the repair asked it: Home Assistant knows more
            # entities than the registry and the state machine hold between
            # them, like the time and date sensors and scenes made on the fly.
            offered = _offered(self.data, "group_unknown_entity_ids")
            dropping = set(
                async_filter_known_entity_ids(
                    self.hass, [member for member in members if member in offered]
                )
            )
            if not dropping:
                return self.async_abort(reason="changed")

            remaining = [member for member in members if member not in dropping]
            if remaining != members:
                self.hass.config_entries.async_update_entry(
                    entry, options={**entry.options, CONF_ENTITIES: remaining}
                )
                # Writing the options is not the end of it. Nothing in the
                # group integration listens for its own entry changing, so
                # the group that is running carries on with the list it was
                # built from until something reloads it. Leave that out and
                # the button reports success, the member is still in the
                # group, and the repair goes on reporting it because it reads
                # the group rather than the stored options.
                await self.hass.config_entries.async_reload(entry.entry_id)
        return self.async_create_entry(data={})


class MinMaxUnknownSourcesFixFlow(_RemoveOrIgnoreFixFlow):
    """Handler for a min/max helper's missing members.

    Prunes the members that no longer exist from the helper's config entry,
    keeping the ones that remain, so the helper carries on working.
    """

    _key = "helper"
    _id_key = "min_max_config_entry_id"

    def _menu_placeholders(self) -> dict[str, str]:
        """Name the helper and its missing members in the menu step."""
        data = self.data or {}
        return {
            "helper": str(data.get("helper", "")),
            "entity_id": async_name_helper_in_the_registry(
                self.hass, str(data.get(self._id_key, ""))
            ),
            "sources": str(data.get("sources", "")),
        }

    async def async_step_remove(
        self,
        _: dict[str, str] | None = None,
    ) -> FlowResult:
        """Remove the members that no longer exist from the helper."""
        entry_id = str((self.data or {}).get("min_max_config_entry_id", ""))
        entry = self.hass.config_entries.async_get_entry(entry_id)
        if entry is not None:
            members = list(entry.options.get(MIN_MAX_ENTITY_IDS) or [])
            # Only the members the issue named, and of those only the ones
            # still gone, asked the way the repair asked it.
            dropping = _offered(
                self.data, "min_max_unknown_sources"
            ) & async_unknown_min_max_members(self.hass, entry)
            if not dropping:
                return self.async_abort(reason="changed")

            remaining = [value for value in members if value not in dropping]
            if remaining != members:
                if len(remaining) < _MIN_MAX_MINIMUM_MEMBERS:
                    # A min/max helper needs at least two members; pruning
                    # would leave too few and break it. Let the user decide.
                    return self.async_abort(reason="too_few_members")
                self.hass.config_entries.async_update_entry(
                    entry, options={**entry.options, MIN_MAX_ENTITY_IDS: remaining}
                )
                # Same as above, and here it is worse than a stale reference:
                # the helper keeps listening to the source somebody just took
                # out, so if that source comes back it drives the value again
                # while the configuration says it is gone.
                await self.hass.config_entries.async_reload(entry.entry_id)
        return self.async_create_entry(data={})


class HelperUnknownSourcesFixFlow(_RemoveOrIgnoreFixFlow):
    """Handler for a helper whose source entities are gone.

    A single-source helper whose source no longer exists is broken; there is
    nothing to prune, so the fix is to remove the whole helper. Warns, up
    front, when the helper is still used by automations or scripts, and is
    honest that those become dangling references, which Spook itself will
    then point out.
    """

    _key = "helper"
    _id_key = "helper_config_entry_id"

    def _menu_placeholders(self) -> dict[str, str]:
        """Name the helper, its missing sources, and where it is used."""
        data = self.data or {}
        return {
            "helper": str(data.get("helper", "")),
            "domain": str(data.get("domain", "")),
            "entity_id": async_name_helper_in_the_registry(
                self.hass, str(data.get(self._id_key, ""))
            ),
            "sources": str(data.get("sources", "")),
            "usage": self._usage_text(str(data.get("helper_config_entry_id", ""))),
        }

    def _usage_text(self, entry_id: str) -> str:
        """Describe which automations and scripts still use the helper."""
        entity_registry = er.async_get(self.hass)
        automations: set[str] = set()
        scripts: set[str] = set()
        for entity in er.async_entries_for_config_entry(entity_registry, entry_id):
            automations.update(
                async_referencing(
                    self.hass, AUTOMATION_DOMAIN, "entities", entity.entity_id
                )
            )
            scripts.update(
                async_referencing(
                    self.hass, SCRIPT_DOMAIN, "entities", entity.entity_id
                )
            )

        if not automations and not scripts:
            return "It is not used by any automation or script."

        parts: list[str] = []
        if automations:
            parts.append(f"{len(automations)} automation{_plural(automations)}")
        if scripts:
            parts.append(f"{len(scripts)} script{_plural(scripts)}")
        return (
            f"It is still used by {' and '.join(parts)}. Removing the helper "
            "leaves them referencing a missing entity, which Spook will then "
            "point out for you."
        )

    async def async_step_remove(
        self,
        _: dict[str, str] | None = None,
    ) -> FlowResult:
        """Remove the whole helper, if it is still as broken as it was shown.

        This deletes the helper outright, so it looks again first, the way
        the repair looked. A source that came back, or a helper pointed at
        something that works since, leaves the helper where it is.
        """
        entry_id = str((self.data or {}).get("helper_config_entry_id", ""))
        if (entry := self.hass.config_entries.async_get_entry(entry_id)) is None:
            return self.async_create_entry(data={})

        # The helper as it was shown, sources and all. One given another
        # source since, even a working one, is not the helper somebody saw.
        offered = _offered(self.data, "helper_unknown_sources")
        configured = _offered(self.data, "helper_configured_sources")
        if (
            not offered
            or configured != async_helper_sources(entry)
            or not offered <= async_unknown_helper_sources(self.hass, entry)
        ):
            return self.async_abort(reason="changed")

        await self.hass.config_entries.async_remove(entry_id)
        return self.async_create_entry(data={})


class OrphanedStatisticsFixFlow(_RemoveOrIgnoreFixFlow):
    """Handler for long-term statistics with no entity left behind them.

    Clearing them is a websocket command the Statistics page calls and no
    action anybody can reach, so without this the only way to act on the
    report is to open that page and work through it by hand, which on a list
    of a couple of hundred is not really an offer at all. #1613.
    """

    _key = "statistics"
    _id_key = "orphaned_statistic_ids"

    def _menu_placeholders(self) -> dict[str, str]:
        """List what the report named, the way the report listed it."""
        return {"statistics": _bulleted(self._offered())}

    def _offered(self) -> list[str]:
        """Return the statistic IDs the report put in front of somebody."""
        written = str((self.data or {}).get(self._id_key, ""))
        return [statistic_id for statistic_id in written.split(",") if statistic_id]

    async def async_step_remove(
        self,
        _: dict[str, str] | None = None,
    ) -> FlowResult:
        """Clear the statistics, after looking again to see if they still go.

        Looked up again rather than taken from the issue, and then kept in
        common with it. An issue sits there until somebody opens it, which
        may be days, and this deletes history: nothing goes that was not on
        the list they read, and nothing goes that has come back since.

        Asked the same way the report asked it, settling time and all. A
        glance would say yes to a sensor that came back long ago and happens
        to be between two brief windows right now, which is the case the
        wait exists for.
        """
        offered = set(self._offered())
        if not offered:
            return self.async_abort(reason="nothing_to_clear")

        still_gone = await async_settled_orphaned_statistic_ids(self.hass)
        clearing = sorted(offered & still_gone)
        if not clearing:
            # Nothing on the list still needs clearing. They may have an
            # entity behind them again, or somebody may have cleared them by
            # hand while the issue sat there. Either way there is nothing to
            # do, and which of the two it was is not worth guessing at.
            return self.async_abort(reason="nothing_to_clear")

        # Queued rather than done: the recorder takes the work on its own
        # thread and says when it has landed. Answering before that would
        # close the issue on the strength of having asked, and a recorder
        # that is wedged would look like a job well done. Home Assistant's
        # own Statistics page waits on exactly this, for exactly this long.
        cleared = asyncio.Event()

        def _done() -> None:
            """Say so from the recorder's thread."""
            self.hass.loop.call_soon_threadsafe(cleared.set)

        get_instance(self.hass).async_clear_statistics(clearing, on_done=_done)

        try:
            async with asyncio.timeout(_CLEARING_TAKES_AT_MOST):
                await cleared.wait()
        except TimeoutError:
            # The work is still queued and will most likely land. Saying it
            # is done would be a guess, and leaving the issue up costs
            # nothing: the next round clears it if the statistics went, and
            # reports them again if they did not.
            LOGGER.debug(
                "Spook asked for %s to be cleared and the recorder has not "
                "said it is done",
                ", ".join(clearing),
            )
            return self.async_abort(reason="clearing_took_too_long")

        LOGGER.debug("Spook cleared orphaned statistics: %s", ", ".join(clearing))

        return self.async_create_entry(data={})


class EnergyUnknownReferencesFixFlow(_RemoveOrIgnoreFixFlow):
    """Handler for entities the energy settings name that are gone.

    Takes them out of the energy settings, each in the way its setting allows:
    a meter takes its source along, a price or a cost is only cleared. What
    goes is what the issue showed and is still unknown when the button is
    pressed, so one that came back meanwhile stays.
    """

    _key = "entities"
    _id_key = "energy_unknown_entity_ids"

    def _menu_placeholders(self) -> dict[str, str]:
        """List the entities the way the issue listed them."""
        return {"entities": str((self.data or {}).get("entities", ""))}

    async def async_step_remove(
        self,
        _: dict[str, str] | None = None,
    ) -> FlowResult:
        """Take the entities out of the energy settings, after looking again."""
        offered = _offered(self.data, self._id_key)
        removing = offered & await async_unknown_energy_entities(self.hass)
        if not removing:
            return self.async_abort(reason="nothing_to_remove")

        manager = await async_get_energy_manager(self.hass)
        preferences = manager.data or {}
        cleaned = energy_preferences_without(preferences, removing)
        if all(value == preferences.get(key) for key, value in cleaned.items()):
            return self.async_abort(reason="nothing_to_remove")

        await manager.async_update(cleaned)  # type: ignore[arg-type]
        LOGGER.debug(
            "Spook took unknown entities out of the energy settings: %s",
            ", ".join(sorted(removing)),
        )

        return self.async_create_entry(data={})


# Remove-or-ignore fix flows, keyed by the data field that identifies their
# leftover registry thing.
_REMOVE_OR_IGNORE_FLOWS: dict[str, type[_RemoveOrIgnoreFixFlow]] = {
    "empty_area_id": EmptyAreaFixFlow,
    "area_sensors_area_id": AreaUnknownSensorsFixFlow,
    "empty_floor_id": EmptyFloorFixFlow,
    "unused_label_id": UnusedLabelFixFlow,
    "unused_blueprint_path": UnusedBlueprintFixFlow,
    "duplicate_resource_url": DuplicateResourceFixFlow,
    "person_entity_id": PersonUnknownDeviceTrackerFixFlow,
    "group_entity_id": GroupUnknownMembersFixFlow,
    "min_max_config_entry_id": MinMaxUnknownSourcesFixFlow,
    "helper_config_entry_id": HelperUnknownSourcesFixFlow,
    "orphaned_statistic_ids": OrphanedStatisticsFixFlow,
    "energy_unknown_entity_ids": EnergyUnknownReferencesFixFlow,
}


async def async_create_fix_flow(
    _hass: HomeAssistant,
    issue_id: str,
    data: dict[str, str | int | float | None] | None,
) -> RepairsFlow:
    """Create flow."""
    if issue_id == "restart_required":
        return RestartRequiredFixFlow()
    if data:
        for key, flow in _REMOVE_OR_IGNORE_FLOWS.items():
            if data.get(key):
                return flow()
    return ConfirmRepairFlow()