pool_bin

#!/usr/bin/python3
# Copyright (c) TurnKey GNU/Linux - https://www.turnkeylinux.org
#
# This file is part of Pool
#
# Pool is free software; you can redistribute it and/or modify it
# under the terms of the GNU Affero General Public License as published by the
# Free Software Foundation; either version 3 of the License, or (at your
# option) any later version.

import argparse
import os
import sys
from collections.abc import Callable
from os.path import basename, dirname, isdir
from os.path import exists as path_exists
from typing import IO, NoReturn

from debian import debfile
from packaging.version import Version
from pool_lib import Pool, PoolError, PoolKernel, logger

exitcode = 0
PROG = "pool"
DEBUG = False
if logger.level == "debug":
    DEBUG = True


def warn(msg: str) -> None:
    global exitcode
    exitcode = 1
    print(f"warning: {msg!s}", file=sys.stderr)


def fatal(
    msg: str | PoolError | FileNotFoundError,
    pool_help: Callable[[IO[str] | None], None] | None = None,
) -> NoReturn:
    print(f"error: {msg!s}", file=sys.stderr)
    if pool_help:
        # argparse.ArgumentParser().print_help(file=None)
        # explicitly passing None to satisfy mypy (otherwise 'too few args')
        pool_help(None)
    sys.exit(1)


def pool_exists(package: str) -> None:
    istrue: bool = False
    logger.debug(f"exists({package=})")
    try:
        istrue = Pool().kernel.exists(package)
    except PoolError as e:
        if DEBUG:
            raise e
        else:
            fatal(e)

    if istrue:
        print("true")
    else:
        print("false")
        sys.exit(1)


def pool_gc(disable_recursion: bool = False) -> None:
    logger.debug(f"gc({disable_recursion=})")
    try:
        Pool().gc(not disable_recursion)
    except PoolError as e:
        if DEBUG:
            raise e
        else:
            fatal(e)


# pool-get function
def read_packages(in_file: str) -> list[str]:
    logger.debug(f"read_packages({in_file=})")
    packages = []
    try:
        with open(in_file) as fob:
            for line in fob.readlines():
                line = line.split("#")[0].strip()
                if not line:
                    continue
                packages.append(line)
        return packages
    except FileNotFoundError as e:
        if DEBUG:
            raise PoolError(e) from e
        else:
            fatal(e)


# #
def pool_get(
    outputdir: str,
    packages: Pool.PackageList | list[str] | None = None,
    inputfile: str | None = None,
    strict: bool = False,
    quiet: bool = False,
    tree: bool = False,
    preserve_buildroot: str | None = None,
    source: bool = False,
) -> NoReturn:
    logger.debug(
        f"get({outputdir=}, {packages=}, {inputfile=}, {strict=},"
        f" {strict=}, {quiet=}, {tree=}, {preserve_buildroot=},"
        f" {source=})"
    )
    this_exitcode = exitcode
    pool = Pool(preserve_buildroot=preserve_buildroot)
    package_list = []

    if not packages:
        packages = []
    if not packages and not inputfile:
        # if no packages specified, get all the newest versions
        packages = pool.list()
    elif inputfile:
        # treat all "packages" as plan files
        for plan_file in packages:
            pkgs_read = read_packages(plan_file)
            if not pkgs_read:
                pkgs_read = []
            package_list.extend(pkgs_read)
    else:
        # assume that it's a list of package names
        package_list = list(packages)

    try:
        packages = pool.get(
            outputdir,
            list(package_list),
            tree_fmt=tree,
            strict=strict,
            source=source,
        )
    except PoolError as e:
        if DEBUG:
            raise e
        else:
            fatal(e)
    if strict and (packages.missing or packages.failed):
        this_exitcode = 1

    if not quiet:
        all_packages = set([*packages, *packages.missing, *packages.failed])

        print("build summary")
        print("=============")
        no_of_successful_pkgs = (
            len(all_packages) - (len(packages.missing) + len(packages.failed))
        )
        print(f"{no_of_successful_pkgs}/{len(all_packages)} builds successful")

        for package in all_packages - set(
            [*packages.failed, *packages.missing]
        ):
            print(f" - {package!r} built successfully")
        for package in packages.missing:
            print(f" - {package!r} not registered")
        for package in packages.failed:
            print(f" - {package!r} failed to build")

    sys.exit(this_exitcode)


