Jerry/tgmsbot
1
0
Fork 0
Code Issues 1 Pull requests Releases Wiki Activity

support context based callback, fix bugs, add cards system

Browse source
This commit is contained in:
JerryXiao 2019-10-15 18:51:46 +08:00
parent 864c72f6ce
commit ae5c7b6eb5
Signed by: Jerry
GPG key ID: 9D9CE43650FF2BAA
4 changed files with 372 additions and 64 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

288
cards.py Normal file
View file

@ -0,0 +1,288 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from telegram import InlineKeyboardMarkup, InlineKeyboardButton
from telegram.ext import run_async
from data import get_player
from random import randrange
from time import time
MAX_LEVEL: int = 100
MID_LEVEL: int = 80
LVL_UP_CARDS: int = 20
def display_username(user, atuser=True, shorten=False, markdown=True):
"""
atuser and shorten has no effect if markdown is True.
"""
name = user.full_name
if markdown:
mdtext = user.mention_markdown(name=user.full_name)
return mdtext
if shorten:
return name
if user.username:
if atuser:
name += " (@{})".format(user.username)
else:
name += " ({})".format(user.username)
return name
def _msg_users(update):
'''
get from_user and reply_to_user
'''
if update.message:
if update.message.reply_to_message:
return (update.message.from_user,
update.message.reply_to_message.from_user)
else:
return (update.message.from_user, None)
else:
return (None, None)
@run_async
def getperm(update, context):
(from_user, reply_to_user) = _msg_users(update)
if not from_user:
return
if reply_to_user:
tuser = reply_to_user
else:
tuser = from_user
tplayer = get_player(int(tuser.id))
update.message.reply_text(f"{display_username(tuser)} 等级为 {tplayer.permission}",
parse_mode="Markdown")
@run_async
def setperm(update, context):
(from_user, reply_to_user) = _msg_users(update)
if not from_user:
return
if reply_to_user:
if context.args and len(context.args) == 1:
try:
new_level = int(context.args[0])
except ValueError:
update.message.reply_text('数字不合法')
return
else:
update.message.reply_text('请指定新的等级')
return
if get_player(int(from_user.id)).permission >= MAX_LEVEL:
tplayer = get_player(int(reply_to_user.id))
tplayer.permission = new_level
tplayer.save()
update.message.reply_text('请求成功')
else:
update.message.reply_text('请求忽略')
else:
update.message.reply_text('请回复被操作人')
@run_async
def lvlup(update, context):
'''
use LVL_UP_CARDS cards to level up 1 lvl
'''
(from_user, reply_to_user) = _msg_users(update)
if not from_user:
return
if reply_to_user:
fplayer = get_player(int(from_user.id))
tplayer = get_player(int(reply_to_user.id))
if fplayer.immunity_cards >= LVL_UP_CARDS:
fplayer.immunity_cards -= LVL_UP_CARDS
if tplayer.permission <= MAX_LEVEL - 2 or tplayer.permission >= MAX_LEVEL:
tplayer.permission += 1
fplayer.save()
tplayer.save()
update.message.reply_text((f"{display_username(from_user)} 消耗了{LVL_UP_CARDS}张免疫卡，"
f"为 {display_username(reply_to_user)} 升了1级"),
parse_mode="Markdown")
else:
update.message.reply_text(f"您的免疫卡不足({fplayer.immunity_cards})，{LVL_UP_CARDS}张免疫卡兑换1等级",
parse_mode="Markdown")
else:
fplayer = get_player(int(from_user.id))
if fplayer.immunity_cards >= LVL_UP_CARDS:
fplayer.immunity_cards -= LVL_UP_CARDS
if fplayer.permission <= MAX_LEVEL - 2 or fplayer.permission >= MAX_LEVEL:
fplayer.permission += 1
fplayer.save()
update.message.reply_text((f"{display_username(from_user)} 消耗了{LVL_UP_CARDS}张免疫卡，"
"为 自己 升了1级"), parse_mode="Markdown")
else:
update.message.reply_text(f"您的免疫卡不足({fplayer.immunity_cards})，{LVL_UP_CARDS}张免疫卡兑换1等级",
parse_mode="Markdown")
@run_async
def transfer_cards(update, context):
(from_user, reply_to_user) = _msg_users(update)
if not from_user:
return
if reply_to_user:
if context.args and len(context.args) == 1:
try:
amount = int(context.args[0])
except ValueError:
update.message.reply_text('数字不合法')
return
else:
update.message.reply_text('请指定数量')
return
if from_user.id == reply_to_user.id:
fplayer = get_player(int(from_user.id))
if fplayer.permission >= MID_LEVEL:
fplayer.immunity_cards += amount
fplayer.save()
update.message.reply_text(f'{display_username(from_user)} 转给了自己{amount}张卡', parse_mode="Markdown")
else:
update.message.reply_text(f'{display_username(from_user)} 转给了自己{amount}张卡', parse_mode="Markdown")
else:
fplayer = get_player(int(from_user.id))
tplayer = get_player(int(reply_to_user.id))
if (amount >= 0 and fplayer.immunity_cards >= amount) or \
(fplayer.permission >= MID_LEVEL and tplayer.permission <= fplayer.permission):
fplayer.immunity_cards -= amount
tplayer.immunity_cards += amount
fplayer.save()
tplayer.save()
update.message.reply_text(f'{display_username(from_user)} 转给了 {display_username(from_user)} {amount}张卡',
parse_mode="Markdown")
else:
update.message.reply_text(f'转账失败，你可能没有这么多卡哦({fplayer.immunity_cards}/{amount})',
parse_mode="Markdown")
else:
update.message.reply_text('请回复被操作人')
@run_async
def rob_cards(update, context):
ROB_TIMEOUT = 10
last_time = context.user_data.setdefault('rob_time', 0.0)
ctime = time()
if ctime - last_time < ROB_TIMEOUT:
update.message.reply_text('别急，你不是刚刚才来过吗')
return
else:
context.user_data['rob_time'] = ctime
(from_user, reply_to_user) = _msg_users(update)
if not from_user:
return
if reply_to_user:
amount = randrange(1, 9)
if from_user.id == reply_to_user.id:
fplayer = get_player(int(from_user.id))
fplayer.immunity_cards -= amount
fplayer.save()
update.message.reply_text(f'{display_username(from_user)} 自己抢走自己{amount}张卡', parse_mode="Markdown")
else:
fplayer = get_player(int(from_user.id))
tplayer = get_player(int(reply_to_user.id))
_fp = fplayer.permission if fplayer.permission > 0 else 0
_tp = tplayer.permission if tplayer.permission > 0 else 0
success_chance = _fp / (_fp + _tp) if _fp + _tp != 0 else 0.5
def __chance(percentage):
if randrange(0,10000)/10000 < percentage:
return True
else:
return False
if __chance(success_chance):
msg_text = "抢劫成功，获得"
else:
msg_text = "抢劫失败，反被抢走"
amount = -amount
fplayer.immunity_cards += amount
tplayer.immunity_cards -= amount
fplayer.save()
tplayer.save()
update.message.reply_text(f'{display_username(from_user)} {msg_text}{abs(amount)}张卡', parse_mode="Markdown")
else:
update.message.reply_text('请回复被操作人')
@run_async
def cards_lottery(update, context):
LOTTERY_TIMEOUT = 10
last_time = context.user_data.setdefault('lottery_time', 0.0)
ctime = time()
if ctime - last_time < LOTTERY_TIMEOUT:
update.message.reply_text('别急，你不是刚刚才来过吗')
return
else:
context.user_data['lottery_time'] = ctime
(from_user, _) = _msg_users(update)
if not from_user:
return
fplayer = get_player(int(from_user.id))
cards = abs(fplayer.immunity_cards) / 3
def __floating(value):
return randrange(5000,15000)/10000 * value
cards = __floating(cards)
cards = int(cards) if cards > 1 else 1
cards *= randrange(-1, 2, 2)
fplayer.immunity_cards += cards
update.message.reply_text(f'您{"获得" if cards >= 0 else "血亏"}了{abs(cards)}张卡')
@run_async
def dist_cards(update, context):
(from_user, _) = _msg_users(update)
if not from_user:
return
try:
if context.args and len(context.args) == 2:
(cards, damount) = [int(a) for a in context.args]
assert (cards > 0 and damount > 0)
fplayer = get_player(int(from_user.id))
fplayer.immunity_cards -= cards
fplayer.save()
red_packets = context.chat_data.setdefault('red_packets', dict())
rphash = str(hash(f"{update.effective_chat.id} {update.effective_message.message_id}"))[:8]
red_packets[rphash] = [cards, damount]
update.message.reply_text(f'{display_username(from_user)}的红包🧧', parse_mode="Markdown",
reply_markup=InlineKeyboardMarkup.from_button(
InlineKeyboardButton(text=f"{cards} / {damount}",
callback_data=f"dist {rphash}")
))
else:
raise ValueError('')
except (ValueError, AssertionError):
update.message.reply_text(f'数字不合法: /dist 卡 红包数量')
@run_async
def dist_cards_btn_click(update, context):
data = update.callback_query.data
user = update.callback_query.from_user
omsg = update.callback_query.message
try:
(_, rphash) = data.split(' ')
red_packets = context.chat_data.setdefault('red_packets', dict())
rp = red_packets.get(str(rphash), None)
if rp:
(cards, damount) = [int(a) for a in rp]
assert (cards > 0 and damount > 0)
def __floating(value):
return randrange(5000,15000)/10000 * value
got_cards = int(__floating(cards/damount))
got_cards = got_cards if got_cards <= cards else cards
got_cards = got_cards if damount != 1 else cards
rp[0] -= got_cards
rp[1] -= 1
(cards, damount) = rp
fplayer = get_player(int(user.id))
fplayer.immunity_cards += cards
fplayer.save()
update.callback_query.answer(text=f"你得到了{got_cards}张卡", show_alert=False)
if cards > 0 and damount > 0:
omsg.reply_markup.inline_keyboard[0][0].text = f"{cards} / {damount}"
omsg.edit_reply_markup(reply_markup=omsg.reply_markup)
else:
raise AssertionError('')
else:
raise AssertionError('')
except (ValueError, AssertionError):
try:
update.callback_query.answer()
except Exception:
pass
omsg.edit_text(omsg.text_markdown + "褪裙了", parse_mode="Markdown", reply_markup=None)

