"""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, })