优化原牌列表,支持v2_cookie以及通过stoken获取cookie_token,修复深渊配队版本号

This commit is contained in:
CMHopeSunshine 2022-12-03 16:23:26 +08:00
parent 8f00cc7174
commit 1163d63a3a
7 changed files with 327 additions and 150 deletions

View File

@ -10,7 +10,7 @@ BASE_API = 'https://www.youchuang.fun'
TEAM_RATE_API = f'{BASE_API}/gamerole/formationRate' TEAM_RATE_API = f'{BASE_API}/gamerole/formationRate'
VERSION_API = 'https://api-cloudgame-static.mihoyo.com/hk4e_cg_cn/gamer/api/getFunctionShieldNew?client_type=1' VERSION_API = 'https://api-cloudgame-static.mihoyo.com/hk4e_cg_cn/gamer/api/getFunctionShieldNew?client_type=1'
VERSION = 3.1 # 尚未更新3.2深渊 VERSION = 3.2
HEADERS = { HEADERS = {
'Host': 'www.youchuang.fun', 'Host': 'www.youchuang.fun',
'Referer': 'https://servicewechat.com/wxce4dbe0cb0f764b3/91/page-frame.html', 'Referer': 'https://servicewechat.com/wxce4dbe0cb0f764b3/91/page-frame.html',

View File

@ -13,8 +13,7 @@ from LittlePaimon.config import config
from LittlePaimon.database import LastQuery, PrivateCookie, PublicCookie, Character, PlayerInfo, DailyNoteSub, \ from LittlePaimon.database import LastQuery, PrivateCookie, PublicCookie, Character, PlayerInfo, DailyNoteSub, \
MihoyoBBSSub MihoyoBBSSub
from LittlePaimon.utils import logger, NICKNAME from LittlePaimon.utils import logger, NICKNAME
from LittlePaimon.utils.api import get_bind_game_info, get_stoken_by_cookie from LittlePaimon.utils.api import get_bind_game_info, get_stoken_by_login_ticket, get_cookie_token_by_stoken
from LittlePaimon.utils.message import recall_message
__plugin_meta__ = PluginMetadata( __plugin_meta__ = PluginMetadata(
name='原神绑定', name='原神绑定',
@ -69,39 +68,54 @@ clear = on_command('清除无效用户', permission=SUPERUSER, block=True, prior
@ysb.handle() @ysb.handle()
async def _(event: MessageEvent, msg: Message = CommandArg()): async def _(event: MessageEvent, msg: Message = CommandArg()):
msg = msg.extract_plain_text().strip() msg = msg.extract_plain_text().strip()
if uid := re.search(r'[125]\d{8}', msg): if uid := re.match(r'[125]\d{8}', msg):
await LastQuery.update_or_create(user_id=str(event.user_id), await LastQuery.update_or_create(user_id=str(event.user_id),
defaults={'uid': uid.group(), 'last_time': datetime.datetime.now()}) defaults={'uid': uid[0], 'last_time': datetime.datetime.now()})
msg = msg.replace(uid.group(), '').strip() msg = msg.replace(uid[0], '').strip()
if not msg: if not msg:
await ysb.finish(f'成功绑定uid为{uid.group()}如果还需绑定cookie可看教程\ndocs.qq.com/doc/DQ3JLWk1vQVllZ2Z1', await ysb.finish(f'成功绑定uid为{uid[0]}如果还需绑定cookie可看教程\ndocs.qq.com/doc/DQ3JLWk1vQVllZ2Z1',
at_sender=True) at_sender=True)
if msg: if msg:
if data := await get_bind_game_info(msg): if msg in {'cookie', 'Cookie', 'ck', 'CK'}:
game_name = data['nickname'] await ysb.finish(f'你在和{NICKNAME}开玩笑嘛请看教程获取Cookie\ndocs.qq.com/doc/DQ3JLWk1vQVllZ2Z1', at_sender=True)
game_uid = data['game_role_id'] if mys_id := re.search(r'(?:(?:login_uid|account_mid|account_id|stmid|ltmid|stuid|ltuid)(?:_v2)?)=(\d+)', msg):
mys_id = data['mys_id'] mys_id = mys_id[1]
await LastQuery.update_or_create(user_id=str(event.user_id),
defaults={'uid': game_uid, 'last_time': datetime.datetime.now()})
logger.info('原神Cookie', '', {'用户': str(event.user_id), 'uid': game_uid}, '成功绑定cookie', True)
if 'login_ticket' in msg and (stoken := await get_stoken_by_cookie(msg)):
await PrivateCookie.update_or_create(user_id=str(event.user_id), uid=game_uid, mys_id=mys_id,
defaults={'cookie': msg,
'stoken': f'stuid={mys_id};stoken={stoken};'})
await ysb.send(f'{game_name}成功绑定cookie{game_uid},开始愉快地享用{NICKNAME}吧!', at_sender=True)
else: else:
await PrivateCookie.update_or_create(user_id=str(event.user_id), uid=game_uid, mys_id=mys_id, await ysb.finish('Cookie无效缺少account_id、login_uid或stuid字段\n获取cookie的教程\ndocs.qq.com/doc/DQ3JLWk1vQVllZ2Z1',
defaults={'cookie': msg})
await ysb.send(f'{game_name}成功绑定cookie{game_uid}但是cookie中没有login_ticket米游币相关功能无法使用哦',
at_sender=True) at_sender=True)
if not isinstance(event, PrivateMessageEvent): cookie_token_match = re.search(r'(?:cookie_token|cookie_token_v2)=([0-9a-zA-Z]+)', msg)
if await recall_message(event): cookie_token = cookie_token_match[1] if cookie_token_match else None
await ysb.finish(f'当前非私聊对话,{NICKNAME}帮你把cookie撤回啦') login_ticket_match = re.search(r'(?:login_ticket|login_ticket_v2)=([0-9a-zA-Z]+)', msg)
login_ticket = login_ticket_match[1] if login_ticket_match else None
stoken_match = re.search(r'(?:stoken|stoken_v2)=([0-9a-zA-Z]+)', msg)
stoken = stoken_match[1] if stoken_match else None
if login_ticket and not stoken:
# 如果有login_ticket但没有stoken就通过login_ticket获取stoken
stoken = await get_stoken_by_login_ticket(login_ticket, mys_id)
if stoken and not cookie_token:
# 如果有stoken但没有cookie_token就通过stoken获取cookie_token
cookie_token = await get_cookie_token_by_stoken(stoken, mys_id)
if not cookie_token:
await ysb.finish('Cookie无效缺少cookie_token或login_ticket字段\n获取cookie的教程\ndocs.qq.com/doc/DQ3JLWk1vQVllZ2Z1', at_sender=True)
if game_info := await get_bind_game_info(f'account_id={mys_id};cookie_token={cookie_token}', mys_id):
if not game_info['list']:
await ysb.finish('该账号尚未绑定任何游戏,请确认账号无误~', at_sender=True)
if not (genshin_games := [{'uid': game['game_role_id'], 'nickname': game['nickname']} for game in game_info['list'] if game['game_id'] == 2]):
await ysb.finish('该账号尚未绑定原神,请确认账号无误~', at_sender=True)
await LastQuery.update_or_create(user_id=str(event.user_id),
defaults={'uid': genshin_games[0]['uid'], 'last_time': datetime.datetime.now()})
send_msg = ''
for info in genshin_games:
send_msg += f'{info["nickname"]}({info["uid"]}) '
await PrivateCookie.update_or_create(user_id=str(event.user_id), uid=info['uid'], mys_id=mys_id,
defaults={'cookie': f'account_id={mys_id};cookie_token={cookie_token}',
'stoken': f'stuid={mys_id};stoken={stoken};' if stoken else None})
await ysb.finish(f'玩家{send_msg.strip()}绑定Cookie{"和Stoken" if stoken else ""}成功{"" if stoken else "当未能绑定Stoken"}'
f'{"当前非私聊对话建议将Cookie撤回哦" if not isinstance(event, PrivateMessageEvent) else ""}', at_sender=True)
else: else:
await ysb.finish(f'当前非私聊对话,{NICKNAME}建议你绑定完将cookie撤回哦') await ysb.finish(
else: 'Cookie无效请确认是否已过期\n获取cookie的教程\ndocs.qq.com/doc/DQ3JLWk1vQVllZ2Z1',
logger.info('原神Cookie', '', {'用户': str(event.user_id)}, '绑定失败cookie已失效', False) at_sender=True)
await ysb.finish('这个cookie无效哦请确认是否正确\n获取cookie的教程\ndocs.qq.com/doc/DQ3JLWk1vQVllZ2Z1\n', at_sender=True)
elif config.CookieWeb_enable: elif config.CookieWeb_enable:
await ysb.finish( await ysb.finish(
f'获取cookie的教程\ndocs.qq.com/doc/DQ3JLWk1vQVllZ2Z1\n获取后,使用[ysb cookie]指令绑定或前往{config.CookieWeb_url}网页添加绑定', f'获取cookie的教程\ndocs.qq.com/doc/DQ3JLWk1vQVllZ2Z1\n获取后,使用[ysb cookie]指令绑定或前往{config.CookieWeb_url}网页添加绑定',
@ -119,7 +133,7 @@ async def _(event: MessageEvent):
if ck: if ck:
msg = f'{event.sender.card or event.sender.nickname}当前绑定情况:\n' msg = f'{event.sender.card or event.sender.nickname}当前绑定情况:\n'
for ck_ in ck: for ck_ in ck:
if await get_bind_game_info(ck_.cookie): if await get_bind_game_info(ck_.cookie, ck_.mys_id):
msg += f'{ck.index(ck_) + 1}.{ck_.uid}(有效)\n' msg += f'{ck.index(ck_) + 1}.{ck_.uid}(有效)\n'
else: else:
msg += f'{ck.index(ck_) + 1}.{ck_.uid}(已失效)\n' msg += f'{ck.index(ck_) + 1}.{ck_.uid}(已失效)\n'
@ -172,14 +186,18 @@ async def _(event: MessageEvent):
useless_private = [] useless_private = []
useless_public = [] useless_public = []
for cookie in private_cookies: for cookie in private_cookies:
if not await get_bind_game_info(cookie.cookie): if not await get_bind_game_info(cookie.cookie, cookie.mys_id):
useless_private.append(cookie.uid) useless_private.append(cookie.uid)
await cookie.delete() await cookie.delete()
await sleep(1) await sleep(1)
for cookie in public_cookies: for cookie in public_cookies:
if not await get_bind_game_info(cookie.cookie): if mys_id := re.search(r'(?:(?:login_uid|account_mid|account_id|stmid|ltmid|stuid|ltuid)(?:_v2)?)=(\d+)', cookie.cookie):
mys_id = mys_id[1]
if not await get_bind_game_info(cookie.cookie, mys_id):
useless_public.append(str(cookie.id)) useless_public.append(str(cookie.id))
await cookie.delete() await cookie.delete()
else:
useless_public.append(str(cookie.id))
await sleep(0.5) await sleep(0.5)
msg = f'当前共{len(public_cookies)}个公共ck{len(private_cookies)}个私人ck。\n' msg = f'当前共{len(public_cookies)}个公共ck{len(private_cookies)}个私人ck。\n'
if useless_public: if useless_public:
@ -196,13 +214,32 @@ async def _(event: MessageEvent):
@pck.handle() @pck.handle()
async def _(event: MessageEvent, msg: Message = CommandArg()): async def _(event: MessageEvent, msg: Message = CommandArg()):
if msg := msg.extract_plain_text().strip(): if msg := msg.extract_plain_text().strip():
if await get_bind_game_info(msg, True): if mys_id := re.search(r'(?:(?:login_uid|account_mid|account_id|stmid|ltmid|stuid|ltuid)(?:_v2)?)=(\d+)', msg):
ck = await PublicCookie.create(cookie=msg) mys_id = mys_id[1]
else:
await pck.finish('Cookie无效缺少account_id、login_uid或stuid字段\n获取cookie的教程\ndocs.qq.com/doc/DQ3JLWk1vQVllZ2Z1',
at_sender=True)
cookie_token_match = re.search(r'(?:cookie_token|cookie_token_v2)=([0-9a-zA-Z]+)', msg)
cookie_token = cookie_token_match[1] if cookie_token_match else None
login_ticket_match = re.search(r'(?:login_ticket|login_ticket_v2)=([0-9a-zA-Z]+)', msg)
login_ticket = login_ticket_match[1] if login_ticket_match else None
stoken_match = re.search(r'(?:stoken|stoken_v2)=([0-9a-zA-Z]+)', msg)
stoken = stoken_match[1] if stoken_match else None
if login_ticket and not stoken:
# 如果有login_ticket但没有stoken就通过login_ticket获取stoken
stoken = await get_stoken_by_login_ticket(login_ticket, mys_id)
if stoken and not cookie_token:
# 如果有stoken但没有cookie_token就通过stoken获取cookie_token
cookie_token = await get_cookie_token_by_stoken(stoken, mys_id)
if not cookie_token:
await pck.finish('Cookie无效缺少cookie_token或login_ticket字段\n获取cookie的教程\ndocs.qq.com/doc/DQ3JLWk1vQVllZ2Z1', at_sender=True)
if await get_bind_game_info(f'account_id={mys_id};cookie_token={cookie_token}', mys_id):
ck = await PublicCookie.create(cookie=f'account_id={mys_id};cookie_token={cookie_token}')
logger.info('原神Cookie', f'{ck.id}号公共cookie', None, '添加成功', True) logger.info('原神Cookie', f'{ck.id}号公共cookie', None, '添加成功', True)
await pck.finish(f'成功添加{ck.id}号公共cookie', at_sender=True) await pck.finish(f'成功添加{ck.id}号公共cookie', at_sender=True)
else: else:
logger.info('原神Cookie', '公共cookie', None, '添加失败cookie已失效', True) logger.info('原神Cookie', '公共cookie', None, '添加失败cookie已失效', True)
await pck.finish('这个cookie无效哦请确认是否正确\n获取cookie的教程\ndocs.qq.com/doc/DQ3JLWk1vQVllZ2Z1\n', at_sender=True) await pck.finish('Cookie无效请确认是否已过期\n获取cookie的教程\ndocs.qq.com/doc/DQ3JLWk1vQVllZ2Z1', at_sender=True)
else: else:
await pck.finish('获取cookie的教程\ndocs.qq.com/doc/DQ3JLWk1vQVllZ2Z1\n获取到后,用[添加公共ck cookie]指令添加', await pck.finish('获取cookie的教程\ndocs.qq.com/doc/DQ3JLWk1vQVllZ2Z1\n获取到后,用[添加公共ck cookie]指令添加',
at_sender=True) at_sender=True)

View File

@ -1,7 +1,7 @@
import datetime import datetime
from nonebot import on_regex, on_command from nonebot import on_regex, on_command
from nonebot.adapters.onebot.v11 import MessageEvent, Message, MessageSegment, GroupMessageEvent from nonebot.adapters.onebot.v11 import MessageEvent, Message, MessageSegment, GroupMessageEvent, PrivateMessageEvent, Bot
from nonebot.adapters.onebot.v11.exception import ActionFailed from nonebot.adapters.onebot.v11.exception import ActionFailed
from nonebot.adapters.onebot.v11.helpers import HandleCancellation from nonebot.adapters.onebot.v11.helpers import HandleCancellation
from nonebot.params import RegexDict, ArgPlainText, CommandArg, Arg from nonebot.params import RegexDict, ArgPlainText, CommandArg, Arg
@ -18,7 +18,7 @@ from LittlePaimon.utils.tool import freq_limiter
from .draw_daily_material import draw_material from .draw_daily_material import draw_material
from .draw_map import init_map, draw_map, get_full_map from .draw_map import init_map, draw_map, get_full_map
from .SereniteaPot import draw_pot_materials from .SereniteaPot import draw_pot_materials
from .card import get_match_card, CARD_API, get_card_list from .card import get_match_card, CARD_API, get_card_resources
__paimon_help__ = { __paimon_help__ = {
'type': '原神Wiki', 'type': '原神Wiki',
@ -216,7 +216,7 @@ def create_wiki_matcher(pattern: str, help_fun: str, help_name: str):
if '武器' in state['type']: if '武器' in state['type']:
state['type'] = '武器' state['type'] = '武器'
# state['img_url'] = 'https://static.cherishmoon.fun/LittlePaimon/WeaponMaps/{}.jpg' # state['img_url'] = 'https://static.cherishmoon.fun/LittlePaimon/WeaponMaps/{}.jpg'
state['img_url'] = 'https://ghproxy.com/https://raw.githubusercontent.com/Nwflower/genshin-atlas/master/weapon/{}.png' state['img_url'] = 'https://github.cherishmoon.fun/https://raw.githubusercontent.com/Nwflower/genshin-atlas/master/weapon/{}.png'
elif '圣遗物' in state['type']: elif '圣遗物' in state['type']:
state['type'] = '圣遗物' state['type'] = '圣遗物'
state['img_url'] = 'https://static.cherishmoon.fun/LittlePaimon/ArtifactMaps/{}.jpg' state['img_url'] = 'https://static.cherishmoon.fun/LittlePaimon/ArtifactMaps/{}.jpg'
@ -230,10 +230,10 @@ def create_wiki_matcher(pattern: str, help_fun: str, help_name: str):
state['type'] = '角色' state['type'] = '角色'
# state['img_url'] = 'https://static.cherishmoon.fun/LittlePaimon/RoleMaterials/{}材料.jpg' # state['img_url'] = 'https://static.cherishmoon.fun/LittlePaimon/RoleMaterials/{}材料.jpg'
state[ state[
'img_url'] = 'https://ghproxy.com/https://raw.githubusercontent.com/Nwflower/genshin-atlas/master/material%20for%20role/{}.png' 'img_url'] = 'https://github.cherishmoon.fun/https://raw.githubusercontent.com/Nwflower/genshin-atlas/master/material%20for%20role/{}.png'
elif state['type'] == '角色图鉴': elif state['type'] == '角色图鉴':
state['type'] = '角色' state['type'] = '角色'
state['img_url'] = 'https://ghproxy.com/https://raw.githubusercontent.com/CMHopeSunshine/GenshinWikiMap/master/results/character_map/{}.jpg' state['img_url'] = 'https://github.cherishmoon.fun/https://raw.githubusercontent.com/CMHopeSunshine/GenshinWikiMap/master/results/character_map/{}.jpg'
elif state['type'] == '收益曲线': elif state['type'] == '收益曲线':
state['type'] = '角色' state['type'] = '角色'
state['img_url'] = 'https://static.cherishmoon.fun/LittlePaimon/blue/{}.jpg' state['img_url'] = 'https://static.cherishmoon.fun/LittlePaimon/blue/{}.jpg'
@ -321,6 +321,8 @@ async def _(state: T_State, name: str = ArgPlainText('name')):
await card_wiki.finish(MessageBuild.Text(f'暂时没有{name}的卡牌图鉴')) await card_wiki.finish(MessageBuild.Text(f'暂时没有{name}的卡牌图鉴'))
if name in matches: if name in matches:
await card_wiki.finish(MessageSegment.image(CARD_API.format(name))) await card_wiki.finish(MessageSegment.image(CARD_API.format(name)))
if len(matches) == 1:
await card_wiki.finish(MessageSegment.image(CARD_API.format(matches[0])))
if 'choice' not in state: if 'choice' not in state:
msg = f'你要查询的卡牌是:\n' msg = f'你要查询的卡牌是:\n'
msg += '\n'.join([f'{int(i) + 1}. {name}' for i, name in enumerate(matches)]) msg += '\n'.join([f'{int(i) + 1}. {name}' for i, name in enumerate(matches)])
@ -353,6 +355,21 @@ async def _(state: T_State, choice: str = ArgPlainText('choice')):
@card_wiki_list.handle() @card_wiki_list.handle()
async def _(): async def _(bot: Bot, event: MessageEvent):
result = await get_card_list() result = await get_card_resources()
await card_wiki_list.finish(result or '暂时没有卡牌图鉴') if not result:
await card_wiki_list.finish('读取七圣召唤卡牌列表失败')
msg = [{'type': 'node', 'data': {'name': NICKNAME, 'uin': event.self_id, 'content': f'{type}\n' + '\n'.join(cards)}} for type, cards in result.items()]
msg.insert(0, {'type': 'node', 'data': {'name': NICKNAME, 'uin': event.self_id, 'content': '七圣召唤卡牌列表如下'}})
try:
if isinstance(event, GroupMessageEvent):
await bot.call_api('send_group_forward_msg', group_id=event.group_id, messages=msg)
elif isinstance(event, PrivateMessageEvent):
await bot.call_api('send_private_forward_msg', user_id=event.user_id, messages=msg)
else:
msg = '七圣召唤卡牌列表:'
for type, cards in result.items():
msg += f'{type}\n' + '\n'.join([' '.join(cards[i:i + 3]) for i in range(0, len(cards), 3)])
await card_wiki_list.send(msg)
except ActionFailed:
await card_wiki_list.finish('七圣召唤卡牌列表发送失败,账号可能被风控')

View File

@ -1,35 +1,27 @@
import contextlib import contextlib
import difflib import difflib
from LittlePaimon.utils import scheduler from typing import Optional
from ruamel import yaml
from LittlePaimon.utils.requests import aiorequests from LittlePaimon.utils.requests import aiorequests
card_list = [] CARD_RESOURCES_API = 'https://github.cherishmoon.fun/https://raw.githubusercontent.com/Nwflower/Atlas/master/resource/text/card.yaml'
CARD_RESOURCES_API = 'https://api.github.com/repos/Nwflower/genshin-atlas/contents/card' CARD_API = 'https://github.cherishmoon.fun/https://raw.githubusercontent.com/Nwflower/genshin-atlas/master/card/{}.png'
CARD_API = 'https://ghproxy.com/https://raw.githubusercontent.com/Nwflower/genshin-atlas/master/card/{}.png'
async def update_card_list(): async def get_card_resources() -> Optional[dict]:
with contextlib.suppress(Exception): with contextlib.suppress(Exception):
resp = await aiorequests.get(CARD_RESOURCES_API) resp = await aiorequests.get(CARD_RESOURCES_API)
if resp.status_code == 200: data = yaml.load(resp.content, Loader=yaml.Loader)
for card in resp.json(): data.pop('召唤')
if (name := card['name'].replace('.png', '')) not in card_list: return data
card_list.append(name)
return None return None
async def get_card_list():
if not card_list:
await update_card_list()
return '七圣召唤原牌列表:\n' + '\n'.join([' '.join(card_list[i:i + 3]) for i in range(0, len(card_list), 3)])
async def get_match_card(name: str): async def get_match_card(name: str):
if not card_list: if not (data := await get_card_resources()):
await update_card_list() return None
return difflib.get_close_matches(name, card_list, cutoff=0.6, n=10) if card_list else None matches = []
for cards in data.values():
matches.extend(difflib.get_close_matches(name, cards, cutoff=0.6, n=10))
@scheduler.scheduled_job('cron', minute='*/30') return matches
async def _():
await update_card_list()

View File

@ -4,7 +4,7 @@ from nonebot import get_driver
from .logger import logger from .logger import logger
from .scheduler import scheduler from .scheduler import scheduler
__version__ = '3.0.0rc4' __version__ = '3.0.0rc5'
DRIVER = get_driver() DRIVER = get_driver()
try: try:

View File

@ -1,3 +1,4 @@
import contextlib
import hashlib import hashlib
import json import json
import random import random
@ -27,6 +28,9 @@ SIGN_INFO_API = 'https://api-takumi.mihoyo.com/event/bbs_sign_reward/info'
SIGN_REWARD_API = 'https://api-takumi.mihoyo.com/event/bbs_sign_reward/home' SIGN_REWARD_API = 'https://api-takumi.mihoyo.com/event/bbs_sign_reward/home'
SIGN_ACTION_API = 'https://api-takumi.mihoyo.com/event/bbs_sign_reward/sign' SIGN_ACTION_API = 'https://api-takumi.mihoyo.com/event/bbs_sign_reward/sign'
AUTHKEY_API = 'https://api-takumi.mihoyo.com/binding/api/genAuthKey' AUTHKEY_API = 'https://api-takumi.mihoyo.com/binding/api/genAuthKey'
STOKEN_API = 'https://api-takumi.mihoyo.com/auth/api/getMultiTokenByLoginTicket'
COOKIE_TOKEN_API = 'https://api-takumi.mihoyo.com/auth/api/getCookieAccountInfoBySToken'
LOGIN_TICKET_INFO_API = 'https://webapi.account.mihoyo.com/Api/cookie_accountinfo_by_loginticket'
def md5(text: str) -> str: def md5(text: str) -> str:
@ -213,27 +217,23 @@ async def get_cookie(user_id: str, uid: str, check: bool = True, own: bool = Fal
return None, '' return None, ''
async def get_bind_game_info(cookie: str, use_for_public: bool = False) -> Optional[dict]: async def get_bind_game_info(cookie: str, mys_id: str):
""" """
通过cookie获取米游社绑定的原神游戏信息 通过cookie获取米游社绑定的原神游戏信息
:param cookie: cookie :param cookie: cookie
:param use_for_public: 是否用于公共cookie :param mys_id: 米游社id
:return: 原神信息 :return: 原神信息
""" """
if mys_id := re.search(r'(account_id|ltuid|stuid|login_uid)=(\d*)', cookie): with contextlib.suppress(Exception):
mys_id = mys_id[2] data = await aiorequests.get(url=GAME_RECORD_API,
data = (await aiorequests.get(url=GAME_RECORD_API,
headers=mihoyo_headers(cookie, f'uid={mys_id}'), headers=mihoyo_headers(cookie, f'uid={mys_id}'),
params={ params={
'uid': mys_id 'uid': mys_id
})).json() })
data = data.json()
nb_logger.debug(data)
if data['retcode'] == 0: if data['retcode'] == 0:
for game_data in data['data']['list']: return data['data']
if game_data['game_id'] == 2:
game_data['mys_id'] = mys_id
return game_data
if use_for_public:
return {'game_biz': 'hk4e_cn', 'mys_id': mys_id}
return None return None
@ -372,19 +372,43 @@ async def get_sign_reward_list() -> dict:
return data return data
async def get_stoken_by_cookie(cookie: str) -> Optional[str]: async def get_stoken_by_login_ticket(login_ticket: str, mys_id: str) -> Optional[str]:
try: with contextlib.suppress(Exception):
if login_ticket := re.search('login_ticket=([0-9a-zA-Z]+)', cookie): data = await aiorequests.get(STOKEN_API,
bbs_cookie_url = 'https://webapi.account.mihoyo.com/Api/cookie_accountinfo_by_loginticket?login_ticket={}' headers={
data = (await aiorequests.get(url=bbs_cookie_url.format(login_ticket[0].split('=')[1]))).json() 'x-rpc-app_version': '2.11.2',
'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 13_2_3 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) miHoYoBBS/2.11.1',
'x-rpc-client_type': '5',
'Referer': 'https://webstatic.mihoyo.com/',
'Origin': 'https://webstatic.mihoyo.com',
},
params={
'login_ticket': login_ticket,
'token_types': '3',
'uid': mys_id
})
data = data.json()
return data['data']['list'][0]['token']
return None
if '成功' in data['data']['msg']:
stuid = data['data']['cookie_info']['account_id'] async def get_cookie_token_by_stoken(stoken: str, mys_id: str) -> Optional[str]:
bbs_cookie_url2 = 'https://api-takumi.mihoyo.com/auth/api/getMultiTokenByLoginTicket?login_ticket={}&token_types=3&uid={}' with contextlib.suppress(Exception):
data2 = (await aiorequests.get(url=bbs_cookie_url2.format(login_ticket[0].split('=')[1], stuid))).json() data = await aiorequests.get(COOKIE_TOKEN_API,
return data2['data']['list'][0]['token'] headers={
except Exception: 'x-rpc-app_version': '2.11.2',
pass 'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 13_2_3 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) miHoYoBBS/2.11.1',
'x-rpc-client_type': '5',
'Referer': 'https://webstatic.mihoyo.com/',
'Origin': 'https://webstatic.mihoyo.com',
'Cookie': f'stuid={mys_id};stoken={stoken}'
},
params={
'uid': mys_id,
'stoken': stoken
})
data = data.json()
return data['data']['cookie_token']
return None return None

View File

@ -1,3 +1,4 @@
import re
import datetime import datetime
from typing import Optional from typing import Optional
@ -6,7 +7,7 @@ from fastapi.responses import JSONResponse
from pydantic import BaseModel from pydantic import BaseModel
from LittlePaimon.database import PublicCookie, PrivateCookie, LastQuery, CookieCache from LittlePaimon.database import PublicCookie, PrivateCookie, LastQuery, CookieCache
from LittlePaimon.utils.api import get_bind_game_info, get_stoken_by_cookie from LittlePaimon.utils.api import get_bind_game_info, get_stoken_by_login_ticket, get_cookie_token_by_stoken
from .utils import authentication from .utils import authentication
route = APIRouter() route = APIRouter()
@ -21,24 +22,52 @@ class BindCookie(BaseModel):
stoken: Optional[str] stoken: Optional[str]
@route.post('/bind_cookie', response_class=JSONResponse) @route.post('/bind_cookie', response_class=JSONResponse)
async def bind_cookie(data: BindCookie): async def bind_cookie(data: BindCookie):
if game_info := await get_bind_game_info(data.cookie): if mys_id := re.search(r'(?:(?:login_uid|account_mid|account_id|stmid|ltmid|stuid|ltuid)(?:_v2)?)=(\d+)',
game_uid = game_info['game_role_id'] data.cookie):
mys_id = game_info['mys_id'] mys_id = mys_id[1]
else:
return {'status': 200, 'msg': 'Cookie无效缺少account_id、login_uid或stuid字段请根据教程重新获取'}
cookie_token_match = re.search(r'(?:cookie_token|cookie_token_v2)=([0-9a-zA-Z]+)', data.cookie)
cookie_token = cookie_token_match[1] if cookie_token_match else None
login_ticket_match = re.search(r'(?:login_ticket|login_ticket_v2)=([0-9a-zA-Z]+)', data.cookie)
login_ticket = login_ticket_match[1] if login_ticket_match else None
stoken_match = re.search(r'(?:stoken|stoken_v2)=([0-9a-zA-Z]+)', data.cookie)
stoken = stoken_match[1] if stoken_match else None
if login_ticket and not stoken:
# 如果有login_ticket但没有stoken就通过login_ticket获取stoken
stoken = await get_stoken_by_login_ticket(login_ticket, mys_id)
if stoken and not cookie_token:
# 如果有stoken但没有cookie_token就通过stoken获取cookie_token
cookie_token = await get_cookie_token_by_stoken(stoken, mys_id)
if not cookie_token:
return {'status': 200, 'msg': 'Cookie无效缺少cookie_token或login_ticket字段请根据教程重新获取'}
if game_info := await get_bind_game_info(f'account_id={mys_id};cookie_token={cookie_token}', mys_id):
if not game_info['list']:
return {'status': 200, 'msg': '该账号尚未绑定任何游戏,请确认账号无误~'}
if not (
genshin_games := [{'uid': game['game_role_id'], 'nickname': game['nickname']} for game in
game_info['list'] if
game['game_id'] == 2]):
return {'status': 200, 'msg': '该账号尚未绑定原神,请确认账号无误~'}
await LastQuery.update_or_create(user_id=data.user_id, await LastQuery.update_or_create(user_id=data.user_id,
defaults={'uid': game_uid, 'last_time': datetime.datetime.now()}) defaults={'uid': genshin_games[0]['uid'],
if 'login_ticket' in data.cookie and (stoken := await get_stoken_by_cookie(data.cookie)): 'last_time': datetime.datetime.now()})
await PrivateCookie.update_or_create(user_id=data.user_id, uid=game_uid, mys_id=mys_id, send_msg = ''
defaults={'cookie': data.cookie, for info in genshin_games:
'stoken': f'stuid={mys_id};stoken={stoken};'}) send_msg += f'{info["nickname"]}({info["uid"]}) '
return {'status': 0, 'msg': f'QQ{data.user_id}的UID{game_uid}的Cookie以及Stoken绑定成功。'} await PrivateCookie.update_or_create(user_id=data.user_id, uid=info['uid'], mys_id=mys_id,
defaults={'cookie': f'account_id={mys_id};cookie_token={cookie_token}',
'stoken': f'stuid={mys_id};stoken={stoken};' if stoken else None})
return {'status': 0, 'msg':
f'QQ{data.user_id}绑定玩家{send_msg.strip()}的Cookie{"和Stoken" if stoken else ""}成功{"" if stoken else "当未能绑定Stoken"}'}
else: else:
await PrivateCookie.update_or_create(user_id=data.user_id, uid=game_uid, mys_id=mys_id, return {'status': 200, 'msg': 'Cookie无效请根据教程重新获取'}
defaults={'cookie': data.cookie})
return {'status': 0, 'msg': f'QQ{data.user_id}的UID{game_uid}绑定Cookie成功但未绑定stoken。'}
else:
return {'status': 200, 'msg': '该Cookie无效请根据教程重新获取'}
@route.get('/get_public_cookies', response_class=JSONResponse, dependencies=[authentication()]) @route.get('/get_public_cookies', response_class=JSONResponse, dependencies=[authentication()])
@ -115,32 +144,82 @@ async def delete_public_cookie(cookie_type: str, id: int):
@route.post('/add_cookie', response_class=JSONResponse, dependencies=[authentication()]) @route.post('/add_cookie', response_class=JSONResponse, dependencies=[authentication()])
async def add_public_cookie(cookie_type: str, force: bool, data: BindCookie): async def add_public_cookie(cookie_type: str, force: bool, data: BindCookie):
if cookie_type == 'public': if cookie_type == 'public':
if force or await get_bind_game_info(data.cookie, True): if not force:
if mys_id := re.search(r'(?:(?:login_uid|account_mid|account_id|stmid|ltmid|stuid|ltuid)(?:_v2)?)=(\d+)',
data.cookie):
mys_id = mys_id[1]
else:
return {'status': 200, 'msg': 'Cookie无效缺少account_id、login_uid或stuid字段'}
cookie_token_match = re.search(r'(?:cookie_token|cookie_token_v2)=([0-9a-zA-Z]+)', data.cookie)
cookie_token = cookie_token_match[1] if cookie_token_match else None
login_ticket_match = re.search(r'(?:login_ticket|login_ticket_v2)=([0-9a-zA-Z]+)', data.cookie)
login_ticket = login_ticket_match[1] if login_ticket_match else None
stoken_match = re.search(r'(?:stoken|stoken_v2)=([0-9a-zA-Z]+)', data.cookie)
stoken = stoken_match[1] if stoken_match else None
if login_ticket and not stoken:
# 如果有login_ticket但没有stoken就通过login_ticket获取stoken
stoken = await get_stoken_by_login_ticket(login_ticket, mys_id)
if stoken and not cookie_token:
# 如果有stoken但没有cookie_token就通过stoken获取cookie_token
cookie_token = await get_cookie_token_by_stoken(stoken, mys_id)
if not cookie_token:
return {'status': 200, 'msg': 'Cookie无效缺少cookie_token或login_ticket字段'}
if await get_bind_game_info(f'account_id={mys_id};cookie_token={cookie_token}', mys_id):
new_cookie = await PublicCookie.create(cookie=f'account_id={mys_id};cookie_token={cookie_token}')
return {'status': 0, 'msg': f'{new_cookie.id}号公共Cookie添加成功'}
else:
return {'status': 200, 'msg': '该Cookie无效请根据教程重新获取'}
else:
new_cookie = await PublicCookie.create(cookie=data.cookie) new_cookie = await PublicCookie.create(cookie=data.cookie)
return {'status': 0, 'msg': f'{new_cookie.id}号公共Cookie添加成功'} return {'status': 0, 'msg': f'{new_cookie.id}号公共Cookie添加成功'}
elif not force:
return {'status': 200, 'msg': '该Cookie无效请根据教程重新获取'}
else: else:
if force: if force:
await PrivateCookie.update_or_create(user_id=data.user_id, uid=data.uid, mys_id=data.mys_id, await PrivateCookie.update_or_create(user_id=data.user_id, uid=data.uid, mys_id=data.mys_id,
defaults={'cookie': data.cookie, 'stoken': data.stoken}) defaults={'cookie': data.cookie, 'stoken': data.stoken})
return {'status': 0, 'msg': f'QQ{data.user_id}的UID{data.uid}的Cookie强制修改成功。'} return {'status': 0, 'msg': f'QQ{data.user_id}的UID{data.uid}的Cookie强制修改成功。'}
elif game_info := await get_bind_game_info(data.cookie): else:
game_uid = game_info['game_role_id'] if mys_id := re.search(r'(?:(?:login_uid|account_mid|account_id|stmid|ltmid|stuid|ltuid)(?:_v2)?)=(\d+)',
mys_id = game_info['mys_id'] data.cookie):
mys_id = mys_id[1]
else:
return {'status': 200, 'msg': 'Cookie无效缺少account_id、login_uid或stuid字段请根据教程重新获取'}
cookie_token_match = re.search(r'(?:cookie_token|cookie_token_v2)=([0-9a-zA-Z]+)', data.cookie)
cookie_token = cookie_token_match[1] if cookie_token_match else None
login_ticket_match = re.search(r'(?:login_ticket|login_ticket_v2)=([0-9a-zA-Z]+)', data.cookie)
login_ticket = login_ticket_match[1] if login_ticket_match else None
stoken_match = re.search(r'(?:stoken|stoken_v2)=([0-9a-zA-Z]+)', data.cookie)
stoken = stoken_match[1] if stoken_match else None
if login_ticket and not stoken:
# 如果有login_ticket但没有stoken就通过login_ticket获取stoken
stoken = await get_stoken_by_login_ticket(login_ticket, mys_id)
if stoken and not cookie_token:
# 如果有stoken但没有cookie_token就通过stoken获取cookie_token
cookie_token = await get_cookie_token_by_stoken(stoken, mys_id)
if not cookie_token:
return {'status': 200, 'msg': 'Cookie无效缺少cookie_token或login_ticket字段请根据教程重新获取'}
if game_info := await get_bind_game_info(f'account_id={mys_id};cookie_token={cookie_token}', mys_id):
if not game_info['list']:
return {'status': 200, 'msg': '该账号尚未绑定任何游戏,请确认账号无误~'}
if not (
genshin_games := [{'uid': game['game_role_id'], 'nickname': game['nickname']} for game in
game_info['list'] if
game['game_id'] == 2]):
return {'status': 200, 'msg': '该账号尚未绑定原神,请确认账号无误~'}
await LastQuery.update_or_create(user_id=data.user_id, await LastQuery.update_or_create(user_id=data.user_id,
defaults={'uid': game_uid, 'last_time': datetime.datetime.now()}) defaults={'uid': genshin_games[0]['uid'],
if 'login_ticket' in data.cookie and (stoken := await get_stoken_by_cookie(data.cookie)): 'last_time': datetime.datetime.now()})
await PrivateCookie.update_or_create(user_id=data.user_id, uid=game_uid, mys_id=mys_id, send_msg = ''
defaults={'cookie': data.cookie, for info in genshin_games:
'stoken': f'stuid={mys_id};stoken={stoken};'}) send_msg += f'{info["nickname"]}({info["uid"]}) '
return {'status': 0, 'msg': f'QQ{data.user_id}的UID{game_uid}的Cookie以及Stoken添加/保存成功。'} await PrivateCookie.update_or_create(user_id=data.user_id, uid=info['uid'], mys_id=mys_id,
defaults={
'cookie': f'account_id={mys_id};cookie_token={cookie_token}',
'stoken': f'stuid={mys_id};stoken={stoken};' if stoken else None})
return {'status': 0, 'msg':
f'QQ{data.user_id}绑定玩家{send_msg.strip()}的Cookie{"和Stoken" if stoken else ""}成功{"" if stoken else "当未能绑定Stoken"}'}
else: else:
await PrivateCookie.update_or_create(user_id=data.user_id, uid=game_uid, mys_id=mys_id, return {'status': 200, 'msg': 'Cookie无效请根据教程重新获取'}
defaults={'cookie': data.cookie})
return {'status': 0, 'msg': f'QQ{data.user_id}的UID{game_uid}的Cookie添加/保存成功但未绑定stoken。'}
else:
return {'status': 200, 'msg': '该Cookie无效请根据教程重新获取'}
@route.post('/update_private_cookie', response_class=JSONResponse, dependencies=[authentication()]) @route.post('/update_private_cookie', response_class=JSONResponse, dependencies=[authentication()])
@ -148,18 +227,46 @@ async def update_cookie(force: bool, data: BindCookie):
if force: if force:
await PrivateCookie.filter(id=data.id).update(**data.dict(exclude={'id'})) await PrivateCookie.filter(id=data.id).update(**data.dict(exclude={'id'}))
return {'status': 0, 'msg': f'QQ{data.user_id}的UID{data.uid}的Cookie强制修改成功。'} return {'status': 0, 'msg': f'QQ{data.user_id}的UID{data.uid}的Cookie强制修改成功。'}
elif game_info := await get_bind_game_info(data.cookie): else:
game_uid = game_info['game_role_id'] if mys_id := re.search(r'(?:(?:login_uid|account_mid|account_id|stmid|ltmid|stuid|ltuid)(?:_v2)?)=(\d+)',
mys_id = game_info['mys_id'] data.cookie):
mys_id = mys_id[1]
else:
return {'status': 200, 'msg': 'Cookie无效缺少account_id、login_uid或stuid字段请根据教程重新获取'}
cookie_token_match = re.search(r'(?:cookie_token|cookie_token_v2)=([0-9a-zA-Z]+)', data.cookie)
cookie_token = cookie_token_match[1] if cookie_token_match else None
login_ticket_match = re.search(r'(?:login_ticket|login_ticket_v2)=([0-9a-zA-Z]+)', data.cookie)
login_ticket = login_ticket_match[1] if login_ticket_match else None
stoken_match = re.search(r'(?:stoken|stoken_v2)=([0-9a-zA-Z]+)', data.cookie)
stoken = stoken_match[1] if stoken_match else None
if login_ticket and not stoken:
# 如果有login_ticket但没有stoken就通过login_ticket获取stoken
stoken = await get_stoken_by_login_ticket(login_ticket, mys_id)
if stoken and not cookie_token:
# 如果有stoken但没有cookie_token就通过stoken获取cookie_token
cookie_token = await get_cookie_token_by_stoken(stoken, mys_id)
if not cookie_token:
return {'status': 200, 'msg': 'Cookie无效缺少cookie_token或login_ticket字段请根据教程重新获取'}
if game_info := await get_bind_game_info(f'account_id={mys_id};cookie_token={cookie_token}', mys_id):
if not game_info['list']:
return {'status': 200, 'msg': '该账号尚未绑定任何游戏,请确认账号无误~'}
if not (
genshin_games := [{'uid': game['game_role_id'], 'nickname': game['nickname']} for game in
game_info['list'] if
game['game_id'] == 2]):
return {'status': 200, 'msg': '该账号尚未绑定原神,请确认账号无误~'}
await LastQuery.update_or_create(user_id=data.user_id, await LastQuery.update_or_create(user_id=data.user_id,
defaults={'uid': game_uid, 'last_time': datetime.datetime.now()}) defaults={'uid': genshin_games[0]['uid'],
if 'login_ticket' in data.cookie and (stoken := await get_stoken_by_cookie(data.cookie)): 'last_time': datetime.datetime.now()})
await PrivateCookie.filter(id=data.id).update(user_id=data.user_id, uid=game_uid, mys_id=mys_id, send_msg = ''
cookie=data.cookie, stoken=f'stuid={mys_id};stoken={stoken};') for info in genshin_games:
return {'status': 0, 'msg': f'QQ{data.user_id}的UID{game_uid}的Cookie以及Stoken添加/保存成功。'} send_msg += f'{info["nickname"]}({info["uid"]}) '
await PrivateCookie.update_or_create(user_id=data.user_id, uid=info['uid'], mys_id=mys_id,
defaults={
'cookie': f'account_id={mys_id};cookie_token={cookie_token}',
'stoken': f'stuid={mys_id};stoken={stoken};' if stoken else None})
return {'status': 0, 'msg':
f'QQ{data.user_id}绑定玩家{send_msg.strip()}的Cookie{"和Stoken" if stoken else ""}成功{"" if stoken else "当未能绑定Stoken"}'}
else: else:
await PrivateCookie.filter(id=data.id).update(user_id=data.user_id, uid=game_uid, mys_id=mys_id, return {'status': 200, 'msg': 'Cookie无效请根据教程重新获取'}
cookie=data.cookie, stoken=None)
return {'status': 0, 'msg': f'QQ{data.user_id}的UID{game_uid}的Cookie添加/保存成功但未获取到stoken。'}
else:
return {'status': 200, 'msg': '该Cookie无效请根据教程重新获取'}