# SPDX-License-Identifier: BUSL-1.1
# Copyright (c) 2024-2026 Ali Sadeghi Aghili
"""Where the setup wizard writes, and how a new run is merged into an old one.

``setup_project.py`` produces *drafts*: aliases, rules and examples suggested
by a model, to be reviewed before they replace anything curated. This module
holds the parts of that rule which need no database and no model, so they can
be tested on their own:

* which directories the wizard refuses to write into without
  ``--overwrite-live`` (:func:`protected_directory`, :func:`refusal_message`),
  and the ``<name>.bak-<stamp>`` copies it makes first (:func:`back_up`);
* how the files of a second run are merged into those of the first, by key,
  without shrinking them (:func:`merge_mapping`, :func:`merge_synonyms`,
  :func:`merge_examples`, :func:`merge_relationships`), with the counts the
  summary prints (:class:`MergeStats`);
* how two earlier entries that the new keys would put on one key become one
  (:func:`collapse_same_key`), so a merge never leaves two entries for one
  table, and how a model's alias list is made usable (:func:`tidy_aliases`);
* how the entries that a run of 6.13 to 6.16.2 wrote into ``ring_aliases`` --
  one per *table*, where a key must be the value of a ring -- are dropped from
  an earlier ``aliases.yaml``, their aliases kept in ``entities.yaml``
  (:func:`drop_table_rings`, :class:`RingCleanup`).

Merging never removes an entry, except that two old entries for one table
become one and that the table-keyed ``ring_aliases`` entries above are dropped
(and counted). An entry the new run produced replaces the old one with the same
key; every other old entry stays as it was.

Public API
----------
DEFAULT_DRAFT_DIR
    Directory name the wizard writes to by default.
protected_directory(output_dir, live_dir, repo_root) -> str | None
refusal_message(output_dir, label) -> str
back_up(paths, stamp) -> list[Path]
MergeStats
merge_mapping(existing, new, rename) -> (dict, MergeStats)
merge_synonyms(existing, new) -> (dict, MergeStats)
merge_examples(existing, new) -> (list, MergeStats)
merge_relationships(existing, new, rename) -> (list, MergeStats)
collapse_same_key(existing, rename, combine) -> (dict, tuple[(dropped, kept), ...])
tidy_aliases(aliases, limit) -> list[str]
RingCleanup, drop_table_rings(ring_aliases, entities, view) -> RingCleanup
"""

from __future__ import annotations

import re
import shutil
from collections.abc import Callable, Iterable, Mapping, Sequence
from dataclasses import dataclass
from pathlib import Path
from typing import Any

from core import config_doctor as doctor
from core import ring_names
from core.persian import normalize_for_matching

__all__ = [
    "ALIAS_LIMIT",
    "DEFAULT_DRAFT_DIR",
    "EXAMPLE_DIR_NAME",
    "MergeStats",
    "RingCleanup",
    "back_up",
    "collapse_same_key",
    "drop_table_rings",
    "hold_back_empty",
    "merge_examples",
    "merge_mapping",
    "merge_relationships",
    "merge_synonyms",
    "protected_directory",
    "refusal_message",
    "tidy_aliases",
]

#: Directory the wizard writes to unless ``--output`` says otherwise. It is
#: git-ignored, like the drafts of ``database/schema_inspector_cli.py``.
DEFAULT_DRAFT_DIR = "project_config_draft"

#: The committed template directory; never a place for drafts built from a
#: real warehouse.
EXAMPLE_DIR_NAME = "project_config.example"

#: Aliases kept per table. A model asked for aliases of one table sometimes
#: answers with every concatenation of the same three words (22 for one table
#: in a real run); a handful of distinct phrasings is what retrieval can use.
ALIAS_LIMIT = 8


# ---------------------------------------------------------------------------
# Protected directories and backups
# ---------------------------------------------------------------------------

