Ctrl K
ComunidadGuías…

core/policy_drafts.py

Rama main · réplica de lectura del repositorio original.

Archivo original ↗Preguntar a KiBot
# SPDX-License-Identifier: BUSL-1.1
# Copyright (c) 2024-2026 Ali Sadeghi Aghili
"""Draft the real values of ``memory_policy.yaml`` and ``session_policy.yaml``.

``scripts/draft_policies.py`` reads the distinct values of one small dimension
column and this module turns them into the two drafts, without touching a
database:

* :func:`usable_values` cuts the values read down to what may be an option: the
  rules of the setup wizard (:mod:`core.value_shapes`: no date, number, code,
  phone number, address, masked value or error text, nothing over 40
  characters), nothing longer than the memory key allows, nothing equal to the
  session policy's ``default_label``, one spelling of each value after the
  project's Persian and Arabic folding, sorted;
* :func:`choose_memory_keys` says which memory keys the dimension belongs to;
* :func:`draft_policies` edits the *text* of the two files, line by line, and
  replaces only ``options`` (and ``filter_key`` / ``column`` when the dimension
  changes). Every comment, every other field, the quoting and the order stay as
  they were, and the result is proved by loading it with the application's own
  validators and comparing it with the input plus exactly those fields.

The invariant the template's comments describe is kept and checked: the memory
options never contain the session policy's ``default_label``, so a pinned value
is always textually different from the unset default.

The same values are also what ``aliases.yaml`` ``ring_aliases`` needs as keys
(the exact stored value of the ring's name column), so :func:`draft_ring_aliases`
adds an entry for each value that has none -- the value itself and, if the
operator names a generic leading word, the value without it -- leaving every
existing entry, ``synonyms`` and comment as they are.

Public API
----------
DraftValues, DraftedPolicies
    The values kept and the reasons for the ones dropped, and the two texts.
usable_values(raw, default_label, max_length) -> DraftValues
choose_memory_keys(policies, requested) -> list[str]
draft_policies(policies, values, filter_key, column, memory_keys) -> DraftedPolicies
DraftedRings
    The drafted ``aliases.yaml`` and what was added to it.
ring_forms(value, generic_words) -> list[str]
draft_ring_aliases(text, values, generic_words) -> DraftedRings
"""

from __future__ import annotations

import re
from collections import Counter
from collections.abc import Iterable, Mapping, Sequence
from dataclasses import dataclass, field

import yaml

from core.persian import normalize_for_matching
from core.policy_values import MEMORY_FILE, SESSION_FILE, Policies
from core.value_shapes import unsuitable_value
from schema_data.yaml_text import (
    Edit,
    LayoutError,
    apply_edits,
    indent_of,
    is_blank_or_comment,
    quote,
)

__all__ = [
    "DraftValues",
    "DraftedPolicies",
    "DraftedRings",
    "choose_memory_keys",
    "draft_policies",
    "draft_ring_aliases",
    "ring_forms",
    "usable_values",
]


@dataclass(frozen=True)
class DraftValues:
    """The values kept as options, and why the others were not."""

    values: tuple[str, ...]
    """One spelling of each usable value, sorted."""
    read: int
    """How many values were given."""
    dropped: Mapping[str, int] = field(default_factory=dict)
    """``{reason: count}``; the reasons are the shape names of :mod:`core.value_shapes`
    (``date``, ``phone``, ``code``, ...), ``long``, ``short``, ``over max_length``,
    ``default label``, ``not text`` and ``duplicate``."""


@dataclass(frozen=True)
class DraftedPolicies:
    """The two drafted files as text, and what was changed in them."""

    memory_text: str
    session_text: str
    memory_keys: tuple[str, ...]
    """The memory keys whose ``options`` were replaced."""
    changed_dimension: bool
    """Whether ``filter_key`` (or a memory key's ``column``) was rewritten."""


def _has_arabic_letter(value: str) -> bool:
    """Whether *value* spells a letter with the Arabic ``ي`` or ``ك`` where Persian writes ``ی`` and ``ک``."""
    return "ي" in value or "ك" in value


