Create storage.py

storage.py

@@ -0,0 +1,56 @@
+import os
+from huggingface_hub import HfApi, create_bucket, list_bucket_tree
+from datetime import datetime
+
+HF_TOKEN = os.environ.get("HF_TOKEN")  # 在 Space Secrets 中设置
+BUCKET_ID = os.environ.get("HF_BUCKET_ID", "your-username/learning-platform-data")
+
+api = HfApi(token=HF_TOKEN)
+
+def ensure_bucket():
+    """确保 Bucket 存在"""
+    try:
+        create_bucket(BUCKET_ID, private=True)
+    except Exception:
+        pass  # 已存在
+
+def upload_file(file_bytes: bytes, filename: str, username: str, content_type: str) -> str:
+    """上传文件到 Bucket，返回存储路径"""
+    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+    safe_name = f"{timestamp}_{filename}"
+    bucket_path = f"users/{username}/{safe_name}"
+
+    from huggingface_hub import batch_bucket_files
+    import tempfile
+
+    with tempfile.NamedTemporaryFile(delete=False, suffix=f"_{filename}") as tmp:
+        tmp.write(file_bytes)
+        tmp_path = tmp.name
+
+    batch_bucket_files(
+        BUCKET_ID,
+        add=[(tmp_path, bucket_path)]
+    )
+    os.unlink(tmp_path)
+    return bucket_path
+
+def list_user_files(username: str) -> list:
+    """列出用户的所有文件"""
+    try:
+        tree = list_bucket_tree(BUCKET_ID, prefix=f"users/{username}/")
+        return [item.rfilename for item in tree if hasattr(item, 'rfilename')]
+    except Exception:
+        return []
+
+def download_file(bucket_path: str) -> str:
+    """下载文件到本地临时路径，返回本地路径"""
+    from huggingface_hub import hf_hub_download
+    return hf_hub_download(
+        repo_id=BUCKET_ID.replace("/", "_"),
+        filename=bucket_path,
+        repo_type="bucket"
+    )
+
+def get_file_url(bucket_path: str) -> str:
+    """获取文件的下载 URL"""
+    return f"https://huggingface.co/buckets/{BUCKET_ID}/resolve/main/{bucket_path}"