privaterelay/cleaner_task.py

"""Framework for tasks that identify data issues and (if possible) clean them up"""

from __future__ import annotations

import string
from abc import ABCMeta, abstractmethod
from typing import Any, Generic, Literal, TypeVar, get_args

from django.db.models import Model, Q
from django.db.models.query import QuerySet

Counts = dict[str, dict[str, int]]
CleanupData = dict[str, Any]

M = TypeVar("M", bound=Model)
CLEAN_GROUP_T = Literal["ok", "needs_cleaning"]

# Define allowed characters for item keys
# {model_plural}.[!]{sub_name1}.[!]{sub_name2}
_KEY_SEP = "."
_NEGATE_PREFIX = "!"
_ITEM_KEY_CHAR_SET = set(
    string.ascii_lowercase + string.digits + "_" + _NEGATE_PREFIX + _KEY_SEP
)


# Used by CleanedItem
_CLEANED_METRIC_NAME = "cleaned"


def _metric_name_for_model(model: type[M]) -> str:
    """The model's metric key, used in metrics and as a dictionary key."""
    return str(model._meta.verbose_name_plural).replace(" ", "_")


class ReportItem(metaclass=ABCMeta):
    """
    An item in a data task report.

    This is the base model in the reporting item hierarchy. Code should use the derived
    classes CleanedItem, DataModelItem, and DataItem.
    """

    def __init__(self, metric_name: str | None = None, report_name: str | None = None):
        """
        Initialize a ReportItem.

        The `metric_name` parameter sets the name of the entry when it appears as a
        `dict` or JSON key. The default is `None`, which omits the entry from
        reports.

        The `report_name` parameter sets the name of the entry when it appears in a
        report for humans. It can be omitted when `metric_name` is None.
        """
        if metric_name and (
            bad_chars := [c for c in metric_name if c not in _ITEM_KEY_CHAR_SET]
        ):
            raise ValueError(
                f"metric_name '{metric_name}' has disallowed character"
                f"{'' if len(bad_chars) == 1 else 's'} '{''.join(sorted(bad_chars))}'"
            )
        if metric_name == "":
            raise ValueError("metric_name is an empty string, should be None")
        if metric_name is None and report_name is not None:
            raise ValueError(f"report_name is '{report_name}', but metric_name is None")
        if report_name == "":
            raise ValueError("report_name is an empty string, should be None")

        self.metric_name = metric_name
        self.report_name = report_name

    @abstractmethod
    def count(self) -> int:
        raise NotImplementedError


class CleanedItem(ReportItem):
    """Represents the results of cleaning a Model."""

    def __init__(self, count: int, report_name: str = "Cleaned") -> None:
        if count < 0:
            raise ValueError("count can not be negative")
        self._count = count
        super().__init__(metric_name=_CLEANED_METRIC_NAME, report_name=report_name)

    def __repr__(self) -> str:
        args = [str(self._count)]
        if self.report_name != "Cleaned":
            args.append(f"report_name={self.report_name!r}")
        return f'{type(self).__name__}({", ".join(args)})'

    def __eq__(self, other: Any) -> bool:
        if isinstance(other, CleanedItem):
            return (
                self.__class__ == other.__class__
                and self._count == other._count
                and self.metric_name == other.metric_name
                and self.report_name == other.report_name
            )
        return NotImplemented

    def count(self) -> int:
        return self._count


class BaseDataItem(ReportItem, Generic[M]):
    """An entry in a data task report backed by a database query."""

    def __init__(
        self,
        model_or_parent: type[M] | BaseDataItem[M],
        filter_by: str | Q | None = None,
        exclude: bool = False,
        metric_name: str | None = None,
        report_name: str | None = None,
    ) -> None:
        if metric_name == _CLEANED_METRIC_NAME:
            raise ValueError(f"metric_name '{metric_name}' is reserved for CleanedItem")

        self._model_or_parent = model_or_parent
        self.filter_by = filter_by
        self.exclude = exclude
        super().__init__(metric_name=metric_name, report_name=report_name)

    def __eq__(self, other: Any) -> bool:
        if isinstance(other, BaseDataItem):
            return (
                self._model_or_parent == other._model_or_parent
                and self.filter_by == other.filter_by
                and self.exclude == other.exclude
                and self.metric_name == other.metric_name
                and self.report_name == other.report_name
            )
        return NotImplemented

    def get_queryset(self) -> QuerySet[M]:
        """Return the Django query for this BaseDataItem."""
        if isinstance(self._model_or_parent, BaseDataItem):
            query = self._model_or_parent.get_queryset()
        else:
            query = self._model_or_parent._default_manager.all()

        if isinstance(self.filter_by, str):
            filter_by = {self.filter_by: True}
            if self.exclude:
                query = query.exclude(**filter_by)
            else:
                query = query.filter(**filter_by)
        elif isinstance(self.filter_by, Q):
            if self.exclude:
                query = query.exclude(self.filter_by)
            else:
                query = query.filter(self.filter_by)
        return query

    def count(self) -> int:
        """Return the number of rows matched for this BaseDataItem."""
        return self.get_queryset().count()


