modules/repo.py

# -*- coding:utf-8 -*-
import os
import sys
import subprocess
from functools import lru_cache
import logging
import gradio as gr
import datetime
import platform

# This file is mainly used to describe repo version info, execute the git command, python pip command, shell command, etc.
# Part of the code in this file is referenced from stable-diffusion-webui/modules/launch_utils.py

python = sys.executable
pip = os.environ.get("PIP", "pip")
git = os.environ.get("GIT", "git")

# Pypi index url
index_url = os.environ.get("INDEX_URL", "")

# Whether to default to printing command output
default_command_live = True


def run(
    command, desc=None, errdesc=None, custom_env=None, live: bool = default_command_live
) -> str:
    if desc is not None:
        print(desc)
    run_kwargs = {
        "args": command,
        "shell": True,
        "env": os.environ if custom_env is None else custom_env,
        "encoding": "utf8",
        "errors": "ignore",
    }

    if not live:
        run_kwargs["stdout"] = run_kwargs["stderr"] = subprocess.PIPE

    result = subprocess.run(**run_kwargs)
    if result.returncode != 0:
        error_bits = [
            f"{errdesc or 'Error running command'}.",
            f"Command: {command}",
            f"Error code: {result.returncode}",
        ]
        if result.stdout:
            error_bits.append(f"stdout: {result.stdout}")
        if result.stderr:
            error_bits.append(f"stderr: {result.stderr}")
        raise RuntimeError("\n".join(error_bits))

    return result.stdout or ""


def run_pip(command, desc=None, pref=None, live=default_command_live):
    # if args.skip_install:
    #     return

    index_url_line = f" --index-url {index_url}" if index_url != "" else ""
    return run(
        f'"{python}" -m pip {command} --prefer-binary{index_url_line}',
        desc=f"{pref} Installing {desc}...",
        errdesc=f"Couldn't install {desc}",
        live=live,
    )


@lru_cache()
def commit_hash():
    try:
        return subprocess.check_output(
            [git, "rev-parse", "HEAD"], shell=False, encoding="utf8"
        ).strip()
    except Exception:
        return "<none>"


def commit_html():
    commit = commit_hash()
    if commit != "<none>":
        short_commit = commit[0:7]
        commit_info = f'<a style="text-decoration:none;color:inherit" href="https://github.com/GaiZhenbiao/ChuanhuChatGPT/commit/{short_commit}">{short_commit}</a>'
    else:
        commit_info = "unknown \U0001F615"
    return commit_info


@lru_cache()
def tag_html():
    try:
        latest_tag = run(f"{git} describe --tags --abbrev=0", live=False).strip()
        try:
            # tag = subprocess.check_output([git, "describe", "--tags", "--exact-match"], shell=False, encoding='utf8').strip()
            tag = run(f"{git} describe --tags --exact-match", live=False).strip()
        except Exception:
            tag = "<edited>"
    except Exception:
        tag = "<none>"

    if tag == "<none>":
        tag_info = "unknown \U0001F615"
    elif tag == "<edited>":
        tag_info = f'<a style="text-decoration:none;color:inherit" href="https://github.com/GaiZhenbiao/ChuanhuChatGPT/releases/tag/{latest_tag}">{latest_tag}</a><span style="font-size:smaller">*</span>'
    else:
        tag_info = f'<a style="text-decoration:none;color:inherit" href="https://github.com/GaiZhenbiao/ChuanhuChatGPT/releases/tag/{tag}">{tag}</a>'

    return tag_info


def repo_tag_html():
    commit_version = commit_html()
    tag_version = tag_html()
    return tag_version if tag_version != "unknown \U0001F615" else commit_version


def versions_html():
    python_version = ".".join([str(x) for x in sys.version_info[0:3]])
    repo_version = repo_tag_html()
    return f"""
        Python: <span title="{sys.version}">{python_version}</span>
         • 
        Gradio: {gr.__version__}
         • 
        <a style="text-decoration:none;color:inherit" href="https://github.com/GaiZhenbiao/ChuanhuChatGPT">ChuanhuChat</a>: {repo_version}
        """


def version_time():
    git = "git"
    cmd = f"{git} log -1 --format=%cd --date=iso-strict"
    commit_time = "unknown"
    try:
        if platform.system() == "Windows":
            # For Windows
            env = dict(os.environ)  # copy the current environment
            env["TZ"] = "UTC"  # set timezone to UTC
            raw_commit_time = subprocess.check_output(
                cmd, shell=True, encoding="utf8", env=env
            ).strip()
        else:
            # For Unix systems
            cmd = f"TZ=UTC {cmd}"
            raw_commit_time = subprocess.check_output(
                cmd, shell=True, encoding="utf8"
            ).strip()

        # Convert the date-time to the desired format
        commit_datetime = datetime.datetime.strptime(
            raw_commit_time, "%Y-%m-%dT%H:%M:%S%z"
        )
        commit_time = commit_datetime.strftime("%Y-%m-%dT%H:%M:%SZ")

        # logging.info(f"commit time: {commit_time}")
    except Exception:
        commit_time = "unknown"
    return commit_time