def usable_values(raw: Iterable[object], default_label: str, max_length: int) -> DraftValues:
    """Cut the values read from a dimension column down to the options they may become.

    Parameters
    ----------
    raw:
        The distinct values read, in any order.
    default_label:
        The session policy's ``default_label``. A value that is the same word
        (after the folding below) is left out: a memory option must differ from it.
    max_length:
        The memory key's ``max_length``; a longer value could never be pinned.

    Returns
    -------
    DraftValues
        Surrounding whitespace is removed, inner spacing is kept. Two values
        that differ only by the letter forms, digits, spacing or case
        :func:`core.persian.normalize_for_matching` folds are one; the spelling
        kept is the first in sorted order, the Persian letters before the Arabic.

    Examples
    --------
    >>> found = usable_values(["Region B", "Region A", " region a ", "2024-05-01", "x", "All regions"],
    ...                       "All regions", 20)
    >>> found.values, found.read
    (('Region A', 'Region B'), 6)
    >>> sorted(found.dropped.items())
    [('date', 1), ('default label', 1), ('duplicate', 1), ('short', 1)]
    >>> usable_values(["علي", "علی"], "", 20).values
    ('علی',)
    >>> usable_values(["a" * 30], "", 20).dropped["over max_length"]
    1
    """
    dropped: Counter[str] = Counter()
    given = 0
    candidates: list[str] = []
    for item in raw:
        given += 1
        if not isinstance(item, str):
            dropped["not text"] += 1
            continue
        value = item.strip()
        why = unsuitable_value(value)
        if why is None and len(value) > max_length:
            why = "over max_length"
        if why is None and default_label and normalize_for_matching(value) == normalize_for_matching(default_label):
            why = "default label"
        if why is not None:
            dropped[why] += 1
            continue
        candidates.append(value)
    kept: dict[str, str] = {}
    for value in sorted(candidates, key=lambda v: (normalize_for_matching(v), _has_arabic_letter(v), v)):
        folded = normalize_for_matching(value)
        if folded in kept:
            dropped["duplicate"] += 1
        else:
            kept[folded] = value
    return DraftValues(tuple(kept.values()), given, dict(dropped))


def choose_memory_keys(policies: Policies, requested: str | None = None) -> list[str]:
    """The memory keys whose ``options`` belong to the session policy's dimension.

    Without *requested*: the keys whose ``filter_key`` is the session policy's
    (the same dimension) and that have a closed set of options. A key with no
    ``options`` takes any value up to its ``max_length`` by design, and closing
    it would change what an analyst can pin, so it is left alone unless it is
    named.

    Parameters
    ----------
    policies:
        From :func:`core.policy_values.read_policies`.
    requested:
        A memory key named by the operator; it is used whatever its options.

    Raises
    ------
    ValueError
        When *requested* is not a key of ``memory_policy.yaml``, or no key
        qualifies (the message says how to name one).

    Examples
    --------
    >>> from core.policy_values import SHIPPED_DIRECTORY, read_policies
    >>> template = read_policies(SHIPPED_DIRECTORY)
    >>> choose_memory_keys(template), choose_memory_keys(template, "scope")
    (['scope'], ['scope'])
    >>> choose_memory_keys(template, "other")
    Traceback (most recent call last):
        ...
    ValueError: memory_policy.yaml has no key 'other' (keys: scope)
    """
    keys = policies.memory.keys
    if requested is not None:
        if requested not in keys:
            raise ValueError(f"{MEMORY_FILE} has no key {requested!r} (keys: {', '.join(keys)})")
        return [requested]
    dimension = policies.session.default_scope.filter_key.strip().casefold()
    chosen = [
        name for name, entry in keys.items()
        if entry.filter_key.strip().casefold() == dimension and entry.options
    ]
    if not chosen:
        raise ValueError(
            f"no key of {MEMORY_FILE} has the filter_key of {SESSION_FILE} default_scope and a closed set of "
            f"options (keys: {', '.join(keys)}); name the one to draft with --memory-key"
        )
    return chosen


