app/api/user_backups_v5.py

"""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