11
data.py
View file

@ -13,25 +13,18 @@ class Player(Model):
wins = IntegerField()
restricted_until = IntegerField()
immunity_cards = IntegerField()
permission = IntegerField()
class Meta:
database = db
def save(self, *args, **kwargs):
super().save(*args, **kwargs)
db.close()
@staticmethod
def db_close():
db.close()
db.connect()
db.create_tables([Player])
db.close()
def get_player(user_id):
db.connect()
player = Player.get_or_none(Player.user_id == user_id)
if player is None:
player = Player.create(user_id=user_id, mines=0, death=0, wins=0,
restricted_until=0, immunity_cards=0)
restricted_until=0, immunity_cards=0, permission=0)
return player
else:
return player

15
data_ram.py
View file

@ -1,28 +1,27 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
db = lambda _=None: None
setattr(db, 'close', lambda _=None: None)
pool = dict()
class Player():
def __init__(self, user_id, mines, death, wins, restricted_until, immunity_cards):
def __init__(self, user_id, mines, death, wins,
restricted_until, immunity_cards, permission):
self.user_id = user_id
self.mines = mines
self.death = death
self.wins = wins
self.restricted_until = restricted_until
self.immunity_cards = immunity_cards
@staticmethod
def save():
pass
@staticmethod
def db_close():
pass
self.permission = permission
def get_player(user_id):
player = pool.get(user_id, None)
if player is None:
player = Player(user_id=user_id, mines=0, death=0, wins=0,
restricted_until=0, immunity_cards=0)
restricted_until=0, immunity_cards=0, permission=0)
pool[user_id] = player
return player
else:

