🐍 Day04 Practice Validator

#!/usr/bin/env python3
"""
Day 4 练习验收脚本
验收内容：Function Calling LLM 客户端

使用方法：
    python day04_practice_validator.py
    python day04_practice_validator.py --file day04_practice.py
"""

import ast
import sys
import time
import subprocess
from pathlib import Path
from typing import List, Tuple

TIMEOUT = 30


class C:
    GREEN = "\033[92m"
    RED = "\033[91m"
    YELLOW = "\033[93m"
    BLUE = "\033[94m"
    CYAN = "\033[96m"
    BOLD = "\033[1m"
    END = "\033[0m"


def 打印(内容: str, 颜色: str = C.END):
    print(f"{C.BOLD}{颜色}{内容}{C.END}")


# ==================== 检查项 ====================

def 检查_pydantic导入(代码: str) -> Tuple[bool, str]:
    """检查 1：是否有 aiohttp 或 httpx（HTTP 客户端）"""
    if "aiohttp" in 代码 or "httpx" in 代码 or "requests" in 代码:
        return True, "✅ 找到 HTTP 客户端库（aiohttp/httpx/requests）"
    return True, "⚠️ 未找到 HTTP 客户端（可用 aiohttp 或 mock）"


def 检查_tools参数(代码: str) -> Tuple[bool, str]:
    """检查 2：是否构建了 OpenAI Function Calling 格式的 tools 参数"""
    关键词 = ["tools", "type", "function", "parameters"]
    找到数 = sum(1 for kw in 关键词 if kw in 代码)
    
    if 找到数 >= 3:
        return True, f"✅ 找到 tools 参数构建代码（{找到数}/4 个关键词）"
    return False, f"❌ 未找到完整的 tools 参数构建（只找到 {找到数}/4 个关键词）"


def 检查_tool_calls解析(代码: str) -> Tuple[bool, str]:
    """检查 3：是否解析 tool_calls"""
    关键词 = ["tool_calls", "function", "arguments", "json.loads"]
    找到数 = sum(1 for kw in 关键词 if kw in 代码)
    
    if 找到数 >= 3:
        return True, f"✅ 找到 tool_calls 解析逻辑（{找到数}/4 个关键词）"
    return False, f"❌ 未找到 tool_calls 解析（需要解析 LLM 返回的 tool_calls）"


def 检查_工具执行(代码: str) -> Tuple[bool, str]:
    """检查 4：是否根据 tool_calls 执行工具"""
    关键词 = ["execute", "call_tool", "执行", "调用"]
    找到 = [kw for kw in 关键词 if kw in 代码]
    
    if 找到:
        return True, f"✅ 找到工具执行逻辑：{', '.join(找到)}"
    return False, "❌ 未找到工具执行逻辑（需要根据 tool_call 调用对应函数）"


def 检查_结果回传(代码: str) -> Tuple[bool, str]:
    """检查 5：是否将工具执行结果返回给 LLM"""
    关键词 = ["role", "tool", "tool_call_id", "Observation"]
    找到数 = sum(1 for kw in 关键词 if kw in 代码)
    
    if 找到数 >= 2:
        return True, f"✅ 找到结果回传逻辑（{找到数}/4 个关键词）"
    return False, "⚠️ 未找到明确的工具结果回传（应追加 role=tool 的消息）"


def 检查_循环结构(代码: str) -> Tuple[bool, str]:
    """检查 6：是否有 Function Calling 循环（可能多轮）"""
    树 = ast.parse(代码)
    有循环 = False
    有_while = False
    
    for node in ast.walk(树):
        if isinstance(node, ast.For):
            有循环 = True
        if isinstance(node, ast.While):
            有_while = True
            有循环 = True
    
    if 有循环:
        msg = "✅ 找到循环结构（"
        if 有_while:
            msg += "while"
        elif 有循环:
            msg += "for"
        msg += "），支持多轮 Function Calling"
        return True, msg
    return True, "⚠️ 未找到循环（单轮 Function Calling 也可以，多轮更好）"


def 检查_错误处理(代码: str) -> Tuple[bool, str]:
    """检查 7：是否有错误处理"""
    树 = ast.parse(代码)
    有_try = False
    有重试 = False
    
    for node in ast.walk(树):
        if isinstance(node, ast.Try):
            有_try = True
        if isinstance(node, ast.FunctionDef):
            for child in ast.walk(node):
                if isinstance(child, ast.Try):
                    有_try = True
        if "retry" in 代码.lower() or "重试" in 代码:
            有重试 = True
    
    if 有_try and 有重试:
        return True, "✅ 有 try/except 错误处理 + 重试机制"
    elif 有_try:
        return True, "✅ 有 try/except 错误处理"
    return False, "❌ 未找到错误处理（建议加 try/except）"


def 检查_类型注解(代码: str) -> Tuple[bool, str]:
    """检查 8：是否有类型注解"""
    树 = ast.parse(代码)
    有注解 = 0
    
    for node in ast.walk(树):
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            if node.returns is not None:
                有注解 += 1
            for arg in node.args.args:
                if arg.annotation is not None:
                    有注解 += 1
    
    if 有注解 >= 3:
        return True, f"✅ 类型注解较好（{有注解} 处注解）"
    return True, f"⚠️ 类型注解较少（{有注解} 处，建议多加）"


# ==================== 运行检查 ====================

def 运行用户代码(文件路径: Path) -> Tuple[bool, str, float]:
    try:
        开始 = time.time()
        结果 = subprocess.run(
            [sys.executable, str(文件路径)],
            capture_output=True, text=True, timeout=TIMEOUT,
        )
        耗时 = time.time() - 开始
        if 结果.returncode == 0:
            return True, 结果.stdout, 耗时
        return False, f"运行失败：\n{结果.stderr[:500]}", 耗时
    except subprocess.TimeoutExpired:
        return False, f"⏰ 运行超时（{TIMEOUT} 秒）", TIMEOUT
    except Exception as e:
        return False, f"运行出错：{e}", 0.0


