#!/www/server/panel/pyenv/bin/python import os import re import shutil import sys import time import requests import traceback import tarfile import subprocess import argparse from concurrent.futures import ThreadPoolExecutor from typing import Optional, Tuple, List, Union, Dict, Iterable from hashlib import sha256, sha1 from platform import system, machine from xml.etree import cElementTree from io import FileIO __all__ = ['pygvm'] class _GvmSTD: out = sys.stdout err = sys.stderr _gvm_std = _GvmSTD() def get_arch() -> Tuple[Optional[str], Optional[str]]: _system_name = system() if _system_name != "Linux": return None, "目前只支持Linux系统使用该功能。" _arch = machine().lower() arch = None if _arch in ("x86_64", "amd64"): arch = "amd64" elif _arch in ("i386", "i486", "i586", "x86"): arch = "386" elif _arch in ("aarch64", "arm64"): arch = "arm64" if arch is None: return None, "未解析出当前处理器的适配的架构。" return arch, None class GVMError(Exception): pass template = """#pygvm export GOROOT="/usr/local/btgo" export PATH="/usr/local/btgo/bin:${PATH}" """ class _Mirror(object): """ 镜像源 """ _cloud = "" name = "" def test_speed(self, time_out: int) -> float: try: s_time = time.time() res = requests.get(self._cloud, timeout=time_out) if res.status_code != 200: return sys.float_info.max return time.time() - s_time except: return sys.float_info.max def get_versions_by_cloud(self, arch: str) -> Tuple[Union[List["GoVersion"], str], Optional[List["GoVersion"]]]: return "请求云端数据错误！请检查网络连接", None def download_url(self, go_v: "GoVersion") -> str: raise NotImplemented() class _StudyGolangMirror(_Mirror): _cloud = "https://studygolang.com/dl" name = "社区源" def get_versions_by_cloud(self, arch: str) -> Tuple[Union[List["GoVersion"], str], Optional[List["GoVersion"]]]: """ 获取云端支持的已归档版本和稳定版本 """ res = requests.get(self._cloud) if res.status_code != 200: return "请求云端数据错误！", None rep = r'(<\s?h3\s?id="stable"\s?>.*\n)(?P(.*\n)+)(\s*<\s?h3\s?id="unstable"\s?>)(.*\n)+' \ r'(?P(\s*)(.*\n)+)((\s*\s*\n){5}\s*" + res.group("stable") + "") history = cElementTree.fromstring(res.group("history")).find('./div[2]') stable_go_versions = self._parser_xml(stable, arch) history_go_versions = self._parser_xml(history, arch) stable_go_versions.sort(key=lambda x: int(x.version.split(".")[1]), reverse=True) history_go_versions.sort(key=lambda x: int(x.version.split(".")[1]), reverse=True) return stable_go_versions, history_go_versions except: traceback.print_exc(file=_gvm_std.err) return "解析错误", None @staticmethod def _parser_xml(datas: cElementTree.Element, arch: str) -> List["GoVersion"]: res_list = [] for data in datas.findall('./div'): v = data.attrib.get("id") need_file = "{}.linux-{}.tar.gz".format(v, arch) for i in data.iterfind('.//tr'): if i.attrib.get("class") == "first": continue file_name = i.find("./td[1]/a").text checksum = i.find("./td[6]/tt").text if need_file == file_name: go_v = GoVersion.new(v, checksum, file_name) if go_v: res_list.append(go_v) return res_list def download_url(self, go_v: "GoVersion") -> str: return "{}/golang/{}".format(self._cloud, go_v.file_name) class _GoOfficialDeVMirror(_Mirror): _cloud = "https://go.dev/dl/" name = "go.dev源" def get_versions_by_cloud(self, arch: str) -> Tuple[Union[List["GoVersion"], str], Optional[List["GoVersion"]]]: """ 获取云端支持的已归档版本和稳定版本 """ res = requests.get(self._cloud) if res.status_code != 200: return "请求云端数据错误！", None rep_stable_versions = re.compile(r'(\s+){3}') rep_history_versions = re.compile(r'(\s+){3}') stable_versions, history_versions = [], [] for i in rep_stable_versions.finditer(res.text): g = self._parser_xml_node(i.group(), arch) if g is not None: stable_versions.append(g) for i in rep_history_versions.finditer(res.text): g = self._parser_xml_node(i.group(), arch) if g is not None: history_versions.append(g) if not stable_versions and history_versions: return "解析错误", None return stable_versions, history_versions @staticmethod def _parser_xml_node(data: str, arch: str) -> Optional["GoVersion"]: try: node = cElementTree.fromstring(data) v = node.attrib.get("id") if v.find("rc") != -1 or v.find("beta") != -1: return None need_file = "{}.linux-{}.tar.gz".format(v, arch) for i in node.iterfind('.//tr'): if i.attrib.get("class") == "first": continue file_name = i.find("./td[1]/a").text checksum = i.find("./td[6]/tt").text if need_file == file_name: go_v = GoVersion.new(v, checksum, file_name) if go_v: return go_v except: return None return None def download_url(self, go_v: "GoVersion") -> str: return "{}/{}".format(self._cloud, go_v.file_name) class _GoOfficialCNMirror(_GoOfficialDeVMirror): _cloud = "https://golang.google.cn/dl/" name = "官方源" # 2026-01-16 验证中科大源无法使用 class _UstcMirror(_Mirror): _cloud = "https://mirrors.ustc.edu.cn/golang/" name = "中科大源" def get_versions_by_cloud(self, arch: str) -> Tuple[Union[List["GoVersion"], str], Optional[List["GoVersion"]]]: """ 获取云端支持的已归档版本和稳定版本 """ res = requests.get(self._cloud) if res.status_code != 200: return "请求云端数据错误！", None rep_version = re.compile(r"(?Pgo[\d.]+)\.linux-%s\.tar\.gz(\.sha256)?)" % arch) v_map = {} for i in rep_version.finditer(res.text): v = i.group("v") if v not in v_map: v_map[v] = [v, "", ""] f = i.group("f") if f.endswith(".sha256"): v_map[v][1] = f else: v_map[v][2] = f try: v_list = self._get_sha256(list(v_map.values())) except: v_list = [] if not v_list: return "解析错误", None return [], v_list def _get_sha256(self, v_list: List[Iterable[str]]) -> List["GoVersion"]: p = ThreadPoolExecutor(10) t_list = [] for idx, (v, s, f) in enumerate(v_list): if not s: continue t = p.submit(self._query_sha256, v_list, s, idx) t_list.append(t) for i in t_list: i.result() p.shutdown() use_list = [] for v, s, f in v_list: g = GoVersion.new(v, s, f) if g: use_list.append(g) return use_list def _query_sha256(self, v_list, s, idx): try: res = requests.get("{}/{}".format(self._cloud, s)) if res.status_code == 200: v_list[idx][1] = res.text except: v_list[idx][1] = "" def download_url(self, go_v: "GoVersion") -> str: return "{}/{}".format(self._cloud, go_v.file_name) def get_cloud() -> _Mirror: _m_list = [ _StudyGolangMirror(), _GoOfficialCNMirror(), _GoOfficialDeVMirror(), # 不稳定禁用 ] v4_file = '/www/server/panel/data/v4.pl' if os.path.exists(v4_file): try: with open(v4_file, 'rt', encoding='utf-8') as fp: if fp.read().strip() == "-4": _m_list = _m_list[1:] except: pass def test_with_time_out(time_out: int) -> _Mirror: p = ThreadPoolExecutor(len(_m_list)) res_task = [] for m in _m_list: res_task.append(p.submit(m.test_speed, time_out)) time_list = [] for idx, t in enumerate(res_task): time_list.append((t.result(timeout=None), idx)) time_list.sort(key=lambda x: x[0]) return _m_list[time_list[0][1]] try: return test_with_time_out(10) except: try: return test_with_time_out(60) except: pass return _Mirror() class GoVersion: __slots__ = ("version", "checksum", "file_name", "_cloud", "bt_gvm_path") def __init__(self, v: str, checksum: str, file_name: str): self.version = v self.checksum = checksum self.file_name = file_name self._cloud: Optional[_Mirror] = None self.bt_gvm_path = "/usr/local/btgojdk" def set_cloud(self, c: _Mirror): self._cloud = c print("正在使用【{}】源进行下载".format(c.name)) def check(self, file) -> bool: """calculate file sha256 """ print("[2/4] 校验文件hash值.....", file=_gvm_std.out, flush=True) sha_obj = sha256() if len(self.checksum) == 64 else sha1() with open(file, 'rb') as f: for chunk in iter(lambda: f.read(4096), b''): sha_obj.update(chunk) hash_value = sha_obj.hexdigest() print("计算值：{}".format(hash_value), file=_gvm_std.out, flush=True) print("验证值：{}".format(self.checksum), file=_gvm_std.out, flush=True) return hash_value == self.checksum @classmethod def new(cls, v: str, checksum: str, file_name: str) -> Optional["GoVersion"]: v_rep = r"(?Pgo1\.\d{1,2}(\.\d{1,2})?)" v_res = re.search(v_rep, v) v = None if v_res: v = v_res.group("target") checksum_rep = r"(?P([0-9a-f]{64})|([0-9a-f]{40}))" checksum_res = re.search(checksum_rep, checksum) checksum = None if checksum_res: checksum = checksum_res.group("target") if not v or not checksum_res: return None return cls(v, checksum, file_name) def __str__(self) -> str: return "[v:{}, checksum:{}]".format(self.version, self.checksum) def serialize(self) -> str: return "({}:{}:{})".format(self.version, self.checksum, self.file_name) def show(self) -> str: return self.version @classmethod def deserialize(cls, data: str) -> Optional["GoVersion"]: if data[0] == "(" and data[-1] == ")": data = data[1:-1] return cls.new(*data.split(":")[:3]) def _download(self) -> bool: cache_dir = os.path.join(self.bt_gvm_path, "cached") if not os.path.exists(cache_dir): os.makedirs(cache_dir) dst = os.path.join(cache_dir, self.file_name) if os.path.exists(dst): print("[1/4] 使用缓存的sdk文件......", file=_gvm_std.out, flush=True) return self.check(dst) print("[1/4] 下载sdk文件......", file=_gvm_std.out, flush=True) response = requests.get(self._cloud.download_url(self), stream=True) total_size = int(response.headers.get('content-length', 0)) print("需要下载的文件大小: %.2fM" % (total_size / (1024 * 1024)), file=_gvm_std.out, flush=True) block_size = 1024 * 1024 downloaded_size = 0 with open(dst, 'wb') as f: for data in response.iter_content(block_size): f.write(data) downloaded_size += len(data) progress = (downloaded_size / total_size) * 100 print(f"Downloading....\t %.2f%% completed" % progress, end='\r', flush=True, file=_gvm_std.out) print('Download complete', file=_gvm_std.out, flush=True) return self.check(dst) def _install(self) -> bool: print("[3/4] 解压安装.....", file=_gvm_std.out, flush=True) tar_filename = os.path.join(self.bt_gvm_path, "cached", self.file_name) target_directory = self.bt_gvm_path try: with tarfile.open(tar_filename, 'r:gz') as tar: tar.extractall(path=target_directory) except: return False src = os.path.join(self.bt_gvm_path, "go") if not os.path.isdir(src): return False os.rename(src, os.path.join(self.bt_gvm_path, self.version)) return True def install(self) -> Tuple[bool, str]: if not os.path.exists(self.bt_gvm_path): os.makedirs(self.bt_gvm_path) print("开始安装......", file=_gvm_std.out, flush=True) dst = os.path.join(self.bt_gvm_path, self.version) if os.path.isdir(dst): return True, "已安装" # 下载文件 if not self._download(): return False, "文件下载并验证失败！" # 安装go if not self._install(): return False, "解压安装失败！" print("[4/4] 使用{}.....".format(self.version), file=_gvm_std.out, flush=True) flag, err = self.use(self.bt_gvm_path, self.version) if not flag: return False, err print("安装完成!", file=_gvm_std.out, flush=True) print("若找不到go指令,可以尝试重新打开终端", file=_gvm_std.out, flush=True) return True, "" @classmethod def use(cls, bt_gvm_path, version) -> Tuple[bool, str]: # 建立软链接 src = os.path.join(bt_gvm_path, version) if not os.path.isdir(src): return False, "{}文件丢失。".format(version) dst = "/usr/local/btgo" if os.path.islink(dst): os.unlink(dst) os.symlink(src, dst) cls.add_to_path() time.sleep(0.1) out, err = cls.test() if version in out: return True, "" else: return False, err @staticmethod def test() -> Tuple[str, str]: # 执行命令并获取输出 os.environ["PATH"] = "/usr/local/btgo/bin:" + os.environ["PATH"] p = subprocess.Popen("go version", stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True, env=os.environ) output, err = p.communicate() return output.decode(), err.decode() @staticmethod def add_to_path(): profile = "/etc/profile" with open(profile, "r") as f: profile_data = f.read() if profile_data.find(template) != -1: return with open(profile, "a") as f: f.write("\n" + template) os.system("source {}".format(profile)) @staticmethod def parse_version(data: str) -> Tuple[bool, str]: v_rep = r"(?P1\.\d{1,2}(\.\d{1,2})?)" v_res = re.search(v_rep, data) v = None if v_res: v = v_res.group("target") return True, "go" + v else: return False, "" class GVM(object): stable_versions: Optional[List[GoVersion]] = None history_versions: Optional[List[GoVersion]] = None arch = "" use_shell = False bt_gvm_path = "/usr/local/btgojdk" now_version = None def __init__(self): self.get_now_version() if not os.path.exists(self.bt_gvm_path): os.makedirs(self.bt_gvm_path) self._cloud = None @staticmethod def check_use(): res = os.popen("lsattr /etc/profile") return res.read().find("--i--") == -1 @property def cloud(self): if not isinstance(self._cloud, _Mirror): self._cloud = get_cloud() return self._cloud def get_now_version(self) -> None: now_version = "/usr/local/btgo" if os.path.islink(now_version): now_version = os.path.basename(os.readlink(now_version)) else: now_version = None self.now_version = now_version def __new__(cls, *args, **kwargs) -> Union["GVM", str]: arch, errmsg = get_arch() if errmsg: return errmsg _gvm = super().__new__(cls, *args, **kwargs) _gvm.arch = arch return _gvm # 获取版本 def get_go_version(self): self.stable_versions, self.history_versions = self._get_versions_by_local() if not self.history_versions: if self.use_shell: print("未找到本地记录文件，将向云端请求GO版本数据，这可能需要一段时间，请稍等", file=_gvm_std.out) self.stable_versions, self.history_versions = self._get_versions_by_cloud() # 缓存数据到本地 if isinstance(self.stable_versions, list): self._save_cached(self.stable_versions, self.history_versions) if not self.history_versions and isinstance(self.stable_versions, str): raise GVMError("网络错误：" + self.stable_versions) return self.stable_versions, self.history_versions def _get_versions_by_local(self) -> [Tuple[Optional[List[GoVersion]], Optional[List[GoVersion]]]]: """ 获取本地已归档版本和稳定版本的缓存数据 """ local_path = "/www/server/panel/data/pygvm" if not os.path.exists(local_path): os.makedirs(local_path) return None, None stable_file = os.path.join(local_path, "stable_versions.txt") history_file = os.path.join(local_path, "history_versions.txt") if not os.path.isfile(stable_file) or not os.path.isfile(history_file): return None, None with open(stable_file, "r") as f: _arch = f.readline().strip() if _arch != self.arch: return None, None stable_versions = [GoVersion.deserialize(line.strip()) for line in f.readlines()] with open(history_file, "r") as f: _arch = f.readline().strip() if _arch != self.arch: return None, None history_versions = [GoVersion.deserialize(line.strip()) for line in f.readlines()] return stable_versions, history_versions def _get_versions_by_cloud(self) -> Tuple[Union[List[GoVersion], str], Optional[List[GoVersion]]]: """ 获取云端支持的已归档版本和稳定版本 """ return self.cloud.get_versions_by_cloud(self.arch) def _save_cached(self, stable_go_versions: List[GoVersion], history_go_versions: List[GoVersion]) -> None: local_path = "/www/server/panel/data/pygvm" if not os.path.exists(local_path): os.makedirs(local_path) with open(os.path.join(local_path, "stable_versions.txt"), "w") as f: f.write(self.arch + "\n") for go_v in stable_go_versions: f.write(go_v.serialize() + "\n") with open(os.path.join(local_path, "history_versions.txt"), "w") as f: f.write(self.arch + "\n") for go_v in history_go_versions: f.write(go_v.serialize() + "\n") @staticmethod def del_cached(): local_path = "/www/server/panel/data/pygvm" stable_file = os.path.join(local_path, "stable_versions.txt") history_file = os.path.join(local_path, "history_versions.txt") if os.path.isfile(stable_file): os.remove(stable_file) if os.path.isfile(history_file): os.remove(history_file) def api_ls(self) -> List[str]: return [i.strip() for i in os.listdir(self.bt_gvm_path) if i.startswith("go1")] def cmd_ls(self) -> None: versions = [i.strip() for i in os.listdir(self.bt_gvm_path) if i.startswith("go1")] if not versions: print("未安装任何一个版本的GO语言SDK") return print("version: ") for i in versions: if i == self.now_version: i += " <- Now use this version" print(" " + i) def api_ls_remote(self, is_all: bool) -> Tuple[Optional[List[GoVersion]], Optional[List[GoVersion]], Optional[str]]: try: self.get_go_version() except GVMError as e: return [], [], str(e) if is_all: self.stable_versions.sort(key=lambda x: int(x.version.split(".")[1]), reverse=True) self.history_versions.sort(key=lambda x: int(x.version.split(".")[1]), reverse=True) return self.stable_versions, self.history_versions, None stable_v = [".".join(i.version.split(".")[:2]) for i in self.stable_versions] res_history = {} for i in self.history_versions: parse_list = i.version.split(".") v = ".".join(parse_list[:2]) if len(parse_list) > 2: # 三级版本号 three_idx = int(parse_list[2]) else: three_idx = 0 # 在稳定版中已存在的就不在展示 if v in stable_v: continue if v in res_history: if three_idx > res_history[v][0]: res_history[v] = [three_idx, i] else: res_history[v] = [three_idx, i] res_h = [] for i in res_history.values(): res_h.append(i[1]) res_h.sort(key=lambda x: int(x.version.split(".")[1]), reverse=True) return self.stable_versions, res_h, None def cmd_ls_remote(self, is_all: bool) -> None: stable, history, err = self.api_ls_remote(is_all) installed = self.api_ls() if err: print("获取版本信息时出错了", file=sys.stderr) print(err, file=sys.stderr) print("Stable Version:") for i in stable: if i.version in installed: i.version += " <- installed" print(" " + i.version) print("History Version:") for i in history: if i.version in installed: i.version += " <- installed" print(" " + i.version) def _get_version(self, version) -> Union[GoVersion, str]: stable, history, err = self.api_ls_remote(True) if err: if self.use_shell: print("获取版本信息时出错了", file=_gvm_std.err) print(err, file=_gvm_std.err) return err for i in stable: if i.version == version: return i for i in history: if i.version == version: return i if self.use_shell: print("未找到对应版本", file=_gvm_std.err) return "未找到对应版本" def api_use(self, version) -> Tuple[bool, str]: versions = self.api_ls() if version not in versions: return False, "未找到对应版本" res = GoVersion.use(self.bt_gvm_path, version) self.get_now_version() return res def cmd_use(self, version) -> None: flag, err = self.api_use(version) if flag: print("已切换到:{}".format(version)) print("若找不到go指令,可以尝试重新打开终端") else: print(err, file=sys.stderr) def api_install(self, version) -> Tuple[bool, str]: go_v = self._get_version(version) if isinstance(go_v, str): return False, go_v go_v.set_cloud(self.cloud) return go_v.install() def cmd_install(self, version) -> None: go_v = self._get_version(version) go_v.set_cloud(self.cloud) if isinstance(go_v, str): pass _, err = go_v.install() if err: print(err, file=sys.stderr) def api_uninstall(self, version: str) -> Tuple[bool, str]: versions = self.api_ls() if version not in versions: return False, "未找到对应版本" next_version = None for i in versions: if i != version: next_version = i break if next_version is not None and version == self.now_version: GoVersion.use(self.bt_gvm_path, next_version) target_dir = os.path.join(self.bt_gvm_path, version) if os.path.exists: shutil.rmtree(target_dir, ignore_errors=True) return True, "" def cmd_uninstall(self, version: str) -> None: flag, err = self.api_uninstall(version) if flag: print("已成功卸载:{}".format(version)) else: print(err, file=sys.stderr) def cmd_clean_cache(self) -> None: local_path = "/www/server/panel/data/pygvm" for i in ("stable_versions.txt", "history_versions.txt"): if os.path.isfile(local_path + "/" + i): os.remove(local_path + "/" + i) cache_dir = os.path.join(self.bt_gvm_path, "cached") if os.path.isdir(cache_dir): shutil.rmtree(cache_dir) return @staticmethod def set_goproxy(proxy: str) -> bool: if os.path.isfile("/usr/local/btgo/bin/go"): try: subprocess.check_output([ "/usr/local/btgo/bin/go", "env", "-w", "GOPROXY=" + proxy, "GO111MODULE=on" ]) return True except Exception: pass return False @staticmethod def get_goproxy() -> str: if os.path.isfile("/usr/local/btgo/bin/go"): try: out = subprocess.check_output(["/usr/local/btgo/bin/go", "env", "GOPROXY"]) if out: return out.decode("utf-8").strip() except: pass return "" @staticmethod def set_std(out: FileIO, err: FileIO) -> None: _gvm_std.out = out _gvm_std.err = err pygvm = GVM() if __name__ == '__main__': parser = argparse.ArgumentParser(description='gvm go语言sdk管理器') # 添加子命令 subparsers = parser.add_subparsers(title='operation', dest='command') # 添加ls子命令 subparsers.add_parser('ls', help='展示已安装的Go语言SDK版本') # 添加ls子命令 subparsers.add_parser('clean-cache', help='清除缓存') # 添加ls_r子命令 parser_ls_r = subparsers.add_parser('ls-remote', help='展示可安装Go语言SDK的版本，默认只展示每个版本中较新的版本') parser_ls_r.add_argument('-a', action='store_true', help='展示可以安装的所有Go语言SDK版本') # 添加use子命令 parser_use = subparsers.add_parser('use', help='使用指定版本') parser_use.add_argument('use_param', type=str, help='完整的版本号') # 添加install子命令 parser_install = subparsers.add_parser('install', help='安装指定版本') parser_install.add_argument('install_param', type=str, help='完整的版本号') # 添加uninstall子命令 parser_uninstall = subparsers.add_parser('uninstall', help='卸载并删除指定版本') parser_uninstall.add_argument('uninstall_param', type=str, help='完整的版本号') input_args = parser.parse_args() gvm = GVM() if isinstance(gvm, str): print(gvm, file=sys.stderr) exit(1) gvm.use_shell = True if input_args.command == 'ls': gvm.cmd_ls() elif input_args.command == "ls-remote": _is_all = True if input_args.a else False gvm.cmd_ls_remote(_is_all) elif input_args.command == "use": _flag, _v = GoVersion.parse_version(input_args.use_param) if _flag: gvm.cmd_use(_v) else: print("版本参数错误，应该是1.xx.xx的结构", file=sys.stderr) elif input_args.command == "install": _flag, _v = GoVersion.parse_version(input_args.install_param) if _flag: gvm.cmd_install(_v) else: print("版本参数错误，应该是1.xx.xx的结构", file=sys.stderr) elif input_args.command == "uninstall": _flag, _v = GoVersion.parse_version(input_args.uninstall_param) if _flag: gvm.cmd_uninstall(_v) else: print("版本参数错误，应该是1.xx.xx的结构", file=sys.stderr) elif input_args.command == "clean-cache": gvm.cmd_clean_cache() else: print("使用pygvm -h 查看操作指令", file=sys.stderr)