#!/usr/bin/python
# 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
import subprocess
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_str = subprocess.run(
["osc", "log", "--xml"], check=True, text=True, stdout=subprocess.PIPE
).stdout
except subprocess.CalledProcessError as exc:
raise RuntimeError(f"Cannot collect log of the package!") from exc
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:
subprocess.check_call(
["osc", "up", "-r", f"{rev}"], stdout=NULL, stderr=subprocess.PIPE
)
subprocess.check_call(["osc", "clean"], stdout=NULL, stderr=subprocess.PIPE)
except subprocess.CalledProcessError as exc:
raise RuntimeError(f"Cannot checkout revision {rev}!") 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")
