Claude Code failure排除 and solution

#!/usr/bin/env python3
"""
Claude Code API errorprocessingexample
"""

import requests
import time

def call_claude_code(prompt, api_key):
    """
    调用Claude Code API并processingerror
    
    Args:
        prompt (str): 提示词
        api_key (str): Claude Code APIkey
        
    Returns:
        dict: APIresponse or errorinformation
    """
    url = "https://api.anthropic.com/v1/messages"
    headers = {
        "Content-Type": "application/json",
        "x-api-key": api_key
    }
    data = {
        "model": "claude-3-opus-20240229",
        "messages": [
            {
                "role": "user",
                "content": prompt
            }
        ],
        "max_tokens": 1000
    }
    
    max_retries = 3
    retry_delay = 1
    
    for attempt in range(max_retries):
        try:
            print(f"Attempt {attempt + 1}/{max_retries}...")
            response = requests.post(url, headers=headers, json=data, timeout=30)
            
            # checkresponsestatus码
            if response.status_code == 200:
                return response.json()
            elif response.status_code == 401:
                return {"error": "authentication失败, 请checkAPIkey"}
            elif response.status_code == 403:
                return {"error": "permission不足, 请checkAPIkeypermission"}
            elif response.status_code == 429:
                print("request频率过 high , 正 in 重试...")
                time.sleep(retry_delay * (2 ** attempt))
                continue
            elif response.status_code == 500:
                print("server in 部error, 正 in 重试...")
                time.sleep(retry_delay * (2 ** attempt))
                continue
            else:
                return {"error": f"request失败, status码: {response.status_code}"}
                
        except requests.exceptions.Timeout:
            print("request超时, 正 in 重试...")
            time.sleep(retry_delay * (2 ** attempt))
            continue
        except requests.exceptions.ConnectionError:
            print("network连接error, 正 in 重试...")
            time.sleep(retry_delay * (2 ** attempt))
            continue
        except Exception as e:
            return {"error": f"发生未知error: {str(e)}"}
    
    return {"error": "已达 to 最 big 重试次数, request失败"}

# usingexample
if __name__ == "__main__":
    api_key = "your-api-key"
    prompt = "creation一个 simple  Pythonfunction, 计算斐波那契数列"
    
    result = call_claude_code(prompt, api_key)
    
    if "error" in result:
        print(f"error: {result['error']}")
    else:
        print("成功!")
        if "content" in result:
            for item in result['content']:
                if item['type'] == 'text':
                    print(item['text'])

#!/usr/bin/env python3
"""
Claude Code performancemonitorexample
"""

import time
import psutil
import requests

def monitor_performance(prompt, api_key):
    """
    monitorClaude Code API调用 performance
    
    Args:
        prompt (str): 提示词
        api_key (str): Claude Code APIkey
        
    Returns:
        dict: performancemonitor结果
    """
    # 记录开始时间 and systemstatus
    start_time = time.time()
    start_cpu = psutil.cpu_percent(interval=1)
    start_memory = psutil.virtual_memory().used / (1024 * 1024 * 1024)  # GB
    
    # 调用Claude Code API
    url = "https://api.anthropic.com/v1/messages"
    headers = {
        "Content-Type": "application/json",
        "x-api-key": api_key
    }
    data = {
        "model": "claude-3-opus-20240229",
        "messages": [
            {
                "role": "user",
                "content": prompt
            }
        ],
        "max_tokens": 1000
    }
    
    response = None
    error = None
    
    try:
        response = requests.post(url, headers=headers, json=data, timeout=60)
        response.raise_for_status()
    except Exception as e:
        error = str(e)
    
    # 记录结束时间 and systemstatus
    end_time = time.time()
    end_cpu = psutil.cpu_percent(interval=1)
    end_memory = psutil.virtual_memory().used / (1024 * 1024 * 1024)  # GB
    
    # 计算performance指标
    execution_time = end_time - start_time
    cpu_using = (start_cpu + end_cpu) / 2
    memory_using = end_memory - start_memory
    
    # 构建结果
    result = {
        "execution_time": execution_time,
        "cpu_using": cpu_using,
        "memory_using": memory_using,
        "prompt_length": len(prompt),
        "error": error
    }
    
    if response and response.status_code == 200:
        content = response.json()
        if "content" in content:
            result["response_length"] = sum(len(item.get("text", "")) for item in content["content"] if item.get("type") == "text")
        result["status_code"] = response.status_code
    elif response:
        result["status_code"] = response.status_code
    
    return result

