| import signal |
| from typing import Literal |
| import subprocess |
| import time |
| import threading |
| import os |
| import signal |
|
|
| from environment import * |
|
|
|
|
| class AppRunner: |
| def __init__(self, run_type: RunType): |
| self.run_type = run_type |
| self.sub_proc = None |
|
|
| def start(self): |
| print(f"start app by type: {self.run_type}") |
| if self.run_type==RunType.electron: |
| cmd_args = [APP_PATH, f"--remote-debugging-port={DEBUG_PORT}"] |
| cwd = None |
| log_file = APP_LOG |
| elif self.run_type == RunType.code: |
| cmd_args = ["python", str(CODE_PATH)] |
| cwd = CODE_DIR |
| log_file = CODE_LOG |
| elif self.run_type == RunType.dev: |
| cmd_args = ["python", str(DEV_PATH)] |
| cwd = DEV_DIR |
| log_file = DEV_LOG |
| else: |
| raise TypeError(f"invalid run_type: {self.run_type}") |
| self.clear_log(log_file) |
| self.sub_proc = subprocess.Popen( |
| cmd_args, |
| cwd=cwd, |
| stdout=subprocess.DEVNULL, |
| stderr=subprocess.DEVNULL, |
| preexec_fn= os.setsid |
| ) |
| print(f"translator started, PID: {self.sub_proc.pid}") |
| time.sleep(10) |
| self.wait_for_app_start(log_file) |
| print(f"translator is ready.") |
| return self.sub_proc |
| def clear_log(self, log_path: Path): |
| if log_path.exists(): |
| log_path.unlink(missing_ok=True) |
|
|
| def wait_for_app_start(self, log_file, timeout = 60, interval=3): |
| for i in range(int(timeout/interval)): |
| ret = subprocess.run(f"tail -n 10 {log_file}", check=True, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, |
| universal_newlines=True) |
| if "Pipeline is ready" in ret.stdout: |
| return True |
| |
| print(f"app is not started yet, retry {i+1}...") |
| time.sleep(interval) |
| print(ret.stdout) |
| self.stop() |
| raise RuntimeError(f"app not started in {timeout}s") |
|
|
| def stop(self): |
| pgid = os.getpgid(self.sub_proc.pid) |
| os.killpg(pgid, signal.SIGTERM) |
| try: |
| time.sleep(3) |
| return_code = self.sub_proc.wait(timeout=10) |
| print(f"进程组已终止，进程退出码: {return_code}") |
| except subprocess.TimeoutExpired: |
| print("超时，进程组可能未完全终止，尝试 SIGKILL...") |
| os.killpg(pgid, signal.SIGKILL) |
| self.sub_proc.wait(timeout=10) |
| print("进程组已被强制杀死。") |
| except Exception as e: |
| print("停止进程失败:", e) |
|
|
| if __name__ == '__main__': |
| app = AppRunner(RunType.code) |
| app.start() |
| print(app.wait_for_app_start()) |
| app.stop() |
