07 - 实战项目¶

"""
AI代码助手 - 项目初始化
"""

import os
import json
from typing import Dict, List
from dataclasses import dataclass

@dataclass
class ProjectConfig:
    """项目配置"""
    name: str
    description: str
    language: str
    framework: str
    created_at: str

@dataclass
class CodeFile:
    """代码文件"""
    name: str
    content: str
    language: str
    created_at: str
    modified_at: str

class AIAssistant:
    """AI代码助手"""

    def __init__(self, api_key: str):
        self.api_key = api_key
        self.projects = {}
        self.current_project = None
        self.config_file = "ai_assistant_config.json"

        # 初始化
        self._initialize()

    def _initialize(self):
        """初始化助手"""
        # 创建配置目录
        os.makedirs(".ai_assistant", exist_ok=True)

        # 加载配置
        if os.path.exists(self.config_file):
            with open(self.config_file, 'r', encoding='utf-8') as f:
                config = json.load(f)
                self.projects = config.get("projects", {})
        else:
            self.projects = {}
            self._save_config()

    def _save_config(self):
        """保存配置"""
        config = {
            "projects": self.projects
        }

        with open(self.config_file, 'w', encoding='utf-8') as f:
            json.dump(config, f, indent=2, ensure_ascii=False)

    def create_project(self, name: str, description: str,
                     language: str = "python",
                     framework: str = "") -> str:
        """
        创建新项目

        Args:
            name: 项目名称
            description: 项目描述
            language: 编程语言
            framework: 框架
        """
        from datetime import datetime

        project_id = f"project_{len(self.projects) + 1}"

        project = ProjectConfig(
            name=name,
            description=description,
            language=language,
            framework=framework,
            created_at=datetime.now().isoformat()
        )

        # 创建项目目录
        project_dir = os.path.join(".ai_assistant", project_id)
        os.makedirs(project_dir, exist_ok=True)

        # 保存项目信息
        self.projects[project_id] = {
            "config": project.__dict__,
            "files": {},
            "history": []
        }

        self.current_project = project_id
        self._save_config()

        return project_id

    def list_projects(self) -> List[Dict]:
        """列出所有项目"""
        projects = []

        for project_id, project_data in self.projects.items():
            projects.append({
                "id": project_id,
                **project_data["config"]
            })

        return projects

    def get_project(self, project_id: str) -> Dict:
        """获取项目详情"""
        if project_id not in self.projects:
            return {"error": "Project not found"}

        return self.projects[project_id]

    def delete_project(self, project_id: str) -> bool:
        """删除项目"""
        if project_id not in self.projects:
            return False

        # 删除项目目录
        project_dir = os.path.join(".ai_assistant", project_id)
        if os.path.exists(project_dir):
            import shutil
            shutil.rmtree(project_dir)

        # 删除项目信息
        del self.projects[project_id]
        self._save_config()

        if self.current_project == project_id:
            self.current_project = None

        return True

# 使用示例
if __name__ == "__main__":
    assistant = AIAssistant(api_key="your_api_key_here")

    # 创建项目
    project_id = assistant.create_project(
        name="我的Web应用",
        description="使用Flask构建的Web应用",
        language="python",
        framework="Flask"
    )

    print(f"项目创建成功！ID: {project_id}")

    # 列出项目
    projects = assistant.list_projects()
    print(f"\n所有项目（共{len(projects)}个）:")
    for project in projects:
        print(f"- {project['name']} ({project['id']})")

    # 获取项目详情
    project_detail = assistant.get_project(project_id)
    print(f"\n项目详情:")
    print(json.dumps(project_detail, indent=2, ensure_ascii=False))

import openai
from typing import Dict, List
from datetime import datetime

