xuthus/LittlePaimon
1
0
Fork 0
mirror of https://github.com/xuthus83/LittlePaimon.git synced 2024-10-21 16:27:15 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

✨ 修复深渊无承伤时的制图失败，暂时移除myb_exchange插件#376

Browse Source
This commit is contained in:
CMHopeSunshine 2023-01-31 15:55:34 +08:00
parent 66530afe46
commit ca768df11b
7 changed files with 286 additions and 624 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

333
LittlePaimon/plugins/Paimon_Abyss/draw_abyss.py
View File

@ -2,9 +2,13 @@ import asyncio
import datetime
from LittlePaimon.database import AbyssInfo
from LittlePaimon.utils.path import RESOURCE_BASE_PATH
from LittlePaimon.utils.image import PMImage, font_manager as fm, load_image
from LittlePaimon.utils.image import PMImage
from LittlePaimon.utils.image import font_manager as fm
from LittlePaimon.utils.image import load_image
from LittlePaimon.utils.message import MessageBuild
from LittlePaimon.utils.path import RESOURCE_BASE_PATH
time_str = ['零', '一', '二', '三', '四', '五', '六', '七', '八', '九', '十', '十一', '十二']
def datetime_to_cn(time: datetime.datetime) -> str:
@ -13,7 +17,6 @@ def datetime_to_cn(time: datetime.datetime) -> str:
:param time: 时间
:return: 中文时间
"""
time_str = ['零', '一', '二', '三', '四', '五', '六', '七', '八', '九', '十', '十一', '十二']
if time.day < 16:
return f'{time_str[time.month]}月上半'
else:
@ -24,7 +27,9 @@ async def draw_abyss_card(info: AbyssInfo):
if not info.total_battle or not info.max_battle:
return '暂无深渊挑战数据，请稍候再试'
# 加载图片素材
img = PMImage(await load_image(RESOURCE_BASE_PATH / 'abyss' / 'bg.png', mode='RGBA'))
img = PMImage(
await load_image(RESOURCE_BASE_PATH / 'abyss' / 'bg.png', mode='RGBA')
)
orange_line = await load_image(RESOURCE_BASE_PATH / 'general' / 'line.png')
# 标题文字
await img.text('深渊战报', 36, 29, fm.get('优设标题黑', 108), '#40342d')
@ -37,58 +42,171 @@ async def draw_abyss_card(info: AbyssInfo):
await img.text(time_str, 438, 41, fm.get('SourceHanSansCN-Bold.otf', 30), 'white')
# UID和昵称
if info.nickname:
await img.text(f'UID{info.uid}', 1040, 72, fm.get('bahnschrift_regular.ttf', 36), '#40342d', 'right')
await img.text(info.nickname, 1040, 114, fm.get('SourceHanSansCN-Bold.otf', 30), '#40342d', 'right')
await img.text(
f'UID{info.uid}',
1040,
72,
fm.get('bahnschrift_regular.ttf', 36),
'#40342d',
'right',
)
await img.text(
info.nickname,
1040,
114,
fm.get('SourceHanSansCN-Bold.otf', 30),
'#40342d',
'right',
)
else:
await img.text(f'UID{info.uid}', 1040, 114, fm.get('bahnschrift_regular.ttf', 36), '#40342d', 'right')
await img.text(
f'UID{info.uid}',
1040,
114,
fm.get('bahnschrift_regular.ttf', 36),
'#40342d',
'right',
)
# 战绩速览标题
await img.paste(orange_line, (40, 164))
await img.text('战绩速览', 63, 176, fm.get('SourceHanSansCN-Bold.otf', 30), 'white')
# logo和生成时间
await img.text(f'CREATED BY LITTLEPAIMON AT {datetime.datetime.now().strftime("%m-%d %H:%M")}',
1025, 178, fm.get('bahnschrift_regular.ttf', 30), '#8c4c2e', 'right')
await img.text(
f'CREATED BY LITTLEPAIMON AT {datetime.datetime.now().strftime("%m-%d %H:%M")}',
1025,
178,
fm.get('bahnschrift_regular.ttf', 30),
'#8c4c2e',
'right',
)
# 数据栏
await img.paste(await load_image(RESOURCE_BASE_PATH / 'abyss' / 'data_line.png'), (40, 246))
await img.paste(
await load_image(RESOURCE_BASE_PATH / 'abyss' / 'data_line.png'), (40, 246)
)
# 获得总星数
await img.text(f'{info.total_star}/36', (47, 150), 357, fm.get('bahnschrift_regular.ttf', 56), '#40342d', 'center')
await img.text(
f'{info.total_star}/36',
(47, 150),
357,
fm.get('bahnschrift_regular.ttf', 56),
'#40342d',
'center',
)
# 战斗次数
await img.text(str(info.total_battle), (209, 312), 357, fm.get('bahnschrift_regular.ttf', 56), '#40342d', 'center')
await img.text(
str(info.total_battle),
(209, 312),
357,
fm.get('bahnschrift_regular.ttf', 56),
'#40342d',
'center',
)
# 最多击破
await img.text(str(info.max_defeat.value), (370, 473), 357, fm.get('bahnschrift_regular.ttf', 56), '#40342d',
'center')
chara_img = PMImage(await load_image(RESOURCE_BASE_PATH / 'avatar' / f'{info.max_defeat.icon}.png', size=(96, 96)))
if info.max_defeat is not None:
await img.text(
str(info.max_defeat.value),
(370, 473),
357,
fm.get('bahnschrift_regular.ttf', 56),
'#40342d',
'center',
)
chara_img = PMImage(
await load_image(
RESOURCE_BASE_PATH / 'avatar' / f'{info.max_defeat.icon}.png',
size=(96, 96),
)
)
await chara_img.to_circle('circle')
await img.paste(chara_img, (373, 248))
# 战技次数
await img.text(str(info.max_normal_skill.value), (532, 635), 357, fm.get('bahnschrift_regular.ttf', 56), '#40342d',
'center')
if info.max_normal_skill is not None:
await img.text(
str(info.max_normal_skill.value),
(532, 635),
357,
fm.get('bahnschrift_regular.ttf', 56),
'#40342d',
'center',
)
chara_img = PMImage(
await load_image(RESOURCE_BASE_PATH / 'avatar' / f'{info.max_normal_skill.icon}.png', size=(96, 96)))
await load_image(
RESOURCE_BASE_PATH / 'avatar' / f'{info.max_normal_skill.icon}.png',
size=(96, 96),
)
)
await chara_img.to_circle('circle')
await img.paste(chara_img, (536, 248))
# 爆发次数
await img.text(str(info.max_energy_skill.value), (693, 796), 357, fm.get('bahnschrift_regular.ttf', 56), '#40342d',
'center')
if info.max_energy_skill is not None:
await img.text(
str(info.max_energy_skill.value),
(693, 796),
357,
fm.get('bahnschrift_regular.ttf', 56),
'#40342d',
'center',
)
chara_img = PMImage(
await load_image(RESOURCE_BASE_PATH / 'avatar' / f'{info.max_energy_skill.icon}.png', size=(96, 96)))
await load_image(
RESOURCE_BASE_PATH / 'avatar' / f'{info.max_energy_skill.icon}.png',
size=(96, 96),
)
)
await chara_img.to_circle('circle')
await img.paste(chara_img, (696, 248))
# 最深抵达
await img.text(str(info.max_floor), (838, 1038), 298, fm.get('bahnschrift_regular.ttf', 60), '#40342d', 'center')
if info.max_floor is not None:
await img.text(
str(info.max_floor),
(838, 1038),
298,
fm.get('bahnschrift_regular.ttf', 60),
'#40342d',
'center',
)
# 最强一击
circle = await load_image(RESOURCE_BASE_PATH / 'general' / 'orange_circle.png')
chara_img = PMImage(await load_image(RESOURCE_BASE_PATH / 'avatar' / f'{info.max_damage.icon}.png', size=(205, 205)))
if info.max_damage is not None:
chara_img = PMImage(
await load_image(
RESOURCE_BASE_PATH / 'avatar' / f'{info.max_damage.icon}.png',
size=(205, 205),
)
)
await chara_img.to_circle('circle')
await img.text('最强一击', 270, 520, fm.get('SourceHanSansCN-Bold.otf', 48), '#40342d')
await img.text(str(info.max_damage.value), 270, 590, fm.get('bahnschrift_bold.ttf', 72, 'Bold'), '#40342d')
await img.text(
'最强一击', 270, 520, fm.get('SourceHanSansCN-Bold.otf', 48), '#40342d'
)
await img.text(
str(info.max_damage.value),
270,
590,
fm.get('bahnschrift_bold.ttf', 72, 'Bold'),
'#40342d',
)
await img.paste(circle, (46, 485))
await img.paste(chara_img, (49, 488))
# 最多承伤
chara_img = PMImage(await load_image(RESOURCE_BASE_PATH / 'avatar' / f'{info.max_take_damage.icon}.png', size=(205, 205)))
if info.info.max_take_damage is not None:
chara_img = PMImage(
await load_image(
RESOURCE_BASE_PATH / 'avatar' / f'{info.max_take_damage.icon}.png',
size=(205, 205),
)
)
await chara_img.to_circle('circle')
await img.text('最多承伤', 791, 520, fm.get('SourceHanSansCN-Bold.otf', 48), '#40342d')
await img.text(str(info.max_take_damage.value), 791, 590, fm.get('bahnschrift_bold.ttf', 72, 'Bold'), '#40342d')
await img.text(
'最多承伤', 791, 520, fm.get('SourceHanSansCN-Bold.otf', 48), '#40342d'
)
await img.text(
str(info.max_take_damage.value),
791,
590,
fm.get('bahnschrift_bold.ttf', 72, 'Bold'),
'#40342d',
)
await img.paste(circle, (560, 485))
await img.paste(chara_img, (563, 488))
# 没有9-12层信息时不会只详细战绩
@ -98,74 +216,159 @@ async def draw_abyss_card(info: AbyssInfo):
await img.paste(orange_line, (40, 747))
await img.text('详细战绩', 63, 759, fm.get('SourceHanSansCN-Bold.otf', 30), 'white')
for i in range(9, 13):
await img.paste(await load_image(RESOURCE_BASE_PATH / 'abyss' / f'floor{i}.png'), (41, 834 + (i - 9) * 194))
await img.paste(
await load_image(RESOURCE_BASE_PATH / 'abyss' / f'floor{i}.png'),
(41, 834 + (i - 9) * 194),
)
if info.floors.get(i):
if info.floors[i].stars:
await img.text('-'.join(map(str, info.floors[i].stars)), 91, 960 + (i - 9) * 194,
fm.get('bahnschrift_bold.ttf', 30, 'Bold'), 'white')
await img.text(
'-'.join(map(str, info.floors[i].stars)),
91,
960 + (i - 9) * 194,
fm.get('bahnschrift_bold.ttf', 30, 'Bold'),
'white',
)
if info.floors[i].battles_up:
up = info.floors[i].battles_up
down = info.floors[i].battles_down
j = 0
for character in up[0]:
await img.paste(
await load_image(RESOURCE_BASE_PATH / 'icon' / f'star{character.rarity}.png', size=(95, 95)),
(192 + (j % 4) * 103, 832 + (i - 9) * 194))
await load_image(
RESOURCE_BASE_PATH / 'icon' / f'star{character.rarity}.png',
size=(95, 95),
),
(192 + (j % 4) * 103, 832 + (i - 9) * 194),
)
await img.paste(
await load_image(RESOURCE_BASE_PATH / 'avatar' / f'{character.icon}.png', size=(95, 95)),
(192 + (j % 4) * 103, 832 + (i - 9) * 194))
await img.draw_rounded_rectangle2((192 + (j % 4) * 103, 903 + (i - 9) * 194), (30, 23), 10,
await load_image(
RESOURCE_BASE_PATH / 'avatar' / f'{character.icon}.png',
size=(95, 95),
),
(192 + (j % 4) * 103, 832 + (i - 9) * 194),
)
await img.draw_rounded_rectangle2(
(192 + (j % 4) * 103, 903 + (i - 9) * 194),
(30, 23),
10,
'#333333',
['ll', 'ur'])
await img.text(str(character.level), (192 + (j % 4) * 103, 192 + (j % 4) * 103 + 30),
['ll', 'ur'],
)
await img.text(
str(character.level),
(192 + (j % 4) * 103, 192 + (j % 4) * 103 + 30),
906 + (i - 9) * 194,
fm.get('bahnschrift_bold.ttf', 20, 'Bold'), 'white', 'center')
fm.get('bahnschrift_bold.ttf', 20, 'Bold'),
'white',
'center',
)
j += 1
j = 0
for character in down[0]:
await img.paste(
await load_image(RESOURCE_BASE_PATH / 'icon' / f'star{character.rarity}.png', size=(95, 95)),
(637 + (j % 4) * 103, 832 + (i - 9) * 194))
await load_image(
RESOURCE_BASE_PATH / 'icon' / f'star{character.rarity}.png',
size=(95, 95),
),
(637 + (j % 4) * 103, 832 + (i - 9) * 194),
)
await img.paste(
await load_image(RESOURCE_BASE_PATH / 'avatar' / f'{character.icon}.png', size=(95, 95)),
(637 + (j % 4) * 103, 832 + (i - 9) * 194))
await img.draw_rounded_rectangle2((637 + (j % 4) * 103, 903 + (i - 9) * 194), (30, 23), 10,
await load_image(
RESOURCE_BASE_PATH / 'avatar' / f'{character.icon}.png',
size=(95, 95),
),
(637 + (j % 4) * 103, 832 + (i - 9) * 194),
)
await img.draw_rounded_rectangle2(
(637 + (j % 4) * 103, 903 + (i - 9) * 194),
(30, 23),
10,
'#333333',
['ll', 'ur'])
await img.text(str(character.level), (637 + (j % 4) * 103, 637 + (j % 4) * 103 + 30),
['ll', 'ur'],
)
await img.text(
str(character.level),
(637 + (j % 4) * 103, 637 + (j % 4) * 103 + 30),
906 + (i - 9) * 194,
fm.get('bahnschrift_bold.ttf', 20, 'Bold'), 'white', 'center')
fm.get('bahnschrift_bold.ttf', 20, 'Bold'),
'white',
'center',
)
j += 1
j = 0
for group in up:
await img.paste(await load_image(RESOURCE_BASE_PATH / 'general' / 'orange_bord_small.png'),
(192 + (j % 3) * 287, 936 + (i - 9) * 194))
await asyncio.gather(*[img.paste(
await img.paste(
await load_image(
RESOURCE_BASE_PATH / 'avatar_side' / f'{group[k].icon.replace("Icon_", "Icon_Side_")}.png',
size=(40, 40)), (190 + (j % 3) * 287 + (k % 4) * 32, 928 + (i - 9) * 194)) for k in
range(len(group))])
await img.text(f'{i}-{j + 1} {info.floors[i].end_times_down[j].strftime("%H:%M:%S")}',
(192 + (j % 3) * 287, 463 + (j % 3) * 287), (978 + (i - 9) * 194),
fm.get('bahnschrift_regular.ttf', 24), '#40342d', 'center')
RESOURCE_BASE_PATH / 'general' / 'orange_bord_small.png'
),
(192 + (j % 3) * 287, 936 + (i - 9) * 194),
)
await asyncio.gather(
*[
img.paste(
await load_image(
RESOURCE_BASE_PATH
/ 'avatar_side'
/ f'{group[k].icon.replace("Icon_", "Icon_Side_")}.png',
size=(40, 40),
),
(
190 + (j % 3) * 287 + (k % 4) * 32,
928 + (i - 9) * 194,
),
)
for k in range(len(group))
]
)
await img.text(
f'{i}-{j + 1} {info.floors[i].end_times_down[j].strftime("%H:%M:%S")}',
(192 + (j % 3) * 287, 463 + (j % 3) * 287),
(978 + (i - 9) * 194),
fm.get('bahnschrift_regular.ttf', 24),
'#40342d',
'center',
)
j += 1
j = 0
for group in down:
await img.paste(await load_image(RESOURCE_BASE_PATH / 'general' / 'orange_bord_small.png'),
(330 + (j % 3) * 287, 936 + (i - 9) * 194))
await asyncio.gather(*[img.paste(
await img.paste(
await load_image(
RESOURCE_BASE_PATH / 'avatar_side' / f'{group[k].icon.replace("Icon_", "Icon_Side_")}.png',
size=(40, 40)), (328 + (j % 3) * 287 + (k % 4) * 32, 928 + (i - 9) * 194)) for k in
range(len(group))])
RESOURCE_BASE_PATH / 'general' / 'orange_bord_small.png'
),
(330 + (j % 3) * 287, 936 + (i - 9) * 194),
)
await asyncio.gather(
*[
img.paste(
await load_image(
RESOURCE_BASE_PATH
/ 'avatar_side'
/ f'{group[k].icon.replace("Icon_", "Icon_Side_")}.png',
size=(40, 40),
),
(
328 + (j % 3) * 287 + (k % 4) * 32,
928 + (i - 9) * 194,
),
)
for k in range(len(group))
]
)
j += 1
else:
await img.paste(await load_image(RESOURCE_BASE_PATH / 'abyss' / 'nock.png'), (192, 834 + (i - 9) * 194))
await img.paste(
await load_image(RESOURCE_BASE_PATH / 'abyss' / 'nock.png'),
(192, 834 + (i - 9) * 194),
)
else:
await img.paste(await load_image(RESOURCE_BASE_PATH / 'abyss' / 'nodata.png'), (192, 834 + (i - 9) * 194))
await img.paste(
await load_image(RESOURCE_BASE_PATH / 'abyss' / 'nodata.png'),
(192, 834 + (i - 9) * 194),
)
return MessageBuild.Image(img, quality=80, mode='RGB')

0
src/plugins/.gitkeep Normal file
View File

44
src/plugins/nonebot_plugin_myb_exchange/README.md
View File

@ -1,44 +0,0 @@
<p align="center" >
<a href="https://github.com/CMHopeSunshine/LittlePaimon/tree/nonebot2"><img src="http://static.cherishmoon.fun/LittlePaimon/readme/logo.png" width="256" height="256" alt="LittlePaimon"></a>
</p>
<h1 align="center">nonebot-plugin-myb-exchange</h1>
<h4 align="center">Nonebot2米游币商品自动兑换插件</h4>
<p align="center" >
<a href="https://pypi.python.org/pypi/nonebot-plugin-myb-exchange">
<img src="https://img.shields.io/pypi/v/nonebot-plugin-myb-exchange" alt="pypi">
</a>
<img src="https://img.shields.io/badge/Python-3.8+-yellow" alt="python">
<img src="https://img.shields.io/badge/Nonebot-2.0.0b5-green" alt="python">
</p>
## 安装
> 需要配合nonebot2使用，安装方法详见[nb文档](https://v2.nonebot.dev/)
- 通过nb脚手架安装（推荐）
```
nb plugin install nonebot-plugin-myb-exchange
```
- 通过pip安装
```
pip install nonebot-plugin-myb-exchange
# 还需要在bot.py加载插件，在bot.py中添加
nonebot.load_plugin("nonebot-plugin-myb-exchange")
```
- 通过poetry安装
```
poetry add nonebot-plugin-myb-exchange
# 还需要在bot.py加载插件，同上
```
## 使用方法
> 由于涉及到cookie、地址等敏感信息，所以仅限**私聊**Bot使用
- 发送`myb`，跟随Bot的指引一步一步录入信息，需要用户提供`米游社cookie、收货地址、商品名称`等信息
- 发送`myb_info`可查看已录入的兑换计划
- 发送`myb_delete`可一键删除所有你的兑换计划
## 成功兑换示范
<img src="https://static.cherishmoon.fun/LittlePaimon/readme/QQ%E5%9B%BE%E7%89%8720220630233519.png" alt="myb">

166
src/plugins/nonebot_plugin_myb_exchange/__init__.py
View File

@ -1,166 +0,0 @@
import re
from nonebot import on_command
from nonebot.adapters.onebot.v11 import PrivateMessageEvent, Message
from nonebot.params import CommandArg, T_State, Arg
from nonebot.plugin import PluginMetadata
from .data_source import get_address, get_goods, save_exchange_info, get_exchange_info, delete_exchange_info
__plugin_meta__ = PluginMetadata(
name="米游币商品自动兑换插件",
description="录入米游币兑换计划，Bot会在对应时间自动帮你抢兑商品",
usage=(
"myb 跟随Bot的指引录入兑换计划\n"
"myb_info 查看当前的兑换计划\n"
"myb_delete 删除你的所有兑换计划\n"
),
extra={
"author": "惜月 <277073121@qq.com>",
"version": "1.0.0",
},
)
myb_exchange = on_command('myb', aliases={'米游币兑换', '米游币商品兑换', '米游社商品兑换'}, priority=6, state={
'pm_name': '米游币兑换',
'pm_usage': 'myb',
'pm_show': True,
'pm_priority': 15,
'pm_description': '录入米游币商品自动兑换计划',
})
myb_info = on_command('myb_info', aliases={'米游币兑换信息', '米游币兑换计划'}, priority=6, state={
'pm_name': '米游币兑换信息',
'pm_usage': 'myb_info',
'pm_show': True,
'pm_priority': 16,
'pm_description': '查看你的米游币兑换计划',
})
myb_delete = on_command('myb_delete', aliases={'米游币兑换删除', '米游币兑换取消'}, priority=6, state={
'pm_name': '米游币兑换删除',
'pm_usage': 'myb_delete',
'pm_show': True,
'pm_priority': 17,
'pm_description': '取消你的米游币兑换计划',
})
@myb_exchange.handle()
async def _(event: PrivateMessageEvent, state: T_State, msg: Message = CommandArg()):
if msg:
msg = msg.extract_plain_text().strip()
if '虚拟' in msg:
state['商品类型'] = '虚拟'
elif '实体' in msg:
state['商品类型'] = '实体'
state['uid'] = None
@myb_exchange.got('商品类型', prompt='请给出要抢的商品类型(虚拟|实体)，例如原石属于虚拟')
async def _(event: PrivateMessageEvent, state: T_State, type: Message = Arg('商品类型')):
type = type.extract_plain_text().strip()
if '虚拟' in type:
state['商品类型'] = '虚拟'
print(state)
elif '实体' in type:
state['商品类型'] = '实体'
state['uid'] = None
else:
await myb_exchange.reject('请给出要抢的商品类型(虚拟|实体)，例如原石属于虚拟')
@myb_exchange.got('uid', prompt='请把虚拟商品要兑换到的游戏uid告诉我')
async def _(event: PrivateMessageEvent, state: T_State, uid: Message = Arg('uid')):
uid = uid.extract_plain_text().strip()
if find_uid := re.search(r'(?P<uid>[125]\d{8})', uid):
state['uid'] = find_uid['uid']
else:
await myb_exchange.reject('这不是有效的uid')
@myb_exchange.got('cookie', prompt='请把米游币cookie给我，cookie获取方式详见：\ndocs.qq.com/doc/DQ3JLWk1vQVllZ2Z1')
async def _(event: PrivateMessageEvent, state: T_State, cookie: Message = Arg('cookie')):
cookie = cookie.extract_plain_text().strip()
address = await get_address(cookie)
if address is None:
await myb_exchange.reject('这个cookie无效，请检查是否以按照正常方法获取')
elif len(address) == 0:
await myb_exchange.finish('你的账号还没有填写收货地址哦，请先去填写收货地址重新再来')
else:
state['cookie'] = cookie
if len(address) == 1:
state['address_id'] = address[0]
else:
state['address_list'] = address
if state['商品类型'] == '虚拟' and 'login_ticket' not in cookie and 'stoken' not in cookie:
await myb_exchange.reject('你的cookie中没有login_ticket字段哦，请尝试退出后重新登录再获取cookie')
@myb_exchange.got('address_id', prompt='回复任意文字继续，接下来回复选择你的收货地址的ID')
async def _(event: PrivateMessageEvent, state: T_State, address_id: Message = Arg('address_id')):
address_id = address_id.extract_plain_text().strip()
flag = False
for add in state['address_list']:
if address_id == add['id']:
state['address_id'] = add
flag = True
break
if not flag:
address_list = ''.join(f'ID：{add["id"]}，{add["地址"]}\n' for add in state['address_list'])
await myb_exchange.reject(f'请选择收货地址ID：\n{address_list}')
@myb_exchange.got('game', prompt='请给出要抢的商品所属游戏名称，有崩坏3|原神|崩坏学园2|未定事件簿|米游社')
async def _(event: PrivateMessageEvent, state: T_State, game: Message = Arg('game')):
game = game.extract_plain_text().strip()
if game in ['崩坏3', 'bh3', '崩崩崩', '三崩子']:
state['goods_list'] = await get_goods('崩坏3')
elif game in ['原神', 'ys']:
state['goods_list'] = await get_goods('原神')
elif game in ['崩坏学园2', 'bh2', '二崩子', '崩坏学院2', '崩崩']:
state['goods_list'] = await get_goods('崩坏学园2')
elif game in ['未定事件簿', 'wdsjb', '未定']:
state['goods_list'] = await get_goods('未定事件簿')
elif game in ['米游社', 'mys']:
state['goods_list'] = await get_goods('米游社')
else:
await myb_exchange.reject('请给出要抢的商品所属游戏名称，有崩坏3|原神|崩坏学园2|未定事件簿|米游社')
@myb_exchange.got('goods_search', prompt='请给出要兑换的商品名，或者其含有的关键词')
async def _(event: PrivateMessageEvent, state: T_State, goods_search: Message = Arg('goods_search')):
goods_search = goods_search.extract_plain_text().strip()
match_goods = [good for good in state['goods_list'] if goods_search in good['name']]
if len(match_goods) == 1:
state['goods'] = match_goods[0]
save_exchange_info(event.user_id, state)
await myb_exchange.finish('兑换计划录入成功，到时候会帮你兑换并告诉你结果，发送 myb_info 可以再次确认兑换信息，发送 myb_delete 可以取消兑换计划')
elif len(match_goods) > 1:
state['goods_search_result'] = match_goods
else:
await myb_exchange.reject('没有相关可兑换的商品，请重新输入')
@myb_exchange.got('goods', prompt='回复任意文字继续，接下来回复选择你想要兑换的商品的ID')
async def _(event: PrivateMessageEvent, state: T_State, msg: Message = Arg('goods')):
msg = msg.extract_plain_text().strip()
for good in state['goods_search_result']:
if msg == good['id']:
state['goods'] = good
save_exchange_info(event.user_id, state)
await myb_exchange.finish('兑换计划录入成功，到时候会帮你兑换并告诉你结果，发送 myb_info 可以再次确认兑换信息，发送 myb_delete 可以取消兑换计划')
good_str = ''.join(f'ID：{good["id"]}, 商品名：{good["name"]}\n' for good in state['goods_search_result'])
await myb_exchange.reject('请选择商品ID：\n' + good_str)
@myb_info.handle()
async def _(event: PrivateMessageEvent):
info = get_exchange_info(str(event.user_id))
await myb_info.finish(info)
@myb_delete.handle()
async def _(event: PrivateMessageEvent):
delete_exchange_info(str(event.user_id))
await myb_delete.finish('米游币兑换计划已全部取消')

210
src/plugins/nonebot_plugin_myb_exchange/data_source.py
View File

@ -1,210 +0,0 @@
import datetime
import random
import re
import string
import time
from asyncio import sleep
from pathlib import Path
from nonebot import require, get_bot, get_driver
from .utils import get, post, load_json, save_json
require('nonebot_plugin_apscheduler')
from nonebot_plugin_apscheduler import scheduler
driver = get_driver()
async def get_address(cookie):
address_url = 'https://api-takumi.mihoyo.com/account/address/list'
header = {
'Accept': 'application/json, text/plain, */*',
'Accept-Encoding': 'gzip, deflate, br',
'Accept-Language': 'zh-CN,zh-Hans;q=0.9',
'Connection': 'keep-alive',
'Cookie': cookie,
'Host': 'api-takumi.mihoyo.com',
'Origin': 'https://user.mihoyo.com',
'Referer': 'https://user.mihoyo.com/',
'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 15_4 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) miHoYoBBS/2.25.1'
}
res = await get(url=address_url, headers=header)
res = res.json()
if res['message'] != 'OK':
return None
address = res['data']['list']
return [{'id': add['id'], '地址': f'姓名:{add["connect_name"]} 电话:{add["connect_mobile"]} 地址:{add["province_name"] + add["city_name"] + add["county_name"] + add["addr_ext"]}'} for add in address]
async def get_goods(game):
game_type = {'崩坏3': 'bh3', '原神': 'hk4e', '崩坏学园2': 'bh2', '未定事件簿': 'nxx', '米游社': 'bbs'}
url = 'https://api-takumi.mihoyo.com/mall/v1/web/goods/list?app_id=1&point_sn=myb&page_size=20&page={page}&game=' + game_type[game]
goods_list = []
goods_list_new = []
page = 1
while True:
res = await get(url=url.format(page=page))
res = res.json()
if not res['data']['list']:
break
else:
goods_list += res['data']['list']
page += 1
for good in goods_list:
if good['next_time'] == 0 and good['type'] == 1:
continue
good_info = {'id': good['goods_id'], 'name': good['goods_name'], 'price': good['price'], 'time': time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(good['next_time']))}
goods_list_new.append(good_info)
return goods_list_new
def save_exchange_info(user_id, state):
info = {'user_id': user_id, '类型': state['商品类型'], 'cookie': state['cookie'], '商品': state['goods'], '地址': state['address_id']}
if info['类型'] == '虚拟':
info['uid'] = state['uid']
t = f"{user_id}-{info['商品']['id']}"
path = Path() / 'data' / 'LittlePaimon' / 'myb_exchange' / f'{t}.json'
path.parent.mkdir(parents=True, exist_ok=True)
save_json(data=info, path=path)
scheduler.add_job(id=t, replace_existing=True, misfire_grace_time=5, func=exchange_action, trigger='date', args=(info,), next_run_time=datetime.datetime.strptime(info['商品']['time'], '%Y-%m-%d %H:%M:%S'))
async def get_bbs_info(info, headers):
url = 'https://api-takumi.mihoyo.com/binding/api/getUserGameRolesByCookie'
res = (await get(url=url, headers=headers)).json()
if res['retcode'] == 0:
data = res['data']['list']
for d in data:
if d['game_uid'] == info['uid']:
return d['game_biz'], d['region']
return None, None
async def get_stoken(cookie):
bbs_cookie_url = 'https://webapi.account.mihoyo.com/Api/cookie_accountinfo_by_loginticket?login_ticket={}'
login_ticket = re.search('login_ticket=[a-zA-Z0-9]{0,100}', cookie)
data = (await get(url=bbs_cookie_url.format(login_ticket.group().split('=')[1]))).json()
if '成功' in data['data']['msg']:
stuid = data['data']['cookie_info']['account_id']
bbs_cookie_url2 = 'https://api-takumi.mihoyo.com/auth/api/getMultiTokenByLoginTicket?login_ticket={}&token_types=3&uid={}'
data2 = (await get(url=bbs_cookie_url2.format(login_ticket.group().split('=')[1], stuid))).json()
return data2['data']['list'][0]['token']
else:
return None
async def exchange_action(info):
url = 'https://api-takumi.mihoyo.com/mall/v1/web/goods/exchange'
headers = {
'Accept': 'application/json, text/plain, */*',
'Accept-Encoding': 'gzip, deflate, br',
'Accept-Language': 'zh-CN,zh-Hans;q=0.9',
'Connection': 'keep-alive',
# 'Content-Length': '88',
'Content-Type': 'application/json;charset=utf-8',
'Cookie': info['cookie'],
'Host': 'api-takumi.mihoyo.com',
'Origin': 'https://webstatic.mihoyo.com',
'Referer': 'https://webstatic.mihoyo.com/',
'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 15_1 like Mac OS X) AppleWebKit/605.1.15 (KHtimeL, like Gecko) miHoYoBBS/2.14.1',
'x-rpc-app_version': '2.14.1',
'x-rpc-channel': 'appstore',
'x-rpc-client_type': '1',
'x-rpc-device_id': ''.join(random.sample(string.ascii_letters + string.digits, 32)).upper(),
'x-rpc-device_model': 'iPhone10,2',
'x-rpc-device_name': ''.join(random.sample(string.ascii_letters + string.digits, random.randrange(5))).upper(),
'x-rpc-sys_version': '15.1'
}
data = {
"app_id": 1,
"point_sn": "myb",
"goods_id": info['商品']['id'],
"exchange_num": 1,
"address_id": info['地址']['id']
}
if info['类型'] == '虚拟':
game_biz, region = await get_bbs_info(info, headers)
data['uid'] = info['uid']
data['game_biz'] = game_biz
data['region'] = region
if 'stoken' not in info['cookie']:
stoken = await get_stoken(info['cookie'])
info['cookie'] += f'stoken={stoken};'
exchange_url = 'https://api-takumi.mihoyo.com/mall/v1/web/goods/exchange'
flag = False
for _ in range(3):
exchange_res = (await post(url=exchange_url, headers=headers, json=data)).json()
if exchange_res['retcode'] == 0:
await get_bot().send_private_msg(user_id=info['user_id'],
message=f'你的米游币商品{info["商品"]["name"]}的兑换成功，结果为:\n{exchange_res["message"]}')
flag = True
break
await sleep(0.2)
try:
if not flag:
await get_bot().send_private_msg(user_id=info['user_id'],
message=f'你的米游币商品{info["商品"]["name"]}兑换失败:\n{exchange_res["message"]}')
except:
pass
path = Path() / 'data' / 'LittlePaimon' / 'myb_exchange' / f"{info['user_id']}-{info['商品']['id']}.json"
path.unlink()
@driver.on_startup
async def _():
path = Path() / 'data' / 'LittlePaimon' / 'myb_exchange'
path.mkdir(parents=True, exist_ok=True)
for exchange_data in path.iterdir():
info = load_json(path=exchange_data)
t = str(exchange_data).replace('data\\LittlePaimon\\myb_exchange\\', '').replace(
'data/LittlePaimon/myb_exchange/', '').replace(
'.json', '')
scheduler.add_job(
id=t,
replace_existing=True,
misfire_grace_time=5,
func=exchange_action,
trigger='date',
args=(info,),
next_run_time=datetime.datetime.strptime(info['商品']['time'], '%Y-%m-%d %H:%M:%S')
)
def get_exchange_info(user_id):
result = ''
path = Path() / 'data' / 'LittlePaimon' / 'myb_exchange'
path.mkdir(parents=True, exist_ok=True)
i = 1
for exchange_data in path.iterdir():
file_name = str(exchange_data).replace('data\\LittlePaimon\\myb_exchange\\', '').replace(
'data/LittlePaimon/myb_exchange/', '')
if file_name.startswith(user_id):
info = load_json(path=exchange_data)
result += f"{i}.{info['商品']['name']} {info['商品']['time']}\n"
if info['类型'] == '虚拟':
result += f"兑换至uid{info['uid']}\n"
else:
result += f"兑换至{info['地址']['地址']}\n"
i += 1
return result or '你还没有米游币兑换计划哦'
def delete_exchange_info(user_id):
path = Path() / 'data' / 'LittlePaimon' / 'myb_exchange'
path.mkdir(parents=True, exist_ok=True)
for exchange_data in path.iterdir():
file_name = str(exchange_data).replace('data\\LittlePaimon\\myb_exchange\\', '').replace(
'data/LittlePaimon/myb_exchange/', '')
if file_name.startswith(user_id):
exchange_data.unlink()

33
src/plugins/nonebot_plugin_myb_exchange/pyproject.toml
View File

@ -1,33 +0,0 @@
[tool.poetry]
name = "nonebot-plugin-myb-exchange"
version = "1.0.0"
description = "Nonebot2米游币商品自动兑换插件"
authors = ["惜月 <277073121@qq.com>"]
readme = "README.md"
license = "AGPL"
keywords = ["nonebot", "nonebot2", "mihoyo"]
homepage = "https://github.com/CMHopeSunshine/LittlePaimon/tree/Bot/src/plugins/nonebot_plugin_myb_exchange"
repository = "https://github.com/CMHopeSunshine/LittlePaimon/tree/Bot/src/plugins/nonebot_plugin_myb_exchange"
documentation = "https://github.com/CMHopeSunshine/LittlePaimon/tree/Bot/src/plugins/nonebot_plugin_myb_exchange#readme"
packages = [
{ include = "nonebot_plugin_myb_exchange/*.py", from = ".."}
]
[[tool.poetry.source]]
name = "ali"
default = true
url = "https://mirrors.aliyun.com/pypi/simple/"
[tool.poetry.dependencies]
python = "^3.8"
nonebot2 = "^2.0.0-beta.4"
nonebot-adapter-onebot = "^2.0.0-beta.1"
nonebot-plugin-apscheduler = "^0.1.2"
httpx = "^0.23.0"
[tool.poetry.dev-dependencies]
[build-system]
requires = ["poetry_core>=1.0.0"]
build-backend = "poetry.core.masonry.api"

88
src/plugins/nonebot_plugin_myb_exchange/utils.py
View File

@ -1,88 +0,0 @@
import json
from pathlib import Path
from typing import Dict, Optional, Any, Union
import httpx
async def get(url: str,
*,
headers: Optional[Dict[str, str]] = None,
params: Optional[Dict[str, Any]] = None,
timeout: Optional[int] = 20,
**kwargs) -> httpx.Response:
"""
说明：
httpx的get请求封装
参数：
:param url: url
:param headers: 请求头
:param params: params
:param timeout: 超时时间
"""
async with httpx.AsyncClient() as client:
return await client.get(url,
headers=headers,
params=params,
timeout=timeout,
**kwargs)
async def post(url: str,
*,
headers: Optional[Dict[str, str]] = None,
params: Optional[Dict[str, Any]] = None,
data: Optional[Dict[str, Any]] = None,
json: Optional[Dict[str, Union[Any, str]]] = None,
timeout: Optional[int] = 20,
**kwargs) -> httpx.Response:
"""
说明：
httpx的post请求封装
参数：
:param url: url
:param headers: 请求头
:param params: params
:param data: data
:param json: json
:param timeout: 超时时间
"""
async with httpx.AsyncClient() as client:
return await client.post(url,
headers=headers,
params=params,
data=data,
json=json,
timeout=timeout,
**kwargs)
def load_json(path: Union[Path, str], encoding: str = 'utf-8') -> dict:
"""
说明：
读取本地json文件，返回json字典。
参数：
:param path: 文件路径
:param encoding: 编码，默认为utf-8
:return: json字典
"""
if isinstance(path, str):
path = Path(path)
if not path.exists():
save_json({}, path, encoding)
return json.load(path.open('r', encoding=encoding))
def save_json(data, path: Union[Path, str], encoding: str = 'utf-8'):
"""
说明：
将数据保存至本地json文件
参数：
:param data: json数据
:param path: 文件路径
:param encoding: 编码，默认为utf-8
"""
if isinstance(path, str):
path = Path(path)
path.parent.mkdir(parents=True, exist_ok=True)
json.dump(data, path.open('w', encoding=encoding), ensure_ascii=False, indent=4)