首次提交

danding_bot/plugins/onmyoji_gacha/web_api.py

@@ -0,0 +1,202 @@
+"""
+onmyoji_gacha 插件的 Web API 接口
+使用 NoneBot 内置的 FastAPI 适配器提供管理员后台接口
+"""
+import os
+from typing import Dict, List, Any, Optional
+from fastapi import APIRouter, Depends, HTTPException, Header, Request
+from fastapi.responses import HTMLResponse, JSONResponse
+from fastapi.staticfiles import StaticFiles
+from fastapi.templating import Jinja2Templates
+from pydantic import BaseModel
+
+from nonebot import get_driver
+from .config import Config
+from .gacha import GachaSystem
+
+# 创建配置实例
+config = Config()
+gacha_system = GachaSystem()
+
+# 创建 FastAPI 路由
+router = APIRouter(prefix="/onmyoji_gacha", tags=["onmyoji_gacha"])
+
+# 设置模板目录
+templates = Jinja2Templates(directory="danding_bot/plugins/onmyoji_gacha/templates")
+
+# 依赖：验证管理员权限
+async def verify_admin_token(authorization: Optional[str] = Header(None)):
+    """验证管理员权限"""
+    print(f"🔐 验证管理员令牌: {authorization}")
+    
+    if not authorization:
+        print("❌ 缺少认证令牌")
+        raise HTTPException(status_code=401, detail="缺少认证令牌")
+    
+    token = authorization.replace("Bearer ", "")
+    print(f"🔑 提取的令牌: {token}")
+    print(f"🎯 期望的令牌: {config.WEB_ADMIN_TOKEN}")
+    
+    if token != config.WEB_ADMIN_TOKEN:
+        print("❌ 令牌验证失败")
+        raise HTTPException(status_code=403, detail="无效的认证令牌")
+    
+    print("✅ 令牌验证成功")
+    return True
+
+# API 响应模型
+class DailyStatsResponse(BaseModel):
+    success: bool
+    date: str
+    stats: Dict[str, Any]
+
+class UserStatsResponse(BaseModel):
+    success: bool
+    user_id: str
+    total_draws: int
+    R_count: int
+    SR_count: int
+    SSR_count: int
+    SP_count: int
+    recent_draws: List[Dict[str, str]]
+
+class RankListResponse(BaseModel):
+    success: bool
+    data: List[Dict[str, Any]]
+
+class AchievementResponse(BaseModel):
+    success: bool
+    user_id: str
+    achievements: Dict[str, Any]
+    progress: Dict[str, Any]
+
+class DailyDetailedRecordsResponse(BaseModel):
+    success: bool
+    date: str
+    records: List[Dict[str, Any]]
+    total_count: int
+
+# 管理后台页面
+@router.get("/admin", response_class=HTMLResponse)
+async def admin_page(request: Request):
+    """管理后台页面"""
+    return templates.TemplateResponse("admin.html", {"request": request})
+
+# API 端点
+@router.get("/api/stats/daily", response_model=DailyStatsResponse, dependencies=[Depends(verify_admin_token)])
+async def get_daily_stats():
+    """获取今日抽卡统计"""
+    result = gacha_system.get_daily_stats()
+    if not result["success"]:
+        return result
+    
+    return {
+        "success": True,
+        "date": result["date"],
+        "stats": result["stats"]
+    }
+
+@router.get("/api/stats/user/{user_id}", response_model=UserStatsResponse, dependencies=[Depends(verify_admin_token)])
+async def get_user_stats(user_id: str):
+    """获取用户抽卡统计"""
+    result = gacha_system.get_user_stats(user_id)
+    if not result["success"]:
+        return {
+            "success": False,
+            "user_id": user_id,
+            "total_draws": 0,
+            "R_count": 0,
+            "SR_count": 0,
+            "SSR_count": 0,
+            "SP_count": 0,
+            "recent_draws": []
+        }
+    
+    return {
+        "success": True,
+        "user_id": user_id,
+        "total_draws": result["total_draws"],
+        "R_count": result["R_count"],
+        "SR_count": result["SR_count"],
+        "SSR_count": result["SSR_count"],
+        "SP_count": result["SP_count"],
+        "recent_draws": result["recent_draws"]
+    }
+
+@router.get("/api/stats/rank", response_model=RankListResponse, dependencies=[Depends(verify_admin_token)])
+async def get_rank_list():
+    """获取抽卡排行榜"""
+    rank_data = gacha_system.get_rank_list()
+    
+    # 转换数据格式
+    formatted_data = []
+    for user_id, stats in rank_data:
+        formatted_data.append({
+            "user_id": user_id,
+            "total_draws": stats["total_draws"],
+            "R_count": stats["R_count"],
+            "SR_count": stats["SR_count"],
+            "SSR_count": stats["SSR_count"],
+            "SP_count": stats["SP_count"],
+            "ssr_sp_total": stats["SSR_count"] + stats["SP_count"]
+        })
+    
+    return {
+        "success": True,
+        "data": formatted_data
+    }
+
+@router.get("/api/achievements/{user_id}", response_model=AchievementResponse, dependencies=[Depends(verify_admin_token)])
+async def get_user_achievements(user_id: str):
+    """获取用户成就信息"""
+    result = gacha_system.get_user_achievements(user_id)
+    if not result["success"]:
+        return {
+            "success": False,
+            "user_id": user_id,
+            "achievements": {},
+            "progress": {}
+        }
+    
+    return {
+        "success": True,
+        "user_id": user_id,
+        "achievements": result["achievements"],
+        "progress": result["progress"]
+    }
+
+@router.get("/api/records/daily", response_model=DailyDetailedRecordsResponse, dependencies=[Depends(verify_admin_token)])
+async def get_daily_detailed_records(date: Optional[str] = None):
+    """获取每日详细抽卡记录"""
+    result = gacha_system.get_daily_detailed_records(date)
+    if not result["success"]:
+        return {
+            "success": False,
+            "date": date or gacha_system.data_manager.get_today_date(),
+            "records": [],
+            "total_count": 0
+        }
+    
+    return {
+        "success": True,
+        "date": result["date"],
+        "records": result["records"],
+        "total_count": result["total_count"]
+    }
+
+# 注册路由到 NoneBot 的 FastAPI 应用
+# 将在插件加载时由 __init__.py 调用
+def register_web_routes():
+    """注册 Web 路由到 NoneBot 的 FastAPI 应用"""
+    try:
+        from nonebot import get_driver
+        driver = get_driver()
+        # 获取 FastAPI 应用实例
+        app = driver.server_app
+        # 注册路由
+        app.include_router(router)
+        print("✅ onmyoji_gacha Web API 路由注册成功")
+        return True
+    except Exception as e:
+        print(f"❌ 注册 Web 路由时出错: {e}")
+        return False