108
tgmsbot.py
View file

@ -6,9 +6,10 @@ from telegram import InlineKeyboardMarkup, InlineKeyboardButton
from telegram.ext import Updater, CommandHandler, CallbackQueryHandler, run_async
from telegram.error import TimedOut as TimedOutError
from numpy import array_equal
# If no peewee orm is installed, try `from data_ram import get_player`
from data import get_player
from random import randint, choice
# If no peewee orm is installed, try `from data_ram import get_player, db`
from data import get_player, db
from random import randint, choice, randrange
from math import log
from threading import Lock
import time
import logging
@ -17,7 +18,7 @@ logging.basicConfig(level=logging.INFO,format='%(asctime)s - %(name)s - %(leveln
logger = logging.getLogger(__name__)
token = "token_here"
updater = Updater(token, workers=8)
updater = Updater(token, workers=8, use_context=True)
job_queue = updater.job_queue
job_queue.start()
@ -29,8 +30,9 @@ WIDTH = 8
MINES = 9
UNOPENED_CELL = "\u2588"
FLAGGED_CELL = "\u259a"
STEPPED_CELL = "*"
FLAGGED_CELL = "\U0001f6a9"
#FLAGGED_CELL = "\u259a"
STEPPED_CELL = "\u2622"
WIN_TEXT_TEMPLATE = "哇所有奇怪的地方都被你打开啦…好羞羞\n" \
"地图：Op {s_op} / Is {s_is} / 3bv {s_3bv}\n操作总数 {ops_count}\n" \
@ -132,13 +134,16 @@ game_manager = GameManager()
@run_async
def send_keyboard(bot, update, args):
def send_keyboard(update, context):
(bot, args) = (context.bot, context.args)
msg = update.message
logger.info("Mine from {0}".format(update.message.from_user.id))
if check_restriction(update.message.from_user):
update.message.reply_text("爆炸这么多次还想扫雷？")
return
# create a game board
if args is None:
args = list()
if len(args) == 3:
height = HEIGHT
width = WIDTH
@ -182,64 +187,74 @@ def send_keyboard(bot, update, args):
bot.send_message(chat_id=msg.chat.id, text="路过的大爷～来扫个雷嘛～", reply_to_message_id=msg.message_id,
parse_mode="Markdown", reply_markup=InlineKeyboardMarkup(keyboard))
def send_help(bot, update):
def send_help(update, context):
logger.debug("Start from {0}".format(update.message.from_user.id))
msg = update.message
msg.reply_text("这是一个扫雷bot\n\n/mine 开始新游戏")
def send_source(bot, update):
def send_source(update, context):
logger.debug("Source from {0}".format(update.message.from_user.id))
update.message.reply_text('Source code: https://git.jerryxiao.cc/Jerry/tgmsbot')
def send_status(bot, update):
def send_status(update, context):
logger.info("Status from {0}".format(update.message.from_user.id))
count = game_manager.count()
update.message.reply_text('当前进行的游戏: {}'.format(count))
def gen_reward(user, negative=True):
''' Reward the player :) '''
def __chance(percentage):
if randrange(0,10000)/10000 < percentage:
return True
else:
return False
def __floating(value):
return randrange(8000,12000)/10000 * value
def __lose_cards(cardnum):
if cardnum <= 6:
return 1
else:
return int(__floating(log(cardnum, 2)))
def __get_cards(cardnum):
if cardnum >= 2:
cards = __floating(1 / log(cardnum, 100))
if cards > 1.0:
return int(cards)
else:
return int(__chance(cards))
else:
return int(__floating(8.0))
# Negative rewards
def restrict_mining(player):
if player.immunity_cards >= 1:
if player.immunity_cards >= 10:
lost_cards = randint(2,4)
elif player.immunity_cards >= 5:
lost_cards = randint(1,3)
else:
lost_cards = 1
lost_cards = __lose_cards(player.immunity_cards)
player.immunity_cards -= lost_cards
if player.immunity_cards >= 0:
ret = "用去{}张免疫卡，还剩{}张".format(lost_cards, player.immunity_cards)
else:
now = int(time.time())
seconds = randint(30, 120)
player.restricted_until = now + seconds
ret = "没有免疫卡了，被限制扫雷{}秒".format(seconds)
player.save()
return ret
# Positive rewards
def give_immunity_cards(player):
rewarded_cards = 0
if player.immunity_cards <= 3:
rewarded_cards = randint(1, 2)
elif player.immunity_cards <= 10:
if randint(1, 5) == 5:
rewarded_cards = 1
elif randint(1, 10) == 10:
rewarded_cards = 1
rewarded_cards = __get_cards(player.immunity_cards)
player.immunity_cards += rewarded_cards
player.save()
if rewarded_cards == 0:
return "共有{}张免疫卡".format(player.immunity_cards)
else:
return "被奖励了{}张免疫卡，共有{}张".format(rewarded_cards, player.immunity_cards)
player = get_player(user.id)
try:
if negative:
player.death += 1
return restrict_mining(player)
else:
player.wins += 1
return give_immunity_cards(player)
finally:
player.save()
def game_count(user):
player = get_player(user.id)
@ -248,7 +263,6 @@ def game_count(user):
def check_restriction(user):
player = get_player(user.id)
player.db_close()
now = int(time.time())
if now >= player.restricted_until:
return False
@ -256,11 +270,10 @@ def check_restriction(user):
return player.restricted_until - now
@run_async
def player_statistics(bot, update):
def player_statistics(update, context):
logger.info("Statistics from {0}".format(update.message.from_user.id))
user = update.message.from_user
player = get_player(user.id)
player.db_close()
mines = player.mines
death = player.death
wins = player.wins
@ -271,7 +284,7 @@ def player_statistics(bot, update):
wins=wins, cards=cards))
def update_keyboard_request(bot, bhash, game, chat_id, message_id):
def update_keyboard_request(context, bhash, game, chat_id, message_id):
current_action_timestamp = time.time()
if current_action_timestamp - game.last_action <= KBD_MIN_INTERVAL:
logger.debug('Rate limit triggered.')
@ -280,8 +293,9 @@ def update_keyboard_request(bot, bhash, game, chat_id, message_id):
context=(bhash, game, chat_id, message_id, current_action_timestamp))
else:
game.last_action = current_action_timestamp
update_keyboard(bot, None, noqueue=(bhash, game, chat_id, message_id))
def update_keyboard(bot, job, noqueue=None):
update_keyboard(context, noqueue=(bhash, game, chat_id, message_id))
def update_keyboard(context, noqueue=None):
(bot, job) = (context.bot, context.job)
if noqueue:
(bhash, game, chat_id, message_id) = noqueue
else:
@ -317,7 +331,8 @@ def update_keyboard(bot, job, noqueue=None):
game.timeouts += 1
@run_async
def handle_button_click(bot, update):
def handle_button_click(update, context):
bot = context.bot
msg = update.callback_query.message
user = update.callback_query.from_user
chat_id = update.callback_query.message.chat.id
@ -340,10 +355,10 @@ def handle_button_click(bot, update):
if game is None:
logger.debug("No game found for hash {}".format(bhash))
return
game.lock.acquire()
try:
if game.stopped:
return
game.lock.acquire()
board = game.board
if board.state == 0:
mmap = None
@ -355,7 +370,7 @@ def handle_button_click(bot, update):
game.lock.release()
game.save_action(user, (row, col))
if not array_equal(board.map, mmap):
update_keyboard_request(bot, bhash, game, chat_id, msg.message_id)
update_keyboard_request(context, bhash, game, chat_id, msg.message_id)
(s_op, s_is, s_3bv) = board.gen_statistics()
ops_count = game.actions_sum()
ops_list = game.get_actions()
@ -394,7 +409,7 @@ def handle_button_click(bot, update):
elif mmap is None or (not array_equal(board.map, mmap)):
game.lock.release()
game.save_action(user, (row, col))
update_keyboard_request(bot, bhash, game, chat_id, msg.message_id)
update_keyboard_request(context, bhash, game, chat_id, msg.message_id)
else:
game.lock.release()
except:
@ -405,12 +420,25 @@ def handle_button_click(bot, update):
raise
from cards import getperm, setperm, lvlup, transfer_cards, rob_cards, cards_lottery, dist_cards, dist_cards_btn_click
updater.dispatcher.add_handler(CommandHandler('getlvl', getperm))
updater.dispatcher.add_handler(CommandHandler('setlvl', setperm))
updater.dispatcher.add_handler(CommandHandler('lvlup', lvlup))
updater.dispatcher.add_handler(CommandHandler('transfer', transfer_cards))
updater.dispatcher.add_handler(CommandHandler('rob', rob_cards))
updater.dispatcher.add_handler(CommandHandler('lottery', cards_lottery))
updater.dispatcher.add_handler(CommandHandler('dist', dist_cards))
updater.dispatcher.add_handler(CallbackQueryHandler(dist_cards_btn_click, pattern=r'dist'))
updater.dispatcher.add_handler(CommandHandler('start', send_help))
updater.dispatcher.add_handler(CommandHandler('mine', send_keyboard, pass_args=True))
updater.dispatcher.add_handler(CommandHandler('mine', send_keyboard))
updater.dispatcher.add_handler(CommandHandler('status', send_status))
updater.dispatcher.add_handler(CommandHandler('stats', player_statistics))
updater.dispatcher.add_handler(CommandHandler('source', send_source))
updater.dispatcher.add_handler(CallbackQueryHandler(handle_button_click))
updater.start_polling()
updater.idle()
try:
updater.start_polling()
updater.idle()
finally:
db.close()