DanDingNoneBot/danding_bot/plugins/onmyoji_gacha/web_api.py

"""
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