saksham
1a4c80958f
FastAPI + Celery + Next.js + Postgres/Redis app with company monitoring, source collection, LLM-based change analysis, enrichment, and account security (Turnstile, escalating lockout, email verification).
57 lines
1.8 KiB
Python
57 lines
1.8 KiB
Python
"""Layer 3: bounded text diff.
|
|
|
|
Runs on noise-stripped text (see noise_filters.py) so navigation/cookie/
|
|
timestamp churn doesn't register as a change. Bounded: only a capped number
|
|
of added/removed lines are kept, so a full page rewrite doesn't produce an
|
|
unbounded diff blob for storage or LLM consumption later.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import difflib
|
|
from dataclasses import dataclass, field
|
|
|
|
from app.change_detection.noise_filters import strip_noise
|
|
|
|
_MAX_DIFF_LINES = 40
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TextDiffResult:
|
|
diff_ratio: float # 0.0 = identical, 1.0 = completely different
|
|
added_lines: list[str] = field(default_factory=list)
|
|
removed_lines: list[str] = field(default_factory=list)
|
|
|
|
@property
|
|
def is_identical(self) -> bool:
|
|
return self.diff_ratio == 0.0
|
|
|
|
|
|
def bounded_text_diff(previous_text: str, current_text: str) -> TextDiffResult:
|
|
previous_clean = strip_noise(previous_text or "")
|
|
current_clean = strip_noise(current_text or "")
|
|
|
|
if previous_clean == current_clean:
|
|
return TextDiffResult(diff_ratio=0.0)
|
|
|
|
previous_lines = [line for line in previous_clean.splitlines() if line.strip()]
|
|
current_lines = [line for line in current_clean.splitlines() if line.strip()]
|
|
|
|
matcher = difflib.SequenceMatcher(a=previous_lines, b=current_lines, autojunk=False)
|
|
similarity = matcher.ratio()
|
|
diff_ratio = round(1.0 - similarity, 4)
|
|
|
|
added: list[str] = []
|
|
removed: list[str] = []
|
|
for tag, i1, i2, j1, j2 in matcher.get_opcodes():
|
|
if tag in ("replace", "delete"):
|
|
removed.extend(previous_lines[i1:i2])
|
|
if tag in ("replace", "insert"):
|
|
added.extend(current_lines[j1:j2])
|
|
|
|
return TextDiffResult(
|
|
diff_ratio=diff_ratio,
|
|
added_lines=added[:_MAX_DIFF_LINES],
|
|
removed_lines=removed[:_MAX_DIFF_LINES],
|
|
)
|