class DataModelItem(BaseDataItem[M]):
    """A BaseDataItem representing the top-level Model"""

    _model_or_parent: type[M]
    metric_name: str
    report_name: str
    filter_by: None

    def __init__(self, model: type[M]) -> None:
        """Initialize a DataModelItem."""
        super().__init__(
            model_or_parent=model,
            metric_name=_metric_name_for_model(model),
            report_name=str(model._meta.verbose_name_plural).title(),
        )

    def __repr__(self) -> str:
        return f"{type(self).__name__}({self._model_or_parent.__name__})"


class DataItem(BaseDataItem[M]):
    """
    A DataItem is a subquery of a DataModelItem or DataItem.

    A top-level model DataItem represents all the items in a table. A query that selects
    some rows is a DataItem with a parent. The specific rows of interest can be
    represented by multiple levels of DataItems, giving context to the specific rows.
    """

    _model_or_parent: BaseDataItem[M]

    def __init__(
        self,
        parent: BaseDataItem[M],
        filter_by: str | Q,
        exclude: bool = False,
        metric_name: str | None = None,
        report_name: str | None = None,
        clean_group: CLEAN_GROUP_T | None = None,
    ) -> None:
        """
        Initialize a DataItem, checking for init-time issues.

        The `filter_by` parameter sets the filter (`filter` is a Python keyword). It can
        be a string, which represents a boolean filter. It can be a Django Q object,
        such as `Q(num_deleted_relay_addresses__gt=5)`.

        The default is to include rows matching the query. If `exclude` is set to
        `True`, then the query is for rows that do not match the filter.

        The `clean_group` parameter identifies the DataItem as a query of interest,
        usually in the context of a CleanerTask. An 'ok' value means the query
        represents rows without a problem, and a 'needs_cleaning' value means the rows
        need fixing.
        """

        if filter_by == "":
            raise ValueError("filter_by is an empty string, should be set")
        if metric_name is None and clean_group is not None:
            raise ValueError(f"clean_group is '{clean_group}', but metric_name is None")
        if clean_group is not None and clean_group not in get_args(CLEAN_GROUP_T):
            raise ValueError(f"clean_group has invalid value '{clean_group}'")
        self.clean_group = clean_group

        super().__init__(
            model_or_parent=parent,
            filter_by=filter_by,
            exclude=exclude,
            metric_name=metric_name,
            report_name=report_name,
        )

    def __repr__(self) -> str:
        if isinstance(self._model_or_parent, DataItem):
            args = [
                f"<{type(self._model_or_parent).__name__}"
                f"(metric_name={self._model_or_parent.metric_name!r}, ...)>"
            ]
        else:
            args = [repr(self._model_or_parent)]
        args.append(repr(self.filter_by))
        if self.exclude:
            args.append(f"exclude={self.exclude!r}")
        if self.metric_name:
            args.append(f"metric_name={self.metric_name!r}")
        if self.report_name:
            args.append(f"report_name={self.report_name!r}")
        if self.clean_group:
            args.append(f"clean_group={self.clean_group!r}")
        return f'{type(self).__name__}({", ".join(args)})'