# ---------------------------------------------------------------------------
# Line-level editing of the two files
# ---------------------------------------------------------------------------

_KEY_LINE = re.compile(r"""^(?P<indent> *)(?P<key>"(?:[^"\\]|\\.)*"|'(?:[^']|'')*'|[^\s#'"\[\]{},&*!|>%@`-][^#:]*?)[ \t]*:[ \t]*(?:\#.*)?$""")
_QUOTED = re.compile(r"""^(?:"(?:[^"\\]|\\.)*"|'(?:[^']|'')*')""")


def _comment_of(text: str) -> str:
    """The ``# ...`` comment that ends the YAML fragment *text*, or ``""`` (a ``#`` in quotes is not one)."""
    inside = ""
    previous = " "
    for position, char in enumerate(text):
        if inside:
            if char == inside and not (inside == '"' and previous == "\\"):
                inside = ""
        elif char in "\"'" and previous in " \t[,":
            inside = char
        elif char == "#" and previous in " \t":
            return text[position:].rstrip()
        previous = char
    return ""


def _top_block(lines: Sequence[str], name: str) -> tuple[int, int]:
    """``(header line, end)`` of the top-level mapping *name*; *end* is one past its last content line."""
    header = next(
        (i for i, line in enumerate(lines) if re.match(rf"^{re.escape(name)}[ \t]*:[ \t]*(?:#.*)?$", line)), None,
    )
    if header is None:
        raise LayoutError(f"no top-level `{name}:` mapping written as a block")
    end = header + 1
    for i in range(header + 1, len(lines)):
        if is_blank_or_comment(lines[i]):
            continue
        if indent_of(lines[i]) == 0:
            break
        end = i + 1
    return header, end


def _entries(lines: Sequence[str], first: int, end: int) -> list[tuple[str, int, int]]:
    """``(key, line, stop)`` of each mapping entry between *first* and *end*, at the indentation of the first."""
    body = [i for i in range(first, end) if not is_blank_or_comment(lines[i])]
    if not body:
        return []
    indent = indent_of(lines[body[0]])
    heads: list[tuple[str, int]] = []
    for i in body:
        match = _KEY_LINE.match(lines[i])
        if match is not None and indent_of(lines[i]) == indent:
            heads.append((str(yaml.safe_load(match.group("key"))), i))
    out: list[tuple[str, int, int]] = []
    for n, (key, at) in enumerate(heads):
        limit = heads[n + 1][1] if n + 1 < len(heads) else end
        last = max(i for i in range(at, limit) if not is_blank_or_comment(lines[i])) + 1
        out.append((key, at, last))
    return out


def _field(lines: Sequence[str], first: int, end: int, name: str) -> tuple[int, int, int] | None:
    """``(line, stop, indent)`` of the field *name* of the mapping whose body is ``lines[first:end]``."""
    body = [i for i in range(first, end) if not is_blank_or_comment(lines[i])]
    if not body:
        return None
    indent = indent_of(lines[body[0]])
    pattern = re.compile(rf"^ {{{indent}}}{re.escape(name)}[ \t]*:(?P<rest>.*)$")
    for i in body:
        match = pattern.match(lines[i])
        if match is None:
            continue
        rest = match.group("rest")
        value = rest.strip()
        stop = i + 1
        if not value or value.startswith("#"):
            for j in range(i + 1, end):
                if is_blank_or_comment(lines[j]):
                    continue
                if indent_of(lines[j]) > indent or (indent_of(lines[j]) == indent and lines[j].lstrip().startswith("- ")):
                    stop = j + 1
                else:
                    break
        return i, stop, indent
    return None


