# -*- coding:utf-8 -*- import os import sys import subprocess from functools import lru_cache import logging import gradio as gr import datetime # 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 "" def commit_html(): commit = commit_hash() if commit != "": short_commit = commit[0:7] commit_info = f'{short_commit}' 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 = "" except Exception: tag = "" if tag == "": tag_info = "unknown \U0001F615" elif tag == "": tag_info = f'{latest_tag}*' else: tag_info = f'{tag}' 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: {python_version}  •  Gradio: {gr.__version__}  •  ChuanhuChat: {repo_version} """ def version_time(): try: commit_time = subprocess.check_output(f"TZ=UTC {git} log -1 --format=%cd --date='format-local:%Y-%m-%dT%H:%M:%SZ'", shell=True, encoding='utf8').strip() # commit_time = run(f"TZ=UTC {git} log -1 --format=%cd --date='format-local:%Y-%m-%dT%H:%M:%SZ'").strip() 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 = "" 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 = "" 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 = "" 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"