#!/usr/bin/env python3 """ YouTube Trending Crawler Benchmark Runner 基于 benchmark.md 的评分维度自动评分 """ import json import re from pathlib import Path SKILL_PATH = Path(__file__).parent / "SKILL.md" def read_skill() -> str: return SKILL_PATH.read_text(encoding="utf-8") def check_method(skill: str, method_name: str) -> bool: """检查是否定义了指定方法""" pattern = rf"def\s+{re.escape(method_name)}\s*\(" return bool(re.search(pattern, skill)) def check_class(skill: str, class_name: str) -> bool: pattern = rf"class\s+{re.escape(class_name)}\s*[\(:]?" return bool(re.search(pattern, skill)) def score_functional_completeness(skill: str) -> dict: """维度1: 功能完整性 (30分)""" score = 0 details = {} # get_trending 方法 (8分) has_get_trending = check_method(skill, "get_trending") details["get_trending"] = has_get_trending if has_get_trending: match = re.search(r"def get_trending\((.*?)\)\s*->", skill, re.DOTALL) if match and "region_code" in match.group(1): score += 8 else: score += 4 else: score += 0 # get_multi_region_trending 方法 (7分) has_multi = check_method(skill, "get_multi_region_trending") details["get_multi_region_trending"] = has_multi if has_multi: if "quota" in skill.lower() and "unique_regions" in skill: score += 7 else: score += 4 else: score += 0 # extract_video_info 方法 (5分) has_extract = check_method(skill, "extract_video_info") details["extract_video_info"] = has_extract if has_extract: required_fields = ["video_id", "title", "view_count", "channel_title", "thumbnails"] extract_body = re.search( r"def extract_video_info\(self.*?\n(.*?)" r"(?=\n def |\n\nclass |\n# --+|$)", skill, re.DOTALL ) body = extract_body.group(1) if extract_body else "" matched = sum(1 for f in required_fields if f in body) score += int(5 * matched / len(required_fields)) else: score += 0 # save_to_json 方法 (5分) has_json = check_method(skill, "save_to_json") details["save_to_json"] = has_json if has_json and "json.dump" in skill: score += 5 elif has_json: score += 3 else: score += 0 # save_to_csv 方法 (5分) has_csv = check_method(skill, "save_to_csv") details["save_to_csv"] = has_csv if has_csv and "csv." in skill: score += 5 elif has_csv: score += 3 else: score += 0 return {"score": min(score, 30), "max": 30, "details": details} def score_code_quality(skill: str) -> dict: """维度2: 代码质量 (25分)""" score = 0 details = {} # 类结构清晰 (5分) has_class = check_class(skill, "YouTubeTrendingCrawler") details["class_structure"] = has_class score += 5 if has_class else 0 # 异常处理 (8分) exceptions = [ "ValueError", "QuotaExceededError", "urllib.error.HTTPError", "urllib.error.URLError", "RuntimeError" ] matched_exceptions = sum(1 for e in exceptions if e in skill) details["exception_handling"] = f"{matched_exceptions}/{len(exceptions)} types" score += int(8 * matched_exceptions / len(exceptions)) # 日志记录 (5分) has_logging = "self.logger" in skill and "logging.getLogger" in skill details["logging"] = has_logging score += 5 if has_logging else 0 # 代码注释 (4分) has_docstring = '"""' in skill docstring_count = skill.count('"""') // 2 # 每对引号算一个 docstring details["docstrings"] = docstring_count score += min(4, docstring_count) if has_docstring else 0 # 类型注解 (3分) type_hints = ["->", "Optional[", "List[", "Dict[", "Any", "Union["] matched_hints = sum(1 for t in type_hints if t in skill) details["type_hints"] = f"{matched_hints}/{len(type_hints)}" score += min(3, int(3 * matched_hints / 4)) return {"score": min(score, 25), "max": 25, "details": details} def score_api_usage(skill: str) -> dict: """维度3: API 使用规范 (20分)""" score = 0 details = {} # 环境变量读取 (5分) has_env = "os.environ.get('YOUTUBE_API_KEY')" in skill details["env_var"] = has_env score += 5 if has_env else 0 # 配额意识 (5分) has_quota = "DAILY_QUOTA" in skill and "quota_used" in skill and "_has_quota" in skill details["quota"] = has_quota score += 5 if has_quota else 0 # 分页处理 (5分) has_pagination = "page_token" in skill and "nextPageToken" in skill details["pagination"] = has_pagination score += 5 if has_pagination else 0 # 错误码处理 (5分) error_codes = ["403", "429", "400"] error_handling_body = re.search( r"def _get_trending_api\(.*?\n(.*?)" r"(?=\n def |\n\nclass |\n# --+|$)", skill, re.DOTALL ) body = error_handling_body.group(1) if error_handling_body else "" matched_codes = sum(1 for code in error_codes if code in body) details["error_codes"] = f"{matched_codes}/{len(error_codes)}" score += int(5 * matched_codes / len(error_codes)) return {"score": min(score, 20), "max": 20, "details": details} def score_data_integrity(skill: str) -> dict: """维度4: 数据完整性 (15分)""" score = 0 details = {} # 必需字段 (8分) required = ["video_id", "title", "view_count", "channel_title", "thumbnails"] extract_body = re.search( r"def extract_video_info\(self.*?\n(.*?)" r"(?=\n def |\n\nclass |\n# --+|$)", skill, re.DOTALL ) body = extract_body.group(1) if extract_body else "" matched = sum(1 for f in required if f in body) details["required_fields"] = f"{matched}/{len(required)}" score += int(8 * matched / len(required)) # 数据格式 (4分) has_json_output = "json.dump" in skill and "ensure_ascii=False" in skill details["json_format"] = has_json_output score += 4 if has_json_output else 2 # 数据清洗 (3分) cleaning_signals = [".strip()", "try:", "except", "if not", "default"] matched_cleaning = sum(1 for s in cleaning_signals if s in body) details["data_cleaning"] = f"{matched_cleaning}/{len(cleaning_signals)}" score += min(3, int(3 * matched_cleaning / 3)) return {"score": min(score, 15), "max": 15, "details": details} def score_extensibility(skill: str) -> dict: """维度5: 扩展性 (10分)""" score = 0 details = {} # 多地区支持 (4分) regions = ["US", "JP", "KR", "CN"] matched_regions = sum(1 for r in regions if r in skill) details["multi_region"] = f"{matched_regions}/{len(regions)}" score += min(4, int(4 * matched_regions / 2)) # 分类过滤 (3分) has_category = "video_category_id" in skill and "categoryId" in skill details["category_filter"] = has_category score += 3 if has_category else 0 # 可配置性 (3分) configurable = ["max_results", "region_code", "api_key", "cache_dir"] matched_config = sum(1 for c in configurable if c in skill) details["configurability"] = f"{matched_config}/{len(configurable)}" score += min(3, int(3 * matched_config / 3)) return {"score": min(score, 10), "max": 10, "details": details} def run_benchmark(iteration: int = 1): skill = read_skill() r1 = score_functional_completeness(skill) r2 = score_code_quality(skill) r3 = score_api_usage(skill) r4 = score_data_integrity(skill) r5 = score_extensibility(skill) total = r1["score"] + r2["score"] + r3["score"] + r4["score"] + r5["score"] max_total = r1["max"] + r2["max"] + r3["max"] + r4["max"] + r5["max"] dimensions = [ {"name": "功能完整性", "score": r1["score"], "weight": 0.30, "details": r1["details"]}, {"name": "代码质量", "score": r2["score"], "weight": 0.25, "details": r2["details"]}, {"name": "API 使用规范", "score": r3["score"], "weight": 0.20, "details": r3["details"]}, {"name": "数据完整性", "score": r4["score"], "weight": 0.15, "details": r4["details"]}, {"name": "扩展性", "score": r5["score"], "weight": 0.10, "details": r5["details"]}, ] weak = [d["name"] for d in dimensions if d["score"] < d["weight"] * max_total * 0.8] result = { "total_score": total, "max_score": max_total, "dimensions": dimensions, "weak_dimensions": weak, "iteration": iteration, "passed": total >= 95 } print(json.dumps(result, ensure_ascii=False, indent=2)) return result if __name__ == "__main__": import sys it = int(sys.argv[1]) if len(sys.argv) > 1 else 1 run_benchmark(it)