# usingexample
if __name__ == "__main__":
    api_key = "your-api-key"
    
    # test不同 long 度 提示词
    test_prompts = [
        "creation一个 simple  Pythonfunction",
        "creation一个Pythonfunction, 计算斐波那契数列, 并package含documentationstring and test用例",
        "creation一个完整 Pythonclass, implementation一个栈datastructure, package含push, pop, peek, is_emptyetc.method, 以及documentationstring and 详细 test用例"
    ]
    
    for i, prompt in enumerate(test_prompts):
        print(f"\ntest {i+1}: 提示词 long 度 = {len(prompt)}")
        result = monitor_performance(prompt, api_key)
        
        print(f"执行时间: {result['execution_time']:.2f} 秒")
        print(f"CPUusing率: {result['cpu_using']:.2f}%")
        print(f"memoryusing变化: {result['memory_using']:.2f} GB")
        if "response_length" in result:
            print(f"response long 度: {result['response_length']} 字符")
        if "error" in result and result["error"]:
            print(f"error: {result['error']}")
        if "status_code" in result:
            print(f"status码: {result['status_code']}")

# security Claude Codeconfigurationexample

# 1. usingenvironmentvariablestoreAPIkey
# .envfile
ANTHROPIC_API_KEY=your-api-key

# 2.  in codeinusingenvironmentvariable
import os
from dotenv import load_dotenv

# 加载environmentvariable
load_dotenv()

# 获取APIkey
api_key = os.getenv("ANTHROPIC_API_KEY")

# 3. usingAPIkey security实践
import requests

def secure_api_call(prompt):
    """
    security调用Claude Code API
    """
    url = "https://api.anthropic.com/v1/messages"
    headers = {
        "Content-Type": "application/json",
        "x-api-key": api_key
    }
    
    # clean提示词in 敏感information
    sanitized_prompt = sanitize_prompt(prompt)
    
    data = {
        "model": "claude-3-opus-20240229",
        "messages": [
            {
                "role": "user",
                "content": sanitized_prompt
            }
        ],
        "max_tokens": 1000,
        "temperature": 0.7
    }
    
    try:
        response = requests.post(url, headers=headers, json=data, timeout=30)
        response.raise_for_status()
        
        # verificationresponse
        content = response.json()
        
        # 审查生成 code
        if "content" in content:
            for item in content["content"]:
                if item.get("type") == "text":
                    # security审查
                    if contains_secure_code(item.get("text", "")):
                        return content
                    else:
                        return {"error": "生成 code未throughsecurity审查"}
        
        return content
        
    except Exception as e:
        # 记录error但不暴露敏感information
        print(f"API调用error: {str(e)}")
        return {"error": "API调用失败"}

def sanitize_prompt(prompt):
    """
    clean提示词in 敏感information
    """
    # 移除APIkey
    import re
    prompt = re.sub(r'(api[_\s-]?key|api[_\s-]?secret)=[\w\d-]+', r'\1=***', prompt)
    
    # 移除邮箱地址
    prompt = re.sub(r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}', '***@***.***', prompt)
    
    # 移除电话号码
    prompt = re.sub(r'\b\d{11}\b', '***-***-****', prompt)
    
    # 移除身份证号码
    prompt = re.sub(r'\b\d{17}[\dXx]\b', '*****************', prompt)
    
    return prompt

