ngfchl/tg-spam-guard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add points lottery shop and panel

Browse Source
This commit is contained in:
ngfchl
2026-07-07 10:32:57 +08:00
parent 9bcecea152
commit 57656a3cc3
4 changed files with 177 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
handlers/bind.py
+11 -6
View File
@@ -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 = (
handlers/fun.py
+90 -4
View File
@@ -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)
main.py
+3
View File
@@ -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", "🗑️ 删除并封禁"),
services/game_service.py
+73 -7
View File
@@ -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"]}