feat✨: 俄罗斯轮盘

zhenxun/plugins/russian/__init__.py

@@ -0,0 +1,186 @@
+from nonebot.adapters import Bot
+from nonebot.plugin import PluginMetadata
+from nonebot_plugin_alconna import Arparma
+from nonebot_plugin_alconna import At as alcAt
+from nonebot_plugin_alconna import Match
+from nonebot_plugin_saa import Image, Text
+from nonebot_plugin_session import EventSession
+
+from zhenxun.configs.utils import PluginExtraData, RegisterConfig
+from zhenxun.services.log import logger
+from zhenxun.utils.depends import UserName
+
+from .command import (
+    _accept_matcher,
+    _rank_matcher,
+    _record_matcher,
+    _refuse_matcher,
+    _russian_matcher,
+    _settlement_matcher,
+    _shoot_matcher,
+)
+from .data_source import Russian, russian_manage
+from .model import RussianUser
+
+__plugin_meta__ = PluginMetadata(
+    name="俄罗斯轮盘",
+    description="虽然是运气游戏，但这可是战场啊少年",
+    usage="""
+    又到了决斗时刻
+    指令：
+        装弹 [金额] [子弹数] ?[at]: 开启游戏，装填子弹，可选自定义金额，或邀请决斗对象
+        接受对决: 接受当前存在的对决
+        拒绝对决: 拒绝邀请的对决
+        开枪: 开出未知的一枪
+        结算: 强行结束当前比赛 (仅当一方未开枪超过30秒时可使用)
+        我的战绩: 对，你的战绩
+        轮盘胜场排行/轮盘败场排行/轮盘欧洲人排行/轮盘慈善家排行/轮盘最高连胜排行/轮盘最高连败排行: 各种排行榜
+        示例：装弹 3 100 @sdd
+        * 注：同一时间群内只能有一场对决 *
+    """.strip(),
+    extra=PluginExtraData(
+        author="HibiKier",
+        version="0.1",
+        menu_type="群内小游戏",
+        configs=[
+            RegisterConfig(
+                key="MAX_RUSSIAN_BET_GOLD",
+                value=1000,
+                help="俄罗斯轮盘最大赌注金额",
+                default_value=1000,
+                type=int,
+            )
+        ],
+    ).dict(),
+)
+
+
+@_russian_matcher.handle()
+async def _(money: int, num: Match[int], at_user: Match[alcAt]):
+    _russian_matcher.set_path_arg("money", money)
+    if num.available:
+        _russian_matcher.set_path_arg("num", num.result)
+    if at_user.available:
+        _russian_matcher.set_path_arg("at_user", at_user.result.target)
+
+
+@_russian_matcher.got_path("num", prompt="请输入装填子弹的数量！(最多6颗)")
+async def _(
+    bot: Bot,
+    session: EventSession,
+    arparma: Arparma,
+    money: int,
+    num: int,
+    at_user: Match[alcAt],
+    uname: str = UserName(),
+):
+    gid = session.id2
+    if not session.id1:
+        await Text("用户id为空...").finish()
+    if not gid:
+        await Text("群组id为空...").finish()
+    _at_user = at_user.result.target if at_user.available else None
+    rus = Russian(
+        at_user=_at_user, player1=(session.id1, uname), money=money, bullet_num=num
+    )
+    result = await russian_manage.add_russian(bot, gid, rus)
+    await result.send()
+    logger.info(
+        f"添加俄罗斯轮盘 装弹: {num}, 金额: {money}",
+        arparma.header_result,
+        session=session,
+    )
+
+
+@_accept_matcher.handle()
+async def _(session: EventSession, arparma: Arparma, uname: str = UserName()):
+    global a
+    gid = session.id2
+    if not session.id1:
+        await Text("用户id为空...").finish()
+    if not gid:
+        await Text("群组id为空...").finish()
+    result = russian_manage.accept(gid, session.id1, uname)
+    await result.send()
+    logger.info(f"俄罗斯轮盘接受对决", arparma.header_result, session=session)
+
+
+@_refuse_matcher.handle()
+async def _(session: EventSession, arparma: Arparma, uname: str = UserName()):
+    gid = session.id2
+    if not session.id1:
+        await Text("用户id为空...").finish()
+    if not gid:
+        await Text("群组id为空...").finish()
+    result = russian_manage.refuse(gid, session.id1, uname)
+    await result.send()
+    logger.info(f"俄罗斯轮盘拒绝对决", arparma.header_result, session=session)
+
+
+@_settlement_matcher.handle()
+async def _(session: EventSession, arparma: Arparma):
+    gid = session.id2
+    if not session.id1:
+        await Text("用户id为空...").finish()
+    if not gid:
+        await Text("群组id为空...").finish()
+    result = await russian_manage.settlement(gid, session.id1, session.platform)
+    await result.send()
+    logger.info(f"俄罗斯轮盘结算", arparma.header_result, session=session)
+
+
+@_shoot_matcher.handle()
+async def _(bot: Bot, session: EventSession, arparma: Arparma, uname: str = UserName()):
+    gid = session.id2
+    if not session.id1:
+        await Text("用户id为空...").finish()
+    if not gid:
+        await Text("群组id为空...").finish()
+    result, settle = await russian_manage.shoot(
+        bot, gid, session.id1, uname, session.platform
+    )
+    await result.send()
+    if settle:
+        await settle.send()
+    logger.info(f"俄罗斯轮盘开枪", arparma.header_result, session=session)
+
+
+@_record_matcher.handle()
+async def _(session: EventSession, arparma: Arparma):
+    gid = session.id2
+    if not session.id1:
+        await Text("用户id为空...").finish()
+    if not gid:
+        await Text("群组id为空...").finish()
+    user, _ = await RussianUser.get_or_create(user_id=session.id1, group_id=gid)
+    await Text(
+        f"俄罗斯轮盘\n"
+        f"总胜利场次：{user.win_count}\n"
+        f"当前连胜：{user.winning_streak}\n"
+        f"最高连胜：{user.max_winning_streak}\n"
+        f"总失败场次：{user.fail_count}\n"
+        f"当前连败：{user.losing_streak}\n"
+        f"最高连败：{user.max_losing_streak}\n"
+        f"赚取金币：{user.make_money}\n"
+        f"输掉金币：{user.lose_money}",
+    ).send(reply=True)
+    logger.info(f"俄罗斯轮盘查看战绩", arparma.header_result, session=session)
+
+
+@_rank_matcher.handle()
+async def _(session: EventSession, arparma: Arparma, rank_type: str, num: int):
+    gid = session.id2
+    if not session.id1:
+        await Text("用户id为空...").finish()
+    if not gid:
+        await Text("群组id为空...").finish()
+    if 51 < num or num < 10:
+        num = 10
+    result = await russian_manage.rank(session.id1, gid, rank_type, num)
+    if isinstance(result, str):
+        await Text(result).finish(reply=True)
+    result.show()
+    await Image(result.pic2bytes()).send(reply=True)
+    logger.info(
+        f"查看轮盘排行: {rank_type} 数量: {num}", arparma.header_result, session=session
+    )

