sync from GitHub d79b7e2: feat: Implement user account management and file synchronization with HF Hub

app/api/user_files.py

@@ -10,62 +10,49 @@
 - GET    /u/api/files/{key}     下载文件（仅 owner）
 - DELETE /u/api/files/{key}     删除文件（仅 owner）
 
-持久化：
-- 文件元信息存于 SQLite file_meta 表（/data/xtc.db，与现有 DB 同卷）
-- 文件内容存于 /data/uploads/（同一持久卷，HF Spaces restart 不丢）
+持久化（与 config_store 同样的混合策略）：
+- 热存储：SQLite（/data/xtc.db）+ 本地磁盘缓存（/data/hf_bucket/...）
+- 冷备份：HF Hub dataset 仓库，按用户隔离的 namespace
+  - user_accounts/<username>                 用户账号 JSON
+  - user_files/<user_id>/<file_key>          文件内容
+  - user_files_meta/<user_id>/<file_key>     文件元信息 JSON
+- 写入时先写本地，再异步推 Hub（Hub 失败不影响主流程）
+- 读取时本地命中即返回；未命中从 Hub 拉取并写本地（用于 Space 重建后恢复）
+- 未配置 HF_TOKEN / HF_CONFIG_REPO 时自动降级为纯本地
 """
 from __future__ import annotations
 
+import asyncio
 import hashlib
+import logging
 import re
 import time
+import urllib.parse
 import uuid
-from pathlib import Path
+from typing import Optional
 
 import bcrypt
 import jwt
-from fastapi import APIRouter, Body, Depends, File, Request, UploadFile
-from fastapi.responses import FileResponse
+from fastapi import APIRouter, Depends, File, Request, UploadFile
+from fastapi.responses import Response
 
 from ..config import get_settings
 from ..database import get_conn
 from ..errors import HttpError
-from ._common import CORS_HEADERS, ok_with_cors
+from ..hf_storage import is_hub_enabled
+from ..services import user_files_store
+from ._common import CORS_HEADERS, ok_with_cors, read_json_body
 
 router = APIRouter(prefix="/u/api", tags=["user-files"])
 
 ALGORITHM = "HS256"
 MAX_FILE_SIZE = 10 * 1024 * 1024  # 10MB
-NS_USER_FILES = "user_files"
+NS_USER_FILES = "user_files"  # file_meta.namespace 字段值
 TOKEN_TTL_SEC = 7 * 24 * 3600  # 7 天
 
 _USERNAME_RE = re.compile(r"^[A-Za-z0-9_\u4e00-\u9fff.\-]{2,32}$")
 
-
-# ===== 文件存储路径 =====
-
-def _uploads_root() -> Path:
-    """上传根目录：优先 /data/uploads，不可写回退 /tmp/uploads。"""
-    for candidate in ("/data/uploads", "/tmp/uploads"):
-        try:
-            p = Path(candidate)
-            p.mkdir(parents=True, exist_ok=True)
-            test = p / ".wtest"
-            test.touch(exist_ok=True)
-            test.unlink(missing_ok=True)
-            return p
-        except Exception:
-            continue
-    p = Path("/tmp/uploads")
-    p.mkdir(parents=True, exist_ok=True)
-    return p
-
-
-def _file_path(key: str) -> Path:
-    """根据 key 返回磁盘路径，做严格白名单校验防穿越。"""
-    if not key or not re.fullmatch(r"[A-Za-z0-9_-]{8,64}", key):
-        raise HttpError("invalid key", status=400, code="bad_request")
-    return _uploads_root() / key
+logger = logging.getLogger(__name__)
 
 
 # ===== JWT =====
@@ -134,10 +121,89 @@ def _verify_password(password: str, hashed: str) -> bool:
         return False
 
 
+# ===== 用户账号：SQLite + Hub 备份/恢复 =====
+
+def _insert_user_to_db(username: str, password_hash: str, now: int) -> int:
+    with get_conn() as conn:
+        cur = conn.execute(
+            "INSERT INTO user_accounts(username, password_hash, created_at) VALUES(?,?,?)",
+            (username, password_hash, now),
+        )
+        return int(cur.lastrowid)
+
+
+def _find_user_in_db(username: str) -> Optional[dict]:
+    with get_conn() as conn:
+        row = conn.execute(
+            "SELECT id, username, password_hash, created_at FROM user_accounts WHERE username = ?",
+            (username,),
+        ).fetchone()
+    return dict(row) if row else None
+
+
+def _restore_user_from_hub(username: str) -> Optional[dict]:
+    """Space 重建后 SQLite 丢失，从 Hub 拉取账号并写回 SQLite。"""
+    obj = user_files_store.restore_user_account(username)
+    if not obj:
+        return None
+    try:
+        # 写回 SQLite（若已存在则忽略）
+        with get_conn() as conn:
+            conn.execute(
+                "INSERT OR IGNORE INTO user_accounts(id, username, password_hash, created_at) "
+                "VALUES(?,?,?,?)",
+                (int(obj["id"]), obj["username"], obj["password_hash"], int(obj["created_at"])),
+            )
+        return obj
+    except Exception as e:
+        print(f"[user_files] restore_user_from_hub write db failed: {e}")
+        return obj
+
+
+# ===== 文件元信息：SQLite + Hub 备份/恢复 =====
+
+def _list_files_from_db(user_id: str) -> list[dict]:
+    with get_conn() as conn:
+        rows = conn.execute(
+            "SELECT key, filename, mime, size, sha256, uploaded_at "
+            "FROM file_meta WHERE uploaded_by = ? AND namespace = ? "
+            "ORDER BY uploaded_at DESC",
+            (user_id, NS_USER_FILES),
+        ).fetchall()
+    return [dict(r) for r in rows]
+
+
+async def _restore_files_from_hub_async(user_id: str) -> int:
+    metas = await user_files_store.restore_file_metas(user_id)
+    if not metas:
+        return 0
+    inserted = 0
+    with get_conn() as conn:
+        for m in metas:
+            try:
+                conn.execute(
+                    "INSERT OR IGNORE INTO file_meta(key, namespace, filename, mime, size, "
+                    "sha256, uploaded_at, uploaded_by, access_key, refs) "
+                    "VALUES(?,?,?,?,?,?,?,?,?,?)",
+                    (
+                        m["key"], NS_USER_FILES, m.get("filename", "unnamed"),
+                        m.get("mime", "application/octet-stream"),
+                        int(m.get("size", 0)), m.get("sha256"),
+                        int(m.get("uploaded_at", 0)), str(user_id),
+                        m.get("access_key", ""), "",
+                    ),
+                )
+                inserted += 1
+            except Exception as e:
+                print(f"[user_files] restore file_meta {m.get('key')} failed: {e}")
+    return inserted
+
+
 # ===== 注册 =====
 
 @router.post("/register")
-async def register(body: dict = Body(default={})):
+async def register(request: Request):
+    body = await read_json_body(request)
     username = str(body.get("username") or "").strip()
     password = str(body.get("password") or "")
     if not username or not password:
@@ -150,18 +216,39 @@ async def register(body: dict = Body(default={})):
     if len(password) < 6 or len(password) > 128:
         raise HttpError("password length must be 6-128", status=400, code="bad_request")
 
+    # 先查 Hub 是否已有该用户（SQLite 重建场景）
+    if not _find_user_in_db(username):
+        hub_user = user_files_store.restore_user_account(username)
+        if hub_user:
+            # 写回 SQLite，之后走"已存在"分支
+            try:
+                with get_conn() as conn:
+                    conn.execute(
+                        "INSERT OR IGNORE INTO user_accounts(id, username, password_hash, created_at) "
+                        "VALUES(?,?,?,?)",
+                        (int(hub_user["id"]), hub_user["username"],
+                         hub_user["password_hash"], int(hub_user["created_at"])),
+                    )
+            except Exception:
+                pass
+
     password_hash = _hash_password(password)
     now = int(time.time())
     try:
-        with get_conn() as conn:
-            cur = conn.execute(
-                "INSERT INTO user_accounts(username, password_hash, created_at) VALUES(?,?,?)",
-                (username, password_hash, now),
-            )
-            user_id = cur.lastrowid
+        user_id = _insert_user_to_db(username, password_hash, now)
     except Exception:
         raise HttpError("username already exists", status=409, code="conflict")
 
+    # 备份到 Hub（本地缓存 + 异步推远端）
+    try:
+        user_files_store.backup_user_account(
+            user_id=user_id, username=username,
+            password_hash=password_hash, created_at=now,
+        )
+        asyncio.create_task(user_files_store.push_user_account(username))
+    except Exception as e:
+        print(f"[user_files] backup_user_account failed: {e}")
+
     token = _issue_user_token(user_id, username)
     return ok_with_cors({
         "token": token,
@@ -172,24 +259,27 @@ async def register(body: dict = Body(default={})):
 # ===== 登录 =====
 
 @router.post("/login")
-async def login(body: dict = Body(default={})):
+async def login(request: Request):
+    body = await read_json_body(request)
     username = str(body.get("username") or "").strip()
     password = str(body.get("password") or "")
     if not username or not password:
         raise HttpError("username and password required", status=400, code="bad_request")
 
-    with get_conn() as conn:
-        row = conn.execute(
-            "SELECT id, username, password_hash FROM user_accounts WHERE username = ?",
-            (username,),
-        ).fetchone()
-    if not row or not _verify_password(password, row["password_hash"]):
+    user = _find_user_in_db(username)
+    # SQLite 没有则尝试从 Hub 恢复（Space 重建场景）
+    if not user:
+        restored = _restore_user_from_hub(username)
+        if restored:
+            user = _find_user_in_db(username) or restored
+
+    if not user or not _verify_password(password, user["password_hash"]):
         raise HttpError("invalid username or password", status=401, code="unauthorized")
 
-    token = _issue_user_token(row["id"], row["username"])
+    token = _issue_user_token(int(user["id"]), user["username"])
     return ok_with_cors({
         "token": token,
-        "user": {"id": row["id"], "username": row["username"]},
+        "user": {"id": int(user["id"]), "username": user["username"]},
     })
 
 
@@ -207,14 +297,12 @@ async def me(user: dict = Depends(require_user)):
 @router.get("/files")
 async def list_my_files(user: dict = Depends(require_user)):
     user_id = str(user["sub"])
-    with get_conn() as conn:
-        rows = conn.execute(
-            "SELECT key, filename, mime, size, sha256, uploaded_at "
-            "FROM file_meta WHERE uploaded_by = ? AND namespace = ? "
-            "ORDER BY uploaded_at DESC",
-            (user_id, NS_USER_FILES),
-        ).fetchall()
-    items = [dict(r) for r in rows]
+    items = _list_files_from_db(user_id)
+    # SQLite 为空时尝试从 Hub 恢复（Space 重建场景）
+    if not items:
+        restored = await _restore_files_from_hub_async(user_id)
+        if restored > 0:
+            items = _list_files_from_db(user_id)
     return ok_with_cors({"items": items, "count": len(items)})
 
 
@@ -239,7 +327,6 @@ async def upload_file(
 
     key = uuid.uuid4().hex
     filename = (file.filename or "unnamed").strip()
-    # 截断过长文件名，避免存储异常
     if len(filename) > 255:
         filename = filename[:255]
     mime = (file.content_type or "application/octet-stream").strip()
@@ -247,11 +334,30 @@ async def upload_file(
     sha256 = hashlib.sha256(content).hexdigest()
     now = int(time.time())
 
-    # 写磁盘
-    p = _file_path(key)
-    p.write_bytes(content)
+    # 写文件内容到本地缓存 + 异步推 Hub
+    try:
+        user_files_store.upload_user_file_content(user_id, key, content)
+        logger.info(
+            "[upload] user=%s uid=%s key=%s filename=%s size=%d saved locally, hub_enabled=%s",
+            username, user_id, key, filename, size, is_hub_enabled(),
+        )
+        asyncio.create_task(user_files_store.push_user_file_content(user_id, key))
+    except Exception as e:
+        logger.error("[upload] store failed user=%s filename=%s: %s", username, filename, e)
+        raise HttpError("failed to store file", status=500, code="internal_error") from e
 
-    # 写元信息
+    # 写元信息到 SQLite
+    meta = {
+        "key": key,
+        "namespace": NS_USER_FILES,
+        "filename": filename,
+        "mime": mime,
+        "size": size,
+        "sha256": sha256,
+        "uploaded_at": now,
+        "uploaded_by": user_id,
+        "access_key": username,
+    }
     with get_conn() as conn:
         conn.execute(
             "INSERT INTO file_meta(key, namespace, filename, mime, size, sha256, "
@@ -260,6 +366,13 @@ async def upload_file(
             (key, NS_USER_FILES, filename, mime, size, sha256, now, user_id, username, ""),
         )
 
+    # 备份元信息到 Hub
+    try:
+        user_files_store.backup_file_meta(user_id, key, meta)
+        asyncio.create_task(user_files_store.push_file_meta(user_id, key))
+    except Exception as e:
+        print(f"[user_files] backup_file_meta failed: {e}")
+
     return ok_with_cors({
         "key": key,
         "filename": filename,
@@ -283,20 +396,49 @@ async def download_file(
             "SELECT filename, mime, uploaded_by FROM file_meta WHERE key = ?",
             (key,),
         ).fetchone()
+    # SQLite 没有则尝试从 Hub 恢复单条 meta（Space 重建场景）
+    if not row:
+        metas = await user_files_store.restore_file_metas(user_id)
+        target = next((m for m in metas if m.get("key") == key), None)
+        if target:
+            try:
+                with get_conn() as conn:
+                    conn.execute(
+                        "INSERT OR IGNORE INTO file_meta(key, namespace, filename, mime, size, "
+                        "sha256, uploaded_at, uploaded_by, access_key, refs) "
+                        "VALUES(?,?,?,?,?,?,?,?,?,?)",
+                        (target["key"], NS_USER_FILES, target.get("filename", "unnamed"),
+                         target.get("mime", "application/octet-stream"),
+                         int(target.get("size", 0)), target.get("sha256"),
+                         int(target.get("uploaded_at", 0)), str(user_id),
+                         target.get("access_key", ""), ""),
+                    )
+            except Exception:
+                pass
+            with get_conn() as conn:
+                row = conn.execute(
+                    "SELECT filename, mime, uploaded_by FROM file_meta WHERE key = ?",
+                    (key,),
+                ).fetchone()
     if not row:
         raise HttpError("file not found", status=404, code="not_found")
     if row["uploaded_by"] != user_id:
         raise HttpError("forbidden", status=403, code="forbidden")
 
-    p = _file_path(key)
-    if not p.exists():
-        raise HttpError("file content missing on disk", status=410, code="gone")
-
-    return FileResponse(
-        path=str(p),
-        media_type=row["mime"] or "application/octet-stream",
-        filename=row["filename"],
-    )
+    # 读文件内容：本地优先，Hub 回退
+    content = user_files_store.download_user_file_content(user_id, key)
+    if content is None:
+        raise HttpError("file content missing", status=410, code="gone")
+
+    # 用 Response 直接返回 bytes（避免依赖本地磁盘路径，兼容 Hub 回退场景）
+    encoded_name = urllib.parse.quote(row["filename"])
+    headers = {
+        **CORS_HEADERS,
+        "Content-Disposition": f"attachment; filename=\"{row['filename']}\"; filename*=UTF-8''{encoded_name}",
+        "Content-Type": row["mime"] or "application/octet-stream",
+        "Content-Length": str(len(content)),
+    }
+    return Response(content=content, status_code=200, headers=headers, media_type=row["mime"] or "application/octet-stream")
 
 
 # ===== 删除 =====
@@ -318,13 +460,14 @@ async def delete_file(
             raise HttpError("forbidden", status=403, code="forbidden")
         conn.execute("DELETE FROM file_meta WHERE key = ?", (key,))
 
-    # 删磁盘文件
+    # 删文件内容 + 元信息（本地 + Hub）
     try:
-        p = _file_path(key)
-        p.unlink(missing_ok=True)
-    except HttpError:
-        raise
-    except Exception:
-        pass
+        user_files_store.delete_user_file_content(user_id, key)
+    except Exception as e:
+        print(f"[user_files] delete_user_file_content failed: {e}")
+    try:
+        user_files_store.delete_file_meta(user_id, key)
+    except Exception as e:
+        print(f"[user_files] delete_file_meta failed: {e}")
 
     return ok_with_cors({"deleted": True, "key": key})

app/hf_storage.py

@@ -105,6 +105,49 @@ def download_bytes(namespace: str, key: str) -> Optional[bytes]:
     return None
 
 
+def delete_local(namespace: str, key: str) -> bool:
+    """仅删本地缓存。返回是否删除了文件。"""
+    p = _local_path(namespace, key)
+    try:
+        if p.exists() and p.is_file():
+            p.unlink(missing_ok=True)
+            return True
+    except Exception:
+        pass
+    return False
+
+
+def delete_from_hub(namespace: str, key: str) -> bool:
+    """删除：本地缓存 + Hub 远端。Hub 失败不影响本地。"""
+    delete_local(namespace, key)
+    if not is_hub_enabled():
+        return True
+    try:
+        _hub_delete(namespace, key)
+        return True
+    except Exception as e:
+        print(f"[hf_storage] delete_from_hub failed ns={namespace} key={key}: {e}")
+        return False
+
+
+def list_local_keys(namespace: str) -> list[str]:
+    """列出某命名空间下本地缓存的所有 key（用于启动恢复）。"""
+    root = _local_root()
+    ns_parts = _safe_parts(namespace)
+    ns_dir = root.joinpath(*ns_parts) if ns_parts else root
+    if not ns_dir.exists() or not ns_dir.is_dir():
+        return []
+    out = []
+    for p in ns_dir.rglob("*"):
+        if p.is_file():
+            try:
+                rel = p.relative_to(ns_dir).as_posix()
+                out.append(rel)
+            except Exception:
+                continue
+    return out
+
+
 # ===== 异步接口（推送 Hub 用）=====
 
 async def push_to_hub(namespace: str, key: str) -> bool:
@@ -171,3 +214,47 @@ def _hub_download(namespace: str, key: str) -> Optional[bytes]:
             return None
         print(f"[hf_storage] hub_download failed: {e}")
         return None
+
+
+def _hub_delete(namespace: str, key: str) -> None:
+    """从 Hub 仓库删除一个文件（不存在视为成功）。"""
+    api, repo_id = _get_hf_api()
+    _ensure_repo(api, repo_id)
+    path_in_repo = _hub_path(namespace, key)
+    try:
+        api.delete_file(
+            path_in_repo=path_in_repo,
+            repo_id=repo_id,
+            repo_type="dataset",
+        )
+    except Exception as e:
+        msg = str(e).lower()
+        if "404" in msg or "not found" in msg or "no such file" in msg:
+            return
+        raise
+
+
+def list_hub_keys(namespace: str) -> list[str]:
+    """列出 Hub 仓库中某命名空间下的所有 key（同步，需在线程中调用）。
+
+    返回 key 列表（已去掉 namespace 前缀）。Hub 未启用或失败时返回空列表。
+    """
+    if not is_hub_enabled():
+        return []
+    try:
+        api, repo_id = _get_hf_api()
+        _ensure_repo(api, repo_id)
+        files = api.list_repo_files(repo_id=repo_id, repo_type="dataset")
+    except Exception as e:
+        print(f"[hf_storage] list_hub_keys failed: {e}")
+        return []
+    ns_parts = _safe_parts(namespace)
+    prefix = "/".join(ns_parts) + "/" if ns_parts else ""
+    out = []
+    for f in files:
+        if not f.startswith(prefix):
+            continue
+        rel = f[len(prefix):]
+        if rel:
+            out.append(rel)
+    return out

app/main.py

@@ -6,6 +6,7 @@ import logging
 from contextlib import asynccontextmanager
 
 from fastapi import FastAPI, Request
+from fastapi.exceptions import RequestValidationError
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse
 
@@ -13,8 +14,41 @@ from .config import get_settings
 from .database import close_db, init_db
 from .db_writer import start as start_db_writer, stop as stop_db_writer
 from .errors import HttpError, http_error_handler, unhandled_exception_handler
+from .hf_storage import is_hub_enabled
 from .http_client import close_http_client
-from .services import config_store, pseudo_store
+from .services import config_store, pseudo_store, user_files_sync
+
+
+async def validation_error_handler(request: Request, exc: RequestValidationError) -> JSONResponse:
+    """把 FastAPI 422 校验错误转为统一格式，避免前端无法解析。"""
+    import json
+    details = []
+    for err in getattr(exc, "errors", lambda: [])():
+        details.append(err)
+    msg = "request validation failed"
+    if details:
+        try:
+            msg = json.dumps(details, ensure_ascii=False, default=str)
+        except Exception:
+            msg = str(details)
+    return JSONResponse(
+        status_code=422,
+        content={
+            "ok": False,
+            "error": {
+                "code": "unprocessable_entity",
+                "message": msg,
+                "status": 422,
+                "retryable": False,
+                "hint": "请求参数格式有误，请检查字段",
+            },
+        },
+        headers={
+            "Access-Control-Allow-Origin": "*",
+            "Access-Control-Allow-Methods": "*",
+            "Access-Control-Allow-Headers": "*",
+        },
+    )
 
 
 @asynccontextmanager
@@ -35,6 +69,21 @@ async def lifespan(app: FastAPI):
         logging.getLogger(__name__).warning("config preload failed: %s", e)
     # 启动伪流式清理任务
     cleanup_task = asyncio.create_task(pseudo_store.cleanup_loop())
+    # 启动用户文件 Hub 定时同步任务
+    s = get_settings()
+    if is_hub_enabled():
+        logging.getLogger(__name__).info(
+            "[startup] Hub sync enabled: repo=%s token=***%s",
+            s.hf_config_repo, s.hf_token[-4:] if s.hf_token else "(empty)",
+        )
+        sync_task = user_files_sync.start_sync_task()
+    else:
+        logging.getLogger(__name__).warning(
+            "[startup] Hub sync DISABLED: HF_TOKEN/HF_CONFIG_REPO not set, "
+            "user files will only persist locally (lost on Space rebuild). "
+            "Set HF_CONFIG_REPO=<your-dataset-repo> and HF_TOKEN=<your-token> to enable."
+        )
+        sync_task = None
     yield
     # 关闭
     cleanup_task.cancel()
@@ -42,6 +91,8 @@ async def lifespan(app: FastAPI):
         await cleanup_task
     except asyncio.CancelledError:
         pass
+    if sync_task is not None:
+        await user_files_sync.stop_sync_task()
     await close_http_client()
     stop_db_writer()
     close_db()
@@ -78,6 +129,7 @@ def create_app() -> FastAPI:
 
     # 异常处理
     app.add_exception_handler(HttpError, http_error_handler)
+    app.add_exception_handler(RequestValidationError, validation_error_handler)
     app.add_exception_handler(Exception, unhandled_exception_handler)
 
     # 路由注册

app/services/user_files_store.py

@@ -0,0 +1,139 @@
+"""用户文件与账号的 HF Hub 持久化备份。
+
+与 config_store 同样的混合策略：
+- SQLite / 本地磁盘为热存储
+- HF Hub dataset 仓库做备份，解决 HF Spaces 免费档 ephemeral 问题
+- 写入时先写本地，再异步推 Hub（Hub 失败不影响主流程）
+- 读取时本地命中即返回；未命中从 Hub 拉取并写本地
+- 未配置 HF_TOKEN / HF_CONFIG_REPO 时自动降级为纯本地
+
+命名空间设计（按用户隔离）：
+- user_accounts            用户账号备份，key=username，value=JSON{id,username,password_hash,created_at}
+- user_files/<user_id>     文件内容，key=file_key，value=bytes
+- user_files_meta/<user_id> 文件元信息，key=file_key，value=JSON
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+from typing import Optional
+
+from ..hf_storage import (
+    delete_from_hub,
+    download_bytes,
+    list_hub_keys,
+    push_to_hub,
+    upload_bytes,
+)
+
+NS_USER_ACCOUNTS = "user_accounts"
+NS_USER_FILES_PREFIX = "user_files"        # 实际: user_files/<user_id>
+NS_USER_FILES_META_PREFIX = "user_files_meta"  # 实际: user_files_meta/<user_id>
+
+
+def _files_ns(user_id: str) -> str:
+    return f"{NS_USER_FILES_PREFIX}/{user_id}"
+
+
+def _meta_ns(user_id: str) -> str:
+    return f"{NS_USER_FILES_META_PREFIX}/{user_id}"
+
+
+# ===== 用户账号备份 =====
+
+def backup_user_account(
+    *, user_id: int, username: str, password_hash: str, created_at: int
+) -> None:
+    """同步备份用户账号到 Hub 本地缓存（远端推送用 async 版）。"""
+    payload = json.dumps(
+        {
+            "id": user_id,
+            "username": username,
+            "password_hash": password_hash,
+            "created_at": created_at,
+        },
+        ensure_ascii=False,
+    ).encode("utf-8")
+    upload_bytes(NS_USER_ACCOUNTS, username, payload)
+
+
+async def push_user_account(username: str) -> bool:
+    """异步推送用户账号到 Hub 远端。"""
+    return await push_to_hub(NS_USER_ACCOUNTS, username)
+
+
+def restore_user_account(username: str) -> Optional[dict]:
+    """从 Hub（本地缓存优先）拉取用户账号。返回 dict 或 None。"""
+    data = download_bytes(NS_USER_ACCOUNTS, username)
+    if not data:
+        return None
+    try:
+        obj = json.loads(data.decode("utf-8"))
+        if isinstance(obj, dict) and obj.get("username") and obj.get("password_hash"):
+            return obj
+    except Exception as e:
+        print(f"[user_files_store] parse user_account failed: {e}")
+    return None
+
+
+# ===== 文件内容 =====
+
+def upload_user_file_content(user_id: str, key: str, content: bytes) -> None:
+    """写文件内容到本地缓存（Hub 远端推送用 async 版）。"""
+    upload_bytes(_files_ns(user_id), key, content)
+
+
+async def push_user_file_content(user_id: str, key: str) -> bool:
+    """异步推送文件内容到 Hub 远端。"""
+    return await push_to_hub(_files_ns(user_id), key)
+
+
+def download_user_file_content(user_id: str, key: str) -> Optional[bytes]:
+    """下载文件内容：本地优先，Hub 回退。"""
+    return download_bytes(_files_ns(user_id), key)
+
+
+def delete_user_file_content(user_id: str, key: str) -> bool:
+    """删除文件内容：本地 + Hub。"""
+    return delete_from_hub(_files_ns(user_id), key)
+
+
+# ===== 文件元信息 =====
+
+def backup_file_meta(user_id: str, key: str, meta: dict) -> None:
+    """同步备份文件元信息到本地缓存。"""
+    payload = json.dumps(meta, ensure_ascii=False).encode("utf-8")
+    upload_bytes(_meta_ns(user_id), key, payload)
+
+
+async def push_file_meta(user_id: str, key: str) -> bool:
+    """异步推送文件元信息到 Hub 远端。"""
+    return await push_to_hub(_meta_ns(user_id), key)
+
+
+def delete_file_meta(user_id: str, key: str) -> bool:
+    """删除文件元信息：本地 + Hub。"""
+    return delete_from_hub(_meta_ns(user_id), key)
+
+
+async def restore_file_metas(user_id: str) -> list[dict]:
+    """从 Hub 拉取该用户所有文件元信息（用于 Space 重建后恢复）。
+
+    流程：list_hub_keys 列出 meta namespace 下所有 key → 逐个 download。
+    """
+    ns = _meta_ns(user_id)
+    keys = await asyncio.to_thread(list_hub_keys, ns)
+    if not keys:
+        return []
+    out: list[dict] = []
+    for k in keys:
+        data = download_bytes(ns, k)
+        if not data:
+            continue
+        try:
+            obj = json.loads(data.decode("utf-8"))
+            if isinstance(obj, dict) and obj.get("key"):
+                out.append(obj)
+        except Exception:
+            continue
+    return out

