#!/usr/bin/env python3 """ Day 3 练习验收脚本 验收内容：生成器、异步生成器、上下文管理器 使用方法： python day03_practice_validator.py python day03_practice_validator.py --file day03_practice.py """ import ast import inspect 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 检查_同步生成器(代码: str) -> Tuple[bool, str]: """检查 1：是否有同步生成器（def + yield）""" 树 = ast.parse(代码) 找到了 = False 详情 = [] for node in ast.walk(树): if isinstance(node, ast.FunctionDef): for child in ast.walk(node): if isinstance(child, ast.Yield) or isinstance(child, ast.YieldFrom): 找到了 = True 详情.append(f"✅ 找到同步生成器：{node.name}（包含 yield）") break if 找到了: return True, "\n".join(详情) return False, "❌ 未找到同步生成器（需要用 def + yield）" def 检查_异步生成器(代码: str) -> Tuple[bool, str]: """检查 2：是否有异步生成器（async def + yield）""" 树 = ast.parse(代码) 找到了 = False 详情 = [] for node in ast.walk(树): if isinstance(node, ast.AsyncFunctionDef): for child in ast.walk(node): if isinstance(child, ast.Yield) or isinstance(child, ast.YieldFrom): 找到了 = True 详情.append(f"✅ 找到异步生成器：{node.name}（async def + yield）") break if 找到了: return True, "\n".join(详情) return False, "❌ 未找到异步生成器（需要用 async def + yield）" def 检查_async_for(代码: str) -> Tuple[bool, str]: """检查 3：是否使用了 async for 迭代异步生成器""" 树 = ast.parse(代码) 找到了 = False for node in ast.walk(树): if isinstance(node, ast.AsyncFor): 找到了 = True break if 找到了: return True, "✅ 找到 async for（正确迭代异步生成器）" return True, "⚠️ 未找到 async for（建议用 async for 迭代异步生成器，也可以用其他方式）" def 检查_上下文管理器类(代码: str) -> Tuple[bool, str]: """检查 4：是否实现了 __enter__/__exit__ 上下文管理器""" 树 = ast.parse(代码) 找到了 = False 详情 = [] for node in ast.walk(树): if isinstance(node, ast.ClassDef): 方法列表 = [m.name for m in node.body if isinstance(m, ast.FunctionDef)] if "__enter__" in 方法列表 and "__exit__" in 方法列表: 找到了 = True 详情.append(f"✅ 找到类上下文管理器：{node.name}（有 __enter__/__exit__）") if 找到了: return True, "\n".join(详情) return False, "❌ 未找到类上下文管理器（需要实现 __enter__/__exit__）" def 检查_contextmanager装饰器(代码: str) -> Tuple[bool, str]: """检查 5：是否使用了 @contextmanager""" 树 = ast.parse(代码) 找到了 = False for node in ast.walk(树): if isinstance(node, ast.FunctionDef): for decorator in node.decorator_list: if (isinstance(decorator, ast.Name) and decorator.id == "contextmanager") or \ (isinstance(decorator, ast.Attribute) and decorator.attr == "contextmanager"): 找到了 = True break if 找到了: return True, "✅ 找到 @contextmanager 装饰器（简洁的上下文管理器实现）" return True, "⚠️ 未找到 @contextmanager（可以用类实现，也可以用 @contextmanager）" def 检查_异步上下文管理器(代码: str) -> Tuple[bool, str]: """检查 6：是否有异步上下文管理器（__aenter__/__aexit__ 或 @asynccontextmanager）""" 树 = ast.parse(代码) 找到了类 = False 找到了装饰器 = False 详情 = [] for node in ast.walk(树): # 检查类实现 if isinstance(node, ast.ClassDef): 方法列表 = [m.name for m in node.body if isinstance(m, (ast.FunctionDef, ast.AsyncFunctionDef))] if "__aenter__" in 方法列表 and "__aexit__" in 方法列表: 找到了类 = True 详情.append(f"✅ 找到异步上下文管理器类：{node.name}") # 检查 @asynccontextmanager if isinstance(node, ast.AsyncFunctionDef): for decorator in node.decorator_list: if (isinstance(decorator, ast.Name) and "async" in decorator.id.lower()) or \ (isinstance(decorator, ast.Attribute) and "async" in decorator.attr.lower()): 找到了装饰器 = True 详情.append(f"✅ 找到 @asynccontextmanager 装饰器") break if 找到了类 or 找到了装饰器: return True, "\n".join(详情) return False, "❌ 未找到异步上下文管理器（需要 __aenter__/__aexit__ 或 @asynccontextmanager）" def 检查_流式输出模拟(代码: str) -> Tuple[bool, str]: """检查 7：生成器是否逐 chunk 产出（多次 yield，不是一次 return）""" 树 = ast.parse(代码) 生成器函数 = [] for node in ast.walk(树): if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): yield_count = sum(1 for child in ast.walk(node) if isinstance(child, (ast.Yield, ast.YieldFrom))) if yield_count >= 2: 生成器函数.append(f"✅ {node.name} 有 {yield_count} 个 yield（逐 chunk 产出）") if 生成器函数: return True, "\n".join(生成器函数) return False, "⚠️ 未找到多次 yield 的生成器（流式输出应该多次 yield）" def 检查_类型注解(代码: str) -> Tuple[bool, str]: """检查 8：是否有类型注解（Generator, AsyncGenerator 等）""" 树 = ast.parse(代码) 有注解 = 0 for node in ast.walk(树): if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): if node.returns is not None: 注解文本 = ast.unparse(node.returns) if hasattr(ast, 'unparse') else str(node.returns) if "Generator" in 注解文本 or "AsyncGenerator" in 注解文本: 有注解 += 1 if 有注解 >= 1: return True, f"✅ 找到 {有注解} 个带生成器类型注解的函数" return True, "⚠️ 未找到 Generator/AsyncGenerator 类型注解（建议加上）" # ==================== 运行检查 ==================== 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}", 耗时 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]}{'...' if len(输出) > 300 else ''}" return False, f"❌ 代码运行失败：\n{输出[:500]}" # ==================== 主逻辑 ==================== def 验收(文件路径: Path) -> int: 打印(f"\n{'='*60}", C.CYAN) 打印(f" Day 3 练习验收：{文件路径.name}", C.CYAN) 打印(f"{'='*60}\n", C.CYAN) if not 文件路径.exists(): 打印(f"❌ 文件不存在：{文件路径}", C.RED) 打印(f" 请创建 {文件路径.name}，或指定正确路径", C.YELLOW) return 0 with open(文件路径, "r", encoding="utf-8") as f: 代码 = f.read() 检查项 = [ ("同步生成器（def + yield）", 检查_同步生成器), ("异步生成器（async def + yield）", 检查_异步生成器), ("async for 使用", 检查_async_for), ("类上下文管理器（__enter__/__exit__）", 检查_上下文管理器类), ("@contextmanager 装饰器", 检查_contextmanager装饰器), ("异步上下文管理器（__aenter__/__aexit__）", 检查_异步上下文管理器), ("流式输出（多次 yield）", 检查_流式输出模拟), ("类型注解（Generator/AsyncGenerator）", 检查_类型注解), ] 分数 = 0 满分 = len(检查项) * 10 + 20 for 名称, 检查函数 in 检查项: 通过, 消息 = 检查函数(代码) 打印(f" [{名称}]", C.BLUE) if 通过: 打印(f" {消息}", C.GREEN) 分数 += 10 else: 打印(f" {消息}", C.RED) # 运行检查 打印(f"\n [代码可运行性]", C.BLUE) 通过, 消息 = 检查_代码可运行(文件路径) if 通过: 打印(f" {消息}", C.GREEN) 分数 += 20 else: 打印(f" {消息}", C.RED) 打印(f"\n{'='*60}", C.CYAN) 颜色 = C.GREEN if 分数 >= 80 else (C.YELLOW if 分数 >= 60 else C.RED) 打印(f" 总分：{分数}/{满分}", 颜色) if 分数 >= 80: 打印(f" 🎉 验收通过！你对生成器和上下文管理器掌握很好！", C.GREEN) elif 分数 >= 60: 打印(f" ⚠️ 基本通过，还有改进空间", C.YELLOW) else: 打印(f" ❌ 还需努力，参考上面提示改进", C.RED) 打印(f"{'='*60}\n", C.CYAN) if 分数 < 80: 打印("📋 改进建议：", C.YELLOW) if not 检查_异步生成器(代码)[0]: 打印(" • 异步生成器：async def 函数体里用 yield", C.YELLOW) if not 检查_异步上下文管理器(代码)[0]: 打印(" • 异步上下文管理器：实现 __aenter__/__aexit__ 或用 @asynccontextmanager", C.YELLOW) if not 检查_流式输出模拟(代码)[0]: 打印(" • 流式输出应该多次 yield，每次产出一个 chunk", C.YELLOW) print() return 分数 def main(): import argparse parser = argparse.ArgumentParser(description="Day 3 练习验收脚本") parser.add_argument("--file", default="day03_practice_template.py", help="练习文件路径") args = parser.parse_args() 文件路径 = Path(args.file) 验收(文件路径) if __name__ == "__main__": main()