class DataModelSpec(Generic[M]):
    """
    Define queries on a Model that can identify issues.

    This provides a higher-level interface for constructing a hierarchy of DataItems.
    The top-level is the most general, and queries are filtered to focus on more
    specific sets of rows as you go down the hierarchy.

    Each DataItem is identified by a key which represents its place in the hierarchy.
    For example, the key for a DataItem for active users created in 2023 might be
    'active.created_in_2023'.

    When exporting the {key: DataItem} dict with `to_data_items`, the keys are turned
    into fully-qualified key by prefixing them with the model key. The model key is
    derived from the `verbose_plural_name` of the model. For example, a fully-qualified
    key would be 'users.active.created_in_2023'.

    https://docs.djangoproject.com/en/4.2/ref/models/options/#verbose-name-plural
    """

    def __init__(
        self,
        model: type[M],
        subdivisions: list[DataBisectSpec] | None = None,
        omit_key_prefixes: list[str] | None = None,
        metric_name_overrides: dict[str, str] | None = None,
        report_name_overrides: dict[str, str] | None = None,
        ok_key: str | None = None,
        needs_cleaning_key: str | None = None,
        cleaned_report_name: str = "Cleaned",
    ) -> None:
        """
        Initialize a DataModelSpec, checking for init-time issues.

        Keyword arguments:

        model - The model for this hierarchy of queries
        subdivisions - The subqueries, as a list of DataBisectSpecs
        omit_key_prefixes - A list of key prefixes that should be omitted from reports.
            Since the key name represents the hierarchy, omitting a prefix removes a
            whole branch of queries and subqueries.
        metric_name_overrides - A dict of keys to the metric names as used in `dict` and
            JSON keys. The default metric name is the last component of the key.
        report_name_overrides - A dict of keys to the human-suitable report names. The
            default report name is the last component of the key, converted to a
            title-cased phrase.
        ok_key - The key for the "ok" DataItem, such as rows that have already been
            cleaned.
        needs_cleaning_key - The key for the "needs_cleaning" DataItem, such as rows
            that need manual or automated cleaning.
        cleaned_report_name - The report name for the cleaned DataItem, added after
            running cleaning. This defaults to "Cleaned".
        """
        if not isinstance(model, type) or not issubclass(model, Model):
            raise ValueError(f"model {model!r} is not a Django model.")

        subkeys: set[str] = set()
        for sub in subdivisions or []:
            for key in sub.get_keys():
                if key in subkeys:
                    raise ValueError("Duplicate key 'active' in subdivisions")
                subkeys.add(key)

        if omit_key_prefixes:
            if "" in omit_key_prefixes:
                raise ValueError(
                    "omit_key_prefixes should not include the empty string"
                )
            for key in omit_key_prefixes:
                if key not in subkeys:
                    raise ValueError(
                        f"omit_key_prefixes key '{key}' not found in subdivision"
                        f" keys {sorted(subkeys)}"
                    )
            for key in (metric_name_overrides or {}).keys():
                if key in omit_key_prefixes:
                    raise ValueError(
                        f"The metric_name_overrides key '{key}'"
                        f" should not be in omit_key_prefixes {omit_key_prefixes}"
                    )
            for key in (report_name_overrides or {}).keys():
                if key in omit_key_prefixes:
                    raise ValueError(
                        f"The report_name_overrides key '{key}'"
                        f" should not be in omit_key_prefixes {omit_key_prefixes}"
                    )
            if ok_key and ok_key in omit_key_prefixes:
                raise ValueError(
                    f"The ok_key '{ok_key}'"
                    f" should not be in omit_key_prefixes {omit_key_prefixes}"
                )
            if needs_cleaning_key and needs_cleaning_key in omit_key_prefixes:
                raise ValueError(
                    f"The needs_cleaning_key '{needs_cleaning_key}'"
                    f" should not be in omit_key_prefixes {omit_key_prefixes}"
                )

        if metric_name_overrides:
            for key in metric_name_overrides.keys():
                if key not in subkeys:
                    raise ValueError(
                        f"metric_name_overrides key '{key}' not found in subdivision"
                        f" keys {sorted(subkeys)}"
                    )
        if report_name_overrides:
            for key in report_name_overrides.keys():
                if key not in subkeys:
                    raise ValueError(
                        f"report_name_overrides key '{key}' not found in subdivision"
                        f" keys {sorted(subkeys)}"
                    )
        if ok_key and ok_key not in subkeys:
            raise ValueError(
                f"ok_key '{ok_key}' not found in subdivision keys {sorted(subkeys)}"
            )
        if needs_cleaning_key and needs_cleaning_key not in subkeys:
            raise ValueError(
                f"needs_cleaning_key '{needs_cleaning_key}' not found in subdivision"
                f" keys {sorted(subkeys)}"
            )

        self.model = model
        self.subdivisions = subdivisions or []
        self.omit_key_prefixes = omit_key_prefixes or []
        self.metric_name_overrides = metric_name_overrides or {}
        self.report_name_overrides = report_name_overrides or {}
        self.ok_key = ok_key
        self.needs_cleaning_key = needs_cleaning_key
        self.cleaned_report_name = cleaned_report_name

    def __repr__(self) -> str:
        args = [f"model={self.model.__name__}"]
        if self.subdivisions:
            args.append(f"subdivisions={self.subdivisions!r}")
        if self.omit_key_prefixes:
            args.append(f"omit_key_prefixes={self.omit_key_prefixes!r}")
        if self.metric_name_overrides:
            args.append(f"metric_name_overrides={self.metric_name_overrides!r}")
        if self.report_name_overrides:
            args.append(f"report_name_overrides={self.report_name_overrides!r}")
        if self.ok_key:
            args.append(f"ok_key={self.ok_key!r}")
        if self.needs_cleaning_key:
            args.append(f"needs_cleaning_key={self.needs_cleaning_key!r}")
        if self.cleaned_report_name != "Cleaned":
            args.append(f"cleaned_report_name={self.cleaned_report_name!r}")
        return f'{type(self).__name__}({", ".join(args)})'

    @property
    def model_key(self) -> str:
        return _metric_name_for_model(self.model)

    def omit_key(self, key: str) -> bool:
        return any(
            key == omit or key.startswith(omit + _KEY_SEP)
            for omit in self.omit_key_prefixes
        )

    def metric_name(self, key: str) -> str | None:
        """Return None (to omit), a friendlier name, or the original name."""
        if self.omit_key(key):
            return None
        return self.metric_name_overrides.get(key, key)

    def report_name(self, key: str) -> str | None:
        if self.omit_key(key):
            return None
        return self.report_name_overrides.get(
            key, key.split(_KEY_SEP)[-1].replace("_", " ").replace("!", "not ").title()
        )

    def clean_group(self, subname: str) -> CLEAN_GROUP_T | None:
        """Identify when the subname is for a key cleaning stat."""
        if subname == self.ok_key:
            return "ok"
        elif subname == self.needs_cleaning_key:
            return "needs_cleaning"
        else:
            return None

    def to_data_items(self) -> dict[str, BaseDataItem[M]]:
        """Converts the spec to a dictionary of DataItems."""
        model_item = DataModelItem(self.model)
        data_items: dict[str, BaseDataItem[M]] = {"": model_item}
        for subdivision in self.subdivisions:
            data_items.update(subdivision.to_data_items(self, data_items).items())

        # For the return dict, prefix with model's metric name
        model_key = model_item.metric_name
        return {
            (f"{model_key}{_KEY_SEP}{key}" if key else model_key): item
            for key, item in data_items.items()
        }


