app/api/user_backups.py

"""用户备份 API：/u/api/backups/*

面向手表端的云端备份/恢复功能，与普通文件(/u/api/files)区分开。

复用 /u/api/* 的账号 JWT 鉴权(require_user)，备份落到独立的 Hub 命名空间
(user_backups/<user_id>)，并在 file_meta 表用 namespace='user_backups' 标记。

别名设计：本地小天才不允许存中文名，故客户端本地用英文文件名，中文别名通过
表单字段 alias 上送，后端存 file_meta.alias 用于展示与下载文件名。

- GET    /u/api/backups              列出当前用户的备份
- POST   /u/api/backups/upload       上传新备份（alias 来自 filename 或表单字段）
- PUT    /u/api/backups/{key}        替换备份内容（同 key 覆盖，刷新 alias/size/sha）
- PUT    /u/api/backups/{key}/alias  重命名备份别名（body {alias}）
- GET    /u/api/backups/{key}        下载备份（仅 owner）
- DELETE /u/api/backups/{key}        删除备份（仅 owner）
"""
from __future__ import annotations

import asyncio
import hashlib
import json
import logging
import time
import urllib.parse

from cachetools import TTLCache
from fastapi import APIRouter, Depends, File, Form, Request, UploadFile
from fastapi.responses import Response

from ..database import get_conn
from ..errors import HttpError
from ..hf_storage import is_hub_enabled
from ..services import user_files_store
from ._common import CORS_HEADERS, ok_with_cors, read_json_body
from .user_files import (
    _make_english_filename,
    _resolve_alias,
    require_user,
)

router = APIRouter(prefix="/u/api/backups", tags=["user-backups"])

# 备份通常比单文件大（含全部会话/配置），放宽到 50MB
MAX_BACKUP_SIZE = 50 * 1024 * 1024

logger = logging.getLogger(__name__)


def _list_backups_from_db(user_id: str) -> list[dict]:
    with get_conn() as conn:
        rows = conn.execute(
            "SELECT key, filename, alias, mime, size, sha256, uploaded_at "
            "FROM file_meta WHERE uploaded_by = ? AND namespace = ? "
            "ORDER BY uploaded_at DESC",
            (user_id, user_files_store.NS_USER_BACKUPS),
        ).fetchall()
    out = []
    for r in rows:
        d = dict(r)
        if not d.get("alias"):
            d["alias"] = d.get("filename", "") or ""
        out.append(d)
    return out


async def _restore_backups_from_hub_async(user_id: str) -> int:
    metas = await user_files_store.restore_backup_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, alias, mime, size, "
                    "sha256, uploaded_at, uploaded_by, access_key, refs) "
                    "VALUES(?,?,?,?,?,?,?,?,?,?,?)",
                    (
                        m["key"], user_files_store.NS_USER_BACKUPS,
                        m.get("filename", "unnamed"), m.get("alias", ""),
                        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_backups] restore backup_meta {m.get('key')} failed: {e}")
    return inserted


def _fetch_backup_content(user_id: str, key: str) -> bytes | None:
    """同步获取备份内容：本地优先，Hub 回退。"""
    return user_files_store.download_user_backup_content(user_id, key)


# 备份 payload 内存缓存：分部分下载时，同一备份会被请求 7 次（config/sessions/
# thoughts/prompts/files/aliasMap/archives），若每次都从 Hub 加载整个 2MB 备份
# + json.loads 解析，7 次重复加载解析导致下载会话极慢（用户反馈 60s 超时）。
# 缓存解析后的 payload，TTL 60s，7 次请求只加载 1 次，后续近乎 0 延迟。
# maxsize=32：同一用户最多 32 个备份缓存条目，足够恢复场景使用。
_backup_payload_cache: "TTLCache[tuple[str, str], dict]" = TTLCache(maxsize=32, ttl=60)


def _invalidate_backup_payload_cache(user_id: str, key: str | None = None) -> None:
    """失效备份 payload 缓存：上传/替换/删除时调用。"""
    if key is not None:
        _backup_payload_cache.pop((user_id, key), None)
    else:
        # 清理该用户所有缓存条目
        for ck in list(_backup_payload_cache.keys()):
            if ck[0] == user_id:
                _backup_payload_cache.pop(ck, None)


