"""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], )