#!/usr/bin/env python3 """Langtianzhi Skills installer. Python 3.9+, standard library only.""" import argparse import hashlib import io import json import os from pathlib import Path, PurePosixPath import re import shutil import stat import sys import tempfile from datetime import datetime, timezone from urllib.parse import urljoin, urlparse from urllib.request import Request, build_opener, HTTPRedirectHandler import zipfile DEFAULT_SITE = "https://skills.langtianzhi.com/" MAX_DOWNLOAD = 50 * 1024 * 1024 IDENTIFIER = re.compile(r"[a-z0-9][a-z0-9-]{0,79}\Z") STATE_NAME = ".langtianzhi-skills.json" class SkillError(Exception): pass def origin(url): parsed = urlparse(url) return parsed.scheme, parsed.hostname, parsed.port def valid_site(url): parsed = urlparse(url) if parsed.username or parsed.password or parsed.query or parsed.fragment: raise SkillError("站点地址不能包含密码、查询参数或片段。") if parsed.scheme != "https" and not ( parsed.scheme == "http" and parsed.hostname in {"127.0.0.1", "localhost", "::1"} ): raise SkillError("仅支持 HTTPS；本地预览可以使用 localhost HTTP。") if not parsed.hostname: raise SkillError("站点地址无效。") return url.rstrip("/") + "/" class SameOriginRedirect(HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): if origin(req.full_url) != origin(newurl): raise SkillError("拒绝跨站重定向，请检查技能站地址。") return super().redirect_request(req, fp, code, msg, headers, newurl) def fetch(site, path, limit=MAX_DOWNLOAD): address = urljoin(site, path) if origin(address) != origin(site): raise SkillError("下载地址必须位于同一个技能站。") request = Request(address, headers={"User-Agent": "LangtianzhiSkills/1.0"}) with build_opener(SameOriginRedirect()).open(request, timeout=30) as response: if int(response.headers.get("Content-Length", "0")) > limit: raise SkillError("下载文件超出大小限制。") data = response.read(limit + 1) if len(data) > limit: raise SkillError("下载文件超出大小限制。") return data def get_catalog(site): data = json.loads(fetch(site, "catalog.json", 5 * 1024 * 1024)) if data.get("schema_version") != 1 or not isinstance(data.get("skills"), list): raise SkillError("不支持的技能目录格式。") seen = set() for skill in data["skills"]: slug = skill.get("id", "") if not IDENTIFIER.fullmatch(slug) or slug in seen: raise SkillError("目录包含无效或重复的技能名称。") seen.add(slug) return data def tree_hashes(directory): result = {} for item in sorted(directory.rglob("*")): if item.is_symlink(): raise SkillError("本地技能包含符号链接，已停止更新。") if item.is_file(): result[item.relative_to(directory).as_posix()] = hashlib.sha256(item.read_bytes()).hexdigest() return result def extract_checked(data, slug, target): """Validate the full archive before creating even one file.""" members = [] seen = set() total = 0 with zipfile.ZipFile(io.BytesIO(data)) as archive: if len(archive.infolist()) > 2000: raise SkillError("技能包包含过多文件。") for entry in archive.infolist(): path = PurePosixPath(entry.filename) if ("\\" in entry.filename or path.is_absolute() or ".." in path.parts or not path.parts or path.parts[0] != slug or ":" in entry.filename): raise SkillError("技能包包含非法路径。") mode = entry.external_attr >> 16 if stat.S_ISLNK(mode) or (stat.S_IFMT(mode) not in {0, stat.S_IFREG, stat.S_IFDIR}): raise SkillError("技能包包含符号链接或特殊文件。") if entry.is_dir(): continue relative = Path(*path.parts[1:]) key = relative.as_posix().casefold() if not relative.parts or key in seen: raise SkillError("技能包包含重复文件名。") if any(part.endswith((".", " ")) or part.upper().split(".")[0] in { "CON", "PRN", "AUX", "NUL", *("COM" + str(i) for i in range(1, 10)), *("LPT" + str(i) for i in range(1, 10)), } for part in relative.parts): raise SkillError("技能包包含不兼容 Windows 的文件名。") seen.add(key) total += entry.file_size if total > MAX_DOWNLOAD: raise SkillError("技能包解压后的体积超出限制。") members.append((entry, relative, mode)) if "skill.md" not in seen: raise SkillError("技能包缺少 SKILL.md。") for entry, relative, mode in members: output = target / relative output.parent.mkdir(parents=True, exist_ok=True) output.write_bytes(archive.read(entry)) output.chmod(0o755 if mode & 0o111 else 0o644) def read_state(target): file = target / STATE_NAME if file.is_symlink(): raise SkillError("安装记录不能是符号链接。") state = json.loads(file.read_text()) if file.exists() else {"schema_version": 1, "skills": {}} if state.get("schema_version") != 1 or not isinstance(state.get("skills"), dict): raise SkillError("本地安装记录格式错误。") return state def write_state(target, state): fd, name = tempfile.mkstemp(prefix=".skills-state-", dir=target) try: with os.fdopen(fd, "w") as handle: json.dump(state, handle, ensure_ascii=False, indent=2) handle.flush() os.fsync(handle.fileno()) os.replace(name, target / STATE_NAME) finally: Path(name).unlink(missing_ok=True) def install(site, skill, target, version=None, replace=False): slug = skill["id"] if not IDENTIFIER.fullmatch(slug): raise SkillError("技能名称无效。") releases = skill.get("releases", []) release = next((item for item in releases if item["version"] == (version or skill["version"])), None) if release is None: raise SkillError("未找到指定版本。") if not re.fullmatch(r"[0-9a-f]{64}", release.get("sha256", "")): raise SkillError("技能包缺少有效的 SHA-256 摘要。") data = fetch(site, release["archive"]) if hashlib.sha256(data).hexdigest() != release["sha256"]: raise SkillError("下载校验失败；本地技能未变更。") target = target.expanduser() if target.is_symlink(): raise SkillError("目标根目录不能是符号链接，请使用真实目录。") target.mkdir(parents=True, exist_ok=True) lock = target / ".langtianzhi-install.lock" try: lock_handle = lock.open("x") except FileExistsError: raise SkillError("有另一个安装任务正在运行；若已退出，请检查后移除安装锁。") from None with lock_handle: try: state = read_state(target) destination = target / slug if destination.is_symlink(): raise SkillError("目标技能目录不能是符号链接。") previous = state["skills"].get(slug) if destination.exists(): if not destination.is_dir(): raise SkillError("目标位置已存在同名文件。") modified = not previous or previous.get("site") != site or tree_hashes(destination) != previous.get("files") if modified and not replace: raise SkillError(f"{slug} 已存在或有本地修改。保留原文件；确认需要替换时增加 --replace（先备份）。") if previous and previous.get("version") == release["version"] and not modified: print(f"✓ {slug} 已是指定版本 {release['version']}") return # Detect a legacy installation to avoid duplicate skills in Codex selectors. legacy = Path.home() / ".codex" / "skills" / slug if target.resolve() == (Path.home() / ".agents" / "skills").resolve() and legacy.exists() and not destination.exists(): raise SkillError(f"发现旧目录中的同名技能：{legacy}。请先整理原有安装，避免重复加载。") backup = None with tempfile.TemporaryDirectory(prefix=".ltz-stage-", dir=target.parent) as work: stage = Path(work) / slug stage.mkdir() extract_checked(data, slug, stage) file_hashes = tree_hashes(stage) if destination.exists(): backup_root = target.parent / "skills-backups" backup_root.mkdir(exist_ok=True) backup = backup_root / (slug + "-" + datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ")) os.replace(destination, backup) try: os.replace(stage, destination) state["skills"][slug] = { "site": site, "version": release["version"], "sha256": release["sha256"], "files": file_hashes, } write_state(target, state) except Exception: if destination.exists(): shutil.rmtree(destination) if backup is not None: os.replace(backup, destination) raise print(f"✓ 已安装 {slug} · {release['version']}") if backup: print(f" 原版本备份：{backup}") finally: lock.unlink(missing_ok=True) def main(argv=None): parser = argparse.ArgumentParser(description="朗天智 Skills：从自己的技能站安装与更新完整技能包") parser.add_argument("--site", default=DEFAULT_SITE, help="技能站的 HTTPS 地址") parser.add_argument("--target", type=Path, default=Path.home() / ".agents" / "skills", help="安装目录") commands = parser.add_subparsers(dest="command", required=True) commands.add_parser("list", help="列出站点中的技能") add = commands.add_parser("install", help="安装技能；不会执行包内脚本") add.add_argument("ids", nargs="+") add.add_argument("--version", help="安装指定版本") add.add_argument("--replace", action="store_true", help="备份后替换已有或修改过的目录") commands.add_parser("sync", help="更新由本站安装的技能；保留本地修改") commands.add_parser("status", help="查看本机安装记录") args = parser.parse_args(argv) site = valid_site(args.site) if args.command == "status": entries = read_state(args.target.expanduser())["skills"] if not entries: print("尚未通过此工具安装技能。") for slug, entry in entries.items(): folder = args.target.expanduser() / slug condition = "已修改" if folder.exists() and tree_hashes(folder) != entry["files"] else ("正常" if folder.exists() else "已移除") print(f"{slug:32} {entry['version']} {condition}") return catalog = get_catalog(site) skills = {skill["id"]: skill for skill in catalog["skills"]} if args.command == "list": for skill in skills.values(): print(f"{skill['id']:32} {skill['title']} · {skill['version']}") return selected = args.ids if args.command == "install" else [ slug for slug, item in read_state(args.target.expanduser())["skills"].items() if item["site"] == site ] if args.command == "install" and args.version and len(selected) != 1: raise SkillError("指定版本时请一次安装一个技能。") unknown = [slug for slug in selected if slug not in skills] if unknown: raise SkillError("站点中不存在这些技能：" + ", ".join(unknown)) if not selected: print("没有需要同步的技能。") for slug in selected: install(site, skills[slug], args.target, getattr(args, "version", None), getattr(args, "replace", False)) if selected: print("安装完成。Codex 会自动检测；未显示时请重启。账号授权与外部工具按技能说明单独配置。") if __name__ == "__main__": try: main() except (SkillError, OSError, ValueError, zipfile.BadZipFile) as error: print(f"错误：{error}", file=sys.stderr) sys.exit(1)