async def _get_backup_payload(user_id: str, key: str) -> dict:
    """获取备份解析后的 payload，带内存缓存。"""
    cache_key = (user_id, key)
    cached = _backup_payload_cache.get(cache_key)
    if cached is not None:
        return cached
    try:
        content = await asyncio.wait_for(
            asyncio.to_thread(_fetch_backup_content, user_id, key),
            timeout=15.0,
        )
    except asyncio.TimeoutError:
        logger.warning("[backup part] content fetch timeout uid=%s key=%s", user_id, key)
        raise HttpError("backup content fetch timeout", status=504, code="upstream_timeout")
    if content is None:
        raise HttpError("backup content missing", status=410, code="gone")
    try:
        payload = json.loads(content.decode("utf-8"))
    except Exception as e:
        logger.warning("[backup part] parse failed uid=%s key=%s: %s", user_id, key, e)
        raise HttpError("backup parse failed", status=500, code="parse_error")
    if not isinstance(payload, dict):
        raise HttpError("backup format invalid", status=500, code="parse_error")
    _backup_payload_cache[cache_key] = payload
    return payload


# ===== 列表 =====

@router.get("")
@router.get("/")
async def list_my_backups(user: dict = Depends(require_user)):
    user_id = str(user["sub"])
    items = _list_backups_from_db(user_id)
    # SQLite 为空时尝试从 Hub 恢复（Space 重建场景）
    if not items and is_hub_enabled():
        try:
            restored = await asyncio.wait_for(
                _restore_backups_from_hub_async(user_id), timeout=8.0
            )
            if restored > 0:
                items = _list_backups_from_db(user_id)
        except asyncio.TimeoutError:
            logger.warning("[backups list] restore from hub timeout uid=%s", user_id)
        except Exception as e:
            logger.warning("[backups list] restore from hub failed uid=%s: %s", user_id, e)
    return ok_with_cors({"items": items, "count": len(items)})


# ===== 上传 =====