def _set_scalar(lines: Sequence[str], where: tuple[int, int, int], name: str, value: str) -> list[Edit]:
    """The edit that gives the one-line field at *where* the value *value*, keeping its comment."""
    line, stop, indent = where
    match = re.match(rf"^( {{{indent}}}{re.escape(name)}[ \t]*:[ \t]*)(?P<rest>.*)$", lines[line])
    assert match is not None
    rest = match.group("rest")
    comment = _comment_of(rest)
    source = rest[: len(rest) - len(comment)].rstrip() if comment else rest.rstrip()
    gap = rest[len(source): len(rest) - len(comment)] if comment else ""
    if stop != line + 1 or not source or source[0] in "|>[{&*!":
        raise LayoutError(f"`{name}:` is not a one-line scalar; edit it by hand")
    quoted = _QUOTED.match(source)
    if quoted is not None and quoted.end() != len(source):
        raise LayoutError(f"`{name}:` is not a one-line scalar; edit it by hand")
    if yaml.safe_load(source) == value:
        return []
    return [(line, line + 1, [match.group(1) + quote(value) + (gap + comment if comment else "")])]


def _set_options(lines: Sequence[str], where: tuple[int, int, int] | None, entry: tuple[int, int, int], values: Sequence[str]) -> list[Edit]:
    """The edit that makes ``options`` the block list of *values*, in place or after the entry's last field."""
    first, last, indent = entry
    if where is None:
        return [(last, last, [f"{' ' * indent}options:", *(f"{' ' * (indent + 2)}- {quote(v)}" for v in values)])]
    line, stop, indent = where
    header = lines[line]
    rest = header.split(":", 1)[1]
    comment = _comment_of(rest)
    flow = rest[: len(rest) - len(comment)].strip() if comment else rest.strip()
    if flow.startswith("[") and "]" not in flow:
        raise LayoutError("`options:` is a flow list written over several lines; edit it by hand")
    items = [lines[j] for j in range(line + 1, stop) if lines[j].lstrip().startswith("- ")]
    item_indent = indent_of(items[0]) if items else indent + 2
    gap = rest[: len(rest) - len(comment)][len(rest[: len(rest) - len(comment)].rstrip()):] if comment else ""
    rewritten = [f"{' ' * indent}options:" + (gap + comment if comment else "")]
    rewritten += [f"{' ' * item_indent}- {quote(v)}" for v in values]
    return [(line, stop, rewritten)]


def _edit_mapping(
    lines: Sequence[str], first: int, end: int, scalars: Mapping[str, str], options: Sequence[str] | None,
) -> list[Edit]:
    """Edits for the mapping whose body is ``lines[first:end]``: the scalars, then ``options``."""
    body = [i for i in range(first, end) if not is_blank_or_comment(lines[i])]
    if not body:
        raise LayoutError("an entry with no field")
    indent = indent_of(lines[body[0]])
    edits: list[Edit] = []
    last = body[-1] + 1
    for name, value in scalars.items():
        where = _field(lines, first, end, name)
        if where is None:
            edits.append((last, last, [f"{' ' * indent}{name}: {quote(value)}"]))
        else:
            edits.extend(_set_scalar(lines, where, name, value))
    if options is not None:
        edits.extend(_set_options(lines, _field(lines, first, end, "options"), (first, last, indent), options))
    return edits


def _join(lines: Sequence[str], eol: str, bom: bool) -> str:
    return ("" if bom else "") + eol.join(lines)


def _split(text: str) -> tuple[list[str], str, bool]:
    """``(lines, line ending, had a byte order mark)`` of *text*."""
    bom = text.startswith("")
    body = text[1:] if bom else text
    eol = "\r\n" if "\r\n" in body else "\n"
    return body.replace("\r\n", "\n").split("\n"), eol, bom


def _edit_memory(text: str, keys: Mapping[str, tuple[Mapping[str, str], Sequence[str]]]) -> str:
    lines, eol, bom = _split(text)
    header, end = _top_block(lines, "keys")
    found = {key: (at, stop) for key, at, stop in _entries(lines, header + 1, end)}
    edits: list[Edit] = []
    for key, (scalars, options) in keys.items():
        if key not in found:
            raise LayoutError(f"{MEMORY_FILE}: key {key!r} is not written as a block entry")
        at, stop = found[key]
        edits.extend(_edit_mapping(lines, at + 1, stop, scalars, options))
    return _join(apply_edits(list(lines), edits), eol, bom)


