diff options
Diffstat (limited to 'ATRI/plugins')
| -rw-r--r-- | ATRI/plugins/bot.py | 159 | ||||
| -rw-r--r-- | ATRI/plugins/character.py | 219 | ||||
| -rw-r--r-- | ATRI/plugins/curse.py | 55 | ||||
| -rw-r--r-- | ATRI/plugins/genshin/__init__.py | 53 | ||||
| -rw-r--r-- | ATRI/plugins/genshin/data_source.py | 121 | ||||
| -rw-r--r-- | ATRI/plugins/hitokoto.py | 76 | ||||
| -rw-r--r-- | ATRI/plugins/setu/__init__.py | 29 | ||||
| -rw-r--r-- | ATRI/plugins/setu/data_source.py | 14 | ||||
| -rw-r--r-- | ATRI/plugins/utils/__init__.py | 76 | ||||
| -rw-r--r-- | ATRI/plugins/utils/data_source.py | 239 | ||||
| -rw-r--r-- | ATRI/plugins/utils/main.bin | bin | 0 -> 3992353 bytes |
11 files changed, 708 insertions, 333 deletions
diff --git a/ATRI/plugins/bot.py b/ATRI/plugins/bot.py deleted file mode 100644 index 42e05d0..0000000 --- a/ATRI/plugins/bot.py +++ /dev/null @@ -1,159 +0,0 @@ -from random import choice -from typing import Optional - -from nonebot.typing import Bot, Event -from nonebot.plugin import on_command, on_regex - -from ATRI.rule import isInBanList, isInDormant, toBot -from ATRI.utils import countList, delListAim, nowTime -from ATRI.config import BOT_CONFIG, NONEBOT_CONFIG - -# ===========================[Begin Command Processing]=========================== - - -class Action: - callRobot = on_regex('|'.join(BOT_CONFIG['callRobot']['command']), - rule=isInBanList() & isInDormant(), - priority=4) - - callMe = on_command(list(NONEBOT_CONFIG['nickname'])[0], - aliases=NONEBOT_CONFIG['nickname'], - rule=isInBanList() & isInDormant()) - - teeTee = on_command(BOT_CONFIG['teetee']['command'][0], - aliases=set(BOT_CONFIG['teetee']['command']), - rule=isInBanList() & isInDormant() & toBot()) - - kani = on_regex('|'.join(BOT_CONFIG['kani']['command']), - rule=isInBanList() & isInDormant()) - - waste = on_regex('|'.join(BOT_CONFIG['waste']['command']), - rule=isInBanList() & isInDormant() & toBot(), - priority=5) - - morning = on_command(BOT_CONFIG['morning']['command'][0], - aliases=set(BOT_CONFIG['morning']['command']), - rule=isInBanList() & isInDormant() & toBot()) - - noon = on_command(BOT_CONFIG['noon']['command'][0], - aliases=set(BOT_CONFIG['noon']['command']), - rule=isInBanList() & isInDormant() & toBot()) - - night = on_command(BOT_CONFIG['night']['command'][0], - aliases=set(BOT_CONFIG['night']['command']), - rule=isInBanList() & isInDormant() & toBot()) - - cantdo = on_regex('|'.join(BOT_CONFIG['cantdo']['command']), - rule=isInBanList() & isInDormant()) - - @callRobot.handle() - async def _(bot: Bot, event: Event, state: dict) -> None: - await bot.send(event, ATRI()._callRobot(event.user_id)) - - @callMe.handle() - async def _(bot: Bot, event: Event, state: dict) -> None: - await bot.send(event, ATRI()._callMe()) - - @teeTee.handle() - async def _(bot: Bot, event: Event, state: dict) -> None: - await bot.send(event, ATRI()._teeTee(event.user_id)) - - @kani.handle() - async def _(bot: Bot, event: Event, state: dict) -> None: - await bot.send(event, ATRI()._kani()) - - @waste.handle() - async def _(bot: Bot, event: Event, state: dict) -> None: - await bot.send(event, ATRI()._waste()) - - @morning.handle() - async def _(bot: Bot, event: Event, state: dict) -> None: - await bot.send(event, ATRI()._morning()) - - @noon.handle() - async def _(bot: Bot, event: Event, state: dict) -> None: - await bot.send(event, ATRI()._noon()) - - @night.handle() - async def _(bot: Bot, event: Event, state: dict) -> None: - await bot.send(event, ATRI()._night()) - - @cantdo.handle() - async def _(bot: Bot, event: Event, state: dict) -> None: - await bot.send(event, ATRI()._cantdo()) - - -# ===========================[End Command Processing]=========================== - -callrobot_list = [] - - -class ATRI(): - def _callRobot(self, user: Optional[int]) -> str: - global callrobot_list - result = '' - for i in range(0, 5): - if countList(callrobot_list, user) == i: - result = choice(BOT_CONFIG['callRobot']['repo'][i]) - callrobot_list.append(user) - if countList(callrobot_list, user) == 5: - callrobot_list = delListAim(callrobot_list, user) - break - else: - continue - return result - - def _callMe(self) -> str: - return choice(BOT_CONFIG['atri']['repo']) - - def _teeTee(self, user: Optional[int]) -> str: - if user in NONEBOT_CONFIG['superusers']: - return choice(BOT_CONFIG['teetee']['repo']['superusers']) - else: - return choice(BOT_CONFIG['teetee']['repo']['user']) - - def _kani(self) -> str: - return choice(BOT_CONFIG['kani']['repo']) - - def _waste(self) -> str: - return choice(BOT_CONFIG['waste']['repo']) - - def _morning(self) -> str: - period = BOT_CONFIG['morning']['repo'] - if period[0]['period'][0] <= nowTime() < period[0]['period'][1]: - return choice(period[0]['repo']) - elif period[1]['period'][0] <= nowTime() < period[1]['period'][1]: - return choice(period[1]['repo']) - elif period[2]['period'][0] <= nowTime() < period[2]['period'][1]: - return choice(period[2]['repo']) - elif period[3]['period'][0] <= nowTime() < period[3]['period'][1]: - return choice(period[3]['repo']) - elif period[4]['period'][0] <= nowTime() < period[4]['period'][1]: - return choice(period[4]['repo']) - else: - return choice(period['error']) - - def _noon(self) -> str: - if BOT_CONFIG['noon']['period'][0] <= nowTime( - ) < BOT_CONFIG['noon']['period'][1]: - return choice(BOT_CONFIG['noon']['repo']) - else: - return choice(BOT_CONFIG['noon']['error']) - - def _night(self) -> str: - period = BOT_CONFIG['night']['repo'] - if period[0]['period'][0] <= nowTime() < period[0]['period'][1]: - return choice(period[0]['repo']) - elif period[1]['period'][0] <= nowTime() < period[1]['period'][1]: - return choice(period[1]['repo']) - elif period[2]['period'][0] <= nowTime() < period[2]['period'][1]: - return choice(period[2]['repo']) - elif period[3]['period'][0] <= nowTime() < period[3]['period'][1]: - return choice(period[3]['repo']) - elif period[4]['period'][0] <= nowTime() < period[4]['period'][1]: - return choice(period[4]['repo']) - else: - return choice(period['error']) - - def _cantdo(self) -> str: - return choice(BOT_CONFIG['cantdo']['repo'])
\ No newline at end of file diff --git a/ATRI/plugins/character.py b/ATRI/plugins/character.py new file mode 100644 index 0000000..93dbc02 --- /dev/null +++ b/ATRI/plugins/character.py @@ -0,0 +1,219 @@ +from random import choice +from typing import Optional + +from nonebot.plugin import on_command, on_regex +from nonebot.adapters.cqhttp import Bot, Event + +from ATRI.rule import is_in_ban_list, is_in_dormant, to_bot +from ATRI.utils import count_list, del_list_aim, now_time +from ATRI.config import BOT_CONFIG, RUNTIME_CONFIG +from ATRI.service.plugin import Plugin + +# ===========================[Begin Command Processing]=========================== + +__plugin_name__ = 'call_robot' +__doc__ = """使用正则触发""" +Plugin.register(__plugin_name__, "func", __doc__, + command=BOT_CONFIG['callRobot']['command']) + +call_robot = on_regex('|'.join(BOT_CONFIG['callRobot']['command']), + rule=is_in_ban_list() & is_in_dormant(), + priority=4) + +@call_robot.handle() +async def _(bot: Bot, event: Event) -> None: + await bot.send(event, ATRI().call_robot(int(event.get_user_id()))) + + +__plugin_name__ = 'call_me' +__doc__ = """正确地呼叫咱""" +Plugin.register(__plugin_name__, "func", __doc__, + RUNTIME_CONFIG['nickname']) + +call_me = on_command(list(RUNTIME_CONFIG['nickname'])[0], + aliases=RUNTIME_CONFIG['nickname'], + rule=is_in_ban_list() & is_in_dormant()) + +@call_me.handle() +async def _(bot: Bot, event: Event) -> None: + await bot.send(event, ATRI().call_me()) + + +__plugin_name__ = 'tee_tee' +__doc__ = """一般人员不许贴！ +需at""" +Plugin.register(__plugin_name__, "func", __doc__, + BOT_CONFIG['teetee']['command']) + +teeTee = on_command(BOT_CONFIG['teetee']['command'][0], + aliases=set(BOT_CONFIG['teetee']['command']), + rule=is_in_ban_list() & is_in_dormant() & to_bot()) + +async def _(bot: Bot, event: Event) -> None: + await bot.send(event, ATRI().tee_tee(int(event.get_user_id()))) + + + +__plugin_name__ = 'kani' +__doc__ = """カニ！カニ！！！ +使用正则匹配""" +Plugin.register(__plugin_name__, "func", __doc__, + BOT_CONFIG['kani']['command']) + +kani = on_regex('|'.join(BOT_CONFIG['kani']['command']), + rule=is_in_ban_list() & is_in_dormant()) + +async def _(bot: Bot, event: Event) -> None: + await bot.send(event, ATRI().kani()) + + +__plugin_name__ = 'waste' +__doc__ = """不准骂咱废物 +使用正则匹配，需at""" +Plugin.register(__plugin_name__, "func", __doc__, + BOT_CONFIG['waste']['command']) + +waste = on_regex('|'.join(BOT_CONFIG['waste']['command']), + rule=is_in_ban_list() & is_in_dormant() & to_bot(), + priority=5) + +async def _(bot: Bot, event: Event) -> None: + await bot.send(event, ATRI().waste()) + + +__plugin_name__ = 'good_morning' +__doc__ = """略带涩气的早安 +需at""" +Plugin.register(__plugin_name__, "func", __doc__, + BOT_CONFIG['morning']['command']) + +morning = on_command(BOT_CONFIG['morning']['command'][0], + aliases=set(BOT_CONFIG['morning']['command']), + rule=is_in_ban_list() & is_in_dormant() & to_bot()) + +async def _(bot: Bot, event: Event) -> None: + await bot.send(event, ATRI().morning()) + + +__plugin_name__ = 'good_noon' +__doc__ = """做白日梦 +需at""" +Plugin.register(__plugin_name__, "func", __doc__, + BOT_CONFIG['noon']['command']) + +noon = on_command(BOT_CONFIG['noon']['command'][0], + aliases=set(BOT_CONFIG['noon']['command']), + rule=is_in_ban_list() & is_in_dormant() & to_bot()) + +async def _(bot: Bot, event: Event) -> None: + await bot.send(event, ATRI().noon()) + + +__plugin_name__ = 'good_night' +__doc__ = """晚安~！ +需at""" +Plugin.register(__plugin_name__, "func", __doc__, + BOT_CONFIG['night']['command']) + +night = on_command(BOT_CONFIG['night']['command'][0], + aliases=set(BOT_CONFIG['night']['command']), + rule=is_in_ban_list() & is_in_dormant() & to_bot()) + +async def _(bot: Bot, event: Event) -> None: + await bot.send(event, ATRI().night()) + + +__plugin_name__ = 'cant_do_anything' +__doc__ = """吃饭第一名！好吃就是高兴！！ +使用正则匹配""" +Plugin.register(__plugin_name__, "func", __doc__, + BOT_CONFIG['cantdo']['command']) + +cantdo = on_regex('|'.join(BOT_CONFIG['cantdo']['command']), + rule=is_in_ban_list() & is_in_dormant()) + +async def _(bot: Bot, event: Event) -> None: + await bot.send(event, ATRI().cantdo()) + + +# ===========================[End Command Processing]============================= + +callrobot_list = [] + + +class ATRI: + def call_robot(self, user: Optional[int]) -> str: + global callrobot_list + result = '' + for i in range(0, 5): + if count_list(callrobot_list, user) == i: + result = choice(BOT_CONFIG['callRobot']['repo'][i]) + callrobot_list.append(user) + if count_list(callrobot_list, user) == 5: + callrobot_list = del_list_aim(callrobot_list, user) + break + else: + continue + return result + + def call_me(self) -> str: + return choice(BOT_CONFIG['atri']['repo']) + + def tee_tee(self, user: Optional[int]) -> str: + if user in RUNTIME_CONFIG['superusers']: + return choice(BOT_CONFIG['teetee']['repo']['superusers']) + else: + return choice(BOT_CONFIG['teetee']['repo']['user']) + + def kani(self) -> str: + return choice(BOT_CONFIG['kani']['repo']) + + def waste(self) -> str: + return choice(BOT_CONFIG['waste']['repo']) + + def morning(self) -> str: + period = BOT_CONFIG['morning']['repo'] + if period[0]['period'][0] <= now_time() < period[0]['period'][1]: + return choice(period[0]['repo']) + elif period[1]['period'][0] <= now_time() < period[1]['period'][1]: + return choice(period[1]['repo']) + elif period[2]['period'][0] <= now_time() < period[2]['period'][1]: + return choice(period[2]['repo']) + elif period[3]['period'][0] <= now_time() < period[3]['period'][1]: + return choice(period[3]['repo']) + elif period[4]['period'][0] <= now_time() < period[4]['period'][1]: + return choice(period[4]['repo']) + else: + return choice(period['error']) + + def noon(self) -> str: + if BOT_CONFIG['noon']['period'][0] <= now_time( + ) < BOT_CONFIG['noon']['period'][1]: + return choice(BOT_CONFIG['noon']['repo']) + else: + return choice(BOT_CONFIG['noon']['error']) + + def night(self) -> str: + period = BOT_CONFIG['night']['repo'] + if period[0]['period'][0] <= now_time() < period[0]['period'][1]: + return choice(period[0]['repo']) + elif period[1]['period'][0] <= now_time() < period[1]['period'][1]: + return choice(period[1]['repo']) + elif period[2]['period'][0] <= now_time() < period[2]['period'][1]: + return choice(period[2]['repo']) + elif period[3]['period'][0] <= now_time() < period[3]['period'][1]: + return choice(period[3]['repo']) + elif period[4]['period'][0] <= now_time() < period[4]['period'][1]: + return choice(period[4]['repo']) + else: + return choice(period['error']) + + def cantdo(self) -> str: + return choice(BOT_CONFIG['cantdo']['repo']) diff --git a/ATRI/plugins/curse.py b/ATRI/plugins/curse.py index e69de29..2fa4986 100644 --- a/ATRI/plugins/curse.py +++ b/ATRI/plugins/curse.py @@ -0,0 +1,55 @@ +from random import choice +from nonebot.plugin import on_command +from nonebot.adapters.cqhttp import Bot + +from ATRI.request import Request +from ATRI.utils import count_list, del_list_aim +from ATRI.config import CURSE_CONFIG +from ATRI.exceptions import InvalidRequest +from ATRI.service.plugin import Plugin +from ATRI.rule import is_in_ban_list, is_in_service, is_in_dormant, to_bot + + + +# ===========================[Begin Command Processing]=========================== + + +__plugin_name__ = 'curse' +__doc__ = """口臭一下""" +Plugin.register(__plugin_name__, "func", __doc__, + CURSE_CONFIG['curse']['command']) + +curse = on_command(CURSE_CONFIG['curse']['command'][0], + aliases=set(CURSE_CONFIG['curse']['command']), + rule=is_in_ban_list() & is_in_dormant() + & is_in_service(__plugin_name__) + & to_bot()) + +async def _(bot: Bot, event) -> None: + await bot.send( + event, await Function().curse(str(event.get_user_id()))) + + +# ===========================[End Command Processing]============================= + +curse_list = [] + + +class Function: + async def curse(self, user: str): + global curse_list + + if count_list(curse_list, user) == 3: + curse_list.append(user) + return choice(CURSE_CONFIG['curse']['times'][3]['repo']) + elif count_list(curse_list, user) == 6: + curse_list = del_list_aim(curse_list, user) + return choice(CURSE_CONFIG['curse']['times'][6]['repo']) + else: + try: + curse_list.append(user) + return str(await Request.get_text( + url=CURSE_CONFIG['curse']['url'])) + except InvalidRequest: + raise InvalidRequest('请求失败') diff --git a/ATRI/plugins/genshin/__init__.py b/ATRI/plugins/genshin/__init__.py deleted file mode 100644 index 4705e0e..0000000 --- a/ATRI/plugins/genshin/__init__.py +++ /dev/null @@ -1,53 +0,0 @@ -import re -from nonebot.plugin import on_command -from nonebot.typing import Bot, Event - -from ATRI.config import GENSHIN_CONFIG -from ATRI.exceptions import InvalidRequestError -from ATRI.rule import isInBanList, isInDormant, isInService - -from .data_source import Genshin - -__plugin_name__ = 'genshin' - -genshin = on_command(GENSHIN_CONFIG['genshin']['command'][0], - aliases=set(GENSHIN_CONFIG['genshin']['command']), - rule=isInBanList() & isInDormant() - & isInService(__plugin_name__)) - - -async def _(bot: Bot, event: Event, state: dict) -> None: - args = str(event.message) - - if args: - state['args'] = args - [email protected]('args', prompt='请告诉咱id以及是官服还是b服嗷~w\n用空格隔开！！') -async def _(bot: Bot, event: Event, state: dict) -> None: - args = state['args'].split(' ') - uid_info = '' - server_id = '' - - if len(args[0]) != 9: - await genshin.finish('抱歉暂时只支持国服呢...') - else: - pass - - if re.findall('[Bb官]服', args[1]): - if re.findall('[Bb]服', args[1]): - server_id = 'cn_qd01' - else: - server_id = 'cn_gf01' - - await bot.send(event, '别急，在找！') - try: - uid_info = Genshin().jsonAnalysis(Genshin().getInfo(server_id=server_id, uid=args[0])) - except InvalidRequestError: - await genshin.finish('找不到呢...咱搜索区域或许出错了...') - - msg0 = (f'{args[0]} Genshin INFO:\n' - f'{uid_info}') - await genshin.finish(msg0) - else: - await genshin.finish('抱歉暂时只支持官服和b服...呐.') diff --git a/ATRI/plugins/genshin/data_source.py b/ATRI/plugins/genshin/data_source.py deleted file mode 100644 index 2bb4572..0000000 --- a/ATRI/plugins/genshin/data_source.py +++ /dev/null @@ -1,121 +0,0 @@ -import time -import json -import random -import string -import hashlib -import requests -from typing import Optional - -from ATRI.request import Request -from ATRI.config import GENSHIN_CONFIG -from ATRI.exceptions import InvalidRequestError - - -mhyVersion = GENSHIN_CONFIG['genshin']['mhyVersion'] - - -class Genshin: - def md5(self, text: str) -> str: - md5 = hashlib.md5() - md5.update(text.encode()) - return md5.hexdigest() - - def getDS(self): - global mhyVersion - if mhyVersion == '2.1.0': - n = self.md5(mhyVersion) - elif mhyVersion == '2.2.1': - n = "cx2y9z9a29tfqvr1qsq6c7yz99b5jsqt" - else: - mhyVersion = "2.2.1" - n = "cx2y9z9a29tfqvr1qsq6c7yz99b5jsqt" - - i = str(int(time.time())) - r = ''.join(random.sample(string.ascii_lowercase + string.digits, 6)) - c = self.md5("salt=" + n + "&t="+ i + "&r=" + r) - return (i + "," + r + "," + c) - - def getInfo(self, server_id: str, uid: str) -> str: - try: - url = GENSHIN_CONFIG['genshin']['url'] + server_id + "&role_id=" + uid - print(url) - headers: dict = { - 'Accept': 'application/json, text/plain, */*', - 'DS': self.getDS(), - 'Origin': 'https://webstatic.mihoyo.com', - 'x-rpc-app_version': mhyVersion, - 'User-Agent': 'Mozilla/5.0 (Linux; Android 9; Unspecified Device) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/39.0.0.0 Mobile Safari/537.36 miHoYoBBS/2.2.0', - 'x-rpc-client_type': '4', - 'Referer': 'https://webstatic.mihoyo.com/app/community-game-records/index.html?v=6', - 'Accept-Encoding': 'gzip, deflate', - 'Accept-Language': 'zh-CN,en-US;q=0.8', - 'X-Requested-With': 'com.mihoyo.hyperion' - } - result = requests.get(url=url,headers=headers, timeout=10) - print(result.text) - return result.text - except InvalidRequestError: - raise InvalidRequestError('请求数据出错') - - def jsonAnalysis(self, a) -> Optional[str]: - print(a) - data = json.loads(a) - if data['retcode'] != 0: - raise InvalidRequestError('请求出错，原因：uid错误/不存在/国服之外') - else: - pass - - character_info = 'Roles:\n' - character_list = data['data']['avatars'] - for i in character_list: - if i["element"] == "None": - character_type = "无属性" - elif i["element"] == "Anemo": - character_type = "风属性" - elif i["element"] == "Pyro": - character_type = "火属性" - elif i["element"] == "Geo": - character_type = "岩属性" - elif i["element"] == "Electro": - character_type = "雷属性" - elif i["element"] == "Cryo": - character_type = "冰属性" - elif i["element"] == "Hydro": - character_type = "水属性" - else: - character_type = "草属性" - - if i["name"] == "旅行者": - if i["image"].find("UI_AvatarIcon_PlayerGirl") != -1: - temp_text = f'* {i["name"]}：\n' - temp_text += f' - [萤——妹妹] {i["level"]}级 {character_type}\n' - - elif i["image"].find("UI_AvatarIcon_PlayerBoy") != -1: - temp_text = f'* {i["name"]}：\n' - temp_text += f' - [空——哥哥] {i["level"]}级 {character_type}\n' - - else: - temp_text = f'* {i["name"]}：\n' - temp_text += f' - [性别判断失败] {i["level"]}级 {character_type}\n' - else: - temp_text = f'* {i["name"]} {i["rarity"]}★角色:\n' - temp_text += f' - {i["level"]}级 好感度({i["fetter"]})级 {character_type}\n' - - character_info += temp_text - - a1 = data["data"]["stats"]["spiral_abyss"] - - account_info = 'Account Info:\n' - account_info += f'- 活跃天数：{data["data"]["stats"]["active_day_number"]} 天\n' - account_info += f'- 达成成就：{data["data"]["stats"]["achievement_number"]} 个\n' - account_info += f'- 获得角色：{data["data"]["stats"]["avatar_number"]}个\n' - account_info += f'- 深渊螺旋：{"没打" if (data["data"]["stats"]["spiral_abyss"] == "-") else f"打到了{a1}"}\n' - account_info += f'* 收集：\n' - account_info += f' - 风神瞳{data["data"]["stats"]["anemoculus_number"]}个 岩神瞳{data["data"]["stats"]["geoculus_number"]}个\n' - account_info += f'* 解锁：\n' - account_info += f' - 传送点{data["data"]["stats"]["way_point_number"]}个 秘境{data["data"]["stats"]["domain_number"]}个\n' - account_info += f'* 共开启宝箱：\n' - account_info += f' - 普通：{data["data"]["stats"]["common_chest_number"]}个 精致：{data["data"]["stats"]["exquisite_chest_number"]}个\n' - account_info += f' - 珍贵：{data["data"]["stats"]["luxurious_chest_number"]}个 华丽：{data["data"]["stats"]["precious_chest_number"]}个' - - return str(character_info + '\r\n' + account_info) diff --git a/ATRI/plugins/hitokoto.py b/ATRI/plugins/hitokoto.py index e69de29..cf03b6f 100644 --- a/ATRI/plugins/hitokoto.py +++ b/ATRI/plugins/hitokoto.py @@ -0,0 +1,76 @@ +import json +from pathlib import Path +from random import choice, randint + +from nonebot.plugin import on_command +from nonebot.adapters.cqhttp import Bot, Event + +from ATRI.exceptions import InvalidLoad +from ATRI.rule import is_in_ban_list, is_in_dormant, is_in_service, to_bot +from ATRI.utils import del_list_aim, count_list +from ATRI.request import Request +from ATRI.config import HITOKOTO_CONFIG +from ATRI.service.plugin import Plugin + +# ===========================[Begin Command Processing]=========================== + + +__plugin_name__ = 'hitokoto' +__doc__ = """一言""" +Plugin.register(__plugin_name__, "func", __doc__, + HITOKOTO_CONFIG['hitokoto']['command']) + +hitokoto = on_command(HITOKOTO_CONFIG['hitokoto']['command'][0], + aliases=set(HITOKOTO_CONFIG['hitokoto']['command']), + rule=is_in_ban_list() & is_in_dormant() + & is_in_service(__plugin_name__) + & to_bot()) + +async def _(bot: Bot, event: Event) -> None: + await bot.send(event, await Function().hitokoto(str(event.get_user_id()))) + + +# ===========================[End Command Processing]============================= + +hitokoto_list = [] +local_path = Path('.') / 'ATRI' / 'data' / 'database' / 'hitokoto' + +class Function: + async def hitokoto(self, user: str): + def local() -> str: + rd = choice(HITOKOTO_CONFIG['hitokoto']['local']['file']) + path = local_path / f"{rd}" + data = {} + try: + data = json.loads(path.read_bytes()) + except InvalidLoad: + raise InvalidLoad('Failed to read file!') + return data[randint(1, len(data)) - 1]['hitokoto'] + + async def link() -> str: + url = HITOKOTO_CONFIG['hitokoto']['link']['url'] + return str( + await Request.get_text( + url=url + ) + ) + + global hitokoto_list + + if count_list(hitokoto_list, user) == 3: + hitokoto_list.append(user) + return choice(HITOKOTO_CONFIG['hitokoto']['times'][3]['repo']) + elif count_list(hitokoto_list, user) == 6: + hitokoto_list = del_list_aim(hitokoto_list, user) + return choice(HITOKOTO_CONFIG['hitokoto']['times'][6]['repo']) + else: + hitokoto_list.append(user) + if HITOKOTO_CONFIG['hitokoto']['link']['use']: + rd = randint(1,2) + if rd == 1: + return await link() + else: + return local() + else: + return local() diff --git a/ATRI/plugins/setu/__init__.py b/ATRI/plugins/setu/__init__.py new file mode 100644 index 0000000..68cb1d2 --- /dev/null +++ b/ATRI/plugins/setu/__init__.py @@ -0,0 +1,29 @@ +from nonebot.plugin import on_regex +from nonebot.adapters.cqhttp import Bot +from nonebot.adapters.cqhttp.event import MessageEvent + +from ATRI.service.plugin import Plugin +from ATRI.config import SETU_CONFIG +from ATRI.rule import is_in_service, is_in_ban_list, is_in_dormant + +# ===========================[Begin Command Processing]=========================== + + +__plugin_name__ = 'setu' +__doc__ = """ +涩图，开冲！ +使用正则匹配 +""" +Plugin.register(plugin_name=__plugin_name__, _type="func", doc=__doc__, command=SETU_CONFIG['setu']['command']) + +setu = on_regex('|'.join(SETU_CONFIG['setu']['command']), + rule=is_in_service(__plugin_name__) & is_in_ban_list() + & is_in_dormant()) + +async def _(bot: Bot, event: MessageEvent) -> None: + await bot.send(event, SETU_CONFIG['setu']['repo']['waiting']) + + + +# ===========================[End Command Processing]=============================
\ No newline at end of file diff --git a/ATRI/plugins/setu/data_source.py b/ATRI/plugins/setu/data_source.py new file mode 100644 index 0000000..c5fffce --- /dev/null +++ b/ATRI/plugins/setu/data_source.py @@ -0,0 +1,14 @@ +from pathlib import Path + +from ATRI.config import SETU_CONFIG + +DATA_PATH = Path('.') / 'ATRI' / 'data' / 'database' + +class Function: + @staticmethod + async def setu_port() -> str: + ... + + @staticmethod + async def setu_local() -> str: + ... diff --git a/ATRI/plugins/utils/__init__.py b/ATRI/plugins/utils/__init__.py new file mode 100644 index 0000000..4204cfb --- /dev/null +++ b/ATRI/plugins/utils/__init__.py @@ -0,0 +1,76 @@ +import re +from nonebot.matcher import Matcher +from nonebot.plugin import on_command +from nonebot.adapters.cqhttp import Bot, Event + +from ATRI.config import UTILS_CONFIG +from ATRI.rule import is_in_ban_list, is_in_dormant, is_in_service + +from .data_source import Function + +# ===========================[Begin Command Processing]=========================== + + +__plugin_name_0__ = 'roll' +roll = on_command(UTILS_CONFIG['utils']['roll']['command'][0], + aliases=set(UTILS_CONFIG['utils']['roll']['command']), + rule=is_in_ban_list() & is_in_dormant() + & is_in_service(__plugin_name_0__)) + +__plugin_name_1__ = 'rcnb' +rcnbEncode = on_command( + UTILS_CONFIG['utils']['rcnb']['encode']['command'][0], + aliases=set( + UTILS_CONFIG['utils']['rcnb']['encode']['command']), + rule=is_in_ban_list() & is_in_dormant() + & is_in_service(__plugin_name_1__)) + +rcnbDecode = on_command( + UTILS_CONFIG['utils']['rcnb']['decode']['command'][0], + aliases=set( + UTILS_CONFIG['utils']['rcnb']['decode']['command']), + rule=is_in_ban_list() & is_in_dormant() + & is_in_service(__plugin_name_1__)) + + +async def _(bot, event: Event, state: dict) -> None: + args = str(event.get_message()).strip() + print(args) + if args: + state['result'] = args + [email protected]('result', prompt='roll参数不能为空！..\ndemo: 1d10 或 2d10+3d10') +async def _(matcher: Matcher, bot: Bot, event: Event, state: dict) -> None: + resu = state['result'] + match = re.match(r'^([\dd+\s]+?)$', resu) + print(match) + if not match: + await matcher.reject('格式不-正-确！\ndemo: 1d10 或 2d10+3d10') + await bot.send(event, Function.roll_dice(par=resu)) + + +async def _(bot, event: Event, state: dict) -> None: + args = str(event.get_message()).strip() + if args: + state['result'] = args + [email protected]('result', prompt='请告诉咱需要加密的字符~！') +async def _(bot: Bot, event: Event, state: dict) -> None: + print(state['result']) + await bot.send(event, Function.RCNB.encode(state['result'])) + + +async def _(bot, event: Event, state: dict) -> None: + args = str(event.get_message()).strip() + if args: + state['result'] = args + [email protected]('result', prompt='请告诉咱需要解密的字符~！') +async def _(bot: Bot, event: Event, state: dict) -> None: + await bot.send(event, Function.RCNB.decode(state['result'])) + + +# ===========================[End Command Processing]============================= diff --git a/ATRI/plugins/utils/data_source.py b/ATRI/plugins/utils/data_source.py new file mode 100644 index 0000000..4d7ddb6 --- /dev/null +++ b/ATRI/plugins/utils/data_source.py @@ -0,0 +1,239 @@ +import re +import os +import json +import random +from math import floor +from pathlib import Path +from zipfile import PyZipFile +from typing import Tuple, Union +from random import choice, randint +from typing import Dict, List + +from ATRI.config import UTILS_CONFIG + + +class Function: + @staticmethod + def roll_dice(par: str) -> str: + result = 0 + proc = '' + proc_list = [] + p = par.split('+') + + for i in p: + args = re.findall(r"(\d{0,10})(?:(d)(\d{1,10}))", i) + args = list(args[0]) + if not args[0]: + args[0] = 1 + if int(args[0]) >= 5000 or int(args[2]) >= 5000: + return choice(UTILS_CONFIG['utils']['roll']['tooBig']['repo']) + for a in range(1, int(args[0]) + 1): + rd = randint(1, int(args[2])) + result = result + rd + if len(proc_list) <= 10: + proc_list.append(rd) + + if len(proc_list) <= 10: + proc += '+'.join(map(str, proc_list)) + elif len(proc_list) >= 10: + proc += choice(UTILS_CONFIG['utils']['roll']['tooLong']['repo']) + else: + proc += str(result) + + msg = f'{par}=({proc})={result}' + print(msg) + return msg + + class Generate: + ''' + (*彩蛋*) + 由于此功能触及法律故不用作上线功能，写于此地只为学习 + 请勿用作网上购物、交易和注册 + 随机身份证根据国家标准(GB11643-1999)生成 + 并不存在所谓： + - 生成在逃犯号码 + ''' + @classmethod + def __init__(cls) -> None: + cls.DATA_PATH = Path('.') / 'ATRI' / 'plugins' / 'utils' / 'main.bin' + + @classmethod + def info_id(cls) -> Tuple[Dict[str, List[str]], Dict[str, str]]: + with PyZipFile(os.path.abspath(cls.DATA_PATH), 'r') as zipFile: + with zipFile.open('name.json', 'r') as f: + name = json.loads(f.read().decode()) + with zipFile.open('area.json', 'r') as f: + area = json.loads(f.read().decode()) + return name, area + + @staticmethod + def number_id(area: int, gender: int, birth: int) -> str: + ''' + 校验码计算公式: (12-∑(Ai×Wi)(mod 11))mod 11 + 验证公式: 请查阅（ISO 7064:1983.MOD 11-2） + ''' + def check_sum(full_code: str): + assert len(full_code) == 17 + check_sum = sum([((1 << (17 - i)) % 11) * int(full_code[i]) + for i in range(0, 17)]) + check_digit = (12 - (check_sum % 11)) % 11 + if check_digit < 10: + return check_digit + else: + return 'X' + + order_code = str(random.randint(10, 99)) + gender_code = str(random.randrange(gender, 10, step=2)) + full_code = str(area) + str(birth) + str(order_code) + str(gender_code) + full_code += str(check_sum(full_code)) + return full_code + + class RCNB: + @classmethod + def __init__(cls) -> None: + cls.cr = 'rRŔŕŖŗŘřƦȐȑȒȓɌɍ' + cls.cc = 'cCĆćĈĉĊċČčƇƈÇȻȼ' + cls.cn = 'nNŃńŅņŇňƝƞÑǸǹȠȵ' + cls.cb = 'bBƀƁƃƄƅßÞþ' + + cls.sr = len(cls.cr) + cls.sc = len(cls.cc) + cls.sn = len(cls.cn) + cls.sb = len(cls.cb) + cls.src = cls.sr * cls.sc + cls.snb = cls.sn * cls.sb + cls.scnb = cls.sc * cls.snb + + @staticmethod + def _div(a: int, b: int) -> int: + return floor(a / b) + + @classmethod + def _encode_byte(cls, i) -> Union[str, None]: + if i > 0xFF: + raise ValueError('ERROR! rc/nb overflow') + + if i > 0x7F: + i = i & 0x7F + return cls.cn[i // cls.sb] + cls.cb[i % cls.sb] + # return f'{cls.cn[i // cls.sb]}{cls.cb[i % cls.sb]}' + # return cls.cn[cls._div(i, cls.sb) + int(cls.cb[i % cls.sb])] + + return cls.cr[i // cls.sc] + cls.cc[i % cls.sc] + # return cls.cr[cls._div(i, cls.sc) + int(cls.cc[i % cls.sc])] + + @classmethod + def _encode_short(cls, i) -> str: + if i > 0xFFFF: + raise ValueError('ERROR! rcnb overflow') + + reverse = False + if i > 0x7FFF: + reverse = True + i = i & 0x7FFF + + char = [ + cls._div(i, cls.scnb), + cls._div(i % cls.scnb, cls.snb), + cls._div(i % cls.snb, cls.sb), i % cls.sb + ] + char = [ + cls.cr[char[0]], cls.cc[char[1]], cls.cn[char[2]], + cls.cb[char[3]] + ] + + if reverse: + return char[2] + char[3] + char[0] + char[1] + + return ''.join(char) + + @classmethod + def _decodeByte(cls, c) -> int: + nb = False + idx = [cls.cr.index(c[0]), cls.cc.index(c[1])] + if idx[0] < 0 or idx[1] < 0: + idx = [cls.cn.index(c[0]), cls.cb.index(c[1])] + nb = True + raise ValueError('ERROR! rc/nb overflow') + + result = idx[0] * cls.sb + idx[1] if nb else idx[0] * cls.sc + idx[1] + if result > 0x7F: + raise ValueError('ERROR! rc/nb overflow') + + return result | 0x80 if nb else 0 + + @classmethod + def _decodeShort(cls, c) -> int: + reverse = c[0] not in cls.cr + if not reverse: + idx = [ + cls.cr.index(c[0]), + cls.cc.index(c[1]), + cls.cn.index(c[2]), + cls.cb.index(c[3]) + ] + else: + idx = [ + cls.cr.index(c[2]), + cls.cc.index(c[3]), + cls.cn.index(c[0]), + cls.cb.index(c[1]) + ] + + if idx[0] < 0 or idx[1] < 0 or idx[2] < 0 or idx[3] < 0: + raise ValueError('ERROR! not rcnb') + + result = (idx[0] * cls.scnb + + idx[1] * cls.snb + + idx[2] * cls.sb + idx[3]) + if result > 0x7FFF: + raise ValueError('ERROR! rcnb overflow') + + result |= 0x8000 if reverse else 0 + return result + + @classmethod + def _encodeBytes(cls, b) -> str: + result = [] + for i in range(0, (len(b) >> 1)): + result.append(cls._encode_short((b[i * 2] << 8 | b[i * 2 + 1]))) + + if len(b) & 1 == 1: + result.append(cls._encode_byte(b[-1])) + + return ''.join(result) + + @classmethod + def encode(cls, s: str, encoding: str = 'utf-8'): + if not isinstance(s, str): + raise ValueError('Please enter str instead of other') + + return cls._encodeBytes(s.encode(encoding)) + + @classmethod + def _decodeBytes(cls, s: str): + if not isinstance(s, str): + raise ValueError('Please enter str instead of other') + + if len(s) & 1: + raise ValueError('ERROR length') + + result = [] + for i in range(0, (len(s) >> 2)): + result.append(bytes([cls._decodeShort(s[i * 4:i * 4 + 4]) >> 8])) + result.append(bytes([cls._decodeShort(s[i * 4:i * 4 + 4]) & 0xFF])) + + if (len(s) & 2) == 2: + result.append(bytes([cls._decodeByte(s[-2:])])) + + return b''.join(result) + + @classmethod + def decode(cls, s: str, encoding: str = 'utf-8') -> str: + if not isinstance(s, str): + raise ValueError('Please enter str instead of other') + + try: + return cls._decodeBytes(s).decode(encoding) + except UnicodeDecodeError: + raise ValueError('Decoding failed') diff --git a/ATRI/plugins/utils/main.bin b/ATRI/plugins/utils/main.bin Binary files differnew file mode 100644 index 0000000..6e74a60 --- /dev/null +++ b/ATRI/plugins/utils/main.bin |