#!/usr/bin/env python3 # -*- coding: utf-8 -*- from telegram import InlineKeyboardMarkup, InlineKeyboardButton from telegram.ext import run_async from random import randrange from time import time import logging logger = logging.getLogger('tgmsbot.cards') # from the main module get_player = lambda *args, **kwargs: None game_manager = None 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): logger.info(f'getperm from {getattr(update.effective_user, "id", None)}') (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}\n" f"口袋里有 {tplayer.immunity_cards} 张免疫卡"), parse_mode="Markdown") @run_async def setperm(update, context): logger.info(f'setperm from {getattr(update.effective_user, "id", None)}') (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 ''' logger.info(f'lvlup from {getattr(update.effective_user, "id", None)}') LVLUP_TIMEOUT = 10 last_time = context.user_data.setdefault('lvlup_time', 0.0) ctime = time() if ctime - last_time < LVLUP_TIMEOUT: update.message.reply_text('别急，你不是刚刚才来过吗\nTips: /lvlup n 可以一次升n级哦') return else: context.user_data['lvlup_time'] = ctime (from_user, reply_to_user) = _msg_users(update) if context.args and len(context.args) == 1: try: amount = int(context.args[0]) except ValueError: update.message.reply_text('数字不合法') return else: amount = 1 if not from_user: return if reply_to_user: fplayer = get_player(int(from_user.id)) tplayer = get_player(int(reply_to_user.id)) amount = abs(amount) if fplayer.immunity_cards >= (used_cards := LVL_UP_CARDS * amount): fplayer.immunity_cards -= used_cards tplayer.permission = MAX_LEVEL - 1 if (_tpp := tplayer.permission + amount) >= MAX_LEVEL - 1 \ and tplayer.permission < MAX_LEVEL else _tpp fplayer.save() tplayer.save() update.message.reply_text((f"{display_username(from_user)} 消耗了{used_cards}张免疫卡，" f"为 {display_username(reply_to_user)} 升了{amount}级"), parse_mode="Markdown") else: update.message.reply_text(f"您的免疫卡不足({fplayer.immunity_cards})，{used_cards}张免疫卡兑换{amount}等级", parse_mode="Markdown") else: fplayer = get_player(int(from_user.id)) if fplayer.immunity_cards >= (used_cards := LVL_UP_CARDS * amount): if amount < 0: if fplayer.permission + amount >= 0: fplayer.immunity_cards += abs(used_cards) else: fplayer.immunity_cards += LVL_UP_CARDS * fplayer.permission else: fplayer.immunity_cards -= abs(used_cards) fplayer.permission = MAX_LEVEL - 1 if (_fpp := fplayer.permission + amount) >= MAX_LEVEL - 1 \ and fplayer.permission < MAX_LEVEL else _fpp fplayer.save() update.message.reply_text((f"{display_username(from_user)} 消耗了{used_cards}张免疫卡，" f"为 自己 升了{amount}级"), parse_mode="Markdown") else: update.message.reply_text(f"您的免疫卡不足({fplayer.immunity_cards})，{used_cards}张免疫卡兑换{amount}等级", parse_mode="Markdown") @run_async def transfer_cards(update, context): logger.info(f'transfer_cards from {getattr(update.effective_user, "id", None)}') (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(reply_to_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): logger.info(f'rob_cards from {getattr(update.effective_user, "id", None)}') 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 MSG_TEXT_SUCCESS = "抢劫成功，获得" MSG_TEXT_FAIL = "抢劫失败，反被抢走" if _fp >= MID_LEVEL and _tp >= MID_LEVEL: cards_amount = int(max(abs(fplayer.immunity_cards), abs(tplayer.immunity_cards)) * randrange(1000,8000)/10000) lvl_amount = int(max(_fp, _tp) * randrange(1000,8000)/10000) if (_tple if (_fp < MAX_LEVEL) ^ (_tple := _tp < MAX_LEVEL) else __chance(success_chance)): msg_text = MSG_TEXT_SUCCESS else: msg_text = MSG_TEXT_FAIL cards_amount = -cards_amount lvl_amount = -lvl_amount fplayer.immunity_cards += cards_amount tplayer.immunity_cards -= cards_amount fplayer.permission = _fpp if (_fpp := _fp + lvl_amount) < MAX_LEVEL or _fp >= MAX_LEVEL else MAX_LEVEL - 1 tplayer.permission = _tpp if (_tpp := _tp - lvl_amount) < MAX_LEVEL or _tp >= MAX_LEVEL else MAX_LEVEL - 1 fplayer.save() tplayer.save() update.message.reply_text((f'{display_username(from_user)} {msg_text}{abs(cards_amount)}张卡, ' f'{abs(lvl_amount)}级'), parse_mode="Markdown") else: if __chance(success_chance): msg_text = MSG_TEXT_SUCCESS else: msg_text = MSG_TEXT_FAIL 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): logger.info(f'cards_lottery from {getattr(update.effective_user, "id", None)}') 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 fplayer.save() update.message.reply_text(f'您{"获得" if cards >= 0 else "血亏"}了{abs(cards)}张卡') @run_async def dist_cards(update, context): logger.info(f'dist_cards from {getattr(update.effective_user, "id", None)}') (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)) assert fplayer.immunity_cards >= cards 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): logger.info(f'dist_cards_btn_click from {getattr(update.effective_user, "id", None)}') data = update.callback_query.data user = update.callback_query.from_user omsg = update.callback_query.message red_packets = context.chat_data.setdefault('red_packets', dict()) try: (_, rphash) = data.split(' ') 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 = 1 if got_cards == 0 and randrange(0,10000)/10000 < 0.2 else got_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 += got_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 def free_mem(job_context): try: red_packets.pop(rphash) except KeyError: pass if rphash: rp = red_packets.get(rphash, [0, 0]) if rp[0] != -1: rp[0] = -1 omsg.edit_text(omsg.text_markdown + "褪裙了", parse_mode="Markdown", reply_markup=None) context.job_queue.run_once(free_mem, 5) @run_async def reveal(update, context): logger.info(f'reveal from {getattr(update.effective_user, "id", None)}') (from_user, _) = _msg_users(update) if not from_user: return if (msg := update.effective_message) and (rmsg := msg.reply_to_message): try: assert (rmarkup := rmsg.reply_markup) and (kbd := rmarkup.inline_keyboard) \ and type((btn := kbd[0][0])) is InlineKeyboardButton and (data := btn.callback_data) data = data.split(' ') data = [int(i) for i in data] (bhash, _, _, _) = data except: msg.reply_text('不是一条有效的消息') return game = game_manager.get_game_from_hash(bhash) if not game: msg.reply_text('这局似乎走丢了呢') return if (mmap := game.board.mmap) is None: msg.reply_text('这局似乎还没开始呢') return def map_to_msg(): ZERO_CELL = '\u23f9' MINE_CELL = '\u2622' NUM_CELL_SUFFIX = '\ufe0f\u20e3' BAD_CELL = "\U0001f21a\ufe0f" msg_text = "" for row in mmap: for cell in row: if cell == 0: msg_text += ZERO_CELL elif cell == 9: msg_text += MINE_CELL elif cell in range(1,9): msg_text += str(cell) + NUM_CELL_SUFFIX else: msg_text += BAD_CELL msg_text += '\n' return msg_text 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 extra_text = "" fplayer.immunity_cards -= cards if fplayer.permission >= MID_LEVEL and fplayer.permission < MAX_LEVEL: lvl = int(randrange(100,3000)/10000 * fplayer.permission) lvl = lvl if lvl > 0 else 1 fplayer.permission -= lvl extra_text = f", {lvl}级" fplayer.save() msg.reply_text(f'本局地图如下:\n\n{map_to_msg()}\n您用去了{cards}张卡{extra_text}') else: msg.reply_text('请回复想要查看的雷区')