app/api/admin_data.py

"""Admin 数据管理 API：会话（跨用户视图）、文件/备份元信息、用户账号管理。

- GET    /admin/api/sessions             列出所有会话
- GET    /admin/api/sessions/{id}        获取任意会话详情
- DELETE /admin/api/sessions/{id}        删除任意会话
- GET    /admin/api/sessions/{id}/messages 查看任意会话消息
- GET    /admin/api/files                列出/统计已上传文件/备份（支持 namespace/user_id 过滤）
- GET    /admin/api/backups              列出备份（= /files?namespace=user_backups 语义糖）
- GET    /admin/api/files/{key}/download 管理员下载任意文件/备份内容
- DELETE /admin/api/files/{key}          管理员删除任意文件/备份
- PUT    /admin/api/files/{key}/alias    管理员重命名别名（文件/备份通用）
- GET    /admin/api/users                列出/统计注册用户
- DELETE /admin/api/users/{user_id}      删除用户（级联删除文件+备份+账号）
- POST   /admin/api/sync-from-hub       触发 Hub→SQLite 全量恢复（手动）
"""
from __future__ import annotations

import asyncio
import logging
import time
import urllib.parse

from fastapi import APIRouter, Depends, Query, Request
from fastapi.responses import Response

from ..database import get_conn
from ..errors import HttpError
from ..hf_storage import is_hub_enabled
from ..services import session_store, user_files_store
from ._common import CORS_HEADERS, ok_with_cors, read_json_body
from .admin import _require_admin

router = APIRouter(prefix="/admin/api", tags=["admin-data"])
logger = logging.getLogger(__name__)


# ===== Hub→SQLite 全量恢复（解决 Space 重建后 SQLite 为空的问题）=====

# 同步状态缓存：(last_ts, in_progress)
_sync_state: dict[str, float] = {
    "users_last_ts": 0.0,
    "files_last_ts": 0.0,
    "backups_last_ts": 0.0,
}
# 同步间隔：60 秒内不重复拉取 Hub（避免管理面板频繁刷新打爆 Hub API）
SYNC_TTL_SEC = 60.0


def _should_sync(kind: str, force: bool) -> bool:
    if force:
        return True
    last = _sync_state.get(kind + "_last_ts", 0.0)
    return (time.time() - last) > SYNC_TTL_SEC


def _mark_synced(kind: str) -> None:
    _sync_state[kind + "_last_ts"] = time.time()


async def _sync_all_users_from_hub() -> int:
    """枚举 Hub 上所有 user_accounts 并写回 SQLite。返回已恢复的用户数。"""
    if not is_hub_enabled():
        return 0
    usernames = await asyncio.to_thread(user_files_store.list_all_hub_usernames)
    if not usernames:
        return 0
    inserted = 0
    for uname in usernames:
        try:
            obj = await asyncio.to_thread(user_files_store.restore_user_account, uname)
            if not obj:
                continue
            with get_conn() as conn:
                cur = 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"])),
                )
                if cur.rowcount > 0:
                    inserted += 1
        except Exception as e:
            logger.warning("[admin sync users] restore %s failed: %s", uname, e)
    logger.info("[admin sync users] restored %d/%d users", inserted, len(usernames))
    return inserted


async def _sync_all_user_files_from_hub(kind: str) -> int:
    """枚举 Hub 上所有用户的 file_meta（或 backup_meta）并写回 SQLite。

    kind: "files" → 普通文件；"backups" → 备份。
    """
    if not is_hub_enabled():
        return 0
    file_uids, backup_uids = await asyncio.to_thread(
        user_files_store.list_all_hub_user_ids_with_files
    )
    if kind == "files":
        uids = file_uids
        restore_fn = user_files_store.restore_file_metas
        ns = user_files_store.NS_USER_FILES
    else:
        uids = backup_uids
        restore_fn = user_files_store.restore_backup_metas
        ns = user_files_store.NS_USER_BACKUPS
    if not uids:
        return 0
    inserted = 0
    for uid in uids:
        try:
            metas = await asyncio.wait_for(restore_fn(uid), timeout=15.0)
            if not metas:
                continue
            with get_conn() as conn:
                for m in metas:
                    try:
                        cur = 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, 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(uid),
                                m.get("access_key", ""), "",
                            ),
                        )
                        if cur.rowcount > 0:
                            inserted += 1
                    except Exception as e:
                        logger.debug("[admin sync %s] meta %s failed: %s", kind, m.get("key"), e)
        except asyncio.TimeoutError:
            logger.warning("[admin sync %s] restore timeout uid=%s", kind, uid)
        except Exception as e:
            logger.warning("[admin sync %s] restore uid=%s failed: %s", kind, uid, e)
    logger.info("[admin sync %s] restored %d metas from %d users", kind, inserted, len(uids))
    return inserted


