# Install-Bensz-Skills 安装报告模板

本文档定义 `install-bensz-skills` 技能结束工作时的报告形式规范。

## 报告结构

```
============================================================
📦 正在安装到 {TARGET}: {target_root}
============================================================

【安装过程】
────────────────────────────────────────────────────────────
{process_messages}

【安装摘要】
────────────────────────────────────────────────────────────
┌────────────────────────┬──────────────┬─────────────────┐
│ Skill 名称              │ 状态         │ 原因            │
├────────────────────────┼──────────────┼─────────────────┤
│ {skill_name}            │ ✅ 已安装    │ 版本已更新...  │
│ {skill_name}            │ ⏭️  跳过     │ 版本未变化     │
└────────────────────────┴──────────────┴─────────────────┘

【辅助技能（已忽略，仅用于开发）】({count} 个)
   • {skill_name} ⏭️ 跳过
     原因: 辅助技能（开发用，不安装到生产环境）

【测试技能（已忽略，仅用于测试）】({count} 个)
   • {skill_name} ⏭️ 跳过
     原因: 测试技能（测试用，不安装到生产环境）

────────────────────────────────────────────────────────────
📊 统计
────────────────────────────────────────────────────────────
普通技能: {installed} 个已安装, {skipped} 个跳过
辅助技能: {auxiliary} 个已忽略（开发用，不安装）
测试技能: {test} 个已忽略（测试用，不安装）

============================================================
🎯 总体安装摘要
============================================================

总计数:
  • 已安装/更新: {unique_installed} 个
  • 跳过: {unique_skipped} 个
辅助技能: {auxiliary} 个已忽略（开发用）
测试技能: {test} 个已忽略（测试用）

{TARGET_UPPER}:
  新安装: {skill_list}
  未变化: {skill_list}

============================================================

📝 安装清单已保存: {relative_manifest_path}
💡 提示: 历史记录保存在 {manifest_parent_dir}/
```

## 报告格式规范

### 1. 技能个数统计（去重）

**核心原则**：统计的是**唯一技能个数**，而非安装次数。

- 如果安装了 3 个技能到 Codex 和 Claude Code，统计应显示 **3 个技能**，而非 6 个
- 使用 `set` 对技能名称去重，确保同一技能在多个平台安装时只计数一次

**实现示例**：

```python
# 正确：基于唯一技能名称统计
installed_skill_names = {s.name for r in reports for s in r.installed_skills}
total_installed = len(installed_skill_names)

# 错误：直接累加安装次数
total_installed = sum(len(r.installed_skills) for r in reports)
```

### 2. 分隔符规范

- **主要分隔线**：60 个 `=` 字符
- **次要分隔线**：60 个 `─` 字符
- **表格边框**：使用 Unicode 字符（`┌─┬┐│├─┼┤└─┴┘`）

### 3. 章节标题

固定章节标题（支持中英文）：

| 章节 | 中文 | 英文 |
|------|------|------|
| 安装过程 | 【安装过程】 | 【Installation Process】 |
| 安装摘要 | 【安装摘要】 | 【Installation Summary】 |
| 辅助技能 | 【辅助技能（已忽略，仅用于开发）】 | 【Auxiliary Skills (Ignored, Dev Only)】 |
| 测试技能 | 【测试技能（已忽略，仅用于测试）】 | 【Test Skills (Ignored, Test Only)】 |
| 统计 | 📊 统计 | 📊 Statistics |
| 总体安装摘要 | 🎯 总体安装摘要 | 🎯 Overall Installation Summary |

### 4. 状态图标

| 状态 | 图标 | 说明 |
|------|------|------|
| 已安装 | ✅ | 新安装或版本更新 |
| 跳过 | ⏭️  | 版本未变化，无需安装 |
| 已忽略 | ⏭️  | 辅助技能或测试技能，不安装 |

### 5. 技能类型说明

| 类型 | 说明 | 是否安装 |
|------|------|---------|
| **普通技能** (normal) | 可安装的技能，会被复制到目标目录 | ✅ 是 |
| **辅助技能** (auxiliary) | 开发辅助工具（如 install-bensz-skills 自身） | ❌ 否 |
| **测试技能** (test) | 测试用技能（如 v202601021343 时间戳目录） | ❌ 否 |

### 6. 消息格式

**操作消息**：
- `installed: {dest}` - 安装成功
- `removed: {dest}` - 删除旧版本
- `removed legacy symlink: {path}` - 移除旧软链接

**表格原因**：
- `版本已更新 (MD5: {md5[:12]})` - 版本变化，已安装
- `版本未变化` - 版本相同，跳过

**列表原因**：
- `辅助技能（开发用，不安装到生产环境）` - 辅助技能跳过原因
- `测试技能（测试用，不安装到生产环境）` - 测试技能跳过原因

### 7. 多语言支持

报告支持中英文切换，基于系统语言自动检测。

**检测优先级**：
1. `LC_ALL` 环境变量
2. `LANG` 环境变量
3. `locale.getdefaultlocale()` 结果
4. 默认英语

## 示例输出

### 中文示例

```
============================================================
📦 正在安装到 CLAUUDE: /Users/xxx/.claude/skills
============================================================

【安装过程】
────────────────────────────────────────────────────────────
installed: /Users/xxx/.claude/skills/nsfc-bib-manager

【安装摘要】
────────────────────────────────────────────────────────────
┌────────────────────────┬──────────────┬─────────────────┐
│ Skill 名称              │ 状态         │ 原因            │
├────────────────────────┼──────────────┼─────────────────┤
│ nsfc-bib-manager        │ ✅ 已安装    │ 版本已更新...  │
│ git-commit              │ ⏭️  跳过     │ 版本未变化     │
└────────────────────────┴──────────────┴─────────────────┘

【辅助技能（已忽略，仅用于开发）】(1 个)
   • install-bensz-skills ⏭️ 跳过

────────────────────────────────────────────────────────────
📊 统计
────────────────────────────────────────────────────────────
普通技能: 1 个已安装, 1 个跳过
辅助技能: 1 个已忽略（开发用，不安装）

============================================================
🎯 总体安装摘要
============================================================

总计数:
  • 已安装/更新: 1 个
  • 跳过: 1 个
辅助技能: 1 个已忽略（开发用）

CLAUUDE:
  新安装: nsfc-bib-manager
  未变化: git-commit

============================================================

📝 安装清单已保存: .bensz-skills/installation/manifests/install-manifest.20250121-120000.json
💡 提示: 历史记录保存在 .bensz-skills/installation/manifests/
```

## 版本历史

- **v5.0.0** (2025-01-21): 新增报告模板规范，修复技能个数统计逻辑（去重）

#!/usr/bin/env python3
"""Cross-platform installer for bensz skills.

This script is intentionally self-contained and uses only the Python standard
library. It downloads skills from GitHub zip archives, selectively extracts the
configured skills subtree when possible, installs changed skills to user-level
Codex and Claude Code directories, and records lightweight manifests for
traceability.
"""
from __future__ import annotations

import argparse
import fnmatch
import hashlib
import json
import os
import re
import shutil
import sys
import tempfile
import time
import urllib.error
import urllib.parse
import urllib.request
import zipfile
import subprocess
from dataclasses import dataclass
from pathlib import Path


class _UnicodeSafeTextStream:
    """Retry only Unicode encoding failures; propagate every other write error."""

    def __init__(self, stream):
        self._stream = stream

    def write(self, text):
        try:
            return self._stream.write(text)
        except UnicodeEncodeError:
            encoding = getattr(self._stream, "encoding", None) or "ascii"
            escaped = text.encode(encoding, errors="backslashreplace").decode(encoding)
            return self._stream.write(escaped)

    def writelines(self, lines):
        for line in lines:
            self.write(line)

    def __getattr__(self, name):
        return getattr(self._stream, name)


def _configure_console_streams() -> None:
    """Keep the host encoding while preventing localized messages from aborting the installer."""

    for stream_name in ("stdout", "stderr"):
        stream = getattr(sys, stream_name, None)
        if stream is None or isinstance(stream, _UnicodeSafeTextStream):
            continue

        reconfigure = getattr(stream, "reconfigure", None)
        if callable(reconfigure):
            try:
                reconfigure(errors="backslashreplace")
                continue
            except (AttributeError, TypeError, ValueError):
                pass

        setattr(sys, stream_name, _UnicodeSafeTextStream(stream))


_configure_console_streams()

MIN_PYTHON = (3, 8)
MANIFEST_SCHEMA_VERSION = 1
FALLBACK_CONFIG_VERSION = "0.7.0"
REMOTE_CONFIG_PATH = "skills/alpha/install-bensz-skills/config.yaml"
INSTALLATION_ROOT_PARTS = (".bensz-skills", "installation")
DOWNLOAD_RETRIES = 3
DOWNLOAD_RETRY_DELAY_SECONDS = 2

DEFAULT_SOURCES = [
    {
        "id": "general",
        "name": "General skills",
        "url": "https://github.com/huangwb8/skills",
        "branch": "main",
        "skills_path": "skills/alpha",
    },
    {
        "id": "research",
        "name": "Research skills",
        "url": "https://github.com/huangwb8/ChineseResearchLaTeX",
        "branch": "main",
        "skills_path": "skills",
    },
    {
        "id": "anthropic-docs",
        "name": "Anthropic document skills",
        "url": "https://github.com/anthropics/skills",
        "branch": "main",
        "skills_path": "skills",
    },
]

FALLBACK_LEGACY_SKILL_NAMES = [
    "make_latex_model",
    "transfer_old_latex_to_new",
    "write-paper-sci",
    "explain-figures",
    "complete_example",
    "write-skill-readme",
]

IGNORE_DIR_NAMES = {
    "__pycache__",
    ".pytest_cache",
    ".mypy_cache",
    ".ruff_cache",
    ".tox",
    ".nox",
    "test",
    "tests",
    "plans",
}
IGNORE_FILE_NAMES = {".DS_Store"}
IGNORE_ROOT_FILE_NAMES = {"readme.md", "changelog.md"}
IGNORE_GLOBS = ("*.pyc", "*.pyo")

MESSAGES = {
    "en": {
        "python_too_old": "Python {current} is too old. Please use Python {required} or newer.",
        "start": "Starting bensz skills installation...",
        "sources": "Selected sources: {sources}",
        "skills": "Selected skills: {skills}",
        "target": "Installing to {label}: {root}",
        "download": "Downloading {name} from {url}",
        "extract_selected": "Extracting only selected path(s): {paths}",
        "download_done": "Downloaded {name}",
        "download_failed": "Failed to download {name}: {error}",
        "extract_failed": "Failed to extract {name}: {error}",
        "skills_missing": "No skills root found for {name} at path: {path}",
        "no_skills": "No installable skills found in: {root}",
        "no_requested_skills": "No requested installable skills found in source: {name}",
        "skill_missing": "Requested skill(s) not found in selected sources: {skills}",
        "skill_not_installable": "Requested skill(s) are non-production and were not installed: {skills}",
        "legacy_config_loaded": "Loaded legacy cleanup list from {path} ({count} names)",
        "config_fallback": "Remote config unavailable; using versioned fallback {version}.",
        "removed_legacy": "Removed legacy skill: {path}",
        "skip_legacy_path": "Skipped legacy path that is not a symlink: {path}",
        "removed": "Removed existing skill: {path}",
        "installed": "Installed: {name}",
        "skipped": "Skipped unchanged: {name}",
        "ignored": "Ignored {count} non-production skill(s)",
        "manifest": "Manifest saved: {path}",
        "summary": "Summary: {installed} installed or updated, {skipped} unchanged, {ignored} ignored",
        "dry_run": "[dry-run] {message}",
        "unknown_source": "Unknown source id(s): {ids}",
        "cleanup": "Temporary files cleaned up.",
        "done": "Installation complete.",
    },
    "zh": {
        "python_too_old": "Python {current} 版本过低，请使用 Python {required} 或更新版本。",
        "start": "开始安装 bensz 技能...",
        "sources": "已选择源: {sources}",
        "skills": "已选择技能: {skills}",
        "target": "正在安装到 {label}: {root}",
        "download": "正在从 {url} 下载 {name}",
        "extract_selected": "仅解压选定路径: {paths}",
        "download_done": "下载完成: {name}",
        "download_failed": "下载失败: {name}: {error}",
        "extract_failed": "解压失败: {name}: {error}",
        "skills_missing": "未找到 {name} 的 skills 根目录: {path}",
        "no_skills": "未发现可安装技能: {root}",
        "no_requested_skills": "该源未发现请求的可安装技能: {name}",
        "skill_missing": "请求的技能未在所选源中找到: {skills}",
        "skill_not_installable": "请求的技能不是生产技能，未安装: {skills}",
        "legacy_config_loaded": "已从 {path} 读取 legacy 清理名单（{count} 个）",
        "config_fallback": "远程配置不可用，使用带版本标记的 fallback {version}。",
        "removed_legacy": "已移除 legacy 技能: {path}",
        "skip_legacy_path": "跳过非软链接 legacy 路径: {path}",
        "removed": "已删除旧技能: {path}",
        "installed": "已安装: {name}",
        "skipped": "跳过未变化: {name}",
        "ignored": "已忽略 {count} 个非生产技能",
        "manifest": "安装清单已保存: {path}",
        "summary": "统计: 已安装或更新 {installed} 个，未变化 {skipped} 个，已忽略 {ignored} 个",
        "dry_run": "[dry-run] {message}",
        "unknown_source": "未知源 ID: {ids}",
        "cleanup": "临时文件已清理。",
        "done": "安装完成。",
    },
}


@dataclass(frozen=True)
class Target:
    label: str
    root: Path
    legacy_link: Path


@dataclass(frozen=True)
class Skill:
    name: str
    src: Path
    md5: str


@dataclass
class InstallResult:
    installed: int = 0
    skipped: int = 0
    ignored: int = 0
    messages: list[str] | None = None
    installed_names: list[str] | None = None
    skipped_names: list[str] | None = None
    ignored_names: list[str] | None = None

    def __post_init__(self) -> None:
        if self.messages is None:
            self.messages = []
        if self.installed_names is None:
            self.installed_names = []
        if self.skipped_names is None:
            self.skipped_names = []
        if self.ignored_names is None:
            self.ignored_names = []


def tr(lang: str, key: str, **kwargs: object) -> str:
    return MESSAGES.get(lang, MESSAGES["en"])[key].format(**kwargs)


def ensure_python(lang: str) -> None:
    if sys.version_info < MIN_PYTHON:
        current = ".".join(str(part) for part in sys.version_info[:3])
        required = ".".join(str(part) for part in MIN_PYTHON)
        raise SystemExit(tr(lang, "python_too_old", current=current, required=required))


def stamp() -> str:
    return time.strftime("%Y%m%d-%H%M%S", time.localtime())


def installation_root() -> Path:
    return Path.home().joinpath(*INSTALLATION_ROOT_PARTS)


SILENT_UPDATE_TTL_SECONDS = 72 * 60 * 60


def _silent_state_path() -> Path:
    return installation_root() / "state" / "silent-update.json"


def _load_silent_state() -> dict:
    try:
        value = json.loads(_silent_state_path().read_text(encoding="utf-8"))
        return value if isinstance(value, dict) else {}
    except (OSError, ValueError, TypeError):
        return {}


def _installed_runtime_manager() -> Path | None:
    for suffix in (
        ".codex/skills/install-bensz-skills/scripts/managed_runtime.py",
        ".claude/skills/install-bensz-skills/scripts/managed_runtime.py",
    ):
        candidate = Path.home() / suffix
        if candidate.is_file():
            return candidate
    return None


def _managed_runtime_python() -> Path:
    prefix = Path.home() / ".bensz-skills" / "envs" / "benszapi"
    return prefix / ("python.exe" if os.name == "nt" else "bin/python")


def _run_managed_runtime(command: str, *, force: bool = False, dry_run: bool = False) -> int:
    manager = _installed_runtime_manager()
    if manager is None:
        print(json.dumps({"ready": False, "error": "managed runtime manager is not installed"}))
        return 1
    arguments = [sys.executable, str(manager), command]
    if force:
        arguments.append("--force-update")
    if dry_run:
        arguments.append("--dry-run")
    return subprocess.run(arguments, check=False).returncode


def _run_silent_update(lang: str) -> int:
    state = _load_silent_state()
    completed = state.get("last_check_completed_at")
    if state.get("schema_version", 0) not in (0, MANIFEST_SCHEMA_VERSION):
        return 0
    if isinstance(completed, (int, float)) and time.time() - completed < SILENT_UPDATE_TTL_SECONDS:
        return 0
    installed = []
    for root in (Path.home() / ".codex/skills", Path.home() / ".claude/skills"):
        try:
            installed.extend(child.name for child in root.iterdir() if child.is_dir() and (child / "SKILL.md").is_file())
        except OSError:
            continue
    installer = next((Path.home() / suffix for suffix in (".codex/skills/install-bensz-skills/scripts/install.py", ".claude/skills/install-bensz-skills/scripts/install.py") if (Path.home() / suffix).is_file()), None)
    if installer is not None:
        manager = installer.with_name("managed_runtime.py")
        if manager.is_file():
            subprocess.run(
                [sys.executable, str(manager), "ensure"],
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
                check=False,
            )
        managed_python = _managed_runtime_python()
        installer_python = managed_python if managed_python.is_file() else Path(sys.executable)
        try:
            supports_silent = "--silent-update" in subprocess.check_output(
                [str(installer_python), str(installer), "--help"], text=True,
                stderr=subprocess.STDOUT, timeout=10,
            )
        except (OSError, subprocess.SubprocessError):
            supports_silent = False
        if supports_silent:
            subprocess.run([str(installer_python), str(installer), "--silent-update"],
                           stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
                           check=False)
            return 0
    if installed:
        # Old installers do not understand --silent-update. Upgrade only the
        # installer from the production source, without expanding the set.
        return main(["--source", "general", "--skill", "install-bensz-skills", "--lang", lang])
    return 0


def path_label(path: Path) -> str:
    try:
        return str(path.relative_to(Path.home()))
    except ValueError:
        return str(path)


