run_benchmark.py
8.65 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
#!/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)