breaker

#!/usr/bin/env python3
# Usage: breaker [package-name] --rebuilds --pkgbreak --bump
# Steps to bump package with shared library:
# 0. Change directory to the root of aosc-os-abbs clone (e.g. /buildroots/xxx/TREE)
# 1. Update package
# 2. Run breaker --pkgbreak (do not generate groups/xxx-rebuilds)
#    or breaker --rebuilds --pkgbreak (generate groups/xxx-rebuilds)
# 3. Copy PKGBREAK to package defines
# 4. Commit
# 5. Run breaker --bump
# You can specify packages to exclude via: --exclude a b c

import requests
import os
import argparse
import logging
import re
import subprocess

logger = logging.getLogger("breaker")


# Locate package & autobuild path for specified package
def search_package_path(package_name: str) -> tuple[str, str]:
    with os.scandir(".") as dir1:
        for section in dir1:
            if section.is_dir() and not section.name.startswith("."):
                with os.scandir(section) as dir2:
                    for package in dir2:
                        if package.is_dir() and os.path.isdir(
                            os.path.join(package, "autobuild")
                        ):
                            defines_path = os.path.join(
                                package.path, "autobuild/defines"
                            )
                            if os.path.exists(defines_path):
                                with open(defines_path, "r") as f:
                                    defines = f.readlines()
                                    for line in defines:
                                        if "PKGNAME=" in line and (
                                            "{}\n".format(package_name) == line[8:]
                                            or '"{}"\n'.format(package_name) == line[8:]
                                        ):
                                            return (
                                                package.path[2:],
                                                defines_path[2:],
                                            )

                            continue

                        # search subpackage, like arch-install-scripts/01-genfstab
                        path = package
                        if os.path.isdir(path) and section.name != "groups":
                            with os.scandir(path) as dir3:
                                for subpackage in dir3:
                                    if (
                                        subpackage.name != "autobuild"
                                        and subpackage.is_dir()
                                    ):
                                        defines_path = os.path.join(
                                            subpackage, "defines"
                                        )
                                        try:
                                            with open(defines_path, "r") as f:
                                                defines = f.readlines()
                                        except:
                                            with open(
                                                os.path.join(
                                                    subpackage, "autobuild/defines"
                                                ),
                                                "r",
                                            ) as f:
                                                defines = f.readlines()
                                        finally:
                                            for line in defines:
                                                if "PKGNAME=" in line and (
                                                    "{}\n".format(package_name)
                                                    == line[8:]
                                                    or '"{}"\n'.format(package_name)
                                                    == line[8:]
                                                ):
                                                    return (
                                                        package.path[2:],
                                                        defines_path[2:],
                                                    )
    # not found
    return None, None


# Get version of package
def get_package_version(package_path: str, defines_path: str) -> str:
    ver = None
    rel = None
    pkgepoch = None

    spec_path = os.path.join(package_path, "spec")
    with open(spec_path, "r") as f:
        for i in f:
            if i.startswith("VER="):
                ver = i.replace("VER=", "").strip()
            if i.startswith("REL="):
                rel = i.replace("REL=", "").strip()

    with open(defines_path, "r") as f:
        for i in f:
            if i.startswith("PKGEPOCH="):
                pkgepoch = i.replace("PKGEPOCH=", "").strip()

    res = None
    if ver is not None:
        res = ver
        if rel is not None:
            res += "-" + rel
        if pkgepoch is not None:
            res = pkgepoch + ":" + res

    return res


