AIRegulation-DocAnalysis/backend/app/infrastructure/perception/regulation_differ.py

"""Deterministic change detection between two versions of a regulation.

This module deliberately contains no LLM call, no network access, and no
embedding lookup. It exists because the previous implementation decided whether
a paragraph had changed by comparing embedding cosine similarity against a 0.85
threshold, which is blind to exactly the edits that matter in regulation.
Measured against the deployed text-embedding-v3 gateway, tightening a braking
limit from 30米 to 20米 scores 0.9153 and relaxing 应当 to 宜 scores 0.9162 —
both far above the threshold, both undetected — while an entirely unrelated
clause scores 0.6862 and is the only thing that fires. Cosine is scale
invariant, so it cannot represent a change in magnitude or certainty
(arXiv:2403.05440, ACM Web Conference 2024); no threshold recovers the signal.

The replacement is the production consensus for legal text: align paragraphs
with a longest-common-subsequence matcher, run a literal character diff on the
aligned pairs, and let cheap deterministic rules decide whether a change is
significant enough to spend an LLM call classifying.
"""

from __future__ import annotations

import re
import unicodedata
from dataclasses import dataclass, field

from diff_match_patch import diff_match_patch
from difflib import SequenceMatcher

from app.config.settings import settings

# Chinese regulatory drafting uses a small, near-unambiguous set of deontic
# markers, so a regex pre-pass identifies legally significant edits without an
# LLM. Adding or removing any of these changes what the provision compels.
_DEONTIC_PATTERN = re.compile(r"应当|须|禁止|不得|可以|允许|宜")

# Matches digit runs including decimals, so "30" -> "20" and "0.85" -> "0.9"
# are both treated as numeric changes.
_NUMBER_PATTERN = re.compile(r"\d+(?:\.\d+)?")

# diff_match_patch operation codes.
_DMP_DELETE = -1
_DMP_INSERT = 1
_DMP_EQUAL = 0


@dataclass(frozen=True)
class ParagraphChange:
    """One detected difference between the old and new version of a regulation.

    `needs_llm` is the gate: it records whether this change is worth the cost of
    an LLM classification call. The deterministic flags that drive it are kept
    on the record so downstream code can act on them even when the LLM call
    fails or is skipped.
    """

    change_type: str
    old_text: str
    new_text: str
    numeric_changed: bool
    deontic_changed: bool
    change_ratio: float
    needs_llm: bool
    # (op, text) pairs from diff_match_patch, for rendering a redline view.
    diff_spans: list[tuple[int, str]] = field(default_factory=list)


def _split_paragraphs(text: str) -> list[str]:
    """Split regulation text into comparable units, dropping blank lines.

    ponytail: newline splitting, not clause parsing. Upgrade to 第X条 / X.X.X
    segmentation only if paragraph granularity proves too coarse in practice.
    """
    return [line.strip() for line in (text or "").split("\n") if line.strip()]


def _numbers_differ(old: str, new: str) -> bool:
    """Report whether the two spans contain a different sequence of numbers."""
    return _NUMBER_PATTERN.findall(old) != _NUMBER_PATTERN.findall(new)


def _deontic_differs(old: str, new: str) -> bool:
    """Report whether obligation markers were added, removed, or swapped."""
    return sorted(_DEONTIC_PATTERN.findall(old)) != sorted(_DEONTIC_PATTERN.findall(new))


def _is_cosmetic(spans: list[tuple[int, str]]) -> bool:
    """Report whether the edit touched nothing but punctuation and whitespace.

    A change ratio alone cannot answer this for Chinese regulation text. Clauses
    run 20-60 characters, so deleting a single 。 is a 4% change and clears any
    threshold low enough to still catch real edits in longer paragraphs. Testing
    what actually changed is both cheaper and exact.
    """
    changed = "".join(text for op, text in spans if op != _DMP_EQUAL)
    # Unicode categories P (punctuation), Z (separator) and C (control) cover
    # Chinese and ASCII punctuation plus every flavour of whitespace.
    return all(unicodedata.category(char)[0] in {"P", "Z", "C"} for char in changed)


