feat: register skill_utils functions (guess_role,load_skills,import_skill,list_imported_skills) via ServerEnv

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
 

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