def _edit_session(text: str, scalars: Mapping[str, str], options: Sequence[str]) -> str:
    lines, eol, bom = _split(text)
    header, end = _top_block(lines, "default_scope")
    edits = _edit_mapping(lines, header + 1, end, scalars, options)
    return _join(apply_edits(list(lines), edits), eol, bom)


def draft_policies(
    policies: Policies,
    values: Sequence[str],
    filter_key: str | None = None,
    column: str | None = None,
    memory_keys: Sequence[str] | None = None,
) -> DraftedPolicies:
    """Draft both files with *values* as the options, changing nothing else.

    The memory keys get ``options`` = *values*; the session policy gets
    ``options`` = its ``default_label`` followed by *values*, the order the
    template uses. *filter_key* and *column*, when given, replace those fields
    in the session policy (``filter_key``) and in the chosen memory keys
    (``filter_key`` and ``column``); ``field_name``, ``default_label``,
    ``clarification_prompt``, ``max_length`` and any other field are never
    touched.

    Parameters
    ----------
    policies:
        The current files, from :func:`core.policy_values.read_policies`.
    values:
        The options, from :func:`usable_values`.
    filter_key:
        The new ``filter_key``, or ``None`` to keep each file's own.
    column:
        The new memory ``column``, or ``None``.
    memory_keys:
        The memory keys to draft; :func:`choose_memory_keys` when ``None``.

    Returns
    -------
    DraftedPolicies

    Raises
    ------
    ValueError
        When *values* is empty or holds the ``default_label``; when a file is
        laid out in a way the line editor cannot edit
        (:class:`~schema_data.yaml_text.LayoutError`, a ``ValueError``); or when
        the result does not load with the application's validators or differs
        from the input in any field but the ones above. Nothing is returned
        then, so nothing wrong is ever written.

    Examples
    --------
    >>> from core.policy_values import SHIPPED_DIRECTORY, read_policies
    >>> drafted = draft_policies(read_policies(SHIPPED_DIRECTORY), ["North", "South"], "Zone", "Title")
    >>> drafted.memory_keys, drafted.changed_dimension
    (('scope',), True)
    >>> print(drafted.session_text.replace(chr(13), "").split("options:")[1].split("clarification")[0].strip())
    - "All rings"
        - "North"
        - "South"
    >>> 'filter_key: "Zone"' in drafted.memory_text and 'column: "Title"' in drafted.memory_text
    True
    """
    from knowledge.config_loader import MemoryPolicyConfig, SessionPolicyConfig, validate_yaml_text

    options = list(values)
    if not options:
        raise ValueError("no value to draft the options from")
    scope = policies.session.default_scope
    folded_default = normalize_for_matching(scope.default_label)
    if any(normalize_for_matching(v) == folded_default for v in options):
        raise ValueError(f"a value equals the default_label of {SESSION_FILE}; a memory option must differ from it")
    chosen = list(memory_keys) if memory_keys is not None else choose_memory_keys(policies)

    memory_scalars = {name: value for name, value in (("filter_key", filter_key), ("column", column)) if value is not None}
    session_scalars = {"filter_key": filter_key} if filter_key is not None else {}
    memory_text = _edit_memory(policies.memory_text, {key: (memory_scalars, options) for key in chosen})
    session_text = _edit_session(policies.session_text, session_scalars, [scope.default_label, *options])

    # Proof: both load as the application loads them, and differ from the input only where intended.
    memory = validate_yaml_text(MEMORY_FILE, memory_text, MemoryPolicyConfig)
    session = validate_yaml_text(SESSION_FILE, session_text, SessionPolicyConfig)
    expected_memory = policies.memory.model_dump()
    for key in chosen:
        expected_memory["keys"][key].update(memory_scalars)
        expected_memory["keys"][key]["options"] = options
    expected_session = policies.session.model_dump()
    expected_session["default_scope"].update(session_scalars)
    expected_session["default_scope"]["options"] = [scope.default_label, *options]
    if memory.model_dump() != expected_memory or session.model_dump() != expected_session:
        raise LayoutError("the edit would change a field other than the ones asked for; edit the files by hand")
    entries = [policies.memory.keys[key] for key in chosen]
    moved = filter_key is not None and (
        scope.filter_key != filter_key or any(entry.filter_key != filter_key for entry in entries)
    )
    changed = moved or (column is not None and any(entry.column != column for entry in entries))
    return DraftedPolicies(memory_text, session_text, tuple(chosen), changed)


