| import os |
| from huggingface_hub import HfApi, create_bucket, list_bucket_tree |
| from datetime import datetime |
|
|
| HF_TOKEN = os.environ.get("HF_TOKEN") |
| 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}" |