@router.post("/sync-from-hub")
async def sync_from_hub(
    request: Request,
    _admin: str = Depends(_require_admin),
):
    """手动触发 Hub→SQLite 全量同步（用户 + 文件 + 备份）。

    body 可选：{kind: "users"|"files"|"backups"|"all"}，默认 "all"。
    返回每类已恢复的条目数。
    """
    body = await read_json_body(request) if request.headers.get("content-type", "").startswith("application/json") else {}
    kind = (body.get("kind") if isinstance(body, dict) else None) or "all"
    result = {"kind": kind, "users": 0, "files": 0, "backups": 0}
    if kind in ("users", "all"):
        result["users"] = await _sync_all_users_from_hub()
        _mark_synced("users")
    if kind in ("files", "all"):
        result["files"] = await _sync_all_user_files_from_hub("files")
        _mark_synced("files")
    if kind in ("backups", "all"):
        result["backups"] = await _sync_all_user_files_from_hub("backups")
        _mark_synced("backups")
    return ok_with_cors(result)


# ===== 命名空间 → 内容/元信息 store 映射 =====

def _is_backup(namespace: str) -> bool:
    return namespace == user_files_store.NS_USER_BACKUPS


def _store_download(namespace: str, user_id: str, key: str):
    """按命名空间选择对应内容 store 下载。"""
    if _is_backup(namespace):
        return user_files_store.download_user_backup_content(user_id, key)
    return user_files_store.download_user_file_content(user_id, key)


def _store_delete(namespace: str, user_id: str, key: str) -> None:
    """按命名空间删除内容 + 元信息（本地 + Hub）。"""
    if _is_backup(namespace):
        try:
            user_files_store.delete_user_backup_content(user_id, key)
        except Exception as e:
            logger.warning("[admin] delete backup content %s failed: %s", key, e)
        try:
            user_files_store.delete_backup_meta(user_id, key)
        except Exception as e:
            logger.warning("[admin] delete backup meta %s failed: %s", key, e)
    else:
        try:
            user_files_store.delete_user_file_content(user_id, key)
        except Exception as e:
            logger.warning("[admin] delete file content %s failed: %s", key, e)
        try:
            user_files_store.delete_file_meta(user_id, key)
        except Exception as e:
            logger.warning("[admin] delete file meta %s failed: %s", key, e)


def _store_backup_meta(namespace: str, user_id: str, key: str, meta: dict) -> None:
    if _is_backup(namespace):
        user_files_store.backup_backup_meta(user_id, key, meta)
    else:
        user_files_store.backup_file_meta(user_id, key, meta)


async def _store_push_meta(namespace: str, user_id: str, key: str) -> None:
    if _is_backup(namespace):
        await user_files_store.push_backup_meta(user_id, key)
    else:
        await user_files_store.push_file_meta(user_id, key)


def _meta_row_to_dict(row) -> dict:
    d = dict(row)
    if not d.get("alias"):
        d["alias"] = d.get("filename", "") or ""
    # 标注 kind 便于前端展示
    d["kind"] = "backup" if _is_backup(d.get("namespace", "")) else "file"
    return d


# ===== 会话 =====

@router.get("/sessions")
async def list_all_sessions(
    _admin: str = Depends(_require_admin),
    limit: int = Query(default=100, ge=1, le=500),
    offset: int = Query(default=0, ge=0),
):
    items = session_store.list_sessions(limit=limit, offset=offset)
    return ok_with_cors({"items": items, "count": len(items)})


@router.get("/sessions/{session_id}")
async def get_any_session(
    session_id: str,
    _admin: str = Depends(_require_admin),
):
    sess = session_store.get_session(session_id)
    if not sess:
        raise HttpError(f"session not found: {session_id}", status=404, code="not_found")
    return ok_with_cors({"session": sess})


