"""v5 备份 API:/u/api/backups/v5/* 方案 C:文件级细粒度 blob + Manifest + ZIP 整包恢复通道。 端点(11 个): - GET /sets 列出当前用户备份集 - POST /sets 创建待提交 set - GET /sets/{setId} 获取指定 set 的完整 manifest - DELETE /sets/{setId} 删除备份集(refs--) - POST /sets/{setId}/manifest 提交 manifest,原子标记为最新 - GET /sets/{setId}/archive ZIP 整包下载(流式) - POST /blobs/upload 上传单个 blob(multipart),返回 key=sha256 - GET /blobs/{key} 下载单个 blob 原始 bytes - GET /blobs/{key}/meta 查询 blob 元信息 - POST /blobs/check 批量查询 blob 是否已存在(按 key) - POST /blobs/batch_check 卸载重装兜底(按 path+size+mtime) """ from __future__ import annotations import asyncio import json import logging import urllib.parse from fastapi import APIRouter, Depends, File, Form, Request, UploadFile from fastapi.responses import FileResponse, JSONResponse, StreamingResponse from pydantic import BaseModel, Field from typing import List, Optional from ..errors import HttpError from ..services import backup_v5_store from ._common import CORS_HEADERS, ok_with_cors, read_json_body from .user_files import require_user, require_user_for_download logger = logging.getLogger(__name__) router = APIRouter(prefix="/u/api/backups/v5", tags=["user-backups-v5"]) MAX_BLOB_SIZE = 10 * 1024 * 1024 # 框架硬限制 def _err(code: str, message: str, status: int = 400, **extra) -> HttpError: detail = {"code": code, "message": message} if extra: detail["details"] = extra return HttpError(message, status=status, code=code) # ===== sets ===== @router.get("/sets") async def list_sets(request: Request): payload = require_user(request) user_id = str(payload.get("sub") or "") if not user_id: raise _err("unauthorized", "missing user", status=401) try: limit = int(request.query_params.get("limit") or "20") offset = int(request.query_params.get("offset") or "0") except Exception: limit, offset = 20, 0 # 本地 DB 为空时触发 Hub 同步(HF Space 重建后 DB 丢失,Hub 上 manifest 仍在) if backup_v5_store.count_sets(user_id) == 0: asyncio.create_task(backup_v5_store._sync_manifests_background()) items = backup_v5_store.list_sets(user_id, limit=limit, offset=offset) return ok_with_cors({"items": items, "count": len(items), "total": len(items)}) @router.post("/sets") async def create_set(request: Request): payload = require_user(request) user_id = str(payload.get("sub") or "") if not user_id: raise _err("unauthorized", "missing user", status=401) body = await read_json_body(request) alias = str(body.get("alias") or "").strip()[:200] try: result = backup_v5_store.create_pending_set(user_id, alias=alias) except ValueError as e: msg = str(e) if msg == "quota_exceeded": raise _err("quota_exceeded", "已达备份集上限(5 份),请先删除旧的备份", status=409) raise _err("create_failed", msg, status=500) # 注意:小天才 fetch.fetch 对非 200 的 2xx(如 201)会触发 fail 回调而非 success, # 导致前端报"网络错误 201 Created"。这里统一返回 200 保证兼容。 return ok_with_cors(result, status=200) @router.get("/sets/{set_id}") async def get_set(set_id: str, request: Request): payload = require_user(request) user_id = str(payload.get("sub") or "") if not user_id: raise _err("unauthorized", "missing user", status=401) owner = backup_v5_store.get_set_owner(set_id) if not owner: raise _err("set_not_found", "备份集不存在", status=404) if owner != user_id: raise _err("forbidden", "无权访问他人备份", status=403) manifest = backup_v5_store.get_set_manifest(set_id) if manifest is None: raise _err("manifest_missing", "manifest 缺失", status=500) return ok_with_cors({"manifest": manifest}) @router.delete("/sets/{set_id}") async def delete_set(set_id: str, request: Request): payload = require_user(request) user_id = str(payload.get("sub") or "") if not user_id: raise _err("unauthorized", "missing user", status=401) owner = backup_v5_store.get_set_owner(set_id) if not owner: raise _err("set_not_found", "备份集不存在", status=404) if owner != user_id: raise _err("forbidden", "无权删除他人备份", status=403) try: result = backup_v5_store.delete_set(set_id, user_id) except ValueError as e: msg = str(e) if msg == "set_not_found": raise _err("set_not_found", "备份集不存在", status=404) if msg == "forbidden": raise _err("forbidden", "无权删除他人备份", status=403) raise _err("delete_failed", msg, status=500) return ok_with_cors({"deleted": True, "setId": set_id, **result}) @router.post("/sets/{set_id}/manifest") async def commit_manifest(set_id: str, request: Request): payload = require_user(request) user_id = str(payload.get("sub") or "") if not user_id: raise _err("unauthorized", "missing user", status=401) body = await read_json_body(request) manifest = body.get("manifest") if not isinstance(manifest, dict): raise _err("invalid_manifest", "manifest 必须是 JSON 对象") # 强制 setId 与 set_id 一致 if str(manifest.get("setId") or "") != set_id: manifest["setId"] = set_id try: result = backup_v5_store.commit_set(set_id, manifest, user_id) except ValueError as e: msg = str(e) if msg == "set_not_pending": raise _err("set_not_pending", "set 不在待提交状态(已提交或已过期)") if msg == "invalid_manifest": raise _err("invalid_manifest", "manifest 格式无效") if msg == "invalid_manifest_version": raise _err("invalid_manifest_version", "manifest.version 必须为 5") if msg == "invalid_manifest_blobs": raise _err("invalid_manifest_blobs", "manifest.blobs 必须是数组") if msg == "empty_blobs": raise _err("empty_blobs", "manifest.blobs 不能为空") if msg.startswith("blob_missing:"): missing = msg.split(":", 1)[1].split(",") raise _err("blob_missing", "manifest 引用了未上传的 blob", missing_keys=missing) if msg == "quota_exceeded": raise _err("quota_exceeded", "已达备份集上限", status=409) raise _err("commit_failed", msg, status=500) return ok_with_cors(result) # ===== ZIP 整包下载 ===== @router.get("/sets/{set_id}/archive") async def download_archive(set_id: str, request: Request): payload = require_user_for_download(request) user_id = str(payload.get("sub") or "") if not user_id: raise _err("unauthorized", "missing user", status=401) owner = backup_v5_store.get_set_owner(set_id) if not owner: raise _err("set_not_found", "备份集不存在", status=404) if owner != user_id: raise _err("forbidden", "无权下载他人备份", status=403) # 检查总大小 manifest = backup_v5_store.get_set_manifest(set_id) if manifest is None: raise _err("manifest_missing", "manifest 缺失", status=500) total_size = sum(int(b.get("size") or 0) for b in manifest.get("blobs", [])) if total_size > backup_v5_store.MAX_ZIP_SIZE: raise _err( "set_too_large", f"备份集 {total_size} 字节超过 ZIP 单文件限制 {backup_v5_store.MAX_ZIP_SIZE} 字节,请走 blob 通道", status=400, totalSize=total_size, maxSize=backup_v5_store.MAX_ZIP_SIZE, ) try: zip_path, zip_size = await asyncio.to_thread( backup_v5_store.build_zip_to_file, set_id ) except ValueError as e: msg = str(e) if msg == "set_not_found": raise _err("set_not_found", "备份集不存在", status=404) if msg == "set_too_large": raise _err("set_too_large", "备份集过大", status=400) raise _err("zip_build_failed", msg, status=500) filename = f"backup_{set_id[:8]}.zip" # 流式返回,传输完成后由 generator 删除临时文件 generator = backup_v5_store.zip_iter_chunks(zip_path) headers = { "Content-Disposition": f'attachment; filename="{filename}"', "Content-Length": str(zip_size), **CORS_HEADERS, } return StreamingResponse(generator, media_type="application/zip", headers=headers) # ===== blobs ===== @router.post("/blobs/upload") async def upload_blob( request: Request, file: UploadFile = File(...), path: str = Form(""), type: str = Form(""), alias: Optional[str] = Form(None), id: Optional[str] = Form(None), ): payload = require_user(request) user_id = str(payload.get("sub") or "") if not user_id: raise _err("unauthorized", "missing user", status=401) # 读文件内容(≤10MB,可一次性读) content = await file.read() if len(content) > MAX_BLOB_SIZE: raise _err("blob_too_large", f"blob 超过 {MAX_BLOB_SIZE} 字节", status=413) # MIME 推断 mime = str(file.content_type or "") if not mime: # 简单根据扩展名推断 ext = (path or file.filename or "").rsplit(".", 1)[-1].lower() if "." in (path or file.filename or "") else "" mime_map = { "json": "application/json", "txt": "text/plain", "md": "text/markdown", "png": "image/png", "jpg": "image/jpeg", "jpeg": "image/jpeg", "gif": "image/gif", "webp": "image/webp", "amr": "audio/amr", "mp3": "audio/mpeg", "wav": "audio/wav", } mime = mime_map.get(ext, "application/octet-stream") try: result = backup_v5_store.store_blob(content, mime, user_id) except ValueError as e: msg = str(e) if msg == "blob_too_large": raise _err("blob_too_large", "blob 过大", status=413) raise _err("store_failed", msg, status=500) return ok_with_cors(result) @router.get("/blobs/{key}") async def download_blob(key: str, request: Request): payload = require_user_for_download(request) user_id = str(payload.get("sub") or "") if not user_id: raise _err("unauthorized", "missing user", status=401) if not _is_sha256_hex(key): raise _err("invalid_key", "key 必须是 64 位 sha256 hex", status=400) blob_path = backup_v5_store.get_blob_local_path(key) if not blob_path or not blob_path.exists(): raise _err("blob_not_found", "blob 不存在", status=404) # 取 mime meta = backup_v5_store.get_blob_meta(key) mime = (meta or {}).get("mime") or "application/octet-stream" filename = blob_path.name headers = { "Content-Disposition": f'attachment; filename="{filename}"', **CORS_HEADERS, } return FileResponse(str(blob_path), media_type=mime, headers=headers) @router.get("/blobs/{key}/meta") async def get_blob_meta(key: str, request: Request): payload = require_user(request) if not payload.get("sub"): raise _err("unauthorized", "missing user", status=401) if not _is_sha256_hex(key): raise _err("invalid_key", "key 必须是 64 位 sha256 hex", status=400) meta = backup_v5_store.get_blob_meta(key) if not meta: raise _err("blob_not_found", "blob 不存在", status=404) return ok_with_cors(meta) @router.post("/blobs/check") async def check_blobs(request: Request): payload = require_user(request) if not payload.get("sub"): raise _err("unauthorized", "missing user", status=401) body = await read_json_body(request) keys = body.get("keys") if not isinstance(keys, list): raise _err("invalid_keys", "keys 必须是数组") if len(keys) > 100: raise _err("too_many_keys", "单次最多 100 个 key") # 校验每个 key 是合法 sha256 hex cleaned = [] for k in keys: s = str(k or "").strip() if not _is_sha256_hex(s): continue cleaned.append(s) result = backup_v5_store.check_blobs(cleaned) return ok_with_cors(result) @router.post("/blobs/batch_check") async def batch_check_blobs(request: Request): payload = require_user(request) if not payload.get("sub"): raise _err("unauthorized", "missing user", status=401) body = await read_json_body(request) prev_set_id = body.get("prevSetId") if prev_set_id is not None: prev_set_id = str(prev_set_id).strip() or None paths = body.get("paths") if not isinstance(paths, list): raise _err("invalid_paths", "paths 必须是数组") if len(paths) > backup_v5_store.BATCH_CHECK_MAX: raise _err("too_many_paths", f"单次最多 {backup_v5_store.BATCH_CHECK_MAX} 个 path") # 清洗 cleaned = [] for p in paths: if not isinstance(p, dict): continue path = str(p.get("path") or "").strip() if not path: continue try: size = int(p.get("size") or 0) except Exception: size = 0 try: mtime = int(p.get("mtime") or 0) except Exception: mtime = 0 cleaned.append({"path": path, "size": size, "mtime": mtime}) result = backup_v5_store.batch_check_blobs(prev_set_id, cleaned) return ok_with_cors(result) def _is_sha256_hex(s: str) -> bool: if len(s) != 64: return False try: int(s, 16) return True except Exception: return False