Genshin-bots/GenshinUID
1
0
Fork 0
mirror of https://github.com/KimigaiiWuyi/GenshinUID.git synced 2025-05-08 04:55:51 +08:00
Code Issues Packages Projects Releases Wiki Activity

💥 为所有命令添加可配置的前缀 (#643)

Browse Source
This commit is contained in:
KimigaiiWuyi 2024-10-04 04:07:34 +08:00
parent 9b80ea76d1
commit c4a2b0d85c
10 changed files with 92 additions and 80 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
GenshinUID/__init__.py Normal file
View File

@ -0,0 +1,3 @@
from gsuid_core.sv import Plugins
Plugins(name="GenshinUID", prefix=["gs"], allow_empty_prefix=False)

2
GenshinUID/genshinuid_check/__init__.py
View File

@ -14,7 +14,7 @@ async def daily_refresh_charData():
await data_backup()
@sv_data_manger.on_fullmatch(('gs清除缓存'))
@sv_data_manger.on_fullmatch(('清除缓存'))
async def send_backup_msg(bot: Bot, ev: Event):
await data_backup()
for item in MAP_DATA.glob('*'):

2
GenshinUID/genshinuid_code/__init__.py
View File

@ -8,7 +8,7 @@ from .data_source import get_code_msg
sv_zzz_code = SV('原神前瞻兑换码')
@sv_zzz_code.on_fullmatch(f'gs兑换码')
@sv_zzz_code.on_fullmatch('兑换码')
async def get_sign_func(bot: Bot, ev: Event):
try:
codes = await get_code_msg()

122
GenshinUID/genshinuid_code/data_source.py
View File

@ -5,62 +5,70 @@ from typing import List, Union, Literal
from datetime import datetime, timezone, timedelta
from httpx import AsyncClient
from gsuid_core.utils.api.mys.api import BBS_URL
TZ = timezone(timedelta(hours=8))
async def get_data(
type: Literal["activity", "index", "code"], data: dict = {}
) -> dict:
"""米哈游接口请求"""
MYJ = (
f'{BBS_URL}/painter/api/user_instant/list?offset=0&size=20&uid=75276550',
) # 米游姬个人中心
url = {
'act_id': MYJ,
'index': 'https://api-takumi.mihoyo.com/event/miyolive/index',
'code': 'https://api-takumi-static.mihoyo.com/event/miyolive/refreshCode',
}
async def get_data(
type: Literal['index', 'code', 'act_id'],
data: dict = {},
) -> dict:
'''米哈游接口请求'''
url = {
"act_id": "https://bbs-api.mihoyo.com/painter/api/user_instant/list?offset=0&size=20&uid=75276550", # 米游姬个人中心
"index": "https://api-takumi.mihoyo.com/event/miyolive/index",
"code": "https://api-takumi-static.mihoyo.com/event/miyolive/refreshCode",
}
async with AsyncClient() as client:
try:
if type == "index":
if type == 'index':
res = await client.get(
url[type], headers={"x-rpc-act_id": data.get("actId", "")}
url[type], headers={'x-rpc-act_id': data.get('actId', '')}
)
elif type == "code":
elif type == 'code':
res = await client.get(
url[type],
params={
"version": data.get("version", ""),
"time": f"{int(time())}",
'version': data.get('version', ''),
'time': f'{int(time())}',
},
headers={"x-rpc-act_id": data.get("actId", "")},
headers={'x-rpc-act_id': data.get('actId', '')},
)
else:
res = await client.get(url[type])
return res.json()
except Exception as e:
return {"error": f"[{e.__class__.__name__}] {type} 接口请求错误"}
return {'error': f'[{e.__class__.__name__}] {type} 接口请求错误'}
async def get_act_id() -> str:
"""获取 ``act_id``"""
'''获取 ``act_id``'''
ret = await get_data("act_id")
if ret.get("error") or ret.get("retcode") != 0:
return ""
ret = await get_data('act_id')
if ret.get('error') or ret.get('retcode') != 0:
return ''
act_id = ""
keywords = ["前瞻特别节目"]
for p in ret["data"]["list"]:
post = p.get("post", {}).get("post", {})
act_id = ''
keywords = ['前瞻特别节目']
for p in ret['data']['list']:
post = p.get('post', {}).get('post', {})
if not post:
continue
if not all(word in post["subject"] for word in keywords):
if not all(word in post['subject'] for word in keywords):
continue
shit = json.loads(post["structured_content"])
shit = json.loads(post['structured_content'])
for segment in shit:
link = segment.get("attributes", {}).get("link", "")
if "观看" in segment.get("insert", "") and link:
matched = findall(r"act_id=(.*?)\&", link)
link = segment.get('attributes', {}).get('link', '')
if '观看' in segment.get('insert', '') and link:
matched = findall(r'act_id=(.*?)\&', link)
if matched:
act_id = matched[0]
if act_id:
@ -70,73 +78,73 @@ async def get_act_id() -> str:
async def get_live_data(act_id: str) -> dict:
"""获取直播数据，尤其是 ``code_ver``"""
'''获取直播数据，尤其是 ``code_ver``'''
ret = await get_data("index", {"actId": act_id})
if ret.get("error") or ret.get("retcode") != 0:
return {"error": ret.get("error") or "前瞻直播数据异常"}
ret = await get_data('index', {'actId': act_id})
if ret.get('error') or ret.get('retcode') != 0:
return {'error': ret.get('error') or '前瞻直播数据异常'}
live_raw = ret["data"]["live"]
live_temp = json.loads(ret["data"]["template"])
live_raw = ret['data']['live']
live_temp = json.loads(ret['data']['template'])
live_data = {
"code_ver": live_raw["code_ver"],
"title": live_raw["title"].replace("特别直播", ""),
"header": live_temp["kvDesktop"],
"room": live_temp["liveConfig"][0]["desktop"],
'code_ver': live_raw['code_ver'],
'title': live_raw['title'].replace('特别直播', ''),
'header': live_temp['kvDesktop'],
'room': live_temp['liveConfig'][0]['desktop'],
}
now = datetime.fromtimestamp(time(), TZ)
start = datetime.strptime(live_raw["start"], "%Y-%m-%d %H:%M:%S").replace(
start = datetime.strptime(live_raw['start'], '%Y-%m-%d %H:%M:%S').replace(
tzinfo=TZ
)
if now < start:
live_data["start"] = live_raw["start"]
live_data['start'] = live_raw['start']
return live_data
async def get_code(version: str, act_id: str) -> Union[dict, List[dict]]:
"""获取兑换码"""
'''获取兑换码'''
ret = await get_data("code", {"version": version, "actId": act_id})
if ret.get("error") or ret.get("retcode") != 0:
return {"error": ret.get("error") or "兑换码数据异常"}
ret = await get_data('code', {'version': version, 'actId': act_id})
if ret.get('error') or ret.get('retcode') != 0:
return {'error': ret.get('error') or '兑换码数据异常'}
code_data = []
for code_info in ret["data"]["code_list"]:
remove_tag = compile("<.*?>")
for code_info in ret['data']['code_list']:
remove_tag = compile('<.*?>')
code_data.append(
{
"items": sub(remove_tag, "", code_info["title"]),
"code": code_info["code"],
'items': sub(remove_tag, '', code_info['title']),
'code': code_info['code'],
}
)
return code_data
async def get_code_msg() -> str:
"""生成最新前瞻直播兑换码消息"""
'''生成最新前瞻直播兑换码消息'''
act_id = await get_act_id()
if not act_id:
return "暂无前瞻直播资讯！"
return '暂无前瞻直播资讯！'
live_data = await get_live_data(act_id)
if live_data.get("error"):
return live_data["error"]
if live_data.get('error'):
return live_data['error']
code_data = await get_code(live_data["code_ver"], act_id)
code_data = await get_code(live_data['code_ver'], act_id)
if isinstance(code_data, dict):
return code_data["error"]
return code_data['error']
code_msg = f'{live_data["title"]}\n'
# 三个兑换码
index = 0
for code in code_data:
index = index + 1
if code.get("code"):
if code.get('code'):
# 该兑换码已开放
code_msg += f'{code["items"]}:\n{code["code"]}\n'
else:
# 该兑换码未开放
code_msg += f"第 {index} 个兑换码暂未发放\n"
code_msg += f'第 {index} 个兑换码暂未发放\n'
return code_msg.strip()

8
GenshinUID/genshinuid_config/__init__.py
View File

@ -13,14 +13,14 @@ from .set_config import set_push_value, set_config_func
sv_self_config = SV('原神配置')
@sv_self_config.on_fullmatch(('gs配置', '原神配置'))
@sv_self_config.on_fullmatch(('配置', '原神配置'))
async def send_config_card(bot: Bot, ev: Event):
logger.info('开始执行[gs配置]')
im = await draw_config_img(ev.bot_id)
await bot.send(im)
@sv_self_config.on_prefix(('gs设置'))
@sv_self_config.on_prefix(('设置'))
async def send_config_ev(bot: Bot, ev: Event):
logger.info('开始执行[设置阈值信息]')
uid = await GsBind.get_uid_by_game(ev.user_id, ev.bot_id)
@ -40,14 +40,14 @@ async def send_config_ev(bot: Bot, ev: Event):
# 开启 自动签到 和 推送树脂提醒 功能
@sv_self_config.on_prefix(('gs开启', 'gs关闭'))
@sv_self_config.on_prefix(('开启', '关闭'))
async def open_switch_func(bot: Bot, ev: Event):
user_id = ev.user_id
config_name = ev.text
logger.info(f'[{user_id}]尝试[{ev.command[2:]}]了[{ev.text}]功能')
if ev.command == 'gs开启':
if '开启' in ev.command:
query = True
gid = ev.group_id if ev.group_id else 'on'
else:

2
GenshinUID/genshinuid_help/__init__.py
View File

@ -10,7 +10,7 @@ from .get_help import ICON, get_core_help
sv_gs_help = SV('gs帮助')
@sv_gs_help.on_fullmatch(('gs帮助'))
@sv_gs_help.on_fullmatch(('gs帮助', '帮助'))
async def send_help_img(bot: Bot, ev: Event):
logger.info('开始执行[gs帮助]')
im = await get_core_help()

3
GenshinUID/genshinuid_help/get_help.py
View File

@ -4,6 +4,7 @@ from typing import Dict, Union
import aiofiles
from PIL import Image
from msgspec import json as msgjson
from gsuid_core.sv import get_plugin_prefix
from gsuid_core.help.model import PluginHelp
from gsuid_core.help.draw_plugin_help import get_help
from gsuid_core.help.draw_new_plugin_help import get_new_help
@ -45,7 +46,7 @@ async def get_core_help() -> Union[bytes, str]:
plugin_info={f'v{GenshinUID_version}': ''},
plugin_icon=Image.open(ICON),
plugin_help=await get_help_data(),
plugin_prefix='',
plugin_prefix=get_plugin_prefix('GenshinUID'),
help_mode='dark',
banner_bg=BANNER_BG,
cag_bg=CAG_BG,

26
GenshinUID/genshinuid_help/help.json
View File

@ -194,12 +194,12 @@
]
},
"私人服务": {
"desc": "需要加 [gs开启/设置] 前缀",
"desc": "需要加 [开启/设置] 前缀",
"data": [
{
"name": "推送",
"desc": "所有推送总开关",
"eg": "gs开启推送",
"eg": "开启推送",
"need_ck": true,
"need_sk": false,
"need_admin": false
@ -207,7 +207,7 @@
{
"name": "自动签到",
"desc": "每晚0:30进行米游社签到",
"eg": "gs开启自动签到",
"eg": "开启自动签到",
"need_ck": true,
"need_sk": false,
"need_admin": false
@ -215,7 +215,7 @@
{
"name": "自动米游币",
"desc": "每晚1:16进行米游币获取",
"eg": "gs开启自动米游币",
"eg": "开启自动米游币",
"need_ck": false,
"need_sk": true,
"need_admin": false
@ -223,7 +223,7 @@
{
"name": "[推送项]",
"desc": "体力/派遣/宝钱/质变仪推送",
"eg": "gs开启体力、gs开启派遣",
"eg": "开启体力",
"need_ck": true,
"need_sk": false,
"need_admin": false
@ -231,7 +231,7 @@
{
"name": "[推送项]阈值",
"desc": "设置推送的阈值",
"eg": "gs设置体力阈值140",
"eg": "设置体力阈值140",
"need_ck": true,
"need_sk": false,
"need_admin": false
@ -327,9 +327,9 @@
"desc": "数据源由AKACV提供",
"data": [
{
"name": "gs开启排名系统",
"name": "开启排名系统",
"desc": "开启排名系统之后可正常使用",
"eg": "gs开启排名系统",
"eg": "开启排名系统",
"need_ck": false,
"need_sk": false,
"need_admin": true
@ -590,9 +590,9 @@
"desc": "有关插件的一些数据",
"data": [
{
"name": "gs配置",
"name": "配置",
"desc": "查看全局的插件配置",
"eg": "gs配置",
"eg": "配置",
"need_ck": false,
"need_sk": false,
"need_admin": false
@ -614,9 +614,9 @@
"need_admin": true
},
{
"name": "gs更新",
"desc": "可尝试gs强制更新",
"eg": "gs强行强制更新",
"name": "更新",
"desc": "可尝试强制更新",
"eg": "强行强制更新",
"need_ck": false,
"need_sk": false,
"need_admin": true

2
GenshinUID/genshinuid_resource/__init__.py
View File

@ -8,7 +8,7 @@ from ..utils.resource.download_all_resource import download_all_resource
sv_download_config = SV('下载资源', pm=2)
@sv_download_config.on_fullmatch(('下载全部资源', 'gs下载全部资源'))
@sv_download_config.on_fullmatch(('下载全部资源'))
async def send_download_resource_msg(bot: Bot, ev: Event):
await bot.send('正在开始下载~可能需要较久的时间!')
im = await download_all_resource()

2
GenshinUID/genshinuid_topup/__init__.py
View File

@ -11,7 +11,7 @@ INPUTTIP = '''请输入正确的商品编号(0~6), 如:原神充值6
也可以直接输入物品名称或价格，如:原神充值月卡 | pay648'''
@sv_topup.on_command(('gsrc', '原神充值', 'pay'))
@sv_topup.on_command(('原神充值', 'pay'))
async def send_qrcode_login(bot: Bot, ev: Event):
await bot.logger.info('开始执行[原神充值]')
value = ev.text