Devine/core/git.py

import asyncio
import shlex
from typing import Tuple

from git import Repo
from git.exc import GitCommandError, InvalidGitRepositoryError

import config

from ..logging import LOGGER


def install_req(cmd: str) -> Tuple[str, str, int, int]:
    async def install_requirements():
        args = shlex.split(cmd)
        process = await asyncio.create_subprocess_exec(
            *args,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        stdout, stderr = await process.communicate()
        return (
            stdout.decode("utf-8", "replace").strip(),
            stderr.decode("utf-8", "replace").strip(),
            process.returncode,
            process.pid,
        )

    return asyncio.get_event_loop().run_until_complete(install_requirements())


def hide_sensitive_info(message: str) -> str:
    # Function to hide sensitive information in messages
    return message.replace(config.UPSTREAM_REPO, "<hidden>")


def git():
    REPO_LINK = config.UPSTREAM_REPO
    if config.GIT_TOKEN:
        GIT_USERNAME = REPO_LINK.split("com/")[1].split("/")[0]
        TEMP_REPO = REPO_LINK.split("https://")[1]
        UPSTREAM_REPO = f"https://{GIT_USERNAME}:{config.GIT_TOKEN}@{TEMP_REPO}"
    else:
        UPSTREAM_REPO = config.UPSTREAM_REPO

    try:
        repo = Repo()
        LOGGER(__name__).info("Git Client Found [VPS DEPLOYER]")
    except InvalidGitRepositoryError:
        repo = Repo.init()
        origin = repo.create_remote("origin", UPSTREAM_REPO)
        origin.fetch()
        repo.create_head(config.UPSTREAM_BRANCH, origin.refs[config.UPSTREAM_BRANCH])
        repo.heads[config.UPSTREAM_BRANCH].set_tracking_branch(
            origin.refs[config.UPSTREAM_BRANCH]
        )
        repo.heads[config.UPSTREAM_BRANCH].checkout()
    except GitCommandError as e:
        LOGGER(__name__).error(f"Invalid Git Command: {e}")
        return

    try:
        origin = repo.remote("origin")
    except ValueError:
        origin = repo.create_remote("origin", UPSTREAM_REPO)
    
    origin.fetch(config.UPSTREAM_BRANCH)
    
    try:
        origin.pull(config.UPSTREAM_BRANCH)
    except GitCommandError as e:
        repo.git.reset("--hard", "FETCH_HEAD")
        LOGGER(__name__).error(f"Error pulling from upstream: {e}")

    install_req("pip3 install --no-cache-dir -r requirements.txt")
    LOGGER(__name__).info(f"Fetching updates from upstream repository: {hide_sensitive_info(UPSTREAM_REPO)}")


# Call the git function
git()