retoorandClaude Opus 5.5
ddf09bea88
FastAPI service on 127.0.0.1:9922 that monitors user-chosen directories with pruned inotify watches, stores versions in a local content-addressed spool indexed in SQLite, coalesces bursts (first + last), and offers history, diff and dry-run-first bulk restore. Includes a CLI with a systemd user unit installer. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
73 lines
2.4 KiB
Python
73 lines
2.4 KiB
Python
"""Text diffs between stored versions or a version and the working copy."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import difflib
|
|
import re
|
|
from typing import Any
|
|
|
|
_WS = re.compile(r"\s+")
|
|
|
|
|
|
def _lines(content: bytes) -> list[str]:
|
|
return content.decode("utf-8", errors="replace").splitlines(keepends=True)
|
|
|
|
|
|
def _normalize(line: str, ignore_whitespace: bool, ignore_eol: bool) -> str:
|
|
if ignore_whitespace:
|
|
return _WS.sub(" ", line).strip()
|
|
if ignore_eol:
|
|
return line.rstrip("\r\n")
|
|
return line
|
|
|
|
|
|
def hunks(
|
|
a: bytes,
|
|
b: bytes,
|
|
context: int = 3,
|
|
ignore_whitespace: bool = False,
|
|
ignore_eol: bool = False,
|
|
) -> list[dict[str, Any]]:
|
|
a_lines, b_lines = _lines(a), _lines(b)
|
|
a_norm = [_normalize(x, ignore_whitespace, ignore_eol) for x in a_lines]
|
|
b_norm = [_normalize(x, ignore_whitespace, ignore_eol) for x in b_lines]
|
|
matcher = difflib.SequenceMatcher(None, a_norm, b_norm, autojunk=False)
|
|
result = []
|
|
for group in matcher.get_grouped_opcodes(context):
|
|
first, last = group[0], group[-1]
|
|
lines = []
|
|
for tag, i1, i2, j1, j2 in group:
|
|
if tag == "equal":
|
|
lines += [{"op": " ", "text": x} for x in a_lines[i1:i2]]
|
|
continue
|
|
if tag in ("replace", "delete"):
|
|
lines += [{"op": "-", "text": x} for x in a_lines[i1:i2]]
|
|
if tag in ("replace", "insert"):
|
|
lines += [{"op": "+", "text": x} for x in b_lines[j1:j2]]
|
|
result.append({
|
|
"old_start": first[1] + 1,
|
|
"old_lines": last[2] - first[1],
|
|
"new_start": first[3] + 1,
|
|
"new_lines": last[4] - first[3],
|
|
"lines": lines,
|
|
})
|
|
return result
|
|
|
|
|
|
def unified(hunk_list: list[dict[str, Any]], from_label: str, to_label: str) -> str:
|
|
if not hunk_list:
|
|
return ""
|
|
out = [f"--- {from_label}\n", f"+++ {to_label}\n"]
|
|
for h in hunk_list:
|
|
out.append(f"@@ -{h['old_start']},{h['old_lines']} +{h['new_start']},{h['new_lines']} @@\n")
|
|
for line in h["lines"]:
|
|
text = line["text"]
|
|
out.append(line["op"] + text + ("" if text.endswith("\n") else "\n\\ No newline at end of file\n"))
|
|
return "".join(out)
|
|
|
|
|
|
def side_by_side_html(a: bytes, b: bytes, from_label: str, to_label: str, context: int = 3) -> str:
|
|
return difflib.HtmlDiff(wrapcolumn=100).make_file(
|
|
_lines(a), _lines(b), from_label, to_label, context=True, numlines=context
|
|
)
|