Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/dulwich/patch.py: 9%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# patch.py -- For dealing with packed-style patches.
2# Copyright (C) 2009-2013 Jelmer Vernooij <jelmer@jelmer.uk>
3#
4# SPDX-License-Identifier: Apache-2.0 OR GPL-2.0-or-later
5# Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU
6# General Public License as published by the Free Software Foundation; version 2.0
7# or (at your option) any later version. You can redistribute it and/or
8# modify it under the terms of either of these two licenses.
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS,
12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13# See the License for the specific language governing permissions and
14# limitations under the License.
15#
16# You should have received a copy of the licenses; if not, see
17# <http://www.gnu.org/licenses/> for a copy of the GNU General Public License
18# and <http://www.apache.org/licenses/LICENSE-2.0> for a copy of the Apache
19# License, Version 2.0.
20#
22"""Classes for dealing with git am-style patches.
24These patches are basically unified diffs with some extra metadata tacked
25on.
26"""
28__all__ = [
29 "DEFAULT_DIFF_ALGORITHM",
30 "FIRST_FEW_BYTES",
31 "DiffAlgorithmNotAvailable",
32 "MailinfoResult",
33 "PatchApplicationFailure",
34 "apply_patch_hunks",
35 "apply_patches",
36 "commit_patch_id",
37 "gen_diff_header",
38 "get_summary",
39 "git_am_patch_split",
40 "is_binary",
41 "mailinfo",
42 "parse_patch_message",
43 "patch_filename",
44 "patch_id",
45 "shortid",
46 "unified_diff",
47 "unified_diff_with_algorithm",
48 "write_blob_diff",
49 "write_commit_diff",
50 "write_commit_patch",
51 "write_object_diff",
52 "write_tree_diff",
53]
55import email.message
56import email.parser
57import email.utils
58import os
59import re
60import time
61from collections.abc import Generator, Sequence
62from dataclasses import dataclass
63from difflib import SequenceMatcher
64from typing import (
65 IO,
66 TYPE_CHECKING,
67 BinaryIO,
68 TextIO,
69)
71if TYPE_CHECKING:
72 from .config import Config
73 from .object_store import BaseObjectStore
74 from .repo import Repo
76from .objects import S_ISGITLINK, Blob, Commit, ObjectID, RawObjectID
78FIRST_FEW_BYTES = 8000
80DEFAULT_DIFF_ALGORITHM = "myers"
83class PatchApplicationFailure(Exception):
84 """Raised when a patch does not apply cleanly."""
87class DiffAlgorithmNotAvailable(Exception):
88 """Raised when a requested diff algorithm is not available."""
90 def __init__(self, algorithm: str, install_hint: str = "") -> None:
91 """Initialize exception.
93 Args:
94 algorithm: Name of the unavailable algorithm
95 install_hint: Optional installation hint
96 """
97 self.algorithm = algorithm
98 self.install_hint = install_hint
99 if install_hint:
100 super().__init__(
101 f"Diff algorithm '{algorithm}' requested but not available. {install_hint}"
102 )
103 else:
104 super().__init__(
105 f"Diff algorithm '{algorithm}' requested but not available."
106 )
109def write_commit_patch(
110 f: IO[bytes],
111 commit: "Commit",
112 contents: str | bytes,
113 progress: tuple[int, int],
114 version: str | None = None,
115 encoding: str | None = None,
116) -> None:
117 """Write a individual file patch.
119 Args:
120 f: File-like object to write to
121 commit: Commit object
122 contents: Contents of the patch
123 progress: tuple with current patch number and total.
124 version: Version string to include in patch header
125 encoding: Encoding to use for the patch
127 Returns:
128 tuple with filename and contents
129 """
130 encoding = encoding or getattr(f, "encoding", "ascii")
131 if encoding is None:
132 encoding = "ascii"
133 if isinstance(contents, str):
134 contents = contents.encode(encoding)
135 (num, total) = progress
136 f.write(
137 b"From "
138 + commit.id
139 + b" "
140 + time.ctime(commit.commit_time).encode(encoding)
141 + b"\n"
142 )
143 f.write(b"From: " + commit.author + b"\n")
144 f.write(
145 b"Date: " + time.strftime("%a, %d %b %Y %H:%M:%S %Z").encode(encoding) + b"\n"
146 )
147 f.write(
148 (f"Subject: [PATCH {num}/{total}] ").encode(encoding) + commit.message + b"\n"
149 )
150 f.write(b"\n")
151 f.write(b"---\n")
152 try:
153 import subprocess
155 p = subprocess.Popen(
156 ["diffstat"], stdout=subprocess.PIPE, stdin=subprocess.PIPE
157 )
158 except (ImportError, OSError):
159 pass # diffstat not available?
160 else:
161 (diffstat, _) = p.communicate(contents)
162 f.write(diffstat)
163 f.write(b"\n")
164 f.write(contents)
165 f.write(b"-- \n")
166 if version is None:
167 from dulwich import __version__ as dulwich_version
169 f.write(b"Dulwich %d.%d.%d\n" % dulwich_version)
170 else:
171 if encoding is None:
172 encoding = "ascii"
173 f.write(version.encode(encoding) + b"\n")
176def _sanitize_subject_for_filename(text: str, max_length: int = 52) -> str:
177 """Sanitize a string for safe use as part of a filename.
179 Matches git's ``format_sanitized_subject`` behavior:
181 - Only ``[A-Za-z0-9._]`` are kept; other characters become ``-``
182 (collapsed across runs).
183 - Consecutive ``.`` are collapsed to a single ``.``.
184 - The result is truncated to ``max_length`` characters.
185 - Trailing ``.`` and ``-`` are stripped.
187 Args:
188 text: Input string (typically a commit subject line).
189 max_length: Maximum length of the returned string.
191 Returns: Sanitized string safe to embed in a filename.
192 """
193 result: list[str] = []
194 # 2 = initial, 1 = saw a non-title char, 0 = saw a title char
195 space = 2
196 i = 0
197 text_len = len(text)
198 while i < text_len:
199 c = text[i]
200 if ("A" <= c <= "Z") or ("a" <= c <= "z") or ("0" <= c <= "9") or c in "._":
201 if space == 1:
202 result.append("-")
203 space = 0
204 result.append(c)
205 if c == ".":
206 while i + 1 < text_len and text[i + 1] == ".":
207 i += 1
208 else:
209 space |= 1
210 i += 1
211 if len(result) >= max_length:
212 break
214 return "".join(result)[:max_length].rstrip(".-")
217def get_summary(commit: "Commit") -> str:
218 """Determine the summary line for use in a filename.
220 Sanitizes the commit subject so it is safe to use as a filename
221 component, matching git's ``format_sanitized_subject`` behavior:
222 characters outside ``[A-Za-z0-9._]`` are replaced with ``-`` (with
223 runs collapsed) and consecutive ``.`` are collapsed. The result is
224 also length-limited to prevent overly long filenames.
226 Args:
227 commit: Commit
228 Returns: Sanitized summary string suitable for use as a filename
229 component.
230 """
231 decoded = commit.message.decode(errors="replace")
232 lines = decoded.splitlines()
233 if not lines:
234 return ""
235 return _sanitize_subject_for_filename(lines[0])
238# Unified Diff
239def _format_range_unified(start: int, stop: int) -> str:
240 """Convert range to the "ed" format."""
241 # Per the diff spec at http://www.unix.org/single_unix_specification/
242 beginning = start + 1 # lines start numbering with one
243 length = stop - start
244 if length == 1:
245 return f"{beginning}"
246 if not length:
247 beginning -= 1 # empty ranges begin at line just before the range
248 return f"{beginning},{length}"
251def unified_diff(
252 a: Sequence[bytes],
253 b: Sequence[bytes],
254 fromfile: bytes = b"",
255 tofile: bytes = b"",
256 fromfiledate: str = "",
257 tofiledate: str = "",
258 n: int = 3,
259 lineterm: str = "\n",
260 tree_encoding: str = "utf-8",
261 output_encoding: str = "utf-8",
262) -> Generator[bytes, None, None]:
263 """difflib.unified_diff that can detect "No newline at end of file" as original "git diff" does.
265 Based on the same function in Python2.7 difflib.py
266 """
267 started = False
268 for group in SequenceMatcher(a=a, b=b).get_grouped_opcodes(n):
269 if not started:
270 started = True
271 fromdate = f"\t{fromfiledate}" if fromfiledate else ""
272 todate = f"\t{tofiledate}" if tofiledate else ""
273 yield f"--- {fromfile.decode(tree_encoding)}{fromdate}{lineterm}".encode(
274 output_encoding
275 )
276 yield f"+++ {tofile.decode(tree_encoding)}{todate}{lineterm}".encode(
277 output_encoding
278 )
280 first, last = group[0], group[-1]
281 file1_range = _format_range_unified(first[1], last[2])
282 file2_range = _format_range_unified(first[3], last[4])
283 yield f"@@ -{file1_range} +{file2_range} @@{lineterm}".encode(output_encoding)
285 for tag, i1, i2, j1, j2 in group:
286 if tag == "equal":
287 for line in a[i1:i2]:
288 yield b" " + line
289 continue
290 if tag in ("replace", "delete"):
291 for line in a[i1:i2]:
292 if not line[-1:] == b"\n":
293 line += b"\n\\ No newline at end of file\n"
294 yield b"-" + line
295 if tag in ("replace", "insert"):
296 for line in b[j1:j2]:
297 if not line[-1:] == b"\n":
298 line += b"\n\\ No newline at end of file\n"
299 yield b"+" + line
302def _get_sequence_matcher(
303 algorithm: str, a: Sequence[bytes], b: Sequence[bytes]
304) -> SequenceMatcher[bytes]:
305 """Get appropriate sequence matcher for the given algorithm.
307 Args:
308 algorithm: Diff algorithm ("myers" or "patience")
309 a: First sequence
310 b: Second sequence
312 Returns:
313 Configured sequence matcher instance
315 Raises:
316 DiffAlgorithmNotAvailable: If patience requested but not available
317 """
318 if algorithm == "patience":
319 try:
320 from patiencediff import PatienceSequenceMatcher
322 return PatienceSequenceMatcher(None, a, b) # type: ignore[no-any-return,unused-ignore]
323 except ImportError:
324 raise DiffAlgorithmNotAvailable(
325 "patience", "Install with: pip install 'dulwich[patiencediff]'"
326 )
327 else:
328 return SequenceMatcher(a=a, b=b)
331def unified_diff_with_algorithm(
332 a: Sequence[bytes],
333 b: Sequence[bytes],
334 fromfile: bytes = b"",
335 tofile: bytes = b"",
336 fromfiledate: str = "",
337 tofiledate: str = "",
338 n: int = 3,
339 lineterm: str = "\n",
340 tree_encoding: str = "utf-8",
341 output_encoding: str = "utf-8",
342 algorithm: str | None = None,
343) -> Generator[bytes, None, None]:
344 """Generate unified diff with specified algorithm.
346 Args:
347 a: First sequence of lines
348 b: Second sequence of lines
349 fromfile: Name of first file
350 tofile: Name of second file
351 fromfiledate: Date of first file
352 tofiledate: Date of second file
353 n: Number of context lines
354 lineterm: Line terminator
355 tree_encoding: Encoding for tree paths
356 output_encoding: Encoding for output
357 algorithm: Diff algorithm to use ("myers" or "patience")
359 Returns:
360 Generator yielding diff lines
362 Raises:
363 DiffAlgorithmNotAvailable: If patience algorithm requested but patiencediff not available
364 """
365 if algorithm is None:
366 algorithm = DEFAULT_DIFF_ALGORITHM
368 matcher = _get_sequence_matcher(algorithm, a, b)
370 started = False
371 for group in matcher.get_grouped_opcodes(n):
372 if not started:
373 started = True
374 fromdate = f"\t{fromfiledate}" if fromfiledate else ""
375 todate = f"\t{tofiledate}" if tofiledate else ""
376 yield f"--- {fromfile.decode(tree_encoding)}{fromdate}{lineterm}".encode(
377 output_encoding
378 )
379 yield f"+++ {tofile.decode(tree_encoding)}{todate}{lineterm}".encode(
380 output_encoding
381 )
383 first, last = group[0], group[-1]
384 file1_range = _format_range_unified(first[1], last[2])
385 file2_range = _format_range_unified(first[3], last[4])
386 yield f"@@ -{file1_range} +{file2_range} @@{lineterm}".encode(output_encoding)
388 for tag, i1, i2, j1, j2 in group:
389 if tag == "equal":
390 for line in a[i1:i2]:
391 yield b" " + line
392 continue
393 if tag in ("replace", "delete"):
394 for line in a[i1:i2]:
395 if not line[-1:] == b"\n":
396 line += b"\n\\ No newline at end of file\n"
397 yield b"-" + line
398 if tag in ("replace", "insert"):
399 for line in b[j1:j2]:
400 if not line[-1:] == b"\n":
401 line += b"\n\\ No newline at end of file\n"
402 yield b"+" + line
405def is_binary(content: bytes) -> bool:
406 """See if the first few bytes contain any null characters.
408 Args:
409 content: Bytestring to check for binary content
410 """
411 return b"\0" in content[:FIRST_FEW_BYTES]
414def shortid(hexsha: bytes | None) -> bytes:
415 """Get short object ID.
417 Args:
418 hexsha: Full hex SHA or None
420 Returns:
421 7-character short ID
422 """
423 if hexsha is None:
424 return b"0" * 7
425 else:
426 return hexsha[:7]
429def patch_filename(p: bytes | None, root: bytes) -> bytes:
430 """Generate patch filename.
432 Args:
433 p: Path or None
434 root: Root directory
436 Returns:
437 Full patch filename
438 """
439 if p is None:
440 return b"/dev/null"
441 else:
442 return root + b"/" + p
445def write_object_diff(
446 f: IO[bytes],
447 store: "BaseObjectStore",
448 old_file: tuple[bytes | None, int | None, ObjectID | None],
449 new_file: tuple[bytes | None, int | None, ObjectID | None],
450 diff_binary: bool = False,
451 diff_algorithm: str | None = None,
452) -> None:
453 """Write the diff for an object.
455 Args:
456 f: File-like object to write to
457 store: Store to retrieve objects from, if necessary
458 old_file: (path, mode, hexsha) tuple
459 new_file: (path, mode, hexsha) tuple
460 diff_binary: Whether to diff files even if they
461 are considered binary files by is_binary().
462 diff_algorithm: Algorithm to use for diffing ("myers" or "patience")
464 Note: the tuple elements should be None for nonexistent files
465 """
466 (old_path, old_mode, old_id) = old_file
467 (new_path, new_mode, new_id) = new_file
468 patched_old_path = patch_filename(old_path, b"a")
469 patched_new_path = patch_filename(new_path, b"b")
471 def content(mode: int | None, hexsha: ObjectID | None) -> Blob:
472 """Get blob content for a file.
474 Args:
475 mode: File mode
476 hexsha: Object SHA
478 Returns:
479 Blob object
480 """
481 if hexsha is None:
482 return Blob.from_string(b"")
483 elif mode is not None and S_ISGITLINK(mode):
484 return Blob.from_string(b"Subproject commit " + hexsha + b"\n")
485 else:
486 obj = store[hexsha]
487 if isinstance(obj, Blob):
488 return obj
489 else:
490 # Fallback for non-blob objects
491 return Blob.from_string(obj.as_raw_string())
493 def lines(content: "Blob") -> list[bytes]:
494 """Split blob content into lines.
496 Args:
497 content: Blob content
499 Returns:
500 List of lines
501 """
502 if not content:
503 return []
504 else:
505 return content.splitlines()
507 f.writelines(
508 gen_diff_header((old_path, new_path), (old_mode, new_mode), (old_id, new_id))
509 )
510 old_content = content(old_mode, old_id)
511 new_content = content(new_mode, new_id)
512 if not diff_binary and (is_binary(old_content.data) or is_binary(new_content.data)):
513 binary_diff = (
514 b"Binary files "
515 + patched_old_path
516 + b" and "
517 + patched_new_path
518 + b" differ\n"
519 )
520 f.write(binary_diff)
521 else:
522 f.writelines(
523 unified_diff_with_algorithm(
524 lines(old_content),
525 lines(new_content),
526 patched_old_path,
527 patched_new_path,
528 algorithm=diff_algorithm,
529 )
530 )
533# TODO(jelmer): Support writing unicode, rather than bytes.
534def gen_diff_header(
535 paths: tuple[bytes | None, bytes | None],
536 modes: tuple[int | None, int | None],
537 shas: tuple[bytes | None, bytes | None],
538) -> Generator[bytes, None, None]:
539 """Write a blob diff header.
541 Args:
542 paths: Tuple with old and new path
543 modes: Tuple with old and new modes
544 shas: Tuple with old and new shas
545 """
546 (old_path, new_path) = paths
547 (old_mode, new_mode) = modes
548 (old_sha, new_sha) = shas
549 if old_path is None and new_path is not None:
550 old_path = new_path
551 if new_path is None and old_path is not None:
552 new_path = old_path
553 old_path = patch_filename(old_path, b"a")
554 new_path = patch_filename(new_path, b"b")
555 yield b"diff --git " + old_path + b" " + new_path + b"\n"
557 if old_mode != new_mode:
558 if new_mode is not None:
559 if old_mode is not None:
560 yield (f"old file mode {old_mode:o}\n").encode("ascii")
561 yield (f"new file mode {new_mode:o}\n").encode("ascii")
562 else:
563 yield (f"deleted file mode {old_mode:o}\n").encode("ascii")
564 yield b"index " + shortid(old_sha) + b".." + shortid(new_sha)
565 if new_mode is not None and old_mode is not None:
566 yield (f" {new_mode:o}").encode("ascii")
567 yield b"\n"
570# TODO(jelmer): Support writing unicode, rather than bytes.
571def write_blob_diff(
572 f: IO[bytes],
573 old_file: tuple[bytes | None, int | None, "Blob | None"],
574 new_file: tuple[bytes | None, int | None, "Blob | None"],
575 diff_algorithm: str | None = None,
576) -> None:
577 """Write blob diff.
579 Args:
580 f: File-like object to write to
581 old_file: (path, mode, hexsha) tuple (None if nonexisting)
582 new_file: (path, mode, hexsha) tuple (None if nonexisting)
583 diff_algorithm: Algorithm to use for diffing ("myers" or "patience")
585 Note: The use of write_object_diff is recommended over this function.
586 """
587 (old_path, old_mode, old_blob) = old_file
588 (new_path, new_mode, new_blob) = new_file
589 patched_old_path = patch_filename(old_path, b"a")
590 patched_new_path = patch_filename(new_path, b"b")
592 def lines(blob: "Blob | None") -> list[bytes]:
593 """Split blob content into lines.
595 Args:
596 blob: Blob object or None
598 Returns:
599 List of lines
600 """
601 if blob is not None:
602 return blob.splitlines()
603 else:
604 return []
606 f.writelines(
607 gen_diff_header(
608 (old_path, new_path),
609 (old_mode, new_mode),
610 (getattr(old_blob, "id", None), getattr(new_blob, "id", None)),
611 )
612 )
613 old_contents = lines(old_blob)
614 new_contents = lines(new_blob)
615 f.writelines(
616 unified_diff_with_algorithm(
617 old_contents,
618 new_contents,
619 patched_old_path,
620 patched_new_path,
621 algorithm=diff_algorithm,
622 )
623 )
626def write_tree_diff(
627 f: IO[bytes],
628 store: "BaseObjectStore",
629 old_tree: ObjectID | None,
630 new_tree: ObjectID | None,
631 diff_binary: bool = False,
632 diff_algorithm: str | None = None,
633) -> None:
634 """Write tree diff.
636 Args:
637 f: File-like object to write to.
638 store: Object store to read from
639 old_tree: Old tree id
640 new_tree: New tree id
641 diff_binary: Whether to diff files even if they
642 are considered binary files by is_binary().
643 diff_algorithm: Algorithm to use for diffing ("myers" or "patience")
644 """
645 changes = store.tree_changes(old_tree, new_tree)
646 for (oldpath, newpath), (oldmode, newmode), (oldsha, newsha) in changes:
647 write_object_diff(
648 f,
649 store,
650 (oldpath, oldmode, oldsha),
651 (newpath, newmode, newsha),
652 diff_binary=diff_binary,
653 diff_algorithm=diff_algorithm,
654 )
657def write_commit_diff(
658 f: IO[bytes],
659 store: "BaseObjectStore",
660 commit: "Commit",
661 diff_binary: bool = False,
662 diff_algorithm: str | None = None,
663) -> None:
664 """Write the diff a commit introduces against its first parent.
666 Args:
667 f: File-like object to write to.
668 store: Object store to read from.
669 commit: Commit whose changes to diff. Root commits (with no parents)
670 are diffed against the empty tree.
671 diff_binary: Whether to diff files even if they are considered binary
672 files by is_binary().
673 diff_algorithm: Algorithm to use for diffing ("myers" or "patience").
674 """
675 if commit.parents:
676 parent = store[commit.parents[0]]
677 assert isinstance(parent, Commit)
678 parent_tree: ObjectID | None = parent.tree
679 else:
680 parent_tree = None
681 write_tree_diff(
682 f,
683 store,
684 parent_tree,
685 commit.tree,
686 diff_binary=diff_binary,
687 diff_algorithm=diff_algorithm,
688 )
691def git_am_patch_split(
692 f: TextIO | BinaryIO, encoding: str | None = None
693) -> tuple["Commit", bytes, bytes | None]:
694 """Parse a git-am-style patch and split it up into bits.
696 Args:
697 f: File-like object to parse
698 encoding: Encoding to use when creating Git objects
699 Returns: Tuple with commit object, diff contents and git version
700 """
701 encoding = encoding or getattr(f, "encoding", "ascii")
702 encoding = encoding or "ascii"
703 contents = f.read()
704 if isinstance(contents, bytes):
705 bparser = email.parser.BytesParser()
706 msg = bparser.parsebytes(contents)
707 else:
708 uparser = email.parser.Parser()
709 msg = uparser.parsestr(contents)
710 return parse_patch_message(msg, encoding)
713def parse_patch_message(
714 msg: email.message.Message, encoding: str | None = None
715) -> tuple["Commit", bytes, bytes | None]:
716 """Extract a Commit object and patch from an e-mail message.
718 Args:
719 msg: An email message (email.message.Message)
720 encoding: Encoding to use to encode Git commits
721 Returns: Tuple with commit object, diff contents and git version
722 """
723 c = Commit()
724 if encoding is None:
725 encoding = "ascii"
726 c.author = msg["from"].encode(encoding)
727 c.committer = msg["from"].encode(encoding)
728 try:
729 patch_tag_start = msg["subject"].index("[PATCH")
730 except ValueError:
731 subject = msg["subject"]
732 else:
733 close = msg["subject"].index("] ", patch_tag_start)
734 subject = msg["subject"][close + 2 :]
735 c.message = (subject.replace("\n", "") + "\n").encode(encoding)
736 first = True
738 body = msg.get_payload(decode=True)
739 if isinstance(body, str):
740 body = body.encode(encoding)
741 if isinstance(body, bytes):
742 lines = body.splitlines(True)
743 else:
744 # Handle other types by converting to string first
745 lines = str(body).encode(encoding).splitlines(True)
746 line_iter = iter(lines)
748 for line in line_iter:
749 if line == b"---\n":
750 break
751 if first:
752 if line.startswith(b"From: "):
753 c.author = line[len(b"From: ") :].rstrip()
754 else:
755 c.message += b"\n" + line
756 first = False
757 else:
758 c.message += line
759 diff = b""
760 for line in line_iter:
761 if line == b"-- \n":
762 break
763 diff += line
764 try:
765 version = next(line_iter).rstrip(b"\n")
766 except StopIteration:
767 version = None
768 return c, diff, version
771def patch_id(diff_data: bytes) -> bytes:
772 """Compute patch ID for a diff.
774 The patch ID is computed by normalizing the diff and computing a SHA1 hash.
775 This follows git's patch-id algorithm which:
776 1. Removes whitespace from lines starting with + or -
777 2. Replaces line numbers in @@ headers with a canonical form
778 3. Computes SHA1 of the result
780 Args:
781 diff_data: Raw diff data as bytes
783 Returns:
784 SHA1 hash of normalized diff (40-byte hex string)
786 TODO: This implementation uses a simple line-by-line approach. For better
787 compatibility with git's patch-id, consider using proper patch parsing that:
788 - Handles edge cases in diff format (binary diffs, mode changes, etc.)
789 - Properly parses unified diff format according to the spec
790 - Matches git's exact normalization algorithm byte-for-byte
791 See git's patch-id.c for reference implementation.
792 """
793 import hashlib
794 import re
796 # Normalize the diff for patch-id computation
797 normalized_lines = []
799 for line in diff_data.split(b"\n"):
800 # Skip diff headers (diff --git, index, ---, +++)
801 if line.startswith(
802 (
803 b"diff --git ",
804 b"index ",
805 b"--- ",
806 b"+++ ",
807 b"new file mode ",
808 b"old file mode ",
809 b"deleted file mode ",
810 b"new mode ",
811 b"old mode ",
812 b"similarity index ",
813 b"dissimilarity index ",
814 b"rename from ",
815 b"rename to ",
816 b"copy from ",
817 b"copy to ",
818 )
819 ):
820 continue
822 # Normalize @@ headers to a canonical form
823 if line.startswith(b"@@"):
824 # Replace line numbers with canonical form
825 match = re.match(rb"^@@\s+-\d+(?:,\d+)?\s+\+\d+(?:,\d+)?\s+@@", line)
826 if match:
827 # Use canonical hunk header without line numbers
828 normalized_lines.append(b"@@")
829 continue
831 # For +/- lines, strip all whitespace
832 if line.startswith((b"+", b"-")):
833 # Keep the +/- prefix but remove all whitespace from the rest
834 if len(line) > 1:
835 # Remove all whitespace from the content
836 content = line[1:].replace(b" ", b"").replace(b"\t", b"")
837 normalized_lines.append(line[:1] + content)
838 else:
839 # Just +/- alone
840 normalized_lines.append(line[:1])
841 continue
843 # Keep context lines and other content as-is
844 if line.startswith(b" ") or line == b"":
845 normalized_lines.append(line)
847 # Join normalized lines and compute SHA1
848 normalized = b"\n".join(normalized_lines)
849 return hashlib.sha1(normalized).hexdigest().encode("ascii")
852def commit_patch_id(
853 store: "BaseObjectStore", commit_id: ObjectID | RawObjectID
854) -> bytes:
855 """Compute patch ID for a commit.
857 Args:
858 store: Object store to read objects from
859 commit_id: Commit ID (40-byte hex string)
861 Returns:
862 Patch ID (40-byte hex string)
863 """
864 from io import BytesIO
866 commit = store[commit_id]
867 assert isinstance(commit, Commit)
869 diff_output = BytesIO()
870 write_commit_diff(diff_output, store, commit)
872 return patch_id(diff_output.getvalue())
875@dataclass
876class MailinfoResult:
877 """Result of mailinfo parsing.
879 Attributes:
880 author_name: Author's name
881 author_email: Author's email address
882 author_date: Author's date (if present in the email)
883 subject: Processed subject line
884 message: Commit message body
885 patch: Patch content
886 message_id: Message-ID header (if -m/--message-id was used)
887 """
889 author_name: str
890 author_email: str
891 author_date: str | None
892 subject: str
893 message: str
894 patch: str
895 message_id: str | None = None
898def _munge_subject(subject: str, keep_subject: bool, keep_non_patch: bool) -> str:
899 """Munge email subject line for commit message.
901 Args:
902 subject: Original subject line
903 keep_subject: If True, keep subject intact (-k option)
904 keep_non_patch: If True, only strip [PATCH] (-b option)
906 Returns:
907 Processed subject line
908 """
909 if keep_subject:
910 return subject
912 result = subject
914 # First remove Re: prefixes (they can appear before brackets)
915 while True:
916 new_result = re.sub(r"^\s*(?:re|RE|Re):\s*", "", result, flags=re.IGNORECASE)
917 if new_result == result:
918 break
919 result = new_result
921 # Remove bracketed strings
922 if keep_non_patch:
923 # Only remove brackets containing "PATCH"
924 # Match each bracket individually anywhere in the string
925 while True:
926 # Remove PATCH bracket, but be careful with whitespace
927 new_result = re.sub(
928 r"\[[^\]]*?PATCH[^\]]*?\](\s+)?", r"\1", result, flags=re.IGNORECASE
929 )
930 if new_result == result:
931 break
932 result = new_result
933 else:
934 # Remove all bracketed strings
935 while True:
936 new_result = re.sub(r"^\s*\[.*?\]\s*", "", result)
937 if new_result == result:
938 break
939 result = new_result
941 # Remove leading/trailing whitespace
942 result = result.strip()
944 # Normalize multiple whitespace to single space
945 result = re.sub(r"\s+", " ", result)
947 return result
950def _find_scissors_line(lines: list[bytes]) -> int | None:
951 """Find the scissors line in message body.
953 Args:
954 lines: List of lines in the message body
956 Returns:
957 Index of scissors line, or None if not found
958 """
959 # The perforation alternatives are written so no two unbounded ``-+``
960 # quantifiers are separated by an optional/empty subpattern. The earlier
961 # form ``(?:>?\s*-+\s*)?(?:8<|>8)?\s*-+\s*`` let a single run of dashes be
962 # split between two ``-+`` groups, so a long non-matching line of dashes
963 # backtracked quadratically (ReDoS) when fed through ``mailinfo`` /
964 # ``git am --scissors`` on an untrusted mailbox. Here a marker is the
965 # mandatory pivot between the two dash runs, and the marker-less case uses a
966 # bounded ``(?:\s+-+)?`` second run, keeping the match linear.
967 scissors_pattern = re.compile(
968 rb"^>?\s*-+(?:\s+-+)?\s*$"
969 rb"|^>?\s*(?:-+\s*)?(?:8<|>8)\s*-+\s*$"
970 rb"|^(?:>?\s*-+\s*)(?:cut here|scissors)(?:\s*-+)?$",
971 re.IGNORECASE,
972 )
974 for i, line in enumerate(lines):
975 if scissors_pattern.match(line.strip()):
976 return i
978 return None
981def git_base85_decode(data: bytes) -> bytes:
982 """Decode Git's base85-encoded binary data.
984 Git uses a custom base85 encoding with its own alphabet and line format.
985 Each line starts with a length byte followed by base85-encoded data.
987 Args:
988 data: Base85-encoded data as bytes (may contain multiple lines)
990 Returns:
991 Decoded binary data
993 Raises:
994 ValueError: If the data is invalid
995 """
996 # Git's base85 alphabet (different from RFC 1924)
997 alphabet = b"0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz!#$%&()*+-;<=>?@^_`{|}~"
999 # Create decode table
1000 decode_table = {}
1001 for i, c in enumerate(alphabet):
1002 decode_table[c] = i
1004 result = bytearray()
1005 lines = data.strip().split(b"\n")
1007 for line in lines:
1008 if not line:
1009 continue
1011 # First character encodes the length of decoded data for this line
1012 if line[0] not in decode_table:
1013 continue
1015 encoded_len = decode_table[line[0]]
1016 if encoded_len == 0:
1017 continue
1019 # Decode the rest of the line
1020 encoded_data = line[1:]
1022 # Process in groups of 5 characters (which encode 4 bytes)
1023 i = 0
1024 decoded_this_line = 0
1025 while i < len(encoded_data) and decoded_this_line < encoded_len:
1026 # Get up to 5 characters
1027 group = encoded_data[i : i + 5]
1028 if len(group) == 0:
1029 break
1031 # Decode 5 base85 digits to a 32-bit value
1032 value = 0
1033 for c in group:
1034 if c not in decode_table:
1035 raise ValueError(f"Invalid base85 character: {chr(c)}")
1036 value = value * 85 + decode_table[c]
1038 # Convert to 4 bytes (big-endian)
1039 bytes_to_add = min(4, encoded_len - decoded_this_line)
1040 decoded_bytes = value.to_bytes(4, byteorder="big")
1041 result.extend(decoded_bytes[:bytes_to_add])
1042 decoded_this_line += bytes_to_add
1043 i += 5
1045 return bytes(result)
1048@dataclass
1049class PatchHunk:
1050 """Represents a single hunk in a unified diff.
1052 Attributes:
1053 old_start: Starting line number in old file
1054 old_count: Number of lines in old file
1055 new_start: Starting line number in new file
1056 new_count: Number of lines in new file
1057 lines: List of diff lines (prefixed with ' ', '+', or '-')
1058 """
1060 old_start: int
1061 old_count: int
1062 new_start: int
1063 new_count: int
1064 lines: list[bytes]
1067@dataclass
1068class FilePatch:
1069 """Represents a patch for a single file.
1071 Attributes:
1072 old_path: Path to old file (None for new files)
1073 new_path: Path to new file (None for deleted files)
1074 old_mode: Mode of old file (None for new files)
1075 new_mode: Mode of new file (None for deleted files)
1076 hunks: List of PatchHunk objects
1077 binary: True if this is a binary patch
1078 rename_from: Original path for renames (None if not a rename)
1079 rename_to: New path for renames (None if not a rename)
1080 copy_from: Source path for copies (None if not a copy)
1081 copy_to: Destination path for copies (None if not a copy)
1082 binary_old: Old binary content for binary patches (base85 encoded)
1083 binary_new: New binary content for binary patches (base85 encoded)
1084 """
1086 old_path: bytes | None
1087 new_path: bytes | None
1088 old_mode: int | None
1089 new_mode: int | None
1090 hunks: list[PatchHunk]
1091 binary: bool = False
1092 rename_from: bytes | None = None
1093 rename_to: bytes | None = None
1094 copy_from: bytes | None = None
1095 copy_to: bytes | None = None
1096 binary_old: bytes | None = None
1097 binary_new: bytes | None = None
1100def parse_unified_diff(diff_text: bytes) -> list[FilePatch]:
1101 """Parse a unified diff into FilePatch objects.
1103 Args:
1104 diff_text: Unified diff content as bytes
1106 Returns:
1107 List of FilePatch objects
1108 """
1109 patches: list[FilePatch] = []
1110 lines = diff_text.split(b"\n")
1111 i = 0
1113 while i < len(lines):
1114 line = lines[i]
1116 # Look for diff header
1117 if line.startswith(b"diff --git "):
1118 # Parse file patch
1119 old_path = None
1120 new_path = None
1121 old_mode = None
1122 new_mode = None
1123 hunks: list[PatchHunk] = []
1124 binary = False
1125 rename_from = None
1126 rename_to = None
1127 copy_from = None
1128 copy_to = None
1129 binary_old = None
1130 binary_new = None
1132 # Parse extended headers
1133 i += 1
1134 while i < len(lines):
1135 line = lines[i]
1137 if line.startswith(b"old file mode "):
1138 old_mode = int(line.split()[-1], 8)
1139 i += 1
1140 elif line.startswith(b"new file mode "):
1141 new_mode = int(line.split()[-1], 8)
1142 i += 1
1143 elif line.startswith(b"deleted file mode "):
1144 old_mode = int(line.split()[-1], 8)
1145 i += 1
1146 elif line.startswith(b"new mode "):
1147 new_mode = int(line.split()[-1], 8)
1148 i += 1
1149 elif line.startswith(b"old mode "):
1150 old_mode = int(line.split()[-1], 8)
1151 i += 1
1152 elif line.startswith(b"rename from "):
1153 rename_from = line[12:].strip()
1154 i += 1
1155 elif line.startswith(b"rename to "):
1156 rename_to = line[10:].strip()
1157 i += 1
1158 elif line.startswith(b"copy from "):
1159 copy_from = line[10:].strip()
1160 i += 1
1161 elif line.startswith(b"copy to "):
1162 copy_to = line[8:].strip()
1163 i += 1
1164 elif line.startswith(b"similarity index "):
1165 # Just skip similarity index for now
1166 i += 1
1167 elif line.startswith(b"dissimilarity index "):
1168 # Just skip dissimilarity index for now
1169 i += 1
1170 elif line.startswith(b"index "):
1171 i += 1
1172 elif line.startswith(b"--- "):
1173 # Parse old file path
1174 path = line[4:].split(b"\t")[0]
1175 if path != b"/dev/null":
1176 old_path = path
1177 i += 1
1178 elif line.startswith(b"+++ "):
1179 # Parse new file path
1180 path = line[4:].split(b"\t")[0]
1181 if path != b"/dev/null":
1182 new_path = path
1183 i += 1
1184 break
1185 elif line.startswith(b"Binary files"):
1186 binary = True
1187 i += 1
1188 break
1189 elif line.startswith(b"GIT binary patch"):
1190 binary = True
1191 i += 1
1192 # Parse binary patch data
1193 while i < len(lines):
1194 line = lines[i]
1195 if line.startswith(b"literal "):
1196 # New binary data
1197 # size = int(line[8:].strip()) # Size information, not currently used
1198 i += 1
1199 binary_data = b""
1200 while i < len(lines):
1201 line = lines[i]
1202 if (
1203 line.startswith(
1204 (b"literal ", b"delta ", b"diff --git ")
1205 )
1206 or not line.strip()
1207 ):
1208 break
1209 binary_data += line + b"\n"
1210 i += 1
1211 binary_new = binary_data
1212 elif line.startswith(b"delta "):
1213 # Delta patch (not supported yet)
1214 i += 1
1215 while i < len(lines):
1216 line = lines[i]
1217 if (
1218 line.startswith(
1219 (b"literal ", b"delta ", b"diff --git ")
1220 )
1221 or not line.strip()
1222 ):
1223 break
1224 i += 1
1225 else:
1226 break
1227 break
1228 else:
1229 i += 1
1230 break
1232 # Parse hunks
1233 if not binary:
1234 while i < len(lines):
1235 line = lines[i]
1237 if line.startswith(b"@@ "):
1238 # Parse hunk header
1239 match = re.match(
1240 rb"@@ -(\d+)(?:,(\d+))? \+(\d+)(?:,(\d+))? @@", line
1241 )
1242 if match:
1243 old_start = int(match.group(1))
1244 old_count = int(match.group(2)) if match.group(2) else 1
1245 new_start = int(match.group(3))
1246 new_count = int(match.group(4)) if match.group(4) else 1
1248 # Parse hunk lines
1249 hunk_lines: list[bytes] = []
1250 i += 1
1251 while i < len(lines):
1252 line = lines[i]
1253 if line.startswith((b" ", b"+", b"-", b"\\")):
1254 hunk_lines.append(line)
1255 i += 1
1256 else:
1257 break
1259 hunks.append(
1260 PatchHunk(
1261 old_start=old_start,
1262 old_count=old_count,
1263 new_start=new_start,
1264 new_count=new_count,
1265 lines=hunk_lines,
1266 )
1267 )
1268 else:
1269 i += 1
1270 elif line.startswith(b"diff --git "):
1271 # Next file patch
1272 break
1273 else:
1274 i += 1
1275 if not line.strip():
1276 # Empty line, might be end of patch or separator
1277 break
1279 patches.append(
1280 FilePatch(
1281 old_path=old_path,
1282 new_path=new_path,
1283 old_mode=old_mode,
1284 new_mode=new_mode,
1285 hunks=hunks,
1286 binary=binary,
1287 rename_from=rename_from,
1288 rename_to=rename_to,
1289 copy_from=copy_from,
1290 copy_to=copy_to,
1291 binary_old=binary_old,
1292 binary_new=binary_new,
1293 )
1294 )
1295 else:
1296 i += 1
1298 return patches
1301def apply_patch_hunks(
1302 patch: FilePatch,
1303 original_lines: list[bytes],
1304) -> list[bytes] | None:
1305 """Apply patch hunks to file content.
1307 Args:
1308 patch: FilePatch object to apply
1309 original_lines: Original file content as list of lines
1311 Returns:
1312 Patched file content as list of lines, or None if patch cannot be applied
1313 """
1314 result = original_lines[:]
1315 offset = 0 # Track line offset as we apply hunks
1317 for hunk in patch.hunks:
1318 # Adjust hunk position by offset
1319 # old_start is 1-indexed; 0 means the hunk inserts at the beginning
1320 target_line = max(hunk.old_start - 1, 0) + offset
1322 # Extract old and new content from hunk
1323 old_content: list[bytes] = []
1324 new_content: list[bytes] = []
1326 for line in hunk.lines:
1327 if line.startswith(b"\\"):
1328 # Skip "\ No newline at end of file" markers
1329 continue
1330 elif line.startswith(b" "):
1331 # Context line - add newline if not present
1332 content = line[1:]
1333 if not content.endswith(b"\n"):
1334 content += b"\n"
1335 old_content.append(content)
1336 new_content.append(content)
1337 elif line.startswith(b"-"):
1338 # Deletion - add newline if not present
1339 content = line[1:]
1340 if not content.endswith(b"\n"):
1341 content += b"\n"
1342 old_content.append(content)
1343 elif line.startswith(b"+"):
1344 # Addition - add newline if not present
1345 content = line[1:]
1346 if not content.endswith(b"\n"):
1347 content += b"\n"
1348 new_content.append(content)
1350 # Verify context matches
1351 if target_line < 0 or target_line + len(old_content) > len(result):
1352 # TODO: Implement fuzzy matching
1353 return None
1355 for i, old_line in enumerate(old_content):
1356 if result[target_line + i] != old_line:
1357 # Context doesn't match
1358 # TODO: Implement fuzzy matching
1359 return None
1361 # Apply the patch
1362 result[target_line : target_line + len(old_content)] = new_content
1364 # Update offset for next hunk
1365 offset += len(new_content) - len(old_content)
1367 return result
1370def _ensure_within_repo(repo_path: bytes, fs_path: bytes, rel_path: bytes) -> None:
1371 """Reject patch target paths that resolve outside the working tree.
1373 Patch headers are untrusted (e.g. ``git am`` of a mailbox), so a name such
1374 as ``../../etc/cron.d/x`` or an absolute path must not be written through
1375 ``os.path.join``. Mirrors git's refusal of paths outside the work tree.
1376 """
1377 repo_root = os.path.realpath(repo_path)
1378 resolved = os.path.realpath(fs_path)
1379 if resolved != repo_root and not resolved.startswith(
1380 repo_root + os.fsencode(os.sep)
1381 ):
1382 raise ValueError(f"patch affects file outside repository: {rel_path!r}")
1385def _apply_rename_or_copy(
1386 r: "Repo",
1387 src_path: bytes,
1388 dst_path: bytes,
1389 strip: int,
1390 patch: FilePatch,
1391 is_rename: bool,
1392 cached: bool,
1393 check: bool,
1394 config: "Config | None",
1395) -> tuple[list[bytes] | None, bool]:
1396 """Apply a rename or copy operation.
1398 Args:
1399 r: Repository object
1400 src_path: Source path
1401 dst_path: Destination path
1402 strip: Number of path components to strip
1403 patch: FilePatch object
1404 is_rename: True for rename, False for copy
1405 cached: Apply to index only, not working tree
1406 check: Check only, don't apply
1407 config: Repository configuration
1409 Returns:
1410 A tuple of (``original_lines``, ``should_continue``) where:
1411 - ``original_lines``: Content lines if hunks need to be applied, None otherwise
1412 - ``should_continue``: True to skip to next patch, False to continue processing
1413 """
1414 from .index import (
1415 ConflictedIndexEntry,
1416 IndexEntry,
1417 cleanup_mode,
1418 index_entry_from_stat,
1419 )
1421 # Strip path components
1422 src_stripped = src_path
1423 dst_stripped = dst_path
1424 if strip > 0:
1425 src_parts = src_path.split(b"/")
1426 if len(src_parts) > strip:
1427 src_stripped = b"/".join(src_parts[strip:])
1428 dst_parts = dst_path.split(b"/")
1429 if len(dst_parts) > strip:
1430 dst_stripped = b"/".join(dst_parts[strip:])
1432 repo_path_bytes = r.path.encode("utf-8") if isinstance(r.path, str) else r.path
1433 src_fs_path = os.path.join(repo_path_bytes, src_stripped)
1434 dst_fs_path = os.path.join(repo_path_bytes, dst_stripped)
1435 _ensure_within_repo(repo_path_bytes, src_fs_path, src_stripped)
1436 _ensure_within_repo(repo_path_bytes, dst_fs_path, dst_stripped)
1438 # Read content from source file
1439 op_name = "rename" if is_rename else "copy"
1440 if os.path.exists(src_fs_path):
1441 with open(src_fs_path, "rb") as f:
1442 content = f.read()
1443 else:
1444 # Try to read from index
1445 index = r.open_index(config=config)
1446 if src_stripped in index:
1447 entry = index[src_stripped]
1448 if not isinstance(entry, ConflictedIndexEntry):
1449 obj = r.object_store[entry.sha]
1450 if isinstance(obj, Blob):
1451 content = obj.data
1452 else:
1453 raise ValueError(
1454 f"Cannot {op_name}: source {src_stripped.decode('utf-8', errors='replace')} not found"
1455 )
1456 else:
1457 raise ValueError(
1458 f"Cannot {op_name}: source {src_stripped.decode('utf-8', errors='replace')} is conflicted"
1459 )
1460 else:
1461 raise ValueError(
1462 f"Cannot {op_name}: source {src_stripped.decode('utf-8', errors='replace')} not found"
1463 )
1465 # If there are hunks, return content as lines for further processing
1466 if patch.hunks:
1467 return content.splitlines(keepends=True), False
1469 # No hunks - pure rename/copy
1470 if check:
1471 return None, True
1473 # Write to destination
1474 if not cached:
1475 os.makedirs(os.path.dirname(dst_fs_path), exist_ok=True)
1476 with open(dst_fs_path, "wb") as f:
1477 f.write(content)
1478 if patch.new_mode is not None:
1479 os.chmod(dst_fs_path, cleanup_mode(patch.new_mode))
1481 # Update index
1482 index = r.open_index(config=config)
1483 blob = Blob.from_string(content)
1484 r.object_store.add_object(blob)
1486 if not cached and os.path.exists(dst_fs_path):
1487 st = os.stat(dst_fs_path)
1488 entry = index_entry_from_stat(st, blob.id)
1489 else:
1490 entry = IndexEntry(
1491 ctime=(0, 0),
1492 mtime=(0, 0),
1493 dev=0,
1494 ino=0,
1495 mode=patch.new_mode or 0o100644,
1496 uid=0,
1497 gid=0,
1498 size=len(content),
1499 sha=blob.id,
1500 flags=0,
1501 )
1503 index[dst_stripped] = entry
1505 # For renames, remove the old file
1506 if is_rename:
1507 if not cached and os.path.exists(src_fs_path):
1508 os.remove(src_fs_path)
1509 if src_stripped in index:
1510 del index[src_stripped]
1512 index.write()
1513 return None, True
1516def apply_patches(
1517 r: "Repo",
1518 patches: list[FilePatch],
1519 cached: bool = False,
1520 reverse: bool = False,
1521 check: bool = False,
1522 strip: int = 1,
1523 three_way: bool = False,
1524 *,
1525 config: "Config | None" = None,
1526) -> None:
1527 """Apply a list of file patches to a repository.
1529 Args:
1530 r: Repository object
1531 patches: List of FilePatch objects to apply
1532 cached: Apply patch to index only, not working tree
1533 reverse: Apply patch in reverse
1534 check: Only check if patch can be applied, don't apply
1535 strip: Number of leading path components to strip (default: 1)
1536 three_way: Fall back to 3-way merge if patch does not apply cleanly
1537 config: Repository configuration. If None, falls back to
1538 ``r.get_config_stack()``.
1540 Raises:
1541 ValueError: If patch cannot be applied
1542 """
1543 from .index import (
1544 ConflictedIndexEntry,
1545 IndexEntry,
1546 cleanup_mode,
1547 index_entry_from_stat,
1548 )
1550 if config is None:
1551 config = r.get_config_stack()
1553 for patch in patches:
1554 # Determine the file path
1555 # For renames/copies without hunks, old_path/new_path may be None
1556 # Use local variables to avoid mutating the patch object
1557 old_path = patch.old_path
1558 new_path = patch.new_path
1560 if new_path is None and old_path is None:
1561 if patch.rename_to is not None:
1562 # Use rename_to for the target path
1563 new_path = patch.rename_to
1564 old_path = patch.rename_from
1565 elif patch.copy_to is not None:
1566 # Use copy_to for the target path
1567 new_path = patch.copy_to
1568 old_path = patch.copy_from
1569 else:
1570 raise ValueError("Patch has no file path")
1572 # Choose path based on operation
1573 file_path: bytes
1574 if new_path is None:
1575 # Deletion
1576 if old_path is None:
1577 raise ValueError("Patch has no file path")
1578 file_path = old_path
1579 elif old_path is None:
1580 # Addition
1581 file_path = new_path
1582 else:
1583 # Modification (use new path)
1584 file_path = new_path
1586 # Strip path components
1587 if strip > 0:
1588 parts = file_path.split(b"/")
1589 if len(parts) > strip:
1590 file_path = b"/".join(parts[strip:])
1592 # Convert to filesystem path
1593 tree_path = file_path
1594 repo_path_bytes = r.path.encode("utf-8") if isinstance(r.path, str) else r.path
1595 fs_path = os.path.join(repo_path_bytes, file_path)
1596 _ensure_within_repo(repo_path_bytes, fs_path, file_path)
1598 # Handle renames and copies
1599 original_lines: list[bytes] | None = None
1600 if patch.rename_from is not None and patch.rename_to is not None:
1601 original_lines, should_continue = _apply_rename_or_copy(
1602 r,
1603 patch.rename_from,
1604 patch.rename_to,
1605 strip,
1606 patch,
1607 is_rename=True,
1608 cached=cached,
1609 check=check,
1610 config=config,
1611 )
1612 if should_continue:
1613 continue
1614 elif patch.copy_from is not None and patch.copy_to is not None:
1615 original_lines, should_continue = _apply_rename_or_copy(
1616 r,
1617 patch.copy_from,
1618 patch.copy_to,
1619 strip,
1620 patch,
1621 is_rename=False,
1622 cached=cached,
1623 check=check,
1624 config=config,
1625 )
1626 if should_continue:
1627 continue
1629 # Handle binary patches
1630 if patch.binary:
1631 if patch.binary_new is not None:
1632 # Decode binary patch
1633 try:
1634 binary_content = git_base85_decode(patch.binary_new)
1635 except (ValueError, KeyError) as e:
1636 raise ValueError(f"Failed to decode binary patch: {e}")
1638 if check:
1639 # Just checking, don't actually apply
1640 continue
1642 # Write binary file
1643 if not cached:
1644 os.makedirs(os.path.dirname(fs_path), exist_ok=True)
1645 with open(fs_path, "wb") as f:
1646 f.write(binary_content)
1647 if patch.new_mode is not None:
1648 os.chmod(fs_path, cleanup_mode(patch.new_mode))
1650 # Update index
1651 index = r.open_index(config=config)
1652 blob = Blob.from_string(binary_content)
1653 r.object_store.add_object(blob)
1655 if not cached and os.path.exists(fs_path):
1656 st = os.stat(fs_path)
1657 entry = index_entry_from_stat(st, blob.id)
1658 else:
1659 entry = IndexEntry(
1660 ctime=(0, 0),
1661 mtime=(0, 0),
1662 dev=0,
1663 ino=0,
1664 mode=patch.new_mode or 0o100644,
1665 uid=0,
1666 gid=0,
1667 size=len(binary_content),
1668 sha=blob.id,
1669 flags=0,
1670 )
1672 index[tree_path] = entry
1673 index.write()
1674 continue
1675 else:
1676 # Old-style "Binary files differ" message without actual patch data
1677 raise NotImplementedError(
1678 "Binary patch detected but no patch data provided (use git diff --binary)"
1679 )
1681 # Read original file content (unless already loaded from rename/copy)
1682 if original_lines is None:
1683 if patch.old_path is None:
1684 # New file
1685 original_lines = []
1686 else:
1687 if os.path.exists(fs_path):
1688 with open(fs_path, "rb") as f:
1689 content = f.read()
1690 original_lines = content.splitlines(keepends=True)
1691 else:
1692 # File doesn't exist - check if it's in the index
1693 try:
1694 index = r.open_index(config=config)
1695 if tree_path in index:
1696 index_entry: IndexEntry | ConflictedIndexEntry = index[
1697 tree_path
1698 ]
1699 if not isinstance(index_entry, ConflictedIndexEntry):
1700 obj = r.object_store[index_entry.sha]
1701 if isinstance(obj, Blob):
1702 original_lines = obj.data.splitlines(keepends=True)
1703 else:
1704 original_lines = []
1705 else:
1706 original_lines = []
1707 else:
1708 original_lines = []
1709 except (KeyError, FileNotFoundError):
1710 original_lines = []
1712 # Reverse patch if requested
1713 if reverse:
1714 # Swap old and new in hunks
1715 for hunk in patch.hunks:
1716 hunk.old_start, hunk.new_start = hunk.new_start, hunk.old_start
1717 hunk.old_count, hunk.new_count = hunk.new_count, hunk.old_count
1718 # Swap +/- prefixes
1719 reversed_lines = []
1720 for line in hunk.lines:
1721 if line.startswith(b"+"):
1722 reversed_lines.append(b"-" + line[1:])
1723 elif line.startswith(b"-"):
1724 reversed_lines.append(b"+" + line[1:])
1725 else:
1726 reversed_lines.append(line)
1727 hunk.lines = reversed_lines
1729 # Apply the patch
1730 assert original_lines is not None
1731 result = apply_patch_hunks(patch, original_lines)
1733 if result is None and three_way:
1734 # Try 3-way merge fallback
1735 from .merge import merge_blobs
1737 # Reconstruct base version from the patch
1738 # Base is what you get by taking only the old lines from hunks
1739 base_lines = []
1740 theirs_lines = []
1742 for hunk in patch.hunks:
1743 for line in hunk.lines:
1744 if line.startswith(b"\\"):
1745 # Skip "\ No newline at end of file" markers
1746 continue
1747 elif line.startswith(b" "):
1748 # Context line - in both base and theirs
1749 content = line[1:]
1750 if not content.endswith(b"\n"):
1751 content += b"\n"
1752 base_lines.append(content)
1753 theirs_lines.append(content)
1754 elif line.startswith(b"-"):
1755 # Deletion - only in base
1756 content = line[1:]
1757 if not content.endswith(b"\n"):
1758 content += b"\n"
1759 base_lines.append(content)
1760 elif line.startswith(b"+"):
1761 # Addition - only in theirs
1762 content = line[1:]
1763 if not content.endswith(b"\n"):
1764 content += b"\n"
1765 theirs_lines.append(content)
1767 # Create blobs for merging
1768 base_content = b"".join(base_lines)
1769 ours_content = b"".join(original_lines)
1770 theirs_content = b"".join(theirs_lines)
1772 base_blob = Blob.from_string(base_content) if base_content else None
1773 ours_blob = Blob.from_string(ours_content) if ours_content else None
1774 theirs_blob = Blob.from_string(theirs_content)
1776 # Perform 3-way merge
1777 merged_content, _had_conflicts = merge_blobs(
1778 base_blob, ours_blob, theirs_blob, path=tree_path
1779 )
1781 result = merged_content.splitlines(keepends=True)
1783 # Note: if _had_conflicts is True, the result contains conflict markers
1784 # Git would exit with error code, but we continue processing
1785 elif result is None:
1786 raise PatchApplicationFailure(
1787 f"Patch does not apply to {file_path.decode('utf-8', errors='replace')}"
1788 )
1790 if check:
1791 # Just checking, don't actually apply
1792 continue
1794 # Write result
1795 result_content = b"".join(result)
1797 if patch.new_path is None:
1798 # File deletion
1799 if not cached and os.path.exists(fs_path):
1800 os.remove(fs_path)
1801 # Remove from index
1802 index = r.open_index(config=config)
1803 if tree_path in index:
1804 del index[tree_path]
1805 index.write()
1806 else:
1807 # File addition or modification
1808 if not cached:
1809 # Write to working tree
1810 os.makedirs(os.path.dirname(fs_path), exist_ok=True)
1811 with open(fs_path, "wb") as f:
1812 f.write(result_content)
1814 # Update file mode if specified
1815 if patch.new_mode is not None:
1816 os.chmod(fs_path, cleanup_mode(patch.new_mode))
1818 # Update index
1819 index = r.open_index(config=config)
1820 blob = Blob.from_string(result_content)
1821 r.object_store.add_object(blob)
1823 # Get file stat for index entry
1824 if not cached and os.path.exists(fs_path):
1825 st = os.stat(fs_path)
1826 entry = index_entry_from_stat(st, blob.id)
1827 else:
1828 # Create a minimal index entry for cached-only changes
1829 entry = IndexEntry(
1830 ctime=(0, 0),
1831 mtime=(0, 0),
1832 dev=0,
1833 ino=0,
1834 mode=patch.new_mode or 0o100644,
1835 uid=0,
1836 gid=0,
1837 size=len(result_content),
1838 sha=blob.id,
1839 flags=0,
1840 )
1842 index[tree_path] = entry
1844 # Handle cleanup for renames with hunks
1845 if patch.rename_from is not None and patch.rename_to is not None:
1846 # Remove old file after successful rename
1847 old_rename_path = patch.rename_from
1848 if strip > 0:
1849 old_parts = old_rename_path.split(b"/")
1850 if len(old_parts) > strip:
1851 old_rename_path = b"/".join(old_parts[strip:])
1853 old_fs_path = os.path.join(
1854 r.path.encode("utf-8") if isinstance(r.path, str) else r.path,
1855 old_rename_path,
1856 )
1858 if not cached and os.path.exists(old_fs_path):
1859 os.remove(old_fs_path)
1860 if old_rename_path in index:
1861 del index[old_rename_path]
1863 index.write()
1866def mailinfo(
1867 msg: email.message.Message | BinaryIO | TextIO,
1868 keep_subject: bool = False,
1869 keep_non_patch: bool = False,
1870 encoding: str | None = None,
1871 scissors: bool = False,
1872 message_id: bool = False,
1873) -> MailinfoResult:
1874 """Extract patch information from an email message.
1876 This function parses an email message and extracts commit metadata
1877 (author, email, subject) and separates the commit message from the
1878 patch content, similar to git mailinfo.
1880 Args:
1881 msg: Email message (email.message.Message object) or file handle to read from
1882 keep_subject: If True, keep subject intact without munging (-k)
1883 keep_non_patch: If True, only strip [PATCH] from brackets (-b)
1884 encoding: Character encoding to use (default: detect from message)
1885 scissors: If True, remove everything before scissors line
1886 message_id: If True, include Message-ID in commit message (-m)
1888 Returns:
1889 MailinfoResult with parsed information
1891 Raises:
1892 ValueError: If message is malformed or missing required fields
1893 """
1894 # Parse message if given a file handle
1895 parsed_msg: email.message.Message
1896 if not isinstance(msg, email.message.Message):
1897 if hasattr(msg, "read"):
1898 content = msg.read()
1899 if isinstance(content, bytes):
1900 bparser = email.parser.BytesParser()
1901 parsed_msg = bparser.parsebytes(content)
1902 else:
1903 sparser = email.parser.Parser()
1904 parsed_msg = sparser.parsestr(content)
1905 else:
1906 raise ValueError("msg must be an email.message.Message or file-like object")
1907 else:
1908 parsed_msg = msg
1910 # Detect encoding from message if not specified
1911 if encoding is None:
1912 encoding = parsed_msg.get_content_charset() or "utf-8"
1914 # Extract author information
1915 from_header = parsed_msg.get("From", "")
1916 if not from_header:
1917 raise ValueError("Email message missing 'From' header")
1919 # Parse "Name <email>" format
1920 author_name, author_email = email.utils.parseaddr(from_header)
1921 if not author_email:
1922 raise ValueError(
1923 f"Could not parse email address from 'From' header: {from_header}"
1924 )
1926 # Extract date
1927 date_header = parsed_msg.get("Date")
1928 author_date = date_header if date_header else None
1930 # Extract and process subject
1931 subject = parsed_msg.get("Subject", "")
1932 if not subject:
1933 subject = "(no subject)"
1935 # Convert Header object to string if needed
1936 subject = str(subject)
1938 # Remove newlines from subject
1939 subject = subject.replace("\n", " ").replace("\r", " ")
1940 subject = _munge_subject(subject, keep_subject, keep_non_patch)
1942 # Extract Message-ID if requested
1943 msg_id = None
1944 if message_id:
1945 msg_id = parsed_msg.get("Message-ID")
1947 # Get message body
1948 body = parsed_msg.get_payload(decode=True)
1949 if body is None:
1950 body = b""
1951 elif isinstance(body, str):
1952 body = body.encode(encoding)
1953 elif not isinstance(body, bytes):
1954 # Handle multipart or other types
1955 body = str(body).encode(encoding)
1957 # Split into lines
1958 lines = body.splitlines(keepends=True)
1960 # Handle scissors
1961 scissors_idx = None
1962 if scissors:
1963 scissors_idx = _find_scissors_line(lines)
1964 if scissors_idx is not None:
1965 # Remove everything up to and including scissors line
1966 lines = lines[scissors_idx + 1 :]
1968 # Separate commit message from patch
1969 # Look for the "---" separator that indicates start of diffstat/patch
1970 message_lines: list[bytes] = []
1971 patch_lines: list[bytes] = []
1972 in_patch = False
1974 for line in lines:
1975 if not in_patch and line == b"---\n":
1976 in_patch = True
1977 patch_lines.append(line)
1978 elif in_patch:
1979 # Stop at signature marker "-- "
1980 if line == b"-- \n":
1981 break
1982 patch_lines.append(line)
1983 else:
1984 message_lines.append(line)
1986 # Build commit message
1987 commit_message = b"".join(message_lines).decode(encoding, errors="replace")
1989 # Clean up commit message
1990 commit_message = commit_message.strip()
1992 # Append Message-ID if requested
1993 if message_id and msg_id:
1994 if commit_message:
1995 commit_message += "\n\n"
1996 commit_message += f"Message-ID: {msg_id}"
1998 # Build patch content
1999 patch_content = b"".join(patch_lines).decode(encoding, errors="replace")
2001 return MailinfoResult(
2002 author_name=author_name,
2003 author_email=author_email,
2004 author_date=author_date,
2005 subject=subject,
2006 message=commit_message,
2007 patch=patch_content,
2008 message_id=msg_id,
2009 )