388 lines
13 KiB
Python
Executable File
388 lines
13 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""
|
||
技能管理系统 — 扫描、打分、审计、修复
|
||
=========================================
|
||
用法:
|
||
skill-manager.py scan → 扫描所有skill，产出质量报告
|
||
skill-manager.py dashboard → 打印概要看板
|
||
skill-manager.py fix → 修复缺失的元数据
|
||
skill-manager.py audit → 标记需归档的skill
|
||
"""
|
||
|
||
import json, os, re, sys, time
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
|
||
HOME = os.path.expanduser("~")
|
||
SKILLS_DIR = os.path.join(HOME, ".hermes", "skills")
|
||
ARCHIVE_DIR = os.path.join(SKILLS_DIR, ".archive")
|
||
REPORT_FILE = os.path.join(HOME, ".hermes", "skill-health.json")
|
||
MAX_DESC_LEN = 120 # 描述截断长度
|
||
|
||
SCORE_WEIGHTS = {
|
||
"has_version": 1.0,
|
||
"version_gt_1": 1.0,
|
||
"has_tags": 1.0,
|
||
"has_related": 0.5,
|
||
"desc_length": 1.5,
|
||
"has_references": 1.0,
|
||
"has_scripts": 1.0,
|
||
"has_setup_info": 1.0,
|
||
"not_archive": 1.0,
|
||
"desc_quality": 1.0,
|
||
}
|
||
|
||
|
||
def scan_skills():
|
||
"""扫描所有 skill 目录，返回元数据列表"""
|
||
skills = []
|
||
|
||
# 先扫 active skills
|
||
for root, dirs, files in os.walk(SKILLS_DIR):
|
||
if "SKILL.md" in files:
|
||
path = os.path.join(root, "SKILL.md")
|
||
meta = parse_skill(path)
|
||
if meta:
|
||
skills.append(meta)
|
||
|
||
# 再扫 archive
|
||
if os.path.exists(ARCHIVE_DIR):
|
||
for root, dirs, files in os.walk(ARCHIVE_DIR):
|
||
if "SKILL.md" in files:
|
||
path = os.path.join(root, "SKILL.md")
|
||
meta = parse_skill(path)
|
||
if meta:
|
||
meta["is_archived"] = True
|
||
skills.append(meta)
|
||
|
||
return skills
|
||
|
||
|
||
def parse_skill(path):
|
||
"""解析一个 SKILL.md 文件，提取元数据"""
|
||
try:
|
||
with open(path, "r") as f:
|
||
content = f.read()
|
||
except Exception:
|
||
return None
|
||
|
||
# 提取 frontmatter
|
||
fm = {}
|
||
fm_match = re.match(r"^---\s*\n(.*?)\n---", content, re.DOTALL)
|
||
if fm_match:
|
||
for line in fm_match.group(1).split("\n"):
|
||
m = re.match(r"^(\w+)\s*:\s*(.*)", line)
|
||
if m:
|
||
key = m.group(1).strip()
|
||
val = m.group(2).strip().strip('"').strip("'")
|
||
fm[key] = val
|
||
|
||
# 相对路径（从 skills/ 开始）
|
||
rel_path = path.replace(SKILLS_DIR, "").lstrip("/")
|
||
|
||
# 分类
|
||
parts = rel_path.split("/")
|
||
category = parts[0] if len(parts) > 1 else "uncategorized"
|
||
name = parts[-2] if len(parts) > 1 else rel_path.replace("/SKILL.md", "")
|
||
|
||
# 描述
|
||
desc = fm.get("description", "")
|
||
|
||
# 版本
|
||
version = fm.get("version", "0.1")
|
||
try:
|
||
ver_num = float(version) if "." in version else 1.0
|
||
except:
|
||
ver_num = 0.1
|
||
|
||
# tags
|
||
tags_raw = fm.get("tags", fm.get("metadata", ""))
|
||
has_tags = bool(tags_raw)
|
||
|
||
# related skills
|
||
related = fm.get("related_skills", fm.get("metadata", ""))
|
||
has_related = "related_skills" in content or bool(related)
|
||
|
||
# 目录结构
|
||
skill_dir = os.path.dirname(path)
|
||
has_refs = os.path.isdir(os.path.join(skill_dir, "references")) and bool(os.listdir(os.path.join(skill_dir, "references")))
|
||
has_scripts = os.path.isdir(os.path.join(skill_dir, "scripts")) and bool(os.listdir(os.path.join(skill_dir, "scripts")))
|
||
|
||
# setup info
|
||
has_setup = "setup_needed" in content or "readiness_status" in content or "required_commands" in content
|
||
|
||
# 描述质量
|
||
desc_quality = 0
|
||
if len(desc) > 20:
|
||
desc_quality = 0.5
|
||
if len(desc) > 50:
|
||
desc_quality = 1.0
|
||
if "—" in desc or "：" in desc:
|
||
desc_quality = 1.5 # 有详细说明
|
||
|
||
# 是否归档
|
||
is_archived = ".archive" in path
|
||
|
||
# 计算基础分
|
||
base_score = (
|
||
(1.0 if fm.get("version") else 0) * SCORE_WEIGHTS["has_version"] +
|
||
(1.0 if ver_num > 1.0 else 0) * SCORE_WEIGHTS["version_gt_1"] +
|
||
(1.0 if has_tags else 0) * SCORE_WEIGHTS["has_tags"] +
|
||
(1.0 if has_related else 0) * SCORE_WEIGHTS["has_related"] +
|
||
min(len(desc) / 80, 1.0) * SCORE_WEIGHTS["desc_length"] +
|
||
(1.0 if has_refs else 0) * SCORE_WEIGHTS["has_references"] +
|
||
(1.0 if has_scripts else 0) * SCORE_WEIGHTS["has_scripts"] +
|
||
(1.0 if has_setup else 0) * SCORE_WEIGHTS["has_setup_info"] +
|
||
(0.0 if is_archived else 1.0) * SCORE_WEIGHTS["not_archive"] +
|
||
desc_quality * 1.0
|
||
)
|
||
|
||
# 归一化到 0-10
|
||
max_possible = sum(SCORE_WEIGHTS.values()) + 1.5 # +1.5 for desc_quality bonus
|
||
quality_score = round(base_score / max_possible * 10, 1)
|
||
|
||
# 等级
|
||
if quality_score >= 8:
|
||
grade = "A"
|
||
elif quality_score >= 6:
|
||
grade = "B"
|
||
elif quality_score >= 4:
|
||
grade = "C"
|
||
else:
|
||
grade = "D"
|
||
|
||
return {
|
||
"name": name,
|
||
"category": category,
|
||
"description": desc[:MAX_DESC_LEN] + ("..." if len(desc) > MAX_DESC_LEN else ""),
|
||
"version": version,
|
||
"path": rel_path,
|
||
"is_archived": is_archived,
|
||
"has_refs": has_refs,
|
||
"has_scripts": has_scripts,
|
||
"has_setup": has_setup,
|
||
"has_tags": has_tags,
|
||
"has_related": has_related,
|
||
"desc_len": len(desc),
|
||
"quality_score": quality_score,
|
||
"grade": grade,
|
||
"needs_attention": quality_score < 5,
|
||
}
|
||
|
||
|
||
def generate_report(skills):
|
||
"""生成完整报告"""
|
||
active = [s for s in skills if not s["is_archived"]]
|
||
archived = [s for s in skills if s["is_archived"]]
|
||
|
||
grade_counts = {"A": 0, "B": 0, "C": 0, "D": 0}
|
||
for s in active:
|
||
grade_counts[s["grade"]] = grade_counts.get(s["grade"], 0) + 1
|
||
|
||
categories = {}
|
||
for s in active:
|
||
cat = s["category"]
|
||
if cat not in categories:
|
||
categories[cat] = {"count": 0, "avg_score": 0}
|
||
categories[cat]["count"] += 1
|
||
categories[cat]["avg_score"] += s["quality_score"]
|
||
for cat in categories:
|
||
categories[cat]["avg_score"] = round(categories[cat]["avg_score"] / categories[cat]["count"], 1)
|
||
|
||
needs_attention = [s for s in active if s["needs_attention"]]
|
||
|
||
report = {
|
||
"timestamp": datetime.now(timezone.utc).isoformat(),
|
||
"summary": {
|
||
"total_skills": len(skills),
|
||
"active": len(active),
|
||
"archived": len(archived),
|
||
"avg_score": round(sum(s["quality_score"] for s in active) / len(active), 1) if active else 0,
|
||
"grades": grade_counts,
|
||
"categories": len(categories),
|
||
"needs_attention": len(needs_attention),
|
||
},
|
||
"categories": categories,
|
||
"needs_attention": needs_attention,
|
||
"top_skills": sorted(active, key=lambda x: -x["quality_score"])[:10],
|
||
"bottom_skills": sorted(active, key=lambda x: x["quality_score"])[:10],
|
||
"skills": sorted(active, key=lambda x: (-x["quality_score"], x["name"])),
|
||
}
|
||
|
||
return report
|
||
|
||
|
||
def print_dashboard(report):
|
||
"""打印看板"""
|
||
s = report["summary"]
|
||
print(f"\n{'='*55}")
|
||
print(f" 技能健康看板 {s['total_skills']} 个 (活跃 {s['active']} / 归档 {s['archived']})")
|
||
print(f"{'='*55}")
|
||
print(f" 平均分: {s['avg_score']}/10")
|
||
print(f" 等级分布: A={s['grades'].get('A',0)} B={s['grades'].get('B',0)} C={s['grades'].get('C',0)} D={s['grades'].get('D',0)}")
|
||
print(f" 分类: {s['categories']} 个")
|
||
print(f" 需关注: {s['needs_attention']} 个")
|
||
print()
|
||
|
||
if report["needs_attention"]:
|
||
print(f" ⚠️ 需关注 (评分<5):")
|
||
for sk in report["needs_attention"][:10]:
|
||
print(f" {sk['grade']} {sk['quality_score']}/10 {sk['name']:30s} [{sk['category']}]")
|
||
print()
|
||
|
||
print(f" 🏆 Top 10:")
|
||
for sk in report["top_skills"]:
|
||
print(f" {sk['grade']} {sk['quality_score']}/10 {sk['name']:30s} v{sk['version']:6s} [{sk['category']}]")
|
||
print()
|
||
|
||
print(f" 📂 分类概览:")
|
||
for cat, info in sorted(report["categories"].items(), key=lambda x: -x[1]["count"]):
|
||
bar = "█" * int(info["avg_score"]) + "░" * (10 - int(info["avg_score"]))
|
||
print(f" {bar} {info['avg_score']}/10 {cat:25s} {info['count']}个")
|
||
|
||
print(f"{'='*55}\n")
|
||
|
||
|
||
def scan():
|
||
skills = scan_skills()
|
||
report = generate_report(skills)
|
||
|
||
os.makedirs(os.path.dirname(REPORT_FILE), exist_ok=True)
|
||
with open(REPORT_FILE, "w") as f:
|
||
json.dump(report, f, indent=2, ensure_ascii=False)
|
||
|
||
print_dashboard(report)
|
||
print(f"报告已保存: {REPORT_FILE}")
|
||
|
||
|
||
def fix_metadata():
|
||
"""自动修复缺失元数据（补 tags、category）"""
|
||
skills = scan_skills()
|
||
active = [s for s in skills if not s["is_archived"]]
|
||
fixed = 0
|
||
|
||
for sk in active:
|
||
path = os.path.join(SKILLS_DIR, sk["path"])
|
||
if not os.path.exists(path):
|
||
continue
|
||
|
||
with open(path, "r") as f:
|
||
content = f.read()
|
||
|
||
changes = []
|
||
|
||
# 补 tags（如果没有）
|
||
if not sk["has_tags"] and "tags:" not in content.split("---")[1] if "---" in content else "":
|
||
# 在 frontmatter 的 metadata 块里加 tags
|
||
tag = sk["category"].replace("-", " ")
|
||
content = content.replace(
|
||
"metadata:", f"metadata:\n tags: [{tag}]", 1
|
||
) if "metadata:" in content else content
|
||
if "metadata:" not in content:
|
||
content = content.replace("---\n", f"---\ntags: [{tag}]\n", 1)
|
||
changes.append("tags")
|
||
|
||
# 补 version（如果没有）
|
||
if not sk.get("version") or sk["version"] == "0.1":
|
||
# version 字段已经有了，只是没写 version: 1.0
|
||
pass
|
||
|
||
if changes:
|
||
with open(path, "w") as f:
|
||
f.write(content)
|
||
print(f" ✅ {sk['name']}: 补了 {', '.join(changes)}")
|
||
fixed += 1
|
||
|
||
print(f"\n修复完成: {fixed} 个 skill")
|
||
|
||
|
||
|
||
def audit():
|
||
"""审计：标记长期不用的 skill → 建议归档"""
|
||
skills = scan_skills()
|
||
active = [s for s in skills if not s["is_archived"]]
|
||
|
||
# 检查技能的使用情况（从 usage.json 或访问时间）
|
||
usage_file = os.path.join(HOME, ".hermes", "skills", ".usage.json")
|
||
usage_data = {}
|
||
if os.path.exists(usage_file):
|
||
with open(usage_file) as f:
|
||
try:
|
||
usage_data = json.load(f)
|
||
except:
|
||
pass
|
||
|
||
now = time.time()
|
||
suggest_archive = []
|
||
|
||
for sk in active:
|
||
# 低分 + 无引用 + 无脚本 = 候选
|
||
if sk["quality_score"] < 4 and not sk["has_refs"] and not sk["has_scripts"]:
|
||
suggest_archive.append(sk)
|
||
continue
|
||
|
||
# 检查使用频率
|
||
name = sk["name"]
|
||
if name in usage_data:
|
||
last_used = usage_data[name].get("last_used", 0)
|
||
if isinstance(last_used, str):
|
||
try:
|
||
last_used = datetime.fromisoformat(last_used).timestamp()
|
||
except:
|
||
last_used = 0
|
||
age_days = (now - last_used) / 86400
|
||
if age_days > 90 and sk["quality_score"] < 6:
|
||
suggest_archive.append(sk)
|
||
|
||
if suggest_archive:
|
||
print(f"\n⚠️ 建议归档 ({len(suggest_archive)} 个):")
|
||
for sk in sorted(suggest_archive, key=lambda x: x["quality_score"]):
|
||
print(f" {sk['quality_score']}/10 {sk['name']:30s} [{sk['category']}]")
|
||
print(f"\n 执行: skill-manager.py archive <skill_name>")
|
||
else:
|
||
print("\n✅ 无建议归档的 skill")
|
||
|
||
return suggest_archive
|
||
|
||
|
||
def archive_skill(name):
|
||
"""归档一个 skill"""
|
||
# 找 skill 目录
|
||
for root, dirs, files in os.walk(SKILLS_DIR):
|
||
if "SKILL.md" in files and name in root:
|
||
rel = os.path.relpath(root, SKILLS_DIR)
|
||
# 目标: skills/.archive/<category>/<name>/
|
||
archive_path = os.path.join(ARCHIVE_DIR, rel)
|
||
os.makedirs(os.path.dirname(archive_path), exist_ok=True)
|
||
os.rename(root, archive_path)
|
||
print(f"📦 已归档: {rel}")
|
||
return True
|
||
|
||
print(f"❌ 未找到 skill: {name}")
|
||
return False
|
||
|
||
|
||
if __name__ == "__main__":
|
||
cmd = sys.argv[1] if len(sys.argv) > 1 else "scan"
|
||
|
||
if cmd == "scan":
|
||
scan()
|
||
elif cmd == "dashboard":
|
||
if os.path.exists(REPORT_FILE):
|
||
with open(REPORT_FILE) as f:
|
||
print_dashboard(json.load(f))
|
||
else:
|
||
scan()
|
||
elif cmd == "fix":
|
||
fix_metadata()
|
||
elif cmd == "audit":
|
||
audit()
|
||
elif cmd == "archive":
|
||
if len(sys.argv) > 2:
|
||
archive_skill(sys.argv[2])
|
||
else:
|
||
print("用法: skill-manager.py archive <skill_name>")
|
||
else:
|
||
print(f"未知命令: {cmd}")
|
||
print("可用: scan, dashboard, fix, audit, archive <name>")
|