zhenxun/plugins/russian/command.py

@@ -0,0 +1,108 @@
+from nonebot_plugin_alconna import Alconna, Args
+from nonebot_plugin_alconna import At as alcAt
+from nonebot_plugin_alconna import on_alconna
+
+from zhenxun.utils.rules import ensure_group
+
+_russian_matcher = on_alconna(
+    Alconna(
+        "俄罗斯轮盘",
+        Args["money", int]["num?", int]["at_user?", alcAt],
+    ),
+    aliases={"装弹"},
+    rule=ensure_group,
+    priority=5,
+    block=True,
+)
+
+_accept_matcher = on_alconna(
+    Alconna("接受对决"),
+    aliases={"接受决斗", "接受挑战"},
+    rule=ensure_group,
+    priority=5,
+    block=True,
+)
+
+_refuse_matcher = on_alconna(
+    Alconna("拒绝对决"),
+    aliases={"拒绝决斗", "拒绝挑战"},
+    rule=ensure_group,
+    priority=5,
+    block=True,
+)
+
+_shoot_matcher = on_alconna(
+    Alconna("开枪"),
+    aliases={"咔", "嘭", "嘣"},
+    rule=ensure_group,
+    priority=5,
+    block=True,
+)
+
+_settlement_matcher = on_alconna(
+    Alconna("结算"),
+    rule=ensure_group,
+    priority=5,
+    block=True,
+)
+
+_record_matcher = on_alconna(
+    Alconna("我的战绩"),
+    rule=ensure_group,
+    priority=5,
+    block=True,
+)
+
+_rank_matcher = on_alconna(
+    Alconna(
+        "russian-rank",
+        Args["rank_type", ["win", "lose", "a", "b", "max_win", "max_lose"]][
+            "num?", int, 10
+        ],
+    ),
+    rule=ensure_group,
+    priority=5,
+    block=True,
+)
+
+_rank_matcher.shortcut(
+    r"轮盘胜场排行(?P<num>\d*)",
+    command="russian-rank",
+    arguments=["win", "{num}"],
+    prefix=True,
+)
+
+_rank_matcher.shortcut(
+    r"轮盘败场排行(?P<num>\d*)",
+    command="russian-rank",
+    arguments=["lose", "{num}"],
+    prefix=True,
+)
+
+_rank_matcher.shortcut(
+    r"轮盘欧洲人排行(?P<num>\d*)",
+    command="russian-rank",
+    arguments=["a", "{num}"],
+    prefix=True,
+)
+
+_rank_matcher.shortcut(
+    r"轮盘慈善家排行(?P<num>\d*)",
+    command="russian-rank",
+    arguments=["b", "{num}"],
+    prefix=True,
+)
+
+_rank_matcher.shortcut(
+    r"轮盘最高连胜排行(?P<num>\d*)",
+    command="russian-rank",
+    arguments=["max_win", "{num}"],
+    prefix=True,
+)
+
+_rank_matcher.shortcut(
+    r"轮盘最高连败排行(?P<num>\d*)",
+    command="russian-rank",
+    arguments=["max_lose", "{num}"],
+    prefix=True,
+)

