Spaces:
Sleeping
Sleeping
File size: 9,412 Bytes
7844386 944f8a1 7844386 944f8a1 7844386 326dd8e 7844386 326dd8e 7844386 326dd8e 7844386 |
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 265 266 267 268 |
#!/usr/bin/env python3
"""
OSS排行榜管理器 - 从OSS读取和更新排行榜数据
"""
import os
import json
import tempfile
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Any, Optional
from .oss_file_manager import OSSFileManager
class OSSLeaderboardManager:
"""OSS排行榜管理器 - 管理存储在OSS中的排行榜数据"""
def __init__(self):
"""初始化OSS排行榜管理器"""
self.oss_manager = OSSFileManager()
# OSS路径配置
self.leaderboard_path = "atlas_eval/leaderboard/"
self.backup_path = "atlas_eval/leaderboard/backup/"
self.leaderboard_file = "leaderboard.json"
# 完整的OSS路径
self.oss_leaderboard_file = f"{self.leaderboard_path}{self.leaderboard_file}"
print(f"📊 OSS排行榜路径: oss://opencompass/{self.oss_leaderboard_file}")
print(f"📦 OSS备份路径: oss://opencompass/{self.backup_path}")
def load_leaderboard_from_oss(self) -> List[Dict[str, Any]]:
"""
从OSS加载排行榜数据
Returns:
排行榜数据列表
"""
try:
print(f"📥 从OSS加载排行榜数据: {self.oss_leaderboard_file}")
# 从OSS下载文件内容
content = self.oss_manager.download_file_content(self.oss_leaderboard_file)
if content:
leaderboard_data = json.loads(content.decode('utf-8'))
print(f"✅ 成功加载 {len(leaderboard_data)} 条排行榜记录")
return leaderboard_data
else:
print("⚠️ OSS中未找到排行榜文件，返回空列表")
return []
except Exception as e:
print(f"❌ 从OSS加载排行榜失败: {e}")
return []
def save_leaderboard_to_oss(self, leaderboard_data: List[Dict[str, Any]],
create_backup: bool = True) -> bool:
"""
保存排行榜数据到OSS
Args:
leaderboard_data: 排行榜数据
create_backup: 是否创建备份
Returns:
是否保存成功
"""
try:
print(f"📤 保存排行榜数据到OSS: {self.oss_leaderboard_file}")
# 创建备份（如果需要且现有文件存在）
if create_backup:
self._create_backup()
# 创建临时文件
with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as temp_file:
json.dump(leaderboard_data, temp_file, indent=2, ensure_ascii=False)
temp_file_path = temp_file.name
try:
# 上传到OSS
success = self.oss_manager.upload_file(
local_file_path=temp_file_path,
oss_file_path=self.oss_leaderboard_file
)
if success:
print(f"✅ 成功保存 {len(leaderboard_data)} 条排行榜记录到OSS")
return True
else:
print("❌ 上传排行榜文件到OSS失败")
return False
finally:
# 清理临时文件
try:
os.unlink(temp_file_path)
except:
pass
except Exception as e:
print(f"❌ 保存排行榜到OSS失败: {e}")
return False
def _create_backup(self) -> bool:
"""
创建当前排行榜文件的备份
Returns:
是否备份成功
"""
try:
# 检查原文件是否存在
if not self.oss_manager.file_exists(self.oss_leaderboard_file):
print("📋 原排行榜文件不存在，跳过备份")
return True
# 生成备份文件名
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
backup_filename = f"leaderboard.json.backup_{timestamp}"
backup_path = f"{self.backup_path}{backup_filename}"
# 复制文件到备份路径
success = self.oss_manager.copy_file(
source_path=self.oss_leaderboard_file,
target_path=backup_path
)
if success:
print(f"📦 创建备份成功: {backup_path}")
return True
else:
print(f"❌ 创建备份失败: {backup_path}")
return False
except Exception as e:
print(f"❌ 创建备份时出错: {e}")
return False
def add_evaluation_result(self, result_data: Dict[str, Any]) -> bool:
"""
添加新的评测结果到排行榜
Args:
result_data: 评测结果数据
Returns:
是否添加成功
"""
try:
# 加载现有排行榜
leaderboard_data = self.load_leaderboard_from_oss()
# 检查是否已存在相同的提交（基于organization和submitted_time）
existing_entry = None
for i, entry in enumerate(leaderboard_data):
if (entry.get("organization") == result_data.get("organization") and
entry.get("submitted_time") == result_data.get("submitted_time")):
existing_entry = i
break
if existing_entry is not None:
print(f"🔄 更新现有排行榜条目: {result_data.get('organization')}")
leaderboard_data[existing_entry] = result_data
else:
print(f"➕ 添加新的排行榜条目: {result_data.get('organization')}")
leaderboard_data.append(result_data)
# 按准确率排序
leaderboard_data.sort(
key=lambda x: x.get("accuracy", 0),
reverse=True
)
# 保存到OSS
return self.save_leaderboard_to_oss(leaderboard_data)
except Exception as e:
print(f"❌ 添加评测结果失败: {e}")
return False
def get_leaderboard_summary(self) -> Dict[str, Any]:
"""
获取排行榜摘要信息
Returns:
排行榜摘要
"""
try:
leaderboard_data = self.load_leaderboard_from_oss()
if not leaderboard_data:
return {"total_entries": 0, "last_updated": None}
# 统计信息
total_entries = len(leaderboard_data)
# 获取最新更新时间
latest_time = None
for entry in leaderboard_data:
eval_time = entry.get("evaluation_timestamp")
if eval_time and (latest_time is None or eval_time > latest_time):
latest_time = eval_time
# 获取最高分
top_scores = {}
if leaderboard_data:
top_entry = leaderboard_data[0] # 已按准确率排序
top_scores = {
"accuracy": top_entry.get("accuracy", 0),
"mg_pass_2": top_entry.get("mg_pass_2", 0),
"mg_pass_4": top_entry.get("mg_pass_4", 0)
}
return {
"total_entries": total_entries,
"last_updated": latest_time,
"top_scores": top_scores,
"oss_path": self.oss_leaderboard_file
}
except Exception as e:
print(f"❌ 获取排行榜摘要失败: {e}")
return {"error": str(e)}
def migrate_local_to_oss(self, local_file_path: str) -> bool:
"""
将本地排行榜文件迁移到OSS
Args:
local_file_path: 本地文件路径
Returns:
是否迁移成功
"""
try:
if not os.path.exists(local_file_path):
print(f"❌ 本地文件不存在: {local_file_path}")
return False
# 读取本地文件
with open(local_file_path, 'r', encoding='utf-8') as f:
leaderboard_data = json.load(f)
print(f"📤 迁移 {len(leaderboard_data)} 条记录到OSS")
# 保存到OSS
return self.save_leaderboard_to_oss(leaderboard_data, create_backup=False)
except Exception as e:
print(f"❌ 迁移文件到OSS失败: {e}")
return False
if __name__ == "__main__":
# 测试OSS排行榜管理器
manager = OSSLeaderboardManager()
# 打印摘要信息
summary = manager.get_leaderboard_summary()
print(f"📊 排行榜摘要: {summary}")
# 测试加载排行榜
leaderboard = manager.load_leaderboard_from_oss()
print(f"📋 排行榜条目数: {len(leaderboard)}")
|