307 lines
11 KiB
Python
307 lines
11 KiB
Python
import random
|
||
from typing import Dict, Tuple, List, Optional
|
||
import os
|
||
from pathlib import Path
|
||
|
||
from .config import Config
|
||
from .data_manager import DataManager
|
||
|
||
config = Config()
|
||
data_manager = DataManager()
|
||
|
||
class GachaSystem:
|
||
def __init__(self):
|
||
self.data_manager = data_manager
|
||
|
||
def draw(self, user_id: str) -> Dict:
|
||
"""执行一次抽卡"""
|
||
# 检查抽卡限制
|
||
if not self.data_manager.check_daily_limit(user_id):
|
||
draws_left = self.data_manager.get_draws_left(user_id)
|
||
return {
|
||
"success": False,
|
||
"message": f"您今日的抽卡次数已用完，每日限制{config.DAILY_LIMIT}次，明天再来吧！"
|
||
}
|
||
|
||
# 抽取稀有度（传递用户ID）
|
||
rarity = self._draw_rarity(user_id)
|
||
|
||
# 从该稀有度中抽取式神
|
||
shikigami_data = self.data_manager.shikigami_data.get(rarity, [])
|
||
if not shikigami_data:
|
||
return {
|
||
"success": False,
|
||
"message": f"系统错误：{rarity}稀有度下没有可用式神"
|
||
}
|
||
|
||
# 随机选择式神
|
||
shikigami = random.choice(shikigami_data)
|
||
|
||
# 记录抽卡
|
||
unlocked_achievements = self.data_manager.record_draw(user_id, rarity, shikigami["name"])
|
||
|
||
# 剩余次数
|
||
draws_left = self.data_manager.get_draws_left(user_id)
|
||
|
||
return {
|
||
"success": True,
|
||
"rarity": rarity,
|
||
"name": shikigami["name"],
|
||
"image_url": shikigami["image_url"],
|
||
"draws_left": draws_left,
|
||
"unlocked_achievements": unlocked_achievements
|
||
}
|
||
|
||
def _draw_rarity(self, user_id: str = None) -> str:
|
||
"""按概率抽取稀有度"""
|
||
# 检查是否是特殊概率用户
|
||
if user_id and user_id in config.SPECIAL_PROBABILITY_USERS:
|
||
# 100%概率抽到SSR或SP，随机选择
|
||
return random.choice(["SSR", "SP"])
|
||
|
||
# 普通用户的概率逻辑
|
||
r = random.random() * 100 # 0-100的随机数
|
||
|
||
cumulative = 0
|
||
for rarity, prob in config.RARITY_PROBABILITY.items():
|
||
cumulative += prob
|
||
if r < cumulative:
|
||
return rarity
|
||
|
||
# 默认返回R，理论上不会执行到这里
|
||
return "R"
|
||
|
||
def get_user_stats(self, user_id: str) -> Dict:
|
||
"""获取用户抽卡统计"""
|
||
user_stats = self.data_manager.get_user_stats()
|
||
|
||
if user_id not in user_stats:
|
||
return {
|
||
"success": False,
|
||
"message": "您还没有抽卡记录哦！"
|
||
}
|
||
|
||
stats = user_stats[user_id]
|
||
return {
|
||
"success": True,
|
||
"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"],
|
||
"recent_draws": stats["draw_history"][-5:] if stats["draw_history"] else []
|
||
}
|
||
|
||
def get_probability_text(self) -> str:
|
||
"""获取概率展示文本"""
|
||
probs = config.RARITY_PROBABILITY
|
||
return f"--- 系统概率 ---\nR: {probs['R']}% | SR: {probs['SR']}% | SSR: {probs['SSR']}% | SP: {probs['SP']}%"
|
||
|
||
def get_rank_list(self) -> List[Tuple[str, Dict[str, int]]]:
|
||
"""获取抽卡排行榜数据"""
|
||
user_stats = self.data_manager.get_user_stats()
|
||
|
||
# 过滤有SSR/SP记录的用户
|
||
ranked_users = [
|
||
(user_id, stats)
|
||
for user_id, stats in user_stats.items()
|
||
if stats.get("SSR_count", 0) > 0 or stats.get("SP_count", 0) > 0
|
||
]
|
||
|
||
# 按SSR+SP总数降序排序
|
||
ranked_users.sort(
|
||
key=lambda x: (x[1].get("SSR_count", 0) + x[1].get("SP_count", 0)),
|
||
reverse=True
|
||
)
|
||
|
||
return ranked_users
|
||
|
||
def get_daily_stats(self) -> Dict:
|
||
"""获取今日抽卡统计"""
|
||
daily_draws = self.data_manager.get_daily_draws()
|
||
today = self.data_manager.get_today_date()
|
||
|
||
if not daily_draws or today not in daily_draws:
|
||
return {
|
||
"success": False,
|
||
"message": "今日还没有人抽卡哦！"
|
||
}
|
||
|
||
today_stats = daily_draws[today]
|
||
total_stats = {
|
||
"total_users": len(today_stats),
|
||
"total_draws": 0,
|
||
"R_count": 0,
|
||
"SR_count": 0,
|
||
"SSR_count": 0,
|
||
"SP_count": 0,
|
||
"user_stats": []
|
||
}
|
||
|
||
# 统计每个用户的抽卡情况
|
||
for user_id, draws in today_stats.items():
|
||
user_stats = {
|
||
"user_id": user_id,
|
||
"total_draws": len(draws),
|
||
"R_count": sum(1 for d in draws if d["rarity"] == "R"),
|
||
"SR_count": sum(1 for d in draws if d["rarity"] == "SR"),
|
||
"SSR_count": sum(1 for d in draws if d["rarity"] == "SSR"),
|
||
"SP_count": sum(1 for d in draws if d["rarity"] == "SP")
|
||
}
|
||
|
||
# 更新总统计
|
||
total_stats["total_draws"] += user_stats["total_draws"]
|
||
total_stats["R_count"] += user_stats["R_count"]
|
||
total_stats["SR_count"] += user_stats["SR_count"]
|
||
total_stats["SSR_count"] += user_stats["SSR_count"]
|
||
total_stats["SP_count"] += user_stats["SP_count"]
|
||
|
||
# 只记录抽到SSR或SP的用户
|
||
if user_stats["SSR_count"] > 0 or user_stats["SP_count"] > 0:
|
||
total_stats["user_stats"].append(user_stats)
|
||
|
||
# 按SSR+SP数量排序用户统计
|
||
total_stats["user_stats"].sort(
|
||
key=lambda x: (x["SSR_count"] + x["SP_count"]),
|
||
reverse=True
|
||
)
|
||
|
||
# 构建稀有度统计
|
||
rarity_stats = {
|
||
"R": total_stats["R_count"],
|
||
"SR": total_stats["SR_count"],
|
||
"SSR": total_stats["SSR_count"],
|
||
"SP": total_stats["SP_count"]
|
||
}
|
||
|
||
# 构建排行榜数据
|
||
top_users = []
|
||
for user_stat in total_stats["user_stats"]:
|
||
top_users.append({
|
||
"user_id": user_stat["user_id"],
|
||
"ssr_count": user_stat["SSR_count"] + user_stat["SP_count"]
|
||
})
|
||
|
||
final_stats = {
|
||
"total_users": total_stats["total_users"],
|
||
"total_draws": total_stats["total_draws"],
|
||
"rarity_stats": rarity_stats,
|
||
"top_users": top_users
|
||
}
|
||
|
||
return {
|
||
"success": True,
|
||
"date": today,
|
||
"stats": final_stats
|
||
}
|
||
|
||
def triple_draw(self, user_id: str) -> Dict:
|
||
"""执行三连抽"""
|
||
# 检查是否有足够的抽卡次数
|
||
draws_left = self.data_manager.get_draws_left(user_id)
|
||
if draws_left < 3:
|
||
return {
|
||
"success": False,
|
||
"message": f"抽卡次数不足，您今日还剩{draws_left}次抽卡机会，三连抽需要3次机会"
|
||
}
|
||
|
||
results = []
|
||
all_unlocked_achievements = []
|
||
|
||
# 执行三次抽卡
|
||
for i in range(3):
|
||
# 抽取稀有度（传递用户ID）
|
||
rarity = self._draw_rarity(user_id)
|
||
|
||
# 从该稀有度中抽取式神
|
||
shikigami_data = self.data_manager.shikigami_data.get(rarity, [])
|
||
if not shikigami_data:
|
||
return {
|
||
"success": False,
|
||
"message": f"系统错误：{rarity}稀有度下没有可用式神"
|
||
}
|
||
|
||
# 随机选择式神
|
||
shikigami = random.choice(shikigami_data)
|
||
|
||
# 记录抽卡
|
||
unlocked_achievements = self.data_manager.record_draw(user_id, rarity, shikigami["name"])
|
||
all_unlocked_achievements.extend(unlocked_achievements)
|
||
|
||
results.append({
|
||
"rarity": rarity,
|
||
"name": shikigami["name"],
|
||
"image_url": shikigami["image_url"]
|
||
})
|
||
|
||
# 剩余次数
|
||
draws_left = self.data_manager.get_draws_left(user_id)
|
||
|
||
return {
|
||
"success": True,
|
||
"results": results,
|
||
"draws_left": draws_left,
|
||
"unlocked_achievements": list(set(all_unlocked_achievements)) # 去重
|
||
}
|
||
|
||
def get_user_achievements(self, user_id: str) -> Dict:
|
||
"""获取用户成就信息"""
|
||
achievement_data = self.data_manager.get_user_achievements(user_id)
|
||
|
||
if not achievement_data["unlocked"] and all(v == 0 for v in achievement_data["progress"].values()):
|
||
return {
|
||
"success": False,
|
||
"message": "您还没有任何成就进度哦！快去抽卡吧！"
|
||
}
|
||
|
||
return {
|
||
"success": True,
|
||
"achievements": achievement_data["unlocked"],
|
||
"progress": achievement_data["progress"]
|
||
}
|
||
|
||
def get_daily_detailed_records(self, date: Optional[str] = None) -> Dict:
|
||
"""获取每日详细抽卡记录"""
|
||
if not date:
|
||
date = self.data_manager.get_today_date()
|
||
|
||
daily_draws = self.data_manager.get_daily_draws()
|
||
|
||
if not daily_draws or date not in daily_draws:
|
||
return {
|
||
"success": False,
|
||
"message": f"{date} 没有抽卡记录"
|
||
}
|
||
|
||
records = []
|
||
for user_id, draws in daily_draws[date].items():
|
||
for draw in draws:
|
||
# 检查这次抽卡是否解锁了成就
|
||
unlocked_achievements = []
|
||
draw_time = draw.get("timestamp", "未知时间")
|
||
|
||
# 获取用户成就信息
|
||
achievement_data = self.data_manager.get_user_achievements(user_id)
|
||
if achievement_data["unlocked"]:
|
||
# 检查是否有在抽卡时间之后解锁的成就
|
||
for achievement_id, achievement_info in achievement_data["unlocked"].items():
|
||
if achievement_info["unlocked_date"] == f"{date} {draw_time}":
|
||
unlocked_achievements.append(achievement_id)
|
||
|
||
records.append({
|
||
"user_id": user_id,
|
||
"draw_time": draw_time,
|
||
"shikigami_name": draw["name"],
|
||
"rarity": draw["rarity"],
|
||
"unlocked_achievements": unlocked_achievements
|
||
})
|
||
|
||
# 按时间排序
|
||
records.sort(key=lambda x: x["draw_time"])
|
||
|
||
return {
|
||
"success": True,
|
||
"date": date,
|
||
"records": records,
|
||
"total_count": len(records)
|
||
} |