aboutsummaryrefslogtreecommitdiffstats
path: root/osc_fast_export.py
blob: 0042100c924186319c98285e839b8b94624d9bc2 (plain) (blame)
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
#!/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 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")

authorsfile = pathlib.Path(".git", "authorsfile.txt")

# For reading section-less config files
# https://stackoverflow.com/a/2819788/164233
def FakeSecHead(fp):
    yield "[asection]\n"
    yield from fp


config = configparser.ConfigParser()
authors = {}
if authorsfile.exists():
    config.read_file(FakeSecHead(open(authorsfile)))
    authors = dict(config.items("asection"))


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 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):
    NULL = open(os.devnull, "wb")
    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) -> 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

    for logentry in osc_log():
        last_mark = print_export(logentry)
    print("done")