def get_current_branch():
    try:
        # branch = run(f"{git} rev-parse --abbrev-ref HEAD").strip()
        branch = subprocess.check_output(
            [git, "rev-parse", "--abbrev-ref", "HEAD"], shell=False, encoding="utf8"
        ).strip()
    except Exception:
        branch = "<none>"
    return branch


def get_latest_release():
    try:
        import requests

        release = requests.get(
            "https://api.github.com/repos/GaiZhenbiao/ChuanhuChatGPT/releases/latest"
        ).json()
        tag = release["tag_name"]
        release_note = release["body"]
        need_pip = release_note.find("requirements reinstall needed") != -1
    except Exception:
        tag = "<none>"
        release_note = ""
        need_pip = False
    return {"tag": tag, "release_note": release_note, "need_pip": need_pip}


def get_tag_commit_hash(tag):
    try:
        import requests

        tags = requests.get(
            "https://api.github.com/repos/GaiZhenbiao/ChuanhuChatGPT/tags"
        ).json()
        commit_hash = [x["commit"]["sha"] for x in tags if x["name"] == tag][0]
    except Exception:
        commit_hash = "<none>"
    return commit_hash


def repo_need_stash():
    try:
        return (
            subprocess.check_output(
                [git, "diff-index", "--quiet", "HEAD", "--"],
                shell=False,
                encoding="utf8",
            ).strip()
            != ""
        )
    except Exception:
        return True


def background_update():
    # {git} fetch --all && ({git} pull https://github.com/GaiZhenbiao/ChuanhuChatGPT.git main -f || ({git} stash && {git} pull https://github.com/GaiZhenbiao/ChuanhuChatGPT.git main -f && {git} stash pop)) && {pip} install -r requirements.txt")
    try:
        latest_release = get_latest_release()
        latest_release_tag = latest_release["tag"]
        latest_release_hash = get_tag_commit_hash(latest_release_tag)
        need_pip = latest_release["need_pip"]
        need_stash = repo_need_stash()

        timestamp = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
        current_branch = get_current_branch()
        updater_branch = f"tmp_{timestamp}"
        backup_branch = f"backup_{timestamp}"
        track_repo = "https://github.com/GaiZhenbiao/ChuanhuChatGPT.git"
        try:
            try:
                run(
                    f"{git} fetch {track_repo}",
                    desc="[Updater] Fetching from github...",
                    live=False,
                )
            except Exception:
                logging.error(
                    f"Update failed in fetching, check your network connection"
                )
                return "failed"

            run(
                f'{git} stash push --include-untracked -m "updater-{timestamp}"',
                desc=f"[Updater] Restoring you local changes on stash updater-{timestamp}",
                live=False,
            ) if need_stash else None

            run(f"{git} checkout -b {backup_branch}", live=False)
            run(f"{git} checkout -b {updater_branch}", live=False)
            run(f"{git} reset --hard FETCH_HEAD", live=False)
            run(
                f"{git} reset --hard {latest_release_hash}",
                desc=f"[Updater] Checking out {latest_release_tag}...",
                live=False,
            )
            run(f"{git} checkout {current_branch}", live=False)

            try:
                run(
                    f"{git} merge --no-edit {updater_branch} -q",
                    desc=f"[Updater] Trying to apply latest update on version {latest_release_tag}...",
                )
            except Exception:
                logging.error(f"Update failed in merging")
                try:
                    run(
                        f"{git} merge --abort",
                        desc="[Updater] Conflict detected, canceling update...",
                    )
                    run(f"{git} reset --hard {backup_branch}", live=False)
                    run(f"{git} branch -D -f {updater_branch}", live=False)
                    run(f"{git} branch -D -f {backup_branch}", live=False)
                    run(f"{git} stash pop", live=False) if need_stash else None
                    logging.error(
                        f"Update failed, but your file was safely reset to the state before the update."
                    )
                    return "failed"
                except Exception as e:
                    logging.error(
                        f"!!!Update failed in resetting, try to reset your files manually. {e}"
                    )
                    return "failed"

            if need_stash:
                try:
                    run(
                        f"{git} stash apply",
                        desc="[Updater] Trying to restore your local modifications...",
                        live=False,
                    )
                except Exception:
                    run(
                        f"{git} reset --hard {backup_branch}",
                        desc="[Updater] Conflict detected, canceling update...",
                        live=False,
                    )
                    run(f"{git} branch -D -f {updater_branch}", live=False)
                    run(f"{git} branch -D -f {backup_branch}", live=False)
                    run(f"{git} stash pop", live=False)
                    logging.error(
                        f"Update failed in applying your local changes, but your file was safely reset to the state before the update."
                    )
                    return "failed"
                run(f"{git} stash drop", live=False)

            run(f"{git} branch -D -f {updater_branch}", live=False)
            run(f"{git} branch -D -f {backup_branch}", live=False)
        except Exception as e:
            logging.error(f"Update failed: {e}")
            return "failed"
        if need_pip:
            try:
                run_pip(
                    f"install -r requirements.txt",
                    pref="[Updater]",
                    desc="requirements",
                    live=False,
                )
            except Exception:
                logging.error(f"Update failed in pip install")
                return "failed"
        return "success"
    except Exception as e:
        logging.error(f"Update failed: {e}")
        return "failed"