class CodeGenerationService:
    """代码生成服务"""

    def __init__(self, api_key: str):
        self.client = openai.OpenAI(api_key=api_key)
        self.model = "gpt-4o"

    def generate_code(self, prompt: str,
                     language: str = "python",
                     context: str = "") -> Dict:
        """
        生成代码

        Args:
            prompt: 提示词
            language: 编程语言
            context: 上下文代码
        """
        system_prompt = f"""你是一个专业的{language}程序员。你的任务是：
1. 根据用户需求生成高质量、可运行的代码
2. 代码应该清晰、易读、有良好的注释
3. 遵循{language}的最佳实践和编码规范
4. 包含必要的错误处理
5. 如果需要，提供使用示例
"""

        if context:
            user_prompt = f"""上下文代码：
~~~{language}
{context}
~~~

用户需求：
{prompt}

请基于上下文代码实现用户需求。"""
        else:
            user_prompt = f"""用户需求：
{prompt}

请生成完整的{language}代码实现。"""

        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": system_prompt},
                    {"role": "user", "content": user_prompt}
                ],
                temperature=0.3,
                max_tokens=2000
            )

            return {
                "success": True,
                "code": response.choices[0].message.content,
                "model": self.model,
                "tokens_used": response.usage.total_tokens,
                "timestamp": datetime.now().isoformat()
            }

        except Exception as e:
            return {
                "success": False,
                "error": str(e),
                "timestamp": datetime.now().isoformat()
            }

    def generate_with_tests(self, prompt: str,
                          language: str = "python") -> Dict:
        """
        生成代码和测试

        Args:
            prompt: 提示词
            language: 编程语言
        """
        system_prompt = f"""你是一个专业的{language}程序员和测试专家。你的任务是：
1. 根据用户需求生成高质量、可运行的代码
2. 同时生成对应的单元测试
3. 代码应该清晰、易读、有良好的注释
4. 测试应该覆盖主要功能和边界情况
"""

        user_prompt = f"""用户需求：
{prompt}

请生成：
1. 完整的{language}代码实现
2. 对应的单元测试代码
3. 使用说明

代码和测试请分开标注。"""

        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": system_prompt},
                    {"role": "user", "content": user_prompt}
                ],
                temperature=0.3,
                max_tokens=3000
            )

            content = response.choices[0].message.content

            # 尝试分离代码和测试
            if "测试" in content or "test" in content.lower():
                parts = content.split("测试", 1) if "测试" in content else content.split("Test", 1)
                code = parts[0]
                tests = parts[1] if len(parts) > 1 else ""
            else:
                code = content
                tests = ""

            return {
                "success": True,
                "code": code,
                "tests": tests,
                "language": language,
                "model": self.model,
                "tokens_used": response.usage.total_tokens,
                "timestamp": datetime.now().isoformat()
            }

        except Exception as e:
            return {
                "success": False,
                "error": str(e),
                "timestamp": datetime.now().isoformat()
            }

# 使用示例
if __name__ == "__main__":
    service = CodeGenerationService(api_key="your_api_key_here")

    # 生成代码
    result = service.generate_code(
        prompt="实现一个快速排序算法，支持自定义比较函数",
        language="python"
    )

    if result["success"]:
        print("代码生成成功！")
        print(f"生成的代码：\n{result['code']}")
        print(f"使用的Token数: {result['tokens_used']}")
    else:
        print(f"代码生成失败: {result['error']}")

    # 生成代码和测试
    result = service.generate_with_tests(
        prompt="实现一个计算器类，支持加减乘除运算",
        language="python"
    )

    if result["success"]:
        print("\n\n代码和测试生成成功！")
        print(f"生成的代码：\n{result['code']}")
        print(f"\n生成的测试：\n{result['tests']}")
        print(f"使用的Token数: {result['tokens_used']}")

import openai
from typing import Dict
from datetime import datetime

