src/gcmd_classifier/classification/redundancy.py

"""Redundancy removal for validated GCMD classification records."""

from __future__ import annotations

from pydantic import BaseModel, ConfigDict, Field

from gcmd_classifier.models import ClassificationRecord, OutputWarning
from gcmd_classifier.vocabulary.index import VocabularyIndex

_STAGE = "redundancy_removal"


class RedundancyResult(BaseModel):
    """Non-redundant classifications plus structured redundancy diagnostics."""

    model_config = ConfigDict(extra="forbid", frozen=True)

    classifications: tuple[ClassificationRecord, ...] = Field(default_factory=tuple)
    removed: tuple[ClassificationRecord, ...] = Field(default_factory=tuple)
    warnings: tuple[OutputWarning, ...] = Field(default_factory=tuple)

    @property
    def retained_count(self) -> int:
        """Number of retained classifications."""
        return len(self.classifications)

    @property
    def removed_count(self) -> int:
        """Number of removed redundant classifications."""
        return len(self.removed)


def remove_redundant_classifications(
    records: tuple[ClassificationRecord, ...] | list[ClassificationRecord],
    vocabulary: VocabularyIndex,
) -> RedundancyResult:
    """Remove exact duplicates and branch-provenance ancestor redundancy."""
    retained: list[ClassificationRecord] = []
    removed: list[ClassificationRecord] = []
    warnings: list[OutputWarning] = []
    seen_uuids: dict[str, ClassificationRecord] = {}
    seen_paths: dict[str, ClassificationRecord] = {}

    for record in records:
        if record.UUID in seen_uuids:
            removed.append(record)
            warnings.append(
                _warning(
                    "DUPLICATE_UUID_REMOVED",
                    "Duplicate UUID classification was removed.",
                    record=record,
                    details={"duplicate_of_branch_id": seen_uuids[record.UUID].branch_id},
                )
            )
            continue
        if record.canonical_path in seen_paths:
            removed.append(record)
            warnings.append(
                _warning(
                    "DUPLICATE_CANONICAL_PATH_REMOVED",
                    "Duplicate canonical-path classification was removed.",
                    record=record,
                    details={"duplicate_of_branch_id": seen_paths[record.canonical_path].branch_id},
                )
            )
            continue
        seen_uuids[record.UUID] = record
        seen_paths[record.canonical_path] = record
        retained.append(record)

    retained, ancestry_removed, ancestry_warnings = _remove_branch_ancestors(retained, vocabulary)
    removed.extend(ancestry_removed)
    warnings.extend(ancestry_warnings)
    return RedundancyResult(
        classifications=tuple(retained),
        removed=tuple(removed),
        warnings=tuple(warnings),
    )


def _remove_branch_ancestors(
    records: list[ClassificationRecord],
    vocabulary: VocabularyIndex,
) -> tuple[list[ClassificationRecord], list[ClassificationRecord], list[OutputWarning]]:
    to_remove: set[int] = set()
    removed: list[ClassificationRecord] = []
    warnings: list[OutputWarning] = []

    for ancestor_index, ancestor in enumerate(records):
        for descendant_index, descendant in enumerate(records):
            if ancestor_index == descendant_index:
                continue
            if not vocabulary.is_ancestor(ancestor.UUID, descendant.UUID):
                continue
            relation = _branch_relation(ancestor.branch_id, descendant.branch_id)
            if relation == "same_lineage":
                if ancestor_index not in to_remove:
                    to_remove.add(ancestor_index)
                    removed.append(ancestor)
                    warnings.append(
                        _warning(
                            "ANCESTOR_REMOVED_SAME_BRANCH",
                            (
                                "Ancestor classification was removed because a deeper descendant "
                                "exists in the same branch lineage."
                            ),
                            record=ancestor,
                            details={
                                "descendant_uuid": descendant.UUID,
                                "descendant_branch_id": descendant.branch_id,
                            },
                        )
                    )
            elif relation == "independent":
                warnings.append(
                    _warning(
                        "ANCESTOR_PRESERVED_INDEPENDENT_BRANCH",
                        (
                            "Ancestor classification was preserved because branch provenance "
                            "indicates an independent classification decision."
                        ),
                        record=ancestor,
                        details={
                            "descendant_uuid": descendant.UUID,
                            "descendant_branch_id": descendant.branch_id,
                        },
                    )
                )
            else:
                warnings.append(
                    _warning(
                        "ANCESTOR_PRESERVED_INSUFFICIENT_PROVENANCE",
                        (
                            "Ancestor classification was preserved because branch provenance "
                            "was insufficient to prove redundancy."
                        ),
                        record=ancestor,
                        details={
                            "descendant_uuid": descendant.UUID,
                            "descendant_branch_id": descendant.branch_id,
                        },
                    )
                )
    retained = [record for index, record in enumerate(records) if index not in to_remove]
    return retained, removed, warnings


def _branch_relation(ancestor_branch_id: str | None, descendant_branch_id: str | None) -> str:
    if not ancestor_branch_id or not descendant_branch_id:
        return "insufficient"
    if ancestor_branch_id == descendant_branch_id:
        return "same_lineage"
    if descendant_branch_id.startswith(f"{ancestor_branch_id}/"):
        return "same_lineage"
    if ancestor_branch_id.startswith(f"{descendant_branch_id}/"):
        return "same_lineage"
    return "independent"


def _warning(
    code: str,
    message: str,
    *,
    record: ClassificationRecord,
    details: dict | None = None,
) -> OutputWarning:
    warning_details = {
        "UUID": record.UUID,
        "canonical_path": record.canonical_path,
        "branch_id": record.branch_id,
    }
    if details:
        warning_details.update(details)
    return OutputWarning(
        code=code,
        message=message,
        stage=_STAGE,
        details=warning_details,
    )