def print_msg(lang: str, key: str, dry_run: bool = False, **kwargs: object) -> None:
    message = tr(lang, key, **kwargs)
    if dry_run:
        message = tr(lang, "dry_run", message=message)
    print(message)


def should_ignore_rel_path(rel_path: Path) -> bool:
    if len(rel_path.parts) == 1 and rel_path.name.lower() in IGNORE_ROOT_FILE_NAMES:
        return True
    if any(part.startswith(".") for part in rel_path.parts):
        return True
    if any(part in IGNORE_DIR_NAMES for part in rel_path.parts):
        return True
    if rel_path.name in IGNORE_FILE_NAMES:
        return True
    if any(fnmatch.fnmatch(rel_path.name, pattern) for pattern in IGNORE_GLOBS):
        return True
    return False


def copytree_ignore(src_root: Path):
    def ignore(current_dir: str, names: list[str]) -> list[str]:
        current_path = Path(current_dir)
        ignored = []
        for name in names:
            try:
                rel_path = (current_path / name).relative_to(src_root)
            except ValueError:
                continue
            if should_ignore_rel_path(rel_path):
                ignored.append(name)
        return ignored

    return ignore


def calculate_md5(skill_dir: Path) -> str:
    hasher = hashlib.md5()
    for file_path in sorted(skill_dir.rglob("*")):
        if not file_path.is_file():
            continue
        rel_path = file_path.relative_to(skill_dir)
        if should_ignore_rel_path(rel_path):
            continue
        hasher.update(str(rel_path).encode("utf-8"))
        hasher.update(b"\0")
        hasher.update(file_path.read_bytes())
    return hasher.hexdigest()


def parse_skill_filter(raw_values: list[str] | None) -> list[str]:
    if not raw_values:
        return []

    skill_names: list[str] = []
    seen: set[str] = set()
    for raw_value in raw_values:
        for part in raw_value.split(","):
            skill_name = part.strip()
            if not skill_name or skill_name in seen:
                continue
            skill_names.append(skill_name)
            seen.add(skill_name)
    return skill_names


def installed_md5(dest_dir: Path, target: Target) -> str | None:
    manifest_file = dest_dir / f".skill-manifest.{target.label}.json"
    if manifest_file.exists():
        try:
            data = json.loads(manifest_file.read_text(encoding="utf-8"))
            value = data.get("md5")
            return str(value) if value else None
        except (OSError, json.JSONDecodeError):
            pass
    if dest_dir.exists():
        try:
            return calculate_md5(dest_dir)
        except OSError:
            return None
    return None


def save_skill_manifest(dest_dir: Path, md5: str, source: str, target: Target) -> None:
    manifest_file = dest_dir / f".skill-manifest.{target.label}.json"
    data = {
        "schema_version": MANIFEST_SCHEMA_VERSION,
        "md5": md5,
        "source": source,
        "installed_at": stamp(),
        "target": target.label,
        "target_root": str(target.root),
        "skills": [{
            "name": dest_dir.name,
            "md5": md5,
            "status": "installed",
            "reason": "",
        }],
    }
    manifest_file.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")


def parse_category(skill_dir: Path) -> str | None:
    skill_md = skill_dir / "SKILL.md"
    try:
        lines = skill_md.read_text(encoding="utf-8").splitlines()
    except OSError:
        return None
    if not lines or lines[0].strip() != "---":
        return None
    for line in lines[1:]:
        if line.strip() == "---":
            return None
        if line.startswith("category:"):
            return line.split(":", 1)[1].strip().strip('"').strip("'").lower()
    return None


def parse_legacy_skill_names_from_text(text: str) -> list[str]:
    lines = text.splitlines()
    names: list[str] = []
    seen: set[str] = set()
    in_legacy_section = False

    for raw_line in lines:
        line_without_comment = raw_line.split("#", 1)[0].rstrip()
        stripped = line_without_comment.strip()
        if not stripped:
            continue

        if not in_legacy_section:
            if stripped == "legacy_skill_names:":
                in_legacy_section = True
            continue

        if not raw_line.startswith((" ", "\t")) and re.match(r"^[A-Za-z0-9_-]+:", stripped):
            break
        if not stripped.startswith("-"):
            continue

        name = stripped[1:].strip().strip('"').strip("'")
        if name and name not in seen:
            names.append(name)
            seen.add(name)

    return names


def parse_legacy_skill_names(config_path: Path) -> list[str]:
    try:
        return parse_legacy_skill_names_from_text(config_path.read_text(encoding="utf-8"))
    except OSError:
        return []


def load_legacy_skill_names_from_source(skills_root: Path) -> tuple[list[str], Path] | None:
    candidate_paths = [
        skills_root / "install-bensz-skills" / "config.yaml",
        skills_root / "config.yaml",
    ]
    for config_path in candidate_paths:
        names = parse_legacy_skill_names(config_path)
        if names:
            return names, config_path
    return None


def skill_type(skill_dir: Path, skills_root: Path) -> str:
    category = parse_category(skill_dir)
    if category in {"auxiliary", "dev", "development"}:
        return "auxiliary"
    if category in {"test", "testing"}:
        return "test"
    if category in {"normal", "production"}:
        return "normal"

    rel_path = skill_dir.relative_to(skills_root)
    if any(part in {"test", "tests"} for part in rel_path.parts):
        return "test"
    name = skill_dir.name.lower()
    if re.match(r"^\d{8}_\d{6}$", name):
        return "test"
    if any(pattern in name for pattern in ("test-", "-test", "_test", "test_")):
        return "test"
    return "normal"


def looks_like_skills_root(root: Path) -> bool:
    try:
        if not root.exists() or not root.is_dir():
            return False
        return any(child.is_dir() and (child / "SKILL.md").is_file() for child in root.iterdir())
    except OSError:
        return False


def resolve_skills_root(repo_root: Path, skills_path: str) -> Path | None:
    raw_path = skills_path.strip()
    requested = repo_root if raw_path in {"", "."} else repo_root / raw_path
    if looks_like_skills_root(requested):
        return requested
    if not raw_path and requested != repo_root and looks_like_skills_root(repo_root):
        return repo_root
    return None


def normalize_archive_path(path: str | None) -> str | None:
    raw_path = (path or "").strip().replace("\\", "/")
    while raw_path.startswith("./"):
        raw_path = raw_path[2:]
    raw_path = raw_path.strip("/")

    if raw_path in {"", "."}:
        return None

    parts = [part for part in raw_path.split("/") if part]
    if any(part == ".." for part in parts):
        return None
    return "/".join(parts)


def build_archive_include_paths(
    skills_path: str,
    skill_names: list[str] | None,
) -> list[str] | None:
    root_path = normalize_archive_path(skills_path)
    if not skill_names:
        return [root_path] if root_path else None

    include_paths: list[str] = []
    for skill_name in skill_names:
        safe_skill_name = normalize_archive_path(skill_name)
        if not safe_skill_name or "/" in safe_skill_name:
            continue
        include_paths.append(
            f"{root_path}/{safe_skill_name}" if root_path else safe_skill_name
        )

    if root_path is None:
        include_paths.append("install-bensz-skills/config.yaml")
    return include_paths or None


def discover_skills(skills_root: Path) -> tuple[list[Skill], int, set[str]]:
    normal: list[Skill] = []
    ignored = 0
    ignored_names: set[str] = set()
    for skill_dir in sorted(skills_root.iterdir()):
        if skill_dir.name.startswith(".") or not skill_dir.is_dir():
            continue
        if not (skill_dir / "SKILL.md").is_file():
            continue
        if skill_type(skill_dir, skills_root) != "normal":
            ignored += 1
            ignored_names.add(skill_dir.name)
            continue
        normal.append(Skill(name=skill_dir.name, src=skill_dir, md5=calculate_md5(skill_dir)))

    by_name: dict[str, list[Path]] = {}
    for skill in normal:
        by_name.setdefault(skill.name, []).append(skill.src)
    collisions = {name: paths for name, paths in by_name.items() if len(paths) > 1}
    if collisions:
        details = "; ".join(f"{name}: {', '.join(str(p) for p in paths)}" for name, paths in collisions.items())
        raise RuntimeError(f"Duplicate skill directory names: {details}")
    return normal, ignored, ignored_names


def github_archive_url(repo_url: str, branch: str) -> str:
    parsed = urllib.parse.urlparse(repo_url)
    parts = [part for part in parsed.path.strip("/").split("/") if part]
    if parsed.netloc.lower() not in {"github.com", "www.github.com"} or len(parts) < 2:
        raise ValueError(f"Only GitHub repository URLs are supported: {repo_url}")
    owner = parts[0]
    repo = parts[1]
    if repo.endswith(".git"):
        repo = repo[:-4]
    branch_ref = urllib.parse.quote(branch, safe="/")
    return f"https://github.com/{owner}/{repo}/archive/refs/heads/{branch_ref}.zip"


def github_raw_file_url(repo_url: str, branch: str, file_path: str) -> str:
    parsed = urllib.parse.urlparse(repo_url)
    parts = [part for part in parsed.path.strip("/").split("/") if part]
    if parsed.netloc.lower() not in {"github.com", "www.github.com"} or len(parts) < 2:
        raise ValueError(f"Only GitHub repository URLs are supported: {repo_url}")
    owner = parts[0]
    repo = parts[1]
    if repo.endswith(".git"):
        repo = repo[:-4]
    branch_ref = urllib.parse.quote(branch, safe="/")
    normalized_path = "/".join(part for part in file_path.split("/") if part)
    return f"https://raw.githubusercontent.com/{owner}/{repo}/{branch_ref}/{normalized_path}"


def summarize_download_error(exc: BaseException) -> str:
    if isinstance(exc, urllib.error.HTTPError):
        return f"HTTP {exc.code}: {exc.reason}"
    if isinstance(exc, urllib.error.URLError):
        return str(exc.reason)
    return str(exc)


def download_bytes(url: str, timeout: int) -> bytes:
    request = urllib.request.Request(url, headers={"User-Agent": "bensz-skills-installer"})
    last_error: BaseException | None = None
    for attempt in range(1, DOWNLOAD_RETRIES + 1):
        try:
            with urllib.request.urlopen(request, timeout=timeout) as response:
                return response.read()
        except (OSError, urllib.error.URLError) as exc:
            last_error = exc
            if attempt >= DOWNLOAD_RETRIES:
                break
            time.sleep(DOWNLOAD_RETRY_DELAY_SECONDS * attempt)

    assert last_error is not None
    raise RuntimeError(summarize_download_error(last_error)) from last_error


def download_file(url: str, dest: Path) -> None:
    temp_dest = dest.with_name(f"{dest.name}.part")
    request = urllib.request.Request(url, headers={"User-Agent": "bensz-skills-installer"})
    last_error: BaseException | None = None
    for attempt in range(1, DOWNLOAD_RETRIES + 1):
        try:
            with urllib.request.urlopen(request, timeout=120) as response:
                with temp_dest.open("wb") as out:
                    shutil.copyfileobj(response, out)
            temp_dest.replace(dest)
            return
        except (OSError, urllib.error.URLError) as exc:
            last_error = exc
            if temp_dest.exists():
                temp_dest.unlink()
            if attempt >= DOWNLOAD_RETRIES:
                break
            time.sleep(DOWNLOAD_RETRY_DELAY_SECONDS * attempt)

    assert last_error is not None
    raise RuntimeError(summarize_download_error(last_error)) from last_error


def download_text(url: str) -> str:
    return download_bytes(url, timeout=30).decode("utf-8")


def load_legacy_skill_names_from_remote_config() -> tuple[list[str], str] | None:
    general_source = next((source for source in DEFAULT_SOURCES if source["id"] == "general"), None)
    if general_source is None:
        return None

    try:
        url = github_raw_file_url(
            general_source["url"],
            general_source["branch"],
            "skills/alpha/install-bensz-skills/config.yaml",
        )
        names = parse_legacy_skill_names_from_text(download_text(url))
    except (OSError, RuntimeError, UnicodeDecodeError, urllib.error.URLError, ValueError):
        return None

    if not names:
        return None
    return names, url


def parse_remote_sources_from_text(text: str) -> list[dict[str, str]]:
    """Parse the small public source contract without requiring PyYAML.

    The bootstrap intentionally accepts only scalar source fields.  Unknown or
    malformed entries are ignored so a partially edited remote config cannot
    replace the safe fallback list with an unusable source.
    """
    sources: list[dict[str, str]] = []
    current: dict[str, str] | None = None
    in_section = False
    allowed = {"id", "name", "url", "branch", "skills_path"}
    for raw_line in text.splitlines():
        stripped = raw_line.split("#", 1)[0].strip()
        if not stripped:
            continue
        if stripped == "remote_sources:":
            in_section = True
            continue
        if not in_section:
            continue
        if not raw_line.startswith((" ", "\t")):
            if stripped.endswith(":"):
                break
            continue
        if stripped.startswith("- "):
            if current and {"id", "url", "branch", "skills_path"} <= current.keys():
                sources.append(current)
            current = {}
            stripped = stripped[2:].strip()
        if ":" not in stripped or current is None:
            continue
        key, value = stripped.split(":", 1)
        key = key.strip()
        if key not in allowed:
            continue
        value = value.strip().strip('"').strip("'")
        if value:
            current[key] = value
    if current and {"id", "url", "branch", "skills_path"} <= current.keys():
        sources.append(current)
    return sources


def load_remote_config_contract() -> tuple[list[dict[str, str]], list[str], str] | None:
    """Load sources and legacy names from the canonical remote config.

    Returns ``None`` on any network or parsing failure; callers then retain the
    versioned fallback and expose its provenance in the run manifest.
    """
    general_source = next((source for source in DEFAULT_SOURCES if source["id"] == "general"), None)
    if general_source is None:
        return None
    try:
        url = github_raw_file_url(
            general_source["url"], general_source["branch"], REMOTE_CONFIG_PATH
        )
        text = download_text(url)
        sources = parse_remote_sources_from_text(text)
        names = parse_legacy_skill_names_from_text(text)
    except (OSError, RuntimeError, UnicodeDecodeError, urllib.error.URLError, ValueError):
        return None
    if not sources:
        return None
    return sources, names, url


def archive_member_relative_path(member_name: str) -> Path | None:
    parts = Path(member_name).parts
    if len(parts) <= 1:
        return None
    return Path(*parts[1:])


def should_extract_archive_member(member_name: str, include_paths: list[str] | None) -> bool:
    if include_paths is None:
        return True

    rel_path = archive_member_relative_path(member_name)
    if rel_path is None:
        return True

    normalized_member = rel_path.as_posix().strip("/")
    for include_path in include_paths:
        normalized_include = include_path.strip("/")
        if (
            normalized_member == normalized_include
            or normalized_member.startswith(f"{normalized_include}/")
        ):
            return True
    return False


def safe_extract_zip(
    zip_path: Path,
    dest_dir: Path,
    include_paths: list[str] | None = None,
) -> Path:
    dest_resolved = dest_dir.resolve()
    with zipfile.ZipFile(zip_path) as archive:
        members = archive.infolist()
        for member in members:
            if not should_extract_archive_member(member.filename, include_paths):
                continue
            target = (dest_dir / member.filename).resolve()
            if target != dest_resolved and dest_resolved not in target.parents:
                raise RuntimeError(f"Unsafe zip path: {member.filename}")
        for member in members:
            if should_extract_archive_member(member.filename, include_paths):
                archive.extract(member, dest_dir)

    top_dirs = [child for child in dest_dir.iterdir() if child.is_dir()]
    if len(top_dirs) == 1:
        return top_dirs[0]
    return dest_dir


def download_source(
    source: dict[str, str],
    temp_root: Path,
    lang: str,
    skill_names: list[str] | None = None,
) -> tuple[Path | None, bool]:
    name = source["name"]
    try:
        url = github_archive_url(source["url"], source["branch"])
    except ValueError as exc:
        print_msg(lang, "download_failed", name=name, error=exc)
        return None, True
    print_msg(lang, "download", name=name, url=url)
    source_dir = temp_root / source["id"]
    source_dir.mkdir(parents=True, exist_ok=True)
    zip_path = source_dir / "source.zip"
    include_paths = build_archive_include_paths(source["skills_path"], skill_names)
    try:
        download_file(url, zip_path)
        if include_paths:
            print_msg(lang, "extract_selected", paths=", ".join(include_paths))
        repo_root = safe_extract_zip(zip_path, source_dir / "extracted", include_paths)
    except (OSError, urllib.error.URLError, zipfile.BadZipFile, RuntimeError) as exc:
        print_msg(lang, "download_failed", name=name, error=exc)
        return None, True
    print_msg(lang, "download_done", name=name)
    skills_root = resolve_skills_root(repo_root, source["skills_path"])
    if skills_root is None and skill_names:
        normalized_skills_path = normalize_archive_path(source["skills_path"])
        requested_root = repo_root if normalized_skills_path is None else repo_root / normalized_skills_path
        requested_root.mkdir(parents=True, exist_ok=True)
        skills_root = requested_root
    return skills_root, False


def remove_path(path: Path) -> None:
    if path.is_symlink() or path.is_file():
        path.unlink()
    else:
        shutil.rmtree(path)


def remove_existing(dest: Path, lang: str, dry_run: bool, result: InstallResult) -> None:
    if not dest.exists() and not dest.is_symlink():
        return
    print_msg(lang, "removed", dry_run=dry_run, path=dest)
    result.messages.append(f"remove existing: {dest}")
    if not dry_run:
        remove_path(dest)


def remove_legacy_skills(
    target: Target,
    active_names: set[str],
    legacy_skill_names: list[str],
    lang: str,
    dry_run: bool,
    result: InstallResult,
) -> None:
    legacy_link = target.legacy_link
    if legacy_link.exists() or legacy_link.is_symlink():
        if legacy_link.is_symlink():
            print_msg(lang, "removed_legacy", dry_run=dry_run, path=legacy_link)
            result.messages.append(f"remove legacy symlink: {legacy_link}")
            if not dry_run:
                legacy_link.unlink()
        else:
            print_msg(lang, "skip_legacy_path", path=legacy_link)

    for name in legacy_skill_names:
        if name in active_names:
            continue
        legacy_path = target.root / name
        if not legacy_path.exists() and not legacy_path.is_symlink():
            continue
        print_msg(lang, "removed_legacy", dry_run=dry_run, path=legacy_path)
        result.messages.append(f"remove legacy skill: {legacy_path}")
        if not dry_run:
            remove_path(legacy_path)


