Files
getDomain/domain-api/app/services/ops_execution_mode_service.py
2026-04-18 23:52:51 +08:00

61 lines
2.3 KiB
Python

from __future__ import annotations
def normalize_execution_modes(raw_modes: object, fallback_mode: str = "remote-agent") -> list[str]:
normalized_modes: list[str] = []
seen: set[str] = set()
fallback = str(fallback_mode or "remote-agent").strip() or "remote-agent"
for item in list(raw_modes or []):
mode = str(item or "").strip()
if not mode or mode in seen:
continue
seen.add(mode)
normalized_modes.append(mode)
if fallback and fallback not in seen:
normalized_modes.insert(0, fallback)
return normalized_modes or [fallback]
def execution_mode_label(execution_mode: str) -> str:
normalized_mode = str(execution_mode or "remote-agent").strip() or "remote-agent"
if normalized_mode == "ssh":
return "SSH"
if normalized_mode == "control-plane":
return "控制面"
if normalized_mode == "local-runtime":
return "本机运行时"
if normalized_mode == "remote-agent":
return "远端 Agent"
return normalized_mode
def build_execution_mode_options(raw_modes: object, fallback_mode: str = "remote-agent") -> list[dict]:
return [
{
"value": mode,
"label": execution_mode_label(mode),
}
for mode in normalize_execution_modes(raw_modes, fallback_mode)
]
def decorate_execution_mode_fields(
payload: dict | None = None,
*,
default_key: str = "default_execution_mode",
modes_key: str = "execution_modes",
) -> dict:
normalized_payload = dict(payload or {})
default_mode = str(normalized_payload.get(default_key) or "remote-agent").strip() or "remote-agent"
execution_mode_options = build_execution_mode_options(normalized_payload.get(modes_key) or [], default_mode)
normalized_payload[default_key] = default_mode
normalized_payload[modes_key] = [str(item.get("value") or "").strip() for item in execution_mode_options if str(item.get("value") or "").strip()]
normalized_payload["default_execution_mode_label"] = execution_mode_label(default_mode)
normalized_payload["execution_mode_options"] = execution_mode_options
normalized_payload["execution_mode_labels"] = [
str(item.get("label") or "").strip()
for item in execution_mode_options
if str(item.get("label") or "").strip()
]
return normalized_payload