def protected_directory(output_dir: Path, live_dir: Path, repo_root: Path) -> str | None:
    """Name the protected directory *output_dir* is, or ``None`` when it is not one.

    The live configuration directory and ``project_config.example/`` are
    protected. Paths are compared after ``resolve()``, so a relative spelling,
    a trailing slash or a symbolic link does not get around the check.

    Args:
        output_dir: Where the wizard was asked to write.
        live_dir: The directory the application reads (``PROJECT_CONFIG_DIR``,
            resolved against *repo_root* when relative).
        repo_root: The repository root, where ``project_config.example/`` lives.

    Returns:
        ``"live"`` for the live directory, ``"example"`` for the template
        directory, else ``None``.

    Raises:
        Nothing for ordinary paths; ``resolve()`` is not strict.

    Examples:
        >>> root = Path("/srv/app")
        >>> protected_directory(Path("/srv/app/project_config"), root / "project_config", root)
        'live'
        >>> protected_directory(Path("/srv/app/project_config.example/"), root / "project_config", root)
        'example'
        >>> protected_directory(Path("/srv/app/project_config_draft"), root / "project_config", root) is None
        True
        >>> protected_directory(Path("/srv/app/x/../project_config"), root / "project_config", root)
        'live'
    """
    resolved = output_dir.resolve()
    if resolved == live_dir.resolve():
        return "live"
    if resolved in {(repo_root / EXAMPLE_DIR_NAME).resolve(), Path(EXAMPLE_DIR_NAME).resolve()}:
        return "example"
    return None


def refusal_message(output_dir: Path, label: str) -> str:
    """The one-paragraph message printed when the wizard refuses to write.

    Args:
        output_dir: The directory that was refused.
        label: What :func:`protected_directory` returned for it.

    Returns:
        The message, ending with how to proceed.

    Raises:
        Nothing. This function never raises.

    Examples:
        >>> text = refusal_message(Path("project_config"), "live")
        >>> "--overwrite-live" in text and "project_config_draft" in text
        True
        >>> "\\n" in refusal_message(Path("x"), "example")
        False
    """
    what = (
        "the live configuration directory the application reads"
        if label == "live"
        else f"the committed template directory ({EXAMPLE_DIR_NAME}/), which git tracks"
    )
    return (
        f"Refusing to write into {output_dir}: it is {what}. The wizard writes drafts of "
        "aliases, rules and examples suggested by a model, and they must be reviewed before "
        "they replace curated files. Run it with the default output "
        f"({DEFAULT_DRAFT_DIR}/) or another scratch directory (--output), review the files "
        "and copy the ones you accept across yourself. To write here anyway, add "
        "--overwrite-live: every file the wizard replaces is first copied to "
        "<name>.bak-<YYYYmmdd-HHMMSS> next to it, and the copies are listed."
    )


def _backup_path(path: Path, stamp: str) -> Path:
    """``<name>.bak-<stamp>`` next to *path*, made unique if that name is taken."""
    candidate = path.with_name(f"{path.name}.bak-{stamp}")
    n = 1
    while candidate.exists():
        n += 1
        candidate = path.with_name(f"{path.name}.bak-{stamp}-{n}")
    return candidate


def back_up(paths: Iterable[Path], stamp: str) -> list[Path]:
    """Copy each existing file in *paths* to ``<name>.bak-<stamp>`` beside it.

    Args:
        paths: Files about to be replaced. One that does not exist is skipped.
        stamp: The timestamp part of the name, ``YYYYmmdd-HHMMSS``.

    Returns:
        The backup files created, in the order of *paths*.

    Raises:
        OSError: When a copy cannot be made; nothing is replaced then.

    Examples:
        >>> import tempfile
        >>> with tempfile.TemporaryDirectory() as tmp:
        ...     f = Path(tmp, "a.yaml"); _ = f.write_text("x")
        ...     made = back_up([f, Path(tmp, "missing.yaml")], "20260101-000000")
        ...     [p.name for p in made], made[0].read_text()
        (['a.yaml.bak-20260101-000000'], 'x')
    """
    made: list[Path] = []
    for path in paths:
        if not path.is_file():
            continue
        target = _backup_path(path, stamp)
        shutil.copy2(path, target)
        made.append(target)
    return made