@router.delete("/sessions/{session_id}")
async def delete_any_session(
    session_id: str,
    _admin: str = Depends(_require_admin),
):
    ok = session_store.delete_session(session_id)
    if not ok:
        raise HttpError(f"session not found: {session_id}", status=404, code="not_found")
    return ok_with_cors({"deleted": True, "id": session_id})


@router.get("/sessions/{session_id}/messages")
async def list_any_session_messages(
    session_id: str,
    _admin: str = Depends(_require_admin),
):
    if not session_store.get_session(session_id):
        raise HttpError(f"session not found: {session_id}", status=404, code="not_found")
    msgs = session_store.list_messages(session_id) or []
    return ok_with_cors({"messages": msgs, "count": len(msgs)})


# ===== 文件 / 备份 列表 =====

def _list_file_meta(
    *, limit: int, offset: int,
    namespace: str | None, user_id: str | None, q: str | None,
) -> tuple[int, list[dict]]:
    """统一列表查询：支持按 namespace / user_id / 别名模糊搜索过滤。"""
    where = []
    args: list = []
    if namespace:
        where.append("namespace = ?")
        args.append(namespace)
    if user_id:
        where.append("uploaded_by = ?")
        args.append(str(user_id))
    if q:
        where.append("(alias LIKE ? OR filename LIKE ? OR access_key LIKE ?)")
        like = f"%{q}%"
        args.extend([like, like, like])
    where_sql = (" WHERE " + " AND ".join(where)) if where else ""

    with get_conn() as conn:
        total = conn.execute(
            "SELECT COUNT(*) AS c FROM file_meta" + where_sql, args
        ).fetchone()["c"]
        items = []
        if limit > 0:
            rows = conn.execute(
                "SELECT key, namespace, filename, alias, mime, size, sha256, "
                "uploaded_at, uploaded_by, access_key "
                "FROM file_meta" + where_sql +
                " ORDER BY uploaded_at DESC LIMIT ? OFFSET ?",
                args + [limit, offset],
            ).fetchall()
            items = [_meta_row_to_dict(r) for r in rows]
    return total, items


@router.get("/files")
async def list_files(
    _admin: str = Depends(_require_admin),
    limit: int = Query(default=100, ge=1, le=500),
    offset: int = Query(default=0, ge=0),
    namespace: str | None = Query(default=None),
    user_id: str | None = Query(default=None, alias="user_id"),
    q: str | None = Query(default=None),
    refresh: int = Query(default=0, ge=0, le=1),
):
    """列出已上传文件/备份元信息。

    - namespace: user_files(普通文件) / user_backups(备份)，缺省=全部
    - user_id: 仅看某用户
    - q: 别名/文件名/账号模糊搜索
    - limit=1 时仅用于概览计数
    - refresh=1 时强制从 Hub 全量同步（默认按 TTL 自动同步）

    Space 重建后 SQLite 为空时，会自动触发 Hub→SQLite 全量恢复，
    避免出现"刷新都是空的"问题。
    """
    # 概览计数（limit=1）时不阻塞触发同步，但仍尝试一次同步保证 count 准确
    if _should_sync("files", bool(refresh)):
        try:
            await _sync_all_user_files_from_hub("files")
            _mark_synced("files")
        except Exception as e:
            logger.warning("[list_files] sync from hub failed: %s", e)
    total, items = _list_file_meta(
        limit=limit, offset=offset, namespace=namespace, user_id=user_id, q=q,
    )
    return ok_with_cors({"items": items, "count": total})


@router.get("/backups")
async def list_backups(
    _admin: str = Depends(_require_admin),
    limit: int = Query(default=100, ge=1, le=500),
    offset: int = Query(default=0, ge=0),
    user_id: str | None = Query(default=None, alias="user_id"),
    q: str | None = Query(default=None),
    refresh: int = Query(default=0, ge=0, le=1),
):
    """列出备份（语义糖：强制 namespace=user_backups）。

    Space 重建后 SQLite 为空时，会自动触发 Hub→SQLite 全量恢复。
    """
    if _should_sync("backups", bool(refresh)):
        try:
            await _sync_all_user_files_from_hub("backups")
            _mark_synced("backups")
        except Exception as e:
            logger.warning("[list_backups] sync from hub failed: %s", e)
    total, items = _list_file_meta(
        limit=limit, offset=offset,
        namespace=user_files_store.NS_USER_BACKUPS, user_id=user_id, q=q,
    )
    return ok_with_cors({"items": items, "count": total})