# pool-info-build functions
def extract_source_name(path: str) -> str | None:
    logger.debug(f"extract_source_name({path=})")
    deb = debfile.DebFile(path)
    if "Source" in deb.debcontrol():
        return deb.debcontrol()["Source"]

    return None


def pkgcache_list_versions(pool: PoolKernel, name: str) -> list[str]:
    logger.debug(f"pkgcache_list_versions({pool=}, {name=})")
    versions = [
        pkgcache_version
        for pkgcache_name, pkgcache_version in pool.pkgcache.list()
        if pkgcache_name == name
    ]

    for subpool in pool.subpools:
        versions += pkgcache_list_versions(subpool, name)

    return versions


def pkgcache_getpath_newest(pool: PoolKernel, name: str) -> str | None:
    logger.debug(f"pkgcache_getpath_newest({pool=}, {name=})")
    versions = pkgcache_list_versions(pool, name)
    if not versions:
        return None
    versions.sort(key=Version)
    version_newest = versions[-1]

    package = pool.fmt_package_id(name, version_newest)
    return pool.getpath_deb(package, build=False)


def binary2source(pool: PoolKernel, package: str) -> str | None:
    """translate package from binary to source"""
    logger.debug(f"binary2source({pool=}, {package=})")
    name, version = pool.parse_package_id(package)
    if version:
        path = pool.getpath_deb(package, build=False)
        if not path:
            return None

        source_name = extract_source_name(path)
        if not source_name:
            return package

        return pool.fmt_package_id(source_name, version)

    # no version, extract source from the most recent binary
    path = pkgcache_getpath_newest(pool, name)
    if not path:
        return None

    source_name = extract_source_name(path)
    if not source_name:
        return name

    return source_name


def getpath_build_log(package: str) -> str | None:
    # XXX This function doesn't appear to be used anywhere?!
    logger.debug(f"getpath_build_log({package=})")
    try:
        pool = Pool().kernel
    except PoolError as e:
        if DEBUG:
            raise e
        else:
            fatal(e)

    path = pool.getpath_build_log(package)
    if path:
        return path

    # maybe package is a binary name?
    # try mapping it to a source name and trying again

    source_package = binary2source(pool, package)
    if source_package:
        path = pool.getpath_build_log(source_package)

    if not path:
        package_desc = repr(package)
        if source_package:
            package_desc = f"{package_desc} ({source_package})"
        fatal(f"no build log for {package_desc}")

    return path


def pool_info_build(pool: PoolKernel, package: str) -> None:
    # TODO - this needs to be completed
    # - all the functionality required should be in the functions above
    logger.debug(f"pool_info({pool=}, {package=})")
    warn(
        "pool-info-build command is incomplete; please see $POOL_PATH/.pool/"
        f"build/buildinfo/{package}_<version>_<arch>.buildinfo for buildlog"
        " and/or pool-info and/or pool-list."
    )


# # pool-info functions
def print_registered(pool: PoolKernel) -> None:
    logger.debug(f"print_registered({pool=})")
    if pool.stocks:
        print("# stocks")
    print_stocks(pool)
    if pool.subpools:
        if pool.stocks:
            print()
        print("# subpools")
        print_subpools(pool)


def print_stocks(pool: PoolKernel) -> None:
    logger.debug(f"print_stocks({pool=})")
    # `Cannot access member "stocks" for type "Pool"`
    # - stocks is a PoolKernel property
    for stock in pool.stocks:
        addr = stock.link
        # `Cannot access member "branch" for type "StockBase"`
        # - branch is a StockBase property; although Stock subclasses
        #   StockBase?!
        if stock.branch:  # type: ignore
            addr += "#" + stock.branch.replace("%2F", "/")  # type: ignore
        print(addr)


def print_subpools(pool: PoolKernel) -> None:
    logger.debug(f"print_subpool({pool=})")
    for subpool in pool.subpools:
        print(subpool.path)


def print_build_root(pool: PoolKernel) -> None:
    logger.debug(f"print_build_root({pool=})")
    print(pool.buildroot)


def print_pkgcache(pool: PoolKernel) -> None:
    logger.debug(f"print_pkgcache({pool=})")
    pool.sync()
    for name, version in pool.pkgcache.list():
        print(f"{name}={version}")