# ---------------------------------------------------------------------------
# ring_aliases of aliases.yaml
# ---------------------------------------------------------------------------

#: Written when there is no ``aliases.yaml`` to add to: both sections the loader requires.
_NEW_ALIASES = "ring_aliases: {}\nsynonyms: {}\n"


@dataclass(frozen=True)
class DraftedRings:
    """The drafted ``aliases.yaml`` as text, and what was added to it."""

    text: str
    added: tuple[str, ...]
    """The values that got a new ``ring_aliases`` entry, in the order written."""
    kept: int
    """Entries the file already had; none of them was changed."""
    stale: int
    """Existing keys that are not among the values read (a count; the entries stay)."""


def ring_forms(value: str, generic_words: Sequence[str] = ()) -> list[str]:
    """The forms an analyst may type for the ring stored as *value*.

    The value itself (``extract_ring`` finds a form inside the question, so the
    stored name is the first thing that works), and one conservative variant: the
    value without a leading generic word the operator named (``Hall Export`` for
    the generic word ``Hall`` gives ``Export``). The variant is made only when a
    whole word is left; words inside the value are never cut, and nothing is
    inferred from the data.

    Parameters
    ----------
    value:
        The exact stored value of the ring's name column.
    generic_words:
        Words that begin many ring names and that analysts leave out (``Hall``,
        ``Ring``); compared whole, with the project's folding.

    Examples
    --------
    >>> ring_forms("Hall Export", ["Hall"])
    ['Hall Export', 'Export']
    >>> ring_forms("Hall Export")
    ['Hall Export']
    >>> ring_forms("Hallway Export", ["Hall"])
    ['Hallway Export']
    >>> ring_forms("Hall", ["Hall"])
    ['Hall']
    """
    forms = [value]
    words = value.split()
    generic = {normalize_for_matching(word) for word in generic_words}
    if len(words) > 1 and normalize_for_matching(words[0]) in generic:
        rest = " ".join(words[1:])
        if normalize_for_matching(rest) not in generic:
            forms.append(rest)
    return forms


def _ring_entry_lines(value: str, forms: Sequence[str], indent: int) -> list[str]:
    pad = " " * indent
    return [f"{pad}{quote(value)}:", *(f"{pad}  - {quote(form)}" for form in forms)]


