版本探测增加重试与数据降级,新增 --bundle 输出模式

patcher/analyze.py

@@ -4,13 +4,15 @@
 """
 import pathlib
 import platform
+import re
 import subprocess
+import time
 
-from . import defio, discovery, llm as llm_mod, ui
+from . import bytecode, defio, discovery, llm as llm_mod, ui
 
 
-def probe_version(binary: str):
-    """跑 binary --version 拿版本号;失败(如被 Gatekeeper 拦)返回 None。"""
+def _try_probe(binary: str):
+    """单次 binary --version 尝试,成功返回版本串,失败返回 None。"""
     try:
         r = subprocess.run(
             [binary, "--version"], capture_output=True, text=True, timeout=30
@@ -21,6 +23,60 @@ def probe_version(binary: str):
         return None
 
 
+def probe_version(binary: str):
+    """跑 binary --version 拿版本号;首次失败等 2 秒重试一次(macOS Gatekeeper
+    首次执行新路径的二进制可能需要额外的公证检查时间)。"""
+    v = _try_probe(binary)
+    if v:
+        return v
+    time.sleep(2)
+    return _try_probe(binary)
+
+
+def remove_quarantine(binary: str):
+    """去掉 macOS quarantine 属性,避免 Gatekeeper 拦截。"""
+    if platform.system() == "Darwin":
+        subprocess.run(["xattr", "-d", "com.apple.quarantine", binary],
+                       capture_output=True)
+
+
+_SEMVER_QUOTED_RE = re.compile(rb'"(\d{1,2}\.\d{1,2}\.\d{1,4})"')
+_SEMVER_RAW_RE = re.compile(rb'(\d{1,2}\.\d{1,2}\.\d{1,4})')
+
+
+def probe_version_from_data(data: bytes):
+    """从二进制数据中提取版本号(当 --version 探测失败时的降级方案)。
+
+    先扫描源码文本中带引号的 semver 模式;bytecode 构建上再验证常量池
+    唯一性,找不到时降级扫不带引号的模式(常量池条目是原始字节)。
+    """
+    candidates = set()
+    for m in _SEMVER_QUOTED_RE.finditer(data):
+        v = m.group(1).decode("ascii", errors="ignore")
+        if int(v.split(".")[0]) <= 19:
+            candidates.add(v)
+    is_bc = bytecode.is_bytecode_build(data)
+    if is_bc:
+        pool_hits = [v for v in candidates if bytecode.unique(data, v)]
+        if len(pool_hits) == 1:
+            return pool_hits[0]
+        extra = set()
+        for m in _SEMVER_RAW_RE.finditer(data):
+            v = m.group(1).decode("ascii", errors="ignore")
+            if v not in candidates and int(v.split(".")[0]) <= 19:
+                extra.add(v)
+        pool_hits = [v for v in extra if bytecode.unique(data, v)]
+        if len(pool_hits) == 1:
+            return pool_hits[0]
+        return None
+    if not candidates:
+        return None
+    shortest = min(candidates, key=len)
+    if len(shortest) <= 10:
+        return shortest
+    return None
+
+
 def safe_label(label: str) -> str:
     """版本标签用于文件名: 过滤 shell/glob 危险字符(如 probe 失败的 '?')。"""
     cleaned = "".join(ch if ch.isalnum() or ch in "._-" else "_" for ch in label)
@@ -88,9 +144,7 @@ def run(binary: str, cfg: dict, project_root: pathlib.Path,
     ui.info(f"{binary} " + ui.dim(f"({len(data):,} bytes)"))
 
     # cp/下载来的文件可能带 quarantine,先去掉,否则后面 --version 探测会被 Gatekeeper 杀
-    if platform.system() == "Darwin":
-        subprocess.run(["xattr", "-d", "com.apple.quarantine", binary],
-                       capture_output=True)
+    remove_quarantine(binary)
 
     display = cfg.get("display", {})
     raw_name = display.get("name", "")
@@ -132,6 +186,10 @@ def run(binary: str, cfg: dict, project_root: pathlib.Path,
 
     ui.step("发现锚点")
     probed = probe_version(binary)
+    if not probed:
+        probed = probe_version_from_data(data)
+        if probed:
+            ui.info(f"--version 探测失败,从二进制数据提取到版本: {probed}")
     patches, report, problems, warnings = discovery.build_patches(
         data, name, version_arg, design, theme_color=theme_color,
         real_version=probed,

patcher/cli.py

@@ -66,6 +66,15 @@ def _cmd_all(args, cfg):
     from . import pipeline
 
     binary = _resolve_binary(args)
+    if args.bundle:
+        if not args.apply:
+            from . import ui
+            ui.fail("--bundle 必须搭配 --apply 使用")
+            return 1
+        return pipeline.bundle(binary, cfg, PROJECT_ROOT,
+                               output_dir=args.output_dir, suffix=args.suffix,
+                               label=args.label, out=args.out, use_llm=args.llm,
+                               force=args.force, yes=args.yes, pty=args.pty)
     if not args.apply:
         rc = pipeline.run(binary, cfg, PROJECT_ROOT,
                           label=args.label, out=args.out, use_llm=args.llm,
@@ -84,7 +93,9 @@ def _cmd_skin(args, cfg):
     from . import interactive
 
     return interactive.run(getattr(args, "binary", None), cfg, PROJECT_ROOT,
-                           yes=args.yes, pty=args.pty)
+                           yes=args.yes, pty=args.pty,
+                           bundle=args.bundle, suffix=args.suffix,
+                           output_dir=args.output_dir)
 
 
 def _cmd_preview(args, cfg):
@@ -122,6 +133,10 @@ def main(argv=None):
     p.add_argument("binary", nargs="?", help="不填则自动定位系统安装")
     p.add_argument("-y", "--yes", action="store_true", help="跳过最终确认")
     p.add_argument("--pty", action="store_true", help="最后抓真实启动画面验证")
+    p.add_argument("--bundle", action="store_true",
+                   help="创建版本目录,放入原始 + 换肤二进制")
+    p.add_argument("--suffix", default="skin", help="换肤文件后缀(默认 skin)")
+    p.add_argument("--output-dir", help="bundle 输出目录(默认在源文件旁建 v版本号/)")
     p.set_defaults(func=_cmd_skin)
 
     p = sub.add_parser("analyze", help="发现锚点,生成补丁定义 TOML")
@@ -158,6 +173,10 @@ def main(argv=None):
                    help="覆盖已存在的定义文件(默认写 .draft.toml)")
     p.add_argument("--pty", action="store_true")
     p.add_argument("-y", "--yes", action="store_true", help="跳过写入前的交互确认")
+    p.add_argument("--bundle", action="store_true",
+                   help="创建版本目录,放入原始 + 换肤二进制")
+    p.add_argument("--suffix", default="skin", help="换肤文件后缀(默认 skin)")
+    p.add_argument("--output-dir", help="bundle 输出目录(默认在源文件旁建 v版本号/)")
     p.set_defaults(func=_cmd_all)
 
     p = sub.add_parser("preview", help="渲染图标设计稿预览")

patcher/interactive.py

@@ -100,7 +100,8 @@ def _ask_hex() -> str:
             ui.warn(str(e))
 
 
-def run(binary, cfg, project_root: pathlib.Path, yes=False, pty=False) -> int:
+def run(binary, cfg, project_root: pathlib.Path, yes=False, pty=False,
+        bundle=False, suffix="skin", output_dir=None) -> int:
     if not sys.stdin.isatty():
         ui.fail("skin 是交互向导,请在终端里运行;非交互请用 all --apply -y")
         return 1
@@ -173,8 +174,13 @@ def run(binary, cfg, project_root: pathlib.Path, yes=False, pty=False) -> int:
         "real_version": not version_s,
     }
 
-    rc = pipeline.run(path, new_cfg, project_root,
-                      label=version, yes=True, pty=pty)
+    if bundle:
+        rc = pipeline.bundle(path, new_cfg, project_root,
+                             output_dir=output_dir, suffix=suffix,
+                             label=version, yes=True, pty=pty)
+    else:
+        rc = pipeline.run(path, new_cfg, project_root,
+                          label=version, yes=True, pty=pty)
     if rc == 0:
         _save_config(project_root, new_cfg)
     return rc

patcher/pipeline.py

@@ -3,6 +3,7 @@
 cli 的 `all` 命令和交互向导 `skin` 都走这里,避免两份拷贝各自漂移。
 """
 import pathlib