# ---------------------------------------------------------------------------
# Merging
# ---------------------------------------------------------------------------

@dataclass(frozen=True)
class MergeStats:
    """What a merge did to one file.

    Attributes:
        kept: Entries of the old file the new run did not touch.
        added: Entries the old file did not have.
        replaced: Entries the new run produced for a key the old file had.
        rekeyed: ``(old key, new key)`` for old entries whose key changed
            because a bare table name is now shared by several schemas. They
            count under *kept* or *replaced* as well.
        held: Keys of old entries kept (they count under *kept*) because this
            run had only an empty answer for them; see :func:`hold_back_empty`.
    """

    kept: int = 0
    added: int = 0
    replaced: int = 0
    rekeyed: tuple[tuple[str, str], ...] = ()
    held: tuple[str, ...] = ()

    def __add__(self, other: "MergeStats") -> "MergeStats":
        """Counts added up, for a file that holds two merged sections.

        Examples:
            >>> (MergeStats(1, 2, 3) + MergeStats(4, 5, 6)).describe()
            'kept 5, added 7, replaced 9'
        """
        return MergeStats(
            self.kept + other.kept,
            self.added + other.added,
            self.replaced + other.replaced,
            self.rekeyed + other.rekeyed,
            self.held + other.held,
        )

    def describe(self) -> str:
        """``"kept 2, added 1, replaced 0"``, plus the re-keyed count when there is one.

        Examples:
            >>> MergeStats(2, 1, 0).describe()
            'kept 2, added 1, replaced 0'
            >>> MergeStats(0, 1, 0, (("Trade", "sales.Trade"),)).describe()
            'kept 0, added 1, replaced 0, re-keyed 1'
        """
        text = f"kept {self.kept}, added {self.added}, replaced {self.replaced}"
        if self.rekeyed:
            text += f", re-keyed {len(self.rekeyed)}"
        return text


def _fold(key: str) -> str:
    return key.casefold()


def hold_back_empty(
    existing: Mapping[str, Any],
    new: Mapping[str, Any],
    is_empty: Callable[[Any], bool],
    rename: Mapping[str, str] | None = None,
) -> tuple[dict[str, Any], tuple[str, ...]]:
    """Drop the new entries that say nothing where the old file already says something.

    A model that failed, or answered with nothing, for one table must not
    erase what an earlier run or a person wrote for it. An entry of *new* is
    dropped when *is_empty* is true for it and false for the old entry with
    the same key (case-insensitive, after *rename*); the merge then keeps the
    old one and counts it as kept.

    Args:
        existing: The old file's mapping.
        new: This run's entries.
        is_empty: Whether an entry carries no answer.
        rename: The renames the merge will apply to *existing*.

    Returns:
        ``(entries to merge, keys held back)``.

    Raises:
        Nothing. This function never raises.

    Examples:
        >>> kept, held = hold_back_empty({"A": ["x"], "B": []}, {"A": [], "B": [], "C": []}, lambda v: not v)
        >>> kept, held
        ({'B': [], 'C': []}, ('A',))
    """
    lowered = {_fold(k): v for k, v in (rename or {}).items()}
    old = {_fold(lowered.get(_fold(k), k)): v for k, v in existing.items()}
    kept: dict[str, Any] = {}
    held: list[str] = []
    for key, value in new.items():
        before = old.get(_fold(key))
        if before is not None and is_empty(value) and not is_empty(before):
            held.append(key)
        else:
            kept[key] = value
    return kept, tuple(held)


