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 _validate_patch_target(r: "Repo", repo_path: bytes, tree_path: bytes) -> bytes:
1386 """Validate a patch target path and return its filesystem path.
1388 ``_ensure_within_repo`` alone only refuses paths that escape the work tree;
1389 a target like ``.git/hooks/pre-commit`` stays inside it and would otherwise
1390 be written (and later executed as a hook). On top of that containment check,
1391 apply the same name and symlink checks git uses for checkout, so patch
1392 application cannot reach the control directory. Mirrors
1393 ``porcelain._checked_worktree_path``.
1395 Returns:
1396 The filesystem path under ``repo_path``, as bytes.
1397 """
1398 from .index import (
1399 InvalidPathError,
1400 get_path_element_validator,
1401 validate_path,
1402 verify_leading_dirs,
1403 )
1405 fs_path = os.path.join(repo_path, tree_path)
1406 _ensure_within_repo(repo_path, fs_path, tree_path)
1407 validator = get_path_element_validator(r.get_config_stack())
1408 if not validate_path(tree_path, validator):
1409 raise ValueError(f"refusing to write unsafe path: {tree_path!r}")
1410 try:
1411 verify_leading_dirs(tree_path, [], repo_path)
1412 except InvalidPathError:
1413 raise ValueError(f"refusing to write through symlink: {tree_path!r}")
1414 return fs_path
1417def _apply_rename_or_copy(
1418 r: "Repo",
1419 src_path: bytes,
1420 dst_path: bytes,
1421 strip: int,
1422 patch: FilePatch,
1423 is_rename: bool,
1424 cached: bool,
1425 check: bool,
1426 config: "Config | None",
1427) -> tuple[list[bytes] | None, bool]:
1428 """Apply a rename or copy operation.
1430 Args:
1431 r: Repository object
1432 src_path: Source path
1433 dst_path: Destination path
1434 strip: Number of path components to strip
1435 patch: FilePatch object
1436 is_rename: True for rename, False for copy
1437 cached: Apply to index only, not working tree
1438 check: Check only, don't apply
1439 config: Repository configuration
1441 Returns:
1442 A tuple of (``original_lines``, ``should_continue``) where:
1443 - ``original_lines``: Content lines if hunks need to be applied, None otherwise
1444 - ``should_continue``: True to skip to next patch, False to continue processing
1445 """
1446 from .index import (
1447 ConflictedIndexEntry,
1448 IndexEntry,
1449 cleanup_mode,
1450 index_entry_from_stat,
1451 )
1453 # Strip path components
1454 src_stripped = src_path
1455 dst_stripped = dst_path
1456 if strip > 0:
1457 src_parts = src_path.split(b"/")
1458 if len(src_parts) > strip:
1459 src_stripped = b"/".join(src_parts[strip:])
1460 dst_parts = dst_path.split(b"/")
1461 if len(dst_parts) > strip:
1462 dst_stripped = b"/".join(dst_parts[strip:])
1464 repo_path_bytes = r.path.encode("utf-8") if isinstance(r.path, str) else r.path
1465 src_fs_path = _validate_patch_target(r, repo_path_bytes, src_stripped)
1466 dst_fs_path = _validate_patch_target(r, repo_path_bytes, dst_stripped)
1468 # Read content from source file
1469 op_name = "rename" if is_rename else "copy"
1470 if os.path.exists(src_fs_path):
1471 with open(src_fs_path, "rb") as f:
1472 content = f.read()
1473 else:
1474 # Try to read from index
1475 index = r.open_index(config=config)
1476 if src_stripped in index:
1477 entry = index[src_stripped]
1478 if not isinstance(entry, ConflictedIndexEntry):
1479 obj = r.object_store[entry.sha]
1480 if isinstance(obj, Blob):
1481 content = obj.data
1482 else:
1483 raise ValueError(
1484 f"Cannot {op_name}: source {src_stripped.decode('utf-8', errors='replace')} not found"
1485 )
1486 else:
1487 raise ValueError(
1488 f"Cannot {op_name}: source {src_stripped.decode('utf-8', errors='replace')} is conflicted"
1489 )
1490 else:
1491 raise ValueError(
1492 f"Cannot {op_name}: source {src_stripped.decode('utf-8', errors='replace')} not found"
1493 )
1495 # If there are hunks, return content as lines for further processing
1496 if patch.hunks:
1497 return content.splitlines(keepends=True), False
1499 # No hunks - pure rename/copy
1500 if check:
1501 return None, True
1503 # Write to destination
1504 if not cached:
1505 os.makedirs(os.path.dirname(dst_fs_path), exist_ok=True)
1506 with open(dst_fs_path, "wb") as f:
1507 f.write(content)
1508 if patch.new_mode is not None:
1509 os.chmod(dst_fs_path, cleanup_mode(patch.new_mode))
1511 # Update index
1512 index = r.open_index(config=config)
1513 blob = Blob.from_string(content)
1514 r.object_store.add_object(blob)
1516 if not cached and os.path.exists(dst_fs_path):
1517 st = os.stat(dst_fs_path)
1518 entry = index_entry_from_stat(st, blob.id)
1519 else:
1520 entry = IndexEntry(
1521 ctime=(0, 0),
1522 mtime=(0, 0),
1523 dev=0,
1524 ino=0,
1525 mode=patch.new_mode or 0o100644,
1526 uid=0,
1527 gid=0,
1528 size=len(content),
1529 sha=blob.id,
1530 flags=0,
1531 )
1533 index[dst_stripped] = entry
1535 # For renames, remove the old file
1536 if is_rename:
1537 if not cached and os.path.exists(src_fs_path):
1538 os.remove(src_fs_path)
1539 if src_stripped in index:
1540 del index[src_stripped]
1542 index.write()
1543 return None, True
1546def apply_patches(
1547 r: "Repo",
1548 patches: list[FilePatch],
1549 cached: bool = False,
1550 reverse: bool = False,
1551 check: bool = False,
1552 strip: int = 1,
1553 three_way: bool = False,
1554 *,
1555 config: "Config | None" = None,
1556) -> None:
1557 """Apply a list of file patches to a repository.
1559 Args:
1560 r: Repository object
1561 patches: List of FilePatch objects to apply
1562 cached: Apply patch to index only, not working tree
1563 reverse: Apply patch in reverse
1564 check: Only check if patch can be applied, don't apply
1565 strip: Number of leading path components to strip (default: 1)
1566 three_way: Fall back to 3-way merge if patch does not apply cleanly
1567 config: Repository configuration. If None, falls back to
1568 ``r.get_config_stack()``.
1570 Raises:
1571 ValueError: If patch cannot be applied
1572 """
1573 from .index import (
1574 ConflictedIndexEntry,
1575 IndexEntry,
1576 cleanup_mode,
1577 index_entry_from_stat,
1578 )
1580 if config is None:
1581 config = r.get_config_stack()
1583 for patch in patches:
1584 # Determine the file path
1585 # For renames/copies without hunks, old_path/new_path may be None
1586 # Use local variables to avoid mutating the patch object
1587 old_path = patch.old_path
1588 new_path = patch.new_path
1590 if new_path is None and old_path is None:
1591 if patch.rename_to is not None:
1592 # Use rename_to for the target path
1593 new_path = patch.rename_to
1594 old_path = patch.rename_from
1595 elif patch.copy_to is not None:
1596 # Use copy_to for the target path
1597 new_path = patch.copy_to
1598 old_path = patch.copy_from
1599 else:
1600 raise ValueError("Patch has no file path")
1602 # Choose path based on operation
1603 file_path: bytes
1604 if new_path is None:
1605 # Deletion
1606 if old_path is None:
1607 raise ValueError("Patch has no file path")
1608 file_path = old_path
1609 elif old_path is None:
1610 # Addition
1611 file_path = new_path
1612 else:
1613 # Modification (use new path)
1614 file_path = new_path
1616 # Strip path components
1617 if strip > 0:
1618 parts = file_path.split(b"/")
1619 if len(parts) > strip:
1620 file_path = b"/".join(parts[strip:])
1622 # Convert to filesystem path
1623 tree_path = file_path
1624 repo_path_bytes = r.path.encode("utf-8") if isinstance(r.path, str) else r.path
1625 fs_path = _validate_patch_target(r, repo_path_bytes, file_path)
1627 # Handle renames and copies
1628 original_lines: list[bytes] | None = None
1629 if patch.rename_from is not None and patch.rename_to is not None:
1630 original_lines, should_continue = _apply_rename_or_copy(
1631 r,
1632 patch.rename_from,
1633 patch.rename_to,
1634 strip,
1635 patch,
1636 is_rename=True,
1637 cached=cached,
1638 check=check,
1639 config=config,
1640 )
1641 if should_continue:
1642 continue
1643 elif patch.copy_from is not None and patch.copy_to is not None:
1644 original_lines, should_continue = _apply_rename_or_copy(
1645 r,
1646 patch.copy_from,
1647 patch.copy_to,
1648 strip,
1649 patch,
1650 is_rename=False,
1651 cached=cached,
1652 check=check,
1653 config=config,
1654 )
1655 if should_continue:
1656 continue
1658 # Handle binary patches
1659 if patch.binary:
1660 if patch.binary_new is not None:
1661 # Decode binary patch
1662 try:
1663 binary_content = git_base85_decode(patch.binary_new)
1664 except (ValueError, KeyError) as e:
1665 raise ValueError(f"Failed to decode binary patch: {e}")
1667 if check:
1668 # Just checking, don't actually apply
1669 continue
1671 # Write binary file
1672 if not cached:
1673 os.makedirs(os.path.dirname(fs_path), exist_ok=True)
1674 with open(fs_path, "wb") as f:
1675 f.write(binary_content)
1676 if patch.new_mode is not None:
1677 os.chmod(fs_path, cleanup_mode(patch.new_mode))
1679 # Update index
1680 index = r.open_index(config=config)
1681 blob = Blob.from_string(binary_content)
1682 r.object_store.add_object(blob)
1684 if not cached and os.path.exists(fs_path):
1685 st = os.stat(fs_path)
1686 entry = index_entry_from_stat(st, blob.id)
1687 else:
1688 entry = IndexEntry(
1689 ctime=(0, 0),
1690 mtime=(0, 0),
1691 dev=0,
1692 ino=0,
1693 mode=patch.new_mode or 0o100644,
1694 uid=0,
1695 gid=0,
1696 size=len(binary_content),
1697 sha=blob.id,
1698 flags=0,
1699 )
1701 index[tree_path] = entry
1702 index.write()
1703 continue
1704 else:
1705 # Old-style "Binary files differ" message without actual patch data
1706 raise NotImplementedError(
1707 "Binary patch detected but no patch data provided (use git diff --binary)"
1708 )
1710 # Read original file content (unless already loaded from rename/copy)
1711 if original_lines is None:
1712 if patch.old_path is None:
1713 # New file
1714 original_lines = []
1715 else:
1716 if os.path.exists(fs_path):
1717 with open(fs_path, "rb") as f:
1718 content = f.read()
1719 original_lines = content.splitlines(keepends=True)
1720 else:
1721 # File doesn't exist - check if it's in the index
1722 try:
1723 index = r.open_index(config=config)
1724 if tree_path in index:
1725 index_entry: IndexEntry | ConflictedIndexEntry = index[
1726 tree_path
1727 ]
1728 if not isinstance(index_entry, ConflictedIndexEntry):
1729 obj = r.object_store[index_entry.sha]
1730 if isinstance(obj, Blob):
1731 original_lines = obj.data.splitlines(keepends=True)
1732 else:
1733 original_lines = []
1734 else:
1735 original_lines = []
1736 else:
1737 original_lines = []
1738 except (KeyError, FileNotFoundError):
1739 original_lines = []
1741 # Reverse patch if requested
1742 if reverse:
1743 # Swap old and new in hunks
1744 for hunk in patch.hunks:
1745 hunk.old_start, hunk.new_start = hunk.new_start, hunk.old_start
1746 hunk.old_count, hunk.new_count = hunk.new_count, hunk.old_count
1747 # Swap +/- prefixes
1748 reversed_lines = []
1749 for line in hunk.lines:
1750 if line.startswith(b"+"):
1751 reversed_lines.append(b"-" + line[1:])
1752 elif line.startswith(b"-"):
1753 reversed_lines.append(b"+" + line[1:])
1754 else:
1755 reversed_lines.append(line)
1756 hunk.lines = reversed_lines
1758 # Apply the patch
1759 assert original_lines is not None
1760 result = apply_patch_hunks(patch, original_lines)
1762 if result is None and three_way:
1763 # Try 3-way merge fallback
1764 from .merge import merge_blobs
1766 # Reconstruct base version from the patch
1767 # Base is what you get by taking only the old lines from hunks
1768 base_lines = []
1769 theirs_lines = []
1771 for hunk in patch.hunks:
1772 for line in hunk.lines:
1773 if line.startswith(b"\\"):
1774 # Skip "\ No newline at end of file" markers
1775 continue
1776 elif line.startswith(b" "):
1777 # Context line - in both base and theirs
1778 content = line[1:]
1779 if not content.endswith(b"\n"):
1780 content += b"\n"
1781 base_lines.append(content)
1782 theirs_lines.append(content)
1783 elif line.startswith(b"-"):
1784 # Deletion - only in base
1785 content = line[1:]
1786 if not content.endswith(b"\n"):
1787 content += b"\n"
1788 base_lines.append(content)
1789 elif line.startswith(b"+"):
1790 # Addition - only in theirs
1791 content = line[1:]
1792 if not content.endswith(b"\n"):
1793 content += b"\n"
1794 theirs_lines.append(content)
1796 # Create blobs for merging
1797 base_content = b"".join(base_lines)
1798 ours_content = b"".join(original_lines)
1799 theirs_content = b"".join(theirs_lines)
1801 base_blob = Blob.from_string(base_content) if base_content else None
1802 ours_blob = Blob.from_string(ours_content) if ours_content else None
1803 theirs_blob = Blob.from_string(theirs_content)
1805 # Perform 3-way merge
1806 merged_content, _had_conflicts = merge_blobs(
1807 base_blob, ours_blob, theirs_blob, path=tree_path
1808 )
1810 result = merged_content.splitlines(keepends=True)
1812 # Note: if _had_conflicts is True, the result contains conflict markers
1813 # Git would exit with error code, but we continue processing
1814 elif result is None:
1815 raise PatchApplicationFailure(
1816 f"Patch does not apply to {file_path.decode('utf-8', errors='replace')}"
1817 )
1819 if check:
1820 # Just checking, don't actually apply
1821 continue
1823 # Write result
1824 result_content = b"".join(result)
1826 if patch.new_path is None:
1827 # File deletion
1828 if not cached and os.path.exists(fs_path):
1829 os.remove(fs_path)
1830 # Remove from index
1831 index = r.open_index(config=config)
1832 if tree_path in index:
1833 del index[tree_path]
1834 index.write()
1835 else:
1836 # File addition or modification
1837 if not cached:
1838 # Write to working tree
1839 os.makedirs(os.path.dirname(fs_path), exist_ok=True)
1840 with open(fs_path, "wb") as f:
1841 f.write(result_content)
1843 # Update file mode if specified
1844 if patch.new_mode is not None:
1845 os.chmod(fs_path, cleanup_mode(patch.new_mode))
1847 # Update index
1848 index = r.open_index(config=config)
1849 blob = Blob.from_string(result_content)
1850 r.object_store.add_object(blob)
1852 # Get file stat for index entry
1853 if not cached and os.path.exists(fs_path):
1854 st = os.stat(fs_path)
1855 entry = index_entry_from_stat(st, blob.id)
1856 else:
1857 # Create a minimal index entry for cached-only changes
1858 entry = IndexEntry(
1859 ctime=(0, 0),
1860 mtime=(0, 0),
1861 dev=0,
1862 ino=0,
1863 mode=patch.new_mode or 0o100644,
1864 uid=0,
1865 gid=0,
1866 size=len(result_content),
1867 sha=blob.id,
1868 flags=0,
1869 )
1871 index[tree_path] = entry
1873 # Handle cleanup for renames with hunks
1874 if patch.rename_from is not None and patch.rename_to is not None:
1875 # Remove old file after successful rename
1876 old_rename_path = patch.rename_from
1877 if strip > 0:
1878 old_parts = old_rename_path.split(b"/")
1879 if len(old_parts) > strip:
1880 old_rename_path = b"/".join(old_parts[strip:])
1882 old_fs_path = os.path.join(
1883 r.path.encode("utf-8") if isinstance(r.path, str) else r.path,
1884 old_rename_path,
1885 )
1887 if not cached and os.path.exists(old_fs_path):
1888 os.remove(old_fs_path)
1889 if old_rename_path in index:
1890 del index[old_rename_path]
1892 index.write()
1895def mailinfo(
1896 msg: email.message.Message | BinaryIO | TextIO,
1897 keep_subject: bool = False,
1898 keep_non_patch: bool = False,
1899 encoding: str | None = None,
1900 scissors: bool = False,
1901 message_id: bool = False,
1902) -> MailinfoResult:
1903 """Extract patch information from an email message.
1905 This function parses an email message and extracts commit metadata
1906 (author, email, subject) and separates the commit message from the
1907 patch content, similar to git mailinfo.
1909 Args:
1910 msg: Email message (email.message.Message object) or file handle to read from
1911 keep_subject: If True, keep subject intact without munging (-k)
1912 keep_non_patch: If True, only strip [PATCH] from brackets (-b)
1913 encoding: Character encoding to use (default: detect from message)
1914 scissors: If True, remove everything before scissors line
1915 message_id: If True, include Message-ID in commit message (-m)
1917 Returns:
1918 MailinfoResult with parsed information
1920 Raises:
1921 ValueError: If message is malformed or missing required fields
1922 """
1923 # Parse message if given a file handle
1924 parsed_msg: email.message.Message
1925 if not isinstance(msg, email.message.Message):
1926 if hasattr(msg, "read"):
1927 content = msg.read()
1928 if isinstance(content, bytes):
1929 bparser = email.parser.BytesParser()
1930 parsed_msg = bparser.parsebytes(content)
1931 else:
1932 sparser = email.parser.Parser()
1933 parsed_msg = sparser.parsestr(content)
1934 else:
1935 raise ValueError("msg must be an email.message.Message or file-like object")
1936 else:
1937 parsed_msg = msg
1939 # Detect encoding from message if not specified
1940 if encoding is None:
1941 encoding = parsed_msg.get_content_charset() or "utf-8"
1943 # Extract author information
1944 from_header = parsed_msg.get("From", "")
1945 if not from_header:
1946 raise ValueError("Email message missing 'From' header")
1948 # Parse "Name <email>" format
1949 author_name, author_email = email.utils.parseaddr(from_header)
1950 if not author_email:
1951 raise ValueError(
1952 f"Could not parse email address from 'From' header: {from_header}"
1953 )
1955 # Extract date
1956 date_header = parsed_msg.get("Date")
1957 author_date = date_header if date_header else None
1959 # Extract and process subject
1960 subject = parsed_msg.get("Subject", "")
1961 if not subject:
1962 subject = "(no subject)"
1964 # Convert Header object to string if needed
1965 subject = str(subject)
1967 # Remove newlines from subject
1968 subject = subject.replace("\n", " ").replace("\r", " ")
1969 subject = _munge_subject(subject, keep_subject, keep_non_patch)
1971 # Extract Message-ID if requested
1972 msg_id = None
1973 if message_id:
1974 msg_id = parsed_msg.get("Message-ID")
1976 # Get message body
1977 body = parsed_msg.get_payload(decode=True)
1978 if body is None:
1979 body = b""
1980 elif isinstance(body, str):
1981 body = body.encode(encoding)
1982 elif not isinstance(body, bytes):
1983 # Handle multipart or other types
1984 body = str(body).encode(encoding)
1986 # Split into lines
1987 lines = body.splitlines(keepends=True)
1989 # Handle scissors
1990 scissors_idx = None
1991 if scissors:
1992 scissors_idx = _find_scissors_line(lines)
1993 if scissors_idx is not None:
1994 # Remove everything up to and including scissors line
1995 lines = lines[scissors_idx + 1 :]
1997 # Separate commit message from patch
1998 # Look for the "---" separator that indicates start of diffstat/patch
1999 message_lines: list[bytes] = []
2000 patch_lines: list[bytes] = []
2001 in_patch = False
2003 for line in lines:
2004 if not in_patch and line == b"---\n":
2005 in_patch = True
2006 patch_lines.append(line)
2007 elif in_patch:
2008 # Stop at signature marker "-- "
2009 if line == b"-- \n":
2010 break
2011 patch_lines.append(line)
2012 else:
2013 message_lines.append(line)
2015 # Build commit message
2016 commit_message = b"".join(message_lines).decode(encoding, errors="replace")
2018 # Clean up commit message
2019 commit_message = commit_message.strip()
2021 # Append Message-ID if requested
2022 if message_id and msg_id:
2023 if commit_message:
2024 commit_message += "\n\n"
2025 commit_message += f"Message-ID: {msg_id}"
2027 # Build patch content
2028 patch_content = b"".join(patch_lines).decode(encoding, errors="replace")
2030 return MailinfoResult(
2031 author_name=author_name,
2032 author_email=author_email,
2033 author_date=author_date,
2034 subject=subject,
2035 message=commit_message,
2036 patch=patch_content,
2037 message_id=msg_id,
2038 )