class CodeReviewService:
    """代码审查服务"""

    def __init__(self, api_key: str):
        self.client = openai.OpenAI(api_key=api_key)
        self.model = "gpt-4o"

    def review_code(self, code: str, language: str = "python") -> Dict:
        """
        审查代码

        Args:
            code: 待审查的代码
            language: 编程语言
        """
        prompt = f"""请全面审查以下{language}代码：

```{language}
{code}
```

请从以下方面进行审查：

1. **代码质量**
   - 可读性
   - 可维护性
   - 命名规范
   - 代码结构

2. **潜在问题**
   - 逻辑错误
   - 边界情况处理
   - 异常处理
   - 资源泄漏

3. **性能问题**
   - 算法复杂度
   - 数据结构选择
   - 性能瓶颈
   - 优化机会

4. **安全问题**
   - 注入漏洞
   - 数据泄露
   - 不安全的操作
   - 权限问题

5. **最佳实践**
   - 设计模式应用
   - 编码规范遵循
   - 文档完整性
   - 测试覆盖

对于每个发现的问题，请提供：
- 问题描述
- 严重程度（高/中/低）
- 具体位置
- 修复建议
- 修复后的代码示例
"""

        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": "你是一个专业的代码审查专家。"},
                    {"role": "user", "content": prompt}
                ],
                temperature=0.3
            )

            return {
                "success": True,
                "review": response.choices[0].message.content,
                "language": language,
                "model": self.model,
                "tokens_used": response.usage.total_tokens,
                "timestamp": datetime.now().isoformat()
            }

        except Exception as e:
            return {
                "success": False,
                "error": str(e),
                "timestamp": datetime.now().isoformat()
            }

    def check_security(self, code: str, language: str = "python") -> Dict:
        """
        安全检查

        Args:
            code: 待检查的代码
            language: 编程语言
        """
        prompt = f"""请检查以下{language}代码的安全问题：

```{language}
{code}
```

请检查以下安全问题：
1. SQL注入
2. XSS攻击
3. CSRF攻击
4. 命令注入
5. 敏感信息泄露
6. 不安全的加密
7. 其他常见安全漏洞

对于每个发现的问题，请提供：
- 问题描述
- 风险等级
- 修复建议
- 修复后的代码
"""

        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": "你是一个专业的安全专家。"},
                    {"role": "user", "content": prompt}
                ],
                temperature=0.2
            )

            return {
                "success": True,
                "security_report": response.choices[0].message.content,
                "language": language,
                "model": self.model,
                "tokens_used": response.usage.total_tokens,
                "timestamp": datetime.now().isoformat()
            }

        except Exception as e:
            return {
                "success": False,
                "error": str(e),
                "timestamp": datetime.now().isoformat()
            }

# 使用示例
if __name__ == "__main__":
    service = CodeReviewService(api_key="your_api_key_here")

    # 示例代码
    code = """
def calculate(a, b, c):
    result = a + b
    result = result * c
    return result

def process_data(data):
    results = []
    for item in data:
        if item > 0:
            processed = item * 2
            results.append(processed)
        else:
            results.append(item)
    return results
"""

    # 代码审查
    print("代码审查：")
    review = service.review_code(code, "python")
    print(review["review"])
    print(f"使用的Token数: {review['tokens_used']}")

    # 安全检查
    print("\n\n安全检查：")
    security = service.check_security(code, "python")
    print(security["security_report"])
    print(f"使用的Token数: {security['tokens_used']}")

import openai
from datetime import datetime
from typing import Dict

class TestGenerationService:
    """测试生成服务"""

    def __init__(self, api_key: str):
        self.client = openai.OpenAI(api_key=api_key)
        self.model = "gpt-4o"

    def generate_unit_tests(self, code: str, language: str = "python") -> Dict:
        """
        生成单元测试

        Args:
            code: 待测试的代码
            language: 编程语言
        """
        prompt = f"""请为以下{language}代码生成完整的单元测试：

~~~{language}
{code}
~~~

请生成：
1. 单元测试
2. 测试每个函数/方法
3. 测试正常情况
4. 测试边界情况
5. 测试异常情况
6. 使用pytest框架
7. 包含测试夹具（如果需要）
8. 添加测试文档字符串
9. 测试应该独立且可重复

请提供完整的、可运行的测试代码。
"""

        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": "你是一个pytest测试专家。"},
                    {"role": "user", "content": prompt}
                ],
                temperature=0.3
            )

            return {
                "success": True,
                "tests": response.choices[0].message.content,
                "language": language,
                "model": self.model,
                "tokens_used": response.usage.total_tokens,
                "timestamp": datetime.now().isoformat()
            }

        except Exception as e:
            return {
                "success": False,
                "error": str(e),
                "timestamp": datetime.now().isoformat()
            }

    def generate_integration_tests(self, api_spec: str) -> Dict:
        """
        生成集成测试

        Args:
            api_spec: API规范
        """
        prompt = f"""请为以下API生成完整的集成测试：

API规范：
{api_spec}

请生成：
1. 集成测试
2. 测试每个端点
3. 测试GET、POST、PUT、DELETE方法
4. 测试成功响应
5. 测试错误响应
6. 测试参数验证
7. 测试认证
8. 测试速率限制
9. 添加测试文档字符串
10. 测试应该独立且可重复

请提供完整的、可运行的测试代码。
"""

        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": "你是一个API测试专家。"},
                    {"role": "user", "content": prompt}
                ],
                temperature=0.3
            )

            return {
                "success": True,
                "tests": response.choices[0].message.content,
                "model": self.model,
                "tokens_used": response.usage.total_tokens,
                "timestamp": datetime.now().isoformat()
            }

        except Exception as e:
            return {
                "success": False,
                "error": str(e),
                "timestamp": datetime.now().isoformat()
            }