def merge_mapping(
    existing: Mapping[str, Any],
    new: Mapping[str, Any],
    rename: Mapping[str, str] | None = None,
) -> tuple[dict[str, Any], MergeStats]:
    """Merge *new* into *existing* by key; keys compare case-insensitively.

    Old entries come first, in their order, under their (possibly renamed)
    key; an entry of *new* with a key already present replaces that value in
    place; the others are appended. Nothing is removed.

    Args:
        existing: The old file's mapping.
        new: The entries this run produced.
        rename: ``{old key: new key}`` for old entries to re-key before the
            merge (a bare name that must now be qualified).

    Returns:
        ``(merged, stats)``.

    Raises:
        Nothing. This function never raises.

    Examples:
        >>> merged, stats = merge_mapping({"A": 1, "B": 2}, {"B": 3, "C": 4})
        >>> merged, stats.describe()
        ({'A': 1, 'B': 3, 'C': 4}, 'kept 1, added 1, replaced 1')
        >>> merged, stats = merge_mapping({"Trade": 1}, {"ref.Trade": 2}, {"Trade": "sales.Trade"})
        >>> merged, stats.rekeyed
        ({'sales.Trade': 1, 'ref.Trade': 2}, (('Trade', 'sales.Trade'),))
        >>> merge_mapping({"a": 1}, {"A": 2})[0]
        {'A': 2}
    """
    lowered_rename = {_fold(k): v for k, v in (rename or {}).items()}
    merged: dict[str, Any] = {}
    index: dict[str, str] = {}
    rekeyed: list[tuple[str, str]] = []
    for key, value in existing.items():
        target = lowered_rename.get(_fold(key), key)
        if target != key:
            rekeyed.append((key, target))
        index[_fold(target)] = target
        merged[target] = value
    untouched = set(merged)
    added = replaced = 0
    for key, value in new.items():
        present = index.get(_fold(key))
        if present is None:
            index[_fold(key)] = key
            merged[key] = value
            added += 1
            continue
        if present == key:
            merged[key] = value
        else:
            # Same table, spelled the way this run spells it; keep its place.
            merged = {(key if k == present else k): (value if k == present else v) for k, v in merged.items()}
            index[_fold(key)] = key
        untouched.discard(present)
        replaced += 1
    return merged, MergeStats(len(untouched), added, replaced, tuple(rekeyed))


def merge_synonyms(
    existing: Mapping[str, Sequence[str]],
    new: Mapping[str, Sequence[str]],
) -> tuple[dict[str, list[str]], MergeStats]:
    """Merge the ``synonyms`` of two runs: a word that is in both keeps both token lists.

    Synonyms are keyed by a word, not by a table, so there is nothing for a
    run to replace: the tokens of the old and the new entry are united (old
    first). Such an entry counts as *replaced* when it gained a token.

    Args:
        existing: The old file's ``synonyms``.
        new: This run's ``synonyms``.

    Returns:
        ``(merged, stats)``.

    Raises:
        Nothing. This function never raises.

    Examples:
        >>> merged, stats = merge_synonyms({"a": ["x"], "b": ["y"]}, {"a": ["x", "z"], "c": ["w"]})
        >>> merged, stats.describe()
        ({'a': ['x', 'z'], 'b': ['y'], 'c': ['w']}, 'kept 1, added 1, replaced 1')
    """
    merged: dict[str, list[str]] = {k: list(v) for k, v in existing.items()}
    index = {_fold(k): k for k in merged}
    untouched = set(merged)
    added = replaced = 0
    for word, tokens in new.items():
        present = index.get(_fold(word))
        if present is None:
            merged[word] = list(tokens)
            index[_fold(word)] = word
            added += 1
            continue
        gained = [t for t in tokens if t not in merged[present]]
        if gained:
            merged[present].extend(gained)
            untouched.discard(present)
            replaced += 1
    return merged, MergeStats(len(untouched), added, replaced)


def _question_key(entry: Mapping[str, Any]) -> str:
    return " ".join(str(entry.get("question", "")).split()).casefold()