zhenxun/plugins/russian/data_source.py

@@ -0,0 +1,479 @@
+import random
+import time
+from datetime import datetime, timedelta
+
+from apscheduler.jobstores.base import JobLookupError
+from nonebot.adapters import Bot
+from nonebot_plugin_apscheduler import scheduler
+from nonebot_plugin_saa import Image, Mention, MessageFactory, Text
+from pydantic import BaseModel
+
+from zhenxun.configs.config import NICKNAME, Config
+from zhenxun.models.group_member_info import GroupInfoUser
+from zhenxun.models.user_console import UserConsole
+from zhenxun.utils.enum import GoldHandle
+from zhenxun.utils.image_utils import BuildImage, BuildMat, MatType, text2image
+from zhenxun.utils.platform import PlatformUtils
+
+from .model import RussianUser
+
+base_config = Config.get("russian")
+
+
+class Russian(BaseModel):
+
+    at_user: str | None
+    """指定决斗对象"""
+    player1: tuple[str, str]
+    """玩家1id, 昵称"""
+    player2: tuple[str, str] | None = None
+    """玩家2id, 昵称"""
+    money: int
+    """金额"""
+    bullet_num: int
+    """子弹数"""
+    bullet_arr: list[int] = []
+    """子弹排列"""
+    bullet_index: int = 0
+    """当前子弹下标"""
+    next_user: str = ""
+    """下一个开枪用户"""
+    time: float = time.time()
+    """创建时间"""
+    win_user: str | None = None
+    """胜利者"""
+
+
+class RussianManage:
+
+    def __init__(self) -> None:
+        self._data: dict[str, Russian] = {}
+
+    def __check_is_timeout(self, group_id: str) -> bool:
+        """检查决斗是否超时
+
+        参数:
+            group_id: 群组id
+
+        返回:
+            bool: 是否超时
+        """
+        if russian := self._data.get(group_id):
+            if russian.time + 30 < time.time():
+                return True
+        return False
+
+    def __random_bullet(self, num: int) -> list[int]:
+        """随机排列子弹
+
+        参数:
+            num: 子弹数量
+
+        返回:
+            list[int]: 子弹排列数组
+        """
+        bullet_list = [0, 0, 0, 0, 0, 0, 0]
+        for i in random.sample([0, 1, 2, 3, 4, 5, 6], num):
+            bullet_list[i] = 1
+        return bullet_list
+
+    def __build_job(
+        self, bot: Bot, group_id: str, is_add: bool = False, platform: str | None = None
+    ):
+        """移除定时任务和构建新定时任务
+
+        参数:
+            bot: Bot
+            group_id: 群组id
+            is_add: 是否添加新定时任务.
+            platform: 平台
+        """
+        try:
+            scheduler.remove_job(f"russian_job_{group_id}")
+        except JobLookupError:
+            pass
+        if is_add:
+            date = datetime.now() + timedelta(seconds=31)
+            scheduler.add_job(
+                self.__auto_end_game,
+                "date",
+                run_date=date.replace(microsecond=0),
+                id=f"russian_job_{group_id}",
+                args=[bot, group_id, platform],
+            )
+
+    async def __auto_end_game(self, bot: Bot, group_id: str, platform: str):
+        """自动结束对决
+
+        参数:
+            bot: Bot
+            group_id: 群组id
+            platform: 平台
+        """
+        result = await self.settlement(group_id, None, platform)
+        if result:
+            await PlatformUtils.send_message(bot, None, group_id, result)
+
+    async def add_russian(
+        self, bot: Bot, group_id: str, rus: Russian
+    ) -> Text | MessageFactory:
+        """添加决斗
+
+        参数:
+            bot: Bot
+            group_id: 群组id
+            rus: Russian
+
+        返回:
+            Text | MessageFactory: 返回消息
+        """
+        russian = self._data.get(group_id)
+        if russian:
+            if russian.time + 30 < time.time():
+                if not russian.player2:
+                    return Text(
+                        f"现在是 {russian.player1[1]} 发起的对决, 请接受对决或等待决斗超时..."
+                    )
+                else:
+                    return Text(
+                        f"{russian.player1[1]} 和 {russian.player2[1]}的对决还未结束！"
+                    )
+            return Text(
+                f"现在是 {russian.player1[1]} 发起的对决\n请等待比赛结束后再开始下一轮..."
+            )
+        max_money = base_config.get("MAX_RUSSIAN_BET_GOLD")
+        if rus.money > max_money:
+            return Text(f"太多了！单次金额不能超过{max_money}！")
+        user = await UserConsole.get_user(rus.player1[0])
+        if user.gold < rus.money:
+            return Text("你没有足够的钱支撑起这场挑战")
+        rus.bullet_arr = self.__random_bullet(rus.bullet_num)
+        self._data[group_id] = rus
+        message_list = []
+        if rus.at_user:
+            user = await GroupInfoUser.get_or_none(
+                user_id=rus.at_user, group_id=group_id
+            )
+            message_list = [
+                Text(f"{rus.player1[1]} 向"),
+                Mention(rus.at_user),
+                Text(
+                    f"发起了决斗！请 {user.user_name if user else rus.at_user} 在30秒内回复‘接受对决’ or ‘拒绝对决’，超时此次决斗作废！"
+                ),
+            ]
+        else:
+            message_list = [
+                Text(
+                    "若30秒内无人接受挑战则此次对决作废【首次游玩请发送 ’俄罗斯轮盘帮助‘ 来查看命令】"
+                )
+            ]
+        result = Text(
+            "咔 " * rus.bullet_num
+            + f"装填完毕\n挑战金额：{rus.money}\n第一枪的概率为：{float(rus.bullet_num) / 7.0 * 100:.2f}%\n"
+        )
+        message_list.insert(0, result)
+        self.__build_job(bot, group_id, True)
+        return MessageFactory(message_list)
+
+    def accept(self, group_id: str, user_id: str, uname: str) -> Text | MessageFactory:
+        """接受对决
+
+        参数:
+            group_id: 群组id
+            user_id: 用户id
+            uname: 用户名称
+
+        返回:
+            Text | MessageFactory: 返回消息
+        """
+        if russian := self._data.get(group_id):
+            if russian.at_user and russian.at_user != user_id:
+                return Text("又不是找你决斗，你接受什么啊！气！")
+            if russian.player2:
+                return Text("当前决斗已被其他玩家接受！请等待下局对决！")
+            russian.player2 = (user_id, uname)
+            russian.next_user = russian.player1[0]
+            return MessageFactory(
+                [
+                    Text("决斗已经开始！请"),
+                    Mention(russian.player1[0]),
+                    Text("先开枪！"),
+                ]
+            )
+        return Text("目前没有进行的决斗，请发送 装弹 开启决斗吧！")
+
+    def refuse(self, group_id: str, user_id: str, uname: str) -> Text | MessageFactory:
+        """拒绝决斗
+
+        参数:
+            group_id: 群组id
+            user_id: 用户id
+            uname: 用户名称
+
+        返回:
+            Text | MessageFactory: 返回消息
+        """
+        if russian := self._data.get(group_id):
+            if russian.at_user:
+                if russian.at_user != user_id:
+                    return Text("又不是找你决斗，你拒绝什么啊！气！")
+                del self._data[group_id]
+                return MessageFactory(
+                    [Mention(russian.player1[0]), Text(f"{uname}拒绝了你的对决！")]
+                )
+            return Text("当前决斗并没有指定对手，无法拒绝哦！")
+        return Text("目前没有进行的决斗，请发送 装弹 开启决斗吧！")
+
+    async def shoot(
+        self, bot: Bot, group_id: str, user_id: str, uname: str, platform: str
+    ) -> tuple[Text | MessageFactory, Text | MessageFactory | None]:
+        """开枪
+
+        参数:
+            bot: Bot
+            group_id: 群组id
+            user_id: 用户id
+            uname: 用户名称
+            platform: 平台
+
+        返回:
+            Text | MessageFactory: 返回消息
+        """
+        if russian := self._data.get(group_id):
+            if not russian.player2:
+                return Text("当前还没有玩家接受对决，无法开枪..."), None
+            if user_id not in [russian.player1[0], russian.player2[0]]:
+                """非玩家1和玩家2发送开枪"""
+                return (
+                    Text(
+                        random.choice(
+                            [
+                                f"不要打扰 {russian.player1[1]} 和 {russian.player2[1]} 的决斗啊！",
+                                f"给我好好做好一个观众！不然{NICKNAME}就要生气了",
+                                f"不要捣乱啊baka{uname}！",
+                            ]
+                        )
+                    ),
+                    None,
+                )
+            if user_id != russian.next_user:
+                """相同玩家连续开枪"""
+                return (
+                    Text(f"你的左轮不是连发的！该 {russian.player2[1]} 开枪了!"),
+                    None,
+                )
+            if russian.bullet_arr[russian.bullet_index] == 1:
+                """去世"""
+                result = Text(
+                    random.choice(
+                        [
+                            '"嘭！"，你直接去世了',
+                            "眼前一黑，你直接穿越到了异世界...(死亡)",
+                            "终究还是你先走一步...",
+                        ]
+                    )
+                )
+                settle = await self.settlement(group_id, user_id, platform)
+                return result, settle
+            else:
+                """存活"""
+                p = (russian.bullet_index + 1) / len(russian.bullet_arr) * 100
+                result = (
+                    random.choice(
+                        [
+                            "呼呼，没有爆裂的声响，你活了下来",
+                            "虽然黑洞洞的枪口很恐怖，但好在没有子弹射出来，你活下来了",
+                            '"咔"，你没死，看来运气不错',
+                        ]
+                    )
+                    + f"\n下一枪中弹的概率: {p:.2f}%, 轮到 "
+                )
+                next_user = (
+                    russian.player2[0]
+                    if russian.next_user == russian.player1[0]
+                    else russian.player1[0]
+                )
+                russian.next_user = next_user
+                russian.bullet_index += 1
+                self.__build_job(bot, group_id, True)
+                return (
+                    MessageFactory([Text(result), Mention(next_user), Text(" 了!")]),
+                    None,
+                )
+        return Text("目前没有进行的决斗，请发送 装弹 开启决斗吧！"), None
+
+    async def settlement(
+        self, group_id: str, user_id: str | None, platform: str | None = None
+    ) -> Text | MessageFactory:
+        """结算
+
+        参数:
+            group_id: 群组id
+            user_id: 用户id
+            platform: 平台
+
+        返回:
+            Text | MessageFactory: 返回消息
+        """
+        if russian := self._data.get(group_id):
+            if not russian.player2:
+                if self.__check_is_timeout(group_id):
+                    del self._data[group_id]
+                    return Text("规定时间内还未有人接受决斗，当前决斗过期...")
+                return Text("决斗还未开始,，无法结算哦...")
+            if user_id and user_id not in [russian.player1[0], russian.player1[0]]:
+                return Text("吃瓜群众不要捣乱！黄牌警告！")
+            if not self.__check_is_timeout(group_id):
+                return Text(
+                    f"{russian.player1[1]} 和 {russian.player1[1]} 比赛并未超时，请继续比赛..."
+                )
+            win_user = None
+            lose_user = None
+            if win_user:
+                russian.next_user = (
+                    russian.player1[0]
+                    if win_user == russian.player2[0]
+                    else russian.player2[0]
+                )
+            if russian.next_user != russian.player1[0]:
+                win_user = russian.player1
+                lose_user = russian.player2
+            else:
+                win_user = russian.player2
+                lose_user = russian.player1
+            if win_user and lose_user:
+                rand = 0
+                if russian.money > 10:
+                    rand = random.randint(0, 5)
+                    fee = int(russian.money * float(rand) / 100)
+                    fee = 1 if fee < 1 and rand != 0 else fee
+                else:
+                    fee = 0
+                winner = await RussianUser.add_count(win_user[0], group_id, "win")
+                loser = await RussianUser.add_count(lose_user[0], group_id, "lose")
+                await RussianUser.money(
+                    win_user[0], group_id, "win", russian.money - fee
+                )
+                await RussianUser.money(lose_user[0], group_id, "lose", russian.money)
+                await UserConsole.add_gold(
+                    win_user[0], russian.money - fee, "russian", platform
+                )
+                await UserConsole.reduce_gold(
+                    lose_user[0], russian.money, GoldHandle.PLUGIN, "russian", platform
+                )
+                result = [Text("这场决斗是 "), Mention(win_user[0]), Text(" 胜利了!")]
+                image = await text2image(
+                    f"结算：\n"
+                    f"\t胜者：{win_user[1]}\n"
+                    f"\t赢取金币：{russian.money - fee}\n"
+                    f"\t累计胜场：{winner.win_count}\n"
+                    f"\t累计赚取金币：{winner.make_money}\n"
+                    f"-------------------\n"
+                    f"\t败者：{lose_user[1]}\n"
+                    f"\t输掉金币：{russian.money}\n"
+                    f"\t累计败场：{loser.fail_count}\n"
+                    f"\t累计输掉金币：{loser.lose_money}\n"
+                    f"-------------------\n"
+                    f"哼哼，{NICKNAME}从中收取了 {float(rand)}%({fee}金币) 作为手续费！\n"
+                    f"子弹排列：{russian.bullet_arr}",
+                    padding=10,
+                    color="#f9f6f2",
+                )
+                result.append(Image(image.pic2bytes()))
+                del self._data[group_id]
+                return MessageFactory(result)
+            return Text("赢家和输家获取错误...")
+        return Text("比赛并没有开始...无法结算...")
+
+    async def __get_x_index(self, users: list[RussianUser], group_id: str):
+        uid_list = [u.user_id for u in users]
+        group_user_list = await GroupInfoUser.filter(
+            user_id__in=uid_list, group_id=group_id
+        ).all()
+        group_user = {gu.user_id: gu.user_name for gu in group_user_list}
+        data = []
+        for uid in uid_list:
+            if uid in group_user:
+                data.append(group_user[uid])
+            else:
+                data.append(uid)
+        return data
+
+    async def rank(
+        self, user_id: str, group_id: str, rank_type: str, num: int
+    ) -> BuildImage | str:
+        x_index = []
+        data = []
+        title = ""
+        x_name = ""
+        if rank_type == "win":
+            users = (
+                await RussianUser.filter(group_id=group_id, win_count__not=0)
+                .order_by("win_count")
+                .limit(num)
+            )
+            x_index = await self.__get_x_index(users, group_id)
+            data = [u.win_count for u in users]
+            title = "胜场排行"
+            x_name = "场次"
+        if rank_type == "lose":
+            users = (
+                await RussianUser.filter(group_id=group_id, fail_count__not=0)
+                .order_by("fail_count")
+                .limit(num)
+            )
+            x_index = await self.__get_x_index(users, group_id)
+            data = [u.fail_count for u in users]
+            title = "败场排行"
+            x_name = "场次"
+        if rank_type == "a":
+            users = (
+                await RussianUser.filter(group_id=group_id, make_money__not=0)
+                .order_by("make_money")
+                .limit(num)
+            )
+            x_index = await self.__get_x_index(users, group_id)
+            data = [u.make_money for u in users]
+            title = "欧洲人排行"
+            x_name = "金币"
+        if rank_type == "b":
+            users = (
+                await RussianUser.filter(group_id=group_id, lose_money__not=0)
+                .order_by("lose_money")
+                .limit(num)
+            )
+            x_index = await self.__get_x_index(users, group_id)
+            data = [u.lose_money for u in users]
+            title = "慈善家排行"
+            x_name = "金币"
+        if rank_type == "max_win":
+            users = (
+                await RussianUser.filter(group_id=group_id, max_winning_streak__not=0)
+                .order_by("max_winning_streak")
+                .limit(num)
+            )
+            x_index = await self.__get_x_index(users, group_id)
+            data = [u.max_winning_streak for u in users]
+            title = "最高连胜排行"
+            x_name = "场次"
+        if rank_type == "max_lose":
+            users = (
+                await RussianUser.filter(group_id=group_id, max_losing_streak__not=0)
+                .order_by("max_losing_streak")
+                .limit(num)
+            )
+            x_index = await self.__get_x_index(users, group_id)
+            data = [u.max_losing_streak for u in users]
+            title = "最高连败排行"
+            x_name = "场次"
+        if not data:
+            return "当前数据为空..."
+        mat = BuildMat(MatType.BARH)
+        mat.x_index = x_index
+        mat.data = data  # type: ignore
+        mat.title = title
+        mat.x_name = x_name
+        return await mat.build()
+
+
+russian_manage = RussianManage()