# 使用示例
if __name__ == "__main__":
    service = TestGenerationService(api_key="your_api_key_here")

    # 示例代码
    code = """
class Calculator:
    def add(self, a, b):
        return a + b

    def subtract(self, a, b):
        return a - b

    def multiply(self, a, b):
        return a * b

    def divide(self, a, b):
        if b == 0:
            raise ValueError("Cannot divide by zero")
        return a / b
"""

    # 生成单元测试
    print("生成单元测试：")
    tests = service.generate_unit_tests(code, "python")
    print(tests["tests"])
    print(f"使用的Token数: {tests['tokens_used']}")

    # API规范
    api_spec = """
用户管理API：

端点：
1. GET /api/users - 获取所有用户
2. POST /api/users - 创建用户
3. GET /api/users/:id - 获取单个用户
4. PUT /api/users/:id - 更新用户
5. DELETE /api/users/:id - 删除用户
"""

    # 生成集成测试
    print("\n\n生成集成测试：")
    integration_tests = service.generate_integration_tests(api_spec)
    print(integration_tests["tests"])
    print(f"使用的Token数: {integration_tests['tokens_used']}")

import openai
from datetime import datetime
from typing import Dict

class DocumentationService:
    """文档生成服务"""

    def __init__(self, api_key: str):
        self.client = openai.OpenAI(api_key=api_key)
        self.model = "gpt-4o"

    def generate_function_docs(self, code: str, language: str = "python") -> Dict:
        """
        生成函数文档

        Args:
            code: 待文档化的代码
            language: 编程语言
        """
        prompt = f"""请为以下{language}代码生成完整的函数文档：

```{language}
{code}
```

要求：
1. 使用Google风格或NumPy风格的docstring
2. 为每个函数生成文档字符串
3. 包含函数概述
4. 列出所有参数及其类型和说明
5. 说明返回值及其类型
6. 列出可能抛出的异常
7. 添加使用示例
8. 添加注意事项和限制

请提供完整的、规范的文档字符串。
"""

        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": f"你是一个{language}文档专家。"},
                    {"role": "user", "content": prompt}
                ],
                temperature=0.3
            )

            return {
                "success": True,
                "documentation": response.choices[0].message.content,
                "language": language,
                "model": self.model,
                "tokens_used": response.usage.total_tokens,
                "timestamp": datetime.now().isoformat()
            }

        except Exception as e:
            return {
                "success": False,
                "error": str(e),
                "timestamp": datetime.now().isoformat()
            }

    def generate_api_docs(self, api_spec: str) -> Dict:
        """
        生成API文档

        Args:
            api_spec: API规范
        """
        prompt = f"""请为以下API生成完整的API文档：

API规范：
{api_spec}

要求：
1. 使用Markdown格式
2. 包含API概述
3. 列出所有端点及其详细信息
4. 包含请求和响应示例
5. 包含认证说明
6. 包含错误代码说明
7. 添加快速开始指南
8. 添加最佳实践建议

请提供完整的、易读的API文档。
"""

        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": "你是一个API文档专家。"},
                    {"role": "user", "content": prompt}
                ],
                temperature=0.3
            )

            return {
                "success": True,
                "documentation": response.choices[0].message.content,
                "model": self.model,
                "tokens_used": response.usage.total_tokens,
                "timestamp": datetime.now().isoformat()
            }

        except Exception as e:
            return {
                "success": False,
                "error": str(e),
                "timestamp": datetime.now().isoformat()
            }