+import shutil
 
 from . import analyze, patch as patch_mod, sign, ui, verify
 
@@ -12,6 +13,7 @@ def run(binary: str, cfg: dict, project_root: pathlib.Path, *,
         force: bool = False, yes: bool = False, pty: bool = False,
         apply: bool = True) -> int:
     """全流程执行;apply=False 时只做 analyze(dry-run)。返回退出码。"""
+    analyze.remove_quarantine(binary)
     label = analyze.safe_label(label or analyze.probe_version(binary) or "unknown")
     out = analyze.resolve_out(project_root, label, out=out, force=force)
     # out 已定好(含 .draft 防覆盖),force=True 让 analyze 直接写它
@@ -30,3 +32,50 @@ def run(binary: str, cfg: dict, project_root: pathlib.Path, *,
         ui.step("全部完成")
         ui.ok(f"可以运行了: {binary}")
     return rc
+
+
+def bundle(binary: str, cfg: dict, project_root: pathlib.Path, *,
+           output_dir: str = None, suffix: str = "skin", **kw) -> int:
+    """创建版本目录,放入原始 + 换肤二进制。
+
+    目录结构: v{版本号}/{原始文件名} + v{版本号}/{原始文件名}{suffix}
+    """
+    analyze.remove_quarantine(binary)
+    src = pathlib.Path(binary)
+    ver = analyze.probe_version(binary)
+    if not ver:
+        data = src.read_bytes()
+        ver = analyze.probe_version_from_data(data)
+    if not ver:
+        ui.fail("版本探测失败,无法确定目录名;请用 --label 显式指定")
+        return 1
+
+    bundle_dir = pathlib.Path(output_dir) if output_dir else src.parent / f"v{ver}"
+    if bundle_dir.exists():
+        ui.warn(f"目录已存在: {bundle_dir}")
+    bundle_dir.mkdir(parents=True, exist_ok=True)
+
+    orig_path = bundle_dir / src.name
+    skin_path = bundle_dir / (src.stem + suffix + src.suffix)
+    if not src.suffix:
+        skin_path = bundle_dir / (src.name + suffix)
+
+    ui.step("创建 bundle")
+    shutil.copy2(binary, orig_path)
+    ui.ok(f"原始文件: {orig_path}")
+    shutil.copy2(binary, skin_path)
+    ui.info(f"待换肤:   {skin_path}")
+
+    rc = run(str(skin_path), cfg, project_root, **kw)
+    if rc != 0:
+        return rc
+
+    backup = pathlib.Path(str(skin_path) + ".orig")
+    if backup.exists():
+        backup.unlink()
+
+    ui.step("bundle 就绪")
+    ui.ok(f"目录: {bundle_dir}")
+    ui.info(f"  原始: {orig_path.name}")
+    ui.info(f"  换肤: {skin_path.name}")
+    return 0

tests/test_core.py

@@ -7,14 +7,16 @@
 import os
 import struct
 import pathlib
+import shutil
 import sys
 import tempfile
 import unittest
+from unittest import mock
 
 sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent))
 
