import io, sys, os, operator, re, base64, hashlib
import patchtheory
def strip_subject(s):
if not s.startswith(b'[PATCH'):
raise ValueError(f'Invalid subject: {s!r}')
i = s.find(b'] ')
if i == -1:
raise ValueError(f'Invalid subject: {s!r}')
return s[i+2:]
rem_patch = re.compile('patch ([0-9a-z]{16})').fullmatch
rem_author = re.compile('[^\0-\x1f]+ <[^\0-\x1f]+@[^\0-\x1f]+>').fullmatch
rem_date = re.compile('[^\0-\x1f]+').fullmatch
rem_title = re.compile('[^\0-\x1f]+').fullmatch
class Patch:
__slots__ = ('author', 'date', 'title', 'body', 'diff', 'data', 'hash', 'id')
def __init__(self, /, *, author, date, title, body, diff):
assert type(author) is str and rem_author(author) is not None
assert type(date) is str and rem_date(date) is not None
assert type(title) is str and rem_title(title) is not None
assert type(body) is str
assert type(diff) is patchtheory.Diff
self.author = author
self.date = date
self.title = title
self.body = body
self.diff = diff
self._set_data()
self._set_id()
@classmethod
def parse(tp, body):
lines = iter(body.split(b'\n'))
line = lines.__next__
l = line()
if l.startswith(b'From '):
l = line()
headers = {}
hdr = None
while True:
if not l:
break
if hdr is not None and l.startswith(b' '):
headers[hdr] += l
else:
hdr,sep,val = l.partition(b':')
if not sep:
raise ValueError(f'Expected header line: {l!r}')
headers[hdr] = val.lstrip()
l = line()
author = headers.pop(b'From').decode('utf-8')
date = headers.pop(b'Date').decode('utf-8')
title = strip_subject(headers.pop(b'Subject')).decode('utf-8')
headers.pop(b'MIME-Version', None)
headers.pop(b'Content-Type', None)
headers.pop(b'Content-Transfer-Encoding', None)
body = []
for l in lines:
if l.startswith(b'diff --git'):
break
body.append(l)
else:
raise RuntimeError('unexpected EOF')
while not body[-1]:
del body[-1]
while body[-1].startswith(b' '):
del body[-1]
if body[-1] != b'---':
raise RuntimeError
del body[-1]
while body and not body[-1]:
del body[-1]
body = b'\n'.join(body).decode('utf-8')
return Patch(
author = author,
date = date,
title = title,
body = body,
diff = patchtheory.Diff.parse_l(l, line),
)
def _set_id(self):
out = io.BytesIO()
w = out.write
w(self.author.encode('utf-8'))
w(b'\n')
w(self.date.encode('utf-8'))
w(b'\n')
w(self.title.encode('utf-8'))
w(b'\n')
self.diff.write_munged(w)
out = out.getvalue()
dgst = hashlib.blake2b(out).digest()
self.id = base64.b32encode(dgst[:10]).lower().decode('ascii')
def as_filename(self):
return '-'.join(re.findall('[0-9A-Za-z]+', self.title))[:72]
def _set_data(self):
out = io.BytesIO()
w = out.write
w(b'From: ')
w(self.author.encode('utf-8'))
w(b'\nDate: ')
w(self.date.encode('utf-8'))
w(b'\nSubject: [PATCH] ')
w(self.title.encode('utf-8'))
w(b'\n\n')
w(self.body.encode('utf-8'))
w(b'\n---\n\n')
self.diff.write(w)
w(b'-- \n0.0.0 patchstate\n\n')
self.data = data = out.getvalue()
self.hash = base64.b32encode(hashlib.blake2b(data).digest()[:20]).lower().decode('ascii')
def revert(self):
r = Patch(
author = self.author,
date = self.date,
title = f'Revert "{self.title}"',
body = self.body,
diff = -self.diff,
)
r._set_data()
r._set_id()
return r
class PatchInfo:
__slots__ = ('patch_id', '_body')
def __init__(self, patch_id, info):
self.patch_id = patch_id
self._body = info
def get_patch(self, repo):
p = repo.patches[self.patch_hash]
if p.id != self.patch_id:
raise KeyError(f'patch id mismatch: {self.patch_id!r} -> {p.id!r}')
if p.title != self.patch_title:
raise KeyError(f'patch title mismatch: {self.patch_title!r} -> {p.title!r}')
return p
@property
def patch_hash(self):
return self._body[0]
@property
def patch_title(self):
return self._body[1]
@property
def mode(self):
return self._body[2]
@property
def info(self):
return self._body[3:]
def update(self, *, patch=None, patch_id=None, patch_hash=None, patch_title=None, mode=None, new_mode=None, info=None):
if patch_title is None:
patch_title = self.patch_title
if patch is not None:
assert patch_id is None
assert patch_hash is None
if patch_title != patch.title:
raise ValueError('patch title change: {patch_title!r} -> {patch.title!r}')
patch_id = patch.id
patch_hash = patch.hash
else:
if patch_id is None:
patch_id = self.patch_id
if patch_hash is None:
patch_hash = self.patch_hash
if info is None:
info = self.info
if mode is None:
mode = self.mode
if new_mode is not None and new_mode != mode:
info.insert(0, f'(was {mode})')
mode = new_mode
return PatchInfo(patch_id, [patch_hash, patch_title, mode, *info])
class Series:
__slots__ = ('info',)
def __init__(self, /, info=None):
if info is None:
info = []
self.info = info
@classmethod
def parse(tp, data):
inf = []
lines = iter(data.split('\n'))
l = next(lines)
while True:
if not l:
try:
l = next(lines)
except StopIteration:
break
continue
m = rem_patch(l)
if m is None:
raise ValueError(f'Invalid header: {l!r}')
pid = m.group(1)
ldata = []
while True:
l = next(lines)
if not l.startswith('\t'):
break
ldata.append(l[1:])
if len(ldata) < 3:
raise ValueError(f'Expected data after {f"patch {pid}"!r}')
while ldata and not ldata[-1]:
del ldata[-1]
inf.append(PatchInfo(pid, ldata))
return tp(info=inf)
def fmt(self):
out = io.StringIO()
w = out.write
for pi in self.info:
w(f'patch {pi.patch_id}')
for line in pi._body:
w('\n\t')
w(line)
w('\n')
return out.getvalue()
class Repository:
__slots__ = ('path', 'patches', 'aliases', 'by_id')
def __init__(self, path):
self.path = path
self.patches = {}
self.by_id = {}
aliases = {}
for fn in os.listdir(path):
if not fn.endswith('.patch'):
continue
fp = os.path.join(path, fn)
with io.open(fp, 'rb') as f:
data = f.read()
try:
p = Patch.parse(data)
except ValueError as exc:
print(f'Failed to parse {fn!r}: {exc}', file=sys.stderr)
os.unlink(fp)
continue
h = p.hash
if fn != f'{h}.patch':
aliases[fn[:-6]] = (h,p)
else:
self.patches[h] = p
self.by_id.setdefault(p.id, []).append(p)
for h,(h2,p) in aliases.items():
assert h != h2
assert h not in self.patches
if h2 in self.patches:
continue
print(f'Migrating patch: {h!r} -> {h2!r}', file=sys.stderr)
self.patches[h2] = p
with io.open(os.path.join(self.path, f'{h2}.patch'), 'xb') as f:
f.write(p.data)
self.aliases = {h: h2 for h,(h2,p) in aliases.items()}
def hash_data(self, data):
data = hashlib.blake2b(data).digest()
return base64.b32encode(data[:20]).lower().decode()
def insert_patch(self, patch):
p = self.patches.setdefault(patch.hash, patch)
if p is patch:
pid = p.id
try:
pd = self.by_id[pid]
except KeyError:
self.by_id[pid] = [p]
else:
print(f'Duplicate patch ID: {pid!r} {p.title}', file=sys.stderr)
pd.append(p)
with io.open(os.path.join(self.path, f'{p.hash}.patch'), 'xb') as f:
f.write(p.data)
return p
def import_dir(self, path):
r = Series()
r.info = inf = []
for fn in sorted(os.listdir(path)):
if not fn.endswith('.patch'):
print(f'Skipping {fn!r}', file=sys.stderr)
continue
with io.open(os.path.join(path, fn), 'rb') as f:
patch = Patch.parse(f.read())
patch = self.insert_patch(patch)
inf.append(PatchInfo(patch.id, [patch.hash, patch.title, 'new']))
return r
def gc(self, keep):
for h in self.patches.keys() - keep:
del self.patches[h]
print(f'Removing patch: {h!r}', file=sys.stderr)
os.unlink(os.path.join(self.path, f'{h}.patch'))
def argparse_next(args, func=...):
def do_argparse(func):
while True:
try:
arg = next(args)
except StopIteration:
return None
if arg == '--':
return next(args, None)
if not arg.startswith('-'):
return arg
arg = func(arg)
if arg is not None:
return arg
if func is not ...:
return do_argparse(func)
return do_argparse
def argparse_all(args, func=...):
def do_argparse(func):
r = []
while True:
try:
arg = next(args)
except StopIteration:
return r
if arg == '--':
r.extend(args)
return r
if not arg.startswith('-'):
r.append(arg)
continue
arg = func(arg)
if arg is not None:
r.extend(arg)
if func is not ...:
return do_argparse(func)
return do_argparse