# ===== 文件 / 备份 管理操作 =====

def _get_meta_row(key: str):
    with get_conn() as conn:
        return conn.execute(
            "SELECT key, namespace, filename, alias, mime, size, sha256, "
            "uploaded_at, uploaded_by, access_key FROM file_meta WHERE key = ?",
            (key,),
        ).fetchone()


@router.get("/files/{key}/download")
async def download_any_file(
    key: str,
    _admin: str = Depends(_require_admin),
):
    """管理员下载任意文件/备份内容（按 namespace 选择内容 store）。"""
    row = _get_meta_row(key)
    if not row:
        raise HttpError(f"item not found: {key}", status=404, code="not_found")
    namespace = row["namespace"]
    owner = str(row["uploaded_by"])

    try:
        content = await asyncio.wait_for(
            asyncio.to_thread(_store_download, namespace, owner, key),
            timeout=30.0,
        )
    except asyncio.TimeoutError:
        raise HttpError("content fetch timeout", status=504, code="upstream_timeout")
    if content is None:
        # 本地未命中且 Hub 未启用 / 内容丢失
        raise HttpError("content missing", status=410, code="gone")

    alias = row["alias"] or row["filename"] or "unnamed"
    encoded = urllib.parse.quote(alias)
    raw_filename = row["filename"] or "unnamed"
    ascii_name = raw_filename
    try:
        ascii_name.encode("latin-1")
    except (UnicodeEncodeError, UnicodeDecodeError):
        ascii_name = key
    headers = {
        **CORS_HEADERS,
        "Content-Disposition": f"attachment; filename=\"{ascii_name}\"; filename*=UTF-8''{encoded}",
        "Content-Type": row["mime"] or "application/octet-stream",
        "Content-Length": str(len(content)),
        "X-XTC-Kind": "backup" if _is_backup(namespace) else "file",
    }
    return Response(content=content, status_code=200, headers=headers,
                    media_type=row["mime"] or "application/octet-stream")


@router.put("/files/{key}/alias")
async def rename_any_alias(
    key: str,
    request: Request,
    _admin: str = Depends(_require_admin),
):
    """管理员重命名文件/备份别名（body {alias}）。"""
    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]

    row = _get_meta_row(key)
    if not row:
        raise HttpError(f"item not found: {key}", status=404, code="not_found")
    namespace = row["namespace"]
    owner = str(row["uploaded_by"])

    with get_conn() as conn:
        cur = conn.execute(
            "UPDATE file_meta SET alias=? WHERE key=?",
            (alias, key),
        )
    if cur.rowcount == 0:
        raise HttpError(f"item not found: {key}", status=404, code="not_found")

    # 同步 Hub 元信息
    meta = {
        "key": key,
        "namespace": namespace,
        "filename": row["filename"],
        "alias": alias,
        "mime": row["mime"],
        "size": row["size"],
        "sha256": row["sha256"],
        "uploaded_at": row["uploaded_at"],
        "uploaded_by": owner,
        "access_key": row["access_key"],
    }
    try:
        _store_backup_meta(namespace, owner, key, meta)
        asyncio.create_task(_store_push_meta(namespace, owner, key))
    except Exception as e:
        logger.warning("[admin rename] backup meta failed key=%s: %s", key, e)

    return ok_with_cors({"key": key, "alias": alias, "kind": "backup" if _is_backup(namespace) else "file"})


@router.delete("/files/{key}")
async def delete_any_file(
    key: str,
    _admin: str = Depends(_require_admin),
):
    """管理员删除任意文件/备份（级联清理 SQLite meta + 内容 + Hub）。"""
    row = _get_meta_row(key)
    if not row:
        raise HttpError(f"item not found: {key}", status=404, code="not_found")
    namespace = row["namespace"]
    owner = str(row["uploaded_by"])
    kind = "backup" if _is_backup(namespace) else "file"

    with get_conn() as conn:
        conn.execute("DELETE FROM file_meta WHERE key = ?", (key,))
    _store_delete(namespace, owner, key)

    logger.info("[admin delete] key=%s kind=%s owner=%s", key, kind, owner)
    return ok_with_cors({"deleted": True, "key": key, "kind": kind})