def merge_examples(
    existing: Sequence[Mapping[str, Any]],
    new: Sequence[Mapping[str, Any]],
) -> tuple[list[dict[str, Any]], MergeStats]:
    """Merge example lists, de-duplicated by question text.

    Questions compare case-insensitively with runs of whitespace collapsed.
    A new example whose question the old list has replaces that example in
    place; the others are appended. An example without a question or an SQL
    text is not an example and is ignored on the new side. A question
    repeated within *new* counts once (the first).

    Args:
        existing: The old file's ``examples``.
        new: This run's examples.

    Returns:
        ``(merged, stats)``.

    Raises:
        Nothing. This function never raises.

    Examples:
        >>> old = [{"question": "How many?", "sql": "S1"}, {"question": "Top 5", "sql": "S2"}]
        >>> fresh = [{"question": "how  many?", "sql": "S3"}, {"question": "New", "sql": "S4"}]
        >>> merged, stats = merge_examples(old, fresh)
        >>> [e["sql"] for e in merged], stats.describe()
        (['S3', 'S2', 'S4'], 'kept 1, added 1, replaced 1')
    """
    merged: list[dict[str, Any]] = [dict(e) for e in existing]
    index: dict[str, int] = {}
    for position, entry in enumerate(merged):
        index.setdefault(_question_key(entry), position)
    untouched = set(range(len(merged)))
    seen: set[str] = set()
    added = replaced = 0
    for entry in new:
        if not (entry.get("question") and entry.get("sql")):
            continue
        key = _question_key(entry)
        if key in seen:
            continue
        seen.add(key)
        if key in index:
            merged[index[key]] = dict(entry)
            untouched.discard(index[key])
            replaced += 1
        else:
            index[key] = len(merged)
            merged.append(dict(entry))
            added += 1
    return merged, MergeStats(len(untouched), added, replaced)


def _endpoints(entry: Mapping[str, Any], rename: Mapping[str, str]) -> frozenset[tuple[str, str]] | None:
    """The two ends of a relationship entry as ``(table, column)``, lower-cased; ``None`` if incomplete."""
    ends: list[tuple[str, str]] = []
    for side in ("from", "to"):
        table = str(entry.get(f"{side}_table") or "").strip()
        column = str(entry.get(f"{side}_column") or "").strip()
        if not (table and column):
            return None
        ends.append((rename.get(_fold(table), table).casefold(), column.casefold()))
    return frozenset(ends)


