#!/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:]))