null: diffkit - unified-diff parse+apply engine (30/30 checks, 3 mutants killed, 8/8 real git diffs byte-exact)
This commit is contained in:
null committed
1 parent
a4698efd73
commit
223d9410ad
7 files changed
+990
No files matched your search
@@ -0,0 +1,391 @@
|
||||
#!/usr/bin/env python3
|
||||
"""diffkit -- dependency-free unified-diff parse + apply engine.
|
||||
|
||||
Design notes
|
||||
------------
|
||||
* Text is modelled as a list of lines WITH their terminators, split only on "\\n"
|
||||
(not str.splitlines, which also splits on \\v \\f \\x1c-\\x1e \\x85 \\u2028 \\u2029).
|
||||
That makes "no newline at end of file" representable exactly: the final element
|
||||
simply has no trailing "\\n".
|
||||
* Hunks are located by exact match of their old-side line sequence, nearest to the
|
||||
position the hunk header claims, with GNU-style fuzz (drop up to N leading /
|
||||
trailing context lines) and unbounded-or-bounded offset drift.
|
||||
* Malformed hunks -- header counts that disagree with the body -- are rejected,
|
||||
never silently guessed at.
|
||||
|
||||
Public API
|
||||
----------
|
||||
parse_unified_diff(text) -> list[FilePatch]
|
||||
apply_file_patch(patch, text, ...) -> str
|
||||
apply_patch(patches, sources, ...) -> dict[str, str | None]
|
||||
apply_one(diff_text, base_text) -> str
|
||||
|
||||
CLI: python3 diffkit.py [-p N] [-F N] [--fuzz N] [patchfile]
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import sys
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Iterable, Mapping, Sequence
|
||||
|
||||
__all__ = [
|
||||
"Hunk",
|
||||
"FilePatch",
|
||||
"DiffParseError",
|
||||
"DiffApplyError",
|
||||
"split_lines",
|
||||
"join_lines",
|
||||
"parse_unified_diff",
|
||||
"apply_file_patch",
|
||||
"apply_patch",
|
||||
"apply_one",
|
||||
]
|
||||
|
||||
DEFAULT_FUZZ = 2 # GNU patch's default maximum fuzz factor
|
||||
|
||||
|
||||
class DiffParseError(Exception):
|
||||
"""The diff text is not well-formed unified diff."""
|
||||
|
||||
|
||||
class DiffApplyError(Exception):
|
||||
"""A well-formed hunk could not be located in the source text."""
|
||||
|
||||
|
||||
_HUNK_RE = re.compile(r"^@@ -(\d+)(?:,(\d+))? \+(\d+)(?:,(\d+))? @@(?: ?(.*))?$")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# text <-> line model
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
def split_lines(text: str) -> list[str]:
|
||||
"""Split on "\\n" only, keeping terminators. "" -> []."""
|
||||
if not text:
|
||||
return []
|
||||
parts = text.split("\n")
|
||||
out = [p + "\n" for p in parts[:-1]]
|
||||
if parts[-1]:
|
||||
out.append(parts[-1]) # final line, no terminator
|
||||
return out
|
||||
|
||||
|
||||
def join_lines(lines: Sequence[str]) -> str:
|
||||
return "".join(lines)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# data model
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
@dataclass
|
||||
class Hunk:
|
||||
old_start: int
|
||||
old_count: int
|
||||
new_start: int
|
||||
new_count: int
|
||||
lines: list[tuple[str, str]] = field(default_factory=list) # (kind, content-with-terminator)
|
||||
section: str = ""
|
||||
|
||||
@property
|
||||
def old_seq(self) -> list[str]:
|
||||
return [c for k, c in self.lines if k in (" ", "-")]
|
||||
|
||||
@property
|
||||
def new_seq(self) -> list[str]:
|
||||
return [c for k, c in self.lines if k in (" ", "+")]
|
||||
|
||||
def validate(self) -> None:
|
||||
o, n = self.old_seq, self.new_seq
|
||||
if len(o) != self.old_count or len(n) != self.new_count:
|
||||
raise DiffApplyError(
|
||||
"hunk header @@ -%d,%d +%d,%d @@ disagrees with its body "
|
||||
"(old lines=%d, new lines=%d)"
|
||||
% (self.old_start, self.old_count, self.new_start, self.new_count,
|
||||
len(o), len(n))
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class FilePatch:
|
||||
old_path: str
|
||||
new_path: str
|
||||
hunks: list[Hunk] = field(default_factory=list)
|
||||
|
||||
|
||||
def _clean_path(raw: str) -> str:
|
||||
"""'a/foo.txt\t2026-10-20 12:00' -> 'a/foo.txt'; strip surrounding quotes."""
|
||||
s = raw.split("\t")[0].strip()
|
||||
if len(s) >= 2 and s.startswith('"') and s.endswith('"'):
|
||||
s = s[1:-1]
|
||||
return s
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# parser
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
def parse_unified_diff(text: str) -> list[FilePatch]:
|
||||
raw = text.split("\n")
|
||||
if raw and raw[-1] == "":
|
||||
raw.pop() # artefact of the trailing newline
|
||||
|
||||
patches: list[FilePatch] = []
|
||||
cur: FilePatch | None = None
|
||||
hunk: Hunk | None = None
|
||||
i, n = 0, len(raw)
|
||||
|
||||
def close_hunk() -> None:
|
||||
nonlocal hunk
|
||||
if hunk is not None:
|
||||
assert cur is not None
|
||||
cur.hunks.append(hunk)
|
||||
hunk = None
|
||||
|
||||
while i < n:
|
||||
line = raw[i]
|
||||
|
||||
if line.startswith("--- "):
|
||||
close_hunk()
|
||||
if i + 1 < n and raw[i + 1].startswith("+++ "):
|
||||
cur = FilePatch(_clean_path(line[4:]), _clean_path(raw[i + 1][4:]), [])
|
||||
patches.append(cur)
|
||||
i += 2
|
||||
continue
|
||||
i += 1 # stray '---' line
|
||||
continue
|
||||
|
||||
if line.startswith("@@ "):
|
||||
if cur is None:
|
||||
raise DiffParseError("hunk header with no preceding file header: %r" % line)
|
||||
m = _HUNK_RE.match(line)
|
||||
if not m:
|
||||
raise DiffParseError("malformed hunk header: %r" % line)
|
||||
close_hunk()
|
||||
os_, oc, ns, nc, section = m.groups()
|
||||
hunk = Hunk(
|
||||
int(os_), 1 if oc is None else int(oc),
|
||||
int(ns), 1 if nc is None else int(nc),
|
||||
[], section or "",
|
||||
)
|
||||
i += 1
|
||||
continue
|
||||
|
||||
if hunk is not None:
|
||||
head = line[:1]
|
||||
if head in (" ", "-", "+"):
|
||||
hunk.lines.append((head, line[1:] + "\n"))
|
||||
i += 1
|
||||
continue
|
||||
if head == "\\":
|
||||
if not hunk.lines:
|
||||
raise DiffParseError("no-newline marker with no preceding line")
|
||||
kind, content = hunk.lines[-1]
|
||||
if not content.endswith("\n"):
|
||||
raise DiffParseError(
|
||||
"no-newline marker after a line that already lacks a newline")
|
||||
hunk.lines[-1] = (kind, content[:-1])
|
||||
i += 1
|
||||
continue
|
||||
if line == "":
|
||||
hunk.lines.append((" ", "\n")) # empty context line
|
||||
i += 1
|
||||
continue
|
||||
close_hunk() # anything else ends the hunk; fall through to outer state
|
||||
|
||||
i += 1 # 'diff --git', 'index', 'new file mode', 'Binary files ...' etc.
|
||||
|
||||
close_hunk()
|
||||
return patches
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# apply
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
def _fuzz_variants(hunk: Hunk, max_fuzz: int) -> Iterable[tuple[int, list[str], list[str], int]]:
|
||||
"""Yield (fuzz_level, old_seq, new_seq, expected_index) for fuzz 0..max_fuzz.
|
||||
|
||||
Fuzz f drops up to f leading and f trailing *context* lines from both sides,
|
||||
which is what GNU patch's -F does.
|
||||
"""
|
||||
base_old, base_new = hunk.old_seq, hunk.new_seq
|
||||
|
||||
lead = 0
|
||||
for k, _ in hunk.lines:
|
||||
if k == " ":
|
||||
lead += 1
|
||||
else:
|
||||
break
|
||||
trail = 0
|
||||
for k, _ in reversed(hunk.lines):
|
||||
if k == " ":
|
||||
trail += 1
|
||||
else:
|
||||
break
|
||||
|
||||
expected = hunk.old_start if hunk.old_count == 0 else hunk.old_start - 1
|
||||
|
||||
for f in range(0, max_fuzz + 1):
|
||||
cut_l = min(f, lead)
|
||||
cut_t = min(f, trail)
|
||||
old = base_old[cut_l:len(base_old) - cut_t] if cut_t else base_old[cut_l:]
|
||||
new = base_new[cut_l:len(base_new) - cut_t] if cut_t else base_new[cut_l:]
|
||||
yield f, old, new, expected + cut_l
|
||||
|
||||
|
||||
def _locate(hay: Sequence[str], needle: Sequence[str], expected: int,
|
||||
index_cache: dict[str, list[int]], max_offset: int | None) -> int | None:
|
||||
"""Index of the occurrence of `needle` in `hay` nearest to `expected`, or None."""
|
||||
n = len(needle)
|
||||
if n == 0:
|
||||
return max(0, min(expected, len(hay)))
|
||||
if n > len(hay):
|
||||
return None
|
||||
|
||||
first = needle[0]
|
||||
positions = index_cache.get(first)
|
||||
if positions is None:
|
||||
positions = [j for j, l in enumerate(hay) if l == first]
|
||||
index_cache[first] = positions
|
||||
|
||||
best: int | None = None
|
||||
best_d: int | None = None
|
||||
for j in positions:
|
||||
if j + n > len(hay):
|
||||
continue
|
||||
if list(hay[j:j + n]) != list(needle):
|
||||
continue
|
||||
d = abs(j - expected)
|
||||
if max_offset is not None and d > max_offset:
|
||||
continue
|
||||
if best_d is None or d < best_d or (d == best_d and best is not None and j < best):
|
||||
best, best_d = j, d
|
||||
return best
|
||||
|
||||
|
||||
def apply_hunk(hay: list[str], hunk: Hunk, index_cache: dict[str, list[int]],
|
||||
max_fuzz: int = DEFAULT_FUZZ, max_offset: int | None = None
|
||||
) -> tuple[list[str], int, int]:
|
||||
"""Return (new_lines, fuzz_used, offset_used). Raises DiffApplyError if unplaceable."""
|
||||
hunk.validate()
|
||||
for f, old, new, expected in _fuzz_variants(hunk, max_fuzz):
|
||||
pos = _locate(hay, old, expected, index_cache, max_offset)
|
||||
if pos is not None:
|
||||
return hay[:pos] + list(new) + hay[pos + len(old):], f, pos - expected
|
||||
raise DiffApplyError(
|
||||
"hunk @@ -%d,%d +%d,%d @@ does not apply (fuzz<=%d)"
|
||||
% (hunk.old_start, hunk.old_count, hunk.new_start, hunk.new_count, max_fuzz)
|
||||
)
|
||||
|
||||
|
||||
def apply_file_patch(patch: FilePatch, text: str, *, max_fuzz: int = DEFAULT_FUZZ,
|
||||
max_offset: int | None = None) -> str:
|
||||
hay = split_lines(text)
|
||||
index_cache: dict[str, list[int]] = {}
|
||||
for hunk in patch.hunks:
|
||||
hay, _f, _o = apply_hunk(hay, hunk, index_cache, max_fuzz, max_offset)
|
||||
return join_lines(hay)
|
||||
|
||||
|
||||
def _strip(path: str, n: int) -> str:
|
||||
if path == "/dev/null" or n <= 0:
|
||||
return path
|
||||
parts = path.split("/")
|
||||
return "/".join(parts[n:]) if len(parts) > n else path
|
||||
|
||||
|
||||
def apply_patch(patches: Sequence[FilePatch], sources: Mapping[str, str], *,
|
||||
strip: int = 0, max_fuzz: int = DEFAULT_FUZZ,
|
||||
max_offset: int | None = None) -> dict[str, str | None]:
|
||||
"""Apply to an in-memory {path: text} map.
|
||||
|
||||
Returns {output_path: new_text}, with None marking a deletion. Creation
|
||||
(old == /dev/null) starts from "".
|
||||
"""
|
||||
out: dict[str, str | None] = {}
|
||||
for p in patches:
|
||||
old = _strip(p.old_path, strip)
|
||||
new = _strip(p.new_path, strip)
|
||||
if old == "/dev/null":
|
||||
src = ""
|
||||
else:
|
||||
if old not in sources:
|
||||
raise DiffApplyError("no source supplied for %r" % old)
|
||||
src = sources[old]
|
||||
if new == "/dev/null":
|
||||
out[old] = None
|
||||
else:
|
||||
out[new] = apply_file_patch(p, src, max_fuzz=max_fuzz, max_offset=max_offset)
|
||||
return out
|
||||
|
||||
|
||||
def apply_one(diff_text: str, base_text: str, *, max_fuzz: int = DEFAULT_FUZZ,
|
||||
max_offset: int | None = None) -> str:
|
||||
"""Parse a single-file diff and apply it. Raises if it is not exactly one file."""
|
||||
patches = parse_unified_diff(diff_text)
|
||||
if len(patches) != 1:
|
||||
raise DiffParseError("expected exactly 1 file patch, got %d" % len(patches))
|
||||
return apply_file_patch(patches[0], base_text, max_fuzz=max_fuzz, max_offset=max_offset)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# CLI
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
def _main(argv: list[str]) -> int:
|
||||
strip, max_fuzz, path = 0, DEFAULT_FUZZ, "-"
|
||||
i = 0
|
||||
while i < len(argv):
|
||||
a = argv[i]
|
||||
if a == "-p" and i + 1 < len(argv):
|
||||
strip = int(argv[i + 1]); i += 2; continue
|
||||
if a.startswith("-p") and len(a) > 2:
|
||||
strip = int(a[2:]); i += 1; continue
|
||||
if a in ("-F", "--fuzz") and i + 1 < len(argv):
|
||||
max_fuzz = int(argv[i + 1]); i += 2; continue
|
||||
if a.startswith("-F") and len(a) > 2:
|
||||
max_fuzz = int(a[2:]); i += 1; continue
|
||||
if a in ("-h", "--help"):
|
||||
print(__doc__); return 0
|
||||
path = a; i += 1
|
||||
|
||||
if path == "-":
|
||||
text = sys.stdin.read()
|
||||
else:
|
||||
try:
|
||||
text = open(path, "r", encoding="utf-8", newline="").read()
|
||||
except OSError as e:
|
||||
print("diffkit: cannot read patch file: %s" % e, file=sys.stderr)
|
||||
return 1
|
||||
try:
|
||||
patches = parse_unified_diff(text)
|
||||
except DiffParseError as e:
|
||||
print("diffkit: parse error: %s" % e, file=sys.stderr)
|
||||
return 1
|
||||
|
||||
rc = 0
|
||||
for p in patches:
|
||||
old = _strip(p.old_path, strip)
|
||||
new = _strip(p.new_path, strip)
|
||||
try:
|
||||
src = "" if old == "/dev/null" else open(old, "r", encoding="utf-8", newline="").read()
|
||||
if new == "/dev/null":
|
||||
import os
|
||||
os.remove(old)
|
||||
print("deleted %s" % old)
|
||||
continue
|
||||
dst = apply_file_patch(p, src, max_fuzz=max_fuzz)
|
||||
with open(new, "w", encoding="utf-8", newline="") as fh:
|
||||
fh.write(dst)
|
||||
print("patched %s" % new)
|
||||
except (OSError, DiffApplyError) as e:
|
||||
print("diffkit: %s: %s" % (old, e), file=sys.stderr)
|
||||
rc = 1
|
||||
return rc
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(_main(sys.argv[1:]))
|
||||
Reference in new issue
Block a user