# ===== 用户账号管理 =====

@router.get("/users")
async def list_users(
    _admin: str = Depends(_require_admin),
    limit: int = Query(default=100, ge=1, le=500),
    offset: int = Query(default=0, ge=0),
    q: str | None = Query(default=None),
    refresh: int = Query(default=0, ge=0, le=1),
):
    """列出注册用户。limit=1 时仅用于概览计数（不返回 password_hash）。

    每个用户分别统计 file_count(普通文件) 与 backup_count(备份)。

    Space 重建后 SQLite 为空时，会自动触发 Hub→SQLite 全量恢复，
    避免出现"刷新都是空的"问题（TTL 60 秒，避免频繁打 Hub）。
    """
    if _should_sync("users", bool(refresh)):
        try:
            await _sync_all_users_from_hub()
            _mark_synced("users")
        except Exception as e:
            logger.warning("[list_users] sync from hub failed: %s", e)
    where = ""
    args: list = []
    if q:
        where = " WHERE username LIKE ?"
        args.append(f"%{q}%")
    with get_conn() as conn:
        total = conn.execute(
            "SELECT COUNT(*) AS c FROM user_accounts" + where, args
        ).fetchone()["c"]
        items = []
        if limit > 0:
            rows = conn.execute(
                "SELECT id, username, created_at FROM user_accounts" + where +
                " ORDER BY created_at DESC LIMIT ? OFFSET ?",
                args + [limit, offset],
            ).fetchall()
            for r in rows:
                d = dict(r)
                uid = str(d["id"])
                fc = conn.execute(
                    "SELECT COUNT(*) AS c FROM file_meta WHERE uploaded_by = ? AND namespace = ?",
                    (uid, user_files_store.NS_USER_FILES),
                ).fetchone()["c"]
                bc = conn.execute(
                    "SELECT COUNT(*) AS c FROM file_meta WHERE uploaded_by = ? AND namespace = ?",
                    (uid, user_files_store.NS_USER_BACKUPS),
                ).fetchone()["c"]
                d["file_count"] = fc
                d["backup_count"] = bc
                items.append(d)
    return ok_with_cors({"items": items, "count": total})


@router.delete("/users/{user_id}")
async def delete_user(
    user_id: int,
    _admin: str = Depends(_require_admin),
):
    """管理员删除用户：级联删除该用户所有文件 + 备份 + 账号。

    会同时清理 SQLite file_meta（全 namespace）、Hub 上的文件/备份内容与元信息、
    user_accounts 行、Hub 上的账号备份。
    """
    with get_conn() as conn:
        row = conn.execute(
            "SELECT username FROM user_accounts WHERE id = ?",
            (user_id,),
        ).fetchone()
    if not row:
        raise HttpError(f"user not found: {user_id}", status=404, code="not_found")
    username = row["username"]
    uid_str = str(user_id)

    # 1. 统计并删除该用户所有 file_meta（全 namespace）
    with get_conn() as conn:
        cnt_row = conn.execute(
            "SELECT COUNT(*) AS c FROM file_meta WHERE uploaded_by = ?", (uid_str,)
        ).fetchone()
        removed = int(cnt_row["c"]) if cnt_row else 0
        conn.execute("DELETE FROM file_meta WHERE uploaded_by = ?", (uid_str,))

    # 2. 删除 Hub 上的文件 + 备份内容/元信息（兜底清理全部 namespace）
    try:
        user_files_store.delete_all_user_data(uid_str)
    except Exception as e:
        logger.warning("[admin delete_user] delete_all_user_data failed: %s", e)

    # 3. 删除 user_accounts 行
    with get_conn() as conn:
        conn.execute("DELETE FROM user_accounts WHERE id = ?", (user_id,))

    # 4. 删除 Hub 上的账号备份
    try:
        user_files_store.delete_user_account(username)
    except Exception as e:
        logger.warning("[admin delete_user] delete_user_account failed: %s", e)

    logger.info(
        "[admin delete_user] uid=%s username=%s deleted, items=%d",
        user_id, username, removed,
    )
    return ok_with_cors({
        "deleted": True,
        "id": user_id,
        "username": username,
        "files_removed": removed,
    })