class DataBisectSpec:
    """Bisect a parent query."""

    def __init__(
        self,
        key: str,
        bisect_by: str | Q,
    ) -> None:
        if key.startswith(_KEY_SEP):
            raise ValueError(f"The key '{key}' should not start with '{_KEY_SEP}'")
        if bad_chars := [c for c in key if c not in _ITEM_KEY_CHAR_SET]:
            raise ValueError(
                f"key '{key}' has disallowed character"
                f"{'' if len(bad_chars) == 1 else 's'} '{''.join(sorted(bad_chars))}'"
            )
        if not bisect_by:
            raise ValueError("Set the bisect_by filter")
        parts = key.split(".")
        for part in parts:
            if _NEGATE_PREFIX in part[1:]:
                raise ValueError(
                    f"In key '{key}', the prefix '{_NEGATE_PREFIX}' is in the"
                    f" middle of subkey '{part}'"
                )
        if parts[-1][0] == _NEGATE_PREFIX:
            raise ValueError(
                f"In key '{key}', the prefix '{_NEGATE_PREFIX}' is not allowed"
                f" in the last subkey '{parts[-1]}'"
            )

        self.key = key
        self.bisect_by = bisect_by

    def __repr__(self) -> str:
        args = [f"key={self.key!r}", f"bisect_by={self.bisect_by!r}"]
        return f'{type(self).__name__}({", ".join(args)})'

    def get_keys(self) -> list[str]:
        if _KEY_SEP in self.key:
            subparent_name, part_name = self.key.rsplit(_KEY_SEP, 1)
            neg_key = f"{subparent_name}{_KEY_SEP}{_NEGATE_PREFIX}{part_name}"
        else:
            subparent_name = ""
            part_name = self.key
            neg_key = f"{_NEGATE_PREFIX}{part_name}"
        return [self.key, neg_key]

    def to_data_items(
        self, model_spec: DataModelSpec[M], existing_items: dict[str, BaseDataItem[M]]
    ) -> dict[str, DataItem[M]]:
        """Return two data items bisecting the parent data."""
        if _KEY_SEP in self.key:
            subparent_name, _ = self.key.rsplit(_KEY_SEP, 1)
        else:
            subparent_name = ""
        parent = existing_items[subparent_name]
        pos_key, neg_key = self.get_keys()
        return {
            pos_key: self._to_bisected_data_item(
                self.key, model_spec, parent, "positive"
            ),
            neg_key: self._to_bisected_data_item(
                neg_key, model_spec, parent, "negative"
            ),
        }

    def _to_bisected_data_item(
        self,
        key: str,
        model_spec: DataModelSpec[M],
        parent: BaseDataItem[M],
        bisect: Literal["positive", "negative"],
    ) -> DataItem[M]:
        """Create one of the bisected data items."""
        return DataItem(
            parent=parent,
            filter_by=self.bisect_by,
            exclude=bisect == "negative",
            metric_name=model_spec.metric_name(key),
            clean_group=model_spec.clean_group(key),
            report_name=model_spec.report_name(key),
        )


