spcd/spcd/__init__.py

"""Shell to Python Continuous Deployment."""

__version__ = "0.0.1"

import sys
from os import environ, pathsep
from pathlib import Path

from rwx import fs
from rwx.log import stream as log
from rwx.ps import run

from spcd import cmd
from spcd.ci import project, projects
from spcd.shell import env
from spcd.util import browse, cat, split, step

COMMANDS_PREFIX = "spcd-"


def clone_project_branch() -> None:
    """Clone project on triggering branch into the current workspace."""
    if not projects.environment.get("GITLAB_CI"):
        step("Clone project branch")
        log.info(projects)
        split()
        log.info(project)
        split()
        log.info(
            """\
%s
↓
""",
            project.url,
        )
        run(
            "git",
            "clone",
            "--branch",
            project.branch,
            "--",
            project.url,
            str(project.root),
        )


def install_actions(path: Path) -> None:
    """Make actions usable in workflows.

    :param path: entry point file
    :type path: Path
    """
    step("Install actions")
    source = Path("act")
    target = path.parent / "actions"
    log.info(source)
    log.info(actions)
    source.symlink_to(target)


def install_commands(path: Path) -> None:
    """Make commands callable in the operating system.

    :param path: entry point that commands will call
    :type path: Path
    """
    step("Install commands")
    user = Path("/usr/local/bin")
    for command in [
        "browse-workspace",
        "build-project",
        "check-project",
        "synchronize",
    ]:
        log.info(command)
        (user / f"{COMMANDS_PREFIX}{command}").symlink_to(path)


def install_python_packages() -> None:
    """Upgrade pip then install extra Python packages."""
    step("Install Python packages")
    log.info("pip")
    run("pip", "install", "--upgrade", "pip")
    split()
    packages = [
        "gitlint",
        "hatch",
        "mypy",
        "pelican",
        "pydoclint",
        "pylint",
        "pytest",
        "PyYAML",
        "ruff",
        "Sphinx",
        "sphinx-rtd-theme",
        "twine",
        "types-PyYAML",
    ]
    for package in packages:
        log.info(package)
    run("pip", "install", *packages)


def list_environment_variables() -> None:
    """List accessible variables and their public contents."""
    step("List environment variables")
    for variable, value in sorted(projects.environment.items()):
        if variable not in ["SPCD", "SPCD_SSH_KEY"]:
            log.info("%s = %s", variable, value)
        else:
            log.info("%s", variable)


def main(main_file: Path) -> None:
    """Entry point to initialize environment or run a specific command.

    :param main_file: entry point file calling this function
    :type main_file: Path
    """
    paths = environ["PATH"].split(pathsep)
    if env.SPCD_PYTHON_VENV_BINARIES not in paths:
        environ["PATH"] = pathsep.join([env.SPCD_PYTHON_VENV_BINARIES, *paths])
    path, *arguments = sys.argv
    name = Path(path).name
    if name == "__main__.py":
        list_environment_variables()
        clone_project_branch()
        set_ssh()
        install_actions(main_file)
        install_commands(main_file)
        install_python_packages()
    else:
        f = getattr(cmd, name.replace("-", "_"))
        f(*arguments)


def set_ssh() -> None:
    """Set things up to enable access to targets through SSH."""
    step("Set SSH")
    # get variables
    ssh_hosts = projects.environment.get("SPCD_SSH_HOSTS")
    ssh_key = projects.environment.get("SPCD_SSH_KEY")
    # set key type
    ssh_type = "ed25519"
    # set home directory
    home = Path("~").expanduser()
    # make home directory
    ssh = home / ".ssh"
    ssh.mkdir(exist_ok=True, parents=True)
    ssh.chmod(0o700)
    # write private key
    key = ssh / f"id_{ssh_type}"
    if ssh_key:
        fs.write(key, ssh_key)
        key.chmod(0o400)
    # write known hosts
    known = ssh / "known_hosts"
    if ssh_hosts:
        fs.write(known, ssh_hosts)
        known.chmod(0o400)
    # display
    browse(ssh)
    if known.exists():
        cat(known)