添加轮盘赌决斗版

shebot/_interact.py

@@ -0,0 +1,149 @@
+from typing import Any, Union, Callable
+from aiocqhttp.typing import Message_T
+from hoshino import logger
+from nonebot.message import CanceledException
+import nonebot
+from hoshino.typing import CQEvent as Event
+from datetime import datetime, timedelta
+from collections import defaultdict
+
+_allsession = {}
+_allaction = defaultdict(dict)
+
+class ActSession:
+    def __init__(self, name: str, group_id: int, user_id: int, max_user: int, expire_time: int, usernum_limit: bool):
+        self.name = name
+        self.bot = nonebot.get_bot()
+        self.group_id = group_id #session所在群聊
+        self.creator = user_id #session的创建者
+        self.usernum_limit = usernum_limit
+        self.users = list([user_id])
+        self.max_user = max_user
+        self.is_valid= True
+        self.last_interaction = None
+        self.expire_time = expire_time
+        self.create_time = datetime.now()
+        self._state = {} 
+        self._actions = {}
+        #self.handle_msg = None
+
+    def __getattr__(self, item) -> Any:
+        return self.state.get(item)
+
+    @property
+    def state(self):
+        """
+        State of the session.
+
+        This contains all named arguments and
+        other session scope temporary values.
+        """
+        return self._state
+
+    @property
+    def actions(self):
+        """
+        Actions of the session.
+
+        This dict contains all actions which
+        can be triggered by certain word
+        """
+        return _allaction.get(self.name) or {}
+
+    @property
+    def handle_msg(self) -> Callable:
+        return interact.allhandler.get(self.name)
+
+    @classmethod
+    def from_event(cls, name: str, event: Event, max_user: int=100, expire_time: int=300, usernum_limit: bool=False):
+        return cls(name, event.group_id, event.user_id, max_user, expire_time, usernum_limit)
+
+    def count_user(self) -> int:
+        return len(self.users)
+
+    def add_user(self, uid: int):
+        #this function should cautiously be used 
+        #because it can not assure user add only one session in the same group
+        #better to use join_session in InteractHandler  
+        if len(self.users) >= self.max_user:
+            raise ValueError('人数超过限制,无法加入')
+        self.users.append(uid)
+
+    def close(self, message: Message_T=None):
+        InteractHandler().close_session(self.group_id, self.name)
+        logger.info(f'interaction session {self.name} has been closed')
+
+    def is_expire(self) -> bool:
+        now = datetime.now()
+        return self.create_time + timedelta(seconds=self.expire_time) < now
+
+    async def send(self, event, message, **kwargs):
+        await self.bot.send(event, message, **kwargs)
+
+    async def finish(self, event, message, **kwargs):
+        await self.send(event, message, **kwargs)
+        raise CanceledException('finished')
+
+class InteractHandler:
+    def __init__(self) -> None:
+        self.allsession = {}
+        global _allaction
+        global _allsession
+        #global _allhandler
+        self.allsession = _allsession
+        self.allaction = _allaction
+        self.allhandler = {}
+
+    def add_session(self, session: ActSession):
+        gid = session.group_id
+        name = session.name
+        if (gid, name) in self.allsession: 
+            raise ValueError(f'{self.allsession[(gid, name)].name} 正在进行中')
+        self.allsession[(gid, name)] = session
+
+    @staticmethod
+    def close_session(group_id: int, name: str):
+        global _allsession
+        if (group_id, name) in _allsession:
+            del _allsession[(group_id, name)]
+
+    def find_session(self, event: Event, name=None) -> ActSession:
+        gid = event.group_id
+        uid = event.user_id
+        if name: #指定名称直接获取
+            return self.allsession.get((gid, name))
+        for k in self.allsession:
+            if gid == k[0]:
+                if not self.allsession[k].usernum_limit or uid in self.allsession[k].users:
+                    return self.allsession[k]
+        return None
+
+    def add_action(self, name: str, trigger_word: Union[str, set]):
+        """
+        用作装饰器
+        """
+        if isinstance(trigger_word, str):
+            trigger_word = (trigger_word,)
+        def deco(func: Callable) -> Callable:
+            for tw in trigger_word:
+                if tw in self.allaction[name]:
+                    raise ValueError('action trigger word duplication')
+                self.allaction[name][tw] = func
+        return deco
+
+    def add_msg_handler(self, session_name: str):
+        """
+        用作装饰器
+        """
+        def deco(func: Callable) -> Callable:
+            self.allhandler[session_name] = func
+        return deco
+
+    def join_session(self, event: Event, session: ActSession):
+        ssn = self.find_session(event)
+        if ssn: #user已经在此session或者其它session中
+            raise ValueError(f'已经在{ssn.name}中，无法再次加入或者加入其它互动')
+        ssn = session
+        ssn.add_user(event.user_id)
+
+interact = InteractHandler()

shebot/interaction_handler.py