def 检查_代码可运行(文件路径: Path) -> Tuple[bool, str]:
    成功, 输出, 耗时 = 运行用户代码(文件路径)
    if 成功:
        return True, f"✅ 代码运行成功（耗时 {耗时:.2f} 秒）\n   输出预览：\n{输出[:300]}"
    return False, f"❌ 代码运行失败：\n{输出[:500]}"


def 检查_function_calling流程(文件路径: Path) -> Tuple[bool, str]:
    """检查：代码是否实现了完整的 Function Calling 流程"""
    检验代码 = """
import sys
sys.path.insert(0, ".")
try:
    import importlib.util
    spec = importlib.util.spec_from_file_location("user_module", r"{}")
    if spec is None:
        print("❌ 无法加载模块")
    else:
        user_module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(user_module)
        
        # 检查是否有 chat_with_function_calling 或类似函数
        funcs = [name for name in dir(user_module) 
                  if "chat" in name.lower() or "function" in name.lower() or "call" in name.lower()]
        if funcs:
            print(f"✅ 找到关键函数：{{', '.join(funcs)}}")
        else:
            print("⚠️ 未找到 chat_with_function_calling 等函数")
except Exception as e:
    print(f"❌ 检查失败：{{e}}")
""".format(str(文件路径).replace("\\", "\\\\"))
    
    try:
        结果 = subprocess.run(
            [sys.executable, "-c", 检验代码],
            capture_output=True, text=True, timeout=TIMEOUT,
        )
        return True, 结果.stdout.strip() or 结果.stderr.strip()
    except Exception as e:
        return False, f"检查失败：{e}"


# ==================== 主逻辑 ====================

def 验收(文件路径: Path) -> int:
    打印(f"\n{'='*60}", C.CYAN)
    打印(f"  Day 4 练习验收：{文件路径.name}", C.CYAN)
    打印(f"{'='*60}\n", C.CYAN)
    
    if not 文件路径.exists():
        打印(f"❌ 文件不存在：{文件路径}", C.RED)
        return 0
    
    with open(文件路径, "r", encoding="utf-8") as f:
        代码 = f.read()
    
    检查项 = [
        ("HTTP 客户端（aiohttp/httpx）", 检查_pydantic导入),
        ("tools 参数构建（OpenAI Function Calling 格式）", 检查_tools参数),
        ("tool_calls 解析", 检查_tool_calls解析),
        ("工具执行逻辑", 检查_工具执行),
        ("工具结果回传给 LLM", 检查_结果回传),
        ("Function Calling 循环（多轮）", 检查_循环结构),
        ("错误处理（try/except + 重试）", 检查_错误处理),
        ("类型注解", 检查_类型注解),
    ]
    
    分数 = 0
    满分 = len(检查项) * 10 + 20
    
    for 名称, 检查函数 in 检查项:
        通过, 消息 = 检查函数(代码)
        打印(f"  [{名称}]", C.BLUE)
        if "✅" in 消息:
            打印(f"  {消息}", C.GREEN)
            分数 += 10
        elif "⚠️" in 消息:
            打印(f"  {消息}", C.YELLOW)
            分数 += 5
        else:
            打印(f"  {消息}", C.RED)
    
    # 运行检查
    打印(f"\n  [代码可运行性]", C.BLUE)
    通过, 消息 = 检查_代码可运行(文件路径)
    打印(f"  {消息}", C.GREEN if 通过 else C.RED)
    if 通过:
        分数 += 10
    
    # Function Calling 流程检查
    打印(f"\n  [Function Calling 流程完整性]", C.BLUE)
    通过, 消息 = 检查_function_calling流程(文件路径)
    打印(f"  {消息}", C.GREEN if 通过 else C.YELLOW)
    if 通过 and "✅" in 消息:
        分数 += 10
    elif 通过:
        分数 += 5
    
    打印(f"\n{'='*60}", C.CYAN)
    颜色 = C.GREEN if 分数 >= 80 else (C.YELLOW if 分数 >= 60 else C.RED)
    打印(f"  总分：{分数}/{满分}", 颜色)
    
    if 分数 >= 80:
        打印(f"  🎉 验收通过！你掌握了 Function Calling！", C.GREEN)
    elif 分数 >= 60:
        打印(f"  ⚠️ 基本通过，建议改进后再继续", C.YELLOW)
    else:
        打印(f"  ❌ 还需努力，参考上面提示改进", C.RED)
    
    打印(f"{'='*60}\n", C.CYAN)
    
    if 分数 < 80:
        打印("📋 改进建议：", C.YELLOW)
        if not 检查_tool_calls解析(代码)[0]:
            打印("  • 解析 LLM 响应中的 tool_calls（function.name 和 function.arguments）", C.YELLOW)
        if not 检查_工具执行(代码)[0]:
            打印("  • 根据 tool_call 的 name 调用对应工具函数", C.YELLOW)
        if not 检查_结果回传(代码)[0]:
            打印("  • 工具执行结果以 role=tool 追加到 messages，再发给 LLM", C.YELLOW)
        if not 检查_错误处理(代码)[0]:
            打印("  • 加 try/except 处理 JSON 解析失败、工具执行失败等情况", C.YELLOW)
        print()
    
    return 分数


def main():
    import argparse
    parser = argparse.ArgumentParser(description="Day 4 练习验收脚本")
    parser.add_argument("--file", default="day04_practice_template.py", help="练习文件路径")
    args = parser.parse_args()
    文件路径 = Path(args.file)
    验收(文件路径)


if __name__ == "__main__":
    main()