# Locate package & autobuild path for specified package
def search_package_path(package_name: str) -> tuple[str, str]:
    with os.scandir(".") as dir1:
        for section in dir1:
            if section.is_dir() and not section.name.startswith("."):
                with os.scandir(section) as dir2:
                    for package in dir2:
                        if package.is_dir() and os.path.isdir(
                            os.path.join(package, "autobuild")
                        ):
                            defines_path = os.path.join(
                                package.path, "autobuild/defines"
                            )
                            if os.path.exists(defines_path):
                                with open(defines_path, "r") as f:
                                    defines = f.readlines()
                                    for line in defines:
                                        if "PKGNAME=" in line and (
                                            "{}\n".format(package_name) == line[8:]
                                            or '"{}"\n'.format(package_name) == line[8:]
                                        ):
                                            return (
                                                package.path[2:],
                                                defines_path[2:],
                                            )

                            continue

                        # search subpackage, like arch-install-scripts/01-genfstab
                        path = package
                        if os.path.isdir(path) and section.name != "groups":
                            with os.scandir(path) as dir3:
                                for subpackage in dir3:
                                    if (
                                        subpackage.name != "autobuild"
                                        and subpackage.is_dir()
                                    ):
                                        defines_path = os.path.join(
                                            subpackage, "defines"
                                        )
                                        try:
                                            with open(defines_path, "r") as f:
                                                defines = f.readlines()
                                        except:
                                            with open(
                                                os.path.join(
                                                    subpackage, "autobuild/defines"
                                                ),
                                                "r",
                                            ) as f:
                                                defines = f.readlines()
                                        finally:
                                            for line in defines:
                                                if "PKGNAME=" in line and (
                                                    "{}\n".format(package_name)
                                                    == line[8:]
                                                    or '"{}"\n'.format(package_name)
                                                    == line[8:]
                                                ):
                                                    return (
                                                        package.path[2:],
                                                        defines_path[2:],
                                                    )
    # not found
    return None, None


# Get version of package
def get_package_version(package_path: str, defines_path: str) -> str:
    ver = None
    rel = None
    pkgepoch = None

    spec_path = os.path.join(package_path, "spec")
    with open(spec_path, "r") as f:
        for i in f:
            if i.startswith("VER="):
                ver = i.replace("VER=", "").strip()
            if i.startswith("REL="):
                rel = i.replace("REL=", "").strip()

    with open(defines_path, "r") as f:
        for i in f:
            if i.startswith("PKGEPOCH="):
                pkgepoch = i.replace("PKGEPOCH=", "").strip()

    res = None
    if ver is not None:
        res = ver
        if rel is not None:
            res += "-" + rel
        if pkgepoch is not None:
            res = pkgepoch + ":" + res

    return res


# Get reverse library dependency for specified package
def get_revdeps(
    package_name: str, alldeps: bool, include: list[str], exclude: list[str],
    sorevdeps: list[str]
) -> list[dict[str, str]]:
    revdeps_list = []
    try:
        response = requests.get(
            "https://packages.aosc.io/revdep/{}?type=json".format(package_name)
        )
        response.raise_for_status()
        package_info = response.json()
        if len(sorevdeps) == 0:
            revdeps_list = [
                package
                for group in package_info["sobreaks"]
                for package in group
                if package not in revdeps_list
            ]
            if package_info["sobreaks_circular"] != None:
                revdeps_list += package_info["sobreaks_circular"]
            if alldeps:
                revdeps_list += [
                    package["package"]
                    for group in package_info["revdeps"]
                    for package in group["deps"]
                    if package["package"] not in revdeps_list
                ]
        else:
            revdeps_list = [
                package
                for soname in sorevdeps
                for package in package_info["sorevdeps"][soname]
                if package not in revdeps_list
            ]
    except:
        logger.warning(
            "Failed to get revdep from packages site, falling back to groups/%s-rebuilds",
            package_name,
        )
        with open("groups/{}-rebuilds".format(package_name), "r") as f:
            for line in f:
                if line.startswith("# revdep:"):
                    revdeps_list = [pkg.strip() for pkg in line[9:].split(",")]
                    break
        if len(revdeps_list) == 0:
            logger.error(
                "Failed to get reverse dependency of %s, and no fallback is available",
                package_name,
            )
            raise

    if len(revdeps_list) == 0:
        logger.info("Package %s has no reverse dependency", package_name)
        exit(0)

    # remove +32 packages
    revdeps_list = [i for i in revdeps_list if not i.endswith("+32")]

    # remove transitional packages/latx
    transitional_pkgs = [
        "gst-plugins-good-1-0",
        "gst-plugins-bad-1-0",
        "gst-plugins-ugly-1-0",
        "latx",
    ]
    revdeps_list = [i for i in revdeps_list if i not in transitional_pkgs]

    # remove excluded packages
    if exclude is not None:
        revdeps_list = [i for i in revdeps_list if i not in exclude]

    # add extra packages
    if include is not None:
        revdeps_list += include

    # dedup and sort
    revdeps_list = sorted(list(set(revdeps_list)))

    # get package versions and filter out removed packages
    res = []
    for pkg in revdeps_list:
        package_path, defines_path = search_package_path(pkg)
        if package_path is None:
            logger.warning("Package %s dropped, skipping", pkg)
            continue

        version = get_package_version(package_path, defines_path)
        res.append(
            {
                "name": pkg,
                "package_path": package_path,
                "defines_path": defines_path,
                "version": version,
            }
        )
    logger.info("Reverse dependencies: %s", ", ".join([pkg["name"] for pkg in res]))
    return res


