docker-configs/discord_bot/utils/claude_code_client.py

import asyncio
import logging
import subprocess
import os
import json
from typing import Optional, Dict, List

logger = logging.getLogger(__name__)

class ClaudeCodeClient:
    """Claude Code CLI客户端"""
    
    def __init__(self, work_dir: str = "/home/will/docker"):
        self.work_dir = work_dir
        self.claude_cli = "claude"  # 假设claude命令已安装
        
    async def run_claude_command(self, command: str, timeout: int = 30) -> Optional[str]:
        """运行Claude Code CLI命令"""
        try:
            # 检查Claude CLI是否可用
            check_process = await asyncio.create_subprocess_shell(
                "which claude",
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE
            )
            await check_process.communicate()
            
            if check_process.returncode != 0:
                logger.warning("Claude CLI未找到，回退到shell命令执行")
                return await self._fallback_shell_execution(command)
            
            # 构建Claude CLI命令
            claude_command = f'claude "{command}"'
            
            # 执行Claude CLI
            process = await asyncio.create_subprocess_shell(
                claude_command,
                cwd=self.work_dir,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE
            )
            
            stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=timeout)
            
            output = ""
            if stdout:
                output += stdout.decode('utf-8', errors='replace')
            if stderr:
                output += "\n" + stderr.decode('utf-8', errors='replace')
            
            if process.returncode == 0:
                return output.strip()
            else:
                logger.error(f"Claude CLI执行失败: {output}")
                return await self._fallback_shell_execution(command)
                
        except asyncio.TimeoutError:
            logger.error(f"Claude CLI超时: {command}")
            return f"⏰ **Claude CLI超时**: 执行时间超过{timeout}秒"
        except Exception as e:
            logger.error(f"Claude CLI执行错误: {e}")
            return await self._fallback_shell_execution(command)
    
    async def _fallback_shell_execution(self, command: str) -> str:
        """回退到shell命令执行"""
        try:
            # 检查是否为直接shell命令
            if self._is_shell_command(command):
                return await self._execute_shell_command(command)
            else:
                # 尝试解释自然语言命令
                return await self._interpret_natural_language(command)
                
        except Exception as e:
            return f"❌ **执行错误**: {str(e)}"
    
    def _is_shell_command(self, command: str) -> bool:
        """判断是否为shell命令"""
        shell_commands = [
            'ls', 'pwd', 'cd', 'cat', 'grep', 'find', 'docker', 'docker-compose',
            'tail', 'head', 'ps', 'df', 'du', 'top', 'htop', 'systemctl', 'service'
        ]
        
        cmd_parts = command.strip().split()
        if cmd_parts:
            return cmd_parts[0] in shell_commands
        return False
    
    async def _execute_shell_command(self, command: str) -> str:
        """直接执行shell命令"""
        try:
            process = await asyncio.create_subprocess_shell(
                command,
                cwd=self.work_dir,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE
            )
            
            stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=30)
            
            output = ""
            if stdout:
                output += stdout.decode('utf-8', errors='replace')
            if stderr:
                output += "\n" + stderr.decode('utf-8', errors='replace')
            
            if process.returncode == 0:
                return f"✅ **命令执行成功**\n```bash\n$ {command}\n{output}\n```"
            else:
                return f"⚠️ **命令执行完成 (退出码: {process.returncode})**\n```bash\n$ {command}\n{output}\n```"
                
        except asyncio.TimeoutError:
            return f"⏰ **命令超时**: 执行时间超过30秒\n```bash\n$ {command}\n```"
        except Exception as e:
            return f"❌ **执行错误**: {str(e)}\n```bash\n$ {command}\n```"
    
    async def _interpret_natural_language(self, command: str) -> str:
        """解释自然语言命令（简单映射）"""
        command_lower = command.lower()
        
        # 自然语言到shell命令的映射
        mappings = {
            "查看当前目录": "ls -la",
            "查看文件": "ls -la", 
            "当前目录": "pwd",
            "当前位置": "pwd",
            "查看容器": "docker ps",
            "docker容器": "docker ps",
            "容器状态": "docker ps",
            "查看进程": "ps aux",
            "磁盘使用": "df -h",
            "存储空间": "df -h",
            "内存使用": "free -h",
            "系统状态": "systemctl status"
        }
        
        for key, cmd in mappings.items():
            if key in command_lower:
                return await self._execute_shell_command(cmd)
        
        # 尝试提取Docker相关操作
        if "启动" in command and "容器" in command:
            # 尝试提取容器名
            words = command.split()
            for word in words:
                if word not in ["启动", "容器", "的"]:
                    container_name = word
                    return await self._execute_shell_command(f"docker start {container_name}")
        
        if "停止" in command and "容器" in command:
            words = command.split()
            for word in words:
                if word not in ["停止", "容器", "的"]:
                    container_name = word
                    return await self._execute_shell_command(f"docker stop {container_name}")
        
        # 未识别的命令
        return f"""🤔 **未能理解命令**: {command}

**建议使用以下格式:**
• 直接shell命令: `ls`, `docker ps`, `pwd`
• 自然语言: "查看当前目录", "查看Docker容器状态"

**常用命令示例:**
```bash
ls -la                    # 查看文件列表
docker ps                 # 查看容器状态
docker-compose ps         # 查看compose服务
df -h                     # 查看磁盘使用
```

💡 **提示**: 当前使用简化的命令解释器，建议直接使用shell命令获得最佳效果"""
    
    async def ask_claude(self, question: str) -> str:
        """使用Claude Code CLI进行对话"""
        try:
            # 首先检查是否有Claude CLI
            check_process = await asyncio.create_subprocess_shell(
                "which claude",
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE
            )
            await check_process.communicate()
            
            if check_process.returncode != 0:
                return "❌ **Claude CLI未安装**: 请先安装Claude Code CLI\n\n可以访问 https://docs.anthropic.com/en/docs/claude-code 了解安装方法"
            
            # 使用Claude CLI进行对话
            claude_command = f'claude "{question}"'
            
            process = await asyncio.create_subprocess_shell(
                claude_command,
                cwd=self.work_dir,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE
            )
            
            stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=60)
            
            if process.returncode == 0 and stdout:
                response = stdout.decode('utf-8', errors='replace').strip()
                return f"🤖 **Claude Code回答:**\n\n{response}"
            else:
                error_msg = stderr.decode('utf-8', errors='replace') if stderr else "未知错误"
                return f"❌ **Claude CLI错误**: {error_msg}"
                
        except asyncio.TimeoutError:
            return f"⏰ **Claude CLI超时**: 响应时间超过60秒"
        except Exception as e:
            logger.error(f"Claude CLI对话失败: {e}")
            return f"❌ **对话错误**: {str(e)}"