class ReportEntry:
    """An entry in a report."""

    def __init__(
        self,
        item: ReportItem,
        count: int,
        depth: int,
        child_keys: list[str],
    ) -> None:
        self.item = item
        self.count = count
        self.depth = depth
        self.child_keys = child_keys

    def __repr__(self) -> str:
        return (
            f"{type(self).__name__}({self.item!r}, {self.count!r}, {self.depth!r},"
            f" {self.child_keys!r})"
        )

    def __eq__(self, other: Any) -> bool:
        if isinstance(other, self.__class__):
            return (
                self.__class__ == other.__class__
                and self.item == other.item
                and self.count == other.count
                and self.depth == other.depth
                and self.child_keys == other.child_keys
            )
        return NotImplemented


class DataIssueTask:
    """Base class for data issue / cleaner tasks."""

    slug: str  # Short name, appropriate for command-line option
    title: str  # Short title for reports
    check_description: str  # A sentence describing what this cleaner is checking.
    can_clean: bool = False  # True if the issue can be automatically cleaned
    data_specification: list[DataModelSpec[Any]] = []  # The specification for this task

    _counts: Counts | None
    _cleanup_data: CleanupData | None
    _cleaned: bool
    _cleaned_report_name: dict[str, str]

    def __init__(self) -> None:
        self._counts = None
        self._cleanup_data = None
        self._cleaned = False
        self._cleaned_report_name = {}
        self.data_items = self._get_data_items()

    def _get_data_items(self) -> dict[str, BaseDataItem[Any]]:
        """Turn the data_specification into a dictionary of names to DataItems."""
        data_items: dict[str, BaseDataItem[Any]] = {}
        for model_spec in self.data_specification:
            if model_spec.model_key in data_items:
                raise ValueError(
                    f"{model_spec!r}\nThe key '{model_spec.model_key}' already exists"
                )
            data_items.update(self._get_data_items_for_model_spec(model_spec))
        return data_items

    def _get_data_items_for_model_spec(
        self, model_spec: DataModelSpec[M]
    ) -> dict[str, BaseDataItem[M]]:
        data_items: dict[str, BaseDataItem[M]] = {}
        self._cleaned_report_name[model_spec.model_key] = model_spec.cleaned_report_name
        for name, item in model_spec.to_data_items().items():
            if (
                self.can_clean
                and isinstance(item, DataItem)
                and item.clean_group == "needs_cleaning"
                and not hasattr(self, f"clean_{model_spec.model_key}")
            ):
                raise ValueError(
                    f"{model_spec!r}\n{item}\n"
                    "This item has clean_group='needs_cleaning', but the"
                    f" cleaning function clean_{model_spec.model_key} is not defined."
                )
            data_items[name] = item
        return data_items

    @property
    def counts(self) -> Counts:
        """Get relevant counts for data issues and prepare to clean if possible."""
        if self._counts is None:
            self._counts, self._cleanup_data = self._get_counts_and_data()
        return self._counts

    def _get_counts_and_data(self) -> tuple[Counts, CleanupData]:
        if not self.data_items:
            return {}, {}

        counts: Counts = {"summary": {"ok": 0, "needs_cleaning": 0}}
        cleanup_data: CleanupData = {}

        for name, data_item in self.data_items.items():
            if not data_item.metric_name:
                continue
            count = data_item.count()

            if isinstance(data_item, DataModelItem):
                counts[name] = {"all": count}
            else:
                model_name = name.split(".")[0]
                counts[model_name][data_item.metric_name] = count

            if isinstance(data_item, DataItem) and data_item.clean_group:
                counts["summary"][data_item.clean_group] += count
                if data_item.clean_group == "needs_cleaning":
                    cleanup_data[model_name] = name
        return counts, cleanup_data

    @property
    def cleanup_data(self) -> CleanupData:
        """Get data needed to clean data issues."""
        if not self.counts:
            return {}
        return self._cleanup_data or {}

    def issues(self) -> int:
        """Return the number of detected data issues."""
        return self.counts["summary"]["needs_cleaning"]

    def get_report_entries(self) -> dict[str, ReportEntry]:
        """
        Return an ordered dict of ReportEntries and related data.

        The key of the returned dict is a dotted path representing the path to the top
        of the hierarchy. If the ReportEntry is a DataModelItem or DataItem, it is the
        same key as used in `data_items`.

        The value of the returned dict is a 4-element tuple:
        - The ReportEntry, which may be a DataModelItem or DataItem
        - The count at this level
        - The key of the parent to this entry, or '' if a top element
        - A list of keys of the child elements of the entry, in sorted order

        Any DataModelItem or DataItem with a blank `metric_name` is omitted. Additional
        ReportEntry items may be added, for example to represent the cleaned data.
        """
        # Pass 1: Gather report items and sorting data.
        # The sorting data:
        # clean_group_keys - what DataItems have .clean_group set
        # count_by_key - the pre-computed count from .counts
        # model_keys - the order models appeared in .data_specification
        # report_items - the items that have a .metric_name
        clean_group_keys: dict[CLEAN_GROUP_T, set[str]] = {
            "ok": set(),
            "needs_cleaning": set(),
        }
        count_by_key: dict[str, int] = {}
        model_keys: list[str] = []
        report_items: dict[str, ReportItem] = {}
        for key, data_item in self.data_items.items():
            if not (metric_name := data_item.metric_name):
                continue
            report_items[key] = data_item
            if _KEY_SEP in key:
                # Handle DataItem
                if not isinstance(data_item, DataItem):  # pragma: no cover
                    raise Exception(
                        f"For key '{key}', expected a DataItem, got {data_item!r}"
                    )
                model_key, _ = key.split(_KEY_SEP, 1)
                count_by_key[key] = self.counts[model_key][metric_name]
                if data_item.clean_group:
                    clean_group_keys[data_item.clean_group].add(key)
            else:
                # Handle DataModelItem
                if not isinstance(data_item, DataModelItem):  # pragma: no cover
                    raise Exception(
                        f"For key '{key}', expected a DataModelItem, got {data_item!r}"
                    )
                model_keys.append(key)
                count_by_key[key] = self.counts[key]["all"]

        # Pass 1.2: Created CleanedItem records
        for key in clean_group_keys["needs_cleaning"]:
            model_key, _ = key.split(_KEY_SEP, 1)
            try:
                clean_count = self.counts[model_key][_CLEANED_METRIC_NAME]
            except KeyError:
                continue
            clean_item = CleanedItem(
                clean_count, report_name=self._cleaned_report_name[model_key]
            )
            clean_key = f"{key}{_KEY_SEP}cleaned"
            count_by_key[clean_key] = clean_count
            report_items[clean_key] = clean_item

        # Pass 2: Create index parts for sorting
        # This determines the sort order between siblings
        _INDEX_PART = tuple[int, int, str]
        sort_index_part: dict[str, _INDEX_PART] = {}
        for key, item in report_items.items():
            # First, sort by if this or a descendant has a cleaning_order
            # None, then ok, then needs_cleaning, then both
            has_ok_descendant = any(
                ok_key.startswith(key) for ok_key in clean_group_keys["ok"]
            )
            has_nc_descendant = any(
                nc_key.startswith(key) for nc_key in clean_group_keys["needs_cleaning"]
            )
            cleaning_order = (1 if has_ok_descendant else 0) + (
                2 if has_nc_descendant else 0
            )

            # Next, sort by negation
            # No negation then negation
            if _KEY_SEP in key:
                key_part = key.rsplit(_KEY_SEP, 1)[1]
            else:
                key_part = key
            if key_part.startswith(_NEGATE_PREFIX):
                neg_order = 1
                key_part = key_part[1:]
            else:
                neg_order = 0

            # Finally, sort by name
            sort_index_part[key] = (cleaning_order, neg_order, key_part)

        # Pass 3: Create the sort index
        # Short paths come before long paths
        # Next, use the index part to sort
        _INDEX_FULL = tuple[int, int, tuple[_INDEX_PART, ...]]
        sort_index: dict[str, _INDEX_FULL] = {}
        for key in report_items.keys():
            key_parts = key.split(_KEY_SEP)
            index_parts: list[_INDEX_PART] = []
            model_index = -1
            while key_parts:
                if len(key_parts) == 1:
                    model_index = model_keys.index(key_parts[0])
                subkey = _KEY_SEP.join(key_parts)
                index_parts.insert(0, sort_index_part[subkey])
                key_parts.pop()
            sort_index[key] = (model_index, len(index_parts), tuple(index_parts))

        # Pass 4: Return the sort dict of ReportEntries
        def get_sort_index(key: str) -> _INDEX_FULL:
            return sort_index[key]

        reports: dict[str, ReportEntry] = {}
        for key in sorted(report_items, key=get_sort_index):
            item = report_items[key]
            reports[key] = ReportEntry(
                item, count_by_key[key], key.count(_KEY_SEP) + 1, []
            )
            if _KEY_SEP in key:
                parent_key = key.rsplit(_KEY_SEP, 1)[0]
                reports[parent_key].child_keys.append(key)
        return reports

    def markdown_report(self) -> str:
        """Return Markdown-formatted report of issues found and (maybe) fixed."""

        lines: list[str] = []
        report_entries = self.get_report_entries()

        # Get the maximum length of children for each entry with children
        max_len_children: dict[str, int] = {}
        for key, entry in report_entries.items():
            if entry.child_keys:
                children = [report_entries[subkey] for subkey in entry.child_keys]
                max_len_children[key] = max(
                    len(child.item.report_name or "") for child in children
                )

        # Output the markdown lines
        for key, entry in report_entries.items():
            if isinstance(entry.item, DataModelItem):
                lines.append(f"{entry.item.report_name}:")
                lines.append(f"  All: {entry.count}")
            else:
                parent_key, _ = key.rsplit(_KEY_SEP, 1)
                parent_entry = report_entries[parent_key]
                if parent_entry.count <= 0:
                    continue
                indent = "  " * entry.depth
                max_len = max_len_children[parent_key]
                lines.append(
                    f"{indent}{entry.item.report_name:<{max_len}}:"
                    f" {self._as_percent(entry.count, parent_entry.count)}"
                )
        return "\n".join(lines)

    @staticmethod
    def _as_percent(part: int, whole: int) -> str:
        """Return value followed by percent of whole, like '5 ( 30.0%)'"""
        if whole <= 0:
            raise ValueError(f"whole ({whole}) can not be less than 0")
        if part < 0:
            raise ValueError(f"part ({part}) can not be negative")
        len_whole = len(str(whole))
        return f"{part:{len_whole}d} ({part / whole:6.1%})"


class CleanerTask(DataIssueTask):
    """Base class for tasks that can clean up detected issues."""

    can_clean = True

    def clean(self) -> int:
        """Clean the detected items, and update counts["summary"]"""
        summary = self.counts["summary"]
        if not self._cleaned:
            summary[_CLEANED_METRIC_NAME] = self._clean()
            self._cleaned = True
        return summary[_CLEANED_METRIC_NAME]

    def _clean(self) -> int:
        """Call the specified cleaners."""
        counts = self.counts
        cleanup_data = self.cleanup_data
        total_cleaned = 0
        for model_name, metric_name in cleanup_data.items():
            clean_item = self.data_items[metric_name]
            cleaner = getattr(self, f"clean_{model_name}")
            count = cleaner(clean_item.get_queryset())
            counts[model_name][_CLEANED_METRIC_NAME] = count
            total_cleaned += count
        return total_cleaned