app/services/user_files_sync.py

@@ -0,0 +1,146 @@
+"""用户文件 Hub 同步后台任务。
+
+定时扫描本地缓存，把未同步到 Hub 的文件推上去。
+作为 asyncio.create_task(push_to_hub) 的兜底：如果异步推送失败或进程崩溃，
+下次定时任务会重新扫描并补推。
+
+仅在 Hub 启用时实际工作（is_hub_enabled()=True）。
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+from pathlib import Path
+
+from ..hf_storage import (
+    _local_path,
+    _local_root,
+    is_hub_enabled,
+    list_hub_keys,
+    push_to_hub,
+)
+
+logger = logging.getLogger(__name__)
+
+# 同步间隔（秒）
+SYNC_INTERVAL_SEC = 300  # 5 分钟
+
+# 需要定时扫描同步的 namespace 前缀
+SYNC_NS_PREFIXES = ["user_accounts", "user_files/", "user_files_meta/"]
+
+_sync_task: asyncio.Task | None = None
+
+
+def _collect_local_files() -> list[tuple[str, str]]:
+    """扫描本地缓存，返回 (namespace, key) 列表。
+
+    namespace 从路径前缀推断：
+    - user_accounts/<username>           → ns="user_accounts", key=<username>
+    - user_files/<user_id>/<file_key>    → ns="user_files/<user_id>", key=<file_key>
+    - user_files_meta/<user_id>/<key>    → ns="user_files_meta/<user_id>", key=<key>
+    """
+    root = _local_root()
+    out: list[tuple[str, str]] = []
+    if not root.exists():
+        return out
+
+    for ns_prefix in SYNC_NS_PREFIXES:
+        parts = ns_prefix.rstrip("/").split("/")
+        ns_dir = root.joinpath(*parts)
+        if not ns_dir.exists() or not ns_dir.is_dir():
+            continue
+        for p in ns_dir.rglob("*"):
+            if not p.is_file():
+                continue
+            try:
+                rel = p.relative_to(ns_dir).as_posix()
+                if not rel:
+                    continue
+                if ns_prefix == "user_accounts/":
+                    ns = "user_accounts"
+                else:
+                    ns = ns_prefix.rstrip("/")
+                out.append((ns, rel))
+            except Exception:
+                continue
+    return out
+
+
+async def _sync_once() -> int:
+    """执行一轮同步：扫描本地 → 对比 Hub → 补推缺失文件。返回补推数量。"""
+    if not is_hub_enabled():
+        return 0
+
+    local_files = await asyncio.to_thread(_collect_local_files)
+    if not local_files:
+        return 0
+
+    # 按 namespace 分组，批量 list_hub_keys 减少调用
+    ns_set: dict[str, list[str]] = {}
+    for ns, key in local_files:
+        ns_set.setdefault(ns, []).append(key)
+
+    pushed = 0
+    for ns, keys in ns_set.items():
+        try:
+            hub_keys = set(await asyncio.to_thread(list_hub_keys, ns))
+        except Exception as e:
+            logger.warning("[user_files_sync] list_hub_keys failed ns=%s: %s", ns, e)
+            continue
+
+        for key in keys:
+            if key in hub_keys:
+                continue
+            # 本地有但 Hub 没有，补推
+            try:
+                ok = await push_to_hub(ns, key)
+                if ok:
+                    pushed += 1
+                    logger.info("[user_files_sync] pushed ns=%s key=%s", ns, key)
+            except Exception as e:
+                logger.warning("[user_files_sync] push failed ns=%s key=%s: %s", ns, key, e)
+
+    if pushed > 0:
+        logger.info("[user_files_sync] sync round done, pushed %d files", pushed)
+    return pushed
+
+
+async def sync_loop():
+    """定时同步循环。"""
+    logger.info(
+        "[user_files_sync] started, interval=%ds, hub_enabled=%s",
+        SYNC_INTERVAL_SEC, is_hub_enabled(),
+    )
+    # 启动后先等一小段时间，让应用初始化完成
+    await asyncio.sleep(10)
+    while True:
+        try:
+            await _sync_once()
+        except Exception as e:
+            logger.warning("[user_files_sync] round failed: %s", e)
+        try:
+            await asyncio.sleep(SYNC_INTERVAL_SEC)
+        except asyncio.CancelledError:
+            break
+
+
+def start_sync_task() -> asyncio.Task:
+    """启动同步后台任务。"""
+    global _sync_task
+    if _sync_task is not None and not _sync_task.done():
+        return _sync_task
+    _sync_task = asyncio.create_task(sync_loop())
+    return _sync_task
+
+
+async def stop_sync_task() -> None:
+    """停止同步后台任务。"""
+    global _sync_task
+    if _sync_task is None:
+        return
+    _sync_task.cancel()
+    try:
+        await _sync_task
+    except asyncio.CancelledError:
+        pass
+    _sync_task = None