def contains_secure_code(code):
    """
    checkcode is 否package含securityissues
    """
    # check恶意code模式
    dangerous_patterns = [
        r'eval\(',
        r'exec\(',
        r'__import__\(["\']os["\']\)',
        r'open\(["\'].*["\'],\s*["\']w["\']\)',
        r'subprocess\.Popen',
        r'os\.system',
        r'os\.popen',
        r'pickle\.load',
        r'marshal\.load',
        r'input\(',
        r'raw_input\(',
        r'open\(["\'].*["\'],\s*["\']r["\']\)',
    ]
    
    import re
    for pattern in dangerous_patterns:
        if re.search(pattern, code, re.IGNORECASE):
            print(f"检测 to 潜 in  securityissues: {pattern}")
            return False
    
    return True

# usingexample
if __name__ == "__main__":
    prompt = "creation一个Pythonfunction, 连接 to datalibrary并queryuserinformation"
    result = secure_api_call(prompt)
    
    if "error" in result:
        print(f"error: {result['error']}")
    else:
        print("成功!")
        if "content" in result:
            for item in result['content']:
                if item['type'] == 'text':
                    print(item['text'])

# Claude Code failure排除toolpackage

## 1. systemcheck脚本
#!/usr/bin/env python3
"""
Claude Code systemcheck脚本
"""

import sys
import os
import platform
import subprocess
import json

def check_system():
    """checksystemenvironment"""
    print("=== systemenvironmentcheck ===")
    
    # systeminformation
    print(f"operationsystem: {platform.system()} {platform.release()}")
    print(f"Pythonversion: {sys.version}")
    print(f"systemarchitecture: {platform.architecture()[0]}")
    print(f"processing器: {platform.processor()}")
    
    # memoryinformation
    try:
        import psutil
        memory = psutil.virtual_memory()
        print(f"总memory: {memory.total / (1024**3):.2f} GB")
        print(f"可用memory: {memory.available / (1024**3):.2f} GB")
        print(f"memoryusing率: {memory.percent}%")
    except ImportError:
        print("psutilmodule未installation, 无法获取memoryinformation")
    
    # diskinformation
    try:
        if platform.system() == "Windows":
            result = subprocess.run(["wmic", "logicaldisk", "get", "size,freespace,caption"], 
                                  capture_output=True, text=True)
            print("diskinformation:")
            print(result.stdout)
        else:
            result = subprocess.run(["df", "-h"], capture_output=True, text=True)
            print("diskinformation:")
            print(result.stdout)
    except Exception as e:
        print(f"获取diskinformation失败: {str(e)}")
    
    # networkinformation
    print("\n=== network连接check ===")
    try:
        # checkClaude API连接
        import requests
        response = requests.get("https://api.anthropic.com/v1", timeout=10)
        print(f"Claude API连接: {'成功' if response.status_code == 401 else '失败'}")
        print(f"status码: {response.status_code}")
    except Exception as e:
        print(f"Claude API连接失败: {str(e)}")
    
    # checknetworklatency
    try:
        if platform.system() == "Windows":
            result = subprocess.run(["ping", "-n", "4", "api.anthropic.com"], 
                                  capture_output=True, text=True)
        else:
            result = subprocess.run(["ping", "-c", "4", "api.anthropic.com"], 
                                  capture_output=True, text=True)
        print("\nnetworklatencytest:")
        print(result.stdout)
    except Exception as e:
        print(f"networklatencytest失败: {str(e)}")

def check_claude_code():
    """checkClaude Codeinstallation"""
    print("\n=== Claude Code installationcheck ===")
    
    # check is 否installation
    try:
        # checkPythonpackage
        import importlib.util
        spec = importlib.util.find_spec("claude_code")
        if spec:
            print("Claude Code Pythonpackage已installation")
        else:
            print("Claude Code Pythonpackage未installation")
    except Exception as e:
        print(f"checkPythonpackage失败: {str(e)}")
    
    # checkenvironmentvariable
    print("\n=== environmentvariablecheck ===")
    api_key = os.environ.get("ANTHROPIC_API_KEY")
    if api_key:
        print(f"ANTHROPIC_API_KEY: {'已设置 (部分隐藏)' if api_key else '未设置'}")
        if api_key:
            print(f"APIkey long 度: {len(api_key)}")
    else:
        print("ANTHROPIC_API_KEY: 未设置")