def draft_ring_aliases(text: str, values: Sequence[str], generic_words: Sequence[str] = ()) -> DraftedRings:
    """Add a ``ring_aliases`` entry for each stored ring value that has none, changing nothing else.

    ``ring_aliases`` maps the exact value of the ring dimension's name column to
    the forms analysts type for it. The entries are added to the *text* of
    ``aliases.yaml`` line by line, after the last entry of the section, so every
    comment, every curated entry, ``synonyms`` and the line endings stay as they
    were; a value that already has an entry (same name after the project's
    folding) is not touched. The forms of an entry are :func:`ring_forms`; a
    shortened form that two values would share is left out, since it would name
    neither. New entries are written longest value first: ``extract_ring``
    returns the first entry with a form in the question, so ``Region AB`` must
    come before ``Region A``.

    Parameters
    ----------
    text:
        The current ``aliases.yaml``; ``""`` when there is none.
    values:
        The stored ring values (:func:`usable_values`).
    generic_words:
        See :func:`ring_forms`.

    Returns
    -------
    DraftedRings

    Raises
    ------
    ValueError
        When the file does not load with the application's validator, is laid
        out in a way the line editor cannot edit (a non-empty ``ring_aliases``
        written inline, :class:`~schema_data.yaml_text.LayoutError`), or the
        result would differ from the input in anything but the added entries.
        Nothing is returned then.

    Examples
    --------
    >>> drafted = draft_ring_aliases('# mine\\nring_aliases:\\n  "Zone A": ["a"]\\nsynonyms:\\n  "annual": ["year"]\\n',
    ...                              ["Zone A", "Zone B", "Zone BB"])
    >>> drafted.added, drafted.kept
    (('Zone BB', 'Zone B'), 1)
    >>> print(drafted.text)
    # mine
    ring_aliases:
      "Zone A": ["a"]
      "Zone BB":
        - "Zone BB"
      "Zone B":
        - "Zone B"
    synonyms:
      "annual": ["year"]
    <BLANKLINE>
    >>> draft_ring_aliases("", ["Zone A"]).text.splitlines()[:3]
    ['ring_aliases:', '  "Zone A":', '    - "Zone A"']
    """
    from knowledge.config_loader import AliasesConfig, validate_yaml_text

    source = text if text.strip() else _NEW_ALIASES
    before = validate_yaml_text("aliases.yaml", source, AliasesConfig).model_dump()
    taken = {normalize_for_matching(key) for key in before["ring_aliases"]}
    wanted = {normalize_for_matching(value) for value in values}
    fresh = [value for value in values if normalize_for_matching(value) not in taken]
    fresh.sort(key=lambda value: (-len(value), normalize_for_matching(value)))
    forms = {value: ring_forms(value, generic_words) for value in fresh}
    shared = Counter(normalize_for_matching(form) for found in forms.values() for form in found[1:])
    forms = {
        value: [f for i, f in enumerate(found) if i == 0 or shared[normalize_for_matching(f)] == 1]
        for value, found in forms.items()
    }

    lines, eol, bom = _split(source)
    inline = next((i for i, line in enumerate(lines) if re.match(r"^ring_aliases[ \t]*:[ \t]*\{[ \t]*\}[ \t]*(?:#.*)?$", line)), None)
    edits: list[Edit]
    if inline is not None:
        comment = _comment_of(lines[inline].split(":", 1)[1])
        edits = [(inline, inline + 1, ["ring_aliases:" + (f" {comment}" if comment else ""), *_all_entries(fresh, forms, 2)])] if fresh else []
    else:
        try:
            header, end = _top_block(lines, "ring_aliases")
        except LayoutError:
            raise LayoutError("`ring_aliases:` is written inline; edit it by hand") from None
        body = [i for i in range(header + 1, end) if not is_blank_or_comment(lines[i])]
        indent = indent_of(lines[body[0]]) if body else 2
        edits = [(end, end, _all_entries(fresh, forms, indent))] if fresh else []
    drafted = _join(apply_edits(list(lines), edits), eol, bom)

    # Proof: it loads as the application loads it, and is the input plus exactly the new entries.
    after = validate_yaml_text("aliases.yaml", drafted, AliasesConfig).model_dump()
    expected = {"ring_aliases": {**before["ring_aliases"], **{v: forms[v] for v in fresh}}, "synonyms": before["synonyms"]}
    if after != expected or list(after["ring_aliases"]) != list(expected["ring_aliases"]):
        raise LayoutError("the edit would change more than the added entries; edit aliases.yaml by hand")
    stale = sum(1 for key in before["ring_aliases"] if normalize_for_matching(key) not in wanted)
    return DraftedRings(drafted, tuple(fresh), len(before["ring_aliases"]), stale)


def _all_entries(fresh: Sequence[str], forms: Mapping[str, Sequence[str]], indent: int) -> list[str]:
    return [line for value in fresh for line in _ring_entry_lines(value, forms[value], indent)]