diff --git a/pipeline_sdlc/adapter.py b/pipeline_sdlc/adapter.py index d94d49e..efbf857 100644 --- a/pipeline_sdlc/adapter.py +++ b/pipeline_sdlc/adapter.py @@ -8,6 +8,13 @@ logger = logging.getLogger(__name__) def load_sdlc_adapter(): """Register all SDLC step types and handlers with pipeline-service engine.""" + from ahserver.serverenv import ServerEnv + from .skill_utils import guess_role, load_skills, import_skill, list_imported_skills + ServerEnv().guess_role = guess_role + ServerEnv().load_skills = load_skills + ServerEnv().import_skill = import_skill + ServerEnv().list_imported_skills = list_imported_skills + from pipeline_service.step_registry import register_step_type from pipeline_service.handler import register_handler diff --git a/pipeline_sdlc/skill_utils.py b/pipeline_sdlc/skill_utils.py new file mode 100644 index 0000000..b48c158 --- /dev/null +++ b/pipeline_sdlc/skill_utils.py @@ -0,0 +1,105 @@ +"""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