zhenxun/plugins/russian/model.py

@@ -0,0 +1,107 @@
+from tortoise import fields
+
+from zhenxun.services.db_context import Model
+
+
+class RussianUser(Model):
+
+    id = fields.IntField(pk=True, generated=True, auto_increment=True)
+    """自增id"""
+    user_id = fields.CharField(255)
+    """用户id"""
+    group_id = fields.CharField(255)
+    """群聊id"""
+    win_count = fields.IntField(default=0)
+    """胜利次数"""
+    fail_count = fields.IntField(default=0)
+    """失败次数"""
+    make_money = fields.IntField(default=0)
+    """赢得金币"""
+    lose_money = fields.IntField(default=0)
+    """输得金币"""
+    winning_streak = fields.IntField(default=0)
+    """当前连胜"""
+    losing_streak = fields.IntField(default=0)
+    """当前连败"""
+    max_winning_streak = fields.IntField(default=0)
+    """最大连胜"""
+    max_losing_streak = fields.IntField(default=0)
+    """最大连败"""
+
+    class Meta:
+        table = "russian_users"
+        table_description = "俄罗斯轮盘数据表"
+        unique_together = ("user_id", "group_id")
+
+    @classmethod
+    async def add_count(cls, user_id: str, group_id: str, itype: str):
+        """添加用户输赢次数
+
+        说明:
+            user_id: 用户id
+            group_id: 群号
+            itype: 输或赢 'win' or 'lose'
+        """
+        user, _ = await cls.get_or_create(user_id=user_id, group_id=group_id)
+        if itype == "win":
+            _max = (
+                user.max_winning_streak
+                if user.max_winning_streak > user.winning_streak + 1
+                else user.winning_streak + 1
+            )
+            user.win_count = user.win_count + 1
+            user.winning_streak = user.winning_streak + 1
+            user.losing_streak = 0
+            user.max_winning_streak = _max
+            await user.save(
+                update_fields=[
+                    "win_count",
+                    "winning_streak",
+                    "losing_streak",
+                    "max_winning_streak",
+                ]
+            )
+        elif itype == "lose":
+            _max = (
+                user.max_losing_streak
+                if user.max_losing_streak > user.losing_streak + 1
+                else user.losing_streak + 1
+            )
+            user.fail_count = user.fail_count + 1
+            user.losing_streak = user.losing_streak + 1
+            user.winning_streak = 0
+            user.max_losing_streak = _max
+            await user.save(
+                update_fields=[
+                    "fail_count",
+                    "winning_streak",
+                    "losing_streak",
+                    "max_losing_streak",
+                ]
+            )
+        return user
+
+    @classmethod
+    async def money(cls, user_id: str, group_id: str, itype: str, count: int):
+        """添加用户输赢金钱
+
+        参数:
+            user_id: 用户id
+            group_id: 群号
+            itype: 输或赢 'win' or 'lose'
+            count: 金钱数量
+        """
+        user, _ = await cls.get_or_create(user_id=str(user_id), group_id=group_id)
+        if itype == "win":
+            user.make_money = user.make_money + count
+        elif itype == "lose":
+            user.lose_money = user.lose_money + count
+        await user.save(update_fields=["make_money", "lose_money"])
+
+    @classmethod
+    async def _run_script(cls):
+        return [
+            "ALTER TABLE russian_users RENAME COLUMN user_qq TO user_id;",  # 将user_qq改为user_id
+            "ALTER TABLE russian_users ALTER COLUMN user_id TYPE character varying(255);",
+            "ALTER TABLE russian_users ALTER COLUMN group_id TYPE character varying(255);",
+        ]

zhenxun/utils/_build_mat.py

@@ -556,7 +556,7 @@ class BuildMat:
         random_color = random.choice(bar_color)
         max_num = max(self.y_index)
         for y_p, y in zip(init_graph.y_point, self.build_data.data):
-            bar_width = int(y / max_num * graph_height)
+            bar_width = int(y / max_num * graph_height) or 1
             bar = BuildImage(bar_width, 18, random_color)
             await mark_image.paste(bar, (y_width + 1, y_p - 9))
             if self.build_data.display_num: