142 lines
5.6 KiB
Python
142 lines
5.6 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""P1-02：窄平台进程适配 —— 显式 shell 契约 + 进程树终止 + 提示词平台段。
|
||
|
||
契约（不依赖 shell=True 的平台默认值）：
|
||
- Windows：字符串命令行 ``cmd.exe /d /s /c "<command>"``（/s 保留内部引号；
|
||
不能传 argv 列表——list2cmdline 的引号转义 cmd 不认）
|
||
- Linux： ``/bin/bash -lc <command>``（登录 shell，可读 profile 别名/函数）
|
||
|
||
进程树终止：
|
||
- Windows：taskkill /F /T（既有等价机制，整树强杀）
|
||
- Linux： start_new_session 独立 POSIX 进程组；SIGTERM 整组 → 宽限期 → SIGKILL 整组
|
||
（安全不变量：仅当 getpgid == 子进程 pid，即确认独立组后才 killpg，
|
||
否则退化为单进程 kill，绝不误杀调用方所在组）
|
||
|
||
系统提示词：通用正文 + 运行时插入的短平台 shell/path 段（占位符
|
||
``{{SHELL_PLATFORM_SECTION}}``）。不维护两份完整提示词。
|
||
"""
|
||
import os
|
||
import signal
|
||
import subprocess
|
||
import time
|
||
|
||
#: 系统提示词中的平台段占位符
|
||
SHELL_SECTION_PLACEHOLDER = "{{SHELL_PLATFORM_SECTION}}"
|
||
|
||
_KILL_GRACE_S = 3.0 # Linux：SIGTERM → SIGKILL 宽限
|
||
_TASKKILL_TIMEOUT_S = 10 # Windows：taskkill 自身超时
|
||
|
||
|
||
def is_windows() -> bool:
|
||
return os.name == "nt"
|
||
|
||
|
||
def shell_command(command: str):
|
||
"""显式 shell 命令（bash 工具的 shell 契约，替代 shell=True 平台默认值）。
|
||
|
||
- Windows：返回**字符串命令行** ``cmd.exe /d /s /c "<command>"``。
|
||
必须走字符串而非 argv 列表：列表形态会被 CPython ``list2cmdline``
|
||
给内部引号加反斜杠转义，cmd.exe 不认 ``\"``；字符串形态下命令行
|
||
原样交给 CreateProcessW，cmd 自行按 /s 规则解析 /c 参数
|
||
（外层引号剥离、内部引号保留，实测引号路径/&&/管道/%VAR% 均正确）。
|
||
- Linux：返回 argv 列表 ``[\ "/bin/bash", "-lc", command]``（POSIX 下
|
||
列表参数无引号转义问题，-l 登录 shell 可读 profile 别名/函数）。
|
||
"""
|
||
if is_windows():
|
||
return 'cmd.exe /d /s /c "' + command + '"'
|
||
return ["/bin/bash", "-lc", command]
|
||
|
||
|
||
def popen_flags() -> dict:
|
||
"""进程树管理所需的 Popen 附加参数：Linux 独立进程组。"""
|
||
if is_windows():
|
||
return {}
|
||
return {"start_new_session": True}
|
||
|
||
|
||
def kill_process_tree(proc, grace_s: float = _KILL_GRACE_S) -> None:
|
||
"""终止 proc 的整棵子进程树；最坏情况也保证返回（不抛异常）。
|
||
|
||
- Windows：taskkill /F /T 整树强杀（既有机制）。
|
||
- Linux：独立进程组 SIGTERM → 宽限 grace_s 秒 → SIGKILL 整组。
|
||
若无法确认独立组（进程已退 / 标志未生效）→ 退化单进程 kill。
|
||
"""
|
||
if proc is None:
|
||
return
|
||
|
||
if is_windows():
|
||
try:
|
||
subprocess.run(["taskkill", "/F", "/T", "/PID", str(proc.pid)],
|
||
capture_output=True, timeout=_TASKKILL_TIMEOUT_S)
|
||
except Exception:
|
||
try:
|
||
proc.kill()
|
||
except Exception:
|
||
pass
|
||
return
|
||
|
||
# ---- POSIX ----
|
||
pgid = None
|
||
try:
|
||
pgid = os.getpgid(proc.pid)
|
||
except Exception:
|
||
pass
|
||
if pgid is None or pgid != proc.pid:
|
||
# 进程已退出，或没有独立组 → 不能 killpg（可能误伤调用方组）
|
||
try:
|
||
proc.kill()
|
||
except Exception:
|
||
pass
|
||
return
|
||
|
||
def _pg(sig: int) -> None:
|
||
try:
|
||
os.killpg(pgid, sig)
|
||
except Exception:
|
||
pass
|
||
|
||
_pg(signal.SIGTERM)
|
||
deadline = time.time() + grace_s
|
||
while time.time() < deadline:
|
||
try:
|
||
if proc.poll() is not None:
|
||
return
|
||
except Exception:
|
||
return
|
||
time.sleep(0.05)
|
||
_pg(signal.SIGKILL)
|
||
|
||
|
||
# ----------------------------------------------------------------------
|
||
# 系统提示词平台段（短段，运行时插入；通用正文保持平台无关）
|
||
# ----------------------------------------------------------------------
|
||
|
||
_WIN_SECTION = """### 1.1 平台 shell 与路径（Windows）
|
||
|
||
- 操作系统：Windows。路径形如 `C:\\Users\\14890\\Desktop\\haocode`；不确定真实路径时先 `dir` / `ls`。
|
||
- bash 命令经 **cmd.exe** 执行（不是 git-bash）；`C:\\Program Files\\Git\\usr\\bin` 在 PATH 上，
|
||
`ls` `grep` `cat` `head` `tail` `wc` `rm` `sed` `awk` 可直接用，`|` `>` `2>&1` `&&` 可用。
|
||
- ⚠️ cmd 陷阱：`;` 不是分隔符（用 `&&` 或换行）；环境变量用 `%VAR%`（`$VAR` 不展开）；
|
||
内建命令不剥单引号（`echo 'x'` 原样带引号）；每次调用都是新进程，`cd` 不跨调用保留
|
||
（写 `cd core && <命令>`）；需要 `for` / `$(...)` / `[ ]` 测试等真实 bash 语义时
|
||
一律 `bash -c "..."`。"""
|
||
|
||
_LINUX_SECTION = """### 1.1 平台 shell 与路径（Linux）
|
||
|
||
- 操作系统：Linux。路径为 POSIX 风格（`/home/user/...`、`/tmp/...`）。
|
||
- bash 命令经 **/bin/bash -lc** 执行（登录 shell）：`;` `&&` `|` `$(...)`、单引号、
|
||
环境变量 `$VAR`、profile 里的别名与函数都可用。
|
||
- 每次调用都是新进程：`cd` 不跨调用保留（写 `cd core && <命令>` 或直接用绝对路径）。"""
|
||
|
||
|
||
def shell_prompt_section() -> str:
|
||
"""当前平台的短 shell/path 段。"""
|
||
return _WIN_SECTION if is_windows() else _LINUX_SECTION
|
||
|
||
|
||
def apply_platform_section(text: str) -> str:
|
||
"""把占位符替换为当前平台段；无占位符（如兜底提示词）时原样返回。"""
|
||
if SHELL_SECTION_PLACEHOLDER in text:
|
||
return text.replace(SHELL_SECTION_PLACEHOLDER, shell_prompt_section())
|
||
return text
|