123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149 |
- """Utilities for reading and generating reflogs."""
- import collections
- from .objects import ZERO_SHA, format_timezone, parse_timezone
- Entry = collections.namedtuple(
- "Entry",
- ["old_sha", "new_sha", "committer", "timestamp", "timezone", "message"],
- )
- def format_reflog_line(old_sha, new_sha, committer, timestamp, timezone, message):
- """Generate a single reflog line.
- Args:
- old_sha: Old Commit SHA
- new_sha: New Commit SHA
- committer: Committer name and e-mail
- timestamp: Timestamp
- timezone: Timezone
- message: Message
- """
- if old_sha is None:
- old_sha = ZERO_SHA
- return (
- old_sha
- + b" "
- + new_sha
- + b" "
- + committer
- + b" "
- + str(int(timestamp)).encode("ascii")
- + b" "
- + format_timezone(timezone)
- + b"\t"
- + message
- )
- def parse_reflog_line(line):
- """Parse a reflog line.
- Args:
- line: Line to parse
- Returns: Tuple of (old_sha, new_sha, committer, timestamp, timezone,
- message)
- """
- (begin, message) = line.split(b"\t", 1)
- (old_sha, new_sha, rest) = begin.split(b" ", 2)
- (committer, timestamp_str, timezone_str) = rest.rsplit(b" ", 2)
- return Entry(
- old_sha,
- new_sha,
- committer,
- int(timestamp_str),
- parse_timezone(timezone_str)[0],
- message,
- )
- def read_reflog(f):
- """Read reflog.
- Args:
- f: File-like object
- Returns: Iterator over Entry objects
- """
- for line in f:
- yield parse_reflog_line(line)
- def drop_reflog_entry(f, index, rewrite=False):
- """Drop the specified reflog entry.
- Args:
- f: File-like object
- index: Reflog entry index (in Git reflog reverse 0-indexed order)
- rewrite: If a reflog entry's predecessor is removed, set its
- old SHA to the new SHA of the entry that now precedes it
- """
- if index < 0:
- raise ValueError("Invalid reflog index %d" % index)
- log = []
- offset = f.tell()
- for line in f:
- log.append((offset, parse_reflog_line(line)))
- offset = f.tell()
- inverse_index = len(log) - index - 1
- write_offset = log[inverse_index][0]
- f.seek(write_offset)
- if index == 0:
- f.truncate()
- return
- del log[inverse_index]
- if rewrite and index > 0 and log:
- if inverse_index == 0:
- previous_new = ZERO_SHA
- else:
- previous_new = log[inverse_index - 1][1].new_sha
- offset, entry = log[inverse_index]
- log[inverse_index] = (
- offset,
- Entry(
- previous_new,
- entry.new_sha,
- entry.committer,
- entry.timestamp,
- entry.timezone,
- entry.message,
- ),
- )
- for _, entry in log[inverse_index:]:
- f.write(
- format_reflog_line(
- entry.old_sha,
- entry.new_sha,
- entry.committer,
- entry.timestamp,
- entry.timezone,
- entry.message,
- )
- )
- f.truncate()
|