a11904b7c5
- 为 pam_deploy_graph 生产代码补充中文模块、类、函数/方法文档字符串 - 将原有英文说明和主要英文异常提示改为中文 - 新增当前整体逻辑结构流程图文档，覆盖模块结构、执行链路、action 路由、人工确认和 checkpoint 续跑 - 新增 Linux 自带运行环境打包脚本，使用 PyInstaller 生成解压即用目录和 tar.gz - 新增 Linux 打包说明，包含构建命令、运行方式、依赖说明和包大小评估 - 同步 README，补充流程图、打包方式、产物路径和大小预估 - 更新相关测试断言以匹配中文错误提示
120 lines
3.4 KiB
Python
120 lines
3.4 KiB
Python
"""通过子进程执行 deploy.sh / deploy.ps1 action。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import subprocess
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from .models import ActionResult
|
||
from .output_parser import parse_script_result
|
||
|
||
|
||
class ScriptActionRunner:
|
||
"""脚本 action runner，负责构造命令、执行脚本并解析结果。"""
|
||
|
||
def __init__(self, script_base_dir: str | Path = "doc_scripts") -> None:
|
||
"""保存脚本所在目录。"""
|
||
self.script_base_dir = Path(script_base_dir)
|
||
|
||
def run(
|
||
self,
|
||
action: str,
|
||
*,
|
||
params: dict[str, Any],
|
||
script_entry: str,
|
||
config_path: str,
|
||
ip: str | None = None,
|
||
hash_code: str | None = None,
|
||
stop_first: bool = False,
|
||
trace_file_path: str | None = None,
|
||
timeout_sec: int | None = None,
|
||
) -> ActionResult:
|
||
"""执行一个脚本 action，并返回统一 ActionResult。"""
|
||
command = self.build_command(
|
||
action,
|
||
script_entry=script_entry,
|
||
config_path=config_path,
|
||
ip=ip,
|
||
hash_code=hash_code,
|
||
stop_first=stop_first,
|
||
trace_file_path=trace_file_path,
|
||
)
|
||
completed = subprocess.run(
|
||
command,
|
||
cwd=str(self.script_base_dir),
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=timeout_sec,
|
||
check=False,
|
||
)
|
||
return parse_script_result(
|
||
action=action,
|
||
stdout=completed.stdout,
|
||
stderr=completed.stderr,
|
||
exit_code=completed.returncode,
|
||
backend="script",
|
||
tool_name=script_entry,
|
||
)
|
||
|
||
def build_command(
|
||
self,
|
||
action: str,
|
||
*,
|
||
script_entry: str,
|
||
config_path: str,
|
||
ip: str | None = None,
|
||
hash_code: str | None = None,
|
||
stop_first: bool = False,
|
||
trace_file_path: str | None = None,
|
||
) -> list[str]:
|
||
"""根据脚本类型构造 action 命令行参数。"""
|
||
if script_entry == "deploy.sh":
|
||
command = [
|
||
"bash",
|
||
"./deploy.sh",
|
||
"--config",
|
||
config_path,
|
||
"--action",
|
||
action,
|
||
]
|
||
if ip:
|
||
command.extend(["--ip", ip])
|
||
if hash_code:
|
||
command.extend(["--hash-code", hash_code])
|
||
if stop_first:
|
||
command.append("--stop-first")
|
||
if trace_file_path:
|
||
command.extend(["--trace-file", trace_file_path])
|
||
return command
|
||
|
||
if script_entry == "deploy.ps1":
|
||
command = [
|
||
"powershell",
|
||
"-File",
|
||
".\\deploy.ps1",
|
||
"-ConfigPath",
|
||
config_path,
|
||
"-Action",
|
||
action,
|
||
]
|
||
if ip:
|
||
command.extend(["-Ip", ip])
|
||
if hash_code:
|
||
command.extend(["-HashCode", hash_code])
|
||
if stop_first:
|
||
command.append("-RollbackStopFirst")
|
||
return command
|
||
|
||
raise ValueError(f"不支持的脚本入口: {script_entry}")
|
||
|
||
|
||
def select_script_entry(os_name: str | None = None) -> str:
|
||
"""根据操作系统选择默认脚本入口。"""
|
||
import platform
|
||
|
||
name = (os_name or platform.system()).lower()
|
||
if "windows" in name:
|
||
return "deploy.ps1"
|
||
return "deploy.sh"
|