#!/usr/bin/env python3
 
import argparse
import logging
import subprocess
import sys
from datetime import datetime
from pathlib import Path
from typing import Any, NoReturn, Optional, Sequence, Iterator
import enum
import contextlib
import textwrap
 
SYSTEMD_REPO = "https://github.com/systemd/systemd"
AUTHOR = "CentOS Hyperscale SIG <centos-devel@centos.org>"
 
 
class LogFormatter(logging.Formatter):
    def __init__(self, fmt: Optional[str] = None, *args: Any, **kwargs: Any) -> None:
        fmt = fmt or "%(message)s"
 
        bold = "\033[0;1;39m" if sys.stderr.isatty() else ""
        gray = "\x1b[38;20m" if sys.stderr.isatty() else ""
        red = "\033[31;1m" if sys.stderr.isatty() else ""
        yellow = "\033[33;1m" if sys.stderr.isatty() else ""
        reset = "\033[0m" if sys.stderr.isatty() else ""
 
        self.formatters = {
            logging.DEBUG: logging.Formatter(f"‣ {gray}{fmt}{reset}"),
            logging.INFO: logging.Formatter(f"‣ {fmt}"),
            logging.WARNING: logging.Formatter(f"‣ {yellow}{fmt}{reset}"),
            logging.ERROR: logging.Formatter(f"‣ {red}{fmt}{reset}"),
            logging.CRITICAL: logging.Formatter(f"‣ {red}{bold}{fmt}{reset}"),
        }
 
        super().__init__(fmt, *args, **kwargs)
 
    def format(self, record: logging.LogRecord) -> str:
        return self.formatters[record.levelno].format(record)
 
 
def run(cmd: Sequence[str], *args: Any, **kwargs: Any) -> subprocess.CompletedProcess:
    try:
        return subprocess.run(cmd, *args, **kwargs, check=True, text=True)
    except FileNotFoundError:
        die(f"{cmd[0]} not found in PATH.")
    except subprocess.CalledProcessError as e:
        logging.error(
            f'"{" ".join(str(s) for s in cmd)}" returned non-zero exit code {e.returncode}.'
        )
        raise e
 
 
def die(message: str) -> NoReturn:
    logging.error(message)
    sys.exit(1)
 
 
@contextlib.contextmanager
def restore(path: Path) -> Iterator[None]:
    old = path.read_text()
    try:
        yield
    finally:
        path.write_text(old)
 
 
def do_cd(args: argparse.Namespace) -> None:
    if not Path(".git").exists():
        die("The cd verb must be run from the rpm git repository")
 
    logging.info("Downloading sources")
    run(
        [
            "spectool",
            "--define",
            f"_sourcedir {Path.cwd()}",
            "--define",
            "branch main",
            "--get-files",
            "systemd.spec",
        ],
    )
 
    # We can't determine the version dynamically in the spec so we retrieve it
    # up front and pass it in via a macro.
    version = run(
        [
            "tar",
            "--gunzip",
            "--extract",
            "--to-stdout",
            "--file=main.tar.gz",
            "systemd-main/meson.version",
        ],
        stdout=subprocess.PIPE,
    ).stdout.strip()
 
    # The timestamp is to ensure the release is always monotonically increasing
    rpmrelease = datetime.now().strftime(r"%Y%m%d%H%M%S")
 
    with restore(Path("systemd.spec")):
        Path("systemd.spec").write_text(
            textwrap.dedent(
                f"""\
                %bcond upstream 1
                %define version_override {version}
                %define release_override {rpmrelease}
                %define branch main
                """
            )
            + Path("systemd.spec").read_text()
        )
 
        if args.repo == "main":
            root = f"centos-stream-hyperscale-{args.release}-x86_64"
        else:
            root = f"centos-stream-hyperscale-{args.repo}-{args.release}-x86_64"
 
        logging.info("Building src.rpm")
        run(
            [
                "mock",
                "--root",
                root,
                "--sources=.",
                "--spec=systemd.spec",
                "--enable-network",
                "--define",
                "%_disable_source_fetch 0",
                "--buildsrpm",
                "--resultdir=.",
            ],
        )
 
    srcrpm = next(Path.cwd().glob("*.src.rpm"))
    logging.info(f"Wrote: {srcrpm}")
 
    run(
        [
            "cbs",
            *(["--cert", args.cert] if args.cert else []),
            "build",
            "--wait",
            "--fail-fast",
            "--skip-tag",
            f"hyperscale{args.release}s-packages-{args.repo}-el{args.release}s",
            str(srcrpm),
        ],
    )
 
    if not args.publish:
        logging.info("Publishing not requested, not tagging builds in testing")
        return
 
    prefix = "hs+fb" if args.repo == "facebook" else "hs"
 
    run(
        [
            "cbs",
            *(["--cert", args.cert] if args.cert else []),
            "tag-build",
            f"hyperscale{args.release}s-packages-{args.repo}-testing",
            f"systemd-{version}-{rpmrelease}.{prefix}.el{args.release}",
        ]
    )
 
 
class Verb(enum.Enum):
    cd = "cd"
 
    def __str__(self) -> str:
        return self.value
 
    def run(self, args: argparse.Namespace) -> None:
        {Verb.cd: do_cd}[self](args)
 
 
def main() -> None:
    handler = logging.StreamHandler(stream=sys.stderr)
    handler.setFormatter(LogFormatter())
    logging.getLogger().addHandler(handler)
    logging.getLogger().setLevel("INFO")
 
    parser = argparse.ArgumentParser()
 
    parser.add_argument(
        "--release",
        help="CentOS Stream release to use (e.g 9)",
        metavar="RELEASE",
        default=9,
        choices=[9, 10],
        type=int,
    )
    parser.add_argument(
        "--repo",
        help="Hyperscale repository to build against",
        choices=["main", "facebook"],
        default="main",
    )
    parser.add_argument(
        "--cert",
        help="Path to the CentOS certificate to use",
        metavar="PATH",
        type=Path,
        default=None,
    )
    parser.add_argument(
        "--publish",
        action="store_true",
        help="Publish results of operation (by default only a dry-run is done)",
    )
    parser.add_argument(
        "verb",
        type=Verb,
        choices=list(Verb),
        help=argparse.SUPPRESS,
    )
 
    args = parser.parse_args()
 
    if args.cert:
        args.cert = args.cert.absolute()
 
    try:
        args.verb.run(args)
    except SystemExit as e:
        sys.exit(e.code)
    except KeyboardInterrupt:
        logging.error("Interrupted")
        sys.exit(1)
    except subprocess.CalledProcessError as e:
        sys.exit(e.returncode)
 
 
if __name__ == "__main__":
    main()