Update annotation.py

annotation.py

@@ -1,51 +1,66 @@
-import re
-import os
-import json
-
-
-def get_user_annotation_file(username):
-    """获取用户特定的标注文件路径"""
-    os.makedirs("annotations", exist_ok=True)
-    safe_username = re.sub(r'[\\/*?:"<>|]', "_", username)
-    return f"annotations/annotation_results_{safe_username}.json"
-
-
-def save_annotations(username_state, annotation_results_state, tasks):
-    """保存标注结果到用户特定的文件"""
-    try:
-        annotation_file = get_user_annotation_file(username_state)
-        save_data = {
-            "total_tasks": len(tasks),
-            "completed_tasks": len(annotation_results_state),
-            "username": username_state,
-            "annotations": []
-        }
-        for task_id, choice in annotation_results_state.items():
-            save_data["annotations"].append({
-                "task_id": task_id,
-                "text": tasks[task_id]["text"],
-                "choice": choice,
-                "audioA_id": f"audioA_{task_id}",
-                "audioB_id": f"audioB_{task_id}",
-                "username": username_state
-            })
-        with open(annotation_file, "w", encoding="utf-8") as f:
-            json.dump(save_data, f, ensure_ascii=False, indent=2)
-        return f"✅ 标注结果已保存到 {annotation_file}\n完成进度: {len(annotation_results_state)}/{len(tasks)}"
-    except Exception as e:
-        return f"❌ 保存失败: {str(e)}"
-
-
-def load_annotations(username):
-    """加载用户特定的标注结果"""
-    try:
-        annotation_file = get_user_annotation_file(username)
-        if os.path.exists(annotation_file):
-            with open(annotation_file, "r", encoding="utf-8") as f:
-                save_data = json.load(f)
-            annotation_results = {ann["task_id"]: ann["choice"] for ann in save_data["annotations"]}
-            return annotation_results
-        else:
-            return {}
-    except Exception as e:
-        return {}
+import re
+import os
+import json
+from huggingface_hub import HfApi, HfFolder, Repository
+
+# 初始化 Hub token
+HF_TOKEN = os.getenv("HF_TOKEN")  # 从 Space secret 里读取
+REPO_ID = "sunnyzjx/annotation_results"  # 你的 dataset repo
+
+def get_user_annotation_file(username):
+    """获取用户特定的标注文件路径"""
+    os.makedirs("annotations", exist_ok=True)
+    safe_username = re.sub(r'[\\/*?:"<>|]', "_", username)
+    return f"annotations/annotation_results_{safe_username}.json"
+
+def save_annotations(username_state, annotation_results_state, tasks):
+    """保存标注结果到本地并推送到 Hub"""
+    try:
+        annotation_file = get_user_annotation_file(username_state)
+        save_data = {
+            "total_tasks": len(tasks),
+            "completed_tasks": len(annotation_results_state),
+            "username": username_state,
+            "annotations": []
+        }
+        for task_id, choice in annotation_results_state.items():
+            save_data["annotations"].append({
+                "task_id": task_id,
+                "text": tasks[task_id]["text"],
+                "choice": choice,
+                "audioA_id": f"audioA_{task_id}",
+                "audioB_id": f"audioB_{task_id}",
+                "username": username_state
+            })
+        # 保存到本地
+        with open(annotation_file, "w", encoding="utf-8") as f:
+            json.dump(save_data, f, ensure_ascii=False, indent=2)
+
+        # 推送到 Hugging Face Hub
+        api = HfApi()
+        api.upload_file(
+            path_or_fileobj=annotation_file,
+            path_in_repo=os.path.basename(annotation_file),
+            repo_id=REPO_ID,
+            repo_type="dataset",
+            token=HF_TOKEN
+        )
+
+        return f"✅ 标注结果已保存并上传到 {REPO_ID}\n完成进度: {len(annotation_results_state)}/{len(tasks)}"
+
+    except Exception as e:
+        return f"❌ 保存失败: {str(e)}"
+
+def load_annotations(username):
+    """加载用户特定的标注结果（仅本地版本，Hub 上的可选扩展）"""
+    try:
+        annotation_file = get_user_annotation_file(username)
+        if os.path.exists(annotation_file):
+            with open(annotation_file, "r", encoding="utf-8") as f:
+                save_data = json.load(f)
+            annotation_results = {ann["task_id"]: ann["choice"] for ann in save_data["annotations"]}
+            return annotation_results
+        else:
+            return {}
+    except Exception as e:
+        return {}