mdpolish/src/mdpolish/edits.py

"""Pure validation and atomic application of exact text edits."""

from __future__ import annotations

from dataclasses import dataclass

from mdpolish.models import (
    Change,
    DocumentSnapshot,
    ModifierInfo,
    ProposalReference,
    ProposedChange,
    TextEdit,
)


class EditValidationError(ValueError):
    """A modifier edit batch cannot be applied safely."""


@dataclass(frozen=True, slots=True)
class AppliedBatch:
    """The new snapshot and audit entries from one atomic modifier batch."""

    snapshot: DocumentSnapshot
    changes: tuple[Change, ...]


@dataclass(frozen=True, slots=True)
class _IndexedEdit:
    proposal_index: int
    edit_index: int
    reason: str
    edit: TextEdit


def _edit_order_key(item: _IndexedEdit) -> tuple[int, int, int, int]:
    edit = item.edit
    return (edit.span.start, edit.span.end, item.proposal_index, item.edit_index)


def _ordered_indexed_edits(
    indexed_edits: tuple[_IndexedEdit, ...],
    *,
    reverse: bool = False,
) -> tuple[_IndexedEdit, ...]:
    """返回执行与评审重放共用的唯一编辑顺序。"""
    return tuple(sorted(indexed_edits, key=_edit_order_key, reverse=reverse))


def _apply_validated_edits(
    snapshot: DocumentSnapshot,
    indexed_edits: tuple[_IndexedEdit, ...],
) -> DocumentSnapshot:
    """应用已经验证的批次, 但不生成审计记录。"""
    if not indexed_edits:
        return snapshot

    markdown = snapshot.markdown
    for item in _ordered_indexed_edits(indexed_edits, reverse=True):
        edit = item.edit
        markdown = markdown[: edit.span.start] + edit.replacement + markdown[edit.span.end :]
    return DocumentSnapshot(markdown)


def _edits_conflict(left: TextEdit, right: TextEdit) -> bool:
    left_span = left.span
    right_span = right.span

    if left_span.is_empty and right_span.is_empty:
        return left_span.start == right_span.start
    if left_span.is_empty:
        return right_span.start <= left_span.start <= right_span.end
    if right_span.is_empty:
        return left_span.start <= right_span.start <= left_span.end
    return max(left_span.start, right_span.start) < min(left_span.end, right_span.end)


def validate_modifier_batch(
    snapshot: DocumentSnapshot,
    proposals: tuple[ProposedChange, ...],
) -> tuple[_IndexedEdit, ...]:
    """Validate a complete modifier batch without changing the snapshot."""
    if not isinstance(snapshot, DocumentSnapshot):
        raise TypeError("snapshot must be a DocumentSnapshot")
    if not isinstance(proposals, tuple):
        raise EditValidationError("modifier proposals must be a tuple")

    indexed_edits: list[_IndexedEdit] = []
    seen_edits: set[TextEdit] = set()
    for proposal_index, proposal in enumerate(proposals):
        if not isinstance(proposal, ProposedChange):
            raise EditValidationError("a modifier batch must contain only ProposedChange values")
        if proposal.snapshot_sha256 != snapshot.sha256:
            raise EditValidationError("a proposal targets a stale document snapshot")
        for edit_index, edit in enumerate(proposal.edits):
            if edit.snapshot_sha256 != snapshot.sha256:
                raise EditValidationError("an edit targets a stale document snapshot")
            if edit.span.end > len(snapshot.markdown):
                raise EditValidationError("an edit span is outside the document snapshot")
            if snapshot.markdown[edit.span.start : edit.span.end] != edit.expected_text:
                raise EditValidationError("an edit's expected_text does not match the document snapshot")
            if edit in seen_edits:
                raise EditValidationError("a modifier batch contains a duplicate edit")
            seen_edits.add(edit)
            indexed_edits.append(
                _IndexedEdit(
                    proposal_index=proposal_index,
                    edit_index=edit_index,
                    reason=proposal.reason,
                    edit=edit,
                )
            )

    for left_index, left in enumerate(indexed_edits):
        for right in indexed_edits[left_index + 1 :]:
            if _edits_conflict(left.edit, right.edit):
                raise EditValidationError("a modifier batch contains conflicting edit ranges")

    return tuple(indexed_edits)


def apply_modifier_batch(
    snapshot: DocumentSnapshot,
    proposals: tuple[ProposedChange, ...],
    modifier: ModifierInfo,
    modifier_position: int,
) -> AppliedBatch:
    """Atomically apply one fully validated modifier batch."""
    indexed_edits = validate_modifier_batch(snapshot, proposals)
    if not indexed_edits:
        return AppliedBatch(snapshot=snapshot, changes=())

    updated_snapshot = _apply_validated_edits(snapshot, indexed_edits)
    report_order = _ordered_indexed_edits(indexed_edits)
    changes = tuple(
        Change(
            modifier_id=modifier.modifier_id,
            modifier_version=modifier.version,
            modifier_position=modifier_position,
            proposal_ref=ProposalReference(
                modifier_position=modifier_position,
                snapshot_sha256=snapshot.sha256,
                proposal_index=item.proposal_index,
            ),
            edit_index=item.edit_index,
            reason=item.reason,
            span=item.edit.span,
            before=item.edit.expected_text,
            after=item.edit.replacement,
            before_sha256=snapshot.sha256,
            after_sha256=updated_snapshot.sha256,
        )
        for item in report_order
    )
    return AppliedBatch(snapshot=updated_snapshot, changes=changes)