def install_skills(
    skills: list[Skill],
    ignored_count: int,
    target: Target,
    source_label: str,
    legacy_skill_names: list[str],
    force: bool,
    dry_run: bool,
    lang: str,
    ignored_names: set[str] | None = None,
) -> InstallResult:
    result = InstallResult(ignored=ignored_count)
    result.ignored_names.extend(sorted(ignored_names or set()))
    active_names = {skill.name for skill in skills}
    remove_legacy_skills(target, active_names, legacy_skill_names, lang, dry_run, result)

    if not dry_run:
        target.root.mkdir(parents=True, exist_ok=True)

    for skill in skills:
        dest = target.root / skill.name
        current_md5 = None if force else installed_md5(dest, target)
        if current_md5 == skill.md5:
            result.skipped += 1
            result.skipped_names.append(skill.name)
            print_msg(lang, "skipped", name=skill.name)
            continue

        remove_existing(dest, lang, dry_run, result)
        print_msg(lang, "installed", dry_run=dry_run, name=skill.name)
        result.messages.append(f"install: {skill.src} -> {dest}")
        if not dry_run:
            shutil.copytree(
                skill.src,
                dest,
                symlinks=False,
                dirs_exist_ok=False,
                ignore=copytree_ignore(skill.src),
            )
            save_skill_manifest(dest, skill.md5, source_label, target)
        result.installed += 1
        result.installed_names.append(skill.name)

    return result


def manifest_dir() -> Path:
    path = installation_root() / "manifests"
    path.mkdir(parents=True, exist_ok=True)
    return path


