From 57656a3cc3d8a1251ca59272d8f0ebab91bc33e9 Mon Sep 17 00:00:00 2001 From: ngfchl Date: Tue, 7 Jul 2026 10:32:57 +0800 Subject: [PATCH] feat: add points lottery shop and panel --- handlers/bind.py | 17 +++++--- handlers/fun.py | 94 ++++++++++++++++++++++++++++++++++++++-- main.py | 3 ++ services/game_service.py | 80 +++++++++++++++++++++++++++++++--- 4 files changed, 177 insertions(+), 17 deletions(-) diff --git a/handlers/bind.py b/handlers/bind.py index e87730f..f44225e 100644 --- a/handlers/bind.py +++ b/handlers/bind.py @@ -10,6 +10,7 @@ from db.models import Action, GameScore from services.bind_service import bind_with_email_token, unbind, get_binding from services.message_utils import safe_reply from services.time_utils import format_dt +from services.game_service import get_scores_summary logger = logging.getLogger("spam_guard") @@ -112,16 +113,20 @@ async def cmd_me(update: Update, context: ContextTypes.DEFAULT_TYPE): """查看当前 TG 用户绑定的授权信息和积分。""" user = update.effective_user binding = await get_binding(user.id) - scores = await GameScore.filter(user_id=user.id).all() + summary = await get_scores_summary(user.id) + scores = summary["scores"] if scores: score_lines = [] - total_score = 0 for item in scores: - total_score += item.score or 0 - score_lines.append(f"🎮 {item.game_type}: {item.score} 分（{item.wins}/{item.total}）") - score_text = f"📊 总积分：{total_score} 分\n" + "\n".join(score_lines) + last = format_dt(item.last_played) if item.last_played else "未参与" + score_lines.append(f"🎮 {item.game_type}: {item.score} 分｜次数 {item.total}｜连续 {item.streak}｜最近 {last}") + checkin = summary["by_type"].get("checkin") + trivia = summary["by_type"].get("trivia") + checkin_text = "✅ 今日签到：已签到" if checkin and checkin.last_played and checkin.last_played.strftime("%Y-%m-%d") == __import__("datetime").datetime.now().strftime("%Y-%m-%d") else "✅ 今日签到：未签到" + trivia_text = f"🎮 答题积分：{trivia.score} 分｜正确 {trivia.wins}/{trivia.total}｜连击 {trivia.streak}" if trivia else "🎮 答题积分：暂无" + score_text = f"📊 总积分：{summary['total']} 分\n{checkin_text}\n{trivia_text}\n" + "\n".join(score_lines) else: - score_text = "📊 积分：暂无记录\n💡 发送 /checkin 签到，或 /trivia 答题获取积分" + score_text = "📊 积分：暂无记录\n✅ 今日签到：未签到\n🎮 答题积分：暂无\n💡 发送 /checkin 签到，或 /trivia 答题获取积分" if not binding: text = ( diff --git a/handlers/fun.py b/handlers/fun.py index d90a679..0240b66 100644 --- a/handlers/fun.py +++ b/handlers/fun.py @@ -2,13 +2,14 @@ import asyncio import logging -from telegram import Update -from telegram.ext import ApplicationHandlerStop, ContextTypes, CommandHandler, MessageHandler, filters +from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update +from telegram.ext import ApplicationHandlerStop, CallbackQueryHandler, ContextTypes, CommandHandler, MessageHandler, filters from db.models import Action from services.game_service import ( get_trivia_question, check_trivia_answer, - get_leaderboard, daily_checkin, + get_leaderboard, daily_checkin, get_scores_summary, + lottery_draw, SHOP_ITEMS, buy_shop_item, ) logger = logging.getLogger("spam_guard") @@ -51,7 +52,7 @@ async def cmd_checkin(update: Update, context: ContextTypes.DEFAULT_TYPE): f"✅ 签到成功！\n" f"📊 积分: {result['score']}\n" f"🔥 连续签到: {result['streak']} 天\n" - f"🎁 签到奖励: +{result['bonus']} 分" + f"🎁 签到奖励: +{result['bonus']} 分（基础 {result.get('base', 0)} + 连签 {result.get('streak_bonus', 0)}）" ) else: reply = await update.message.reply_text("❌ 你今天已经签到过了") @@ -105,6 +106,86 @@ async def cmd_rank(update: Update, context: ContextTypes.DEFAULT_TYPE): asyncio.create_task(auto_delete(reply, seconds=60)) +async def cmd_lottery(update: Update, context: ContextTypes.DEFAULT_TYPE): + user = update.effective_user + result = await lottery_draw(user.id, user.username or user.first_name or str(user.id)) + if not result["ok"]: + reply = await update.message.reply_text(f"🎁 抽奖需要 {result.get('cost', 10)} 积分，当前积分不足。\n发送 /checkin 或 /trivia 先赚积分吧。") + else: + reply = await update.message.reply_text( + "🎁 抽奖结果\n" + f"🎯 {result['prize']}\n" + f"💸 消耗：{result['cost']} 分\n" + f"🎉 奖励：+{result['reward']} 分\n" + f"📊 当前总积分：{result['total']} 分" + ) + asyncio.create_task(auto_delete(update.message, seconds=60)) + asyncio.create_task(auto_delete(reply, seconds=60)) + + +async def cmd_shop(update: Update, context: ContextTypes.DEFAULT_TYPE): + user = update.effective_user + summary = await get_scores_summary(user.id) + text = "🛒 积分商城\n━━━━━━━━━━━━\n" + text += f"📊 当前总积分：{summary['total']} 分\n\n" + for item_id, item in SHOP_ITEMS.items(): + text += f"{item_id}. {item['name']} — {item['cost']} 分\n {item['desc']}\n" + text += "\n购买：/buy 商品编号，例如 /buy 1" + reply = await update.message.reply_text(text) + asyncio.create_task(auto_delete(update.message, seconds=60)) + asyncio.create_task(auto_delete(reply, seconds=60)) + + +async def cmd_buy(update: Update, context: ContextTypes.DEFAULT_TYPE): + if not context.args: + reply = await update.message.reply_text("用法：/buy 商品编号，例如 /buy 1") + else: + result = await buy_shop_item(update.effective_user.id, context.args[0]) + if result["ok"]: + item = result["item"] + await Action.create(action="SHOP_BUY", user_id=update.effective_user.id, username=update.effective_user.username, details={"item": item}) + reply = await update.message.reply_text(f"✅ 兑换成功：{item['name']}\n📊 剩余总积分：{result['total']} 分") + else: + item = result.get("item") + extra = f"，需要 {item['cost']} 分" if item else "" + reply = await update.message.reply_text(f"❌ 兑换失败：{result['reason']}{extra}") + asyncio.create_task(auto_delete(update.message, seconds=60)) + asyncio.create_task(auto_delete(reply, seconds=60)) + + +async def cmd_panel(update: Update, context: ContextTypes.DEFAULT_TYPE): + keyboard = InlineKeyboardMarkup([ + [InlineKeyboardButton("👤 我的信息", callback_data="panel:me"), InlineKeyboardButton("✅ 签到", callback_data="panel:checkin")], + [InlineKeyboardButton("🎁 抽奖", callback_data="panel:lottery"), InlineKeyboardButton("🛒 商城", callback_data="panel:shop")], + [InlineKeyboardButton("📖 帮助", callback_data="panel:help")], + ]) + reply = await update.message.reply_text("🎛️ 操作面板\n请选择要执行的操作：", reply_markup=keyboard) + asyncio.create_task(auto_delete(update.message, seconds=60)) + asyncio.create_task(auto_delete(reply, seconds=180)) + + +async def handle_panel_callback(update: Update, context: ContextTypes.DEFAULT_TYPE): + query = update.callback_query + await query.answer() + user = query.from_user + action = query.data.split(":", 1)[1] + if action == "checkin": + result = await daily_checkin(user.id, user.username or user.first_name or str(user.id)) + text = "❌ 你今天已经签到过了" if not result["ok"] else f"✅ 签到成功！奖励 +{result['bonus']} 分，连续 {result['streak']} 天" + elif action == "lottery": + result = await lottery_draw(user.id, user.username or user.first_name or str(user.id)) + text = f"❌ {result.get('reason')}，抽奖需要 {result.get('cost', 10)} 分" if not result["ok"] else f"🎁 {result['prize']}\n当前总积分：{result['total']} 分" + elif action == "shop": + summary = await get_scores_summary(user.id) + text = f"🛒 积分商城\n当前总积分：{summary['total']} 分\n" + "\n".join([f"{k}. {v['name']} — {v['cost']} 分" for k, v in SHOP_ITEMS.items()]) + "\n\n购买请发送 /buy 商品编号" + elif action == "me": + summary = await get_scores_summary(user.id) + text = f"👤 我的积分\n总积分：{summary['total']} 分\n" + "\n".join([f"{s.game_type}: {s.score} 分" for s in summary['scores']]) + else: + text = "📖 常用：/me /checkin /trivia /lottery /shop /panel" + await query.edit_message_text(text) + + async def handle_trivia_answer(update: Update, context: ContextTypes.DEFAULT_TYPE): """处理 /trivia 后的用户答题。""" msg = update.effective_message @@ -160,5 +241,10 @@ def register_fun(app): app.add_handler(CommandHandler("trivia", cmd_trivia, filters.ChatType.GROUPS)) app.add_handler(CommandHandler("score", cmd_score, filters.ChatType.GROUPS)) app.add_handler(CommandHandler("rank", cmd_rank, filters.ChatType.GROUPS)) + app.add_handler(CommandHandler("lottery", cmd_lottery, filters.ChatType.GROUPS)) + app.add_handler(CommandHandler("shop", cmd_shop, filters.ChatType.GROUPS)) + app.add_handler(CommandHandler("buy", cmd_buy, filters.ChatType.GROUPS)) + app.add_handler(CommandHandler("panel", cmd_panel, filters.ChatType.GROUPS)) + app.add_handler(CallbackQueryHandler(handle_panel_callback, pattern=r"^panel:")) app.add_handler(MessageHandler(filters.TEXT & filters.ChatType.GROUPS & ~filters.COMMAND, handle_trivia_answer), group=-1) app.add_handler(MessageHandler(filters.TEXT & filters.ChatType.GROUPS, handle_trigger), group=2) diff --git a/main.py b/main.py index 8f3a7c4..94fab53 100644 --- a/main.py +++ b/main.py @@ -103,6 +103,9 @@ async def start_tg_bot(): BotCommand("stats", "📊 今日统计"), BotCommand("checkin", "✅ 每日签到"), BotCommand("trivia", "🎮 知识答题"), + BotCommand("lottery", "🎁 积分抽奖"), + BotCommand("shop", "🛒 积分商城"), + BotCommand("panel", "🎛️ 操作面板"), BotCommand("bind", "🔗 绑定账号"), BotCommand("me", "👤 查看授权信息"), BotCommand("dban", "🗑️ 删除并封禁"), diff --git a/services/game_service.py b/services/game_service.py index 7be3827..2f385e2 100644 --- a/services/game_service.py +++ b/services/game_service.py @@ -7,19 +7,27 @@ 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 @@ -36,7 +44,7 @@ async def check_trivia_answer(user_id: int, username: str, answer: str, correct_ score.wins += 1 score.streak += 1 if score.streak > 1: - score.score += 5 # 连击奖励 + score.score += 5 else: score.streak = 0 score.last_played = datetime.now() @@ -52,12 +60,70 @@ 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"} + return {"ok": False, "reason": "already_checked", "score": score.score, "streak": score.streak} score.total += 1 - score.score += 5 - score.streak += 1 - bonus = min(score.streak, 7) # 连续签到最多 +7 + 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} + 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"]}