@@ -0,0 +1,34 @@
+from nonebot.message import CanceledException, message_preprocessor
+
+from ._interact import interact, ActSession
+from hoshino import logger
+from hoshino.typing import CQEvent, HoshinoBot
+
+@message_preprocessor
+async def handler_interaction(bot: HoshinoBot, ev: CQEvent, _):
+    if interact.find_session(ev):
+        session = interact.find_session(ev)
+
+        if ev.raw_message == 'exit' and ev.user_id == session.creator: #创建者选择退出
+            session.close()
+            await session.finish(ev, f'{session.name}已经结束，欢迎下次再玩！')
+
+        if session.is_expire():
+            session.close()
+            await bot.send(ev, f'时间已到，{session.name}自动结束')
+
+        func = session.actions.get(ev.raw_message) if ev.user_id in session.users else None
+
+        if func:
+            logger.info(f'triggered interaction action {func.__name__}')
+            try:
+                await func(ev, session)
+            except CanceledException as e:
+                logger.info(e)
+            except Exception as ex:
+                logger.exception(ex)
+            raise CanceledException('handled by interact handler')
+        elif session.handle_msg:
+            await session.handle_msg(ev, session)
+        else:
+            pass

shebot/roullette/__init__.py

@@ -0,0 +1,82 @@
+from aiocqhttp.message import MessageSegment
+from hoshino.service import Service
+from hoshino.typing import HoshinoBot, CQEvent as Event
+from .._interact import interact, ActSession
+from hoshino.util import silence
+from random import randint, shuffle
+from itertools import cycle
+
+sv = Service('俄罗斯轮盘赌')
+
+@sv.on_fullmatch(('轮盘赌', '俄罗斯轮盘赌'))
+async def roulette(bot: HoshinoBot, ev: Event):
+    try:
+        session = ActSession.from_event('俄罗斯轮盘赌', ev, max_user=3, usernum_limit=True)
+        interact.add_session(session)
+        await bot.send(ev, '游戏开始,目前有1位玩家，还缺1名玩家，发送"参与轮盘赌"加入游戏')
+    except ValueError as e:
+        await bot.finish(ev, f'{e}')
+
+@sv.on_fullmatch('参与轮盘赌')
+async def join_roulette(bot: HoshinoBot, ev: Event):
+    session = interact.find_session(ev, name='俄罗斯轮盘赌')
+    if not session: #session未创建
+        await bot.send(ev, '游戏未创建，发送轮盘赌或者俄罗斯轮盘赌创建游戏')
+        return #不处理
+    try:
+        interact.join_session(ev, session)
+        await bot.send(ev, f'成功加入,目前有{session.count_user()}位玩家,发送“开始”进行游戏')
+
+    except ValueError as e:
+        await bot.finish(ev, f'{e}')
+
+@interact.add_action('俄罗斯轮盘赌', (f'{MessageSegment.face(169)}', '开枪'))
+async def fire(event: Event, session: ActSession):
+    if not session.state.get('started'):
+        await session.finish(event, '请先发送“开始”进行游戏')
+
+    if not session.pos:
+        session.state['pos'] = randint(1, 6) #拨动轮盘，pos为第几发是子弹 """
+    if not session.state.get('times'):
+        session.state['times'] = 1
+
+    if event.user_id != session.state.get('turn'):
+        await session.finish(event, '现在枪不在你手上哦~')
+
+    pos = session.pos
+    times = session.times
+    if pos == times: #shoot
+        session.close()
+        await session.send(event, '枪响了，你死了！')
+        await silence(event, 60)
+    elif times == 5:
+        session.close()
+        user = session.rotate.__next__()
+        await session.send(event, f'你长舒了一口气，并反手击毙了{MessageSegment.at(user)}')
+        await session.bot.set_group_ban(group_id=event.group_id, user_id=user, duration=60)
+    else:
+        session.state['times'] += 1
+        session.state['turn'] = session.rotate.__next__()
+        await session.send(event, f'无事发生,轮到{MessageSegment.at(session.state["turn"])}开枪')
+
+@interact.add_action('俄罗斯轮盘赌', '开始')
+async def start_roulette(event: Event, session: ActSession):
+    if session.count_user() < 2:
+        await session.finish(event, '人数不足')
+
+    if not session.state.get('started'):
+        session.state['started'] = True
+        rule = """
+        轮盘容量为6，但只填充了一发子弹，请参与游戏的双方轮流发送开枪，枪响结束
+        """.strip()
+        if not session.rotate: #user轮流
+            shuffle(session.users)
+            session.state['rotate'] = cycle(session.users)
+        if not session.turn:
+            session.state['turn'] = session.rotate.__next__()
+        await session.send(event, f'游戏开始,{rule}现在请{MessageSegment.at(session.state["turn"])}开枪')
+    else:
+        await session.send(event, '游戏已经开始了')
+
+
+