# 使用示例
if __name__ == "__main__":
    service = DocumentationService(api_key="your_api_key_here")

    # 示例代码
    code = """
def calculate_average(numbers):
    if not numbers:
        raise ValueError("Numbers list cannot be empty")
    return sum(numbers) / len(numbers)

def find_max(numbers):
    if not numbers:
        return None
    return max(numbers)
"""

    # 生成函数文档
    print("生成函数文档：")
    docs = service.generate_function_docs(code, "python")
    print(docs["documentation"])
    print(f"使用的Token数: {docs['tokens_used']}")

    # API规范
    api_spec = """
用户管理API：

端点：
- GET /api/users - 获取所有用户
- POST /api/users - 创建用户
- GET /api/users/:id - 获取单个用户
- PUT /api/users/:id - 更新用户
- DELETE /api/users/:id - 删除用户

认证：JWT Bearer Token
响应格式：JSON
"""

    # 生成API文档
    print("\n\n生成API文档：")
    api_docs = service.generate_api_docs(api_spec)
    print(api_docs["documentation"])
    print(f"使用的Token数: {api_docs['tokens_used']}")

"""
AI代码助手 - 完整应用
"""

from flask import Flask, request, jsonify
from typing import Dict, List
from datetime import datetime
import os

app = Flask(__name__)

# 服务实例
code_generation_service = CodeGenerationService(os.getenv("OPENAI_API_KEY"))
code_review_service = CodeReviewService(os.getenv("OPENAI_API_KEY"))
test_generation_service = TestGenerationService(os.getenv("OPENAI_API_KEY"))
documentation_service = DocumentationService(os.getenv("OPENAI_API_KEY"))

@app.route('/api/generate', methods=['POST'])
def generate_code():
    """生成代码"""
    data = request.json

    prompt = data.get('prompt', '')
    language = data.get('language', 'python')
    context = data.get('context', '')

    result = code_generation_service.generate_code(prompt, language, context)

    return jsonify(result)

@app.route('/api/review', methods=['POST'])
def review_code():
    """审查代码"""
    data = request.json

    code = data.get('code', '')
    language = data.get('language', 'python')

    result = code_review_service.review_code(code, language)

    return jsonify(result)

@app.route('/api/generate-tests', methods=['POST'])
def generate_tests():
    """生成测试"""
    data = request.json

    code = data.get('code', '')
    language = data.get('language', 'python')

    result = test_generation_service.generate_unit_tests(code, language)

    return jsonify(result)

@app.route('/api/generate-docs', methods=['POST'])
def generate_docs():
    """生成文档"""
    data = request.json

    code = data.get('code', '')
    language = data.get('language', 'python')

    result = documentation_service.generate_function_docs(code, language)

    return jsonify(result)

@app.route('/api/batch', methods=['POST'])
def batch_process():
    """批量处理"""
    data = request.json

    operations = data.get('operations', [])
    results = []

    for operation in operations:
        op_type = operation.get('type')

        if op_type == 'generate':
            result = code_generation_service.generate_code(
                operation.get('prompt', ''),
                operation.get('language', 'python'),
                operation.get('context', '')
            )
        elif op_type == 'review':
            result = code_review_service.review_code(
                operation.get('code', ''),
                operation.get('language', 'python')
            )
        elif op_type == 'test':
            result = test_generation_service.generate_unit_tests(
                operation.get('code', ''),
                operation.get('language', 'python')
            )
        elif op_type == 'docs':
            result = documentation_service.generate_function_docs(
                operation.get('code', ''),
                operation.get('language', 'python')
            )
        else:
            result = {
                "success": False,
                "error": f"Unknown operation type: {op_type}"
            }

        results.append({
            "operation": op_type,
            "result": result
        })

    return jsonify({
        "success": True,
        "results": results,
        "timestamp": datetime.now().isoformat()
    })

@app.route('/health', methods=['GET'])
def health_check():
    """健康检查"""
    return jsonify({
        "status": "healthy",
        "timestamp": datetime.now().isoformat()
    })

if __name__ == '__main__':
    app.run(debug=True, host='0.0.0.0', port=5000)