# Generate groups/xxx-rebuilds content
def gen_rebuilds_list_string(revdep_list: list[dict[str, str]], args) -> str:
    rebuilds_path_list = [pkg["package_path"] for pkg in revdep_list]
    rebuilds_path_list = [
        i for i in rebuilds_path_list if i != None and len(i.split("/")) == 2
    ]
    # dedup
    rebuilds_path_list = list(set(rebuilds_path_list))
    logger.info("Found package paths: %s", ", ".join(rebuilds_path_list))

    generate_args = ["breaker"]
    if args.exclude is not None:
        generate_args.append("--exclude")
        generate_args += args.exclude
    if args.include is not None:
        generate_args.append("--include")
        generate_args += args.include
    if args.alldeps:
        generate_args.append("--alldeps")

    return (
        f"# Auto-generated by {' '.join(generate_args)} begin, do not edit\n"
        + "# revdep: "
        + ", ".join([pkg["name"] for pkg in revdep_list])
        + "\n"
        + "\n".join(sorted(rebuilds_path_list))
        + "\n"
        + "# Auto-generated by breaker end, do not edit\n"
    )


# Save groups/xxx-rebuilds file
def write_rebuilds(package_name: str, rebuilds_path_list_str: str) -> None:
    with open("groups/{}-rebuilds".format(package_name), "w") as f:
        f.write(rebuilds_path_list_str)
        logger.info("groups/{}-rebuilds created".format(package_name))


# Generate PKGBREAK for a list of packages
def gen_pkgbreak_string(revdeps: list[dict[str, str]]) -> str:
    max_line_size = 68
    pkgbreak_list = ["{}<={}".format(pkg["name"], pkg["version"]) for pkg in revdeps]
    buffer = []
    buffer2 = []
    for i in pkgbreak_list:
        buffer.append(i)
        if len(" ".join(buffer)) < max_line_size:
            buffer2.append(i)
        else:
            buffer = [i]
            buffer2.append("\\\n")
            buffer2.append(" " * 9 + " ".join(buffer))
    return 'PKGBREAK="{}"'.format(" ".join(buffer2))


# Bump version in spec and create git commit
def bump_rel(reason, name, spec_path):
    contents = []
    has_rel = False
    with open(spec_path, "r") as f:
        for line in f:
            if "REL=" in line:
                cur_rel = line.split("=")[-1].strip()
                # https://stackoverflow.com/a/17888240/2148614
                int_rel = int(re.compile("(\d+)").match(cur_rel).group(1))
                new_rel = str(int_rel + 1)
                contents.append(line.replace(cur_rel, new_rel))
                has_rel = True
            elif "REL=" not in line:
                contents.append(line)

    if not has_rel:
        contents.insert(1, "REL=1\n")
    with open(spec_path, "w") as f:
        f.writelines(contents)

    os.system(f"git add {spec_path}")
    os.system(f'git commit -m "{name}: bump REL due to {reason}"')