def print_stock_inventory(stock_inventory: list[tuple[str, str, str]]) -> None:
    logger.debug(f"print_stock_inventory({stock_inventory=})")
    package_width = max([len(vals[0]) for vals in stock_inventory])
    stock_name_width = max([len(vals[1]) for vals in stock_inventory])

    for package, stock_name, _ in stock_inventory:
        print(
            f"{package.ljust(package_width)}"
            f" {stock_name.ljust(stock_name_width)}"
            f" relative_path"
        )


def print_stock_sources(pool: PoolKernel) -> None:
    logger.debug(f"print_stock_sources({pool=})")
    pool.sync()

    stock_inventory = []
    for stock in pool.stocks:
        for path, versions in stock.sources:
            for version in versions:
                package = f"{basename(path)}={version}"
                relative_path = dirname(path)
                stock_inventory.append((package, stock.name, relative_path))

    if stock_inventory:
        print_stock_inventory(stock_inventory)


def print_stock_binaries(pool: PoolKernel) -> None:
    logger.debug(f"print_stock_binaries({pool=})")
    pool.sync()

    stock_inventory = []
    for stock in pool.stocks:
        for path in stock.binaries:
            package = basename(path)
            relative_path = dirname(path)
            stock_inventory.append((package, stock.name, relative_path))

    if stock_inventory:
        print_stock_inventory(stock_inventory)


def print_build_logs(pool: PoolKernel) -> None:
    logger.debug(f"print_build_logs({pool=})")
    for log_name, log_version in pool.build_logs:
        print(log_name + "=" + log_version)


def pool_info(
    function: Callable[[PoolKernel], None],
    recursive: bool = False,
    pool: PoolKernel | None = None,
) -> None:
    logger.debug(f"pool_info({function=}, {recursive=}, {pool=})")
    try:
        if pool is None:
            pool = PoolKernel()
            pool.drop_privileges()
    except PoolError as e:
        if DEBUG:
            raise e
        else:
            fatal(e)

    if recursive:
        print(f"### POOL_DIR={pool.path}")

    function(pool)
    if recursive:
        for subpool in pool.subpools:
            print()
            pool_info(function, recursive, subpool)


def pool_init(buildroot: str | None) -> None:
    logger.debug(f"pool_init({buildroot=})")
    try:
        Pool.init_create(buildroot)
    except PoolError as e:
        if DEBUG:
            raise e
        else:
            fatal(e)


def pool_list(
    globs: list[str] | None = None,
    all_versions: bool = False,
    name_only: bool = False,
    verbose: bool = False,
) -> None:
    logger.debug(
        f"pool_list({globs=}, {all_versions=}, {name_only=}, {verbose=})"
    )
    if not globs:
        globs = []
    packages = Pool().list(all_versions, *globs, verbose=verbose)
    for glob in packages.missing:
        print(f"warning: {glob}: no matching packages", file=sys.stderr)

    for package in packages:
        if name_only:
            print(Pool.parse_package_id(package)[0])
        else:
            print(package)


def pool_register(stock: str) -> None:
    logger.debug(f"pool_register({stock=})")
    logger.debug(f"{stock=!r}")
    try:
        Pool().register(stock)
    except PoolError as e:
        if DEBUG:
            raise e
        else:
            fatal(e)


def pool_unregister(stock: str) -> None:
    logger.debug(f"pool_unregister({stock=})")
    try:
        Pool().unregister(stock)
    except PoolError as e:
        if DEBUG:
            raise e
        else:
            fatal(e)


def pool_refresh(stock: str) -> None:
    logger.debug(f"pool_refresh({stock=})")
    pool_unregister(stock)
    print("Garbage collecting stable package cache")
    pool_gc()  # TODO - just garbage collect the refreshed stock
    pool_register(stock)


