git-importer/git-importer.py

#!/usr/bin/python3

import argparse
import logging
import pathlib
import shutil
import sys

import osc.core

from lib.exporter import Exporter
from lib.importer import Importer

URL_OBS = "https://api.opensuse.org"
URL_IBS = "https://api.suse.de"

# The order is relevant (from older to newer initial codebase)
PROJECTS = [
    ("openSUSE:Factory", "factory", URL_OBS),
    # ("SUSE:SLE-12:GA", "SLE_12", URL_IBS),
    # ("SUSE:SLE-12:Update", "SLE_12", URL_IBS),
    # ("SUSE:SLE-12-SP1:GA", "SLE_12_SP1", URL_IBS),
    # ("SUSE:SLE-12-SP1:Update", "SLE_12_SP1", URL_IBS),
    # ("SUSE:SLE-12-SP2:GA", "SLE_12_SP2", URL_IBS),
    # ("SUSE:SLE-12-SP2:Update", "SLE_12_SP2", URL_IBS),
    # ("SUSE:SLE-12-SP3:GA", "SLE_12_SP3", URL_IBS),
    # ("SUSE:SLE-12-SP3:Update", "SLE_12_SP3", URL_IBS),
    # ("SUSE:SLE-12-SP4:GA", "SLE_12_SP4", URL_IBS),
    # ("SUSE:SLE-12-SP4:Update", "SLE_12_SP4", URL_IBS),
    # ("SUSE:SLE-12-SP5:GA", "SLE_12_SP5", URL_IBS),
    # ("SUSE:SLE-12-SP5:Update", "SLE_12_SP5", URL_IBS),
    # ("SUSE:SLE-15:GA", "SLE_15", URL_IBS),
    # ("SUSE:SLE-15:Update", "SLE_15", URL_IBS),
    # ("SUSE:SLE-15-SP1:GA", "SLE_15_SP1", URL_IBS),
    # ("SUSE:SLE-15-SP1:Update", "SLE_15_SP1", URL_IBS),
    # ("SUSE:SLE-15-SP2:GA", "SLE_15_SP2", URL_IBS),
    # ("SUSE:SLE-15-SP2:Update", "SLE_15_SP2", URL_IBS),
    # ("SUSE:SLE-15-SP3:GA", "SLE_15_SP3", URL_IBS),
    # ("SUSE:SLE-15-SP3:Update", "SLE_15_SP3", URL_IBS),
    # ("SUSE:SLE-15-SP4:GA", "SLE_15_SP4", URL_IBS),
    # ("SUSE:SLE-15-SP4:Update", "SLE_15_SP4", URL_IBS),
]


def main():
    parser = argparse.ArgumentParser(description="OBS history importer into git")
    parser.add_argument("package", help="OBS package name")
    parser.add_argument(
        "-r",
        "--repodir",
        required=False,
        type=pathlib.Path,
        help="Local git repository directory",
    )
    parser.add_argument(
        "-f",
        "--force",
        action="store_true",
        help="If the repository directory exists, remove it",
    )
    parser.add_argument(
        "-a",
        "--search-ancestor",
        action="store_true",
        help="Search closest ancestor candidate for initial commit",
    )
    parser.add_argument(
        "-d",
        "--rebase-devel",
        action="store_true",
        help="The devel project with be rebased after a merge",
    )
    parser.add_argument(
        "-g",
        "--gc",
        metavar="N",
        type=int,
        default=200,
        help="Garbage recollect and pack the git history each N commits",
    )
    parser.add_argument(
        "--level",
        "-l",
        default="INFO",
        help="logging level",
    )
    parser.add_argument(
        "--db",
        action="store_true",
        help="Import revisions into database only",
    )
    parser.add_argument(
        "--export",
        action="store_true",
        help="Export database fields for the given package as JSON",
    )

    args = parser.parse_args()

    if args.level:
        numeric_level = getattr(logging, args.level.upper(), None)
        if not isinstance(numeric_level, int):
            print(f"Invalid log level: {args.level}")
            sys.exit(-1)
        logging.basicConfig(level=numeric_level)
        if numeric_level == logging.DEBUG:
            osc.conf.config["debug"] = True
            requests_log = logging.getLogger("requests.packages.urllib3")
            requests_log.setLevel(logging.DEBUG)
            requests_log.propagate = True

    if args.export:
        Exporter(args.package).run()
        return

    if not args.repodir:
        args.repodir = pathlib.Path(args.package)

    if args.repodir.exists() and not args.force:
        print(f"Repository {args.repodir} already present")
        sys.exit(-1)
    elif args.repodir.exists() and args.force:
        logging.info(f"Removing old repository {args.repodir}")
        shutil.rmtree(args.repodir)

    # TODO: use a CLI parameter to describe the projects
    importer = Importer(
        PROJECTS, args.package, args.repodir, args.search_ancestor, args.rebase_devel
    )

    if args.db:
        importer.import_into_db()
        return
    importer.import_all_revisions(args.gc)


if __name__ == "__main__":
    main()