from core.pocket_flow import AsyncNode from core.logger import logger from .cmd_node import async_run_cmd from pathlib import Path class GitNode(AsyncNode): def set_id(self, id): self.id = id async def prep_async(self, shared): return { "client": { "host": self.params[self.id].get("host"), "user": self.params[self.id].get("user"), "password": self.params[self.id].get("password"), "port": self.params[self.id].get("port"), "sudo": self.params[self.id].get("sudo", False), }, "git": { "url": self.params[self.id].get("git_url"), "user": self.params[self.id].get("git_user"), "password": self.params[self.id].get("git_password"), "branch": self.params[self.id].get("git_branch"), "path": self.params[self.id].get("git_path"), "clone_type": self.params[self.id].get("git_clone_type", "http/https"), "depth": self.params[self.id].get("git_clone_depth", 1) }, } async def exec_async(self, prep_res): client:dict = prep_res.get("client") git:dict = prep_res.get("git") path = git.get("path") git_path = str(Path(path) / ".git") parent_path = str(Path(path).parent) clone_url:str = git.get('url') clone_type = git.get("clone_type") clone_branch = git.get("branch") clone_depth = git.get("depth") if clone_type == "http/https": _info = clone_url.split("://") user = git.get("user") password = git.get("password") clone_url = f"{_info[0]}://{user}:{password}@{_info[-1]}" cmd = ( f"mkdir -p {parent_path} \n" f"cd {parent_path} \n" f"[ ! -d '{git_path}' ] && git clone -b {clone_branch} --single-branch --depth=1 --quiet {clone_url} \n" f"cd {path} \n" "git stash \n" f"git fetch --depth=1 origin {clone_branch} --quiet \n" f"git switch -C {clone_branch} FETCH_HEAD" ) if clone_depth == 0: cmd = f"{cmd} \n[ -e .git/shallow ] && git fetch --unshallow --quiet" logger.info(cmd) err = [] out = [] async for stream_type, line in async_run_cmd(**client, cmd=cmd): if stream_type == "stderr": err.append(str(line)) else: out.append(str(line)) err_str = "\n".join(err) out_str = "\n".join(out) # 补充一下密码失效问题 if "Sorry, try again." in out_str and not err_str: err_str = "SSH connection failed no password was provided" return {"err": err_str, "out": out_str} return None async def post_async(self, shared, prep_res, exec_res): shared[self.id] = exec_res if exec_res.get("err"): return "false" return "true"