def save_run_manifest(records: list[dict[str, object]], dry_run: bool, lang: str) -> None:
    if dry_run:
        print(json.dumps({"runs": records}, ensure_ascii=False, indent=2))
        return
    path = manifest_dir() / f"install-manifest.{stamp()}.json"
    path.write_text(json.dumps({"runs": records}, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    print_msg(lang, "manifest", path=path_label(path))


def select_sources(
    source_ids: str | None,
    available_sources: list[dict[str, str]] | None = None,
) -> list[dict[str, str]]:
    sources = available_sources or DEFAULT_SOURCES
    if not source_ids:
        return sources
    wanted = {item.strip() for item in source_ids.split(",") if item.strip()}
    known = {source["id"] for source in sources}
    unknown = sorted(wanted - known)
    if unknown:
        raise ValueError(",".join(unknown))
    return [source for source in sources if source["id"] in wanted]


def build_targets(args: argparse.Namespace) -> list[Target]:
    install_codex = args.codex or (not args.codex and not args.claude)
    install_claude = args.claude or (not args.codex and not args.claude)
    home = Path.home()
    targets = []
    if install_codex:
        targets.append(Target("codex", home / ".codex" / "skills", home / ".codex" / "skills" / "pipeline-skills"))
    if install_claude:
        targets.append(Target("claude", home / ".claude" / "skills", home / ".claude" / "skills" / "pipeline-skills"))
    return targets


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Install bensz skills using only the Python standard library.",
    )
    parser.add_argument("--codex", action="store_true", help="Install to Codex only.")
    parser.add_argument("--claude", action="store_true", help="Install to Claude Code only.")
    parser.add_argument("--force", action="store_true", help="Reinstall even when MD5 is unchanged.")
    parser.add_argument("--dry-run", action="store_true", help="Print actions without writing files.")
    parser.add_argument("--check", action="store_true", help="Alias for --dry-run.")
    parser.add_argument("--source", help="Comma-separated source ids. Available: general,research,anthropic-docs.")
    parser.add_argument("--skill", action="append", default=[], help="Install only selected skill names. Repeat or comma-separate values.")
    parser.add_argument("--lang", choices=["en", "zh"], default="en", help="Installer language. Default: en.")
    parser.add_argument("--silent-update", action="store_true", help="Refresh installed production skills after the 72-hour TTL.")
    parser.add_argument("--ensure-runtime", action="store_true", help="Create or update the managed benszapi Conda runtime.")
    parser.add_argument("--runtime-status", action="store_true", help="Inspect the managed benszapi runtime without changing it.")
    parser.add_argument("--force-runtime-update", action="store_true", help="Ignore the runtime update TTL.")
    return parser


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    lang = args.lang
    ensure_python(lang)
    if args.runtime_status:
        if any((args.ensure_runtime, args.force_runtime_update, args.silent_update, args.codex,
                args.claude, args.force, args.dry_run, args.check, args.source, args.skill)):
            parser.error("--runtime-status cannot be combined with other operations")
        return _run_managed_runtime("status")
    if args.ensure_runtime or args.force_runtime_update:
        if any((args.silent_update, args.codex, args.claude, args.force, args.check, args.source, args.skill)):
            parser.error("managed runtime options cannot be combined with skill installation options")
        if _installed_runtime_manager() is None:
            if args.dry_run:
                print(json.dumps({
                    "ready": False,
                    "dry_run": True,
                    "would_install": "install-bensz-skills and managed benszapi runtime",
                }))
                return 0
            install_code = main(["--source", "general", "--skill", "install-bensz-skills", "--lang", lang])
            if install_code != 0:
                return install_code
        return _run_managed_runtime(
            "ensure",
            force=args.force_runtime_update,
            dry_run=args.dry_run,
        )
    if args.silent_update:
        if any((args.codex, args.claude, args.force, args.dry_run, args.check, args.source, args.skill,
                args.ensure_runtime, args.runtime_status, args.force_runtime_update)):
            parser.error("--silent-update cannot be combined with install or filter options")
        return _run_silent_update(lang)
    dry_run = bool(args.dry_run or args.check)
    selected_skill_names = parse_skill_filter(args.skill)
    selected_skill_set = set(selected_skill_names)
    matched_installable_names: set[str] = set()
    matched_non_installable_names: set[str] = set()

    config_provenance = f"fallback:{FALLBACK_CONFIG_VERSION}"
    configured_sources = list(DEFAULT_SOURCES)
    legacy_skill_names = list(FALLBACK_LEGACY_SKILL_NAMES)
    remote_contract = load_remote_config_contract()
    if remote_contract is not None:
        configured_sources, legacy_skill_names, config_provenance = remote_contract
    else:
        print_msg(lang, "config_fallback", version=FALLBACK_CONFIG_VERSION)
    try:
        sources = select_sources(args.source, configured_sources)
    except ValueError as exc:
        print_msg(lang, "unknown_source", ids=exc)
        return 1

    targets = build_targets(args)
    print_msg(lang, "start")
    print_msg(lang, "sources", sources=", ".join(source["id"] for source in sources))
    if selected_skill_names:
        print_msg(lang, "skills", skills=", ".join(selected_skill_names))

    total_installed = 0
    total_skipped = 0
    total_ignored = 0
    source_errors = 0
    legacy_config_loaded = remote_contract is not None
    records: list[dict[str, object]] = []

    remote_legacy_names = (
        None
        if remote_contract is not None
        else load_legacy_skill_names_from_remote_config()
    )
    if remote_legacy_names is not None:
        legacy_skill_names, legacy_config_url = remote_legacy_names
        config_provenance = legacy_config_url
        legacy_config_loaded = True
        print_msg(
            lang,
            "legacy_config_loaded",
            path=legacy_config_url,
            count=len(legacy_skill_names),
        )

    with tempfile.TemporaryDirectory(prefix="bensz-skills-install-") as tmp:
        temp_root = Path(tmp)
        for source in sources:
            skills_root, download_failed = download_source(
                source,
                temp_root,
                lang,
                selected_skill_names,
            )

            if skills_root is None:
                source_errors += 1
                if not download_failed:
                    print_msg(lang, "skills_missing", name=source["name"], path=source["skills_path"])
                continue

            loaded_legacy_names = load_legacy_skill_names_from_source(skills_root)
            if loaded_legacy_names is not None and not legacy_config_loaded:
                legacy_skill_names, legacy_config_path = loaded_legacy_names
                legacy_config_loaded = True
                print_msg(
                    lang,
                    "legacy_config_loaded",
                    path=path_label(legacy_config_path),
                    count=len(legacy_skill_names),
                )

            skills, ignored, ignored_names = discover_skills(skills_root)
            selected_ignored_names = set(ignored_names)
            if selected_skill_names:
                matched_installable_names.update(skill.name for skill in skills if skill.name in selected_skill_set)
                matched_non_installable_names.update(ignored_names & selected_skill_set)
                skills = [skill for skill in skills if skill.name in selected_skill_set]
                selected_ignored_names &= selected_skill_set
                ignored = len(selected_ignored_names)

            total_ignored += ignored
            if not skills:
                if selected_skill_names:
                    print_msg(lang, "no_requested_skills", name=source["name"])
                else:
                    print_msg(lang, "no_skills", root=skills_root)
                continue

            source_label = f"{source['id']} ({source['url']}@{source['branch']}:{source['skills_path']})"
            for target in targets:
                print()
                print_msg(lang, "target", label=target.label, root=target.root)
                result = install_skills(
                    skills=skills,
                    ignored_count=ignored,
                    target=target,
                    source_label=source_label,
                    legacy_skill_names=legacy_skill_names,
                    force=args.force,
                    dry_run=dry_run,
                    lang=lang,
                    ignored_names=selected_ignored_names,
                )
                total_installed += result.installed
                total_skipped += result.skipped
                if result.ignored:
                    print_msg(lang, "ignored", count=result.ignored)
                records.append(
                    {
                        "schema_version": MANIFEST_SCHEMA_VERSION,
                        "source": source_label,
                        "config_source": config_provenance,
                        "target": target.label,
                        "target_root": str(target.root),
                        "installed_count": result.installed,
                        "skipped_count": result.skipped,
                        "ignored_count": result.ignored,
                        "skills": [
                            {
                                "name": skill.name,
                                "md5": skill.md5,
                                "status": "installed" if skill.name in result.installed_names else "skipped",
                                "reason": "" if skill.name in result.installed_names else "unchanged",
                            }
                            for skill in skills
                        ] + [
                            {
                                "name": name,
                                "md5": calculate_md5(skills_root / name),
                                "status": "ignored",
                                "reason": "non-production",
                            }
                            for name in result.ignored_names
                        ],
                        "messages": result.messages,
                    }
                )

    print()
    save_run_manifest(records, dry_run, lang)
    print_msg(lang, "cleanup")
    print_msg(lang, "summary", installed=total_installed, skipped=total_skipped, ignored=total_ignored)

    exit_code = 1 if source_errors else 0
    if selected_skill_names:
        missing_names = [
            name for name in selected_skill_names
            if name not in matched_installable_names and name not in matched_non_installable_names
        ]
        non_installable_names = [
            name for name in selected_skill_names
            if name in matched_non_installable_names and name not in matched_installable_names
        ]
        if missing_names:
            print_msg(lang, "skill_missing", skills=", ".join(missing_names))
            exit_code = 1
        if non_installable_names:
            print_msg(lang, "skill_not_installable", skills=", ".join(non_installable_names))
            exit_code = 1

    print_msg(lang, "done")
    return exit_code


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Internationalization (i18n) module for install-bensz-skills.

自动检测用户系统语言并返回相应的本地化消息。
"""
from __future__ import annotations

import locale
import os
from dataclasses import dataclass
from typing import Callable


# 支持的语言列表
SUPPORTED_LANGUAGES = ["en", "zh"]


@dataclass(frozen=True)
class Messages:
    """本地化消息集合。"""

    # 通用消息
    arg_help_description: str
    arg_help_dry_run: str
    arg_help_codex: str
    arg_help_claude: str
    arg_help_force: str
    arg_help_skill: str
    arg_help_remote: str
    arg_help_check: str
    arg_help_auto: str
    arg_help_source_filter: str

    # 错误消息
    error_no_skills_found: str
    error_skill_filter_missing: str
    error_skill_filter_not_installable: str
    error_skill_name_collision: str
    error_source_root_not_found: str

    # 安装过程消息
    installing_to_target: str
    removed_legacy_symlink: str
    removed_legacy_skill: str
    skip_legacy_path: str
    removed_existing: str
    installed: str
    dry_run_prefix: str

    # 表格相关消息
    table_header_skill: str
    table_header_status: str
    table_header_reason: str
    table_status_installed: str
    table_status_skipped: str
    table_reason_no_change: str
    table_reason_updated: str
    table_separator: str

    # 安装报告消息（固定格式）
    report_section_process: str
    report_section_summary: str
    report_section_ignored: str
    report_no_actions: str
    report_statistics: str

    # 安装摘要消息（兼容保留）
    summary_header: str
    summary_installed: str
    summary_skipped: str
    summary_reason: str
    summary_total_header: str
    summary_total_counts: str
    summary_installed_count: str
    summary_skipped_count: str
    summary_new_install: str
    summary_unchanged: str
    summary_manifest_saved: str

    # 其他消息
    manifest_preview: str

    # 远程安装消息
    remote_check_intro: str
    remote_auto_intro: str
    remote_source_prompt: str
    remote_source_prompt_recommended: str
    remote_download_progress: str
    remote_download_complete: str
    remote_download_failed: str
    remote_skills_path_missing: str
    remote_compare_header: str
    remote_compare_new: str
    remote_compare_updated: str
    remote_compare_unchanged: str
    remote_confirm_install: str
    remote_confirm_install_detail: str
    remote_temp_dir: str
    remote_cleanup_complete: str
    remote_no_updates: str
    remote_user_cancelled: str
    remote_git_not_found: str
    remote_config_not_found: str
    remote_config_error: str
    remote_source_filter_selected: str
    remote_source_filter_invalid: str
    remote_source_update_failed: str
    remote_source_stale_cache: str


# 英文消息
MESSAGES_EN = Messages(
    arg_help_description="Install all skills from this repo to Codex/Claude Code user-level skills directories (copy-based, with MD5 versioning).",
    arg_help_dry_run="Print actions without writing anything.",
    arg_help_codex="Install to Codex only (default: both Codex and Claude Code).",
    arg_help_claude="Install to Claude Code only (default: both Codex and Claude Code).",
    arg_help_force="Force re-install all skills, ignoring MD5 check.",
    arg_help_skill="Install/update only the named skill. Can be repeated or comma-separated.",
    arg_help_remote="Enable remote installation mode (download skills from GitHub).",
    arg_help_check="Check mode (interactive confirmation before installing).",
    arg_help_auto="Auto mode (force install without confirmation).",
    arg_help_source_filter="Filter remote sources by ID (e.g., --general, --research).",
    error_no_skills_found="No installable skills found (scanned root: {root})",
    error_skill_filter_missing="Requested skill(s) not found: {skills}",
    error_skill_filter_not_installable="Requested skill(s) are not installable normal skills: {skills}",
    error_skill_name_collision="Detected skill directory name conflicts (basename duplicated), cannot install safely:",
    error_source_root_not_found="Failed to auto-detect skills source root. Run this script inside a project containing ./skills/alpha (the installer checks the current directory and its parents), or pass --source explicitly (use ./skills/beta only when beta is intentional).",
    installing_to_target="Installing to {TARGET}: {root}",
    removed_legacy_symlink="removed legacy symlink: {path}",
    removed_legacy_skill="removed legacy skill: {path}",
    skip_legacy_path="skip legacy path (not a symlink): {path}",
    removed_existing="removed: {dest}",
    installed="installed: {dest}",
    dry_run_prefix="[dry-run] ",
    # 表格相关
    table_header_skill="Skill Name",
    table_header_status="Status",
    table_header_reason="Reason",
    table_status_installed="✅ Installed",
    table_status_skipped="⏭️  Skipped",
    table_reason_no_change="No version change",
    table_reason_updated="Version updated (MD5: {md5})",
    table_separator="├─",
    # 安装报告消息（固定格式）
    report_section_process="\n【Installation Process】",
    report_section_summary="\n【Installation Summary】",
    report_section_ignored="Ignored Directories (test/ and tests/)",
    report_no_actions="No actions taken (all skills up-to-date)",
    report_statistics="📊 Statistics: {installed} installed, {skipped} skipped",
    # 安装摘要消息（兼容保留）
    summary_header="\n📊 Installation Summary - {TARGET}",
    summary_installed="\n✅ Installed/Updated ({count} skills):",
    summary_skipped="\n⏭️  Skipped ({count} skills):",
    summary_reason="     Reason: {reason}",
    summary_total_header="\n🎯 Overall Installation Summary",
    summary_total_counts="\nTotal counts:",
    summary_installed_count="  • Installed/Updated: {count} skills",
    summary_skipped_count="  • Skipped: {count} skills",
    summary_new_install="  New install: {skills}",
    summary_unchanged="  Unchanged: {skills}",
    summary_manifest_saved="📝 Installation manifest saved: {path}",
    manifest_preview="[dry-run] manifest preview:",
    # 远程安装消息
    remote_check_intro="\n🌐 Remote Check Mode: Download and compare remote skills before installation.",
    remote_auto_intro="\n🌐 Remote Auto Mode: Automatically download and install remote skills.",
    remote_source_prompt="Do you want to install skills from '{name}'? ({description}) [y/N]: ",
    remote_source_prompt_recommended="Do you want to install skills from '{name}'? ({description}) [Recommended] [Y/n]: ",
    remote_download_progress="Downloading {name} from {url}...",
    remote_download_complete="Download complete: {name}",
    remote_download_failed="Download failed: {name} - {error}",
    remote_skills_path_missing="Download complete but skills_path not found: {name} - {path}",
    remote_compare_header="\n【Remote Skills Comparison Report】",
    remote_compare_new="🟢 New Skills ({count})",
    remote_compare_updated="🟡 Updatable Skills ({count})",
    remote_compare_unchanged="⚪ Latest Skills ({count})",
    remote_confirm_install="Confirm to install/update these skills? [y/N]: ",
    remote_confirm_install_detail="  New: {new_count}, Updated: {updated_count}, Unchanged: {unchanged_count}",
    remote_temp_dir="Temporary directory: {path}",
    remote_cleanup_complete="Temporary directory cleaned up.",
    remote_no_updates="No updates available.",
    remote_user_cancelled="User cancelled.",
    remote_git_not_found="Error: Git command not found. Please install Git first.",
    remote_config_not_found="Error: Config file not found: {path}",
    remote_config_error="Error: Failed to load config file: {error}",
    remote_source_filter_selected="Selected sources: {sources}",
    remote_source_filter_invalid="Warning: Invalid source ID(s) ignored: {invalid_ids}",
    remote_source_update_failed="Remote source update failed: {sources}",
    remote_source_stale_cache="Remote source freshness was not verified; used existing cache: {sources}",
)

# 中文消息
MESSAGES_ZH = Messages(
    arg_help_description="将本仓库的所有 skills 安装到 Codex/Claude Code 用户级 skills 目录（基于复制，使用 MD5 版本控制）。",
    arg_help_dry_run="打印操作但不写入任何内容。",
    arg_help_codex="仅安装到 Codex（默认：同时安装到 Codex 和 Claude Code）。",
    arg_help_claude="仅安装到 Claude Code（默认：同时安装到 Codex 和 Claude Code）。",
    arg_help_force="强制重新安装所有 skills，忽略 MD5 检查。",
    arg_help_skill="仅安装/更新指定 skill。可重复传入，也可用逗号分隔。",
    arg_help_remote="启用远程安装模式（从 GitHub 下载技能）。",
    arg_help_check="检查模式（安装前交互式确认）。",
    arg_help_auto="自动模式（强制安装，无需确认）。",
    arg_help_source_filter="按 ID 过滤远程源（如 --general、--research）。",
    error_no_skills_found="未发现可安装的 skills（扫描根目录：{root}）",
    error_skill_filter_missing="未找到指定 skill：{skills}",
    error_skill_filter_not_installable="指定 skill 不是可安装的普通技能：{skills}",
    error_skill_name_collision="检测到 skill 目录名冲突（basename 重复），无法安全安装：",
    error_source_root_not_found="未能自动识别 skills 源目录：请在包含 ./skills/alpha 的项目目录或其子目录中运行（安装器会检查当前目录及祖先目录）；beta 必须显式使用 --source，或直接传入 --source 指定其它源目录。",
    installing_to_target="正在安装到 {TARGET}: {root}",
    removed_legacy_symlink="已移除旧软链接: {path}",
    removed_legacy_skill="已移除 legacy 技能: {path}",
    skip_legacy_path="跳过旧路径（非软链接）: {path}",
    removed_existing="已删除: {dest}",
    installed="已安装: {dest}",
    dry_run_prefix="[dry-run] ",
    # 表格相关
    table_header_skill="Skill 名称",
    table_header_status="状态",
    table_header_reason="原因",
    table_status_installed="✅ 已安装",
    table_status_skipped="⏭️  跳过",
    table_reason_no_change="版本未变化",
    table_reason_updated="版本已更新 (MD5: {md5})",
    table_separator="├─",
    # 安装报告消息（固定格式）
    report_section_process="\n【安装过程】",
    report_section_summary="\n【安装摘要】",
    report_section_ignored="已忽略的目录 (test/ 和 tests/)",
    report_no_actions="无需操作（所有 skills 均为最新版本）",
    report_statistics="📊 统计：已安装 {installed} 个，跳过 {skipped} 个",
    # 安装摘要消息（兼容保留）
    summary_header="\n📊 安装摘要 - {TARGET}",
    summary_installed="\n✅ 已安装/更新 ({count} 个):",
    summary_skipped="\n⏭️  跳过 ({count} 个):",
    summary_reason="     原因: {reason}",
    summary_total_header="\n🎯 总体安装摘要",
    summary_total_counts="\n总计数:",
    summary_installed_count="  • 已安装/更新: {count} 个",
    summary_skipped_count="  • 跳过: {count} 个",
    summary_new_install="  新安装: {skills}",
    summary_unchanged="  未变化: {skills}",
    summary_manifest_saved="📝 安装清单已保存: {path}",
    manifest_preview="[dry-run] manifest preview:",
    # 远程安装消息
    remote_check_intro="\n🌐 远程检查模式：下载并对比远程技能后再安装。",
    remote_auto_intro="\n🌐 远程自动模式：自动下载并安装远程技能。",
    remote_source_prompt="是否要安装来自 '{name}' 的技能？({description}) [y/N]: ",
    remote_source_prompt_recommended="是否要安装来自 '{name}' 的技能？({description}) [推荐] [Y/n]: ",
    remote_download_progress="正在从 {url} 下载 {name}...",
    remote_download_complete="下载完成: {name}",
    remote_download_failed="下载失败: {name} - {error}",
    remote_skills_path_missing="下载完成但未找到 skills_path: {name} - {path}",
    remote_compare_header="\n【远程技能对比报告】",
    remote_compare_new="🟢 新增技能 ({count}个)",
    remote_compare_updated="🟡 可更新技能 ({count}个)",
    remote_compare_unchanged="⚪ 最新技能 ({count}个)",
    remote_confirm_install="确认安装/更新这些技能？ [y/N]: ",
    remote_confirm_install_detail="  新增: {new_count}个, 更新: {updated_count}个, 最新: {unchanged_count}个",
    remote_temp_dir="临时目录: {path}",
    remote_cleanup_complete="临时目录已清理。",
    remote_no_updates="没有可用的更新。",
    remote_user_cancelled="用户取消。",
    remote_git_not_found="错误: 未找到 Git 命令。请先安装 Git。",
    remote_config_not_found="错误: 配置文件不存在: {path}",
    remote_config_error="错误: 加载配置文件失败: {error}",
    remote_source_filter_selected="已选择源: {sources}",
    remote_source_filter_invalid="警告: 无效的源 ID 已忽略: {invalid_ids}",
    remote_source_update_failed="远程源更新失败: {sources}",
    remote_source_stale_cache="远程源未确认最新，已使用旧缓存: {sources}",
)


def detect_system_language() -> str:
    """检测系统语言设置。

    优先级：
    1. LC_ALL 环境变量
    2. LANG 环境变量
    3. locale.getdefaultlocale() 结果
    4. 默认英语

    Returns:
        语言代码（'zh' 或 'en'）
    """
    # 尝试从环境变量获取
    for var in ["LC_ALL", "LANG"]:
        lang = os.environ.get(var, "")
        if lang:
            # 提取语言代码（如 'zh_CN.UTF-8' -> 'zh'）
            code = lang.split("_")[0].split(".")[0].lower()
            if code in SUPPORTED_LANGUAGES:
                return code
            # 处理 'zh' 的变体
            if code.startswith("zh"):
                return "zh"

    # 尝试从 locale 获取
    try:
        default_locale = locale.getdefaultlocale()
        if default_locale and default_locale[0]:
            lang_code = default_locale[0].split("_")[0].lower()
            if lang_code in SUPPORTED_LANGUAGES:
                return lang_code
            if lang_code.startswith("zh"):
                return "zh"
    except (ValueError, AttributeError):
        pass

    # 默认返回英语
    return "en"


class Translator:
    """翻译器，提供统一的本地化消息访问接口。"""

    def __init__(self, lang_code: str | None = None) -> None:
        """初始化翻译器。

        Args:
            lang_code: 语言代码，如果为 None 则自动检测
        """
        if lang_code is None:
            lang_code = detect_system_language()

        # 确保语言代码受支持
        if lang_code not in SUPPORTED_LANGUAGES:
            lang_code = "en"

        self._lang_code = lang_code
        self._messages = MESSAGES_ZH if lang_code == "zh" else MESSAGES_EN

    @property
    def lang_code(self) -> str:
        """返回当前语言代码。"""
        return self._lang_code

    def get(self, attr: str, *args, **kwargs) -> str:
        """获取本地化消息。

        Args:
            attr: Messages 数据类的属性名
            *args: format() 的位置参数
            **kwargs: format() 的关键字参数

        Returns:
            格式化后的本地化消息
        """
        template = getattr(self._messages, attr)
        if args or kwargs:
            return template.format(*args, **kwargs)
        return template

    def __getattr__(self, name: str) -> Callable[..., str]:
        """提供统一的消息访问方法。

        所有消息都作为可调用对象返回，无论是否需要参数。
        不带参数调用: t.message()
        带参数调用: t.message(key=value)

        例如：
            t.arg_help_description()  # 返回字符串
            t.summary_header(TARGET='TEST')  # 返回格式化字符串
        """
        # 首先尝试从 _messages 获取原始值
        if hasattr(self._messages, name):
            value = getattr(self._messages, name)
            # 返回一个可调用对象，无论是否包含占位符
            return lambda *args, **kwargs: value.format(*args, **kwargs) if (args or kwargs) else value
        # 如果找不到，抛出 AttributeError
        raise AttributeError(f"'{type(self).__name__}' object has no attribute '{name}'")


# 全局翻译器实例（延迟初始化）
_global_translator: Translator | None = None


def get_translator() -> Translator:
    """获取全局翻译器实例（单例模式）。"""
    global _global_translator
    if _global_translator is None:
        _global_translator = Translator()
    return _global_translator


def set_language(lang_code: str) -> None:
    """设置全局语言（用于测试）。"""
    global _global_translator
    _global_translator = Translator(lang_code)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations

import argparse
import contextlib
import io
import fnmatch
import hashlib
import json
import os
import shutil
import subprocess
import sys
import time
from contextlib import contextmanager
from dataclasses import dataclass
from pathlib import Path

MIN_PYTHON = (3, 11)


def ensure_python() -> None:
    """Fail early when the full installer runs outside its support range."""
    if sys.version_info < MIN_PYTHON:
        current = ".".join(str(part) for part in sys.version_info[:3])
        required = ".".join(str(part) for part in MIN_PYTHON)
        raise SystemExit(
            f"The full local installer requires Python {required}+ "
            f"(current: {current}). On Python 3.8-3.10, use bootstrap_install.py."
        )


ensure_python()


# 设置 UTF-8 编码，解决 Windows GBK 环境下的 emoji 显示问题
if sys.platform == 'win32':
    import codecs
    sys.stdout = codecs.getwriter('utf-8')(sys.stdout.buffer, 'strict')
    sys.stderr = codecs.getwriter('utf-8')(sys.stderr.buffer, 'strict')

# 添加 scripts 目录到 Python 路径，以便导入 i18n
_scripts_dir = Path(__file__).resolve().parent
if str(_scripts_dir) not in sys.path:
    sys.path.insert(0, str(_scripts_dir))

from i18n import get_translator
from remove_legacy_skills import (
    load_legacy_skill_names as _load_legacy_skill_names,
    remove_legacy_skills as _remove_legacy_skills,
)
from managed_runtime import ManagedRuntimeError, ensure as ensure_managed_runtime, status as managed_runtime_status

_INSTALLATION_ROOT_PARTS = (".bensz-skills", "installation")
MANIFEST_SCHEMA_VERSION = 1
SILENT_UPDATE_STATE_SCHEMA_VERSION = 1
SILENT_UPDATE_TTL_SECONDS = 72 * 60 * 60


@dataclass(frozen=True)
class Target:
    label: str
    root: Path
    legacy_link: Path


@dataclass
class SkillType:
    """技能类型枚举。"""
    AUXILIARY: str = "auxiliary"  # 辅助技能（开发用，不安装）
    NORMAL: str = "normal"        # 普通技能（可安装）
    TEST: str = "test"            # 测试技能（测试用，不安装）


@dataclass
class SkillInfo:
    name: str
    src: Path
    dest: Path
    md5: str
    skill_type: str = SkillType.NORMAL  # 技能类型
    installed: bool = False
    skipped: bool = False
    reason: str = ""


@dataclass
class SkillComparison:
    """技能对比结果。"""
    name: str
    remote_md5: str
    local_md5: str | None
    status: str  # "new", "updated", "unchanged"
    remote_path: Path
    local_path: Path | None


@dataclass
class RemoteDownloadResult:
    """Result of preparing a remote skills source."""
    skills_dir: Path | None
    stale_cache_used: bool = False
    failed: bool = False
    error: str = ""


def _now_stamp() -> str:
    return time.strftime("%Y%m%d-%H%M%S", time.localtime())


def _is_symlink(path: Path) -> bool:
    try:
        return path.is_symlink()
    except OSError:
        return False


def _get_installation_root() -> Path:
    """返回 install-bensz-skills 的统一工作根目录。"""
    return Path.home().joinpath(*_INSTALLATION_ROOT_PARTS)


def _get_installation_root_label() -> str:
    """返回相对用户主目录的工作根目录标签。"""
    return str(Path(*_INSTALLATION_ROOT_PARTS))


def _get_state_dir() -> Path:
    return _get_installation_root() / "state"


def _get_silent_update_state_path() -> Path:
    return _get_state_dir() / "silent-update.json"


def _load_silent_update_state() -> dict:
    path = _get_silent_update_state_path()
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
        return data if isinstance(data, dict) else {}
    except (OSError, json.JSONDecodeError, TypeError):
        return {}


def _redact_error(value: object) -> str:
    text = str(value).replace(str(Path.home()), "~")
    for token in ("token=", "password=", "api_key=", "authorization:"):
        while token.lower() in text.lower():
            start = text.lower().index(token.lower())
            end = text.find(" ", start)
            text = text[:start] + token + "[redacted]" + (text[end:] if end >= 0 else "")
    return text[:500]


def _write_silent_update_state(*, result: str, error: str = "", **fields: object) -> None:
    state_dir = _get_state_dir()
    state_dir.mkdir(parents=True, exist_ok=True)
    path = _get_silent_update_state_path()
    payload = _load_silent_update_state()
    now = int(time.time())
    payload.update({
        "schema_version": SILENT_UPDATE_STATE_SCHEMA_VERSION,
        "last_check_started_at": payload.get("last_check_started_at", now),
        "last_check_completed_at": now,
        "last_check_completed_at_iso": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(now)),
        "last_result": result,
        **fields,
    })
    if error:
        payload["last_error"] = _redact_error(error)
    else:
        payload.pop("last_error", None)
    temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
    try:
        temporary.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
        os.replace(temporary, path)
    finally:
        temporary.unlink(missing_ok=True)


@contextmanager
def _silent_update_lock():
    """Serialize automatic checks without requiring a third-party package."""
    lock_path = _get_state_dir() / "silent-update.lock"
    _get_state_dir().mkdir(parents=True, exist_ok=True)
    handle = lock_path.open("a+", encoding="utf-8")
    try:
        try:
            import fcntl
            fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
        except (ImportError, BlockingIOError, OSError):
            yield False
            return
        yield True
    finally:
        try:
            if "fcntl" in sys.modules:
                fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
        finally:
            handle.close()


def _installed_skill_names(target: str | None = None) -> list[str]:
    names: set[str] = set()
    roots = {"codex": Path.home() / ".codex/skills", "claude": Path.home() / ".claude/skills"}
    for root in (roots.get(target),) if target else roots.values():
        if root is None:
            continue
        try:
            for child in root.iterdir():
                if child.is_dir() and (child / "SKILL.md").is_file():
                    names.add(child.name)
        except OSError:
            continue
    return sorted(names)


def _run_silent_update(*, t: get_translator().__class__) -> int:
    """Run at most one non-interactive update per 72-hour window."""
    with _silent_update_lock() as acquired:
        if not acquired:
            return 0
        state = _load_silent_update_state()
        if state.get("schema_version", 0) not in (0, SILENT_UPDATE_STATE_SCHEMA_VERSION):
            return 0
        completed = state.get("last_check_completed_at")
        if isinstance(completed, (int, float)) and time.time() - completed < SILENT_UPDATE_TTL_SECONDS:
            return 0
        platform_skills = {platform: _installed_skill_names(platform) for platform in ("codex", "claude")}
        output = io.StringIO()
        try:
            _write_silent_update_state(
                result="running",
                last_check_started_at=int(time.time()),
                platforms=platform_skills,
                source={"id": "general", "repository": "https://github.com/huangwb8/skills", "branch": "main", "skills_path": "skills/alpha"},
            )
            with contextlib.redirect_stdout(output), contextlib.redirect_stderr(output):
                runtime = ensure_managed_runtime()
                code = 0
                if any(platform_skills.values()):
                    code = _remote_install_main(
                        auto_mode=True,
                        install_codex=bool(platform_skills["codex"]),
                        install_claude=bool(platform_skills["claude"]),
                        source_filter=["general"],
                        platform_skill_filters=platform_skills,
                        skill_filter=sorted({name for names in platform_skills.values() for name in names}),
                        t=t,
                    )
            _write_silent_update_state(result="success" if code == 0 else "degraded",
                                       failure_kind="none" if code == 0 else "remote-or-install",
                                       error=output.getvalue(), platforms=platform_skills,
                                       runtime=runtime,
                                       source={"id": "general", "skills_path": "skills/alpha"})
        except Exception as exc:
            try:
                _write_silent_update_state(result="failed", error=str(exc))
            except OSError:
                pass
        return 0


_IGNORE_DIR_NAMES = {
    "__pycache__",
    ".pytest_cache",
    ".mypy_cache",
    ".ruff_cache",
    ".tox",
    ".nox",
    "test",
    "tests",
    "plans",
}
_IGNORE_FILE_NAMES = {".DS_Store"}
_IGNORE_ROOT_FILE_NAMES = {"readme.md", "changelog.md"}
_IGNORE_GLOBS = ("*.pyc", "*.pyo")


def _copytree_ignore(src_root: Path):
    """Return a copytree ignore callback based on installable relative paths."""

    def _ignore(current_dir: str, names: list[str]) -> list[str]:
        current_path = Path(current_dir)
        ignored: list[str] = []
        for name in names:
            try:
                rel_path = (current_path / name).relative_to(src_root)
            except ValueError:
                continue
            if _should_ignore_rel_path(rel_path):
                ignored.append(name)
        return ignored

    return _ignore


def _should_ignore_rel_path(rel_path: Path) -> bool:
    if len(rel_path.parts) == 1 and rel_path.name.lower() in _IGNORE_ROOT_FILE_NAMES:
        return True
    if any(part.startswith(".") for part in rel_path.parts):
        return True
    if any(part in _IGNORE_DIR_NAMES for part in rel_path.parts):
        return True
    if rel_path.name in _IGNORE_FILE_NAMES:
        return True
    if any(fnmatch.fnmatch(rel_path.name, pattern) for pattern in _IGNORE_GLOBS):
        return True
    return False


def _path_is_within(path: Path, root: Path) -> bool:
    """Return True if path is root or inside root (after resolve)."""
    try:
        path = path.resolve()
        root = root.resolve()
    except OSError:
        return False
    return path == root or root in path.parents


def _looks_like_skills_root(root: Path) -> bool:
    """Heuristic: a skills root contains at least one immediate subdir with SKILL.md."""
    try:
        if not root.exists() or not root.is_dir():
            return False
        for child in root.iterdir():
            if child.is_dir() and (child / "SKILL.md").is_file():
                return True
    except OSError:
        return False
    return False


def _detect_default_source_roots(script_path: Path, include_legacy: bool = False) -> list[Path]:
    """Detect a sensible default source root for local install.

    Goal: allow running this installer from a system-installed location (e.g. ~/.codex/skills)
    while still installing skills from the *current project* (cwd).
    """
    cwd = Path.cwd().resolve()
    candidates: list[Path] = []

    # Production layout: only the canonical alpha channel is selected implicitly.
    # Walk up from cwd so a system-installed installer also works when invoked
    # from a project subdirectory (or from inside ./skills/alpha). Historical
    # pipelines paths are considered only with the explicit migration flag.
    for base in (cwd, *cwd.parents):
        candidates_to_check = [base / "skills" / "alpha"]
        if include_legacy:
            candidates_to_check.insert(0, base / "pipelines" / "skills" / "alpha")
        for p in candidates_to_check:
            if _looks_like_skills_root(p):
                candidates.append(p)
                break
        if candidates:
            break

    # Fallback to "repo-local" layout when this script lives in the same checkout.
    repo_candidate = script_path.parents[2] if len(script_path.parents) > 2 else Path()
    installed_roots = [
        Path.home() / ".codex" / "skills",
        Path.home() / ".claude" / "skills",
    ]
    if not any(_path_is_within(repo_candidate, r) for r in installed_roots):
        if _looks_like_skills_root(repo_candidate) and repo_candidate.name == "alpha":
            candidates.append(repo_candidate)

    # De-dup while keeping order.
    deduped: list[Path] = []
    seen: set[Path] = set()
    for p in candidates:
        try:
            rp = p.resolve()
        except OSError:
            continue
        if rp not in seen:
            deduped.append(rp)
            seen.add(rp)
    return deduped


def _print_skill_table(
    installed_skills: list[SkillInfo],
    skipped_skills: list[SkillInfo],
    t: get_translator().__class__,
) -> None:
    """以表格形式打印 skills 安装结果。

    表格格式：
    ┌──────────────────────────────┬──────────────┬─────────────────────────────────┐
    │ Skill 名称                   │ 状态         │ 原因                             │
    ├──────────────────────────────┼──────────────┼─────────────────────────────────┤
    │ systematic-literature-review │ ✅ 已安装    │ 版本已更新 (MD5: xxx...)        │
    │ knit-rmd-html                │ ⏭️  跳过     │ 版本未变化                      │
    └──────────────────────────────┴──────────────┴─────────────────────────────────┘
    """
    if not installed_skills and not skipped_skills:
        return

    # 获取列标题
    header_skill = t.table_header_skill()
    header_status = t.table_header_status()
    header_reason = t.table_header_reason()

    # 计算实际显示宽度（中文/emoji 以双宽估算）
    def display_width(s: str) -> int:
        return sum(2 if ord(c) > 127 else 1 for c in s)

    def pad_display(s: str, width: int) -> str:
        padding = width - display_width(s)
        return s + (" " * max(padding, 0))

    # 计算列宽（基于显示宽度）
    all_skills = installed_skills + skipped_skills
    max_name_display = max((display_width(skill.name) for skill in all_skills), default=20)
    name_width = max(max_name_display, display_width(header_skill)) + 2

    # 计算原因列的最大宽度（考虑中文和英文）
    reason_samples = []
    for skill in all_skills:
        if skill.installed:
            reason_samples.append(t.table_reason_updated(md5=skill.md5[:12]))
        else:
            reason_samples.append(t.table_reason_no_change())

    max_reason_display = max((display_width(r) for r in reason_samples), default=20)
    reason_width = max(max_reason_display, display_width(header_reason)) + 4

    # 状态列宽度（考虑 emoji）
    status_samples = [header_status, t.table_status_installed(), t.table_status_skipped()]
    status_width = max(display_width(s) for s in status_samples) + 2

    # 构建分隔线
    separator = "─" * name_width + "┬" + "─" * status_width + "┬" + "─" * reason_width
    top_border = "┌" + separator + "┐"
    bottom_border = "└" + separator.replace("┬", "┴") + "┘"
    row_separator = "├" + separator.replace("┬", "┼") + "┤"

    # 打印表格
    print()
    print(top_border)

    # 表头
    print(f"│ {pad_display(header_skill, name_width)} │ {pad_display(header_status, status_width)} │ {pad_display(header_reason, reason_width)} │")
    print(row_separator)

    # 按状态排序：已安装的在前，跳过的在后
    sorted_skills = sorted(all_skills, key=lambda s: not s.installed)

    for skill in sorted_skills:
        if skill.installed:
            status = t.table_status_installed()
            reason = t.table_reason_updated(md5=skill.md5[:12])
        else:
            status = t.table_status_skipped()
            reason = t.table_reason_no_change()

        print(f"│ {pad_display(skill.name, name_width)} │ {pad_display(status, status_width)} │ {pad_display(reason, reason_width)} │")

    print(bottom_border)


def _calculate_skill_md5(skill_dir: Path) -> str:
    """计算 skill 目录的 MD5 哈希值。

    基于可安装文件生成稳定哈希，排除 tests/plans 等不参与安装的内容。
    """
    hasher = hashlib.md5()
    for file in sorted(skill_dir.rglob("*")):
        if not file.is_file():
            continue
        rel_path = file.relative_to(skill_dir)
        if _should_ignore_rel_path(rel_path):
            continue
        hasher.update(str(rel_path).encode("utf-8"))
        hasher.update(b"\0")
        hasher.update(file.read_bytes())
    return hasher.hexdigest()


def _get_installed_md5(dest_dir: Path, target: Target) -> str | None:
    """获取已安装 skill 的 MD5 值。

    从平台特定的 manifest 文件读取（如 .skill-manifest.claude.json），
    或回退到计算目录内容的 MD5。

    Args:
        dest_dir: 技能目标目录
        target: 目标平台信息（codex/claude）
    """
    # 平台特定的 manifest 文件名（避免不同平台的版本记录互相干扰）
    manifest_file = dest_dir / f".skill-manifest.{target.label}.json"
    if manifest_file.exists():
        try:
            data = json.loads(manifest_file.read_text(encoding="utf-8"))
            return data.get("md5")
        except (json.JSONDecodeError, KeyError):
            pass

    # 回退方案：尝试直接计算目录 MD5
    if dest_dir.exists():
        try:
            return _calculate_skill_md5(dest_dir)
        except Exception:
            pass

    return None


def _save_skill_manifest(dest_dir: Path, md5: str, source: str | Path, target: Target) -> None:
    """保存 skill 的版本信息到平台特定的 manifest 文件。

    Args:
        dest_dir: 技能目标目录
        md5: 技能内容的 MD5 哈希值
        source: 技能源目录路径或稳定来源标识
        target: 目标平台信息（codex/claude）
    """
    # 平台特定的 manifest 文件名
    manifest_file = dest_dir / f".skill-manifest.{target.label}.json"
    manifest_data = {
        "schema_version": MANIFEST_SCHEMA_VERSION,
        "md5": md5,
        "source": str(source),
        "source_id": "general" if "skills/alpha" in str(source) else str(source),
        "installed_at": _now_stamp(),
        "target": target.label,
        "target_root": str(target.root),
        "skills": [{
            "name": dest_dir.name,
            "md5": md5,
            "status": "installed",
            "reason": "",
        }],
    }
    manifest_file.write_text(json.dumps(manifest_data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")


def _get_skill_category_from_yaml(skill_dir: Path) -> str | None:
    """从 SKILL.md 的 YAML frontmatter 读取 category 字段。

    Returns:
        category 值（如 "auxiliary", "normal", "test"），如果不存在则返回 None
    """
    skill_md = skill_dir / "SKILL.md"
    if not skill_md.exists():
        return None

    try:
        content = skill_md.read_text(encoding="utf-8")
        lines = content.split("\n")
        if not lines or lines[0].strip() != "---":
            return None

        in_frontmatter = False
        for line in lines:
            stripped = line.strip()
            if stripped == "---":
                if not in_frontmatter:
                    in_frontmatter = True
                    continue
                break
            if in_frontmatter and line.startswith("category:"):
                category = line.split(":", 1)[1].strip().strip('"').strip("'")
                return category.lower()
    except Exception:
        pass

    return None


def _determine_skill_type(skill_dir: Path, skills_root: Path) -> str:
    """确定技能的类型（auxiliary/normal/test）。

    判断优先级：
    1. YAML frontmatter 中的 category 字段（最高优先级）
    2. 基于目录名和路径的启发式规则

    Args:
        skill_dir: 技能目录路径
        skills_root: skills 根目录

    Returns:
        技能类型：SkillType.AUXILIARY, SkillType.NORMAL, 或 SkillType.TEST
    """
    # 优先级1：从 YAML 读取 category（最高优先级，显式声明优先于启发式规则）
    category = _get_skill_category_from_yaml(skill_dir)
    if category:
        if category in {"auxiliary", "dev", "development"}:
            return SkillType.AUXILIARY
        elif category in {"test", "testing"}:
            return SkillType.TEST
        elif category in {"normal", "production"}:
            return SkillType.NORMAL
        # 其他值：继续检查启发式规则（向后兼容）

    # 优先级2：检查是否在 test/ 或 tests/ 目录下
    rel_path = skill_dir.relative_to(skills_root)
    if any(part in {"test", "tests"} for part in rel_path.parts):
        return SkillType.TEST

    # 优先级3：基于目录名的启发式规则
    dir_name = skill_dir.name.lower()

    # 辅助技能识别规则（基于目录名，不包含特殊排除）
    # 技能类型应优先通过 YAML frontmatter 中的 category 字段控制
    # 此处仅作为后备识别规则

    # 测试技能识别规则（时间戳格式）
    import re
    if re.match(r"^\d{8}_\d{6}$", dir_name):
        return SkillType.TEST

    # 测试技能识别规则（目录名包含 test）
    test_patterns = ["test-", "-test", "_test", "test_"]
    if any(pattern in dir_name for pattern in test_patterns):
        return SkillType.TEST

    # 默认：普通技能
    return SkillType.NORMAL


def _is_test_skill_dir(skill_dir: Path) -> bool:
    """判断是否为测试技能目录（向后兼容）。

    注意：此函数已废弃，请使用 _determine_skill_type() 替代。
    保留此函数是为了向后兼容旧代码。
    """
    return _determine_skill_type(skill_dir, skill_dir.parents[1]) == SkillType.TEST


def _find_skill_dirs(skills_root: Path, exclude_names: set[str]) -> dict[str, list[Path]]:
    """发现所有技能目录并按类型分类。

    只扫描顶级 skill 目录（即 skills_root 的直接子目录），跳过子目录中的 skill。
    例如：awesome-code/agents/xxx 中的 SKILL.md 会被跳过。

    Returns:
        包含三个键的字典：
        - "normal": 普通技能列表（可安装）
        - "auxiliary": 辅助技能列表（不安装）
        - "test": 测试技能列表（不安装）
    """
    # 按类型分组的技能目录
    skill_dirs_by_type: dict[str, list[Path]] = {
        SkillType.NORMAL: [],
        SkillType.AUXILIARY: [],
        SkillType.TEST: [],
    }

    # 只扫描顶级目录（skills_root 的直接子目录）
    for skill_dir in sorted(skills_root.iterdir()):
        # 跳过排除的目录名
        if skill_dir.name in exclude_names:
            continue
        # 跳过隐藏目录
        if skill_dir.name.startswith("."):
            continue
        # 跳过非目录
        if not skill_dir.is_dir():
            continue

        # 检查是否是 skill 目录（包含 SKILL.md）
        skill_md = skill_dir / "SKILL.md"
        if not skill_md.exists():
            continue

        # 跳过隐藏路径（以防万一）
        if any(part.startswith(".") for part in skill_dir.relative_to(skills_root).parts):
            continue

        # 确定技能类型
        skill_type = _determine_skill_type(skill_dir, skills_root)
        skill_dirs_by_type[skill_type].append(skill_dir)

    # Ensure no basename collisions (we install by directory name).
    # 仅检查普通技能的冲突（因为只有普通技能会被安装）
    by_name: dict[str, list[Path]] = {}
    for d in skill_dirs_by_type[SkillType.NORMAL]:
        by_name.setdefault(d.name, []).append(d)
    collisions = {name: paths for name, paths in by_name.items() if len(paths) > 1}
    if collisions:
        msg = ["检测到 skill 目录名冲突（basename 重复），无法安全安装："]
        for name, paths in sorted(collisions.items()):
            msg.append(f"- {name}: " + ", ".join(str(p) for p in paths))
        raise SystemExit("\n".join(msg))

    return skill_dirs_by_type


def _parse_skill_filter(raw_values: list[str] | None) -> list[str]:
    """解析 --skill 参数，支持重复传入和逗号分隔。"""
    if not raw_values:
        return []

    skill_names: list[str] = []
    seen: set[str] = set()
    for raw_value in raw_values:
        for part in raw_value.split(","):
            skill_name = part.strip()
            if not skill_name or skill_name in seen:
                continue
            skill_names.append(skill_name)
            seen.add(skill_name)
    return skill_names


def _filter_skill_dirs_by_names(
    skill_dirs_by_type: dict[str, list[Path]],
    skill_names: list[str],
) -> tuple[dict[str, list[Path]], list[str], list[str]]:
    """按 skill 目录名过滤扫描结果。

    Returns:
        (过滤后的目录、未找到的名称、存在但不可安装的名称)
    """
    if not skill_names:
        return skill_dirs_by_type, [], []

    selected_names = set(skill_names)
    existing_names: set[str] = set()
    normal_names = {path.name for path in skill_dirs_by_type[SkillType.NORMAL]}

    filtered: dict[str, list[Path]] = {
        SkillType.NORMAL: [],
        SkillType.AUXILIARY: [],
        SkillType.TEST: [],
    }
    for skill_type in [SkillType.NORMAL, SkillType.AUXILIARY, SkillType.TEST]:
        for skill_dir in skill_dirs_by_type[skill_type]:
            existing_names.add(skill_dir.name)
            if skill_dir.name in selected_names:
                filtered[skill_type].append(skill_dir)

    missing_names = [name for name in skill_names if name not in existing_names]
    not_installable_names = [
        name for name in skill_names
        if name in existing_names and name not in normal_names
    ]
    return filtered, missing_names, not_installable_names


# ============================================================================
# 远程安装相关函数
# ============================================================================

def _load_config(config_path: Path) -> dict:
    """加载配置文件。

    Returns:
        配置字典
    """
    if not config_path.exists():
        raise FileNotFoundError(config_path)

    try:
        import yaml
    except ImportError as exc:
        raise RuntimeError("缺少 PyYAML 依赖，请先运行 `python3 -m pip install pyyaml`") from exc

    try:
        with open(config_path, "r", encoding="utf-8") as f:
            return yaml.safe_load(f) or {}
    except Exception as exc:
        raise RuntimeError(f"配置文件解析失败: {exc}") from exc


def _check_git_available() -> bool:
    """检查 Git 命令是否可用。"""
    try:
        subprocess.run(
            ["git", "--version"],
            capture_output=True,
            check=True,
            text=True
        )
        return True
    except (subprocess.CalledProcessError, FileNotFoundError):
        return False


def _create_temp_dir() -> Path:
    """创建临时目录。

    Returns:
        临时目录路径
    """
    temp_dir = _get_installation_root() / "tmp-remote-install"
    if temp_dir.exists():
        shutil.rmtree(temp_dir)
    temp_dir.mkdir(parents=True, exist_ok=True)
    return temp_dir


def _cleanup_temp_dir(temp_dir: Path) -> None:
    """清理临时目录。"""
    if temp_dir.exists():
        shutil.rmtree(temp_dir)


def _get_remote_cache_root() -> Path:
    """Return persistent cache root for remote source repositories."""
    cache_root = _get_installation_root() / "cache" / "remote-sources"
    cache_root.mkdir(parents=True, exist_ok=True)
    return cache_root


def _sanitize_source_name(name: str, url: str) -> str:
    """生成安全且稳定的临时目录名。"""
    sanitized = name.strip()
    for sep in (os.sep, os.altsep, "/", "\\"):
        if sep:
            sanitized = sanitized.replace(sep, "-")
    sanitized = sanitized.replace(" ", "-").strip(".-")
    sanitized = "-".join(part for part in sanitized.split("-") if part)
    if not sanitized:
        sanitized = "source"
    suffix = hashlib.md5(url.encode("utf-8")).hexdigest()[:8] if url else "local"
    return f"{sanitized}-{suffix}"


def _get_remote_repo_cache_dir(name: str, url: str, branch: str) -> Path:
    """Return stable cache directory for a remote source and branch."""
    cache_key = hashlib.md5(f"{url}@{branch}".encode("utf-8")).hexdigest()[:12]
    safe_name = _sanitize_source_name(name, url).rsplit("-", 1)[0]
    return _get_remote_cache_root() / f"{safe_name}-{cache_key}"


def _format_remote_source_label(source_config: dict) -> str:
    """生成可写入 manifest 的稳定来源标识。"""
    name = source_config.get("name", "unknown")
    url = source_config.get("url", "")
    branch = source_config.get("branch", "main")
    skills_path = source_config.get("skills_path", "skills")
    if url:
        return f"{name} ({url}@{branch}:{skills_path})"
    return f"{name} (@{branch}:{skills_path})"


def _resolve_remote_skills_root(repo_root: Path, skills_path: str | None) -> Path | None:
    """解析远程仓库中的 skills 根目录。

    优先使用配置中的 skills_path；如果该路径不存在，但仓库根目录本身
    就是一个 skills 根目录，则回退到仓库根目录。
    """
    raw_path = (skills_path or "").strip()
    requested_path = repo_root if raw_path in {"", "."} else (repo_root / raw_path)

    if requested_path.exists() and _looks_like_skills_root(requested_path):
        return requested_path

    if not raw_path and repo_root != requested_path and _looks_like_skills_root(repo_root):
        return repo_root

    return None


def _normalize_git_path(path: str | None) -> str | None:
    """Return a safe relative Git path, or None when the repo root is needed."""
    raw_path = (path or "").strip().replace("\\", "/")
    while raw_path.startswith("./"):
        raw_path = raw_path[2:]
    raw_path = raw_path.strip("/")

    if raw_path in {"", "."}:
        return None

    parts = [part for part in raw_path.split("/") if part]
    if any(part == ".." for part in parts):
        return None

    return "/".join(parts)


def _build_sparse_checkout_paths(
    skills_path: str | None,
    skill_names: list[str] | None = None,
) -> list[str] | None:
    """Return sparse checkout paths, narrowed to requested skills when possible."""
    root_path = _normalize_git_path(skills_path)
    if not skill_names:
        return [root_path] if root_path else None

    sparse_paths: list[str] = []
    for skill_name in skill_names:
        safe_skill_name = _normalize_git_path(skill_name)
        if not safe_skill_name or "/" in safe_skill_name:
            continue
        sparse_paths.append(
            f"{root_path}/{safe_skill_name}" if root_path else safe_skill_name
        )

    if sparse_paths:
        return sparse_paths
    return [root_path] if root_path else None


def _path_or_nearby_contains_skills(path: Path) -> bool:
    """Return True when path, a direct child, or a parent is a usable skills root."""
    candidates = [path, path.parent]
    if path.is_dir():
        try:
            candidates.extend(child for child in path.iterdir() if child.is_dir())
        except OSError:
            pass

    seen: set[Path] = set()
    for candidate in candidates:
        try:
            resolved = candidate.resolve()
        except OSError:
            continue
        if resolved in seen:
            continue
        seen.add(resolved)

        if (candidate / "SKILL.md").is_file() or _looks_like_skills_root(candidate):
            return True

    return False


def _cached_repo_has_usable_content(repo_dir: Path, sparse_paths: list[str] | None) -> bool:
    """Return True if an existing cache can still satisfy a remote install run."""
    if not (repo_dir / ".git").is_dir():
        return False

    if not sparse_paths:
        return _path_or_nearby_contains_skills(repo_dir)

    return any(_path_or_nearby_contains_skills(repo_dir / path) for path in sparse_paths)


REMOTE_GIT_RETRIES = 3
REMOTE_GIT_RETRY_DELAY_SECONDS = 2
REMOTE_GIT_LOW_SPEED_LIMIT = "1024"
REMOTE_GIT_LOW_SPEED_TIME = "30"


def _git_env() -> dict[str, str]:
    """Return a Git environment tuned for non-interactive remote updates."""
    env = os.environ.copy()
    env.setdefault("GIT_TERMINAL_PROMPT", "0")
    env.setdefault("GIT_HTTP_LOW_SPEED_LIMIT", REMOTE_GIT_LOW_SPEED_LIMIT)
    env.setdefault("GIT_HTTP_LOW_SPEED_TIME", REMOTE_GIT_LOW_SPEED_TIME)
    return env


def _summarize_git_error(exc: BaseException) -> str:
    """Return a compact error string for retry/failure messages."""
    if isinstance(exc, subprocess.TimeoutExpired):
        cmd = " ".join(str(part) for part in exc.cmd)
        return f"timeout after {exc.timeout}s: {cmd}"

    if isinstance(exc, subprocess.CalledProcessError):
        stderr = (exc.stderr or "").strip()
        stdout = (exc.stdout or "").strip()
        detail = stderr or stdout or f"exit code {exc.returncode}"
        return detail.splitlines()[-1] if detail else f"exit code {exc.returncode}"

    return str(exc)


def _run_git_command(
    cmd: list[str],
    *,
    timeout: int,
    retries: int = REMOTE_GIT_RETRIES,
) -> subprocess.CompletedProcess[str]:
    """Run a Git command with retries for transient network failures."""
    last_error: BaseException | None = None
    for attempt in range(1, retries + 1):
        try:
            return subprocess.run(
                cmd,
                capture_output=True,
                check=True,
                text=True,
                timeout=timeout,
                env=_git_env(),
            )
        except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as exc:
            last_error = exc
            if attempt >= retries:
                break
            print(
                f"Git command failed ({attempt}/{retries}), retrying: "
                f"{_summarize_git_error(exc)}"
            )
            time.sleep(REMOTE_GIT_RETRY_DELAY_SECONDS * attempt)

    assert last_error is not None
    raise last_error


def _clone_remote_repo(
    *,
    url: str,
    branch: str,
    repo_dir: Path,
    sparse_paths: list[str] | None,
) -> None:
    """Clone a remote repository, using sparse checkout when a subdir is enough."""
    clone_cmd = [
        "git", "clone",
        "--depth", "1",
        "--no-tags",
        "--branch", branch,
        "--single-branch",
    ]
    if sparse_paths:
        clone_cmd.extend(["--filter=blob:none", "--sparse"])
    clone_cmd.extend([url, str(repo_dir)])

    _run_git_command(
        clone_cmd,
        timeout=300,
    )

    if sparse_paths:
        _run_git_command(
            ["git", "-C", str(repo_dir), "sparse-checkout", "set", *sparse_paths],
            timeout=300,
        )


def _clone_remote_repo_with_sparse_fallback(
    *,
    url: str,
    branch: str,
    repo_dir: Path,
    sparse_paths: list[str] | None,
) -> None:
    """Try sparse checkout first, then fall back to a full shallow clone."""
    try:
        _clone_remote_repo(
            url=url,
            branch=branch,
            repo_dir=repo_dir,
            sparse_paths=sparse_paths,
        )
    except (subprocess.CalledProcessError, subprocess.TimeoutExpired):
        if not sparse_paths:
            raise
        shutil.rmtree(repo_dir, ignore_errors=True)
        _clone_remote_repo(
            url=url,
            branch=branch,
            repo_dir=repo_dir,
            sparse_paths=None,
        )


def _update_cached_remote_repo(
    *,
    url: str,
    branch: str,
    repo_dir: Path,
    sparse_paths: list[str] | None,
) -> bool:
    """Create or update a cached remote repository.

    Returns True when an existing cache had to be reused because GitHub update
    failed. The caller can then surface the run as degraded instead of fresh.
    """
    if not (repo_dir / ".git").is_dir():
        if repo_dir.exists():
            shutil.rmtree(repo_dir)
        _clone_remote_repo_with_sparse_fallback(
            url=url,
            branch=branch,
            repo_dir=repo_dir,
            sparse_paths=sparse_paths,
        )
        return False

    try:
        _run_git_command(
            ["git", "-C", str(repo_dir), "remote", "set-url", "origin", url],
            timeout=60,
            retries=1,
        )
        _run_git_command(
            ["git", "-C", str(repo_dir), "fetch", "--depth", "1", "--no-tags", "origin", branch],
            timeout=300,
        )
        if sparse_paths:
            _run_git_command(
                ["git", "-C", str(repo_dir), "sparse-checkout", "set", *sparse_paths],
                timeout=300,
            )
        else:
            _run_git_command(
                ["git", "-C", str(repo_dir), "sparse-checkout", "disable"],
                timeout=300,
            )
        _run_git_command(
            ["git", "-C", str(repo_dir), "checkout", "--force", "FETCH_HEAD"],
            timeout=300,
        )
    except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as exc:
        if _cached_repo_has_usable_content(repo_dir, sparse_paths):
            print(
                "Git update failed; using existing remote cache: "
                f"{_summarize_git_error(exc)}"
            )
            return True
        shutil.rmtree(repo_dir, ignore_errors=True)
        _clone_remote_repo_with_sparse_fallback(
            url=url,
            branch=branch,
            repo_dir=repo_dir,
            sparse_paths=sparse_paths,
        )
        return False

    return False


def _download_remote_source(
    source_config: dict,
    temp_dir: Path,
    t: get_translator().__class__,
    skill_names: list[str] | None = None,
) -> RemoteDownloadResult:
    """从 GitHub 下载远程技能源到临时目录。

    Args:
        source_config: 源配置字典
        temp_dir: 临时目录
        t: 翻译器

    Returns:
        远程源准备结果；失败时 failed=True
    """
    name = source_config.get("name", "unknown")
    url = source_config.get("url", "")
    branch = source_config.get("branch", "main")
    skills_path = source_config.get("skills_path", "skills")

    print(t.remote_download_progress(name=name, url=url))

    # 持久缓存远程仓库；重复远程更新时用浅 fetch 代替从零 clone。
    repo_root = _get_remote_repo_cache_dir(name, url, branch)
    sparse_paths = _build_sparse_checkout_paths(skills_path, skill_names)
    requested_skills_root = repo_root
    normalized_skills_path = _normalize_git_path(skills_path)
    if normalized_skills_path:
        requested_skills_root = repo_root / normalized_skills_path

    try:
        # 如果配置指向仓库子目录，优先只拉取该子目录；如果指定了 --skill，
        # 进一步只拉取目标 skill 目录，减少远程更新等待时间。
        stale_cache_used = _update_cached_remote_repo(
            url=url,
            branch=branch,
            repo_dir=repo_root,
            sparse_paths=sparse_paths,
        )
    except subprocess.TimeoutExpired:
        print(t.remote_download_failed(name=name, error="timeout"))
        return RemoteDownloadResult(None, failed=True, error="timeout")
    except subprocess.CalledProcessError as e:
        error = _summarize_git_error(e)
        print(t.remote_download_failed(name=name, error=error))
        return RemoteDownloadResult(None, failed=True, error=error)
    except Exception as e:
        error = str(e)
        print(t.remote_download_failed(name=name, error=error))
        return RemoteDownloadResult(None, failed=True, error=error)

    # 返回技能目录路径
    skills_dir = _resolve_remote_skills_root(repo_root, skills_path)
    if skills_dir is None and skill_names:
        requested_skills_root.mkdir(parents=True, exist_ok=True)
        skills_dir = requested_skills_root
    elif skills_dir is None and sparse_paths:
        # 保留历史回退：若配置路径不存在但仓库根目录本身就是 skills 根目录，
        # 需要完整浅克隆后才能识别根目录。
        try:
            stale_cache_used = stale_cache_used or _update_cached_remote_repo(
                url=url,
                branch=branch,
                repo_dir=repo_root,
                sparse_paths=None,
            )
        except subprocess.TimeoutExpired:
            print(t.remote_download_failed(name=name, error="timeout"))
            return RemoteDownloadResult(None, stale_cache_used=stale_cache_used, failed=True, error="timeout")
        except subprocess.CalledProcessError as e:
            error = _summarize_git_error(e)
            print(t.remote_download_failed(name=name, error=error))
            return RemoteDownloadResult(None, stale_cache_used=stale_cache_used, failed=True, error=error)
        except Exception as e:
            error = str(e)
            print(t.remote_download_failed(name=name, error=error))
            return RemoteDownloadResult(None, stale_cache_used=stale_cache_used, failed=True, error=error)
        skills_dir = _resolve_remote_skills_root(repo_root, skills_path)

    if skills_dir is None:
        print(t.remote_skills_path_missing(name=name, path=skills_path))
        return RemoteDownloadResult(None, stale_cache_used=stale_cache_used, failed=True, error=f"skills_path not found: {skills_path}")

    print(t.remote_download_complete(name=name))
    return RemoteDownloadResult(skills_dir, stale_cache_used=stale_cache_used)


def _compare_remote_skills(
    remote_skills_dir: Path,
    local_target: Path,
    target: Target,
    t: get_translator().__class__,
    skill_names: list[str] | None = None,
    md5_cache: dict[Path, str] | None = None,
) -> list[SkillComparison]:
    """对比远程技能与本地已安装技能。

    Args:
        remote_skills_dir: 远程技能目录
        local_target: 本地目标目录（如 ~/.claude/skills）
        target: 目标平台信息
        t: 翻译器

    Returns:
        技能对比结果列表
    """
    comparisons: list[SkillComparison] = []

    # 发现远程技能
    remote_skill_dirs_by_type = _find_skill_dirs(remote_skills_dir, exclude_names=set())
    if skill_names:
        remote_skill_dirs_by_type, _, _ = _filter_skill_dirs_by_names(
            remote_skill_dirs_by_type,
            skill_names,
        )

    for remote_skill_dir in remote_skill_dirs_by_type.get(SkillType.NORMAL, []):
        skill_name = remote_skill_dir.name
        remote_path_key = remote_skill_dir.resolve()
        if md5_cache is not None and remote_path_key in md5_cache:
            remote_md5 = md5_cache[remote_path_key]
        else:
            remote_md5 = _calculate_skill_md5(remote_skill_dir)
            if md5_cache is not None:
                md5_cache[remote_path_key] = remote_md5
        local_skill_dir = local_target / skill_name

        if local_skill_dir.exists():
            # 技能已存在，检查是否需要更新
            local_md5 = _get_installed_md5(local_skill_dir, target)
            if local_md5 != remote_md5:
                status = "updated"
            else:
                status = "unchanged"
        else:
            # 新技能
            local_md5 = None
            status = "new"

        comparisons.append(SkillComparison(
            name=skill_name,
            remote_md5=remote_md5,
            local_md5=local_md5,
            status=status,
            remote_path=remote_skill_dir,
            local_path=local_skill_dir if local_skill_dir.exists() else None,
        ))

    return comparisons


def _print_comparison_report(
    comparisons: list[SkillComparison],
    t: get_translator().__class__,
) -> None:
    """打印远程技能对比报告。

    Args:
        comparisons: 技能对比结果列表
        t: 翻译器
    """
    print()
    print(t.remote_compare_header())
    print("─" * 60)

    # 按状态分组
    new_skills = [c for c in comparisons if c.status == "new"]
    updated_skills = [c for c in comparisons if c.status == "updated"]
    unchanged_skills = [c for c in comparisons if c.status == "unchanged"]

    # 打印新增技能
    if new_skills:
        print()
        print(t.remote_compare_new(count=len(new_skills)))
        for skill in new_skills:
            print(f"   • {skill.name}")

    # 打印可更新技能
    if updated_skills:
        print()
        print(t.remote_compare_updated(count=len(updated_skills)))
        for skill in updated_skills:
            print(f"   • {skill.name}")

    # 打印最新技能
    if unchanged_skills:
        print()
        print(t.remote_compare_unchanged(count=len(unchanged_skills)))
        for skill in unchanged_skills:
            print(f"   • {skill.name}")


def _prompt_user_install(
    comparisons: list[SkillComparison],
    source_name: str,
    t: get_translator().__class__,
) -> bool:
    """询问用户是否确认安装/更新远程技能。

    Args:
        comparisons: 技能对比结果列表
        source_name: 源名称
        t: 翻译器

    Returns:
        True 表示用户确认，False 表示取消
    """
    new_count = len([c for c in comparisons if c.status == "new"])
    updated_count = len([c for c in comparisons if c.status == "updated"])
    unchanged_count = len([c for c in comparisons if c.status == "unchanged"])

    # 如果没有需要安装或更新的技能，跳过
    if new_count == 0 and updated_count == 0:
        print(t.remote_no_updates())
        return False

    print()
    print(t.remote_confirm_install_detail(
        new_count=new_count,
        updated_count=updated_count,
        unchanged_count=unchanged_count,
    ))
    response = input(t.remote_confirm_install()).strip().lower()

    return response in ("y", "yes", "是")


def _prompt_source_install(
    source_config: dict,
    t: get_translator().__class__,
) -> bool:
    """询问用户是否安装某个源。

    Args:
        source_config: 源配置
        t: 翻译器

    Returns:
        True 表示用户确认，False 表示跳过
    """
    name = source_config.get("name", "")
    description = source_config.get("description", "")
    recommended = source_config.get("recommended", False)

    if recommended:
        prompt = t.remote_source_prompt_recommended(name=name, description=description)
        default = "y"
    else:
        prompt = t.remote_source_prompt(name=name, description=description)
        default = "n"

    response = input(prompt).strip().lower()
    if not response:
        return default == "y"

    return response in ("y", "yes", "是")


def _install_remote_skills(
    comparisons: list[SkillComparison],
    target: Target,
    force: bool,
    source_label: str,
    legacy_skill_names: list[str],
    t: get_translator().__class__,
) -> InstallReport:
    """安装远程技能。

    Args:
        comparisons: 技能对比结果列表
        target: 目标平台
        force: 是否强制安装
        source_label: 远程来源标识（用于 manifest）
        t: 翻译器

    Returns:
        安装报告
    """
    installed_skills: list[SkillInfo] = []
    skipped_skills: list[SkillInfo] = []
    process_messages: list[str] = []

    # 处理旧的软链接
    legacy_msg = _safe_remove_legacy_symlink(target.legacy_link, dry_run=False, t=t)
    if legacy_msg:
        process_messages.append(legacy_msg)

    target.root.mkdir(parents=True, exist_ok=True)
    process_messages.extend(
        _remove_legacy_skills(
            target_root=target.root,
            legacy_skill_names=legacy_skill_names,
            dry_run=False,
            active_skill_names={comparison.name for comparison in comparisons},
        )
    )

    for comparison in comparisons:
        if comparison.status == "unchanged" and not force:
            skipped_skills.append(SkillInfo(
                name=comparison.name,
                src=comparison.remote_path,
                dest=target.root / comparison.name,
                md5=comparison.remote_md5,
                skill_type=SkillType.NORMAL,
                skipped=True,
                reason=t.table_reason_no_change(),
            ))
            continue

        dest_dir = target.root / comparison.name
        stage_root = _get_installation_root() / "staging" / f"{os.getpid()}-{target.label}"
        stage_root.mkdir(parents=True, exist_ok=True)
        staged = stage_root / comparison.name
        try:
            if staged.exists():
                shutil.rmtree(staged)
            shutil.copytree(comparison.remote_path, staged, symlinks=False, ignore=_copytree_ignore(comparison.remote_path))
            if dest_dir.exists() or dest_dir.is_symlink():
                backup = stage_root / f".old-{comparison.name}"
                os.replace(dest_dir, backup)
            os.replace(staged, dest_dir)
            backup = stage_root / f".old-{comparison.name}"
            if backup.exists() or backup.is_symlink():
                shutil.rmtree(backup) if backup.is_dir() and not backup.is_symlink() else backup.unlink()
        except OSError as exc:
            backup = stage_root / f".old-{comparison.name}"
            if not dest_dir.exists() and (backup.exists() or backup.is_symlink()):
                os.replace(backup, dest_dir)
            process_messages.append(f"staging failed for {comparison.name}: {_redact_error(exc)}")
            continue

        _save_skill_manifest(dest_dir, comparison.remote_md5, source_label, target)

        reason_msg = t.table_reason_updated(md5=comparison.remote_md5[:12])
        installed_skills.append(SkillInfo(
            name=comparison.name,
            src=comparison.remote_path,
            dest=dest_dir,
            md5=comparison.remote_md5,
            skill_type=SkillType.NORMAL,
            installed=True,
            reason=reason_msg,
        ))

    return InstallReport(
        target_label=target.label,
        target_root=target.root,
        installed_skills=installed_skills,
        skipped_skills=skipped_skills,
        auxiliary_skills=[],
        test_skills=[],
        process_messages=process_messages,
    )


def _remote_install_main(
    *,
    auto_mode: bool,
    install_codex: bool,
    install_claude: bool,
    source_filter: list[str] | None = None,
    skill_filter: list[str] | None = None,
    available_source_ids: list[str] | None = None,
    legacy_skill_names: list[str] | None = None,
    platform_skill_filters: dict[str, list[str]] | None = None,
    t: get_translator().__class__,
) -> int:
    """远程安装主流程。

    Args:
        auto_mode: 是否自动模式（无需确认）
        install_codex: 是否安装到 Codex
        install_claude: 是否安装到 Claude Code
        source_filter: 要安装的源 ID 列表（None 表示安装所有源）
        skill_filter: 要安装的 skill 名称列表（None 表示安装所有 skill）
        available_source_ids: 配置中可用的源 ID 列表
        t: 翻译器

    Returns:
        退出代码
    """
    # 打印介绍
    if auto_mode:
        print(t.remote_auto_intro())
    else:
        print(t.remote_check_intro())

    # 检查 Git
    if not _check_git_available():
        print(t.remote_git_not_found())
        return 1

    # 加载配置
    config_path = Path(__file__).resolve().parents[1] / "config.yaml"
    try:
        config = _load_config(config_path)
    except FileNotFoundError:
        print(t.remote_config_not_found(path=config_path))
        return 1
    except RuntimeError as exc:
        print(t.remote_config_error(error=str(exc)))
        return 1

    remote_sources = config.get("remote_sources", [])
    if not remote_sources:
        print(t.remote_config_error(error=f"未配置 remote_sources: {config_path}"))
        return 1

    # 应用源过滤
    if source_filter:
        # 检查无效的源 ID
        valid_ids = {s.get("id", "") for s in remote_sources if s.get("id")}
        invalid_ids = [sid for sid in source_filter if sid not in valid_ids]
        if invalid_ids:
            print(t.remote_source_filter_invalid(invalid_ids=", ".join(invalid_ids)))

        # 过滤源
        filtered_sources = [s for s in remote_sources if s.get("id") in source_filter]
        if not filtered_sources:
            print(t.remote_config_error(error=f"没有找到匹配的源，请检查源 ID: {', '.join(source_filter)}"))
            return 1

        remote_sources = filtered_sources
        print(t.remote_source_filter_selected(sources=", ".join(s.get("name", s.get("id", "")) for s in remote_sources)))

    # 创建临时目录
    temp_dir = _create_temp_dir()
    print(t.remote_temp_dir(path=temp_dir.relative_to(Path.home())))

    # 确定目标
    targets: list[Target] = []
    home = Path.home()
    if install_codex:
        targets.append(Target(
            label="codex",
            root=home / ".codex/skills",
            legacy_link=home / ".codex/skills/pipeline-skills",
        ))
    if install_claude:
        targets.append(Target(
            label="claude",
            root=home / ".claude/skills",
            legacy_link=home / ".claude/skills/pipeline-skills",
        ))

    all_reports: list[InstallReport] = []
    matched_skill_names: set[str] = set()
    failed_sources: list[str] = []
    stale_cache_sources: list[str] = []

    try:
        # 遍历每个远程源
        for source in remote_sources:
            source_name = source.get("name", "unknown")
            source_label = _format_remote_source_label(source)

            # 询问用户是否安装（check 模式）
            if not auto_mode:
                if not _prompt_source_install(source, t):
                    print(t.remote_user_cancelled())
                    continue

            # 下载远程源
            remote_result = _download_remote_source(source, temp_dir, t, skill_names=skill_filter)
            if remote_result.failed or remote_result.skills_dir is None:
                failed_sources.append(source_name)
                continue
            if remote_result.stale_cache_used:
                stale_cache_sources.append(source_name)
            remote_skills_dir = remote_result.skills_dir
            remote_md5_cache: dict[Path, str] = {}

            # 对每个目标进行处理
            for target in targets:
                print(f"\n{'=' * 60}")
                print(f"📦 {t.installing_to_target(TARGET=target.label.upper(), root=target.root)}")
                print(f"{'=' * 60}")

                # 对比远程与本地技能
                target_skill_filter = (
                    platform_skill_filters.get(target.label)
                    if platform_skill_filters is not None
                    else skill_filter
                )
                comparisons = _compare_remote_skills(
                    remote_skills_dir,
                    target.root,
                    target,
                    t,
                    skill_names=target_skill_filter,
                    md5_cache=remote_md5_cache,
                )
                matched_skill_names.update(comparison.name for comparison in comparisons)

                if not comparisons:
                    if not skill_filter:
                        print(t.error_no_skills_found(root=remote_skills_dir))
                    continue

                # 打印对比报告（check 模式）
                if not auto_mode:
                    _print_comparison_report(comparisons, t)

                    # 询问用户是否确认安装
                    if not _prompt_user_install(comparisons, source_name, t):
                        print(t.remote_user_cancelled())
                        continue

                # 安装技能
                report = _install_remote_skills(
                    comparisons,
                    target,
                    force=auto_mode,
                    source_label=source_label,
                    legacy_skill_names=[] if skill_filter else legacy_skill_names,
                    t=t,
                )
                all_reports.append(report)

                # 打印报告
                _print_report(report, t)

    finally:
        # 清理临时目录
        _cleanup_temp_dir(temp_dir)
        print()
        print(t.remote_cleanup_complete())

    # 输出总体摘要
    if all_reports:
        print(f"\n{'=' * 60}")
        print(t.summary_total_header())
        print(f"{'=' * 60}")

        installed_skill_names = {s.name for r in all_reports for s in r.installed_skills}
        skipped_skill_names = {s.name for r in all_reports for s in r.skipped_skills}
        total_installed = len(installed_skill_names)
        total_skipped = len(skipped_skill_names)

        print(t.summary_total_counts())
        print(t.summary_installed_count(count=total_installed))
        print(t.summary_skipped_count(count=total_skipped))

        print(f"{'=' * 60}\n")

        # Remote runs use the same public manifest contract as local runs.
        manifest_dir = _get_manifest_dir()
        manifest_path = manifest_dir / f"install-manifest.{_now_stamp()}.json"
        manifest_path.write_text(
            json.dumps(
                {"runs": [r.to_manifest_dict(source="remote") for r in all_reports]},
                ensure_ascii=False,
                indent=2,
            )
            + "\n",
            encoding="utf-8",
        )
        print(t.summary_manifest_saved(path=manifest_path.relative_to(Path.home())))

    if skill_filter:
        missing_names = [name for name in skill_filter if name not in matched_skill_names]
        if missing_names:
            print(t.error_skill_filter_missing(skills=", ".join(missing_names)))
            return 1

    if failed_sources:
        print(t.remote_source_update_failed(sources=", ".join(failed_sources)))
        return 1

    if stale_cache_sources:
        print(t.remote_source_stale_cache(sources=", ".join(stale_cache_sources)))
        return 1

    return 0


@dataclass
class InstallReport:
    """安装报告数据类。"""
    target_label: str
    target_root: Path
    installed_skills: list[SkillInfo]
    skipped_skills: list[SkillInfo]
    auxiliary_skills: list[SkillInfo] = None  # 辅助技能（被忽略）
    test_skills: list[SkillInfo] = None  # 测试技能（被忽略）
    process_messages: list[str] = None  # 安装过程中的消息
    removed_legacy: bool = False
    removed_existing: list[str] = None

    def __post_init__(self):
        if self.auxiliary_skills is None:
            self.auxiliary_skills = []
        if self.test_skills is None:
            self.test_skills = []
        if self.process_messages is None:
            self.process_messages = []
        if self.removed_existing is None:
            self.removed_existing = []

    def to_manifest_dict(self, source: str | Path | None = None) -> dict:
        """转换为可序列化的字典格式（用于 manifest 文件）。"""
        skills_list = []
        for skill in self.installed_skills:
            skills_list.append({
                "name": skill.name,
                "src": str(skill.src),
                "dest": str(skill.dest),
    

{
  "schema_version": 1,
  "environment": {
    "name": "benszapi",
    "prefix": ".bensz-skills/envs/benszapi",
    "python": "3.12"
  },
  "update_ttl_hours": 72,
  "packages": [
    {
      "id": "bsk",
      "distribution": "bensz-skill-kernel",
      "command": "bsk",
      "version_policy": "latest-production",
      "health_checks": [
        ["--version"],
        ["diagnostics"],
        ["capabilities"]
      ]
    }
  ]
}

#!/usr/bin/env python3
"""Manage the Bensz-owned Conda runtime without shell activation.

This module intentionally uses only the Python 3.8 standard library so the
bootstrap installer can use it before the managed Python is available.
"""

from __future__ import print_function

import argparse
import contextlib
import json
import os
import shutil
import stat
import subprocess
import time
from pathlib import Path


CONFIG_NAME = "managed-runtime.json"
STATE_SCHEMA_VERSION = 1
LOCK_TIMEOUT_SECONDS = 30
STALE_LOCK_SECONDS = 30 * 60


class ManagedRuntimeError(RuntimeError):
    """A stable, user-actionable managed-runtime failure."""


def _config_path():
    return Path(__file__).resolve().with_name(CONFIG_NAME)


def load_config(path=None):
    source = Path(path) if path is not None else _config_path()
    try:
        data = json.loads(source.read_text(encoding="utf-8"))
    except (OSError, ValueError) as exc:
        raise ManagedRuntimeError("cannot load managed runtime configuration") from exc
    if data.get("schema_version") != 1:
        raise ManagedRuntimeError("unsupported managed runtime configuration schema")
    environment = data.get("environment")
    packages = data.get("packages")
    if not isinstance(environment, dict) or not isinstance(packages, list) or not packages:
        raise ManagedRuntimeError("managed runtime configuration is incomplete")
    return data


def runtime_prefix(config, home=None):
    root = Path(home) if home is not None else Path.home()
    configured = config["environment"].get("prefix", ".bensz-skills/envs/benszapi")
    path = Path(str(configured))
    if path.is_absolute():
        raise ManagedRuntimeError("managed runtime prefix must be relative to the user home")
    resolved = (root / path).resolve()
    home_resolved = root.resolve()
    try:
        resolved.relative_to(home_resolved)
    except ValueError as exc:
        raise ManagedRuntimeError("managed runtime prefix escapes the user home") from exc
    return resolved


def runtime_python(prefix):
    if os.name == "nt":
        return prefix / "python.exe"
    return prefix / "bin" / "python"


def runtime_command(prefix, command):
    if os.name == "nt":
        scripts = prefix / "Scripts"
        for suffix in (".exe", ".cmd", ".bat", ""):
            candidate = scripts / (command + suffix)
            if candidate.exists():
                return candidate
        return scripts / (command + ".exe")
    return prefix / "bin" / command


def state_path(home=None):
    root = Path(home) if home is not None else Path.home()
    return root / ".bensz-skills" / "installation" / "state" / "managed-runtime.json"


def _read_state(home=None):
    try:
        value = json.loads(state_path(home).read_text(encoding="utf-8"))
        return value if isinstance(value, dict) else {}
    except (OSError, ValueError, TypeError):
        return {}


def _write_state(payload, home=None):
    path = state_path(home)
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_name(".%s.%s.tmp" % (path.name, os.getpid()))
    data = dict(payload)
    data["schema_version"] = STATE_SCHEMA_VERSION
    try:
        temporary.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
        os.replace(str(temporary), str(path))
    finally:
        try:
            temporary.unlink()
        except FileNotFoundError:
            pass


@contextlib.contextmanager
def _runtime_lock(home=None):
    path = state_path(home).with_name("managed-runtime.lock")
    path.parent.mkdir(parents=True, exist_ok=True)
    token = "%s:%s" % (os.getpid(), time.time_ns())
    deadline = time.monotonic() + LOCK_TIMEOUT_SECONDS
    acquired = False
    while time.monotonic() < deadline:
        try:
            descriptor = os.open(str(path), os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
            with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
                handle.write(token)
            acquired = True
            break
        except FileExistsError:
            try:
                if time.time() - path.stat().st_mtime > STALE_LOCK_SECONDS:
                    path.unlink()
                    continue
            except FileNotFoundError:
                continue
            time.sleep(0.1)
    if not acquired:
        raise ManagedRuntimeError("managed runtime is being updated by another process")
    try:
        yield
    finally:
        try:
            if path.read_text(encoding="utf-8") == token:
                path.unlink()
        except FileNotFoundError:
            pass


def _path_label(path, home=None):
    root = (Path(home) if home is not None else Path.home()).resolve()
    try:
        return "~/" + str(Path(path).resolve().relative_to(root))
    except ValueError:
        return "<external>"


def find_conda():
    explicit = os.environ.get("BENSZ_CONDA_EXE") or os.environ.get("CONDA_EXE")
    if explicit:
        candidate = Path(explicit).expanduser()
        if candidate.is_file():
            return str(candidate)
    for name in ("conda", "mamba", "micromamba"):
        candidate = shutil.which(name)
        if candidate:
            return candidate
    raise ManagedRuntimeError(
        "Conda, Mamba, or Micromamba was not found; install one or set BENSZ_CONDA_EXE"
    )


def _run(command, timeout=300):
    try:
        return subprocess.run(
            [str(part) for part in command],
            capture_output=True,
            text=True,
            check=False,
            timeout=timeout,
        )
    except (OSError, subprocess.SubprocessError) as exc:
        raise ManagedRuntimeError("managed runtime command could not be executed") from exc


def _checked_run(command, label, timeout=300):
    result = _run(command, timeout=timeout)
    if result.returncode != 0:
        detail = (result.stderr or result.stdout or "").strip().splitlines()
        suffix = (": " + _redact_detail(detail[-1])[:300]) if detail else ""
        raise ManagedRuntimeError(label + " failed" + suffix)
    return result


def _redact_detail(value):
    return str(value).replace(str(Path.home()), "~")


def _package_versions(python, packages):
    names = [str(item["distribution"]) for item in packages]
    script = (
        "import importlib.metadata as m,json; "
        "print(json.dumps({n:m.version(n) for n in %r},sort_keys=True))" % names
    )
    result = _checked_run([python, "-c", script], "managed package version check", timeout=30)
    try:
        value = json.loads(result.stdout)
    except ValueError as exc:
        raise ManagedRuntimeError("managed package version check returned invalid JSON") from exc
    return value


def _health_check(prefix, config):
    python = runtime_python(prefix)
    if not python.is_file():
        raise ManagedRuntimeError("managed Conda environment has no Python interpreter")
    versions = _package_versions(python, config["packages"])
    for package in config["packages"]:
        executable = runtime_command(prefix, str(package["command"]))
        if not executable.exists():
            raise ManagedRuntimeError("managed command is missing: " + str(package["command"]))
        for arguments in package.get("health_checks", []):
            _checked_run([executable] + list(arguments), "health check for " + str(package["id"]), timeout=60)
    return versions


def _needs_update(config, home=None, force=False):
    if force:
        return True
    state = _read_state(home)
    if state.get("schema_version") != STATE_SCHEMA_VERSION:
        return True
    completed = state.get("last_check_completed_at")
    ttl = int(config.get("update_ttl_hours", 72)) * 60 * 60
    return not isinstance(completed, (int, float)) or time.time() - completed >= ttl


def _has_package_drift(packages, home=None):
    recorded = _read_state(home).get("packages")
    return not isinstance(recorded, dict) or recorded != packages


def _write_posix_launcher(path, target):
    content = "#!/bin/sh\nexec %s \"$@\"\n" % _shell_quote(str(target))
    temporary = path.with_name(".%s.%s.tmp" % (path.name, os.getpid()))
    temporary.write_text(content, encoding="utf-8")
    temporary.chmod(temporary.stat().st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
    os.replace(str(temporary), str(path))


def _shell_quote(value):
    return "'" + value.replace("'", "'\"'\"'") + "'"


def _install_launchers(prefix, config, home=None):
    root = Path(home) if home is not None else Path.home()
    bin_dir = root / ".bensz-skills" / "bin"
    bin_dir.mkdir(parents=True, exist_ok=True)
    installed = {}
    for package in config["packages"]:
        command = str(package["command"])
        target = runtime_command(prefix, command)
        if os.name == "nt":
            launcher = bin_dir / (command + ".cmd")
            launcher.write_text('@echo off\r\n"%s" %%*\r\n' % target, encoding="utf-8")
        else:
            launcher = bin_dir / command
            _write_posix_launcher(launcher, target)
        installed[command] = _path_label(launcher, home)
    return installed


def status(config=None, home=None):
    config = config or load_config()
    prefix = runtime_prefix(config, home)
    result = {
        "schema_version": STATE_SCHEMA_VERSION,
        "environment": str(config["environment"].get("name", "benszapi")),
        "prefix": _path_label(prefix, home),
        "ready": False,
        "packages": {},
    }
    try:
        result["packages"] = _health_check(prefix, config)
        result["ready"] = True
    except ManagedRuntimeError as exc:
        result["reason"] = str(exc)
    return result


def _ensure_unlocked(config, home=None, force_update=False):
    prefix = runtime_prefix(config, home)
    python = runtime_python(prefix)
    created = False
    updated = False

    if not python.is_file():
        if prefix.exists() and (not prefix.is_dir() or any(prefix.iterdir())):
            raise ManagedRuntimeError(
                "managed runtime prefix exists but is not a valid Conda environment: "
                + _path_label(prefix, home)
            )
        conda = find_conda()
        create_command = [
            conda,
            "create",
            "--yes",
            "--prefix",
            str(prefix),
            "python=" + str(config["environment"].get("python", "3.12")),
            "pip",
        ]
        prefix.parent.mkdir(parents=True, exist_ok=True)
        _checked_run(create_command, "managed Conda environment creation", timeout=900)
        if not python.is_file():
            raise ManagedRuntimeError("Conda reported success but the managed Python is missing")
        created = True

    current = status(config, home)
    healthy = current.get("ready", False)
    drifted = healthy and _has_package_drift(current.get("packages", {}), home)
    if created or not healthy or drifted or _needs_update(config, home, force_update):
        specs = [str(item["distribution"]) for item in config["packages"]]
        _checked_run(
            [python, "-m", "pip", "install", "--disable-pip-version-check", "--no-input", "--upgrade"] + specs,
            "managed package update",
            timeout=900,
        )
        updated = True

    versions = _health_check(prefix, config)
    launchers = _install_launchers(prefix, config, home)
    now = int(time.time())
    payload = {
        "last_check_completed_at": now,
        "last_check_completed_at_iso": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(now)),
        "last_result": "success",
        "environment": str(config["environment"].get("name", "benszapi")),
        "prefix": _path_label(prefix, home),
        "packages": versions,
        "launchers": launchers,
    }
    _write_state(payload, home)
    return dict(payload, ready=True, created=created, updated=updated)


def ensure(config=None, home=None, force_update=False, dry_run=False):
    config = config or load_config()
    prefix = runtime_prefix(config, home)
    current = status(config, home)
    if dry_run:
        return {
            "ready": current.get("ready", False),
            "dry_run": True,
            "would_create": None if runtime_python(prefix).is_file() else _path_label(prefix, home),
            "would_update": (
                force_update
                or not current.get("ready", False)
                or _has_package_drift(current.get("packages", {}), home)
                or _needs_update(config, home)
            ),
            "would_install": [item["distribution"] for item in config["packages"]],
        }
    with _runtime_lock(home):
        return _ensure_unlocked(config, home, force_update)


def build_parser():
    parser = argparse.ArgumentParser(description="Manage the Bensz-owned Conda runtime")
    commands = parser.add_subparsers(dest="command", required=True)
    ensure_parser = commands.add_parser("ensure", help="create, update, and validate the runtime")
    ensure_parser.add_argument("--force-update", action="store_true")
    ensure_parser.add_argument("--dry-run", action="store_true")
    commands.add_parser("status", help="inspect the runtime without modifying it")
    return parser


def main(argv=None):
    args = build_parser().parse_args(argv)
    try:
        result = ensure(force_update=args.force_update, dry_run=args.dry_run) if args.command == "ensure" else status()
    except ManagedRuntimeError as exc:
        print(json.dumps({"ready": False, "error": str(exc)}, ensure_ascii=False))
        return 1
    print(json.dumps(result, ensure_ascii=False, indent=2))
    return 0 if result.get("ready") or result.get("dry_run") else 1


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import shutil
from pathlib import Path

from i18n import get_translator


def load_legacy_skill_names(config_path: Path) -> list[str]:
    """从配置文件读取 legacy skill 名单。"""
    if not config_path.exists():
        return []

    try:
        import yaml
    except ImportError as exc:
        raise RuntimeError("缺少 PyYAML 依赖，请先运行 `python3 -m pip install pyyaml`") from exc

    try:
        config = yaml.safe_load(config_path.read_text(encoding="utf-8")) or {}
    except Exception as exc:
        raise RuntimeError(f"配置文件解析失败: {exc}") from exc

    raw_names = config.get("legacy_skill_names", [])
    if raw_names is None:
        return []
    if not isinstance(raw_names, list):
        raise RuntimeError("配置项 legacy_skill_names 必须是列表")

    legacy_skill_names: list[str] = []
    seen: set[str] = set()
    for item in raw_names:
        name = str(item).strip()
        if not name or name in seen:
            continue
        legacy_skill_names.append(name)
        seen.add(name)
    return legacy_skill_names


def _remove_path(path: Path) -> None:
    if path.is_symlink() or path.is_file():
        path.unlink()
        return
    shutil.rmtree(path)


def remove_legacy_skills(
    *,
    target_root: Path,
    legacy_skill_names: list[str],
    dry_run: bool,
    active_skill_names: set[str] | None = None,
) -> list[str]:
    """删除目标平台目录中已弃用的 skill 目录。"""
    t = get_translator()
    messages: list[str] = []
    active_names = {name for name in (active_skill_names or set()) if name}

    for skill_name in legacy_skill_names:
        if skill_name in active_names:
            continue

        legacy_path = target_root / skill_name
        if not legacy_path.exists() and not legacy_path.is_symlink():
            continue

        if dry_run:
            messages.append(f"{t.get('dry_run_prefix')}remove legacy skill: {legacy_path}")
            continue

        _remove_path(legacy_path)
        messages.append(t.removed_legacy_skill(path=legacy_path))

    return messages


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="删除 Codex/Claude Code 中已弃用的 legacy skills。")
    parser.add_argument("--config", type=Path, default=Path(__file__).resolve().parents[1] / "config.yaml", help="legacy skill 配置文件路径")
    parser.add_argument("--codex", action="store_true", help="仅清理 Codex")
    parser.add_argument("--claude", action="store_true", help="仅清理 Claude Code")
    parser.add_argument("--dry-run", action="store_true", help="仅预览，不实际删除")
    args = parser.parse_args(argv)

    try:
        legacy_skill_names = load_legacy_skill_names(args.config)
    except RuntimeError as exc:
        print(f"错误: {exc}")
        return 1

    if not legacy_skill_names:
        print("未配置 legacy_skill_names，无需清理。")
        return 0

    install_codex = args.codex or (not args.codex and not args.claude)
    install_claude = args.claude or (not args.codex and not args.claude)

    targets: list[tuple[str, Path]] = []
    home = Path.home()
    if install_codex:
        targets.append(("CODEX", home / ".codex" / "skills"))
    if install_claude:
        targets.append(("CLAUDE", home / ".claude" / "skills"))

    removed_any = False
    for label, target_root in targets:
        print(f"\n[{label}] {target_root}")
        messages = remove_legacy_skills(
            target_root=target_root,
            legacy_skill_names=legacy_skill_names,
            dry_run=args.dry_run,
        )
        if not messages:
            print("未发现需要清理的 legacy skills。")
            continue
        removed_any = True
        for message in messages:
            print(message)

    if not removed_any:
        print("\n所有目标平台都没有发现 legacy skills。")

    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Fast remote skill version checker and updater.

Only remote installation is affected. Local-source installation remains owned by
install.py and keeps its existing MD5-based behavior.
"""
from __future__ import annotations

import argparse
import json
import re
import subprocess
import sys
import tempfile
from concurrent.futures import ThreadPoolExecutor
import urllib.parse
import urllib.error
import urllib.request
from dataclasses import dataclass
from pathlib import Path

SCRIPT_DIR = Path(__file__).resolve().parent
SKILL_ROOT = SCRIPT_DIR.parent
CONFIG_PATH = SKILL_ROOT / "config.yaml"
DEFAULT_SOURCE_FILTER = "huangwb8"
TARGETS = {"codex": Path.home() / ".codex" / "skills", "claude": Path.home() / ".claude" / "skills"}


@dataclass(frozen=True)
class RemoteSkill:
    source: dict[str, str]
    name: str
    version: tuple[int, ...]
    remote_path: str


def parse_version(value: str | None) -> tuple[int, ...] | None:
    """Parse numeric dotted versions without adding a runtime dependency."""
    if not value:
        return None
    match = re.fullmatch(r"\s*v?(\d+(?:\.\d+)*)\s*", value)
    return tuple(int(part) for part in match.group(1).split(".")) if match else None


def version_is_newer(remote: tuple[int, ...], local: tuple[int, ...] | None) -> bool:
    if local is None:
        return True
    width = max(len(remote), len(local))
    return (remote + (0,) * (width - len(remote))) > (local + (0,) * (width - len(local)))


def _config_value(text: str, key: str) -> str | None:
    match = re.search(rf"^\s*{re.escape(key)}:\s*[\"']?([^\"'\n#]+?)[\"']?\s*$", text, re.MULTILINE)
    return match.group(1).strip() if match else None


def load_sources(config_path: Path = CONFIG_PATH) -> list[dict[str, str]]:
    """Read the small remote_sources contract without requiring PyYAML."""
    text = config_path.read_text(encoding="utf-8")
    sources: list[dict[str, str]] = []
    current: dict[str, str] | None = None
    in_sources = False
    for line in text.splitlines():
        if line.strip() == "remote_sources:":
            in_sources = True
            continue
        if not in_sources:
            continue
        if line and not line[0].isspace() and line.strip().endswith(":"):
            break
        item = re.match(r"^\s*-\s+id:\s*[\"']?([^\"'\n]+?)[\"']?\s*$", line)
        if item:
            if current and current.get("url"):
                sources.append(current)
            current = {"id": item.group(1).strip()}
            continue
        if current:
            field = re.match(r"^\s+(url|branch|skills_path|name):\s*[\"']?([^\"'\n]+?)[\"']?\s*$", line)
            if field:
                current[field.group(1)] = field.group(2).strip()
    if current and current.get("url"):
        sources.append(current)
    return sources


def github_tree(source: dict[str, str]) -> list[str]:
    parsed = urllib.parse.urlparse(source["url"])
    parts = [part for part in parsed.path.split("/") if part]
    if parsed.netloc.lower() != "github.com" or len(parts) < 2:
        raise ValueError(f"unsupported remote source: {source['url']}")
    owner, repo = parts[0], parts[1].removesuffix(".git")
    branch = urllib.parse.quote(source.get("branch", "main"), safe="")
    api_url = f"https://api.github.com/repos/{owner}/{repo}/git/trees/{branch}?recursive=1"
    request = urllib.request.Request(api_url, headers={"Accept": "application/vnd.github+json", "User-Agent": "bensz-skill-updater"})
    try:
        with urllib.request.urlopen(request, timeout=20) as response:
            payload = json.load(response)
        return [item["path"] for item in payload.get("tree", []) if item.get("type") == "blob"]
    except urllib.error.HTTPError:
        # GitHub's unauthenticated API is rate-limited. Git's protocol is the
        # authoritative fallback and avoids a second API dependency.
        with tempfile.TemporaryDirectory(prefix="bensz-skill-tree-") as temp:
            repo = Path(temp) / "repo"
            subprocess.run(
                ["git", "clone", "--depth", "1", "--filter=blob:none", "--no-checkout",
                 "--branch", source.get("branch", "main"), source["url"], str(repo)],
                check=True, capture_output=True, text=True, timeout=120,
            )
            result = subprocess.run(
                ["git", "-C", str(repo), "ls-tree", "-r", "--name-only", "HEAD"],
                check=True, capture_output=True, text=True, timeout=30,
            )
            return result.stdout.splitlines()


def raw_url(source: dict[str, str], path: str) -> str:
    parsed = urllib.parse.urlparse(source["url"])
    parts = [part for part in parsed.path.split("/") if part]
    owner, repo = parts[0], parts[1].removesuffix(".git")
    branch = urllib.parse.quote(source.get("branch", "main"), safe="")
    quoted_path = "/".join(urllib.parse.quote(part, safe="") for part in path.split("/"))
    return f"https://raw.githubusercontent.com/{owner}/{repo}/{branch}/{quoted_path}"


def mirror_raw_url(source: dict[str, str], path: str) -> str | None:
    mirror = source.get("mirror_url", "").rstrip("/")
    if not mirror:
        return None
    return f"{mirror}/raw/{source.get('branch', 'main')}/{path}"


def fetch_text(url: str) -> str:
    request = urllib.request.Request(url, headers={"User-Agent": "bensz-skill-updater"})
    with urllib.request.urlopen(request, timeout=20) as response:
        return response.read().decode("utf-8")


def fetch_preferred_text(source: dict[str, str], path: str) -> str:
    """Use a configured mirror first; fall back to GitHub on any mirror error."""
    mirror = mirror_raw_url(source, path)
    if mirror:
        try:
            return fetch_text(mirror)
        except Exception:
            pass
    return fetch_text(raw_url(source, path))


def _fetch_version(source: dict[str, str], path: str) -> tuple[str, tuple[int, ...] | None]:
    text = fetch_preferred_text(source, path)
    return path, parse_version(_config_value(text, "version"))


def discover_remote_skills(source: dict[str, str], tree: list[str] | None = None) -> list[RemoteSkill]:
    root = source.get("skills_path", "skills").strip("/")
    prefix = f"{root}/" if root and root != "." else ""
    paths = tree if tree is not None else github_tree(source)
    result: list[RemoteSkill] = []
    for path in paths:
        if not path.startswith(prefix) or not path.endswith("/config.yaml"):
            continue
        relative = path[len(prefix):]
        parts = relative.split("/")
        if len(parts) != 2 or parts[1] != "config.yaml":
            continue
        name = parts[0]
        version = None
        try:
            version = parse_version(_config_value(fetch_preferred_text(source, path), "version"))
        except Exception:
            continue
        if version is not None:
            result.append(RemoteSkill(source, name, version, path))
    return result


def local_version(name: str, target_roots: list[Path]) -> tuple[int, ...] | None:
    versions: list[tuple[int, ...]] = []
    for root in target_roots:
        config = root / name / "config.yaml"
        if config.is_file():
            version = parse_version(_config_value(config.read_text(encoding="utf-8"), "version"))
            if version is not None:
                versions.append(version)
    return min(versions) if versions else None


def installer_path() -> Path:
    candidates = [root / "install-bensz-skills" / "scripts" / "install.py" for root in TARGETS.values()]
    for candidate in candidates:
        if candidate.is_file():
            return candidate
    return SCRIPT_DIR / "install.py"


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="Check and update newer versions from selected remote skill sources.")
    parser.add_argument("--source-contains", action="append", default=None, help="Only sources whose id/name/url contains this string; repeatable.")
    parser.add_argument("--all-sources", action="store_true", help="Disable the default source substring filter.")
    parser.add_argument("--skill", action="append", default=[], help="Only check these skill names; repeatable or comma-separated.")
    parser.add_argument("--check-only", action="store_true", help="Report newer skills without installing them.")
    parser.add_argument("--codex", action="store_true", help="Compare Codex installation only.")
    parser.add_argument("--claude", action="store_true", help="Compare Claude Code installation only.")
    args = parser.parse_args(argv)
    targets = [TARGETS[label] for label, selected in (("codex", args.codex), ("claude", args.claude)) if selected] or list(TARGETS.values())
    filters = [item.lower() for item in (args.source_contains or [DEFAULT_SOURCE_FILTER])]
    requested = {name.strip() for value in args.skill for name in value.split(",") if name.strip()}
    newer: dict[str, list[str]] = {}
    for source in load_sources():
        haystack = " ".join(source.get(key, "") for key in ("id", "name", "url")).lower()
        if not args.all_sources and not any(value in haystack for value in filters):
            continue
        try:
            tree = github_tree(source)
            # Fetch all version manifests concurrently: network latency, not CPU,
            # dominates this check. The tree itself remains sourced from GitHub.
            root = source.get("skills_path", "skills").strip("/")
            prefix = f"{root}/" if root and root != "." else ""
            candidates = []
            for path in tree:
                if path.startswith(prefix) and path.endswith("/config.yaml"):
                    relative = path[len(prefix):].split("/")
                    if len(relative) == 2 and relative[1] == "config.yaml":
                        candidates.append(path)
            skills = []
            with ThreadPoolExecutor(max_workers=min(16, max(1, len(candidates)))) as pool:
                for path, version in pool.map(lambda item: _fetch_version(source, item), candidates):
                    if version is not None:
                        skills.append(RemoteSkill(source, path[len(prefix):].split("/")[0], version, path))
        except Exception as exc:
            print(f"[WARN] {source.get('id', 'unknown')}: {exc}", file=sys.stderr)
            continue
        for skill in skills:
            if requested and skill.name not in requested:
                continue
            current = local_version(skill.name, targets)
            if version_is_newer(skill.version, current):
                newer.setdefault(skill.source["id"], []).append(skill.name)
                print(f"[UPDATE] {skill.name}: local {current or 'missing'} -> remote {'.'.join(map(str, skill.version))}")
    if not newer or args.check_only:
        return 0
    command = [sys.executable, str(installer_path()), "--remote", "--auto"]
    for source_id, names in newer.items():
        command.append(f"--{source_id}")
        for name in sorted(set(names)):
            command.extend(["--skill", name])
    if args.codex:
        command.append("--codex")
    elif args.claude:
        command.append("--claude")
    return subprocess.run(command, check=False).returncode


if __name__ == "__main__":
    raise SystemExit(main())

"""install-bensz-skills scripts package."""

from .install import main  # noqa: F401

__all__ = ["main"]