ngfchl
130 lines
4.7 KiB
Python
130 lines
4.7 KiB
Python
"""小游戏服务"""
|
|
import random
|
|
import logging
|
|
from datetime import datetime
|
|
|
|
from db.models import GameScore
|
|
|
|
logger = logging.getLogger("spam_guard")
|
|
|
|
TRIVIA_QUESTIONS = [
|
|
{"q": "收割机项目的 GitHub 地址是？", "a": "github.com/ngfchl/go-harvest", "opts": ["github.com/ngfchl/go-harvest", "github.com/ngfchl/harvest", "github.com/harvest/go-harvest"]},
|
|
{"q": "以下哪个不是 PT 站的常见等级？", "a": "Diamond", "opts": ["User", "Power User", "Diamond", "Elite"]},
|
|
{"q": "Docker 默认的日志驱动是？", "a": "json-file", "opts": ["json-file", "syslog", "journald", "fluentd"]},
|
|
]
|
|
|
|
SHOP_ITEMS = {
|
|
"1": {"name": "补签卡", "cost": 50, "desc": "预留权益：后续可用于补签"},
|
|
"2": {"name": "群内称号券", "cost": 80, "desc": "预留权益：联系管理员兑换称号"},
|
|
"3": {"name": "好运盲盒", "cost": 30, "desc": "兑换后记录一次好运盲盒"},
|
|
}
|
|
|
|
|
|
async def get_or_create_score(user_id: int, username: str, game_type: str) -> GameScore:
|
|
score, _ = await GameScore.get_or_create(
|
|
user_id=user_id, game_type=game_type,
|
|
defaults={"username": username},
|
|
)
|
|
if username and score.username != username:
|
|
score.username = username
|
|
await score.save()
|
|
return score
|
|
|
|
|
|
async def get_trivia_question() -> dict:
|
|
return random.choice(TRIVIA_QUESTIONS)
|
|
|
|
|
|
async def check_trivia_answer(user_id: int, username: str, answer: str, correct_answer: str) -> dict:
|
|
score = await get_or_create_score(user_id, username, "trivia")
|
|
score.total += 1
|
|
is_correct = answer.strip().lower() in correct_answer.strip().lower()
|
|
if is_correct:
|
|
score.score += 10
|
|
score.wins += 1
|
|
score.streak += 1
|
|
if score.streak > 1:
|
|
score.score += 5
|
|
else:
|
|
score.streak = 0
|
|
score.last_played = datetime.now()
|
|
await score.save()
|
|
return {"correct": is_correct, "score": score.score, "streak": score.streak}
|
|
|
|
|
|
async def get_leaderboard(game_type: str = "trivia", limit: int = 10) -> list:
|
|
return await GameScore.filter(game_type=game_type).order_by("-score").limit(limit).all()
|
|
|
|
|
|
async def daily_checkin(user_id: int, username: str) -> dict:
|
|
score = await get_or_create_score(user_id, username, "checkin")
|
|
today = datetime.now().strftime("%Y-%m-%d")
|
|
if score.last_played and score.last_played.strftime("%Y-%m-%d") == today:
|
|
return {"ok": False, "reason": "already_checked", "score": score.score, "streak": score.streak}
|
|
score.total += 1
|
|
base = random.randint(5, 12)
|
|
streak_bonus = min(score.streak + 1, 7)
|
|
bonus = base + streak_bonus
|
|
score.score += bonus
|
|
score.streak += 1
|
|
score.wins += 1
|
|
score.last_played = datetime.now()
|
|
await score.save()
|
|
return {"ok": True, "score": score.score, "streak": score.streak, "bonus": bonus, "base": base, "streak_bonus": streak_bonus}
|
|
|
|
|
|
async def get_scores_summary(user_id: int) -> dict:
|
|
scores = await GameScore.filter(user_id=user_id).all()
|
|
total = sum((s.score or 0) for s in scores)
|
|
by_type = {s.game_type: s for s in scores}
|
|
return {"scores": scores, "total": total, "by_type": by_type}
|
|
|
|
|
|
async def deduct_points(user_id: int, cost: int) -> bool:
|
|
summary = await get_scores_summary(user_id)
|
|
if summary["total"] < cost:
|
|
return False
|
|
remaining = cost
|
|
for score in sorted(summary["scores"], key=lambda x: x.score or 0, reverse=True):
|
|
if remaining <= 0:
|
|
break
|
|
take = min(score.score, remaining)
|
|
score.score -= take
|
|
remaining -= take
|
|
await score.save()
|
|
return True
|
|
|
|
|
|
async def lottery_draw(user_id: int, username: str) -> dict:
|
|
cost = 10
|
|
if not await deduct_points(user_id, cost):
|
|
return {"ok": False, "reason": "积分不足", "cost": cost}
|
|
roll = random.random()
|
|
reward = 0
|
|
prize = "谢谢参与"
|
|
if roll < 0.08:
|
|
reward, prize = 60, "一等奖 +60"
|
|
elif roll < 0.25:
|
|
reward, prize = 25, "二等奖 +25"
|
|
elif roll < 0.55:
|
|
reward, prize = 12, "三等奖 +12"
|
|
score = await get_or_create_score(user_id, username, "lottery")
|
|
score.total += 1
|
|
if reward:
|
|
score.wins += 1
|
|
score.score += reward
|
|
score.last_played = datetime.now()
|
|
await score.save()
|
|
summary = await get_scores_summary(user_id)
|
|
return {"ok": True, "cost": cost, "reward": reward, "prize": prize, "total": summary["total"]}
|
|
|
|
|
|
async def buy_shop_item(user_id: int, item_id: str) -> dict:
|
|
item = SHOP_ITEMS.get(item_id)
|
|
if not item:
|
|
return {"ok": False, "reason": "商品不存在"}
|
|
if not await deduct_points(user_id, item["cost"]):
|
|
return {"ok": False, "reason": "积分不足", "item": item}
|
|
summary = await get_scores_summary(user_id)
|
|
return {"ok": True, "item": item, "total": summary["total"]}
|