-from patcher import (bytecode, defio, discovery, glyphpack, locate,
-                     patch as patch_mod, tomlmini)
+from patcher import (analyze, bytecode, defio, discovery, glyphpack, locate,
+                     patch as patch_mod, pipeline, tomlmini)
 
 
 def esc(s: str) -> str:
@@ -634,5 +636,131 @@ def test_sniff(self):
             self.assertEqual(locate.sniff_kind(str(pathlib.Path(td) / "x")), "unknown")
 
 
+class TestProbeVersionRetry(unittest.TestCase):
+    def test_succeeds_on_second_attempt(self):
+        calls = []
+
+        def fake_probe(binary):
+            calls.append(1)
+            return None if len(calls) == 1 else "2.1.281"
+
+        with mock.patch("patcher.analyze._try_probe", side_effect=fake_probe), \
+             mock.patch("patcher.analyze.time.sleep"):
+            result = analyze.probe_version("/fake/binary")
+        self.assertEqual(result, "2.1.281")
+        self.assertEqual(len(calls), 2)
+
+    def test_returns_none_after_both_fail(self):
+        with mock.patch("patcher.analyze._try_probe", return_value=None), \
+             mock.patch("patcher.analyze.time.sleep"):
+            result = analyze.probe_version("/fake/binary")
+        self.assertIsNone(result)
+
+    def test_returns_immediately_on_first_success(self):
+        with mock.patch("patcher.analyze._try_probe", return_value="1.0.0"):
+            result = analyze.probe_version("/fake/binary")
+        self.assertEqual(result, "1.0.0")
+
+
+class TestProbeVersionFromData(unittest.TestCase):
+    def test_bytecode_build(self):
+        data = make_bytecode_bundle(version="2.1.281")
+        result = analyze.probe_version_from_data(data)
+        self.assertEqual(result, "2.1.281")
+
+    def test_non_bytecode_build(self):
+        data = b'"3.0.42"' + os.urandom(200)
+        result = analyze.probe_version_from_data(data)
+        self.assertEqual(result, "3.0.42")
+
+    def test_no_match(self):
+        data = os.urandom(500)
+        result = analyze.probe_version_from_data(data)
+        self.assertIsNone(result)
+
+    def test_rejects_high_major_version(self):
+        data = b'"99.1.0"' + os.urandom(200)
+        result = analyze.probe_version_from_data(data)
+        self.assertIsNone(result)
+
+
+class TestBundle(unittest.TestCase):
+    def _make_patchable_binary(self, td):
+        """创建一个可供 pipeline 打补丁的合成二进制(只需能通过 analyze)。"""
+        binary = pathlib.Path(td) / "claude"
+        data = make_bytecode_bundle(version="2.1.300")
+        binary.write_bytes(data)
+        binary.chmod(0o755)
+        return str(binary)
+
+    def test_bundle_creates_directory_structure(self):
+        with tempfile.TemporaryDirectory() as td:
+            binary = self._make_patchable_binary(td)
+            project = pathlib.Path(__file__).resolve().parent.parent
+            bundle_dir = pathlib.Path(td) / "v2.1.300"
+
+            with mock.patch("patcher.analyze.probe_version", return_value="2.1.300"), \
+                 mock.patch("patcher.pipeline.run", return_value=0):
+                rc = pipeline.bundle(binary, {}, project)
+
+            self.assertEqual(rc, 0)
+            self.assertTrue(bundle_dir.exists())
+            self.assertTrue((bundle_dir / "claude").exists())
+            self.assertTrue((bundle_dir / "claudeskin").exists())
+
+    def test_bundle_custom_suffix(self):
+        with tempfile.TemporaryDirectory() as td:
+            binary = self._make_patchable_binary(td)
+            project = pathlib.Path(__file__).resolve().parent.parent
+
+            with mock.patch("patcher.analyze.probe_version", return_value="2.1.300"), \
+                 mock.patch("patcher.pipeline.run", return_value=0):
+                rc = pipeline.bundle(binary, {}, project, suffix="mod")
+
+            bundle_dir = pathlib.Path(td) / "v2.1.300"
+            self.assertEqual(rc, 0)
+            self.assertTrue((bundle_dir / "claudemod").exists())
+
+    def test_bundle_custom_output_dir(self):
+        with tempfile.TemporaryDirectory() as td:
+            binary = self._make_patchable_binary(td)
+            project = pathlib.Path(__file__).resolve().parent.parent
+            custom = pathlib.Path(td) / "custom_out"
+
+            with mock.patch("patcher.analyze.probe_version", return_value="2.1.300"), \
+                 mock.patch("patcher.pipeline.run", return_value=0):
+                rc = pipeline.bundle(binary, {}, project, output_dir=str(custom))
+
+            self.assertEqual(rc, 0)
+            self.assertTrue(custom.exists())
+            self.assertTrue((custom / "claude").exists())
+            self.assertTrue((custom / "claudeskin").exists())
+
+    def test_bundle_fails_on_probe_failure(self):
+        with tempfile.TemporaryDirectory() as td:
+            binary = pathlib.Path(td) / "claude"
+            binary.write_bytes(os.urandom(500))
+            project = pathlib.Path(__file__).resolve().parent.parent
+
+            with mock.patch("patcher.analyze.probe_version", return_value=None), \
+                 mock.patch("patcher.analyze.probe_version_from_data", return_value=None):
+                rc = pipeline.bundle(str(binary), {}, project)
+
+            self.assertEqual(rc, 1)
+
+    def test_bundle_original_untouched(self):
+        with tempfile.TemporaryDirectory() as td:
+            binary = self._make_patchable_binary(td)
+            original_data = pathlib.Path(binary).read_bytes()
+            project = pathlib.Path(__file__).resolve().parent.parent
+            bundle_dir = pathlib.Path(td) / "v2.1.300"
+
+            with mock.patch("patcher.analyze.probe_version", return_value="2.1.300"), \
+                 mock.patch("patcher.pipeline.run", return_value=0):
+                pipeline.bundle(binary, {}, project)
+
+            self.assertEqual((bundle_dir / "claude").read_bytes(), original_data)
+
+
 if __name__ == "__main__":
     unittest.main()