class RegulationDiffer:
    """Align two regulation versions and classify what changed, without an LLM."""

    def __init__(self, min_change_ratio: float | None = None) -> None:
        """Store the gate threshold, defaulting to the configured value.

        The explicit argument exists so tests never depend on the deployed .env.
        """
        self._min_change_ratio = (
            settings.perception_diff_min_change_ratio
            if min_change_ratio is None
            else min_change_ratio
        )
        self._dmp = diff_match_patch()

    def diff(self, old_text: str, new_text: str) -> list[ParagraphChange]:
        """Return every changed paragraph between two versions.

        Unchanged paragraphs are not returned. An empty old version means there
        is no baseline to compare against — the caller's first crawl — so no
        changes are reported rather than the whole document being called new.
        """
        old_paras = _split_paragraphs(old_text)
        new_paras = _split_paragraphs(new_text)

        if not old_paras or not new_paras:
            return []

        # autojunk=False is load-bearing: the default treats any element
        # appearing in over 1% of a sequence of 200+ items as junk, and
        # regulations repeat boilerplate paragraphs that alignment depends on
        # as anchors.
        matcher = SequenceMatcher(None, old_paras, new_paras, autojunk=False)

        changes: list[ParagraphChange] = []
        for tag, i1, i2, j1, j2 in matcher.get_opcodes():
            if tag == "equal":
                continue
            if tag == "insert":
                changes.extend(self._added(p) for p in new_paras[j1:j2])
            elif tag == "delete":
                changes.extend(self._removed(p) for p in old_paras[i1:i2])
            elif tag == "replace":
                changes.extend(self._replaced(old_paras[i1:i2], new_paras[j1:j2]))

        return changes

    def _added(self, paragraph: str) -> ParagraphChange:
        """Build a record for a provision present only in the new version."""
        return ParagraphChange(
            change_type="added",
            old_text="",
            new_text=paragraph,
            numeric_changed=False,
            deontic_changed=bool(_DEONTIC_PATTERN.search(paragraph)),
            change_ratio=1.0,
            # A new provision always carries new obligations, so it is always
            # worth classifying.
            needs_llm=True,
            diff_spans=[(_DMP_INSERT, paragraph)],
        )

    def _removed(self, paragraph: str) -> ParagraphChange:
        """Build a record for a provision dropped from the new version."""
        return ParagraphChange(
            change_type="removed",
            old_text=paragraph,
            new_text="",
            numeric_changed=False,
            deontic_changed=bool(_DEONTIC_PATTERN.search(paragraph)),
            change_ratio=1.0,
            needs_llm=True,
            diff_spans=[(_DMP_DELETE, paragraph)],
        )

    def _replaced(self, old_block: list[str], new_block: list[str]) -> list[ParagraphChange]:
        """Compare a run of rewritten paragraphs pairwise, reporting the remainder.

        SequenceMatcher emits `replace` for a whole run at once, and the two
        sides may differ in length. Pairing by position within the run is safe
        here because alignment has already established that this run as a whole
        corresponds; any surplus on either side is a genuine insertion or
        deletion.
        """
        results: list[ParagraphChange] = []
        for index in range(max(len(old_block), len(new_block))):
            if index >= len(old_block):
                results.append(self._added(new_block[index]))
            elif index >= len(new_block):
                results.append(self._removed(old_block[index]))
            else:
                results.append(self._modified(old_block[index], new_block[index]))
        return results

    def _modified(self, old: str, new: str) -> ParagraphChange:
        """Character-diff an aligned pair and decide whether it warrants an LLM call."""
        spans = self._dmp.diff_main(old, new)
        # Merges single-character edits into human-meaningful chunks so the
        # redline view and the change ratio both reflect real edits.
        self._dmp.diff_cleanupSemantic(spans)

        changed_chars = sum(len(text) for op, text in spans if op != _DMP_EQUAL)
        denominator = max(len(old), len(new), 1)
        change_ratio = changed_chars / denominator

        numeric_changed = _numbers_differ(old, new)
        deontic_changed = _deontic_differs(old, new)

        # A changed limit or obligation marker is always significant no matter
        # how few characters moved. Everything else must be substantive and
        # clear the ratio gate to be worth a model call.
        significant = numeric_changed or deontic_changed or (
            not _is_cosmetic(spans) and change_ratio >= self._min_change_ratio
        )

        return ParagraphChange(
            change_type="modified",
            old_text=old,
            new_text=new,
            numeric_changed=numeric_changed,
            deontic_changed=deontic_changed,
            change_ratio=change_ratio,
            needs_llm=significant,
            diff_spans=[(op, text) for op, text in spans],
        )