106 lines
3.9 KiB
Python
106 lines
3.9 KiB
Python
"""Skill management utilities for SDLC pipeline."""
|
|
import os
|
|
|
|
|
|
def guess_role(title):
|
|
"""Guess agent role from task title keywords."""
|
|
title_lower = (title or '').lower()
|
|
keywords = {
|
|
'design': ['设计', 'design', '架构', '方案', '需求', '原型', 'ui', 'ux'],
|
|
'develop': ['开发', '编码', '实现', '编写', 'develop', 'code', 'build', '重构', '修复'],
|
|
'test': ['测试', 'test', '验证', '检查', 'review', '评审'],
|
|
'deploy': ['部署', '发布', 'deploy', 'release', '上线', '配置'],
|
|
'ops': ['运维', '监控', 'ops', '日志', '备份', '迁移'],
|
|
}
|
|
for role, kws in keywords.items():
|
|
for kw in kws:
|
|
if kw in title_lower:
|
|
return role
|
|
return 'develop'
|
|
|
|
|
|
def load_skills(skills_dir, role=None):
|
|
"""Scan skills_dir for SKILL.md files. If role given, load common/ + {role}/ only."""
|
|
skills = []
|
|
if not skills_dir:
|
|
return skills
|
|
dirs_to_scan = ['common']
|
|
if role:
|
|
dirs_to_scan.append(role)
|
|
else:
|
|
try:
|
|
dirs_to_scan = [d for d in os.listdir(skills_dir)
|
|
if os.path.isdir(os.path.join(skills_dir, d))]
|
|
except Exception:
|
|
return skills
|
|
for subdir in dirs_to_scan:
|
|
subdir_path = os.path.join(skills_dir, subdir)
|
|
if not os.path.isdir(subdir_path):
|
|
continue
|
|
try:
|
|
for name in os.listdir(subdir_path):
|
|
skill_path = os.path.join(subdir_path, name)
|
|
skill_md = os.path.join(skill_path, 'SKILL.md')
|
|
if os.path.isdir(skill_path) and os.path.isfile(skill_md):
|
|
try:
|
|
with open(skill_md, 'r') as f:
|
|
content = f.read()
|
|
if len(content) > 8000:
|
|
content = content[:8000] + '\n\n... (truncated)'
|
|
skills.append({'name': name, 'role': subdir, 'content': content})
|
|
except Exception:
|
|
pass
|
|
except Exception:
|
|
pass
|
|
return skills
|
|
|
|
|
|
def import_skill(skills_dir, source_path, role=None):
|
|
"""Import a SKILL.md file into skills_dir/{role}/{name}/SKILL.md.
|
|
Returns (success, message, skill_name)."""
|
|
if not skills_dir:
|
|
return False, 'skills_dir not configured', ''
|
|
src = os.path.abspath(source_path)
|
|
if not os.path.exists(src):
|
|
return False, '路径不存在: ' + source_path, ''
|
|
if os.path.isfile(src) and src.endswith('.md'):
|
|
skill_name = os.path.splitext(os.path.basename(src))[0]
|
|
with open(src, 'r') as f:
|
|
content = f.read()
|
|
elif os.path.isdir(src):
|
|
md = os.path.join(src, 'SKILL.md')
|
|
if not os.path.isfile(md):
|
|
return False, '目录中未找到 SKILL.md: ' + src, ''
|
|
skill_name = os.path.basename(src)
|
|
with open(md, 'r') as f:
|
|
content = f.read()
|
|
else:
|
|
return False, '源文件必须是 .md 文件或包含 SKILL.md 的目录', ''
|
|
if not role:
|
|
role = guess_role(skill_name)
|
|
dest_dir = os.path.join(skills_dir, role, skill_name)
|
|
os.makedirs(dest_dir, exist_ok=True)
|
|
dest = os.path.join(dest_dir, 'SKILL.md')
|
|
with open(dest, 'w') as f:
|
|
f.write(content)
|
|
return True, '已导入 ' + role + '/' + skill_name, skill_name
|
|
|
|
|
|
def list_imported_skills(skills_dir):
|
|
"""List all skills currently in skills_dir."""
|
|
result = {}
|
|
if not skills_dir or not os.path.isdir(skills_dir):
|
|
return result
|
|
for role_dir in os.listdir(skills_dir):
|
|
rp = os.path.join(skills_dir, role_dir)
|
|
if not os.path.isdir(rp):
|
|
continue
|
|
skills = []
|
|
for sn in os.listdir(rp):
|
|
sp = os.path.join(rp, sn)
|
|
if os.path.isdir(sp) and os.path.isfile(os.path.join(sp, 'SKILL.md')):
|
|
skills.append(sn)
|
|
if skills:
|
|
result[role_dir] = skills
|
|
return result
|