app/api/user_files.py

"""用户文件管理 API：/u/api/*

面向终端用户的文件管理（注册/登录/上传/下载/删除/列表）。

- POST   /u/api/register        注册用户名 + 密码
- POST   /u/api/login           登录，返回 JWT
- GET    /u/api/me              当前用户信息（校验 token）
- GET    /u/api/files           列出当前用户的文件
- POST   /u/api/upload          上传单个文件（≤10MB）
- GET    /u/api/files/{key}     下载文件（仅 owner）
- DELETE /u/api/files/{key}     删除文件（仅 owner）

持久化（与 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 typing import Optional

import bcrypt
import jwt
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 ..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"  # file_meta.namespace 字段值
TOKEN_TTL_SEC = 7 * 24 * 3600  # 7 天

_USERNAME_RE = re.compile(r"^[A-Za-z0-9_\u4e00-\u9fff.\-]{2,32}$")

logger = logging.getLogger(__name__)


def _make_english_filename(original: str, key: str) -> str:
    """根据原始文件名生成安全的英文存储文件名，保留扩展名。

    规则：f_{key前8位}_{时间戳后6位}.{ext}
    例如 original="我的文件.txt" -> "f_a1b2c3d4_8858a1.txt"
    """
    raw = str(original or "").strip()
    # 提取扩展名（只保留 ASCII 字母数字，最多 8 位）
    ext = ""
    dot_idx = raw.rfind(".")
    if dot_idx > 0:
        raw_ext = raw[dot_idx + 1:].lower()
        safe_ext = "".join(c for c in raw_ext if c.isalnum())[:8]
        if safe_ext:
            ext = "." + safe_ext
    short_key = key[:8] if key else "00000000"
    short_ts = str(int(time.time()))[-6:]
    return f"f_{short_key}_{short_ts}{ext}"


def _resolve_alias(row_or_meta) -> str:
    """从 DB 行或 meta dict 解析展示用的别名：优先 alias，回退 filename。"""
    if not row_or_meta:
        return ""
    keys = row_or_meta.keys() if hasattr(row_or_meta, "keys") else []
    if "alias" in keys:
        alias = row_or_meta["alias"]
        if alias:
            return alias
    if "filename" in keys:
        return row_or_meta["filename"] or ""
    return ""


# ===== JWT =====

def _issue_user_token(user_id: int, username: str) -> str:
    settings = get_settings()
    if not settings.has_jwt_secret:
        raise HttpError(
            "XTC_JWT_SECRET not configured, cannot issue user token",
            status=500,
            code="server_misconfigured",
        )
    now = int(time.time())
    payload = {
        "type": "user_file",
        "sub": str(user_id),
        "username": username,
        "iat": now,
        "exp": now + TOKEN_TTL_SEC,
    }
    return jwt.encode(payload, settings.xtc_jwt_secret, algorithm=ALGORITHM)


def _verify_user_token(token: str) -> dict:
    settings = get_settings()
    if not settings.has_jwt_secret:
        raise HttpError("server misconfigured", status=500, code="server_misconfigured")
    try:
        payload = jwt.decode(token, settings.xtc_jwt_secret, algorithms=[ALGORITHM])
    except jwt.PyJWTError:
        raise HttpError("invalid or expired token", status=401, code="unauthorized")
    if payload.get("type") != "user_file":
        raise HttpError("invalid token type", status=401, code="unauthorized")
    if not payload.get("sub") or not payload.get("username"):
        raise HttpError("invalid token payload", status=401, code="unauthorized")
    return payload


# ===== 依赖 =====

def _extract_token(request: Request) -> str:
    # 优先 query 参数（小天才手表等平台会拦截自定义 header，只能走 query）
    q = request.query_params.get("token") or ""
    q = q.strip()
    if q:
        return q
    auth = request.headers.get("authorization") or ""
    auth = auth.strip()
    if auth.lower().startswith("bearer "):
        return auth[7:].strip()
    return (request.headers.get("x-xtc-user-token") or "").strip()


def require_user(request: Request) -> dict:
    token = _extract_token(request)
    if not token:
        raise HttpError("missing token", status=401, code="unauthorized")
    return _verify_user_token(token)


# ===== 密码哈希 =====

def _hash_password(password: str) -> str:
    return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")


def _verify_password(password: str, hashed: str) -> bool:
    try:
        return bcrypt.checkpw(password.encode("utf-8"), hashed.encode("utf-8"))
    except Exception:
        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, alias, mime, size, sha256, uploaded_at "
            "FROM file_meta WHERE uploaded_by = ? AND namespace = ? "
            "ORDER BY uploaded_at DESC",
            (user_id, NS_USER_FILES),
        ).fetchall()
    out = []
    for r in rows:
        d = dict(r)
        # alias 为空时回退到 filename（兼容旧数据）
        if not d.get("alias"):
            d["alias"] = d.get("filename", "") or ""
        out.append(d)
    return out


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, alias, mime, size, "
                    "sha256, uploaded_at, uploaded_by, access_key, refs) "
                    "VALUES(?,?,?,?,?,?,?,?,?,?,?)",
                    (
                        m["key"], NS_USER_FILES, 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_files] restore file_meta {m.get('key')} failed: {e}")
    return inserted


# ===== 注册 =====

@router.post("/register")
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:
        raise HttpError("username and password required", status=400, code="bad_request")
    if not _USERNAME_RE.fullmatch(username):
        raise HttpError(
            "username must be 2-32 chars, only letters/digits/_-. or Chinese",
            status=400, code="bad_request",
        )
    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:
        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,
        "user": {"id": user_id, "username": username},
    })


# ===== 登录 =====

@router.post("/login")
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")

    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(int(user["id"]), user["username"])
    return ok_with_cors({
        "token": token,
        "user": {"id": int(user["id"]), "username": user["username"]},
    })


# ===== 当前用户 =====

@router.get("/me")
async def me(user: dict = Depends(require_user)):
    return ok_with_cors({
        "user": {"id": int(user["sub"]), "username": user["username"]}
    })


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

@router.get("/files")
async def list_my_files(user: dict = Depends(require_user)):
    user_id = str(user["sub"])
    items = _list_files_from_db(user_id)
    # SQLite 为空时尝试从 Hub 恢复（Space 重建场景）
    # 加超时保护，避免 Hub 网络问题导致前端一直卡在"拉取列表"
    if not items and is_hub_enabled():
        try:
            restored = await asyncio.wait_for(
                _restore_files_from_hub_async(user_id), timeout=8.0
            )
            if restored > 0:
                items = _list_files_from_db(user_id)
        except asyncio.TimeoutError:
            logger.warning("[list] restore from hub timeout uid=%s, return empty", user_id)
        except Exception as e:
            logger.warning("[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_file(
    user: dict = Depends(require_user),
    file: UploadFile = File(...),
):
    user_id = str(user["sub"])
    username = user["username"]

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

    key = uuid.uuid4().hex
    alias = (file.filename or "unnamed").strip()  # 原始文件名（可能含中文），用于展示
    if len(alias) > 255:
        alias = alias[:255]
    filename = _make_english_filename(alias, key)  # 随机英文名，用于本地保存
    mime = (file.content_type or "application/octet-stream").strip()
    size = len(content)
    sha256 = hashlib.sha256(content).hexdigest()
    now = int(time.time())

    # 写文件内容到本地缓存 + 异步推 Hub
    try:
        user_files_store.upload_user_file_content(user_id, key, content)
        logger.info(
            "[upload] user=%s uid=%s key=%s alias=%s filename=%s size=%d saved locally, hub_enabled=%s",
            username, user_id, key, alias, 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 alias=%s: %s", username, alias, e)
        raise HttpError("failed to store file", status=500, code="internal_error") from e

    # 写元信息到 SQLite
    meta = {
        "key": key,
        "namespace": NS_USER_FILES,
        "filename": filename,
        "alias": alias,
        "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, alias, mime, size, sha256, "
            "uploaded_at, uploaded_by, access_key, refs) "
            "VALUES(?,?,?,?,?,?,?,?,?,?,?)",
            (key, NS_USER_FILES, filename, alias, 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,
        "alias": alias,
        "mime": mime,
        "size": size,
        "sha256": sha256,
        "uploaded_at": now,
    })


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

@router.get("/files/{key}")
async def download_file(
    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 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_file_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"], NS_USER_FILES, 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 FROM file_meta WHERE key = ?",
                        (key,),
                    ).fetchone()
        except asyncio.TimeoutError:
            logger.warning("[download] restore meta timeout uid=%s key=%s", user_id, key)
        except Exception as e:
            logger.warning("[download] restore meta failed uid=%s key=%s: %s", user_id, key, e)
    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")

    # 读文件内容：本地优先，Hub 回退（同步网络 IO 放到线程池，加超时保护）
    try:
        content = await asyncio.wait_for(
            asyncio.to_thread(user_files_store.download_user_file_content, user_id, key),
            timeout=15.0,
        )
    except asyncio.TimeoutError:
        logger.warning("[download] content fetch timeout uid=%s key=%s", user_id, key)
        raise HttpError("file content fetch timeout", status=504, code="upstream_timeout")
    if content is None:
        raise HttpError("file content missing", status=410, code="gone")

    # Content-Disposition 用 alias（原始名）展示，fallback filename
    disp_name = _resolve_alias(row)
    if not disp_name:
        disp_name = row["filename"] or "unnamed"
    encoded_name = urllib.parse.quote(disp_name)
    # filename= 只允许 ASCII（latin-1 编码），含非 ASCII 字符会导致编码错误；
    # 用英文存储名作为降级，浏览器优先解析 filename*= 取真实别名。
    # 旧文件 filename 可能是中文（alias 机制之前上传的），需降级到 key
    raw_filename = row["filename"] if "filename" in (row.keys() if hasattr(row, "keys") else []) else ""
    ascii_name = raw_filename or "unnamed"
    try:
        ascii_name.encode("latin-1")
    except (UnicodeEncodeError, UnicodeDecodeError):
        # filename 含非 ASCII（旧数据），降级用 key（纯 hex）
        ascii_name = key
    headers = {
        **CORS_HEADERS,
        "Content-Disposition": f"attachment; filename=\"{ascii_name}\"; 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")


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

@router.delete("/files/{key}")
async def delete_file(
    key: str,
    user: dict = Depends(require_user),
):
    user_id = str(user["sub"])
    with get_conn() as conn:
        row = conn.execute(
            "SELECT 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")
        conn.execute("DELETE FROM file_meta WHERE key = ?", (key,))

    # 删文件内容 + 元信息（本地 + Hub）
    try:
        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})


# ===== 注销账号（仅 Web 端使用，需密码确认）=====

@router.delete("/account")
async def delete_account(
    request: Request,
    user: dict = Depends(require_user),
):
    """注销账号：删除该用户所有文件 + 账号本身。

    需要请求体 {password: "..."} 二次确认，避免误操作。
    手表端不会调用此接口（UI 不暴露）。
    """
    body = await read_json_body(request)
    password = str(body.get("password") or "")
    if not password:
        raise HttpError("password required to confirm deletion", status=400, code="bad_request")

    user_id = str(user["sub"])
    username = user["username"]

    # 校验密码
    user_row = _find_user_in_db(username)
    if not user_row or not _verify_password(password, user_row["password_hash"]):
        raise HttpError("password incorrect", status=401, code="unauthorized")

    # 1. 删除该用户所有文件 + 备份的 SQLite meta（不区分 namespace，全部清掉）
    removed_count = 0
    with get_conn() as conn:
        row = conn.execute(
            "SELECT COUNT(*) AS c FROM file_meta WHERE uploaded_by = ?",
            (user_id,),
        ).fetchone()
        removed_count = int(row["c"]) if row else 0
        conn.execute("DELETE FROM file_meta WHERE uploaded_by = ?", (user_id,))

    # 2. 删除 Hub 上的文件 + 备份内容/元信息（兜底清理全部 namespace）
    try:
        user_files_store.delete_all_user_data(user_id)
    except Exception as e:
        print(f"[user_files] delete_account delete_all_user_data failed: {e}")

    # 4. 删除 user_accounts 行
    try:
        with get_conn() as conn:
            conn.execute("DELETE FROM user_accounts WHERE id = ?", (int(user_id),))
    except Exception as e:
        print(f"[user_files] delete_account delete user_accounts row failed: {e}")

    # 5. 删除 Hub 上的账号备份
    try:
        user_files_store.delete_user_account(username)
    except Exception as e:
        print(f"[user_files] delete_account delete_user_account failed: {e}")

    logger.info(
        "[delete_account] user=%s uid=%s deleted, items=%d", username, user_id, removed_count
    )
    return ok_with_cors({"deleted": True, "username": username, "files_removed": removed_count})