(cmd: Union[List[str], str], call_log: Optional[Callable[[str], None]] = None, user='')
| 35 | |
| 36 | |
| 37 | def _run_command_with_call_log(cmd: Union[List[str], str], call_log: Optional[Callable[[str], None]] = None, user='') -> Optional[str]: |
| 38 | if not callable(call_log): |
| 39 | call_log = lambda x: print(x) |
| 40 | |
| 41 | if user and user != "root": |
| 42 | res = pwd.getpwnam(user) |
| 43 | uid = res.pw_uid |
| 44 | gid = res.pw_gid |
| 45 | |
| 46 | def pre_exec_fn(): |
| 47 | os.setgid(gid) |
| 48 | os.setuid(uid) |
| 49 | else: |
| 50 | pre_exec_fn = None |
| 51 | |
| 52 | # 执行命令 |
| 53 | try: |
| 54 | process = subprocess.Popen( |
| 55 | cmd, |
| 56 | stdout=subprocess.PIPE, |
| 57 | stderr=subprocess.STDOUT, |
| 58 | text=True, |
| 59 | shell=True, |
| 60 | preexec_fn=pre_exec_fn |
| 61 | ) |
| 62 | |
| 63 | # 实时读取输出 |
| 64 | while True: |
| 65 | output = process.stdout.readline() # type: ignore |
| 66 | if output == '' and process.poll() is not None: |
| 67 | break |
| 68 | if output: |
| 69 | call_log(output.strip()) |
| 70 | |
| 71 | # 检查返回码 |
| 72 | if process.returncode != 0: |
| 73 | error_msg = "执行失败，返回码：{}".format(process.returncode) |
| 74 | call_log(error_msg) |
| 75 | return error_msg |
| 76 | |
| 77 | return None |
| 78 | |
| 79 | except Exception as e: |
| 80 | error_msg = str(e) |
| 81 | call_log(error_msg) |
| 82 | return error_msg |
| 83 | |
| 84 | |
| 85 | def python_manager_path() -> str: |
no test coverage detected