first build nexus

#recycle/openclaw/yunce/star-tasksender/scripts/intent_parser.py

@@ -0,0 +1,191 @@
+# -*- coding: utf-8 -*-
+"""
+意图解析模块 - Intent Parser
+将自然语言指令转换为结构化任务
+"""
+
+import re
+from typing import Dict, Optional
+
+
+class IntentParser:
+    """意图解析器"""
+    
+    # 指令模式定义
+    PATTERNS = {
+        "code_review": {
+            "keywords": ["审查", "review", "代码审查", "review code"],
+            "target": "yunce",
+            "action": "code_review",
+            "param_extractor": "extract_repo_info"
+        },
+        "deploy": {
+            "keywords": ["部署", "deploy", "发布"],
+            "target": "prometheus",
+            "action": "deploy",
+            "param_extractor": "extract_deploy_info"
+        },
+        "status_check": {
+            "keywords": ["检查状态", "status", "查看状态", "状态"],
+            "target": "atlas",
+            "action": "status_check",
+            "param_extractor": "extract_target"
+        },
+        "data_analysis": {
+            "keywords": ["分析", "analysis", "数据分析"],
+            "target": "atlas",
+            "action": "data_analysis",
+            "param_extractor": "extract_data_info"
+        },
+        "file_operation": {
+            "keywords": ["复制", "移动", "删除", "copy", "move", "delete"],
+            "target": "oracle",
+            "action": "file_operation",
+            "param_extractor": "extract_file_info"
+        }
+    }
+    
+    def parse_intent(self, user_input: str) -> Dict:
+        """
+        解析用户输入，返回结构化意图
+        
+        Args:
+            user_input: 用户输入的自然语言
+            
+        Returns:
+            {
+                "action": "code_review",
+                "target": "yunce",
+                "params": {...}
+            }
+        """
+        user_input = user_input.strip()
+        
+        # 遍历所有模式，匹配关键词
+        for pattern_name, pattern_config in self.PATTERNS.items():
+            for keyword in pattern_config["keywords"]:
+                if keyword in user_input:
+                    # 提取参数
+                    extractor_name = pattern_config["param_extractor"]
+                    extractor = getattr(self, extractor_name)
+                    params = extractor(user_input)
+                    
+                    return {
+                        "action": pattern_config["action"],
+                        "target": pattern_config["target"],
+                        "params": params
+                    }
+        
+        # 无法识别
+        return {
+            "error": "无法理解指令",
+            "original_input": user_input
+        }
+    
+    def extract_repo_info(self, text: str) -> Dict:
+        """从文本中提取仓库信息"""
+        result = {}
+        
+        # 提取仓库名
+        repo_match = re.search(r'(?:仓库|repo|项目)[:\s]*(\S+)', text)
+        if not repo_match:
+            repo_match = re.search(r'(?:审查|review)\s+(\S+)', text)
+        
+        if repo_match:
+            result["repo"] = repo_match.group(1).strip('，。')
+        
+        # 提取分支
+        branch_match = re.search(r'(?:分支|branch)[:\s]*(\S+)', text)
+        if branch_match:
+            result["branch"] = branch_match.group(1).strip('，。')
+        else:
+            result["branch"] = "main"
+            
+        return result
+    
+    def extract_deploy_info(self, text: str) -> Dict:
+        """从文本中提取部署信息"""
+        result = {}
+        
+        # 提取服务名
+        service_match = re.search(r'(?:服务|service|部署)[:\s]*(\S+)', text)
+        if not service_match:
+            service_match = re.search(r'部署\s+(\S+)', text)
+            
+        if service_match:
+            result["service"] = service_match.group(1).strip('，。')
+        
+        # 提取环境
+        if "生产" in text or "prod" in text.lower():
+            result["env"] = "prod"
+        elif "测试" in text or "test" in text.lower():
+            result["env"] = "test"
+        else:
+            result["env"] = "dev"
+            
+        return result
+    
+    def extract_target(self, text: str) -> Dict:
+        """提取目标信息"""
+        result = {}
+        
+        # 尝试提取目标名称
+        target_match = re.search(r'(?:服务器|server|目标)[:\s]*(\S+)', text)
+        if not target_match:
+            target_match = re.search(r'检查\s+(\S+)', text)
+            
+        if target_match:
+            result["target"] = target_match.group(1).strip('，。')
+        
+        return result
+    
+    def extract_data_info(self, text: str) -> Dict:
+        """提取数据分析信息"""
+        result = {}
+        
+        # 提取数据源
+        data_match = re.search(r'(?:数据|data|分析)[:\s]*(\S+)', text)
+        if not data_match:
+            data_match = re.search(r'分析\s+(\S+)', text)
+            
+        if data_match:
+            result["data"] = data_match.group(1).strip('，。')
+            
+        return result
+    
+    def extract_file_info(self, text: str) -> Dict:
+        """提取文件操作信息"""
+        result = {}
+        
+        # 提取文件路径
+        path_match = re.search(r'[:\s](\S+\.\S+)', text)
+        if path_match:
+            result["path"] = path_match.group(1)
+        
+        # 提取操作类型
+        if "复制" in text or "copy" in text.lower():
+            result["operation"] = "copy"
+        elif "移动" in text or "move" in text.lower():
+            result["operation"] = "move"
+        elif "删除" in text or "delete" in text.lower():
+            result["operation"] = "delete"
+            
+        return result
+
+
+# 测试
+if __name__ == "__main__":
+    parser = IntentParser()
+    
+    test_cases = [
+        "帮我审查 my-project 仓库",
+        "部署 test-server 到生产",
+        "检查服务器状态",
+        "分析销售数据"
+    ]
+    
+    for test in test_cases:
+        result = parser.parse_intent(test)
+        print(f"输入: {test}")
+        print(f"输出: {result}")
+        print("-" * 40)