path: root/osc_fast_export.py

#!/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")