Hugging Face's logo

Spaces:
zlcy
/
study
Paused

App Files Community
study / services /skill_service.py
xiaolongmr
fix: harden HF persistence and runtime workflows
34f333f
Raw History Blame Contribute Delete
110 kB
from __future__ import annotations
import copy
import io
import ipaddress
from functools import wraps
import hashlib
import json
import mimetypes
import os
from pathlib import Path, PurePosixPath
import re
import secrets
import shutil
import socket
import subprocess
import sys
import tempfile
import threading
import time
from typing import Any
from urllib.parse import urlparse
import uuid
import zipfile
import httpx
from services.config import BASE_DIR, DATA_DIR
MAX_ARCHIVE_BYTES = 25 * 1024 * 1024
MAX_FILE_BYTES = 8 * 1024 * 1024
MAX_FILES = 500
# An administrator-built export bundle carries every Skill together with its
# showcase media, so it is orders of magnitude larger than a Skill downloaded
# from a repository. Only the administrator-only import endpoint uses these
# relaxed limits; public installs keep the tight ones above.
MAX_IMPORT_ARCHIVE_BYTES = 256 * 1024 * 1024
MAX_IMPORT_FILES = 5000
MAX_RESOURCE_BYTES = 512 * 1024
MAX_SCRIPT_ARGS = 48
MAX_SCRIPT_ARG_CHARS = 2048
MAX_SCRIPT_OUTPUT_BYTES = 25 * 1024 * 1024
MEDIA_PARSER_OUTPUT_BYTES = max(
MAX_SCRIPT_OUTPUT_BYTES,
int(os.environ.get("CHATGPT2API_MEDIA_PARSER_MAX_BYTES", 512 * 1024 * 1024)),
)
MEDIA_PARSER_ARTIFACT_TTL_SECONDS = max(
900,
int(os.environ.get("CHATGPT2API_MEDIA_PARSER_TTL_SECONDS", 6 * 60 * 60)),
)
MEDIA_PARSER_OWNER_STORAGE_BYTES = max(
MEDIA_PARSER_OUTPUT_BYTES,
int(os.environ.get("CHATGPT2API_MEDIA_PARSER_OWNER_STORAGE_BYTES", 1024 * 1024 * 1024)),
)
MAX_SCRIPT_OUTPUT_FILES = 24
MAX_SCRIPT_STDIO_CHARS = 20000
DEFAULT_SCRIPT_TIMEOUT_SECONDS = 45
MAX_SCRIPT_TIMEOUT_SECONDS = 120
SKILLS_DIR = DATA_DIR / "skills"
BUNDLED_SKILLS_DIR = BASE_DIR / "bundled_skills"
BUNDLED_SKILL_VERSION_FILE = ".bundled-version"
SKILL_RUNS_DIR = DATA_DIR / "files" / "skills"
SAFE_NAME_RE = re.compile(r"[^a-zA-Z0-9._-]+")
FRONTMATTER_RE = re.compile(r"^---\s*\n(?P<body>.*?)\n---\s*(?:\n|$)", re.DOTALL)
TEXT_RESOURCE_EXTENSIONS = {
".md", ".mdx", ".txt", ".json", ".jsonl", ".yaml", ".yml", ".toml", ".ini",
".py", ".js", ".mjs", ".cjs", ".ts", ".tsx", ".css", ".html", ".svg",
}
SKILL_PRESENTATION_FILE = "skill-ui.json"
SKILL_PRESENTATION_OVERRIDE_FILE = ".presentation-overrides.json"
SKILL_PRESENTATION_ASSET_DIR = ".presentation-assets"
# Updating a Skill replaces its directory wholesale. Before that happens the
# previous copy is kept so an administrator can roll back a bad release.
SKILL_BACKUP_DIR = ".backups"
MAX_SKILL_BACKUPS = 3
SKILL_EXPORT_FORMAT = "chatsam-skills-export"
SKILL_EXPORT_FORMAT_VERSION = 1
EXPORT_MANIFEST_FILE = "manifest.json"
# Administrator state that survives an overwrite: script permissions, the
# showcase overrides, uploaded showcase media and the user-facing visibility.
PRESERVED_STATE_FILES = (
".permissions.json",
SKILL_PRESENTATION_OVERRIDE_FILE,
".visibility.json",
)
PRESERVED_STATE_DIRS = (SKILL_PRESENTATION_ASSET_DIR,)
SKILL_WORKFLOW_FILE = "skill-workflow.json"
PRESENTATION_SOURCE_FILES = {
SKILL_PRESENTATION_FILE,
SKILL_PRESENTATION_OVERRIDE_FILE,
SKILL_WORKFLOW_FILE,
}
PRESENTATION_SUMMARY_KEYS = (
"version",
"title",
"description",
"category",
"tags",
"capabilities",
"origin",
"cover",
"asset_count",
"example_count",
"input_count",
"output_count",
"workflow_count",
)
PRESENTATION_ASSET_EXTENSIONS = {
".png", ".jpg", ".jpeg", ".webp", ".gif", ".svg",
".html", ".htm", ".pdf", ".mp4", ".webm", ".mp3", ".wav",
}
PRESENTATION_MEDIA_ROOTS = {"assets", "examples", "example", "demo", "demos", "preview", "previews"}
SCRIPT_EXTENSIONS = {".py", ".js", ".mjs", ".cjs"}
USER_GUIDE_DOC_RE = re.compile(
r"^(?:readme|usage|quickstart|getting[-_ ]started|user[-_ ]guide|how[-_ ]to)(?:[._-].*)?\.(?:md|mdx|txt)$",
re.IGNORECASE,
)
GUIDE_CACHE_FILE = ".user-guide-cache.json"
SKILL_FAVORITES_FILE = "skill-favorites.json"
# Caches and build leftovers add nothing to an export bundle.
EXPORT_EXCLUDED_NAMES = {GUIDE_CACHE_FILE, ".DS_Store"}
EXPORT_EXCLUDED_PARTS = {"__pycache__"}
# A Skill package is not always a single directory. Repositories keep runnable
# code beside the Skill folder, or ship it in a monorepo package, and describe
# the layout in Markdown instead of in SKILL.md. These rules decide which
# documents are read while deciding what belongs to the package.
PACKAGE_DOC_RE = re.compile(
r"^(?:readme|usage|quickstart|getting[-_ ]started|user[-_ ]guide)(?:[._-].*)?\.(?:md|mdx|txt)$",
re.IGNORECASE,
)
PACKAGE_DOCUMENT_SUFFIXES = {".md", ".mdx", ".txt"}
MAX_PACKAGE_DOCUMENTS = 40
# ``scripts/`` already has a fixed meaning inside a Skill — it is where the
# runtime looks for controlled scripts. A repository-level ``scripts/`` holds
# development tooling, so re-homing it would produce ``scripts/scripts/`` and mix
# unrelated files into the Skill's own script list.
RESERVED_PACKAGE_ROOT_NAMES = {"scripts"}
class _KeyedRunLocks:
"""Serialize only duplicate idempotent runs inside this API process."""
def __init__(self) -> None:
self._guard = threading.Lock()
self._locks: dict[str, tuple[threading.Lock, int]] = {}
def acquire(self, key: str) -> threading.Lock:
with self._guard:
lock, references = self._locks.get(key, (threading.Lock(), 0))
self._locks[key] = (lock, references + 1)
lock.acquire()
return lock
def release(self, key: str, lock: threading.Lock) -> None:
lock.release()
with self._guard:
current, references = self._locks.get(key, (lock, 1))
if current is lock and references <= 1:
self._locks.pop(key, None)
elif current is lock:
self._locks[key] = (lock, references - 1)
_SCRIPT_RUN_LOCKS = _KeyedRunLocks()
def _serialize_duplicate_runs(function):
@wraps(function)
def wrapped(self, skill_id, identity, *args, **kwargs):
client_run_id = str(kwargs.get("client_run_id") or "").strip()
if not client_run_id:
return function(self, skill_id, identity, *args, **kwargs)
owner = _safe_name(str(identity.get("id") or "user"), "user")
key = hashlib.sha256(f"{owner}:{skill_id}:{client_run_id}".encode("utf-8")).hexdigest()
lock = _SCRIPT_RUN_LOCKS.acquire(key)
try:
return function(self, skill_id, identity, *args, **kwargs)
finally:
_SCRIPT_RUN_LOCKS.release(key, lock)
return wrapped
class SkillInstallError(ValueError):
pass
def _safe_name(value: str, fallback: str = "skill") -> str:
normalized = SAFE_NAME_RE.sub("-", value.strip()).strip("-._").lower()
return (normalized or fallback)[:80]
def _frontmatter(skill_file: Path) -> dict[str, str]:
try:
text = skill_file.read_text(encoding="utf-8")
except (OSError, UnicodeError) as exc:
raise SkillInstallError(f"无法读取 {skill_file.name}: {exc}") from exc
match = FRONTMATTER_RE.match(text)
values: dict[str, str] = {}
if match:
for raw_line in match.group("body").splitlines():
key, separator, value = raw_line.partition(":")
if separator and key.strip() in {
"name", "description", "author", "version", "homepage",
"repository", "source", "source_url",
}:
values[key.strip()] = value.strip().strip("\"'")
values.setdefault("name", skill_file.parent.name)
values.setdefault("description", "")
return values
def _presentation_category(skill_id: str, name: str, description: str) -> str:
text = f"{skill_id} {name} {description}".lower()
if re.search(r"\b(?:image|photo|poster|banner|logo|art|draw|paint|gif|wallpaper|holographic)\b|视觉|图片|海报|作图|封面|壁纸|摄影|插画|卡片", text):
return "视觉创作"
if re.search(r"\b(?:ui|ux|frontend|website|web)\b|界面|交互|网页", text):
return "界面设计"
if re.search(r"\b(?:document|pdf|ppt|slide|sheet|excel)\b|文档|表格|演示", text):
return "文档效率"
if re.search(r"\b(?:search|research)\b|搜索|调研", text):
return "搜索研究"
return "专业工作流"
def _asset_kind(path: Path, mime_type: str) -> str:
suffix = path.suffix.lower()
if mime_type.startswith("image/"):
return "image"
if mime_type.startswith("video/"):
return "video"
if mime_type.startswith("audio/"):
return "audio"
if suffix in {".html", ".htm"}:
return "html"
if suffix == ".pdf":
return "pdf"
return "file"
def _string_list(value: Any, *, limit: int = 16) -> list[str]:
if isinstance(value, str):
values = [value]
elif isinstance(value, list):
values = [str(item) for item in value if isinstance(item, (str, int, float))]
else:
return []
result: list[str] = []
for value_item in values:
clean = value_item.strip()
if clean and clean not in result:
result.append(clean[:160])
if len(result) >= limit:
break
return result
def _http_url(value: object) -> str:
clean = str(value or "").strip()
parsed = urlparse(clean)
return clean if parsed.scheme in {"http", "https"} and parsed.hostname else ""
def _require_public_remote_url(value: str) -> str:
clean = _http_url(value)
if not clean:
raise SkillInstallError("安装说明地址必须是有效的 http/https URL")
hostname = urlparse(clean).hostname or ""
try:
addresses = {item[4][0] for item in socket.getaddrinfo(hostname, None)}
except OSError as exc:
raise SkillInstallError("安装说明地址无法解析") from exc
for address in addresses:
ip = ipaddress.ip_address(address)
if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_multicast or ip.is_reserved or ip.is_unspecified:
raise SkillInstallError("安装说明地址不能指向本机或内网")
return clean
def _skill_origin(skill_dir: Path, metadata: dict[str, str], manifest: dict[str, Any]) -> dict[str, Any]:
install_metadata: dict[str, Any] = {}
try:
raw_install = json.loads((skill_dir / ".install.json").read_text(encoding="utf-8"))
if isinstance(raw_install, dict):
install_metadata = raw_install
except (OSError, UnicodeError, ValueError):
pass
package: dict[str, Any] = {}
try:
raw_package = json.loads((skill_dir / "package.json").read_text(encoding="utf-8"))
if isinstance(raw_package, dict):
package = raw_package
except (OSError, UnicodeError, ValueError):
pass
manifest_source = manifest.get("source")
source_config = manifest_source if isinstance(manifest_source, dict) else {}
package_repository = package.get("repository")
if isinstance(package_repository, dict):
package_repository = package_repository.get("url")
installed_source = str(install_metadata.get("source") or "").strip()
url = next((candidate for candidate in (
_http_url(source_config.get("url")),
_http_url(source_config.get("repository")),
_http_url(metadata.get("source_url")),
_http_url(metadata.get("repository")),
_http_url(metadata.get("homepage")),
_http_url(package_repository),
_http_url(package.get("homepage")),
_http_url(installed_source),
) if candidate), "")
if url.startswith("git+"):
url = url[4:]
url = url.removesuffix(".git")
package_author = package.get("author")
if isinstance(package_author, dict):
package_author = package_author.get("name")
author = str(
source_config.get("author")
or manifest.get("author")
or metadata.get("author")
or package_author
or ""
).strip()[:160]
github_match = re.match(r"https?://github\.com/([^/]+)/([^/#?]+)", url, re.IGNORECASE)
if not author and github_match:
author = github_match.group(1)
if github_match:
source_type = "github"
default_label = "GitHub 开源"
elif installed_source.startswith("upload:"):
source_type = "upload"
default_label = "本地上传"
elif url:
source_type = "url"
default_label = "外部来源"
else:
source_type = "local"
default_label = "本地 Skill"
return {
"type": source_type,
"label": str(source_config.get("label") or source_config.get("name") or default_label).strip()[:120],
"url": url,
"author": author,
"version": str(source_config.get("version") or manifest.get("version") or metadata.get("version") or package.get("version") or "").strip()[:80],
"open_source": bool(source_config.get("openSource", source_config.get("open_source", source_type == "github"))),
}
def _safe_relative_path(value: str) -> Path:
clean = str(value or "").strip().replace("\\", "/")
path = PurePosixPath(clean)
if not clean or path.is_absolute() or ".." in path.parts or any(not part for part in path.parts):
raise SkillInstallError("Skill 文件路径无效")
return Path(*path.parts)
def _safe_output_name(value: str) -> str:
name = Path(str(value or "").replace("\\", "/")).name
clean = re.sub(r"[^A-Za-z0-9._-]+", "-", name).strip(".-")
if not clean:
raise SkillInstallError("输出文件名无效")
return clean[:160]
def _is_text_resource(path: Path) -> bool:
return path.suffix.lower() in TEXT_RESOURCE_EXTENSIONS
def _script_kind(path: Path) -> str:
suffix = path.suffix.lower()
if suffix == ".py":
return "python"
if suffix in {".js", ".mjs", ".cjs"}:
return "node"
return ""
class _ScannedFile:
"""A single file discovered by one recursive walk of a Skill directory."""
__slots__ = ("path", "relative", "name", "suffix", "size", "mtime_ns")
def __init__(self, path: Path, relative: str, size: int, mtime_ns: int) -> None:
self.path = path
self.relative = relative
self.name = path.name
self.suffix = path.suffix.lower()
self.size = size
self.mtime_ns = mtime_ns
@property
def is_presentation_source(self) -> bool:
return self.relative in PRESENTATION_SOURCE_FILES
@property
def is_user_document(self) -> bool:
return bool(USER_GUIDE_DOC_RE.match(self.name.lower()))
@property
def is_presentation_asset(self) -> bool:
return self.suffix in PRESENTATION_ASSET_EXTENSIONS
def _scan_skill_files(skill_dir: Path) -> list[_ScannedFile]:
"""Walk a Skill directory exactly once and reuse the result everywhere.
``Path.rglob`` is a Python-level recursive scan and every call re-stats each
entry, so listing one Skill used to cost four full walks plus three reads of
the same SKILL.md. ``os.scandir`` reuses the stat data the OS already handed
back, and every consumer below shares this single result instead.
"""
found: list[_ScannedFile] = []
pending: list[Path] = [skill_dir]
while pending:
current = pending.pop()
try:
with os.scandir(current) as entries:
for entry in entries:
try:
if entry.is_dir(follow_symlinks=False):
pending.append(Path(entry.path))
continue
if not entry.is_file():
continue
info = entry.stat()
except OSError:
continue
path = Path(entry.path)
found.append(_ScannedFile(
path,
path.relative_to(skill_dir).as_posix(),
info.st_size,
info.st_mtime_ns,
))
except OSError:
continue
# Preserve the historical ordering: the guide fingerprint must stay stable
# for unchanged directories so existing cached guides remain valid.
found.sort(key=lambda item: item.path)
return found
def _guide_fingerprint(files: list[_ScannedFile]) -> str:
"""Cheap version of everything that can change a Skill's rendered surface."""
digest = hashlib.sha256()
for item in files:
if not (
item.relative == "SKILL.md"
or item.is_user_document
or item.is_presentation_source
or item.is_presentation_asset
):
continue
digest.update(f"{item.relative}:{item.size}:{item.mtime_ns}\n".encode("utf-8"))
return digest.hexdigest()[:20]
def _content_fingerprint(files: list[_ScannedFile]) -> str:
"""Version of every file in the directory, used to key the describe cache."""
digest = hashlib.sha256()
for item in files:
digest.update(f"{item.relative}:{item.size}:{item.mtime_ns}\n".encode("utf-8"))
return digest.hexdigest()[:20]
def _presentation_summary(presentation: dict[str, Any]) -> dict[str, Any]:
return {
key: copy.deepcopy(presentation[key])
for key in PRESENTATION_SUMMARY_KEYS
if key in presentation
}
def _guide_document_score(path: str) -> tuple[int, str]:
normalized = path.replace("\\", "/").lower()
name = normalized.rsplit("/", 1)[-1]
score = 20 if "/" in normalized else 0
if re.search(r"zh[-_.]?(?:cn|hans)|chinese|中文", name):
score -= 12
if name.startswith("readme"):
score -= 8
if re.match(r"^(?:usage|quickstart|getting[-_ ]started|user[-_ ]guide)", name):
score -= 6
if normalized.startswith("_package-docs/"):
score -= 4
return score, normalized
def _ancestor_directories(skill_file: Path, search_root: Path) -> list[Path]:
"""Directories from the Skill folder up to and including the search root."""
ancestors: list[Path] = []
search_root_resolved = search_root.resolve()
current = skill_file.parent.parent
while True:
try:
current.resolve().relative_to(search_root_resolved)
except ValueError:
break
ancestors.append(current)
if current.resolve() == search_root_resolved:
break
current = current.parent
return ancestors
def _package_documents(skill_file: Path, search_root: Path) -> list[Path]:
"""Return the documents that describe this Skill package.
SKILL.md is the authoritative protocol file, but repositories also document
how to run a Skill in ``references/*.md`` beside it and in the README above
it. All of those can name directories the Skill needs at run time, so all of
them are read when working out what the package contains.
"""
documents = [skill_file]
taken = 0
for scanned in _scan_skill_files(skill_file.parent):
if taken >= MAX_PACKAGE_DOCUMENTS:
break
if scanned.suffix not in PACKAGE_DOCUMENT_SUFFIXES:
continue
taken += 1
documents.append(scanned.path)
for ancestor in _ancestor_directories(skill_file, search_root):
try:
entries = sorted(ancestor.iterdir())
except OSError:
continue
for candidate in entries:
if candidate.is_file() and PACKAGE_DOC_RE.match(candidate.name):
documents.append(candidate)
return documents
def _referenced_package_roots(skill_file: Path, search_root: Path) -> list[Path]:
"""Find repository directories that belong to this Skill package.
Some repositories keep SKILL.md in a small definition directory while the
executable implementation lives beside it (for example
``media-parser/`` + ``astrbot_plugin_media_parser/``), and monorepo layouts
ship a companion service in a sibling package (``skills/holo-card/`` +
``packages/holo-card-api/``). Installing only the SKILL.md parent produces a
Skill that can explain commands but cannot run them, so every directory named
as ``dir/...`` by the package documents is included when it really exists
next to the Skill.
Copying an ancestor would drag the Skill folder into its own ``scripts``
directory, and copying a directory that holds another SKILL.md would install
a second, competing Skill — both are skipped.
"""
segments: list[str] = []
for document in _package_documents(skill_file, search_root):
try:
content = document.read_text(encoding="utf-8")[:MAX_RESOURCE_BYTES]
except (OSError, UnicodeError):
continue
for match in re.finditer(r"(?<![\w.:-])([A-Za-z0-9][A-Za-z0-9._-]*)/[A-Za-z0-9._-]+(?:/[A-Za-z0-9._-]+)*", content):
segment = match.group(1)
if segment not in segments and segment not in {skill_file.parent.name, "http", "https"}:
segments.append(segment)
if len(segments) >= 256:
break
if len(segments) >= 256:
break
skill_dir = skill_file.parent.resolve()
roots: list[Path] = []
search_root_resolved = search_root.resolve()
for current in _ancestor_directories(skill_file, search_root):
for segment in segments:
candidate = (current / segment).resolve()
try:
candidate.relative_to(search_root_resolved)
except ValueError:
continue
if not candidate.exists() or candidate in roots:
continue
if candidate.name in RESERVED_PACKAGE_ROOT_NAMES:
continue
if candidate == skill_dir or candidate in skill_dir.parents or skill_dir in candidate.parents:
continue
if next(candidate.rglob("SKILL.md"), None) is not None:
continue
roots.append(candidate)
return roots
def _validate_zip_member(info: zipfile.ZipInfo) -> PurePosixPath:
path = PurePosixPath(info.filename.replace("\\", "/"))
if path.is_absolute() or ".." in path.parts:
raise SkillInstallError(f"压缩包包含不安全路径：{info.filename}")
mode = info.external_attr >> 16
if mode and (mode & 0o170000) == 0o120000:
raise SkillInstallError(f"压缩包不允许符号链接：{info.filename}")
if info.file_size > MAX_FILE_BYTES:
raise SkillInstallError(f"文件过大：{info.filename}")
return path
def _is_export_excluded(relative: str) -> bool:
parts = relative.split("/")
if parts[-1] in EXPORT_EXCLUDED_NAMES or parts[-1].endswith(".pyc"):
return True
return any(part in EXPORT_EXCLUDED_PARTS for part in parts)
def _export_bundle_sources(extracted: Path) -> dict[str, dict[str, Any]]:
"""Return {skill_id: manifest entry} when the archive is an export bundle.
A bundle produced by ``export_skills`` is the unit of migration between
deployments: it carries each Skill's recorded origin so the importing
deployment can keep updating it from the same place. Importing must
therefore not relabel the Skill as a local upload — which only happens for
archives that actually declare the bundle format.
"""
path = extracted / EXPORT_MANIFEST_FILE
if not path.is_file():
return {}
try:
payload = json.loads(path.read_text(encoding="utf-8"))
except (OSError, UnicodeError, ValueError):
return {}
if not isinstance(payload, dict) or payload.get("format") != SKILL_EXPORT_FORMAT:
return {}
entries: dict[str, dict[str, Any]] = {}
for item in payload.get("skills") or []:
if isinstance(item, dict) and str(item.get("id") or "").strip():
entries[str(item["id"]).strip()] = item
return entries
def _restore_preserved_state(target: Path, backup: Path) -> None:
"""Copy administrator state from an overwrite snapshot into the fresh install.
A reinstall brings new code, but the administrator's decisions about the
Skill — which scripts may run, how it is presented, which media was uploaded
and whether end users see it — belong to the installation, not to the
package, so they are carried across.
"""
for name in PRESERVED_STATE_FILES:
source = backup / name
if source.is_file():
shutil.copy2(source, target / name)
for name in PRESERVED_STATE_DIRS:
source = backup / name
if not source.is_dir():
continue
destination = target / name
if destination.exists():
shutil.rmtree(destination, ignore_errors=True)
shutil.copytree(source, destination)
def _format_size_limit(limit: int) -> str:
return f"{limit // (1024 * 1024)}MB"
def _extract_archive(
payload: bytes,
destination: Path,
*,
subpath: str = "",
max_bytes: int = MAX_ARCHIVE_BYTES,
max_files: int = MAX_FILES,
) -> None:
if len(payload) > max_bytes:
raise SkillInstallError(f"ZIP 不能超过 {_format_size_limit(max_bytes)}")
try:
archive = zipfile.ZipFile(io.BytesIO(payload))
except zipfile.BadZipFile as exc:
raise SkillInstallError("上传内容不是有效 ZIP") from exc
infos = archive.infolist()
if subpath:
wanted = PurePosixPath(subpath).parts
infos = [
info for info in infos
if tuple(PurePosixPath(info.filename.replace("\\", "/")).parts[1:1 + len(wanted)]) == wanted
]
if not infos:
raise SkillInstallError(f"仓库中找不到技能目录：{subpath}")
if len(infos) > max_files:
raise SkillInstallError(f"ZIP 文件数量不能超过 {max_files}")
total_size = sum(info.file_size for info in infos)
if total_size > max_bytes:
raise SkillInstallError(f"ZIP 解压后不能超过 {_format_size_limit(max_bytes)}")
for info in infos:
relative = _validate_zip_member(info)
if not relative.parts:
continue
target = destination.joinpath(*relative.parts)
if info.is_dir():
target.mkdir(parents=True, exist_ok=True)
continue
target.parent.mkdir(parents=True, exist_ok=True)
with archive.open(info) as source, target.open("wb") as output:
shutil.copyfileobj(source, output)
def _github_download_candidates(url: str) -> list[tuple[str, str]]:
parsed = urlparse(url)
if parsed.netloc.lower() not in {"github.com", "www.github.com"}:
return [(url, "")]
parts = [part for part in parsed.path.split("/") if part]
if len(parts) < 2:
raise SkillInstallError("GitHub URL 必须包含 owner/repository")
owner, repository = parts[0], parts[1].removesuffix(".git")
if len(parts) >= 5 and parts[2] == "tree":
ref = parts[3]
subpath = "/".join(parts[4:])
return [(f"https://codeload.github.com/{owner}/{repository}/zip/{ref}", subpath)]
return [
(f"https://codeload.github.com/{owner}/{repository}/zip/refs/heads/main", ""),
(f"https://codeload.github.com/{owner}/{repository}/zip/refs/heads/master", ""),
]
class SkillService:
def __init__(
self,
root: Path = SKILLS_DIR,
runs_root: Path = SKILL_RUNS_DIR,
favorites_path: Path | None = None,
) -> None:
self.root = root
self.runs_root = runs_root
self.favorites_path = favorites_path or (
DATA_DIR / SKILL_FAVORITES_FILE
if root == SKILLS_DIR
else root.parent / SKILL_FAVORITES_FILE
)
self._favorites_lock = threading.RLock()
# Keyed by (resolved directory, guide version). The cache always holds the
# full presentation; summaries are projected from it so the list view and
# the detail view no longer build the same Skill twice.
self._presentation_cache: dict[tuple[str, str], dict[str, Any]] = {}
self._describe_cache: dict[tuple[str, str, bool], dict[str, Any]] = {}
self._run_cleanup_lock = threading.RLock()
self._last_run_cleanup_at = 0.0
self._media_link_lock = threading.RLock()
self._media_links: dict[str, dict[str, Any]] = {}
def sync_bundled_skills(self, source_root: Path = BUNDLED_SKILLS_DIR) -> list[str]:
"""Install the Skills shipped with the application into runtime storage.
Runtime storage may be a Hugging Face persistent volume, whereas the
bundled directory lives in the immutable application image. Refresh
the application-owned files on every deployment while retaining the
administrator-controlled visibility, permissions and presentation
overrides already stored on the volume.
"""
if not source_root.is_dir():
return []
self.root.mkdir(parents=True, exist_ok=True)
synced: list[str] = []
for source in sorted(path for path in source_root.iterdir() if path.is_dir()):
skill_file = source / "SKILL.md"
if not skill_file.is_file():
continue
metadata = _frontmatter(skill_file)
skill_id = _safe_name(metadata["name"], _safe_name(source.name))
target = self.root / skill_id
digest = hashlib.sha256()
for resource in sorted(path for path in source.rglob("*") if path.is_file()):
relative = resource.relative_to(source).as_posix()
digest.update(relative.encode("utf-8"))
digest.update(b"\0")
digest.update(resource.read_bytes())
digest.update(b"\0")
version = digest.hexdigest()
version_file = target / BUNDLED_SKILL_VERSION_FILE
if version_file.is_file() and version_file.read_text(encoding="utf-8").strip() == version:
continue
staging = self.root / f".{skill_id}.bundled-{os.getpid()}-{uuid.uuid4().hex[:6]}"
previous = self.root / f".{skill_id}.previous-{os.getpid()}-{uuid.uuid4().hex[:6]}"
try:
shutil.copytree(source, staging)
(staging / BUNDLED_SKILL_VERSION_FILE).write_text(version + "\n", encoding="utf-8")
if target.is_dir():
_restore_preserved_state(staging, target)
os.replace(target, previous)
os.replace(staging, target)
shutil.rmtree(previous, ignore_errors=True)
synced.append(skill_id)
except Exception:
shutil.rmtree(staging, ignore_errors=True)
if previous.exists() and not target.exists():
os.replace(previous, target)
raise
self._presentation_cache.clear()
self._describe_cache.clear()
return synced
def register_media_link(
self,
url: str,
*,
headers: dict[str, object] | None,
name: str,
media_type: str,
) -> dict[str, Any]:
clean_url = _require_public_remote_url(url)
allowed_headers = {"user-agent", "accept", "accept-language", "referer", "origin"}
clean_headers = {
str(key): str(value)
for key, value in (headers or {}).items()
if str(key).lower() in allowed_headers and str(value).strip()
}
token = secrets.token_urlsafe(32)
expires_at = time.time() + MEDIA_PARSER_ARTIFACT_TTL_SECONDS
with self._media_link_lock:
current = time.time()
self._media_links = {
key: value
for key, value in self._media_links.items()
if float(value.get("expires_at") or 0) > current
}
self._media_links[token] = {
"url": clean_url,
"headers": clean_headers,
"name": _safe_output_name(name),
"media_type": str(media_type or "application/octet-stream"),
"expires_at": expires_at,
}
return {
"name": _safe_output_name(name),
"type": str(media_type or "application/octet-stream"),
"size": 0,
"url": f"/api/skills/media/{token}",
"expires_at": expires_at,
"source_url": clean_url,
"streamed": True,
}
def resolve_media_link(self, token: str) -> dict[str, Any]:
clean_token = str(token or "").strip()
if not re.fullmatch(r"[A-Za-z0-9_-]{32,128}", clean_token):
raise SkillInstallError("媒体下载链接无效")
with self._media_link_lock:
record = self._media_links.get(clean_token)
if not record or float(record.get("expires_at") or 0) <= time.time():
self._media_links.pop(clean_token, None)
raise SkillInstallError("媒体下载链接已过期，请重新解析")
return dict(record)
def _media_stream_artifacts(self, stdout: str) -> list[dict[str, Any]]:
try:
payload = json.loads(stdout)
except (TypeError, ValueError):
return []
results = payload.get("results") if isinstance(payload, dict) else []
if not isinstance(results, list):
return []
artifacts: list[dict[str, Any]] = []
for result_index, result in enumerate(results):
if not isinstance(result, dict) or result.get("error"):
continue
title = _safe_name(str(result.get("title") or f"media-{result_index + 1}"), f"media-{result_index + 1}")
videos = result.get("videos") if isinstance(result.get("videos"), list) else []
images = result.get("images") if isinstance(result.get("images"), list) else []
for index, media in enumerate(videos):
if not isinstance(media, dict) or not media.get("direct_url"):
continue
artifacts.append(self.register_media_link(
str(media["direct_url"]),
headers=media.get("headers") if isinstance(media.get("headers"), dict) else {},
name=f"{title}-video-{index + 1}.mp4",
media_type="video/mp4",
))
for index, media in enumerate(images):
if not isinstance(media, dict) or not media.get("direct_url"):
continue
artifacts.append(self.register_media_link(
str(media["direct_url"]),
headers=media.get("headers") if isinstance(media.get("headers"), dict) else {},
name=f"{title}-image-{index + 1}.jpg",
media_type="image/jpeg",
))
return artifacts[:MAX_SCRIPT_OUTPUT_FILES]
def cleanup_expired_runs(self, *, force: bool = False, now: float | None = None) -> int:
"""Remove expired temporary Skill artifacts and idempotency records."""
current = time.time() if now is None else float(now)
with self._run_cleanup_lock:
monotonic_now = time.monotonic()
if not force and monotonic_now - self._last_run_cleanup_at < 60:
return 0
self._last_run_cleanup_at = monotonic_now
removed = 0
if self.runs_root.is_dir():
for marker in self.runs_root.glob("*/*/.expires-at"):
try:
expires_at = float(marker.read_text(encoding="utf-8").strip())
except (OSError, UnicodeError, ValueError):
continue
if expires_at > current:
continue
run_dir = marker.parent.resolve()
try:
run_dir.relative_to(self.runs_root.resolve())
except ValueError:
continue
shutil.rmtree(run_dir, ignore_errors=True)
removed += 1
cache_root = self.runs_root / ".runs"
if cache_root.is_dir():
for cache_path in cache_root.glob("*.json"):
try:
payload = json.loads(cache_path.read_text(encoding="utf-8"))
expires_at = float(payload.get("expires_at") or 0) if isinstance(payload, dict) else 0
except (OSError, UnicodeError, ValueError, TypeError):
continue
legacy_run_dir: Path | None = None
if isinstance(payload, dict) and payload.get("skill_id") == "media-parser" and not expires_at:
artifacts = payload.get("artifacts") if isinstance(payload.get("artifacts"), list) else []
first_url = next((str(item.get("url") or "") for item in artifacts if isinstance(item, dict)), "")
prefix = "/files/skills/"
if first_url.startswith(prefix):
relative = PurePosixPath(first_url[len(prefix):])
if len(relative.parts) >= 3 and ".." not in relative.parts:
candidate = (self.runs_root / Path(*relative.parts[:2])).resolve()
try:
candidate.relative_to(self.runs_root.resolve())
if candidate.is_dir():
legacy_run_dir = candidate
except ValueError:
legacy_run_dir = None
if legacy_run_dir is not None:
expires_at = legacy_run_dir.stat().st_mtime + MEDIA_PARSER_ARTIFACT_TTL_SECONDS
(legacy_run_dir / ".expires-at").write_text(str(expires_at), encoding="utf-8")
payload["expires_at"] = expires_at
cache_path.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")
if expires_at and expires_at <= current:
if legacy_run_dir is not None:
shutil.rmtree(legacy_run_dir, ignore_errors=True)
removed += 1
cache_path.unlink(missing_ok=True)
return removed
def _enforce_media_owner_quota(self, owner: str) -> int:
owner_root = (self.runs_root / owner).resolve()
try:
owner_root.relative_to(self.runs_root.resolve())
except ValueError:
return 0
if not owner_root.is_dir():
return 0
runs: list[tuple[float, Path, int]] = []
total = 0
for marker in owner_root.glob("*/.expires-at"):
run_dir = marker.parent
try:
size = sum(
path.stat().st_size
for path in run_dir.rglob("*")
if path.is_file() and path.name != ".expires-at"
)
created = run_dir.stat().st_mtime
except OSError:
continue
total += size
runs.append((created, run_dir, size))
removed = 0
for _created, run_dir, size in sorted(runs):
if total <= MEDIA_PARSER_OWNER_STORAGE_BYTES:
break
shutil.rmtree(run_dir, ignore_errors=True)
total -= size
removed += 1
return removed
def list(self) -> list[dict[str, Any]]:
self.root.mkdir(parents=True, exist_ok=True)
items: list[dict[str, Any]] = []
for skill_file in sorted(self.root.glob("*/SKILL.md")):
# Installations are assembled in a hidden staging directory and
# atomically renamed when complete. Never expose a partial Skill.
if skill_file.parent.name.startswith("."):
continue
skill_dir = skill_file.parent
files = _scan_skill_files(skill_dir)
metadata = _frontmatter(skill_file)
install_metadata = self._install_metadata(skill_dir)
permissions = self._permissions(skill_dir)
enabled = self._visibility(skill_dir)
guide_version = _guide_fingerprint(files)
presentation = self._presentation_for(
skill_dir,
metadata,
guide_version=guide_version,
files=files,
summary_only=True,
)
items.append({
"id": skill_dir.name,
"name": metadata["name"],
"description": metadata["description"],
"source": str(install_metadata.get("source") or "manual"),
"installed_at": str(install_metadata.get("installed_at") or ""),
"file_count": len(files),
"scripts_allowed": bool(permissions.get("allow_scripts")),
"allowed_scripts": list(permissions.get("allowed_scripts") or []),
"enabled": enabled,
"guide_version": guide_version,
"presentation_summary": presentation,
})
return items
@staticmethod
def _read_presentation_manifest(skill_dir: Path) -> dict[str, Any]:
"""Return skill-ui.json, or an empty config when it is absent or unusable."""
path = skill_dir / SKILL_PRESENTATION_FILE
if not path.is_file():
return {}
try:
value = json.loads(path.read_text(encoding="utf-8"))
except (OSError, UnicodeError, ValueError):
return {}
return value if isinstance(value, dict) else {}
@staticmethod
def _read_presentation_overrides(skill_dir: Path) -> dict[str, Any]:
"""Return the administrator overrides, ignoring unreadable or malformed files."""
path = skill_dir / SKILL_PRESENTATION_OVERRIDE_FILE
if not path.is_file():
return {}
try:
value = json.loads(path.read_text(encoding="utf-8"))
except (OSError, UnicodeError, ValueError):
return {}
return value if isinstance(value, dict) else {}
@staticmethod
def _merge_presentation_config(base: dict[str, Any], overrides: dict[str, Any]) -> dict[str, Any]:
result = dict(base)
for key, value in overrides.items():
if key == "display" and isinstance(value, dict):
current = result.get("display") if isinstance(result.get("display"), dict) else {}
result["display"] = {**current, **value}
else:
result[key] = value
return result
@staticmethod
def _safe_presentation_asset(skill_dir: Path, value: object) -> Path | None:
if not isinstance(value, str) or not value.strip():
return None
try:
relative = _safe_relative_path(value.strip())
except SkillInstallError:
return None
target = (skill_dir / relative).resolve()
try:
target.relative_to(skill_dir.resolve())
except ValueError:
return None
if not target.is_file() or target.stat().st_size > MAX_FILE_BYTES:
return None
return target
def _build_presentation(
self,
skill_dir: Path,
metadata: dict[str, str],
*,
files: list[_ScannedFile] | None = None,
) -> dict[str, Any]:
scanned = files if files is not None else _scan_skill_files(skill_dir)
manifest = self._read_presentation_manifest(skill_dir)
overrides = self._read_presentation_overrides(skill_dir)
manifest = self._merge_presentation_config(manifest, overrides)
display = manifest.get("display") if isinstance(manifest.get("display"), dict) else {}
origin = _skill_origin(skill_dir, metadata, manifest)
skill_id = skill_dir.name
name = str(display.get("title") or metadata.get("name") or skill_id).strip()
description = str(display.get("description") or metadata.get("description") or "").strip()
category = str(display.get("category") or _presentation_category(skill_id, name, description)).strip()
tags = _string_list(display.get("tags") or manifest.get("tags"), limit=12)
content = ""
try:
content = (skill_dir / "SKILL.md").read_text(encoding="utf-8")[:50000]
except (OSError, UnicodeError):
content = ""
searchable = f"{name}\n{description}\n{content}".lower()
capabilities = _string_list(manifest.get("capabilities"), limit=8)
inferred_capabilities: list[str] = []
if re.search(r"web[_ -]?search|联网|搜索", searchable):
inferred_capabilities.append("web_search")
if re.search(r"image[_ -]?(?:generation|edit)|生图|图生图|图片生成|图片编辑|poster|photo|logo|art", searchable):
inferred_capabilities.append("image")
if any((skill_dir / folder).is_dir() for folder in ("scripts",)):
inferred_capabilities.append("script")
if re.search(r"html|网页|interactive|交互", searchable):
inferred_capabilities.append("html")
if not capabilities:
capabilities = inferred_capabilities or ["chat"]
asset_entries: list[dict[str, Any]] = []
seen_assets: set[str] = set()
def add_asset(raw: object, *, default_role: str = "gallery") -> None:
item = raw if isinstance(raw, dict) else {"path": raw}
path_value = item.get("path") if isinstance(item, dict) else None
target = self._safe_presentation_asset(skill_dir, path_value)
if target is None or target.suffix.lower() not in PRESENTATION_ASSET_EXTENSIONS:
return
relative = target.relative_to(skill_dir.resolve()).as_posix()
if relative in seen_assets:
return
seen_assets.add(relative)
mime_type = mimetypes.guess_type(target.name)[0] or "application/octet-stream"
asset_entries.append({
"path": relative,
"name": str(item.get("title") or item.get("name") or target.stem).strip()[:120],
"description": str(item.get("description") or "").strip()[:500],
"type": mime_type,
"kind": _asset_kind(target, mime_type),
"size": target.stat().st_size,
"role": str(item.get("role") or default_role).strip()[:40],
})
explicit_cover = display.get("cover") or manifest.get("cover")
if explicit_cover:
add_asset({"path": explicit_cover, "role": "cover", "title": name}, default_role="cover")
raw_assets = manifest.get("assets")
if isinstance(raw_assets, dict):
raw_assets = raw_assets.get("gallery") or raw_assets.get("items") or []
if isinstance(raw_assets, list):
for item in raw_assets[:24]:
add_asset(item)
readme_documents = sorted(
(item for item in scanned if item.is_user_document),
key=lambda item: _guide_document_score(item.relative),
)[:4]
readme_text = ""
for readme in readme_documents:
if readme.size > MAX_RESOURCE_BYTES:
continue
try:
text_value = readme.path.read_text(encoding="utf-8")
except (OSError, UnicodeError):
continue
readme_text += f"\n{text_value[:12000]}"
for match in re.finditer(r"!\[[^\]]*\]\(([^)\s]+)(?:\s+[^)]*)?\)", text_value):
source = match.group(1).strip("<>\"'")
if re.match(r"^[a-z]+://", source, re.IGNORECASE):
continue
try:
relative_to_skill = (readme.path.parent / source).resolve().relative_to(skill_dir.resolve()).as_posix()
except (ValueError, OSError):
continue
add_asset(relative_to_skill)
if len(asset_entries) < 24:
for item in scanned:
if not item.is_presentation_asset:
continue
parts = PurePosixPath(item.relative).parts
if not parts or parts[0].lower() not in PRESENTATION_MEDIA_ROOTS:
continue
add_asset(item.relative)
if len(asset_entries) >= 24:
break
cover = next((item for item in asset_entries if item["role"] == "cover"), None)
if cover is None:
cover = next((item for item in asset_entries if item["kind"] == "image"), None)
if cover:
cover = {**cover, "role": "cover"}
raw_examples = manifest.get("examples") if isinstance(manifest.get("examples"), list) else []
examples: list[dict[str, Any]] = []
for index, raw in enumerate(raw_examples[:8]):
if not isinstance(raw, dict):
continue
prompt = str(raw.get("prompt") or "").strip()
if not prompt:
continue
examples.append({
"id": str(raw.get("id") or f"example-{index + 1}")[:80],
"title": str(raw.get("title") or f"示例 {index + 1}")[:120],
"description": str(raw.get("description") or "")[:500],
"prompt": prompt[:4000],
"assets": _string_list(raw.get("assets"), limit=8),
"attachment_hint": str(raw.get("attachmentHint") or raw.get("attachment_hint") or "")[:300],
})
if not examples:
example_source = readme_text or content
candidates = [
match.group(1).strip()
for match in re.finditer(r"```(?:text|prompt|markdown)?\s*\n([\s\S]{12,1200}?)\n```", example_source, re.IGNORECASE)
]
candidates.extend(match.group(0).strip() for match in re.finditer(rf"\${re.escape(skill_id)}\b[^\n]{{4,500}}", example_source, re.IGNORECASE))
for index, prompt in enumerate(candidates[:4]):
if not prompt.startswith(f"${skill_id}"):
prompt = f"${skill_id} {prompt}"
examples.append({"id": f"auto-{index + 1}", "title": f"示例 {index + 1}", "description": "", "prompt": prompt, "assets": [], "attachment_hint": ""})
if not examples:
examples.append({
"id": "starter",
"title": "从你的需求开始",
"description": description,
"prompt": f"${skill_id} 请根据我接下来提供的要求完成任务",
"assets": [],
"attachment_hint": "如任务依赖图片或文件，请一并上传。",
})
def normalize_specs(key: str, defaults: list[dict[str, Any]]) -> list[dict[str, Any]]:
raw_specs = manifest.get(key)
if not isinstance(raw_specs, list):
return defaults
result: list[dict[str, Any]] = []
for index, raw in enumerate(raw_specs[:12]):
if isinstance(raw, str):
raw = {"label": raw}
if not isinstance(raw, dict) or not str(raw.get("label") or raw.get("name") or "").strip():
continue
result.append({
"id": str(raw.get("id") or f"{key}-{index + 1}")[:80],
"label": str(raw.get("label") or raw.get("name"))[:120],
"type": str(raw.get("type") or "text")[:40],
"required": bool(raw.get("required", False)),
"description": str(raw.get("description") or "")[:500],
})
return result or defaults
default_inputs = [{"id": "request", "label": "任务要求", "type": "text", "required": True, "description": "说明目标、风格、尺寸或交付要求。"}]
input_searchable = f"{searchable}\n{readme_text.lower()}"
has_image_input = "image" in capabilities and bool(re.search(
r"supplied image|uploaded image|user[- ]provided image|source image|reference image|"
r"参考图|上传图片|上传照片|原图|照片",
input_searchable,
))
# Auto-discovered image inputs are suggestions, never hard gates. An
# administrator can explicitly mark an image input as required through
# skill-ui.json or the presentation overrides editor.
image_input_required = False
if has_image_input:
default_inputs.append({
"id": "image",
"label": "参考图片",
"type": "image",
"required": image_input_required,
"description": "上传需要处理或参考的图片。" if image_input_required else "如需基于现有画面处理，可上传参考图片。",
})
inputs = normalize_specs("inputs", default_inputs)
default_outputs: list[dict[str, Any]] = []
if "image" in capabilities:
default_outputs.append({"id": "image", "label": "图片结果", "type": "image", "required": False, "description": "可预览、下载并继续作为参考图使用。"})
if "html" in capabilities:
default_outputs.append({"id": "html", "label": "互动 HTML", "type": "html", "required": False, "description": "在安全沙箱中预览，可手动启用交互。"})
if "script" in capabilities:
default_outputs.append({"id": "file", "label": "文件产物", "type": "file", "required": False, "description": "由受控脚本生成的可下载文件。"})
if not default_outputs:
default_outputs.append({"id": "answer", "label": "结构化结果", "type": "text", "required": False, "description": "在对话中直接交付。"})
outputs = normalize_specs("outputs", default_outputs)
workflow: list[dict[str, Any]] = []
workflow_summary = ""
workflow_path = skill_dir / SKILL_WORKFLOW_FILE
if workflow_path.is_file():
try:
raw_workflow = json.loads(workflow_path.read_text(encoding="utf-8"))
workflow_summary = str(raw_workflow.get("summary") or "").strip()[:500] if isinstance(raw_workflow, dict) else ""
tasks = raw_workflow.get("tasks") if isinstance(raw_workflow, dict) else []
if not isinstance(tasks, list) or not tasks:
raise ValueError("tasks missing")
for index, task in enumerate(tasks[:32]):
if not isinstance(task, dict):
continue
workflow.append({
"id": str(task.get("id") or f"step-{index + 1}")[:80],
"title": str(task.get("title") or task.get("id") or f"步骤 {index + 1}")[:120],
"description": str(task.get("description") or task.get("prompt") or "")[:500],
"capability": str(task.get("capability") or ("image" if "image" in capabilities else "task"))[:40],
})
postprocess = raw_workflow.get("postprocess") if isinstance(raw_workflow, dict) else None
if isinstance(postprocess, dict) and postprocess.get("script"):
workflow.append({"id": "postprocess", "title": str(postprocess.get("title") or "整理最终产物")[:120], "description": "将前面步骤的结果组装为最终文件。", "capability": "script"})
except (OSError, UnicodeError, ValueError, TypeError):
# A malformed workflow file simply falls back to the plain Skill flow.
workflow = []
result = {
"version": 1,
"title": name,
"description": description,
"category": category,
"tags": tags,
"capabilities": capabilities,
"origin": origin,
"cover": cover,
"asset_count": len(asset_entries),
"example_count": len(examples),
"input_count": len(inputs),
"output_count": len(outputs),
"workflow_count": len(workflow),
}
return {
**result,
"assets": asset_entries,
"examples": examples,
"inputs": inputs,
"outputs": outputs,
"workflow": {"summary": workflow_summary, "steps": workflow},
"manifest": {
"present": bool(manifest),
"path": SKILL_PRESENTATION_FILE,
"overridden": bool(overrides),
"override_path": SKILL_PRESENTATION_OVERRIDE_FILE,
},
}
def _presentation_for(
self,
skill_dir: Path,
metadata: dict[str, str],
*,
guide_version: str | None = None,
files: list[_ScannedFile] | None = None,
summary_only: bool = False,
) -> dict[str, Any]:
"""Return the cached full presentation, projecting a summary on demand.
The summary used to live under its own cache key, which meant the list
request and the detail request could never share a build. Both views now
read one entry, so opening a Skill normally reuses work already done.
"""
version = guide_version or (_guide_fingerprint(files) if files is not None else self._guide_version(skill_dir))
key = (str(skill_dir.resolve()), version)
cached = self._presentation_cache.get(key)
if cached is None:
cached = self._build_presentation(skill_dir, metadata, files=files)
self._presentation_cache[key] = cached
if len(self._presentation_cache) > 256:
self._presentation_cache = dict(list(self._presentation_cache.items())[-128:])
return _presentation_summary(cached) if summary_only else cached
def presentation(self, skill_id: str) -> dict[str, Any]:
skill_dir = self._skill_dir(skill_id)
metadata = _frontmatter(skill_dir / "SKILL.md")
return copy.deepcopy(self._presentation_for(skill_dir, metadata))
def _invalidate_caches(self, skill_dir: Path) -> None:
prefix = str(skill_dir.resolve())
self._presentation_cache = {
key: value for key, value in self._presentation_cache.items() if key[0] != prefix
}
self._describe_cache = {
key: value for key, value in self._describe_cache.items() if key[0] != prefix
}
def reindex_presentation(self, skill_id: str) -> dict[str, Any]:
skill_dir = self._skill_dir(skill_id)
self._invalidate_caches(skill_dir)
return self.presentation(skill_id)
def update_presentation(self, skill_id: str, value: dict[str, Any]) -> dict[str, Any]:
skill_dir = self._skill_dir(skill_id)
allowed = {"display", "assets", "examples", "inputs", "outputs", "capabilities", "tags", "author", "version", "source"}
current = self._read_presentation_overrides(skill_dir)
incoming = {key: item for key, item in value.items() if key in allowed}
payload = self._merge_presentation_config(current, incoming)
display = payload.get("display")
if display is not None and not isinstance(display, dict):
raise SkillInstallError("display 必须是对象")
if isinstance(display, dict):
payload["display"] = {
key: item for key, item in display.items()
if key in {"title", "description", "category", "cover", "tags"}
}
for key in ("assets", "examples", "inputs", "outputs", "capabilities", "tags"):
if key in payload and not isinstance(payload[key], list):
raise SkillInstallError(f"{key} 必须是数组")
path = skill_dir / SKILL_PRESENTATION_OVERRIDE_FILE
temporary = path.with_suffix(f"{path.suffix}.{uuid.uuid4().hex}.tmp")
temporary.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
os.replace(temporary, path)
return self.reindex_presentation(skill_id)
def upload_presentation_assets(
self,
skill_id: str,
uploads: list[tuple[bytes, str, str]],
) -> list[dict[str, Any]]:
skill_dir = self._skill_dir(skill_id)
if not uploads:
raise SkillInstallError("请选择要上传的展示图片")
if len(uploads) > 12:
raise SkillInstallError("一次最多上传 12 张展示图片")
target_dir = skill_dir / SKILL_PRESENTATION_ASSET_DIR
target_dir.mkdir(parents=True, exist_ok=True)
results: list[dict[str, Any]] = []
allowed_extensions = {".png", ".jpg", ".jpeg", ".webp", ".gif", ".svg"}
for content, original_name, content_type in uploads:
if not content or len(content) > MAX_FILE_BYTES:
raise SkillInstallError(f"展示图片 {original_name or '未命名文件'} 为空或超过 8MB")
suffix = Path(original_name or "image").suffix.lower()
if suffix not in allowed_extensions or not str(content_type or "").lower().startswith("image/"):
raise SkillInstallError(f"展示素材只支持 PNG、JPG、WebP、GIF 和 SVG：{original_name}")
stem = _safe_name(Path(original_name).stem, "showcase")[:48]
filename = f"{stem}-{uuid.uuid4().hex[:10]}{suffix}"
target = target_dir / filename
target.write_bytes(content)
relative = target.relative_to(skill_dir).as_posix()
mime_type = mimetypes.guess_type(target.name)[0] or content_type or "application/octet-stream"
results.append({
"path": relative,
"name": Path(original_name).stem[:120] or stem,
"description": "",
"type": mime_type,
"kind": "image",
"size": len(content),
"role": "gallery",
})
self.reindex_presentation(skill_id)
return results
def _read_favorites(self) -> dict[str, list[str]]:
if not self.favorites_path.is_file():
return {}
try:
value = json.loads(self.favorites_path.read_text(encoding="utf-8"))
except (OSError, UnicodeError, ValueError):
return {}
if not isinstance(value, dict):
return {}
result: dict[str, list[str]] = {}
for owner, items in value.items():
if isinstance(owner, str) and isinstance(items, list):
result[owner] = [str(item) for item in items if isinstance(item, str)]
return result
def favorite_ids(self, owner_id: str) -> list[str]:
owner = str(owner_id or "").strip()
if not owner:
return []
with self._favorites_lock:
known = {item["id"] for item in self.catalog()}
return [item for item in self._read_favorites().get(owner, []) if item in known]
def set_favorite(self, owner_id: str, skill_id: str, *, favorite: bool) -> list[str]:
self.ensure_enabled(skill_id)
owner = str(owner_id or "").strip()
if not owner:
raise SkillInstallError("用户身份无效")
with self._favorites_lock:
value = self._read_favorites()
items = list(dict.fromkeys(value.get(owner, [])))
if favorite and skill_id not in items:
items.append(skill_id)
if not favorite:
items = [item for item in items if item != skill_id]
value[owner] = items
self.favorites_path.parent.mkdir(parents=True, exist_ok=True)
temporary = self.favorites_path.with_suffix(f"{self.favorites_path.suffix}.{uuid.uuid4().hex}.tmp")
temporary.write_text(json.dumps(value, ensure_ascii=False, indent=2), encoding="utf-8")
os.replace(temporary, self.favorites_path)
return items
def resolve_asset(self, skill_id: str, asset_path: str) -> tuple[Path, str]:
skill_dir = self._skill_dir(skill_id)
target = self._safe_presentation_asset(skill_dir, asset_path)
if target is None or target.suffix.lower() not in PRESENTATION_ASSET_EXTENSIONS:
raise SkillInstallError("Skill 展示素材不存在、过大或类型不受支持")
return target, mimetypes.guess_type(target.name)[0] or "application/octet-stream"
def catalog(self, *, include_content: bool = False) -> list[dict[str, Any]]:
items = [item for item in self.list() if item.get("enabled", True)]
if not include_content:
return items
for item in items:
skill_file = self.root / str(item["id"]) / "SKILL.md"
try:
item["content"] = skill_file.read_text(encoding="utf-8")[:50000]
except (OSError, UnicodeError):
item["content"] = ""
return items
def _skill_dir(self, skill_id: str) -> Path:
safe_id = _safe_name(skill_id)
if safe_id != skill_id or not safe_id or safe_id.startswith("."):
raise SkillInstallError("Skill ID 无效")
target = (self.root / safe_id).resolve()
root = self.root.resolve()
try:
target.relative_to(root)
except ValueError as exc:
raise SkillInstallError("Skill 路径无效") from exc
if not target.is_dir() or not (target / "SKILL.md").is_file():
raise SkillInstallError("Skill 不存在")
return target
@staticmethod
def _install_metadata(skill_dir: Path) -> dict[str, Any]:
"""Return .install.json — where the Skill came from and when."""
path = skill_dir / ".install.json"
if not path.is_file():
return {}
try:
value = json.loads(path.read_text(encoding="utf-8"))
except (OSError, ValueError):
return {}
return value if isinstance(value, dict) else {}
def _install_record(
self,
*,
staging: Path,
target: Path,
source: str,
bundle_entry: dict[str, Any] | None,
package_roots: list[str],
) -> dict[str, Any]:
"""Build the .install.json written for a freshly installed Skill.
The recorded origin is what decides whether an administrator can pull
updates, so provenance carried by the package must survive the install:
a migrated export bundle keeps the origin of the deployment it came
from (that is the whole point of migrating), and re-uploading a plain
ZIP over an existing Skill keeps that Skill's known origin instead of
downgrading it to an anonymous local upload. Only archives that declare
the bundle format may set their own origin — a repository that happens
to ship a .install.json must not redirect the update address.
"""
import datetime
incoming = self._install_metadata(staging) if bundle_entry is not None else {}
manifest_entry = bundle_entry or {}
preserved_source = str(incoming.get("source") or manifest_entry.get("source") or "").strip()
preserved_installed_at = str(incoming.get("installed_at") or manifest_entry.get("installed_at") or "").strip()
origin = "bundle" if preserved_source else ""
if not preserved_source and source.startswith("upload:") and target.is_dir():
previous = self._install_metadata(target)
previous_source = str(previous.get("source") or "").strip()
if previous_source:
preserved_source = previous_source
preserved_installed_at = str(previous.get("installed_at") or "").strip()
origin = "previous"
record: dict[str, Any] = dict(incoming) if origin == "bundle" else {}
record["source"] = preserved_source or source
record["installed_at"] = (
preserved_installed_at or datetime.datetime.now(datetime.timezone.utc).isoformat()
)
record["package_roots"] = package_roots
if origin:
record["imported_at"] = datetime.datetime.now(datetime.timezone.utc).isoformat()
record["imported_from"] = source
record["origin_preserved"] = origin
return record
@staticmethod
def _permissions(skill_dir: Path) -> dict[str, Any]:
path = skill_dir / ".permissions.json"
raw: dict[str, Any] = {}
if path.is_file():
try:
value = json.loads(path.read_text(encoding="utf-8"))
if isinstance(value, dict):
raw = value
except (OSError, ValueError):
raw = {}
allowed = [
str(item).replace("\\", "/")
for item in raw.get("allowed_scripts") or []
if isinstance(item, str) and item.strip()
]
try:
timeout = min(MAX_SCRIPT_TIMEOUT_SECONDS, max(5, int(raw.get("timeout_seconds") or DEFAULT_SCRIPT_TIMEOUT_SECONDS)))
except (TypeError, ValueError):
timeout = DEFAULT_SCRIPT_TIMEOUT_SECONDS
return {
"allow_scripts": bool(raw.get("allow_scripts")),
"allowed_scripts": allowed,
"timeout_seconds": timeout,
}
@staticmethod
def _visibility(skill_dir: Path) -> bool:
path = skill_dir / ".visibility.json"
if not path.is_file():
return True
try:
value = json.loads(path.read_text(encoding="utf-8"))
except (OSError, ValueError):
return True
return bool(value.get("enabled", True)) if isinstance(value, dict) else True
@staticmethod
def _guide_version(skill_dir: Path) -> str:
return _guide_fingerprint(_scan_skill_files(skill_dir))
def ensure_enabled(self, skill_id: str) -> None:
skill_dir = self._skill_dir(skill_id)
if not self._visibility(skill_dir):
raise SkillInstallError("Skill 已被管理员隐藏")
def guide_context(self, skill_id: str) -> dict[str, Any]:
skill_dir = self._skill_dir(skill_id)
files = _scan_skill_files(skill_dir)
metadata = _frontmatter(skill_dir / "SKILL.md")
documents: list[dict[str, str]] = []
candidates = sorted(
(item for item in files if item.is_user_document),
key=lambda item: _guide_document_score(item.relative),
)[:4]
for item in candidates:
if item.size > MAX_RESOURCE_BYTES:
continue
try:
content = item.path.read_text(encoding="utf-8")
except (OSError, UnicodeError):
continue
documents.append({"path": item.relative, "content": content[:10000]})
return {
"id": skill_id,
"name": metadata["name"],
"description": metadata["description"],
"version": _guide_fingerprint(files),
"content": (skill_dir / "SKILL.md").read_text(encoding="utf-8")[:MAX_RESOURCE_BYTES],
"documents": documents,
}
def read_cached_guide(self, skill_id: str) -> dict[str, Any] | None:
skill_dir = self._skill_dir(skill_id)
path = skill_dir / GUIDE_CACHE_FILE
if not path.is_file():
return None
try:
value = json.loads(path.read_text(encoding="utf-8"))
except (OSError, ValueError):
return None
version = self._guide_version(skill_dir)
guide = str(value.get("guide") or "").strip() if isinstance(value, dict) else ""
if not guide or value.get("version") != version:
return None
return {"guide": guide, "version": version, "cached": True}
def write_cached_guide(self, skill_id: str, *, version: str, guide: str) -> dict[str, Any]:
skill_dir = self._skill_dir(skill_id)
current_version = self._guide_version(skill_dir)
clean_guide = guide.strip()
if version != current_version or not clean_guide:
raise SkillInstallError("Skill 文档已更新，请重新生成使用指南")
(skill_dir / GUIDE_CACHE_FILE).write_text(
json.dumps({"version": current_version, "guide": clean_guide}, ensure_ascii=False),
encoding="utf-8",
)
return {"guide": clean_guide, "version": current_version, "cached": False}
def describe(self, skill_id: str, *, include_content: bool = True) -> dict[str, Any]:
return self._describe(skill_id, include_content=include_content)[0]
def describe_with_etag(self, skill_id: str, *, include_content: bool = True) -> tuple[dict[str, Any], str]:
payload, version = self._describe(skill_id, include_content=include_content)
return payload, f'"{version}"'
def _describe(self, skill_id: str, *, include_content: bool) -> tuple[dict[str, Any], str]:
skill_dir = self._skill_dir(skill_id)
files = _scan_skill_files(skill_dir)
content_version = _content_fingerprint(files)
cache_key = (str(skill_dir.resolve()), content_version, include_content)
cached = self._describe_cache.get(cache_key)
if cached is not None:
return copy.deepcopy(cached), content_version
metadata = _frontmatter(skill_dir / "SKILL.md")
permissions = self._permissions(skill_dir)
resources: list[dict[str, Any]] = []
scripts: list[dict[str, Any]] = []
for item in files:
if item.name in {
".install.json",
".permissions.json",
".visibility.json",
GUIDE_CACHE_FILE,
SKILL_PRESENTATION_OVERRIDE_FILE,
}:
continue
kind = _script_kind(item.path)
entry = {
"path": item.relative,
"size": item.size,
"type": mimetypes.guess_type(item.name)[0] or "application/octet-stream",
"readable": item.suffix in TEXT_RESOURCE_EXTENSIONS and item.size <= MAX_RESOURCE_BYTES,
}
resources.append(entry)
if item.relative.startswith("scripts/") and kind:
scripts.append({
**entry,
"kind": kind,
"allowed": bool(permissions["allow_scripts"]) and (
not permissions["allowed_scripts"] or item.relative in permissions["allowed_scripts"]
),
})
result: dict[str, Any] = {
"id": skill_id,
"name": metadata["name"],
"description": metadata["description"],
"resources": resources,
"scripts": scripts,
"permissions": permissions,
"enabled": self._visibility(skill_dir),
"presentation": self._presentation_for(
skill_dir,
metadata,
guide_version=_guide_fingerprint(files),
files=files,
),
}
if include_content:
result["content"] = (skill_dir / "SKILL.md").read_text(encoding="utf-8")[:MAX_RESOURCE_BYTES]
self._describe_cache[cache_key] = result
if len(self._describe_cache) > 128:
self._describe_cache = dict(list(self._describe_cache.items())[-64:])
return copy.deepcopy(result), content_version
def read_resource(self, skill_id: str, resource_path: str) -> dict[str, Any]:
skill_dir = self._skill_dir(skill_id)
relative = _safe_relative_path(resource_path)
target = (skill_dir / relative).resolve()
try:
target.relative_to(skill_dir)
except ValueError as exc:
raise SkillInstallError("资源路径超出 Skill 目录") from exc
if not target.is_file():
raise SkillInstallError("Skill 资源不存在")
size = target.stat().st_size
if size > MAX_RESOURCE_BYTES:
raise SkillInstallError(f"Skill 文本资源不能超过 {MAX_RESOURCE_BYTES // 1024}KB")
if not _is_text_resource(target):
raise SkillInstallError("该资源不是可读取的文本文件")
try:
content = target.read_text(encoding="utf-8")
except UnicodeError as exc:
raise SkillInstallError("Skill 资源不是 UTF-8 文本") from exc
return {
"skill_id": skill_id,
"path": relative.as_posix(),
"content": content,
"size": size,
}
def set_permissions(
self,
skill_id: str,
*,
allow_scripts: bool,
allowed_scripts: list[str] | None = None,
timeout_seconds: int = DEFAULT_SCRIPT_TIMEOUT_SECONDS,
) -> dict[str, Any]:
skill_dir = self._skill_dir(skill_id)
known_scripts = {
path.relative_to(skill_dir).as_posix()
for path in skill_dir.rglob("*")
if path.is_file() and path.relative_to(skill_dir).as_posix().startswith("scripts/") and _script_kind(path)
}
requested: list[str] = []
for item in allowed_scripts or []:
relative = _safe_relative_path(item).as_posix()
if relative not in known_scripts:
raise SkillInstallError(f"脚本不存在或类型不受支持：{relative}")
if relative not in requested:
requested.append(relative)
payload = {
"allow_scripts": bool(allow_scripts),
"allowed_scripts": requested,
"timeout_seconds": min(MAX_SCRIPT_TIMEOUT_SECONDS, max(5, int(timeout_seconds))),
}
(skill_dir / ".permissions.json").write_text(
json.dumps(payload, ensure_ascii=False, indent=2),
encoding="utf-8",
)
return self.describe(skill_id, include_content=False)
def set_visibility(self, skill_id: str, *, enabled: bool) -> dict[str, Any]:
skill_dir = self._skill_dir(skill_id)
(skill_dir / ".visibility.json").write_text(
json.dumps({"enabled": bool(enabled)}, ensure_ascii=False, indent=2),
encoding="utf-8",
)
return self.describe(skill_id, include_content=False)
@_serialize_duplicate_runs
def run_script(
self,
skill_id: str,
identity: dict[str, object],
*,
script_path: str,
args: list[str],
uploads: list[tuple[bytes, str, str]],
client_run_id: str = "",
) -> dict[str, Any]:
self.cleanup_expired_runs()
skill_dir = self._skill_dir(skill_id)
permissions = self._permissions(skill_dir)
if not permissions["allow_scripts"]:
raise PermissionError("管理员尚未允许这个 Skill 执行脚本")
relative = _safe_relative_path(script_path)
relative_posix = relative.as_posix()
if not relative_posix.startswith("scripts/"):
raise SkillInstallError("只允许执行 Skill 的 scripts/ 目录中的脚本")
if permissions["allowed_scripts"] and relative_posix not in permissions["allowed_scripts"]:
raise PermissionError("该脚本不在管理员白名单中")
script = (skill_dir / relative).resolve()
try:
script.relative_to(skill_dir)
except ValueError as exc:
raise SkillInstallError("脚本路径超出 Skill 目录") from exc
kind = _script_kind(script)
if not script.is_file() or not kind:
raise SkillInstallError("脚本不存在或类型不受支持")
if len(args) > MAX_SCRIPT_ARGS:
raise SkillInstallError(f"脚本参数不能超过 {MAX_SCRIPT_ARGS} 个")
owner = _safe_name(str(identity.get("id") or "user"), "user")
clean_client_run_id = str(client_run_id or "").strip()
if len(clean_client_run_id) > 200 or "\x00" in clean_client_run_id:
raise SkillInstallError("client_run_id 无效")
cache_path: Path | None = None
if clean_client_run_id:
cache_key = hashlib.sha256(f"{owner}:{skill_id}:{clean_client_run_id}".encode("utf-8")).hexdigest()
cache_path = self.runs_root / ".runs" / f"{cache_key}.json"
if cache_path.is_file():
try:
cached = json.loads(cache_path.read_text(encoding="utf-8"))
if isinstance(cached, dict):
expires_at = float(cached.get("expires_at") or 0)
cached_artifacts = cached.get("artifacts") if isinstance(cached.get("artifacts"), list) else []
cached_run_id = _safe_name(str(cached.get("run_id") or ""), "")
def cached_artifact_exists(item: object) -> bool:
if not isinstance(item, dict):
return True
artifact_url = str(item.get("url") or "")
if item.get("streamed") and artifact_url.startswith("/api/skills/media/"):
try:
self.resolve_media_link(artifact_url.rsplit("/", 1)[-1])
return True
except SkillInstallError:
return False
return (
self.runs_root
/ owner
/ cached_run_id
/ _safe_output_name(str(item.get("name") or "artifact"))
).is_file()
artifacts_exist = all(cached_artifact_exists(item) for item in cached_artifacts)
if (expires_at and expires_at <= time.time()) or not artifacts_exist:
cache_path.unlink(missing_ok=True)
else:
return cached
except (OSError, ValueError):
pass
run_id = f"{int(time.time())}-{uuid.uuid4().hex[:12]}"
durable_dir = self.runs_root / owner / run_id
with tempfile.TemporaryDirectory(prefix=f"skill-{skill_id}-") as temp_name:
workspace = Path(temp_name).resolve()
input_dir = workspace / "inputs"
output_dir = workspace / "outputs"
home_dir = workspace / "home"
input_dir.mkdir()
output_dir.mkdir()
home_dir.mkdir()
input_paths: list[Path] = []
total_upload_bytes = 0
for index, (data, name, _mime_type) in enumerate(uploads):
total_upload_bytes += len(data)
if total_upload_bytes > MAX_SCRIPT_OUTPUT_BYTES:
raise SkillInstallError("脚本输入文件总大小不能超过 25MB")
safe_name = _safe_output_name(name or f"input-{index + 1}")
target = input_dir / f"{index + 1:02d}-{safe_name}"
target.write_bytes(data)
input_paths.append(target)
output_pattern = re.compile(r"\{\{output:([^{}]+)\}\}")
input_pattern = re.compile(r"\{\{input:(\d+)\}\}")
def resolve_argument(raw: str) -> str:
value = str(raw)
# Arguments are passed as an argv list without a shell. Keep
# NUL forbidden, but allow line breaks so parser Skills can
# process pasted share text containing several URLs.
if len(value) > MAX_SCRIPT_ARG_CHARS or "\x00" in value:
raise SkillInstallError("脚本参数无效或过长")
used_placeholder = False
def input_replacement(match: re.Match[str]) -> str:
nonlocal used_placeholder
used_placeholder = True
index = int(match.group(1))
if index < 0 or index >= len(input_paths):
raise SkillInstallError(f"输入文件索引不存在：{index}")
return str(input_paths[index])
def output_replacement(match: re.Match[str]) -> str:
nonlocal used_placeholder
used_placeholder = True
return str(output_dir / _safe_output_name(match.group(1)))
value = input_pattern.sub(input_replacement, value)
value = output_pattern.sub(output_replacement, value)
for token, replacement in {
"{{workspace}}": workspace,
"{{skill}}": skill_dir,
"{{outputs}}": output_dir,
}.items():
if token in value:
used_placeholder = True
value = value.replace(token, str(replacement))
if "{{" in value or "}}" in value:
raise SkillInstallError("脚本参数包含未知占位符")
if not used_placeholder:
candidate = Path(value)
if candidate.is_absolute() or PurePosixPath(value.replace("\\", "/")).is_absolute() or ".." in PurePosixPath(value.replace("\\", "/")).parts:
raise SkillInstallError("脚本参数中的文件路径必须使用受控占位符")
return value
resolved_args = [resolve_argument(item) for item in args]
if kind == "python":
# Keep Python isolated from user/site configuration while still
# allowing a Skill entry point to import sibling modules such as
# scripts/core.py. The bootstrap is fixed by the host; neither
# the Skill nor the model can inject code into it.
bootstrap = (
"import runpy,sys; "
"sys.path.insert(0,sys.argv[1]); "
"sys.argv=sys.argv[2:]; "
"runpy.run_path(sys.argv[0],run_name='__main__')"
)
command = [
sys.executable,
"-I",
"-c",
bootstrap,
str(script.parent),
str(script),
*resolved_args,
]
else:
node = shutil.which("node")
if not node:
raise SkillInstallError("服务器没有可用的 Node.js 运行时")
command = [node, str(script), *resolved_args]
env = {
"PATH": os.environ.get("PATH", ""),
"PYTHONIOENCODING": "utf-8",
"PYTHONUTF8": "1",
"HOME": str(home_dir),
"USERPROFILE": str(home_dir),
"TEMP": str(workspace),
"TMP": str(workspace),
"SKILL_DIR": str(skill_dir),
"SKILL_WORKSPACE": str(workspace),
"SKILL_OUTPUT_DIR": str(output_dir),
}
# Windows needs these variables to initialise Winsock. Omitting
# them from the deliberately small child environment makes even
# ``import asyncio`` fail with WinError 10106 before the Skill's
# own code starts. They identify system directories only and do
# not broaden the script's data or credential access.
if os.name == "nt":
for name in ("SYSTEMROOT", "WINDIR", "COMSPEC", "PATHEXT"):
value = os.environ.get(name, "")
if value:
env[name] = value
started_at = time.monotonic()
try:
completed = subprocess.run(
command,
cwd=workspace,
env=env,
stdin=subprocess.DEVNULL,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
encoding="utf-8",
errors="replace",
timeout=int(permissions["timeout_seconds"]),
check=False,
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
)
except subprocess.TimeoutExpired as exc:
raise SkillInstallError(f"脚本执行超过 {permissions['timeout_seconds']} 秒，已终止") from exc
# Empty cache markers/lock files are implementation details, not
# user-facing artifacts. Do not render them as bogus downloads.
output_candidates = [
path for path in output_dir.rglob("*")
if path.is_file() and path.stat().st_size > 0
]
if len(output_candidates) > MAX_SCRIPT_OUTPUT_FILES:
raise SkillInstallError(f"脚本输出文件不能超过 {MAX_SCRIPT_OUTPUT_FILES} 个")
total_output_bytes = sum(path.stat().st_size for path in output_candidates)
output_limit = MEDIA_PARSER_OUTPUT_BYTES if skill_id.lower() == "media-parser" else MAX_SCRIPT_OUTPUT_BYTES
if total_output_bytes > output_limit:
raise SkillInstallError(f"脚本输出文件总大小不能超过 {output_limit // 1024 // 1024}MB")
artifacts: list[dict[str, Any]] = []
expires_at: float | None = None
if output_candidates:
durable_dir.mkdir(parents=True, exist_ok=False)
for source in output_candidates:
name = _safe_output_name(source.name)
target = durable_dir / name
suffix = 2
while target.exists():
target = durable_dir / f"{Path(name).stem}-{suffix}{Path(name).suffix}"
suffix += 1
shutil.copy2(source, target)
try:
relative_url = target.relative_to(DATA_DIR / "files").as_posix()
except ValueError:
# Custom roots are used by tests and embedders. Production
# keeps runs under DATA_DIR/files and uses the first path.
relative_url = (Path("skills") / owner / run_id / target.name).as_posix()
artifacts.append({
"name": target.name,
"type": mimetypes.guess_type(target.name)[0] or "application/octet-stream",
"size": target.stat().st_size,
"url": f"/files/{relative_url}",
})
if skill_id.lower() == "media-parser":
expires_at = time.time() + MEDIA_PARSER_ARTIFACT_TTL_SECONDS
(durable_dir / ".expires-at").write_text(str(expires_at), encoding="utf-8")
self._enforce_media_owner_quota(owner)
elif skill_id.lower() == "media-parser" and completed.returncode == 0:
artifacts.extend(self._media_stream_artifacts(completed.stdout))
if artifacts:
expires_at = max(float(item.get("expires_at") or 0) for item in artifacts)
result = {
"run_id": run_id,
"skill_id": skill_id,
"script": relative_posix,
"status": "success" if completed.returncode == 0 else "error",
"exit_code": completed.returncode,
"duration_ms": int((time.monotonic() - started_at) * 1000),
"stdout": completed.stdout[-MAX_SCRIPT_STDIO_CHARS:],
"stderr": completed.stderr[-MAX_SCRIPT_STDIO_CHARS:],
"artifacts": artifacts,
}
if expires_at is not None:
result["expires_at"] = expires_at
if cache_path is not None:
cache_path.parent.mkdir(parents=True, exist_ok=True)
staging = cache_path.with_suffix(f".tmp-{os.getpid()}")
staging.write_text(json.dumps(result, ensure_ascii=False), encoding="utf-8")
os.replace(staging, cache_path)
return result
def _backup_skill(self, skill_dir: Path) -> Path:
"""Snapshot a Skill before it is overwritten and keep the newest few.
An update is destructive: the previous files are deleted and replaced.
Keeping a rolling snapshot means a bad release can be rolled back by
copying the directory back, and it gives ``_restore_preserved_state`` a
single source for the administrator state that has to survive.
"""
import datetime
backups_root = self.root / SKILL_BACKUP_DIR / skill_dir.name
backups_root.mkdir(parents=True, exist_ok=True)
stamp = datetime.datetime.now(datetime.timezone.utc).strftime("%Y%m%d-%H%M%S")
destination = backups_root / f"{stamp}-{uuid.uuid4().hex[:6]}"
try:
shutil.copytree(skill_dir, destination)
except OSError as exc:
shutil.rmtree(destination, ignore_errors=True)
raise SkillInstallError(f"无法备份现有 Skill，已中止覆盖：{exc}") from exc
self._prune_skill_backups(backups_root)
return destination
@staticmethod
def _prune_skill_backups(backups_root: Path) -> None:
try:
snapshots = sorted(
(path for path in backups_root.iterdir() if path.is_dir()),
key=lambda path: path.name,
)
except OSError:
return
for stale in snapshots[:-MAX_SKILL_BACKUPS]:
shutil.rmtree(stale, ignore_errors=True)
def install_zip(
self,
payload: bytes,
*,
source: str,
overwrite: bool = False,
subpath: str = "",
max_bytes: int = MAX_ARCHIVE_BYTES,
max_files: int = MAX_FILES,
) -> list[dict[str, Any]]:
import datetime
self.root.mkdir(parents=True, exist_ok=True)
with tempfile.TemporaryDirectory(prefix="chatgpt2api-skill-") as temp_name:
extracted = Path(temp_name)
_extract_archive(payload, extracted, subpath=subpath, max_bytes=max_bytes, max_files=max_files)
search_root = extracted
if subpath:
matches = [path for path in extracted.rglob(subpath.split("/")[-1]) if path.is_dir() and path.as_posix().endswith(subpath)]
if not matches:
raise SkillInstallError(f"仓库中找不到技能目录：{subpath}")
search_root = matches[0]
skill_files = sorted(search_root.rglob("SKILL.md"))
if not skill_files:
raise SkillInstallError("ZIP 中没有找到 SKILL.md")
bundle_skills = _export_bundle_sources(extracted)
installed_ids: list[str] = []
discovered_ids: set[str] = set()
for skill_file in skill_files:
# A parent skill owns its nested resources; do not install a
# nested SKILL.md twice when a package intentionally includes one.
if any(parent.name == "SKILL.md" for parent in skill_file.parents):
continue
metadata = _frontmatter(skill_file)
skill_id = _safe_name(metadata["name"], _safe_name(skill_file.parent.name))
# Release snapshots and examples sometimes contain older copies
# of the same Skill. Prefer the first (normally shallowest)
# definition instead of overwriting a complete runtime with an
# archived duplicate later in the same package.
if skill_id in discovered_ids:
continue
discovered_ids.add(skill_id)
target = self.root / skill_id
if target.exists() and not overwrite:
raise SkillInstallError(f"Skill 已存在：{skill_id}")
# Overwriting deletes the previous copy outright, so snapshot it
# first. The snapshot doubles as the source of the administrator
# state that must survive a reinstall: script permissions,
# showcase overrides, uploaded showcase media and visibility.
backup_dir = self._backup_skill(target) if target.is_dir() else None
staging = self.root / f".{skill_id}.installing-{os.getpid()}"
if staging.exists():
shutil.rmtree(staging)
try:
shutil.copytree(skill_file.parent, staging)
package_roots = _referenced_package_roots(skill_file, search_root)
for package_root in package_roots:
# External implementations become controlled Skill
# scripts. Keeping them below scripts/ makes them
# discoverable to the runtime and subject to the same
# administrator whitelist as native Skill scripts.
destination = staging / "scripts" / package_root.name
if package_root.is_dir():
shutil.copytree(package_root, destination, dirs_exist_ok=True)
elif package_root.is_file():
destination.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(package_root, destination)
# A repository may keep its human-facing README/USAGE next
# to (or above) the Skill directory rather than beside
# SKILL.md. Preserve those package docs so the chat UI can
# explain the Skill to end users after installation.
copied_doc_names: set[str] = set()
package_docs_dir = staging / "_package-docs"
for ancestor in _ancestor_directories(skill_file, search_root):
for candidate in sorted(ancestor.iterdir()):
if not candidate.is_file() or not PACKAGE_DOC_RE.match(candidate.name):
continue
normalized_name = candidate.name.lower()
if normalized_name in copied_doc_names:
continue
copied_doc_names.add(normalized_name)
package_docs_dir.mkdir(parents=True, exist_ok=True)
shutil.copy2(candidate, package_docs_dir / candidate.name)
(staging / ".install.json").write_text(json.dumps(
self._install_record(
staging=staging,
target=target,
source=source,
bundle_entry=bundle_skills.get(skill_id),
package_roots=[path.name for path in package_roots],
),
ensure_ascii=False,
indent=2,
), encoding="utf-8")
if target.exists():
shutil.rmtree(target)
os.replace(staging, target)
if backup_dir is not None:
_restore_preserved_state(target, backup_dir)
finally:
if staging.exists():
shutil.rmtree(staging, ignore_errors=True)
installed_ids.append(skill_id)
if not installed_ids:
raise SkillInstallError("没有可安装的 Skill")
installed = {item["id"]: item for item in self.list()}
return [installed[item] for item in installed_ids if item in installed]
def install_url(self, url: str, *, overwrite: bool = False) -> list[dict[str, Any]]:
clean_url = url.strip()
parsed = urlparse(clean_url)
if parsed.scheme not in {"http", "https"}:
raise SkillInstallError("仅支持 http/https URL")
last_error: Exception | None = None
for download_url, subpath in _github_download_candidates(clean_url):
try:
with httpx.Client(follow_redirects=True, timeout=45) as client:
response = client.get(download_url, headers={"User-Agent": "chatgpt2api-skill-installer/1.0"})
response.raise_for_status()
payload = response.content
return self.install_zip(payload, source=clean_url, overwrite=overwrite, subpath=subpath)
except SkillInstallError as exc:
# A valid branch/archive was already found. Do not mask an
# installation result such as "already exists" by probing a
# fallback branch that may return 404.
if "Skill 已存在：" in str(exc):
raise
last_error = exc
except httpx.HTTPError as exc:
last_error = exc
raise SkillInstallError(f"下载或安装失败：{last_error}")
def update_skill(self, skill_id: str) -> list[dict[str, Any]]:
"""Reinstall a Skill from the source recorded when it was installed.
Only remotely-sourced Skills can be refreshed. A Skill that arrived as an
uploaded archive has no address to fetch from, so the administrator has
to supply a new archive instead.
"""
skill_dir = self._skill_dir(skill_id)
source = str(self._install_metadata(skill_dir).get("source") or "").strip()
if not source or source.startswith("upload:"):
raise SkillInstallError("这个 Skill 由本地上传安装，没有可用的远端地址，请重新上传新版 ZIP")
if not self._is_direct_install_url(source):
raise SkillInstallError(f"该 Skill 的来源不支持自动更新：{source}")
installed = self.install_url(source, overwrite=True)
self._invalidate_caches(skill_dir)
return installed
def export_skills(self, skill_ids: list[str] | None = None) -> tuple[bytes, str]:
"""Bundle Skills — code plus administrator state — into one archive.
The result is the unit of migration between deployments: installing it
elsewhere reproduces the Skills exactly, including script permissions,
showcase overrides, uploaded showcase media and visibility, because the
installer copies each Skill directory verbatim.
"""
import datetime
if skill_ids:
directories = [self._skill_dir(item) for item in dict.fromkeys(skill_ids)]
else:
directories = [self._skill_dir(str(item["id"])) for item in self.list()]
if not directories:
raise SkillInstallError("没有可导出的 Skill")
manifest_skills: list[dict[str, Any]] = []
buffer = io.BytesIO()
with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED, compresslevel=6) as archive:
for skill_dir in directories:
metadata = _frontmatter(skill_dir / "SKILL.md")
install_metadata = self._install_metadata(skill_dir)
exported = 0
for item in _scan_skill_files(skill_dir):
if _is_export_excluded(item.relative):
continue
archive.write(item.path, f"{skill_dir.name}/{item.relative}")
exported += 1
manifest_skills.append({
"id": skill_dir.name,
"name": metadata["name"],
"description": metadata["description"],
"source": str(install_metadata.get("source") or "manual"),
"installed_at": str(install_metadata.get("installed_at") or ""),
"file_count": exported,
})
archive.writestr(EXPORT_MANIFEST_FILE, json.dumps({
"format": SKILL_EXPORT_FORMAT,
"format_version": SKILL_EXPORT_FORMAT_VERSION,
"exported_at": datetime.datetime.now(datetime.timezone.utc).isoformat(),
"skill_count": len(manifest_skills),
"skills": manifest_skills,
}, ensure_ascii=False, indent=2))
stamp = datetime.datetime.now().strftime("%Y%m%d-%H%M%S")
# Naming the file after a Skill only makes sense when one was picked
# explicitly; a "export everything" request keeps the bundle name even
# when the installation happens to hold a single Skill.
filename = (
f"skill-{directories[0].name}-{stamp}.zip"
if skill_ids and len(directories) == 1
else f"skills-export-{stamp}.zip"
)
return buffer.getvalue(), filename
@staticmethod
def _instruction_urls(text: str) -> list[str]:
result: list[str] = []
for match in re.finditer(r"https?://[^\s<>\]\[()\"']+", text, re.IGNORECASE):
value = match.group(0).rstrip(".,;:!?，。；：！？")
if value not in result:
result.append(value)
return result[:24]
@staticmethod
def _is_direct_install_url(value: str) -> bool:
parsed = urlparse(value)
path = parsed.path.lower()
return (parsed.hostname or "").lower() == "github.com" or path.endswith(".zip")
def _read_install_instruction(self, url: str) -> str:
current = _require_public_remote_url(url)
with httpx.Client(follow_redirects=False, timeout=25) as client:
for _ in range(6):
response = client.get(current, headers={"User-Agent": "chatgpt2api-skill-installer/1.0", "Accept": "text/markdown,text/plain;q=0.9,*/*;q=0.2"})
if response.is_redirect:
location = response.headers.get("location")
if not location:
raise SkillInstallError("安装说明重定向缺少目标地址")
current = _require_public_remote_url(str(httpx.URL(current).join(location)))
continue
response.raise_for_status()
if len(response.content) > MAX_RESOURCE_BYTES:
raise SkillInstallError("安装说明文件超过 512KB 限制")
content_type = response.headers.get("content-type", "").lower()
if not any(value in content_type for value in ("text/", "markdown", "json", "yaml", "octet-stream")):
raise SkillInstallError("安装说明不是可读取的文本文件")
try:
return response.content.decode("utf-8")
except UnicodeDecodeError as exc:
raise SkillInstallError("安装说明必须使用 UTF-8 编码") from exc
raise SkillInstallError("安装说明重定向次数过多")
def install_instruction(self, instruction: str, *, overwrite: bool = False) -> dict[str, Any]:
clean = str(instruction or "").strip()
if not clean or len(clean) > 8000:
raise SkillInstallError("安装请求为空或过长")
supplied_urls = self._instruction_urls(clean)
if not supplied_urls:
raise SkillInstallError("没有找到安装说明、GitHub 或 ZIP 地址")
candidates = [url for url in supplied_urls if self._is_direct_install_url(url)]
instruction_urls = [url for url in supplied_urls if url not in candidates]
inspected: list[str] = []
for instruction_url in instruction_urls[:3]:
document = self._read_install_instruction(instruction_url)
inspected.append(instruction_url)
for candidate in self._instruction_urls(document):
if self._is_direct_install_url(candidate) and candidate not in candidates:
candidates.append(candidate)
if not candidates:
raise SkillInstallError("安装说明中没有找到可受控安装的 GitHub 仓库或 ZIP 地址；不会执行网页中的 shell/npx 命令")
errors: list[str] = []
for candidate in candidates[:8]:
try:
installed = self.install_url(candidate, overwrite=overwrite)
return {"installed": installed, "resolved_url": candidate, "instruction_urls": inspected}
except SkillInstallError as exc:
existing_match = re.search(r"Skill 已存在：([a-z0-9._-]+)", str(exc), re.IGNORECASE)
if existing_match and not overwrite:
existing = next((item for item in self.list() if item["id"] == existing_match.group(1)), None)
if existing:
return {"installed": [existing], "resolved_url": candidate, "instruction_urls": inspected, "already_installed": True}
errors.append(str(exc))
raise SkillInstallError(f"安装说明已读取，但候选 Skill 均安装失败：{'；'.join(errors[-3:])}")
def delete(self, skill_id: str) -> None:
target = self._skill_dir(skill_id)
shutil.rmtree(target)
skill_service = SkillService()