392 lines
13 KiB
Python
392 lines
13 KiB
Python
#!/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:]))
|