ngfchl
400 lines
17 KiB
Python
400 lines
17 KiB
Python
"""气氛组 + 小游戏 handler"""
|
||
import asyncio
|
||
import logging
|
||
|
||
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_scores_summary,
|
||
SHOP_ITEMS, buy_shop_item,
|
||
parse_lottery_spec, create_lottery_event, lottery_condition_text, lottery_end_text,
|
||
count_lottery_participants, join_lottery_event, draw_lottery_event,
|
||
)
|
||
|
||
logger = logging.getLogger("spam_guard")
|
||
PANEL_IDLE_SECONDS = 60
|
||
|
||
# 关键词触发
|
||
def panel_keyboard():
|
||
return 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")],
|
||
])
|
||
|
||
|
||
def schedule_panel_idle_delete(context: ContextTypes.DEFAULT_TYPE, message, seconds: int = PANEL_IDLE_SECONDS):
|
||
tasks = context.bot_data.setdefault("panel_idle_tasks", {})
|
||
key = (message.chat_id, message.message_id)
|
||
old_task = tasks.pop(key, None)
|
||
if old_task and not old_task.done():
|
||
old_task.cancel()
|
||
|
||
async def _delete_later():
|
||
try:
|
||
await asyncio.sleep(seconds)
|
||
await message.delete()
|
||
except asyncio.CancelledError:
|
||
return
|
||
except Exception:
|
||
pass
|
||
finally:
|
||
tasks.pop(key, None)
|
||
|
||
tasks[key] = asyncio.create_task(_delete_later())
|
||
|
||
|
||
TRIGGER_REACTIONS = {
|
||
"谢谢": "👍❤️",
|
||
"感谢": "👍❤️🎉",
|
||
"哈哈": "😂",
|
||
"笑死": "😂💀",
|
||
"牛逼": "🔥💪",
|
||
"nb": "🔥💪",
|
||
"666": "🔥",
|
||
"晚安": "🌙✨",
|
||
}
|
||
|
||
|
||
async def handle_trigger(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
"""关键词触发气氛组"""
|
||
msg = update.effective_message
|
||
if not msg or not msg.text:
|
||
return
|
||
|
||
text = msg.text.strip()
|
||
for keyword, emoji in TRIGGER_REACTIONS.items():
|
||
if text == keyword or text.startswith(keyword):
|
||
try:
|
||
await msg.react(emoji)
|
||
except Exception:
|
||
pass
|
||
break
|
||
|
||
|
||
async def cmd_checkin(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
"""每日签到"""
|
||
user = update.effective_user
|
||
result = await daily_checkin(user.id, user.username or user.first_name)
|
||
if result["ok"]:
|
||
reply = await update.message.reply_text(
|
||
f"✅ 签到成功！\n"
|
||
f"📊 积分: {result['score']}\n"
|
||
f"🔥 连续签到: {result['streak']} 天\n"
|
||
f"🎁 签到奖励: +{result['bonus']} 分（基础 {result.get('base', 0)} + 连签 {result.get('streak_bonus', 0)}）"
|
||
)
|
||
else:
|
||
reply = await update.message.reply_text("❌ 你今天已经签到过了")
|
||
asyncio.create_task(auto_delete(update.message, seconds=60))
|
||
asyncio.create_task(auto_delete(reply, seconds=60))
|
||
|
||
|
||
async def cmd_trivia(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
"""收割机知识答题"""
|
||
q = await get_trivia_question()
|
||
text = f"❓ {q['q']}\n\n"
|
||
for i, opt in enumerate(q["opts"], 1):
|
||
text += f"{i}. {opt}\n"
|
||
text += f"\n💡 请回复序号或答案文字"
|
||
context.user_data["trivia_answer"] = q["a"]
|
||
context.user_data["trivia_options"] = q["opts"]
|
||
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_score(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
"""查看积分"""
|
||
from db.models import GameScore
|
||
user = update.effective_user
|
||
scores = await GameScore.filter(user_id=user.id).all()
|
||
if not scores:
|
||
reply = await update.message.reply_text("📊 你还没有积分记录\n💡 签到 /checkin 或答题 /trivia 开始赚积分")
|
||
else:
|
||
text = "📊 你的积分\n━━━━━━━━━━━━\n"
|
||
for s in scores:
|
||
text += f"🎮 {s.game_type}: {s.score} 分 (胜率 {s.wins}/{s.total})\n"
|
||
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_rank(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
"""排行榜"""
|
||
top = await get_leaderboard("trivia", 10)
|
||
if not top:
|
||
reply = await update.message.reply_text("🏆 暂无排行数据")
|
||
else:
|
||
text = "🏆 知识答题排行榜\n━━━━━━━━━━━━━━━━━━\n"
|
||
medals = ["🥇", "🥈", "🥉"]
|
||
for i, s in enumerate(top):
|
||
medal = medals[i] if i < 3 else f" {i+1}."
|
||
text += f"{medal} @{s.username or s.user_id} — {s.score} 分\n"
|
||
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 build_lottery_text(event, count: int | None = None) -> str:
|
||
if count is None:
|
||
count = await count_lottery_participants(event.id)
|
||
return (
|
||
"🎁 群内抽奖\n"
|
||
"━━━━━━━━━━━━\n"
|
||
f"🎯 奖品：{event.prize}\n"
|
||
f"👤 发起人：@{event.creator_name or event.creator_id}\n"
|
||
f"✅ 参与条件：{await lottery_condition_text(event)}\n"
|
||
f"⏱️ 结束条件：{await lottery_end_text(event)}\n"
|
||
f"👥 当前参与：{count} 人\n\n"
|
||
"点击下方按钮参与抽奖。"
|
||
)
|
||
|
||
|
||
def lottery_keyboard(event_id: int) -> InlineKeyboardMarkup:
|
||
return InlineKeyboardMarkup([[InlineKeyboardButton("🎁 参与抽奖", callback_data=f"lottery:join:{event_id}")]])
|
||
|
||
|
||
async def schedule_lottery_draw(context: ContextTypes.DEFAULT_TYPE, event_id: int, seconds: int):
|
||
async def _run():
|
||
try:
|
||
await asyncio.sleep(seconds)
|
||
result = await draw_lottery_event(event_id)
|
||
event = result.get("event")
|
||
if not event or not event.message_id:
|
||
return
|
||
if result.get("ok"):
|
||
winner = result["winner"]
|
||
text = (
|
||
"🎉 抽奖已开奖\n"
|
||
"━━━━━━━━━━━━\n"
|
||
f"🎯 奖品：{event.prize}\n"
|
||
f"👥 参与人数：{result['count']} 人\n"
|
||
f"🏆 中奖用户：@{winner.username or winner.first_name or winner.user_id}"
|
||
)
|
||
else:
|
||
text = f"🎁 抽奖已结束\n━━━━━━━━━━━━\n🎯 奖品：{event.prize}\n❌ {result.get('reason', '未开奖')}"
|
||
await context.bot.edit_message_text(chat_id=event.chat_id, message_id=event.message_id, text=text)
|
||
except Exception as e:
|
||
logger.warning(f"⚠️ 自动开奖失败: {type(e).__name__}: {e}")
|
||
asyncio.create_task(_run())
|
||
|
||
|
||
async def create_lottery_from_text(update: Update, context: ContextTypes.DEFAULT_TYPE, spec_text: str):
|
||
user = update.effective_user
|
||
msg = update.effective_message
|
||
spec = parse_lottery_spec(spec_text)
|
||
event = await create_lottery_event(
|
||
chat_id=msg.chat_id,
|
||
creator_id=user.id,
|
||
creator_name=user.username or user.first_name or str(user.id),
|
||
spec=spec,
|
||
)
|
||
sent = await msg.reply_text(await build_lottery_text(event), reply_markup=lottery_keyboard(event.id))
|
||
event.message_id = sent.message_id
|
||
await event.save()
|
||
if event.end_type == "minutes":
|
||
await schedule_lottery_draw(context, event.id, event.end_value * 60)
|
||
return sent
|
||
|
||
|
||
async def cmd_lottery(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
parts = (update.message.text or "").split(maxsplit=1)
|
||
if len(parts) > 1:
|
||
try:
|
||
await create_lottery_from_text(update, context, parts[1])
|
||
except Exception as e:
|
||
reply = await update.message.reply_text(f"❌ 抽奖格式错误：{e}\n\n示例：/lottery 月卡一张 | all | people:3")
|
||
asyncio.create_task(auto_delete(reply, seconds=90))
|
||
else:
|
||
context.user_data["awaiting_lottery_spec"] = True
|
||
reply = await update.message.reply_text(
|
||
"🎁 发起群内抽奖\n"
|
||
"请发送：奖品 | 参与条件 | 结束条件\n\n"
|
||
"示例：\n"
|
||
"月卡一张 | all | people:3\n"
|
||
"邀请码 | checkin | minutes:10\n"
|
||
"徽章 | points:100 | people:5\n\n"
|
||
"参与条件：all / checkin / points:N\n"
|
||
"结束条件：people:N / minutes:N"
|
||
)
|
||
asyncio.create_task(auto_delete(reply, seconds=120))
|
||
asyncio.create_task(auto_delete(update.message, 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):
|
||
reply = await update.message.reply_text(
|
||
"🎛️ 操作面板\n请选择要执行的操作：\n\n⏱️ 60 秒无操作后自动消失。",
|
||
reply_markup=panel_keyboard(),
|
||
)
|
||
asyncio.create_task(auto_delete(update.message, seconds=60))
|
||
schedule_panel_idle_delete(context, reply)
|
||
|
||
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":
|
||
context.user_data["awaiting_lottery_spec"] = True
|
||
text = (
|
||
"🎁 发起群内抽奖\n"
|
||
"请在群里发送：奖品 | 参与条件 | 结束条件\n\n"
|
||
"示例：月卡一张 | all | people:3\n"
|
||
"参与条件：all / checkin / points:N\n"
|
||
"结束条件：people:N / minutes:N"
|
||
)
|
||
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 + "\n\n⏱️ 60 秒无操作后自动消失。",
|
||
reply_markup=panel_keyboard(),
|
||
)
|
||
schedule_panel_idle_delete(context, query.message)
|
||
|
||
|
||
async def handle_trivia_answer(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
"""处理 /trivia 后的用户答题。"""
|
||
msg = update.effective_message
|
||
user = update.effective_user
|
||
if not msg or not msg.text or not user or user.is_bot:
|
||
return
|
||
|
||
if context.user_data.get("awaiting_lottery_spec"):
|
||
try:
|
||
await create_lottery_from_text(update, context, msg.text.strip())
|
||
context.user_data.pop("awaiting_lottery_spec", None)
|
||
asyncio.create_task(auto_delete(msg, seconds=60))
|
||
except Exception as e:
|
||
reply = await msg.reply_text(f"❌ 抽奖格式错误：{e}\n请重新发送，或发送 /panel 重新开始。")
|
||
asyncio.create_task(auto_delete(reply, seconds=90))
|
||
raise ApplicationHandlerStop
|
||
|
||
correct_answer = context.user_data.get("trivia_answer")
|
||
options = context.user_data.get("trivia_options") or []
|
||
if not correct_answer:
|
||
return
|
||
|
||
raw_answer = msg.text.strip()
|
||
answer = raw_answer
|
||
if raw_answer.isdigit():
|
||
idx = int(raw_answer) - 1
|
||
if 0 <= idx < len(options):
|
||
answer = options[idx]
|
||
|
||
result = await check_trivia_answer(user.id, user.username or user.first_name or str(user.id), answer, correct_answer)
|
||
context.user_data.pop("trivia_answer", None)
|
||
context.user_data.pop("trivia_options", None)
|
||
|
||
if result["correct"]:
|
||
text = (
|
||
"✅ 回答正确！\n"
|
||
f"🎯 正确答案：{correct_answer}\n"
|
||
f"📊 当前积分：{result['score']}\n"
|
||
f"🔥 连击：{result['streak']}"
|
||
)
|
||
else:
|
||
text = (
|
||
"❌ 回答错误\n"
|
||
f"🎯 正确答案：{correct_answer}\n"
|
||
f"📊 当前积分：{result['score']}"
|
||
)
|
||
reply = await msg.reply_text(text)
|
||
asyncio.create_task(auto_delete(msg, seconds=60))
|
||
asyncio.create_task(auto_delete(reply, seconds=60))
|
||
raise ApplicationHandlerStop
|
||
|
||
|
||
async def handle_lottery_callback(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
query = update.callback_query
|
||
await query.answer()
|
||
parts = query.data.split(":")
|
||
event_id = int(parts[-1])
|
||
user = query.from_user
|
||
result = await join_lottery_event(event_id, user.id, user.username or "", user.first_name or "")
|
||
if not result["ok"]:
|
||
await query.answer(result["reason"], show_alert=True)
|
||
return
|
||
event = result["event"]
|
||
if result["should_draw"]:
|
||
draw = await draw_lottery_event(event.id)
|
||
if draw.get("ok"):
|
||
winner = draw["winner"]
|
||
text = (
|
||
"🎉 抽奖已开奖\n"
|
||
"━━━━━━━━━━━━\n"
|
||
f"🎯 奖品：{event.prize}\n"
|
||
f"👥 参与人数：{draw['count']} 人\n"
|
||
f"🏆 中奖用户：@{winner.username or winner.first_name or winner.user_id}"
|
||
)
|
||
await query.edit_message_text(text)
|
||
else:
|
||
await query.edit_message_text(f"🎁 抽奖已结束\n❌ {draw.get('reason', '未开奖')}")
|
||
else:
|
||
await query.edit_message_text(await build_lottery_text(event, result["count"]), reply_markup=lottery_keyboard(event.id))
|
||
await query.answer(f"参与成功，当前 {result['count']} 人")
|
||
|
||
|
||
async def auto_delete(message, seconds=60):
|
||
try:
|
||
await asyncio.sleep(seconds)
|
||
await message.delete()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def register_fun(app):
|
||
app.add_handler(CommandHandler("checkin", cmd_checkin, filters.ChatType.GROUPS))
|
||
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(CallbackQueryHandler(handle_lottery_callback, pattern=r"^lottery:"))
|
||
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)
|