def merge_relationships(
    existing: Sequence[Mapping[str, Any]],
    new: Sequence[Mapping[str, Any]],
    rename: Mapping[str, str] | None = None,
) -> tuple[list[dict[str, Any]], MergeStats]:
    """Merge relationship lists, de-duplicated by their endpoints.

    Two entries are the same relationship when they join the same two
    ``(table, column)`` ends, in either direction. A new entry replaces the
    old one in place; the others are appended. An entry marked as inferred
    (it has a ``basis``) never replaces one that is not: a person has
    reviewed that one, or the database declares it.

    Args:
        existing: The old file's ``relationships``.
        new: This run's relationships.
        rename: ``{old table key: new key}`` applied to the old entries'
            ``from_table`` and ``to_table`` first.

    Returns:
        ``(merged, stats)``.

    Raises:
        Nothing. This function never raises.

    Examples:
        >>> old = [{"from_table": "A", "from_column": "b_id", "to_table": "B", "to_column": "id", "join_hint": "J1"}]
        >>> fresh = [{"from_table": "B", "from_column": "ID", "to_table": "a", "to_column": "B_ID", "join_hint": "J2"},
        ...          {"from_table": "A", "from_column": "c_id", "to_table": "C", "to_column": "id", "join_hint": "J3"}]
        >>> merged, stats = merge_relationships(old, fresh)
        >>> [e["join_hint"] for e in merged], stats.describe()
        (['J2', 'J3'], 'kept 0, added 1, replaced 1')
        >>> guess = [dict(fresh[0], join_hint="J4", basis="naming convention")]
        >>> merge_relationships(old, guess)[0][0]["join_hint"]
        'J1'
    """
    lowered = {_fold(k): v for k, v in (rename or {}).items()}
    merged: list[dict[str, Any]] = []
    rekeyed: list[tuple[str, str]] = []
    for entry in existing:
        item = dict(entry)
        for side in ("from_table", "to_table"):
            old = str(item.get(side) or "")
            target = lowered.get(_fold(old))
            if target and target != old:
                item[side] = target
                if (old, target) not in rekeyed:
                    rekeyed.append((old, target))
        merged.append(item)
    index: dict[frozenset[tuple[str, str]], int] = {}
    for position, item in enumerate(merged):
        ends = _endpoints(item, {})
        if ends is not None:
            index.setdefault(ends, position)
    untouched = set(range(len(merged)))
    added = replaced = 0
    for entry in new:
        ends = _endpoints(entry, {})
        position = index.get(ends) if ends is not None else None
        if position is None:
            if ends is not None:
                index[ends] = len(merged)
            merged.append(dict(entry))
            added += 1
        elif entry.get("basis") and not merged[position].get("basis"):
            continue
        else:
            merged[position] = dict(entry)
            untouched.discard(position)
            replaced += 1
    return merged, MergeStats(len(untouched), added, replaced, tuple(rekeyed))


def collapse_same_key(
    existing: Mapping[str, Any],
    rename: Mapping[str, str],
    combine: Callable[[Any, Any], Any],
) -> tuple[dict[str, Any], tuple[tuple[str, str], ...]]:
    """Make one entry of the old entries that *rename* would put on the same key.

    :func:`merge_mapping` re-keys old entries and, when two end up on one key,
    keeps whichever it saw last. Two earlier entries for one table (a bare one
    from the inspector's draft and a qualified one from an earlier wizard run)
    are exactly that case, and neither should be lost: they are combined here,
    before the merge. Keys compare case-insensitively.

    The entry whose key already is the target keeps its key, and so does the
    first of the group when none is; the others are dropped, and their data is
    folded into it with *combine*, in file order.

    Parameters
    ----------
    existing:
        The old file's mapping.
    rename:
        ``{old key: new key}`` the merge will apply.
    combine:
        ``combine(kept, other)`` returns the entry that replaces *kept* once
        *other* is folded into it.

    Returns
    -------
    tuple
        ``(entries, dropped)``: the mapping without the duplicates, and
        ``(dropped key, kept key)`` for each entry folded away.

    Raises
    ------
    Nothing. This function never raises.

    Examples
    --------
    >>> old = {"Trade": ["a"], "sales.Trade": ["b"], "Customer": ["c"]}
    >>> merged, dropped = collapse_same_key(old, {"sales.Trade": "Trade"}, lambda keep, other: keep + other)
    >>> merged, dropped
    ({'Trade': ['a', 'b'], 'Customer': ['c']}, (('sales.Trade', 'Trade'),))
    >>> collapse_same_key({"A": 1, "B": 2}, {}, lambda keep, other: keep)[1]
    ()
    """
    lowered = {_fold(k): v for k, v in rename.items()}

    def final(key: str) -> str:
        return _fold(lowered.get(_fold(key), key))

    groups: dict[str, list[str]] = {}
    for key in existing:
        groups.setdefault(final(key), []).append(key)
    result: dict[str, Any] = {}
    dropped: list[tuple[str, str]] = []
    for key in existing:
        group = groups[final(key)]
        if len(group) == 1:
            result[key] = existing[key]
        elif key == group[0]:
            keeper = next((k for k in group if _fold(k) == final(key)), key)
            value = existing[keeper]
            for other in group:
                if other != keeper:
                    value = combine(value, existing[other])
                    dropped.append((other, keeper))
            result[keeper] = value
    return result, tuple(dropped)


def _alias_key(text: str) -> str:
    """The words of an alias, folded and without punctuation: what makes two aliases the same."""
    return " ".join(re.findall(r"\w+", normalize_for_matching(text)))


def tidy_aliases(aliases: Iterable[Any] | None, limit: int = ALIAS_LIMIT) -> list[str]:
    """A model's alias list made usable: strings, no repeats, at most *limit*.

    Aliases are compared after folding (case, Persian and Arabic letter and
    digit forms, punctuation, spacing), and the first spelling of each is
    kept. When more than *limit* remain, an alias whose words are all in a
    shorter alias of the list, or that has the same words in another order,
    is dropped (``"bank name list"`` beside ``"bank name"``), and if that is
    not enough the shortest are kept. The order of what is left is the
    model's.

    Parameters
    ----------
    aliases:
        What the model returned; anything that is not text is ignored.
    limit:
        The most aliases to keep.

    Returns
    -------
    list[str]
        The aliases to write.

    Raises
    ------
    Nothing. This function never raises.

    Examples
    --------
    >>> tidy_aliases(["Bank", "bank", " Bank  ", "banks", None, 3, ""])
    ['Bank', 'banks', '3']
    >>> many = ["bank name", "bank name list", "list bank name", "name bank", "bank", "bank list",
    ...         "list of bank", "banks", "bank names", "bank name code", "the bank"]
    >>> tidy_aliases(many, 4)
    ['bank', 'banks']
    >>> tidy_aliases(["alpha beta", "gamma", "delta epsilon zeta", "eta theta"], 2)
    ['alpha beta', 'gamma']
    >>> tidy_aliases(["a b", "a b c"], 8)
    ['a b', 'a b c']
    """
    seen: dict[str, str] = {}
    for raw in aliases or []:
        if isinstance(raw, bool) or not isinstance(raw, (str, int, float)):
            continue
        text = " ".join(str(raw).split())
        key = _alias_key(text)
        if key and key not in seen:
            seen[key] = text
    items = list(seen.items())
    if len(items) <= limit:
        return [text for _, text in items]
    words = [frozenset(key.split()) for key, _ in items]
    kept = [
        i for i in range(len(items))
        if not any(
            j != i and words[j] <= words[i] and (len(words[j]) < len(words[i]) or j < i)
            for j in range(len(items))
        )
    ]
    if len(kept) > limit:
        kept = sorted(sorted(kept, key=lambda i: (len(words[i]), i))[:limit])
    return [items[i][1] for i in kept]


# ---------------------------------------------------------------------------
# ring_aliases entries that are tables
# ---------------------------------------------------------------------------

@dataclass(frozen=True)
class RingCleanup:
    """What :func:`drop_table_rings` did to an earlier ``aliases.yaml``.

    Attributes:
        ring_aliases: The entries that stay: every one whose key is not a table.
        entities: The entities, with the aliases of the dropped entries added.
        dropped: The keys dropped (table names, not data), in file order.
        carried: Entities that gained aliases from a dropped entry.
        created: Entities made for a table that had none.
        unplaced: Dropped keys whose aliases went nowhere: the key names no
            single table, so there was no entity to put them in.
    """

    ring_aliases: dict[str, list[str]]
    entities: dict[str, dict[str, Any]]
    dropped: tuple[str, ...] = ()
    carried: int = 0
    created: int = 0
    unplaced: tuple[str, ...] = ()


def _union_aliases(first: Iterable[str], second: Iterable[str]) -> list[str]:
    """*first* then what *second* adds, without repeats (case ignored)."""
    seen: dict[str, str] = {}
    for item in (*first, *second):
        seen.setdefault(str(item).casefold(), str(item))
    return list(seen.values())


def _table_key(view: doctor.SchemaView, key: str) -> str:
    """The key of the one table of *view* that *key* names, or ``""``."""
    found = doctor.resolve_table_keys(view, key)
    if len(found) == 1:
        return found[0]
    if not found:
        spelled = doctor.entity_key_findings(view, (), [key]).unknown_rings
        return next((right for _, right in spelled if right), "")
    return ""


def drop_table_rings(
    ring_aliases: Mapping[str, list[str]],
    entities: Mapping[str, Mapping[str, Any]],
    view: doctor.SchemaView,
) -> RingCleanup:
    """Drop the ``ring_aliases`` entries that name a table; keep their aliases as the table's entity aliases.

    A ``ring_aliases`` key is the exact value of a ring's name column. A run of
    the wizard of 6.13 to 6.16.2 wrote one entry per table instead, keyed by
    the table and holding the table's aliases, which are also in
    ``entities.yaml``. An entry whose key names a table of *view* by its key,
    by a bare name that finds one key, or by a ``Schema.Table`` spelling
    (:func:`core.ring_names.ring_keys_naming_tables`) is dropped. Every other
    entry -- a real ring, whatever words its name shares with a table -- is
    returned as it was.

    Nothing is lost: the aliases of a dropped entry are added to the entity of
    that table (the entity has them already when the entry was written by the
    wizard), and an entity is made for a table that has none.

    Parameters
    ----------
    ring_aliases:
        ``aliases.yaml`` ``ring_aliases`` of the earlier run.
    entities:
        ``entities.yaml`` of the earlier run, as the wizard reads it
        (``{key: {"table", "schema", "aliases", ...}}``).
    view:
        The tables the wizard knows: this run's, the earlier entities' and
        ``schema.yaml``'s (:func:`core.ring_names.table_view`).

    Returns
    -------
    RingCleanup
        With *entities* itself when no entry was dropped.

    Examples
    --------
    >>> view = ring_names.table_view([("Hall", "ref", "Hall"), ("sales.Trade", "sales", "Trade")])
    >>> rings = {"North Hall": ["north"], "Hall": ["hall"], "Trade": ["deal"], "Region A": ["a"]}
    >>> entities = {"Hall": {"table": "Hall", "schema": "ref", "aliases": ["hall"]}}
    >>> cleaned = drop_table_rings(rings, entities, view)
    >>> list(cleaned.ring_aliases), cleaned.dropped
    (['North Hall', 'Region A'], ('Hall', 'Trade'))
    >>> cleaned.carried, cleaned.created, cleaned.entities["sales.Trade"]["aliases"]
    (0, 1, ['deal'])
    >>> drop_table_rings({"Region A": ["a"]}, entities, view).entities is entities
    True
    """
    doomed = ring_names.ring_keys_naming_tables(view, ring_aliases)
    if not doomed:
        return RingCleanup(dict(ring_aliases), entities)  # type: ignore[arg-type]
    gone = set(doomed)
    grown = {key: dict(entry) for key, entry in entities.items()}
    named = {key.casefold(): key for key in grown}
    by_key = {table.key: table for table in view.tables}
    carried = created = 0
    unplaced: list[str] = []
    for key in doomed:
        aliases = list(ring_aliases[key])
        table_key = _table_key(view, key)
        target = named.get(key.casefold()) or (named.get(table_key.casefold()) if table_key else None)
        if target is not None:
            before = list(grown[target].get("aliases") or [])
            after = _union_aliases(before, aliases)
            if len(after) > len(before):
                grown[target]["aliases"] = after
                carried += 1
        elif table_key:
            table = by_key[table_key]
            grown[table_key] = {"table": table.name, "schema": table.schema or None, "aliases": aliases}
            named[table_key.casefold()] = table_key
            created += 1
        else:
            unplaced.append(key)
    return RingCleanup(
        {key: list(value) for key, value in ring_aliases.items() if key not in gone},
        grown, tuple(doomed), carried, created, tuple(unplaced),
    )