def main():
    logging.basicConfig(level=logging.INFO)
    parser = argparse.ArgumentParser(
        description="Breaker make bumping shared library easy"
    )
    parser.add_argument("package", help="Package name to update")
    parser.add_argument(
        "-a",
        "--alldeps",
        action="store_true",
        help="Include all reverse dependencies, not only library reverse dependencies",
    )
    parser.add_argument(
        "-r", "--rebuilds", action="store_true", help="Generate groups/xxx-rebuilds"
    )
    parser.add_argument(
        "-p", "--pkgbreak", action="store_true", help="Generate PKGBREAK"
    )
    parser.add_argument(
        "-s", "--sorevdeps", nargs="+", help="Use specific soname to generate reverse dependencies"
    )
    parser.add_argument(
        "-b",
        "--bump",
        action="store_true",
        help="Bump dependencies one in a git commit",
    )
    parser.add_argument(
        "-d", "--debug", action="store_true", help="Enable debug message"
    )
    parser.add_argument(
        "-e", "--exclude", nargs="+", help="Exclude packages from reverse dependencies"
    )
    parser.add_argument(
        "-i", "--include", nargs="+", help="Add extra packages to reverse dependencies"
    )

    args = parser.parse_args()

    if args.exclude is not None:
        logger.info("Excluding packages: %s", " ".join(args.exclude))

    if args.include is not None:
        logger.info("Extra packages: %s", " ".join(args.include))

    if args.debug:
        logger.setLevel(logging.DEBUG)

    if args.bump and (args.rebuilds or args.pkgbreak):
        logger.error("Do not use --bump with --rebuilds or --pkgbreak")
        exit(1)

    # get info of new package
    package_path, defines_path = search_package_path(args.package)
    version = get_package_version(package_path, defines_path)

    revdeps = get_revdeps(args.package, args.alldeps, args.include,
                          args.exclude, args.sorevdeps)

    if args.rebuilds:
        revdeps_path_list_str = gen_rebuilds_list_string(revdeps, args)
        write_rebuilds(args.package, revdeps_path_list_str)

    if args.pkgbreak:
        pkgbreak = gen_pkgbreak_string(revdeps)

    if args.bump:
        logger.info("Bumping reverse depedencies")
        reason = f"{args.package} update to {version}"
        for pkg in revdeps:
            spec_path = os.path.join(pkg["package_path"], "spec")
            bump_rel(reason, os.path.basename(pkg["package_path"]), spec_path)
        logger.info("Created %s git commits", len(revdeps))

    if args.pkgbreak:
        logger.info("Generated PKGBREAK:")
        print(pkgbreak)

    logger.info("Using acbs-build to reorder packages")
    output = subprocess.check_output(
        ["sudo", "ciel", "shell", "-i", "main", "--", "acbs-build", "-ep", args.package]
        + [os.path.basename(pkg["package_path"]) for pkg in revdeps],
        encoding="utf-8",
        stderr=subprocess.STDOUT,
    )
    for line in output.split("\n"):
        if "ACBS has saved your build queue to" in line:
            file = line.split(" ")[-1].removesuffix("\x1b[0m")
            with open(file, "r") as f:
                pkgs = [pkg.strip() for pkg in f if pkg.strip() != args.package]
            os.remove(file)

            pkgs.insert(0, f"{args.package}:-pkgbreak")
            pkgs.append(args.package)

            logger.info("BuildIt command to open pr:")
            current_branch = subprocess.check_output(
                ["git", "symbolic-ref", "--short", "HEAD"], encoding="utf-8"
            ).strip()
            print(
                f"/openpr {args.package}: update to {version};{current_branch};{','.join(pkgs)}"
            )

            logger.info("Ciel command to build:")
            print(f"sudo ciel build -i main {' '.join(pkgs)}")
            break


if __name__ == "__main__":
    main()