def main() -> None:
    # support for "one word" '/usr/bin/pool-COMMAND'
    # - symlinks to '/usr/bin/pool'; COMMAND becomes the first argument

    command = os.path.split(sys.argv[0])[-1]
    sys.argv[0] = PROG
    if command.startswith(PROG + "-"):
        subcommand = command[len(PROG + "-"):]
        sys.argv.insert(1, subcommand)

    env_vars = (
        "env vars:"
        "\n    POOL_DIR\t\tPath to the pool directory (defaults to '.')"
        "\n    POOL_LOG_LEVEL\tSet log level for pool (no logging by default)"
        "\n    DEBUG\t\tGlobal 'debug' log level (overrides 'POOL_LOG_LEVEL')"
        "\n    DECKDEBUILD_*\tEnv vars will be forwarded to deckdebuild"
    )

    parser = argparse.ArgumentParser(
        prog=PROG,
        formatter_class=argparse.RawDescriptionHelpFormatter,
        description=(
            "Maintain a pool of packages from source and binary stocks"
        ),
        epilog=env_vars,
    )
    subparsers = parser.add_subparsers(dest="cmd")

    # pool-exists
    parser_exists = subparsers.add_parser(
        "exists",
        help=(
            "Check if package exists in pool (Prints true/false; exit code"
            " 0/1 respectively)"
        ),
    )
    parser_exists.add_argument("package", help="Package to check for")
    parser_exists.set_defaults(func=pool_exists)

    # pool-gc
    parser_gc = subparsers.add_parser(
        "gc",
        help="Garbage collect stale data from the pool's caches",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=env_vars,
    )
    parser_gc.add_argument(
        "-R",
        "--disable-recursion",
        action="store_true",
        help="Disable recursive garbage collection of subpools",
    )
    parser_gc.set_defaults(func=pool_gc)

    # pool-get
    parser_get = subparsers.add_parser(
        "get",
        help="Get packages from pool",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=env_vars,
    )
    parser_get.add_argument(
        "-i",
        "--input",
        dest="inputfile",
        action="store_true",
        help="Read packages from file(s) - one package[=version] per line.",
    )
    parser_get.add_argument(
        "-s",
        "--strict",
        action="store_true",
        help="fatal error on missing packages",
    )
    parser_get.add_argument(
        "-q",
        "--quiet",
        action="store_true",
        help="suppress warnings about missing packages",
    )
    parser_get.add_argument(
        "-t",
        "--tree",
        action="store_true",
        help="output packages in a Debian apt repo like filesystem tree",
    )

    buildroot_cleanup_args = parser_get.add_mutually_exclusive_group()
    buildroot_cleanup_args.add_argument(
        "-e",
        "--preserve-buildroot-on-error",
        action="store_const",
        const="on-error",
        dest="preserve_buildroot",
        help="leave build chroot intact after build if failure (default)",
    )
    buildroot_cleanup_args.add_argument(
        "-p",
        "--preserve-buildroot-always",
        action="store_const",
        const="always",
        dest="preserve_buildroot",
        help="always leave build chroot intact after build",
    )
    buildroot_cleanup_args.add_argument(
        "-n",
        "--preserve-buildroot-never",
        action="store_const",
        const="never",
        dest="preserve_buildroot",
        help="never leave build chroot intact after build",
    )
    parser_get.add_argument(
        "-o",
        "--source",
        action="store_true",
        help="build source packages in addition to binary packages",
    )
    parser_get.add_argument("outputdir", help="Output directory")
    parser_get.add_argument(
        "packages",
        nargs="+",
        default=[],
        help=(
            "Package(s) or file containing packages (optionally with versions)"
        ),
    )
    parser_get.set_defaults(func=pool_get)

    # pool-info-build
    parser_info_build = subparsers.add_parser(
        "info-build", help="Prints source build log for package"
    )
    parser_info_build.add_argument("package", help="Package name")
    parser_info_build.set_defaults(func=pool_info_build)

    # pool-info
    parser_info = subparsers.add_parser(
        "info",
        help="Prints pool info",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=env_vars,
    )
    # pool-info conflicting args
    parser_info_conflicts = parser_info.add_mutually_exclusive_group()
    parser_info_conflicts.add_argument(
        "--registered",
        dest="function",
        action="store_const",
        const=print_registered,
        default=print_registered,
        help="Prints list of registered stocks and subpools (default)",
    )
    parser_info_conflicts.add_argument(
        "--stocks",
        dest="function",
        action="store_const",
        const=print_stocks,
        help="Prints list of registered stocks",
    )
    parser_info_conflicts.add_argument(
        "--subpools",
        dest="function",
        action="store_const",
        const=print_subpools,
        help="Prints list of registered subpools",
    )
    parser_info_conflicts.add_argument(
        "--build-root",
        dest="function",
        action="store_const",
        const=print_build_root,
        help="Prints build-root",
    )
    parser_info_conflicts.add_argument(
        "--build-logs",
        dest="function",
        action="store_const",
        const=print_build_logs,
        help="Prints a list of build logs for source packages",
    )
    parser_info_conflicts.add_argument(
        "--pkgcache",
        dest="function",
        action="store_const",
        const=print_pkgcache,
        help="Prints list of cached packages",
    )
    parser_info_conflicts.add_argument(
        "--stock-sources",
        dest="function",
        action="store_const",
        const=print_stock_sources,
        help="Prints list of package sources in registered stocks",
    )
    parser_info_conflicts.add_argument(
        "--stock-binaries",
        dest="function",
        action="store_const",
        const=print_stock_binaries,
        help="Prints list of package binaries in registered stocks",
    )
    # pool-info non-conflicting arg
    parser_info.add_argument(
        "-r",
        "--recursive",
        action="store_true",
        help="Lookup pool info recursively in subpools",
    )
    parser_info.set_defaults(func=pool_info)

    # pool-init
    parser_init = subparsers.add_parser(
        "init",
        help="Initialize a new pool",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=env_vars,
    )
    parser_init_conflicts = parser_init.add_mutually_exclusive_group()
    parser_init_conflicts.add_argument(
        "buildroot",
        nargs="?",
        default=None,
        help="/path/to/build-chroot",
    )
    parser_init_conflicts.add_argument(
        "-N",
        "--no-buildroot",
        action="store_true",
        help="create pool with no buildroot (for pool of pools or pre-built"
        " packages)"
    )
    parser_init.set_defaults(func=pool_init)

    # pool-list
    parser_list = subparsers.add_parser(
        "list",
        help="List packages in pool",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=env_vars,
    )
    parser_list.add_argument(
        "-a",
        "--all-versions",
        action="store_true",
        help="print all available versions of a package in the pool"
        " (default: print the newest versions only)",
    )
    parser_list.add_argument(
        "-v",
        "--verbose",
        action="store_true",
        help="show warnings for skipped package versions",
    )
    parser_list.add_argument(
        "-n",
        "--name-only",
        action="store_true",
        help="print only the names of packages in the pool",
    )
    parser_list.add_argument("globs", nargs="*", help="package-glob(s)")
    parser_list.set_defaults(func=pool_list)

    # pool-register
    parser_register = subparsers.add_parser(
        "register",
        help="Register a package stock into the pool",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=env_vars,
    )
    parser_register.add_argument("stock", help="/path/to/stock[#branch]")
    parser_register.set_defaults(func=pool_register)

    # pool-unregister
    parser_unregister = subparsers.add_parser(
        "unregister",
        help="Unregister a package stock from the pool",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=env_vars,
    )
    parser_unregister.add_argument("stock", help="/path/to/stock[#branch]")
    parser_unregister.set_defaults(func=pool_unregister)

    # pool-refresh
    parser_refresh = subparsers.add_parser(
        "refresh",
        help="Refresh a package stock by unregistering/reregistering it from"
        " (only makes sense for git stocks)",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=env_vars,
    )
    parser_refresh.add_argument("stock", help="/path/to/stock[#branch]")
    parser_refresh.set_defaults(func=pool_refresh)

    args = parser.parse_args()
    logger.debug(f"{args=}")
    if args.cmd:
        func = args.func
        args_dict = vars(args)
        if args.cmd == "init":
            # argparse handles error if both buildroot and --no-buildroot, but
            # to force user to make explicit choice, error if neither set
            if args.buildroot is None and not args.no_buildroot:
                fatal(
                    "pool-init requires one of /path/to/buildroot or"
                    " --no-buildroot"
                )
            # if no_buildroot == True then buildroot is None
            del args_dict["no_buildroot"]
        # only pass debug for pool-get
        if args.cmd != "get" and "debug" in args:
            del args_dict["debug"]
        del args_dict["cmd"]
        del args_dict["func"]
        if "outputdir" in args:
            if not path_exists(args.outputdir):
                fatal(f"{args.outputdir} does not exist")
            elif not isdir(args.outputdir):
                fatal(f"{args.outputdir} exists, but is not a directory")
        try:
            func(**args_dict)
        except PoolError as e:
            fatal(e)
    else:
        fatal("Subcommand required.", parser.print_help)


if __name__ == "__main__":
    main()