1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
|
#!/usr/bin/python3
# https://git-scm.com/book/en/v2/Git-and-Other-Systems-Migrating-to-Git#_custom_importer
import collections
import configparser
from datetime import datetime
import logging
import os.path
import pathlib
from subprocess import CalledProcessError, PIPE, run
import sys
from typing import Dict, List
import xml.etree.ElementTree as ET
logging.basicConfig(format="%(levelname)s:%(funcName)s:%(message)s",
level=logging.INFO)
log = logging.getLogger("osc_fast_export")
NULL = open(os.devnull, "wb")
# For reading section-less config files
# https://stackoverflow.com/a/2819788/164233
def FakeSecHead(fp):
yield "[asection]\n"
yield from fp
class LogEntry(
collections.namedtuple("LogEntry", ["rev", "md5", "author", "date", "msg"])
):
def __str__(self):
return (
f"{self.rev}, {self.md5[:12]}, {authors.get(self.author, '<none>')},"
+ f" {datetime.isoformat(self.date)}:\n{self.msg}"
)
def get_authors() -> Dict[str, str]:
config = configparser.ConfigParser()
authors = {}
authorsfile = pathlib.Path(".osc", "authorsfile.txt")
if authorsfile.exists():
config.read_file(FakeSecHead(open(authorsfile)))
authors = dict(config.items("asection"))
return authors
def osc_log() -> List[LogEntry]:
try:
log_pid = run(["osc", "log", "--xml"], check=True, text=True,
stdout=PIPE, stderr=PIPE
)
except CalledProcessError as exc:
raise RuntimeError(
f"Cannot collect log of the package!\nReported issue was:\n{log_pid.stderr}") from exc
log_str = log_pid.stdout
tree = ET.fromstring(log_str)
log_list = [
LogEntry(
int(entry.attrib["revision"]),
entry.attrib["srcmd5"],
entry.findtext("author"),
datetime.strptime(entry.findtext("date"), "%Y-%m-%d %H:%M:%S"),
entry.findtext("msg"),
)
for entry in tree.iter("logentry")
]
log_list.reverse()
return log_list
def checkout_revision(rev: int):
try:
osc_pid = run(
["osc", "up", "-r", f"{rev}"], stdout=PIPE, stderr=PIPE, check=True
)
except CalledProcessError as exc:
raise RuntimeError(
f"Cannot checkout revision {rev}!\nReported issue was:\n{osc_pid.stderr}") from exc
try:
osc_pid = run(["osc", "clean"], stdout=PIPE, stderr=PIPE, check=True)
except CalledProcessError as exc:
raise RuntimeError(
f"Cannot clean checkout!\nReported issue was:\n{osc_pid.stderr}") from exc
def print_export(entry: LogEntry, authors: Dict[str, str]) -> int:
mark = entry.rev
author = authors.get(entry.author, "<none>")
checkout_revision(entry.rev)
print("commit refs/heads/master")
print(f"mark :{mark}")
if entry.md5:
print(f"original-oid {entry.md5}")
print(f"committer {author} {entry.date:%s} +0000")
print(f"data {len(entry.msg)}\n{entry.msg}")
if last_mark:
print(f"from :{last_mark}")
# Create actual content of the commit
# It is easier just to wipe out everything and include files again.
print("deleteall")
for dirpath, dirnames, filenames in os.walk("."):
# TODO are osc notes (aka osc comment) a thing?
dirpath = os.path.relpath(dirpath, ".")
if dirpath.startswith(".osc"):
continue
# It seems git-fast-export doesn't export directories at all
# and git-fast-import just creates them when needed.
# if dirpath != '.':
# # create directory
# print(f'M 040000 inline {dirpath}')
for fn in filenames:
fname = os.path.relpath(os.path.join(dirpath, fn), ".")
log.debug("dirpath = %s, fname = %s", dirpath, fname)
fstat = f"{os.stat(fname).st_mode:o}"
print(f"M {fstat} inline {fname}")
with open(fname, "rb") as inf:
dt = inf.read()
print(f"data {len(dt)}")
sys.stdout.flush()
with os.fdopen(sys.stdout.fileno(), "wb",
closefd=False) as stdout:
stdout.write(dt)
stdout.flush()
print("")
return mark
if __name__ == "__main__":
last_mark = None
authors = get_authors()
for logentry in osc_log():
last_mark = print_export(logentry, authors)
print("done")
|