@router.post("/upload")
async def upload_backup(
    user: dict = Depends(require_user),
    file: UploadFile = File(...),
    alias: str | None = Form(default=None),
):
    user_id = str(user["sub"])
    username = user["username"]

    content = await file.read()
    if len(content) > MAX_BACKUP_SIZE:
        raise HttpError(
            f"backup too large: {len(content)} bytes > 50MB ({MAX_BACKUP_SIZE} bytes)",
            status=413, code="too_large",
        )
    if not content:
        raise HttpError("empty backup", status=400, code="bad_request")

    key = hashlib.sha256(content).hexdigest()  # 备份内容哈希作 key，便于去重识别
    # alias：表单字段优先，回退原始文件名（可能含中文）
    raw_alias = (alias or file.filename or "unnamed").strip()
    if len(raw_alias) > 255:
        raw_alias = raw_alias[:255]
    filename = _make_english_filename(raw_alias, key)  # 随机英文名，本地保存用
    mime = (file.content_type or "application/octet-stream").strip()
    size = len(content)
    sha256 = key
    now = int(time.time())

    # 同一 alias 视为「替换」：先删旧 key 的内容/元信息（不同 key 时）
    old_key = None
    with get_conn() as conn:
        row = conn.execute(
            "SELECT key FROM file_meta WHERE uploaded_by = ? AND namespace = ? AND alias = ?",
            (user_id, user_files_store.NS_USER_BACKUPS, raw_alias),
        ).fetchone()
        if row and row["key"] != key:
            old_key = row["key"]
    if old_key:
        try:
            user_files_store.delete_user_backup_content(user_id, old_key)
            user_files_store.delete_backup_meta(user_id, old_key)
            with get_conn() as conn:
                conn.execute("DELETE FROM file_meta WHERE key = ?", (old_key,))
        except Exception as e:
            logger.warning("[backup upload] cleanup old alias=%s key=%s: %s", raw_alias, old_key, e)
        _invalidate_backup_payload_cache(user_id, old_key)

    # 写内容到本地缓存 + 异步推 Hub
    try:
        user_files_store.upload_user_backup_content(user_id, key, content)
        logger.info(
            "[backup upload] user=%s uid=%s key=%s alias=%s filename=%s size=%d, hub=%s",
            username, user_id, key, raw_alias, filename, size, is_hub_enabled(),
        )
        asyncio.create_task(user_files_store.push_user_backup_content(user_id, key))
    except Exception as e:
        logger.error("[backup upload] store failed user=%s alias=%s: %s", username, raw_alias, e)
        raise HttpError("failed to store backup", status=500, code="internal_error") from e
    _invalidate_backup_payload_cache(user_id, key)

    meta = {
        "key": key,
        "namespace": user_files_store.NS_USER_BACKUPS,
        "filename": filename,
        "alias": raw_alias,
        "mime": mime,
        "size": size,
        "sha256": sha256,
        "uploaded_at": now,
        "uploaded_by": user_id,
        "access_key": username,
    }
    with get_conn() as conn:
        # 同 key（内容完全相同）覆盖旧 meta
        conn.execute(
            "INSERT INTO file_meta(key, namespace, filename, alias, mime, size, sha256, "
            "uploaded_at, uploaded_by, access_key, refs) "
            "VALUES(?,?,?,?,?,?,?,?,?,?,?) "
            "ON CONFLICT(key) DO UPDATE SET filename=excluded.filename, alias=excluded.alias, "
            "mime=excluded.mime, size=excluded.size, sha256=excluded.sha256, "
            "uploaded_at=excluded.uploaded_at, access_key=excluded.access_key",
            (key, user_files_store.NS_USER_BACKUPS, filename, raw_alias, mime,
             size, sha256, now, user_id, username, ""),
        )

    try:
        user_files_store.backup_backup_meta(user_id, key, meta)
        asyncio.create_task(user_files_store.push_backup_meta(user_id, key))
    except Exception as e:
        print(f"[user_backups] backup_backup_meta failed: {e}")

    return ok_with_cors({
        "key": key,
        "filename": filename,
        "alias": raw_alias,
        "mime": mime,
        "size": size,
        "sha256": sha256,
        "uploaded_at": now,
    })


# ===== 替换内容（同 key 覆盖）=====

@router.put("/{key}")
async def replace_backup(
    key: str,
    user: dict = Depends(require_user),
    file: UploadFile = File(...),
    alias: str | None = Form(default=None),
):
    user_id = str(user["sub"])
    with get_conn() as conn:
        row = conn.execute(
            "SELECT uploaded_by, namespace FROM file_meta WHERE key = ?",
            (key,),
        ).fetchone()
    if not row:
        raise HttpError("backup not found", status=404, code="not_found")
    if row["uploaded_by"] != user_id or row["namespace"] != user_files_store.NS_USER_BACKUPS:
        raise HttpError("forbidden", status=403, code="forbidden")

    content = await file.read()
    if len(content) > MAX_BACKUP_SIZE:
        raise HttpError(
            f"backup too large: {len(content)} bytes > 50MB",
            status=413, code="too_large",
        )
    if not content:
        raise HttpError("empty backup", status=400, code="bad_request")

    raw_alias = (alias or file.filename or "unnamed").strip()
    if len(raw_alias) > 255:
        raw_alias = raw_alias[:255]
    filename = _make_english_filename(raw_alias, key)
    mime = (file.content_type or "application/octet-stream").strip()
    size = len(content)
    sha256 = hashlib.sha256(content).hexdigest()
    now = int(time.time())
    username = user["username"]

    try:
        user_files_store.upload_user_backup_content(user_id, key, content)
        asyncio.create_task(user_files_store.push_user_backup_content(user_id, key))
    except Exception as e:
        logger.error("[backup replace] store failed key=%s: %s", key, e)
        raise HttpError("failed to store backup", status=500, code="internal_error") from e
    _invalidate_backup_payload_cache(user_id, key)

    with get_conn() as conn:
        conn.execute(
            "UPDATE file_meta SET filename=?, alias=?, mime=?, size=?, sha256=?, "
            "uploaded_at=?, access_key=? WHERE key=?",
            (filename, raw_alias, mime, size, sha256, now, username, key),
        )

    meta = {
        "key": key,
        "namespace": user_files_store.NS_USER_BACKUPS,
        "filename": filename,
        "alias": raw_alias,
        "mime": mime,
        "size": size,
        "sha256": sha256,
        "uploaded_at": now,
        "uploaded_by": user_id,
        "access_key": username,
    }
    try:
        user_files_store.backup_backup_meta(user_id, key, meta)
        asyncio.create_task(user_files_store.push_backup_meta(user_id, key))
    except Exception as e:
        print(f"[user_backups] replace backup_meta failed: {e}")

    return ok_with_cors({
        "key": key, "filename": filename, "alias": raw_alias,
        "mime": mime, "size": size, "sha256": sha256, "uploaded_at": now,
    })