def main():
    """主function"""
    print("Claude Code systemchecktool")
    print("=" * 50)
    
    check_system()
    check_claude_code()
    
    print("\n=== checkcompletion ===")
    print("请根据以 on information排查issues")

if __name__ == "__main__":
    main()

## 2. loganalysistool
#!/usr/bin/env python3
"""
Claude Code loganalysistool
"""

import re
import datetime
from collections import Counter

def analyze_log(log_file):
    """analysisClaude Codelogfile"""
    try:
        with open(log_file, 'r', encoding='utf-8') as f:
            logs = f.readlines()
    except Exception as e:
        print(f"读取logfile失败: {str(e)}")
        return
    
    print(f"analysislogfile: {log_file}")
    print(f"log行数: {len(logs)}")
    
    # 提取errorinformation
    errors = []
    warnings = []
    info = []
    
    error_pattern = re.compile(r'(ERROR|Error|error)', re.IGNORECASE)
    warning_pattern = re.compile(r'(WARNING|Warning|warning)', re.IGNORECASE)
    
    for line in logs:
        if error_pattern.search(line):
            errors.append(line.strip())
        elif warning_pattern.search(line):
            warnings.append(line.strip())
        else:
            info.append(line.strip())
    
    print(f"error数量: {len(errors)}")
    print(f"warning数量: {len(warnings)}")
    print(f"information数量: {len(info)}")
    
    # analysiserrorclass型
    if errors:
        print("\n=== erroranalysis ===")
        error_types = Counter()
        for error in errors:
            # 提取errorclass型
            match = re.search(r'(ERROR|Error|error):\s*(.*?)(?:\s*at|$)', error)
            if match:
                error_type = match.group(2).strip()[:100]
                error_types[error_type] += 1
        
        # 显示 before 10个最common error
        print("最common error:")
        for error_type, count in error_types.most_common(10):
            print(f"{count}: {error_type}")
    
    # analysis时间分布
    print("\n=== 时间分布analysis ===")
    time_pattern = re.compile(r'(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})')
    timestamps = []
    
    for line in logs:
        match = time_pattern.search(line)
        if match:
            try:
                timestamp = datetime.datetime.strptime(match.group(1), '%Y-%m-%d %H:%M:%S')
                timestamps.append(timestamp)
            except Exception:
                pass
    
    if timestamps:
        # 按 small 时statistics
        hour_counts = Counter()
        for timestamp in timestamps:
            hour_key = timestamp.strftime('%Y-%m-%d %H:00')
            hour_counts[hour_key] += 1
        
        print("log时间分布:")
        for hour, count in sorted(hour_counts.items()):
            print(f"{hour}: {count} 条")
    
    # 保存analysis结果
    analysis_file = f"log_analysis_{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.txt"
    with open(analysis_file, 'w', encoding='utf-8') as f:
        f.write(f"loganalysis结果\n")
        f.write(f"analysis时间: {datetime.datetime.now()}\n")
        f.write(f"logfile: {log_file}\n")
        f.write(f"log行数: {len(logs)}\n")
        f.write(f"error数量: {len(errors)}\n")
        f.write(f"warning数量: {len(warnings)}\n")
        f.write(f"information数量: {len(info)}\n\n")
        
        if errors:
            f.write("=== errorlist ===\n")
            for error in errors[:50]:  # 只保存 before 50个error
                f.write(f"{error}\n")
    
    print(f"\nanalysis结果已保存 to : {analysis_file}")

def main():
    """主function"""
    import sys
    if len(sys.argv) != 2:
        print("用法: python analyze_log.py ")
        return
    
    log_file = sys.argv[1]
    analyze_log(log_file)

if __name__ == "__main__":
    main()