wandering1
842d711ed6
用户裁定「我不想看到编号」。测试验证层级这一族的整族清退分三道并行：本笔是**代码与 CI 道** （CI 步骤与门禁输出面），代码注释面另走、文档面由文档道走。 口径（旧 → 语义名 → ASCII 段名，段名只进日志/旗标/job id/文件名）： L0 静态门禁 static ｜ L0a 钩子解释器预检 preflight ｜ L0b 提交信息门禁 commitmsg L0c 体积门禁 size ｜ L0.5 主机侧纯逻辑单测 host-unit ｜ L0.5c 覆盖率 coverage - scripts/ci_local.py 43 站点：段标题与 [SUMMARY] 标签改 ASCII 段名（static: / host-unit:），函数 run_l0/run_l05 → run_static_gates/run_host_unit_tests， 收尾「已跑：」列表用段名。旗标改 --static-only / --host-unit-only，**旧旗标 --l0-only / --l05-only 保留为 argparse 别名**（肌肉记忆与既有脚本片段不断档， 实测 --help 两名列出）。 - .gitea/workflows/ci.yml 42 站点：job id l0-gates → static-gates、 l05-host-unit-tests → host-unit-tests；步骤名与 step-summary 标题改语义名； 日志名 l0-pre-commit.log → static-pre-commit.log、l05-<模块>.log → host-unit-<模块>.log；::error::/::notice:: 文案同步。yaml.safe_load 核过 两个 job id 与步骤名。 host-unit-tests；步骤名与 step-summary 标题改语义名；日志名 l0-pre-commit.log → static-pre-commit.log、l05-<模块>.log → host-unit-<模块>.log；::error::/::notice:: 文案同步。yaml.safe_load 核过两个 job id 与步骤名。 - 门禁脚本 18 站点：check_registration（登记簿理由串 7）、check_test_coverage 5、 README 8、check_generated_fresh 2、ram_baseline 2、check_size_gate 1、 ci_find_python.ps1 2、ci_runner_bootstrap.ps1 2、.pre-commit-config.yaml 2、 requirements.txt 1（保持纯 ASCII）。 - skills 7 站点：task-worktree（含 [SUMMARY] 样例两行）、mem-budget、ui-verify、 task_worktree.py 的实跑标签。 - AGENTS.md 3 站点：CI 段「只跑静态门禁 + 主机侧纯逻辑单测」、自查命令段、 「2026-10-11 静态门禁全灭」事件名。 - 归档：编号治理历史.md 新增 §10 换算表 + **撞族警示**（同形 L0/L1/L2/L3 还属 UI 设计 令牌层 / UI 菜单层级 / 采集链 / 链路时延段 / CubeMX 三层 / 硬件位号与 FTP 应答字面量 等六个族，其中 6 处在字符串字面量里，改了就是改行为——故全程只按行内置换、先判族）。 - docs/产品需求 §7.4 测试分层表：去号 + 顺带修掉上一批**错族替换疤**（该表 L1/L2 行曾被 采集链层名「语义取数与有效性 / 全周傅氏相量」「保护判据」顶替，现按内容改写为板端测试 套件 / 主机侧回归两行）。同文档 §1.5 G5 判据行同步。 门禁：ci_local --help 两面旗标并列；py_compile 7 个 .py 全过；yaml 解析通过； check_ids_new --worktree rc=0；git diff --check 干净；17 个改动文件 FFFD=0、无 CR、 BOM 与原文件一致（含 requirements.txt 纯 ASCII）。本族无机器识别面（单字母+数字是 新增行门禁的自述盲区），口径来自自建逐行扫描器，复扫本族形态归零。 - 交接说明：文档道 teammate 在改 CI 指南时中断，剩余 9 处双写疤与 29 处 CI 标识 （job id / 日志名 / 汇总行字面量）由 Lead 接手改完，逐处仍走断言式行内置换。
917 lines
37 KiB
Python
917 lines
37 KiB
Python
#!/usr/bin/env python3
|
||
"""task-worktree — 多任务并行开发的空间管理（git worktree 生命周期）
|
||
|
||
执行体 = 本脚本（纯标准库）。规格见 docs/环境与部署/并行任务工作树方案.md，
|
||
用法与铁律见同目录 SKILL.md。
|
||
|
||
退出码：0=成功 / 1=操作失败 / 2=前置或环境错误 / 3=用法错误
|
||
"""
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import re
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
import time
|
||
from pathlib import Path
|
||
|
||
# 输出统一 UTF-8，避免被控制台代码页（cp936）转码破坏
|
||
for _stream in (sys.stdout, sys.stderr):
|
||
try:
|
||
_stream.reconfigure(encoding="utf-8")
|
||
except Exception:
|
||
pass
|
||
|
||
EXIT_OK, EXIT_FAIL, EXIT_ENV, EXIT_USAGE = 0, 1, 2, 3
|
||
|
||
BRANCH_PREFIX = "task/"
|
||
MAX_CONCURRENT = 3
|
||
LOCK_NAMES = ("build", "bench")
|
||
# 按锁分设陈旧窗口（小时）——兜底窗口应与「操作真实时长」同量级：
|
||
# build：编译是秒~分钟级（keil_build.py 自持自释，持有窗口 ≈ UV4 运行时长，
|
||
# 上限 = --timeout，默认 600s）；崩溃泄漏时应在数十分钟内即可接管 → 1h。
|
||
# bench：会话级（flash → @test → 探针，天然几十分钟~数小时）→ 保留 6h。
|
||
# 另：锁 payload 里持有者可用 `ttl` 字段自报窗口（见 lock_acquire(ttl=...)），
|
||
# 优先于这里的默认值——这样 build 锁的窗口能自动跟随其 --timeout。
|
||
LOCK_TTL_HOURS = {"build": 1.0, "bench": 6.0}
|
||
DEFAULT_LOCK_TTL_HOURS = 6.0
|
||
JUNCTION_SPECS = (
|
||
(".venv", "Python 环境"),
|
||
("simulator/SDL2", "模拟器 SDL2 SDK"),
|
||
)
|
||
FILE_ATTRIBUTE_REPARSE_POINT = 0x400
|
||
|
||
ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]*$")
|
||
SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9-]*$")
|
||
|
||
|
||
class Fail(Exception):
|
||
"""操作失败（退出码 1）"""
|
||
|
||
|
||
class EnvError(Exception):
|
||
"""前置或环境错误（退出码 2）"""
|
||
|
||
|
||
# --------------------------------------------------------------------- 仓库定位
|
||
|
||
|
||
def _main_worktree_of(start: Path):
|
||
"""借 git 反查「主工作树」路径——脚本可能被从任务工作树里调用，不能只靠 __file__"""
|
||
for extra in (["--path-format=absolute"], []):
|
||
r = subprocess.run(
|
||
["git", "-C", str(start), "rev-parse", *extra, "--git-common-dir"],
|
||
capture_output=True, text=True, encoding="utf-8", errors="replace",
|
||
stdin=subprocess.DEVNULL,
|
||
)
|
||
if r.returncode == 0 and r.stdout.strip():
|
||
common = Path(r.stdout.strip())
|
||
if not common.is_absolute():
|
||
common = (start / common).resolve()
|
||
return common.parent if common.name == ".git" else common
|
||
return None
|
||
|
||
|
||
_SCRIPT_ROOT = Path(__file__).resolve().parents[4]
|
||
REPO_ROOT = _main_worktree_of(_SCRIPT_ROOT) or _SCRIPT_ROOT
|
||
WT_ROOT = Path(os.environ.get("DTU_WT_ROOT") or (REPO_ROOT.parent / "_wt"))
|
||
BASE_BRANCH = os.environ.get("DTU_TASK_BASE", "v2-on-ui")
|
||
|
||
STATE_DIR = WT_ROOT / ".state"
|
||
LOCKS_DIR = STATE_DIR / "locks"
|
||
|
||
|
||
# --------------------------------------------------------------------- 基础工具
|
||
|
||
|
||
def git(*args, cwd=None, check=True, env=None):
|
||
cmd = ["git"] + [str(a) for a in args]
|
||
full_env = dict(os.environ)
|
||
if env:
|
||
full_env.update(env)
|
||
r = subprocess.run(
|
||
cmd, cwd=str(cwd or REPO_ROOT), capture_output=True, text=True,
|
||
encoding="utf-8", errors="replace", stdin=subprocess.DEVNULL, env=full_env,
|
||
)
|
||
if check and r.returncode != 0:
|
||
raise Fail("git {} 失败（rc={}）\n{}".format(
|
||
" ".join(cmd[1:]), r.returncode, (r.stdout + r.stderr).strip()))
|
||
return r
|
||
|
||
|
||
def git_out(*args, **kw):
|
||
return git(*args, **kw).stdout.strip()
|
||
|
||
|
||
def worktree_entries():
|
||
out = git_out("worktree", "list", "--porcelain")
|
||
entries, cur = [], None
|
||
for line in out.splitlines():
|
||
if line.startswith("worktree "):
|
||
if cur:
|
||
entries.append(cur)
|
||
cur = {"path": Path(line[len("worktree "):]), "branch": None}
|
||
elif cur is None:
|
||
continue
|
||
elif line.startswith("branch refs/heads/"):
|
||
cur["branch"] = line[len("branch refs/heads/"):]
|
||
if cur:
|
||
entries.append(cur)
|
||
return entries
|
||
|
||
|
||
def main_worktree() -> Path:
|
||
"""集成点（合并发生的地方）。默认 = 主工作树；可用环境变量覆盖，便于演练/测试。"""
|
||
override = os.environ.get("DTU_INTEGRATION_WORKTREE")
|
||
if override:
|
||
p = Path(override)
|
||
if not p.is_absolute():
|
||
p = (REPO_ROOT / p).resolve()
|
||
if not p.is_dir():
|
||
raise EnvError("DTU_INTEGRATION_WORKTREE 不存在：{}".format(p))
|
||
return p
|
||
entries = worktree_entries()
|
||
if not entries:
|
||
raise EnvError("无法解析 git worktree 列表")
|
||
return entries[0]["path"]
|
||
|
||
|
||
def venv_python(tree: Path = None) -> Path:
|
||
"""优先用该工作树自己的 .venv（任务树里是指向主树的联接），否则退回主工作树"""
|
||
for root in (tree, REPO_ROOT):
|
||
if root is None:
|
||
continue
|
||
p = Path(root) / ".venv" / "Scripts" / "python.exe"
|
||
if p.is_file():
|
||
return p
|
||
return Path(sys.executable)
|
||
|
||
|
||
def run_script(script: str, args, cwd: Path, label: str, tail: int = 40) -> bool:
|
||
"""跑项目脚本并回报。输出可能上千行，只回显尾部（结论与报错都在尾部）
|
||
|
||
⚠ 脚本必须取自 **cwd 那棵树**，不能固定用 REPO_ROOT：
|
||
项目脚本普遍以 `__file__` 反推仓库根（如 `scripts/ci_local.py`、
|
||
`scripts/check_registration.py`），若拿主树的脚本、却给集成树的 cwd，
|
||
脚本仍会去校验主树——门禁就测错了对象（2026-10-12 实测踩到：
|
||
报出主树才有的 iec104 未注册缺口，而集成树里根本没那些文件）。
|
||
"""
|
||
script_path = Path(cwd) / script
|
||
if not script_path.is_file():
|
||
script_path = REPO_ROOT / script
|
||
cmd = [str(venv_python(Path(cwd))), str(script_path)] + [str(x) for x in args]
|
||
r = subprocess.run(
|
||
cmd, cwd=str(cwd), capture_output=True, text=True,
|
||
encoding="utf-8", errors="replace", stdin=subprocess.DEVNULL,
|
||
)
|
||
out = (r.stdout + r.stderr).strip()
|
||
print("[{}] {}".format("ok" if r.returncode == 0 else "!!", label))
|
||
if out:
|
||
lines = out.splitlines()
|
||
if len(lines) > tail:
|
||
print(" …（前 {} 行略）".format(len(lines) - tail))
|
||
lines = lines[-tail:]
|
||
print(" " + "\n ".join(lines))
|
||
return r.returncode == 0
|
||
|
||
|
||
# --------------------------------------------------------------------- 台账
|
||
|
||
|
||
def state_file(task_id: str) -> Path:
|
||
return STATE_DIR / (task_id + ".json")
|
||
|
||
|
||
def load_state(task_id: str) -> dict:
|
||
p = state_file(task_id)
|
||
if not p.is_file():
|
||
raise EnvError("任务 {} 不存在（缺台账 {}）".format(task_id, p))
|
||
return json.loads(p.read_text(encoding="utf-8"))
|
||
|
||
|
||
def save_state(st: dict) -> None:
|
||
STATE_DIR.mkdir(parents=True, exist_ok=True)
|
||
state_file(st["id"]).write_text(
|
||
json.dumps(st, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
|
||
|
||
def all_states():
|
||
if not STATE_DIR.is_dir():
|
||
return []
|
||
out = []
|
||
for p in sorted(STATE_DIR.glob("*.json")):
|
||
try:
|
||
out.append(json.loads(p.read_text(encoding="utf-8")))
|
||
except Exception:
|
||
continue
|
||
return out
|
||
|
||
|
||
# --------------------------------------------------------------------- 独占锁
|
||
|
||
|
||
def lock_age_hours(info) -> float:
|
||
try:
|
||
return max(0.0, (time.time() - float(info.get("epoch") or 0)) / 3600.0)
|
||
except (TypeError, ValueError):
|
||
return float("inf")
|
||
|
||
|
||
def lock_ttl_hours(name: str, info=None) -> float:
|
||
"""锁 `name` 的陈旧判定窗口（小时）。
|
||
|
||
优先用锁 payload 里持有者自报的 `ttl`（例如 keil_build.py 按其 --timeout 声明），
|
||
否则回落到该锁的默认窗口（build 1h / bench 6h）。
|
||
"""
|
||
if info:
|
||
try:
|
||
v = info.get("ttl")
|
||
if v:
|
||
return max(0.0, float(v))
|
||
except (TypeError, ValueError):
|
||
pass
|
||
return LOCK_TTL_HOURS.get(name, DEFAULT_LOCK_TTL_HOURS)
|
||
|
||
|
||
def lock_is_stale(name: str, info) -> bool:
|
||
return lock_age_hours(info) > lock_ttl_hours(name, info)
|
||
|
||
|
||
def lock_path(name: str) -> Path:
|
||
return LOCKS_DIR / (name + ".lock")
|
||
|
||
|
||
def lock_read(name: str):
|
||
p = lock_path(name)
|
||
if not p.is_file():
|
||
return None
|
||
try:
|
||
return json.loads(p.read_text(encoding="utf-8"))
|
||
except Exception:
|
||
return {"task": "?", "pid": 0, "epoch": 0, "at": "?"}
|
||
|
||
|
||
def lock_acquire(name: str, task: str, wait: int = 0, force: bool = False, note: str = "",
|
||
ttl: float = None):
|
||
"""抢占独占资源。
|
||
|
||
⚠ 锁**不**绑定进程存活：`lock acquire` 本身是短命进程，而用锁的烧录/构建
|
||
在另一个进程里跑。因此判「陈旧」只看**时效**（超过该锁的 TTL），外加显式
|
||
`--force`；pid 仅作记录便于排障。TTL 按锁分设（build 1h / bench 6h），
|
||
持有者也可用 `ttl`（小时）自报更精确的窗口（如 build 按 --timeout 声明）。
|
||
"""
|
||
LOCKS_DIR.mkdir(parents=True, exist_ok=True)
|
||
p = lock_path(name)
|
||
deadline = time.time() + max(0, wait)
|
||
while True:
|
||
try:
|
||
fd = os.open(str(p), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
|
||
except FileExistsError:
|
||
info = lock_read(name) or {}
|
||
if info.get("task") == task:
|
||
return True, info
|
||
age = lock_age_hours(info)
|
||
ttl_h = lock_ttl_hours(name, info)
|
||
if force or age > ttl_h:
|
||
print("[lock] 接管 {} 锁（原持有者 {}，已持有 {:.1f} 小时 / TTL {:.1f} 小时{}）".format(
|
||
name, info.get("task"), age, ttl_h, "，--force" if force else "，视为陈旧"))
|
||
try:
|
||
p.unlink()
|
||
except OSError:
|
||
pass
|
||
continue
|
||
if time.time() >= deadline:
|
||
return False, info
|
||
print("[lock] {} 被 {} 占用（已持有 {:.1f} / TTL {:.1f} 小时），等待…".format(
|
||
name, info.get("task"), age, ttl_h))
|
||
time.sleep(2)
|
||
else:
|
||
now = time.time()
|
||
payload = {
|
||
"name": name, "task": task, "pid": os.getpid(),
|
||
"host": os.environ.get("COMPUTERNAME", "?"),
|
||
"epoch": now,
|
||
"at": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(now)),
|
||
"note": note,
|
||
"ttl": ttl,
|
||
}
|
||
os.write(fd, (json.dumps(payload, ensure_ascii=False) + "\n").encode("utf-8"))
|
||
os.close(fd)
|
||
return True, payload
|
||
|
||
|
||
def lock_release(name: str, task=None, force: bool = False):
|
||
p = lock_path(name)
|
||
if not p.is_file():
|
||
return True, "锁 {} 未被持有".format(name)
|
||
info = lock_read(name) or {}
|
||
if not force and task and info.get("task") not in (task, None):
|
||
return False, "锁 {} 由 {} 持有，与 {} 不符；强制释放请加 --force".format(
|
||
name, info.get("task"), task)
|
||
try:
|
||
p.unlink()
|
||
except OSError as e:
|
||
return False, "释放锁 {} 失败：{}".format(name, e)
|
||
return True, "已释放锁 {}（原持有者 {}）".format(name, info.get("task"))
|
||
|
||
|
||
# --------------------------------------------------------------------- 目录联接
|
||
|
||
|
||
def is_reparse(p: Path) -> bool:
|
||
try:
|
||
st = os.lstat(str(p))
|
||
except OSError:
|
||
return False
|
||
return bool(getattr(st, "st_file_attributes", 0) & FILE_ATTRIBUTE_REPARSE_POINT)
|
||
|
||
|
||
def make_junction(link: Path, target: Path) -> None:
|
||
"""创建 Windows 目录联接（不需要管理员权限，也不是符号链接）
|
||
|
||
走 `mklink /J`——这是唯一不需要 SeCreateSymbolicLinkPrivilege 的目录链接形式。
|
||
⚠ 实测（2026-10-12）：直接用 ctypes 调 `DeviceIoControl(FSCTL_SET_REPARSE_POINT)`
|
||
会被系统拒绝（Win32 4392 / ERROR_REPARSE_TAG_INVALID，缓冲区经 hexdump 核对无误），
|
||
而 `mklink /J` 正常，故采用后者。参数以**列表**交给 subprocess，
|
||
含空格的路径（本项目 `DTU Code`）由 subprocess 自动加引号，不经手写引号。
|
||
"""
|
||
if os.name != "nt":
|
||
raise EnvError("目录联接是 Windows 专属能力；当前平台 {}".format(os.name))
|
||
if os.path.lexists(str(link)):
|
||
raise Fail("联接位置已被占用：{}".format(link))
|
||
if not target.exists():
|
||
raise Fail("联接源不存在：{}".format(target))
|
||
link.parent.mkdir(parents=True, exist_ok=True)
|
||
r = subprocess.run(
|
||
["cmd", "/c", "mklink", "/J", str(link), str(target)],
|
||
capture_output=True, stdin=subprocess.DEVNULL,
|
||
)
|
||
if r.returncode != 0 or not is_reparse(link):
|
||
detail = (r.stdout or b"").decode("mbcs", "replace").strip()
|
||
raise Fail("创建目录联接失败：{} -> {}\n rc={} {}".format(
|
||
link, target, r.returncode, detail))
|
||
|
||
|
||
def remove_junction(link: Path) -> bool:
|
||
"""只摘掉联接本身，绝不触碰目标内容（os.rmdir = RemoveDirectoryW）"""
|
||
if not is_reparse(link):
|
||
return False
|
||
try:
|
||
os.rmdir(str(link))
|
||
return True
|
||
except OSError as e:
|
||
print("[warn] 摘除联接失败 {}：{}".format(link, e))
|
||
return False
|
||
|
||
|
||
def bootstrap(wt: Path):
|
||
"""建立共享联接。单个联接失败只告警不中止——避免留下半成品任务空间"""
|
||
made, failed = [], []
|
||
for rel, desc in JUNCTION_SPECS:
|
||
target = REPO_ROOT / rel
|
||
if not target.exists():
|
||
print("[skip] {}（{}）不存在，跳过联接".format(rel, desc))
|
||
continue
|
||
try:
|
||
make_junction(wt / rel, target)
|
||
except (Fail, EnvError) as e:
|
||
failed.append(rel)
|
||
print("[WARN] {} 联接失败：{}".format(rel, e))
|
||
continue
|
||
made.append(rel)
|
||
print("[ok] 联接 {} -> {}".format(rel, target))
|
||
if failed:
|
||
print()
|
||
print("!! {} 个共享联接未建成；该任务里跑 `.venv\\Scripts\\python.exe` 会找不到解释器。".format(len(failed)))
|
||
for rel in failed:
|
||
print(" 手工补：cmd /c mklink /J \"{}\\{}\" \"{}\\{}\"".format(wt, rel, REPO_ROOT, rel))
|
||
return made
|
||
|
||
|
||
# --------------------------------------------------------------------- 子进程脚本
|
||
|
||
|
||
def regen(cwd: Path) -> None:
|
||
run_script("scripts/gen_task_table.py", ["--write"], cwd, "重生成 FreeRTOS 任务表")
|
||
if list((Path(cwd) / "MDK-ARM").glob("**/*.map")):
|
||
run_script("scripts/check_mem_numbers.py", ["--write"], cwd, "同步内存/Flash 快照")
|
||
else:
|
||
print("[skip] 未找到 .map，跳过内存快照同步（需先 @build）")
|
||
|
||
|
||
def run_gates(cwd: Path, skip_build: bool) -> bool:
|
||
ok = run_script("scripts/ci_local.py", [], cwd, "静态门禁 + 主机侧纯逻辑单测 本地等价 CI")
|
||
if skip_build:
|
||
print("[skip] 按 --skip-build 跳过 @build")
|
||
return ok
|
||
got, info = lock_acquire("build", "__done__", wait=900)
|
||
if not got:
|
||
print("!! build 锁被 {} 占用（UV4 不能并发构建），跳过 @build".format(info.get("task")))
|
||
return False
|
||
# 告知 @build「调用方已持锁」：新版 keil_build.py 自持自释（见其 docstring），
|
||
# 读到本变量即跳过，避免自我死锁；**旧版根本不看此变量、也不会自锁**，
|
||
# 故此处对「工作树里还是旧脚本」的在途任务同样安全（向后兼容）。
|
||
os.environ["DTU_LOCK_HELD"] = "build"
|
||
try:
|
||
if not run_script(".dsh/skills/keil-build/scripts/keil_build.py", [], cwd,
|
||
"@build（0 错 0 警）"):
|
||
ok = False
|
||
finally:
|
||
os.environ.pop("DTU_LOCK_HELD", None)
|
||
lock_release("build", "__done__", force=True)
|
||
return ok
|
||
|
||
|
||
# --------------------------------------------------------------------- 合并辅助
|
||
|
||
|
||
def write_merge_message(st: dict, base: str, cwd: Path) -> Path:
|
||
raw = git_out("log", "--format=%s", "{}..{}".format(base, st["branch"]), cwd=cwd)
|
||
subjects = [s.strip() for s in raw.splitlines() if s.strip()]
|
||
title = "merge: {} 合入 {}".format(st["branch"], base)
|
||
if len(title) > 100:
|
||
title = title[:97] + "..."
|
||
lines = [title, "", "任务 {}（{} 个提交）经 task-worktree 回收：".format(st["id"], len(subjects))]
|
||
for s in subjects:
|
||
lines.append("- " + (s[:73] + "..." if len(s) > 76 else s))
|
||
if st.get("note"):
|
||
note = st["note"]
|
||
lines.append("")
|
||
lines.append("备注：" + (note[:73] + "..." if len(note) > 76 else note))
|
||
fd, path = tempfile.mkstemp(prefix="taskwt-merge-", suffix=".txt")
|
||
os.close(fd)
|
||
Path(path).write_text("\n".join(lines) + "\n", encoding="utf-8", newline="\n")
|
||
return Path(path)
|
||
|
||
|
||
def audit_residue(base_before: str, branch: str, cwd: Path) -> None:
|
||
"""并集合并残留审计：双方都改过的文件会被 git 自动并成并集"""
|
||
mb = git_out("merge-base", base_before, branch, cwd=cwd)
|
||
task_files = set(filter(None, git_out("diff", "--name-only", mb, branch, cwd=cwd).splitlines()))
|
||
base_files = set(filter(None, git_out("diff", "--name-only", mb, base_before, cwd=cwd).splitlines()))
|
||
both = sorted(task_files & base_files)
|
||
if not both:
|
||
print("[ok] 没有「双方都改过」的文件，无并集残留风险")
|
||
return
|
||
print("以下 {} 个文件双方都改过，git 会自动并成并集，需人工核对：".format(len(both)))
|
||
for f in both[:30]:
|
||
print(" - " + f)
|
||
if len(both) > 30:
|
||
print(" …（其余 {} 个略）".format(len(both) - 30))
|
||
hits = []
|
||
for f in both:
|
||
p = cwd / f
|
||
if not p.is_file():
|
||
continue
|
||
try:
|
||
lines = p.read_text(encoding="utf-8", errors="replace").splitlines()
|
||
except OSError:
|
||
continue
|
||
dups = [i + 1 for i in range(1, len(lines))
|
||
if lines[i] == lines[i - 1] and lines[i].strip()]
|
||
if dups:
|
||
hits.append((f, dups))
|
||
if hits:
|
||
print()
|
||
print("!! 连续重复行（并集合并残留嫌疑）：")
|
||
for f, dups in hits:
|
||
print(" - {} 第 {} 行".format(f, ", ".join(str(d) for d in dups[:8])))
|
||
else:
|
||
print("[ok] 未检测到连续重复行")
|
||
print("提示：权威判据是 base / 两个 tip 三方对比（git show <ref>:<file>），别只看结果文件。")
|
||
|
||
|
||
def harvest_memory(st: dict) -> None:
|
||
"""任务工作树会被删除，先把它自己的记忆归并回主工作树，避免丢上下文"""
|
||
src = Path(st["path"]) / ".workbuddy-ai" / "memory"
|
||
if not src.is_dir():
|
||
return
|
||
dst_dir = REPO_ROOT / ".workbuddy-ai" / "memory"
|
||
dst_dir.mkdir(parents=True, exist_ok=True)
|
||
for f in sorted(src.glob("*.md")):
|
||
text = f.read_text(encoding="utf-8", errors="replace").strip()
|
||
if not text:
|
||
continue
|
||
dst = dst_dir / f.name
|
||
marker = "## [task-worktree 归并自任务 {} / {}]".format(st["id"], st["slug"])
|
||
with dst.open("a", encoding="utf-8", newline="\n") as fh:
|
||
fh.write("\n" + marker + "\n\n" + text + "\n")
|
||
print("[ok] 记忆归并 {} -> {}".format(f.name, dst))
|
||
|
||
|
||
def teardown_worktree(st: dict, force: bool = False, base: str = None) -> None:
|
||
wt = Path(st["path"])
|
||
base = base or st.get("base")
|
||
# 1) 先摘联接——绝不能让 git 递归进联接去删目标内容（.venv / SDL2 会被一起删掉）
|
||
for rel in st.get("junctions", []):
|
||
if remove_junction(wt / rel):
|
||
print("[ok] 已摘除联接 {}".format(rel))
|
||
# 2) 归并任务侧记忆
|
||
if wt.is_dir():
|
||
harvest_memory(st)
|
||
# 3) 移除工作树
|
||
r = git("worktree", "remove", *(["--force"] if force else []), str(wt), check=False)
|
||
if r.returncode != 0:
|
||
raise Fail("移除工作树失败：{}\n（提示：先处理未跟踪文件，或加 --force）".format(
|
||
(r.stdout + r.stderr).strip()))
|
||
# 4) 删分支。⚠ 不用 `git branch -d`——它按**环境 HEAD** 判断是否已合并，
|
||
# 集成点被 DTU_INTEGRATION_WORKTREE 改走时会误判。改为显式核对 base。
|
||
if force or git("merge-base", "--is-ancestor", st["branch"], base,
|
||
check=False).returncode == 0:
|
||
r = git("branch", "-D", st["branch"], check=False)
|
||
if r.returncode != 0:
|
||
print("[warn] 删除分支 {} 失败：{}".format(
|
||
st["branch"], (r.stdout + r.stderr).strip()))
|
||
else:
|
||
print("[ok] 已删除分支 {}".format(st["branch"]))
|
||
else:
|
||
print("[warn] 分支 {} 未并入 {}，保留不删（确认可弃再 `git branch -D {}`）".format(
|
||
st["branch"], base, st["branch"]))
|
||
state_file(st["id"]).unlink(missing_ok=True)
|
||
|
||
|
||
# --------------------------------------------------------------------- 子命令
|
||
|
||
|
||
def cmd_new(a) -> int:
|
||
if not ID_RE.match(a.task_id):
|
||
raise EnvError("任务 ID 非法：{}（字母数字开头，可用 . _ -）".format(a.task_id))
|
||
if not SLUG_RE.match(a.slug):
|
||
raise EnvError("slug 非法：{}（小写字母数字与连字符）".format(a.slug))
|
||
branch = "{}{}-{}".format(BRANCH_PREFIX, a.task_id, a.slug)
|
||
wt = WT_ROOT / "{}-{}".format(a.task_id, a.slug)
|
||
|
||
if state_file(a.task_id).is_file():
|
||
raise EnvError("任务 {} 已存在（{}）".format(a.task_id, state_file(a.task_id)))
|
||
if os.path.lexists(str(wt)):
|
||
raise EnvError("目录已存在：{}".format(wt))
|
||
if git("rev-parse", "--verify", "--quiet", "refs/heads/" + branch, check=False).returncode == 0:
|
||
raise EnvError("分支已存在：{}".format(branch))
|
||
if git("rev-parse", "--verify", "--quiet", a.base, check=False).returncode != 0:
|
||
raise EnvError("基线分支不存在：{}".format(a.base))
|
||
|
||
active = [s for s in all_states() if Path(s["path"]).is_dir()]
|
||
if len(active) >= MAX_CONCURRENT and not a.force:
|
||
raise EnvError("已有 {} 个活跃任务（上限 {}）；先 done/abort，或加 --force".format(
|
||
len(active), MAX_CONCURRENT))
|
||
|
||
WT_ROOT.mkdir(parents=True, exist_ok=True)
|
||
print("[1/3] 建工作树（分支 {}，基线 {}{}）…".format(
|
||
branch, a.base, "，跳过 LFS smudge" if a.lean else ""))
|
||
git("worktree", "add", "-b", branch, str(wt), a.base,
|
||
env={"GIT_LFS_SKIP_SMUDGE": "1"} if a.lean else None)
|
||
|
||
print("[2/3] 建立共享联接…")
|
||
junctions = [] if a.no_bootstrap else bootstrap(wt)
|
||
|
||
st = {
|
||
"id": a.task_id, "slug": a.slug, "branch": branch, "path": str(wt),
|
||
"base": a.base, "created_at": time.strftime("%Y-%m-%d %H:%M:%S"),
|
||
"junctions": junctions, "needs_bench": bool(a.needs_bench),
|
||
"bench_pass": False, "note": a.note or "", "lean": bool(a.lean),
|
||
}
|
||
save_state(st)
|
||
print("[3/3] 台账已写入 {}".format(state_file(a.task_id)))
|
||
print()
|
||
print("下一步：把新会话的工作目录切到")
|
||
print(" " + str(wt))
|
||
print("在该目录里干活、跑 @build 与 ci_local.py（并行安全）；")
|
||
print("要动台架先 `lock acquire bench --task {}`，完事 `lock release bench --task {} --bench-pass`。".format(
|
||
a.task_id, a.task_id))
|
||
print("回收用 `done {}`（**必须在主工作树执行**）。".format(a.task_id))
|
||
return EXIT_OK
|
||
|
||
|
||
def cmd_list(a) -> int:
|
||
states = all_states()
|
||
if not states:
|
||
print("（没有任务空间。用 `new <ID> <slug>` 建一个）")
|
||
return EXIT_OK
|
||
print("{:<12} {:<32} {:>5} {:>5} {:>4} {}".format(
|
||
"任务", "分支", "落后", "领先", "脏", "台架"))
|
||
for st in states:
|
||
wt = Path(st["path"])
|
||
if not wt.is_dir():
|
||
print("{:<12} {:<32} {:>5} {:>5} {:>4} {}".format(
|
||
st["id"], st["branch"], "-", "-", "-", "[工作树缺失]"))
|
||
continue
|
||
ab = git_out("rev-list", "--left-right", "--count",
|
||
"{}...{}".format(st["base"], st["branch"]), cwd=wt, check=False).split()
|
||
behind, ahead = (ab[0], ab[1]) if len(ab) == 2 else ("?", "?")
|
||
dirty = len(git_out("status", "--porcelain", cwd=wt, check=False).splitlines())
|
||
bench = "PASS" if st.get("bench_pass") else ("需测" if st.get("needs_bench") else "-")
|
||
print("{:<12} {:<32} {:>5} {:>5} {:>4} {}".format(
|
||
st["id"], st["branch"], behind, ahead, dirty, bench))
|
||
print()
|
||
for n in LOCK_NAMES:
|
||
info = lock_read(n)
|
||
if not info:
|
||
print("锁 {:<6} 空闲".format(n))
|
||
else:
|
||
print("锁 {:<6} {} 持有（已 {:.1f} 小时，于 {}）".format(
|
||
n, info.get("task"), lock_age_hours(info), info.get("at")))
|
||
return EXIT_OK
|
||
|
||
|
||
def cmd_status(a) -> int:
|
||
st = load_state(a.task_id)
|
||
wt = Path(st["path"])
|
||
print("任务 {} ({})".format(st["id"], st["slug"]))
|
||
print("分支 {} ← 基线 {}".format(st["branch"], st["base"]))
|
||
print("工作树 {}{}".format(wt, "" if wt.is_dir() else " [缺失]"))
|
||
print("建立于 {}".format(st.get("created_at")))
|
||
if st.get("note"):
|
||
print("备注 {}".format(st["note"]))
|
||
if wt.is_dir():
|
||
dirty = git_out("status", "--porcelain", cwd=wt, check=False).splitlines()
|
||
print("未提交 {} 项".format(len(dirty)))
|
||
for line in dirty[:10]:
|
||
print(" " + line)
|
||
ab = git_out("rev-list", "--left-right", "--count",
|
||
"{}...{}".format(st["base"], st["branch"]), cwd=wt, check=False).split()
|
||
if len(ab) == 2:
|
||
print("相对基线 落后 {} / 领先 {} 个提交".format(ab[0], ab[1]))
|
||
print("台架 {}{}".format(
|
||
"需要，" if st.get("needs_bench") else "不涉及，",
|
||
"已记录 PASS" if st.get("bench_pass") else "尚无 PASS 记录"))
|
||
if st.get("junctions"):
|
||
print("联接 {}".format(", ".join(st["junctions"])))
|
||
return EXIT_OK
|
||
|
||
|
||
def cmd_sync(a) -> int:
|
||
st = load_state(a.task_id)
|
||
wt = Path(st["path"])
|
||
base = a.base or st["base"]
|
||
if not wt.is_dir():
|
||
raise EnvError("工作树不存在：{}".format(wt))
|
||
if git_out("status", "--porcelain", cwd=wt, check=False):
|
||
raise Fail("任务工作树有未提交改动，先提交或 stash 再 sync")
|
||
r = git("rebase", base, cwd=wt, check=False)
|
||
if r.returncode != 0:
|
||
print((r.stdout + r.stderr).strip())
|
||
raise Fail("rebase 有冲突：在 {} 里解决后 `git rebase --continue`，放弃用 `git rebase --abort`".format(wt))
|
||
print("[ok] 任务 {} 已 rebase 到 {}".format(st["id"], base))
|
||
return EXIT_OK
|
||
|
||
|
||
def cmd_lock(a) -> int:
|
||
if a.action == "status":
|
||
for n in LOCK_NAMES:
|
||
info = lock_read(n)
|
||
if not info:
|
||
print("{:<6} 空闲（TTL {:.1f}h）".format(n, lock_ttl_hours(n)))
|
||
else:
|
||
age = lock_age_hours(info)
|
||
ttl_h = lock_ttl_hours(n, info)
|
||
mark = " [陈旧，可被自动接管]" if age > ttl_h else ""
|
||
print("{:<6} {} 持有（pid={}，于 {}，已 {:.1f} / TTL {:.1f} 小时，备注 {}）{}".format(
|
||
n, info.get("task"), info.get("pid"), info.get("at"), age,
|
||
ttl_h, info.get("note") or "-", mark))
|
||
return EXIT_OK
|
||
|
||
if a.name not in LOCK_NAMES:
|
||
raise EnvError("未知资源 {}（可用：{}）".format(a.name, " / ".join(LOCK_NAMES)))
|
||
|
||
if a.action == "acquire":
|
||
if not a.task:
|
||
raise EnvError("acquire 需要 --task <ID>")
|
||
got, info = lock_acquire(a.name, a.task, wait=a.wait, force=a.force,
|
||
note=a.note or "", ttl=a.ttl)
|
||
if not got:
|
||
print("获取失败：{} 被 {} 占用（pid={}，于 {}）".format(
|
||
a.name, info.get("task"), info.get("pid"), info.get("at")))
|
||
return EXIT_FAIL
|
||
print("已获取 {} 锁（task={} pid={} ttl={}）".format(
|
||
a.name, a.task, os.getpid(),
|
||
"{:.2f}h".format(a.ttl) if a.ttl else "默认"))
|
||
return EXIT_OK
|
||
|
||
if a.bench_pass:
|
||
if not a.task:
|
||
raise EnvError("--bench-pass 需要 --task <ID>")
|
||
st = load_state(a.task)
|
||
st["bench_pass"] = True
|
||
save_state(st)
|
||
print("[ok] 已为任务 {} 记录台架 PASS".format(a.task))
|
||
ok, msg = lock_release(a.name, a.task, force=a.force)
|
||
print(msg)
|
||
return EXIT_OK if ok else EXIT_FAIL
|
||
|
||
|
||
def release_stale_locks() -> None:
|
||
"""`done` 前置：陈旧锁自动接管；仍被有效占用的则拒绝收口。
|
||
|
||
「陈旧」= 持有时间 > 该锁 TTL（build 1h / bench 6h，或持有者自报的 ttl）。
|
||
崩溃/遗忘留下的锁不该阻塞收口——故自动释放而非硬失败（2026-10-12）。
|
||
"""
|
||
for n in LOCK_NAMES:
|
||
info = lock_read(n)
|
||
if not info:
|
||
continue
|
||
age = lock_age_hours(info)
|
||
ttl_h = lock_ttl_hours(n, info)
|
||
if age > ttl_h:
|
||
ok, msg = lock_release(n, force=True)
|
||
print("[i] {} 锁已陈旧（{} 持有 {:.1f}h > TTL {:.1f}h），自动接管：{}".format(
|
||
n, info.get("task"), age, ttl_h, msg))
|
||
continue
|
||
raise Fail("独占资源 {} 仍被 {} 占用（已 {:.1f} / TTL {:.1f} 小时），先 release".format(
|
||
n, info.get("task"), age, ttl_h))
|
||
|
||
|
||
def cmd_done(a) -> int:
|
||
st = load_state(a.task_id)
|
||
wt = Path(st["path"])
|
||
branch = st["branch"]
|
||
base = a.base or st["base"]
|
||
main = main_worktree()
|
||
|
||
print("=== 前置检查 ===")
|
||
if not wt.is_dir():
|
||
raise EnvError("工作树不存在：{}".format(wt))
|
||
if git_out("status", "--porcelain", cwd=wt, check=False):
|
||
raise Fail("任务工作树有未提交改动，先提交")
|
||
if git_out("status", "--porcelain", cwd=main, check=False):
|
||
raise Fail("主工作树不干净（{}）——合并会把未提交改动卷进来".format(main))
|
||
cur = git_out("rev-parse", "--abbrev-ref", "HEAD", cwd=main)
|
||
if cur != base:
|
||
raise Fail("主工作树当前在 {}，不是集成分支 {}；先切过去再 done".format(cur, base))
|
||
release_stale_locks()
|
||
|
||
ahead = int(git_out("rev-list", "--count", "{}..{}".format(base, branch), cwd=wt) or "0")
|
||
already = git("merge-base", "--is-ancestor", branch, base, cwd=main, check=False).returncode == 0
|
||
if ahead == 0 and not already:
|
||
raise Fail("任务分支相对 {} 没有新提交，无需合并".format(base))
|
||
if already and ahead == 0:
|
||
print("[i] {} 已合入 {}，跳过合并，直接清场".format(branch, base))
|
||
print("=== 清场 ===")
|
||
teardown_worktree(st, force=a.force, base=base)
|
||
print("\n完成（分支此前已合入）。")
|
||
return EXIT_OK
|
||
|
||
if st.get("needs_bench") and not st.get("bench_pass"):
|
||
raise Fail("任务标记 --needs-bench 但无台架 PASS 记录；跑完 @test 后执行 "
|
||
"`lock release bench --task {} --bench-pass`".format(st["id"]))
|
||
print("[ok] 前置检查通过（领先 {} 共 {} 个提交）".format(base, ahead))
|
||
|
||
print("=== 合并预演（git merge-tree，不改动任何东西）===")
|
||
r = git("merge-tree", "--write-tree", base, branch, cwd=main, check=False)
|
||
if r.returncode != 0:
|
||
print((r.stdout + r.stderr).strip())
|
||
raise Fail("预演发现冲突：到任务工作树里 `git merge {}` 解决并提交后重跑 done".format(base))
|
||
tree = (r.stdout.splitlines() or ["?"])[0].strip()
|
||
print("[ok] 无冲突，预演结果树 {}".format(tree))
|
||
if a.dry_run:
|
||
print()
|
||
print("--dry-run：到此为止，未做任何改动。")
|
||
return EXIT_OK
|
||
|
||
print("=== 合入主干（{}，--no-ff）===".format(main))
|
||
base_before = git_out("rev-parse", "HEAD", cwd=main)
|
||
msgfile = write_merge_message(st, base, wt)
|
||
r = git("merge", "--no-ff", "--no-edit", branch, "-F", str(msgfile), cwd=main, check=False)
|
||
msgfile.unlink(missing_ok=True)
|
||
if r.returncode != 0:
|
||
raise Fail("合并失败：\n{}".format((r.stdout + r.stderr).strip()))
|
||
print((r.stdout or "").strip())
|
||
base_after = git_out("rev-parse", "HEAD", cwd=main)
|
||
|
||
print("=== 合并残留审计 ===")
|
||
audit_residue(base_before, branch, cwd=main)
|
||
|
||
print("=== 重生成受管产物 ===")
|
||
regen(main)
|
||
|
||
if a.skip_gates:
|
||
print("=== 门禁（--skip-gates 已跳过）===")
|
||
gates_ok = True
|
||
else:
|
||
print("=== 门禁 ===")
|
||
gates_ok = run_gates(main, skip_build=a.skip_build)
|
||
|
||
if not gates_ok:
|
||
print()
|
||
print("!! 门禁未全绿：工作树与分支**保留**，未清场。")
|
||
print(" 主干已含合并提交 {}".format(base_after[:12]))
|
||
print(" 可选：a) 在主工作树修正后追加提交（推荐）")
|
||
print(" b) 回退本次合并：git -C \"{}\" reset --hard {}".format(main, base_before[:12]))
|
||
return EXIT_FAIL
|
||
|
||
print("=== 清场 ===")
|
||
teardown_worktree(st, force=a.force, base=base)
|
||
print()
|
||
print("完成。主干 {} 现为 {}。".format(base, base_after[:12]))
|
||
print("push 需人工许可，本次未推送。")
|
||
return EXIT_OK
|
||
|
||
|
||
def cmd_abort(a) -> int:
|
||
st = load_state(a.task_id)
|
||
wt = Path(st["path"])
|
||
if wt.is_dir():
|
||
dirty = git_out("status", "--porcelain", cwd=wt, check=False).splitlines()
|
||
ahead = int(git_out("rev-list", "--count",
|
||
"{}..{}".format(st["base"], st["branch"]), cwd=wt,
|
||
check=False) or "0")
|
||
if (dirty or ahead) and not a.force:
|
||
print("任务 {} 有 {} 项未提交改动、{} 个未合并提交，abort 会**永久丢弃**：".format(
|
||
st["id"], len(dirty), ahead))
|
||
for line in dirty[:20]:
|
||
print(" " + line)
|
||
raise Fail("确认丢弃请加 --force")
|
||
teardown_worktree(st, force=True)
|
||
print("\n已放弃任务 {}（工作树与分支均已删除）".format(st["id"]))
|
||
return EXIT_OK
|
||
|
||
|
||
# --------------------------------------------------------------------- 入口
|
||
|
||
|
||
def build_parser():
|
||
ap = argparse.ArgumentParser(
|
||
prog="task_worktree.py",
|
||
description="多任务并行开发的空间管理（git worktree 生命周期）")
|
||
sub = ap.add_subparsers(dest="cmd")
|
||
|
||
p = sub.add_parser("new", help="新建任务空间")
|
||
p.add_argument("task_id")
|
||
p.add_argument("slug")
|
||
p.add_argument("--base", default=BASE_BRANCH, help="基线分支（默认 %(default)s）")
|
||
p.add_argument("--lean", action="store_true",
|
||
help="跳过 Git LFS smudge（省约 38MB，参考图变成指针文件）")
|
||
p.add_argument("--no-bootstrap", action="store_true", help="不建共享联接")
|
||
p.add_argument("--needs-bench", action="store_true",
|
||
help="需台架验证；无 PASS 记录则 done 拒绝合并")
|
||
p.add_argument("--note", default="")
|
||
p.add_argument("--force", action="store_true", help="忽略并发上限")
|
||
p.set_defaults(func=cmd_new)
|
||
|
||
p = sub.add_parser("list", help="列出所有任务空间")
|
||
p.set_defaults(func=cmd_list)
|
||
|
||
p = sub.add_parser("status", help="单个任务详情")
|
||
p.add_argument("task_id")
|
||
p.set_defaults(func=cmd_status)
|
||
|
||
p = sub.add_parser("sync", help="rebase 追平主干")
|
||
p.add_argument("task_id")
|
||
p.add_argument("--base", default=None)
|
||
p.set_defaults(func=cmd_sync)
|
||
|
||
p = sub.add_parser("lock", help="独占资源锁（build / bench）")
|
||
p.add_argument("action", choices=["acquire", "release", "status"])
|
||
p.add_argument("name", nargs="?", choices=list(LOCK_NAMES))
|
||
p.add_argument("--task", default=None)
|
||
p.add_argument("--wait", type=int, default=0, help="等待秒数（0=不等待）")
|
||
p.add_argument("--force", action="store_true")
|
||
p.add_argument("--note", default="")
|
||
p.add_argument("--ttl", type=float, default=None,
|
||
help="本次持有自报的陈旧窗口（小时）；缺省用该锁默认值（build 1h / bench 6h）")
|
||
p.add_argument("--bench-pass", action="store_true",
|
||
help="release bench 时顺带为该任务记录台架 PASS")
|
||
p.set_defaults(func=cmd_lock)
|
||
|
||
p = sub.add_parser("done", help="回收：预演 → 合入主干 → 审计 → 重生成 → 门禁 → 清场")
|
||
p.add_argument("task_id")
|
||
p.add_argument("--base", default=None)
|
||
p.add_argument("--dry-run", action="store_true", help="只预演，不做任何改动")
|
||
p.add_argument("--skip-build", action="store_true", help="跳过 @build")
|
||
p.add_argument("--skip-gates", action="store_true", help="跳过全部门禁")
|
||
p.add_argument("--force", action="store_true", help="清场时强制移除工作树")
|
||
p.set_defaults(func=cmd_done)
|
||
|
||
p = sub.add_parser("abort", help="放弃任务（不合并）")
|
||
p.add_argument("task_id")
|
||
p.add_argument("--force", action="store_true")
|
||
p.set_defaults(func=cmd_abort)
|
||
return ap
|
||
|
||
|
||
def main(argv=None) -> int:
|
||
ap = build_parser()
|
||
a = ap.parse_args(argv)
|
||
if not getattr(a, "func", None):
|
||
ap.print_help()
|
||
return EXIT_USAGE
|
||
if not (REPO_ROOT / ".git").exists():
|
||
print("!! 无法定位仓库根（推导为 {}）".format(REPO_ROOT), file=sys.stderr)
|
||
return EXIT_ENV
|
||
try:
|
||
return a.func(a)
|
||
except EnvError as e:
|
||
print("前置/环境错误：{}".format(e), file=sys.stderr)
|
||
return EXIT_ENV
|
||
except Fail as e:
|
||
print("失败：{}".format(e), file=sys.stderr)
|
||
return EXIT_FAIL
|
||
except KeyboardInterrupt:
|
||
print("已中断", file=sys.stderr)
|
||
return EXIT_FAIL
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|