# ===== 重命名别名 =====

@router.put("/{key}/alias")
async def rename_backup(
    key: str,
    request: Request,
    user: dict = Depends(require_user),
):
    body = await read_json_body(request)
    alias = str(body.get("alias") or "").strip()
    if not alias:
        raise HttpError("alias is required", status=400, code="bad_request")
    if len(alias) > 255:
        alias = alias[:255]

    user_id = str(user["sub"])
    with get_conn() as conn:
        row = conn.execute(
            "SELECT uploaded_by, namespace, filename, mime, size, sha256, uploaded_at, access_key "
            "FROM file_meta WHERE key = ?",
            (key,),
        ).fetchone()
        if not row:
            raise HttpError("backup not found", status=404, code="not_found")
        if row["uploaded_by"] != user_id or row["namespace"] != user_files_store.NS_USER_BACKUPS:
            raise HttpError("forbidden", status=403, code="forbidden")
        conn.execute(
            "UPDATE file_meta SET alias=? WHERE key=?",
            (alias, key),
        )

    meta = {
        "key": key,
        "namespace": user_files_store.NS_USER_BACKUPS,
        "filename": row["filename"],
        "alias": alias,
        "mime": row["mime"],
        "size": row["size"],
        "sha256": row["sha256"],
        "uploaded_at": row["uploaded_at"],
        "uploaded_by": user_id,
        "access_key": row["access_key"],
    }
    try:
        user_files_store.backup_backup_meta(user_id, key, meta)
        asyncio.create_task(user_files_store.push_backup_meta(user_id, key))
    except Exception as e:
        print(f"[user_backups] rename backup_meta failed: {e}")

    return ok_with_cors({"key": key, "alias": alias})


# ===== 下载 =====

@router.get("/{key}")
async def download_backup(
    key: str,
    user: dict = Depends(require_user),
):
    user_id = str(user["sub"])
    with get_conn() as conn:
        row = conn.execute(
            "SELECT filename, alias, mime, uploaded_by, namespace FROM file_meta WHERE key = ?",
            (key,),
        ).fetchone()
    # SQLite 没有则尝试从 Hub 恢复单条 meta（Space 重建场景）
    if not row and is_hub_enabled():
        try:
            metas = await asyncio.wait_for(
                user_files_store.restore_backup_metas(user_id), timeout=8.0
            )
            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, alias, mime, size, "
                            "sha256, uploaded_at, uploaded_by, access_key, refs) "
                            "VALUES(?,?,?,?,?,?,?,?,?,?,?)",
                            (target["key"], user_files_store.NS_USER_BACKUPS,
                             target.get("filename", "unnamed"), target.get("alias", ""),
                             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, alias, mime, uploaded_by, namespace FROM file_meta WHERE key = ?",
                        (key,),
                    ).fetchone()
        except asyncio.TimeoutError:
            logger.warning("[backup download] restore meta timeout uid=%s key=%s", user_id, key)
        except Exception as e:
            logger.warning("[backup download] restore meta failed uid=%s key=%s: %s", user_id, key, e)
    if not row:
        raise HttpError("backup not found", status=404, code="not_found")
    if row["uploaded_by"] != user_id or row["namespace"] != user_files_store.NS_USER_BACKUPS:
        raise HttpError("forbidden", status=403, code="forbidden")

    try:
        content = await asyncio.wait_for(
            asyncio.to_thread(_fetch_backup_content, user_id, key),
            timeout=15.0,
        )
    except asyncio.TimeoutError:
        logger.warning("[backup download] content fetch timeout uid=%s key=%s", user_id, key)
        raise HttpError("backup content fetch timeout", status=504, code="upstream_timeout")
    if content is None:
        raise HttpError("backup content missing", status=410, code="gone")

    # 下载文件名：纯英文 backup_<key前8位>.json
    # 不用 alias（可能含中文，小天才本地无法存中文名），
    # 也不用 filename 字段（f_xxx_xxx 无扩展名，可读性差）。
    # 浏览器/客户端直接用此名保存，避免中文存储问题。
    download_name = "backup_" + str(key)[:8] + ".json"
    headers = {
        **CORS_HEADERS,
        "Content-Disposition": f"attachment; filename=\"{download_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")


# ===== 分部分下载（避免手表一次性加载整个大备份导致内存溢出崩溃） =====

_VALID_PARTS = {"config", "sessions", "prompts", "files", "thoughts", "aliasMap", "archives"}


@router.get("/{key}/part/{part_name}")
async def download_backup_part(
    key: str,
    part_name: str,
    user: dict = Depends(require_user),
):
    """分部分下载备份：返回备份 JSON 中指定字段的独立 JSON。

    手表 JS 引擎内存有限，一次性 readText 1.6MB + JSON.parse 创建大对象，
    内存峰值过高会触发系统级崩溃重启（非普通 OOM）。
    拆分成 config/sessions/prompts/files/thoughts 五部分单独下载，
    每部分通常 < 500KB，内存峰值大幅降低。
    """
    user_id = str(user["sub"])
    if part_name not in _VALID_PARTS:
        raise HttpError(f"invalid part: {part_name}", status=400, code="bad_request")

    with get_conn() as conn:
        row = conn.execute(
            "SELECT uploaded_by, namespace FROM file_meta WHERE key = ?",
            (key,),
        ).fetchone()
    if not row:
        raise HttpError("backup not found", status=404, code="not_found")
    if row["uploaded_by"] != user_id or row["namespace"] != user_files_store.NS_USER_BACKUPS:
        raise HttpError("forbidden", status=403, code="forbidden")

    # 使用内存缓存：7 次分部分下载只加载解析 1 次，后续近乎 0 延迟
    payload = await _get_backup_payload(user_id, key)

    part_data = payload.get(part_name)
    # v2 备份可能没有 prompts/files 字段，返回 null（前端按 hasPrompts/hasFiles=false 处理）
    if part_data is None:
        part_json = "null"
    else:
        part_json = json.dumps(part_data, ensure_ascii=False)

    return Response(
        content=part_json.encode("utf-8"),
        status_code=200,
        headers={**CORS_HEADERS, "Content-Type": "application/json"},
        media_type="application/json",
    )


# ===== 删除 =====

@router.delete("/{key}")
async def delete_backup(
    key: str,
    user: dict = Depends(require_user),
):
    user_id = str(user["sub"])
    with get_conn() as conn:
        row = conn.execute(
            "SELECT uploaded_by, namespace FROM file_meta WHERE key = ?",
            (key,),
        ).fetchone()
        if not row:
            raise HttpError("backup not found", status=404, code="not_found")
        if row["uploaded_by"] != user_id or row["namespace"] != user_files_store.NS_USER_BACKUPS:
            raise HttpError("forbidden", status=403, code="forbidden")
        conn.execute("DELETE FROM file_meta WHERE key = ?", (key,))

    try:
        user_files_store.delete_user_backup_content(user_id, key)
    except Exception as e:
        print(f"[user_backups] delete_user_backup_content failed: {e}")
    try:
        user_files_store.delete_backup_meta(user_id, key)
    except Exception as e:
        print(f"[user_backups] delete_backup_meta failed: {e}")
    _invalidate_backup_payload_cache(user_id, key)

    return ok_with_cors({"deleted": True, "key": key})