Genshin-bots/StarRailUID
1
0
Fork 0
mirror of https://github.com/baiqwerdvd/StarRailUID.git synced 2025-05-05 03:03:45 +08:00
Code Issues Packages Projects Releases Wiki Activity

✨支持铁道签到

Browse Source
This commit is contained in:
qwerdvd 2023-04-27 22:09:07 +08:00
parent cbd6937cfc
commit a1a7f9c8b0
14 changed files with 874 additions and 85 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
StarRailUID/sruid_utils/api/mys/api.py
View File

@ -2,7 +2,11 @@
OLD_URL = "https://api-takumi.mihoyo.com"
STAR_RAIL_SIGN_INFO_URL = f'{OLD_URL}/event/luna/extra_info'
STAR_RAIL_SIGN_URL = f'{OLD_URL}/event/luna/extra_award'
STAR_RAIL_SIGN_INFO_URL = f'{OLD_URL}/event/luna/info'
STAR_RAIL_SIGN_LIST_URL = f'{OLD_URL}/event/luna/home'
STAR_RAIL_SIGN_EXTRA_INFO_URL = f'{OLD_URL}/event/luna/extra_info'
STAR_RAIL_SIGN_URL = f'{OLD_URL}/event/luna/sign'
# CREATE_QRCODE = f'{OLD_URL}/event/bbs_sign_reward/gen_auth_code'
_API = locals()

46
StarRailUID/sruid_utils/api/mys/models.py Normal file
View File

@ -0,0 +1,46 @@
################
# 签到相关 #
################
from typing import Any, List, TypedDict
class MysSign(TypedDict):
code: str
risk_code: int
gt: str
challenge: str
success: int
is_risk: bool
class SignInfo(TypedDict):
total_sign_day: int
today: str
is_sign: bool
is_sub: bool
region: str
sign_cnt_missed: int
short_sign_day: int
class SignAward(TypedDict):
icon: str
name: str
cnt: int
class SignExtraAward(TypedDict):
has_extra_award: bool
start_time: str
end_time: str
list: List[Any] # TODO
start_timestamp: str
end_timestamp: str
class SignList(TypedDict):
month: int
awards: List[SignAward]
biz: str
resign: bool
short_extra_award: SignExtraAward

10
StarRailUID/starrailuid_signin/__init__.py
View File

@ -8,7 +8,7 @@ from gsuid_core.models import Event
from gsuid_core.aps import scheduler
from gsuid_core.logger import logger
from ..utils.database import get_sqla
from ..utils.api import get_sqla
from .sign import sign_in, daily_sign
from ..utils.error_reply import UID_HINT
from ....GenshinUID.GenshinUID.genshinuid_config.gs_config import gsconfig
@ -31,11 +31,11 @@ async def sign_at_night():
async def get_sign_func(bot: Bot, ev: Event):
await bot.logger.info('[SR签到]QQ号: {}'.format(ev.user_id))
sqla = get_sqla(ev.bot_id)
uid = await sqla.get_bind_uid(ev.user_id)
if uid is None:
sr_uid = await sqla.get_bind_sruid(ev.user_id)
if sr_uid is None:
return await bot.send(UID_HINT)
await bot.logger.info('[SR签到]UID: {}'.format(uid))
await bot.send(await sign_in(uid))
await bot.logger.info('[SR签到]UID: {}'.format(sr_uid))
await bot.send(await sign_in(sr_uid))
@sv_sign_config.on_fullmatch('sr全部重签')

51
StarRailUID/starrailuid_signin/sign.py
View File

@ -5,8 +5,8 @@ from copy import deepcopy
from gsuid_core.gss import gss
from gsuid_core.logger import logger
from ..utils.api import get_sqla
from ..utils.mys_api import mys_api
from ..utils.database import get_sqla
from ....GenshinUID.GenshinUID.genshinuid_config.gs_config import gsconfig
private_msg_list = {}
@ -15,17 +15,17 @@ already = 0
# 签到函数
async def sign_in(uid: str) -> str:
logger.info(f'[SR签到] {uid} 开始执行签到')
async def sign_in(sr_uid: str) -> str:
logger.info(f'[SR签到] {sr_uid} 开始执行签到')
# 获得签到信息
sign_info = await mys_api.get_sign_info(uid)
sign_info = await mys_api.get_sign_info(sr_uid)
# 初步校验数据
if isinstance(sign_info, int):
logger.warning(f'[SR签到] {uid} 出错, 请检查Cookies是否过期！')
logger.warning(f'[SR签到] {sr_uid} 出错, 请检查Cookies是否过期！')
return '签到失败...请检查Cookies是否过期！'
# 检测是否已签到
if sign_info['is_sign']:
logger.info(f'[SR签到] {uid} 该用户今日已签到,跳过...')
logger.info(f'[SR签到] {sr_uid} 该用户今日已签到,跳过...')
global already
already += 1
day_of_month = int(sign_info['today'].split('-')[-1])
@ -37,10 +37,10 @@ async def sign_in(uid: str) -> str:
Header = {}
for index in range(4):
# 进行一次签到
sign_data = await mys_api.mys_sign(uid=uid, header=Header)
sign_data = await mys_api.mys_sign(uid=sr_uid, header=Header)
# 检测数据
if isinstance(sign_data, int):
logger.warning(f'[SR签到] {uid} 出错, 请检查Cookies是否过期！')
logger.warning(f'[SR签到] {sr_uid} 出错, 请检查Cookies是否过期！')
return 'sr签到失败...请检查Cookies是否过期！'
if 'risk_code' in sign_data:
# 出现校验码
@ -54,11 +54,11 @@ async def sign_in(uid: str) -> str:
Header['x-rpc-challenge'] = ch
Header['x-rpc-validate'] = vl
Header['x-rpc-seccode'] = f'{vl}|jordan'
logger.info(f'[SR签到] {uid} 已获取验证码, 等待时间{delay}秒')
logger.info(f'[SR签到] {sr_uid} 已获取验证码, 等待时间{delay}秒')
await asyncio.sleep(delay)
else:
delay = 605 + random.randint(1, 120)
logger.info(f'[SR签到] {uid} 未获取验证码,等待{delay}秒后重试...')
logger.info(f'[SR签到] {sr_uid} 未获取验证码,等待{delay}秒后重试...')
await asyncio.sleep(delay)
continue
else:
@ -67,13 +67,13 @@ async def sign_in(uid: str) -> str:
# 成功签到!
else:
if index == 0:
logger.info(f'[SR签到] {uid} 该用户无校验码!')
logger.info(f'[SR签到] {sr_uid} 该用户无校验码!')
else:
logger.info(f'[SR签到] [无感验证] {uid} 该用户重试 {index} 次验证成功!')
logger.info(f'[SR签到] [无感验证] {sr_uid} 该用户重试 {index} 次验证成功!')
break
elif (int(str(uid)[0]) > 5) and (sign_data['data']['code'] == 'ok'):
elif (int(str(sr_uid)[0]) > 5) and (sign_data['data']['code'] == 'ok'):
# 国际服签到无risk_code字段
logger.info(f'[SR国际服签到] {uid} 签到成功!')
logger.info(f'[SR国际服签到] {sr_uid} 签到成功!')
break
else:
# 重试超过阈值
@ -82,13 +82,13 @@ async def sign_in(uid: str) -> str:
# 签到失败
else:
im = 'sr签到失败!'
logger.warning(f'[SR签到] {uid} 签到失败, 结果: {im}')
logger.warning(f'[SR签到] {sr_uid} 签到失败, 结果: {im}')
return im
# 获取签到列表
sign_list = await mys_api.get_sign_list(uid)
new_sign_info = await mys_api.get_sign_info(uid)
sign_list = await mys_api.get_sign_list(sr_uid)
new_sign_info = await mys_api.get_sign_info(sr_uid)
if isinstance(sign_list, int) or isinstance(new_sign_info, int):
logger.warning(f'[SR签到] {uid} 出错, 请检查Cookies是否过期！')
logger.warning(f'[SR签到] {sr_uid} 出错, 请检查Cookies是否过期！')
return 'sr签到失败...请检查Cookies是否过期！'
# 获取签到奖励物品，拿旧的总签到天数 + 1 为新的签到天数，再 -1 即为今日奖励物品的下标
getitem = sign_list['awards'][int(sign_info['total_sign_day']) + 1 - 1]
@ -103,16 +103,18 @@ async def sign_in(uid: str) -> str:
sign_missed -= 1
sign_missed = sign_info.get('sign_cnt_missed') or sign_missed
im = f'{mes_im}!\n{get_im}\n本月漏签次数：{sign_missed}'
logger.info(f'[SR签到] {uid} 签到完成, 结果: {mes_im}, 漏签次数: {sign_missed}')
logger.info(f'[SR签到] {sr_uid} 签到完成, 结果: {mes_im}, 漏签次数: {sign_missed}')
return im
async def single_daily_sign(bot_id: str, uid: str, gid: str, qid: str):
im = await sign_in(uid)
async def single_daily_sign(bot_id: str, sr_uid: str, gid: str, qid: str):
im = await sign_in(sr_uid)
if gid == 'on':
if qid not in private_msg_list:
private_msg_list[qid] = []
private_msg_list[qid].append({'bot_id': bot_id, 'uid': uid, 'msg': im})
private_msg_list[qid].append(
{'bot_id': bot_id, 'uid': sr_uid, 'msg': im}
)
else:
# 向群消息推送列表添加这个群
if gid not in group_msg_list:
@ -149,7 +151,10 @@ async def daily_sign():
if user.sign_switch != 'off':
tasks.append(
single_daily_sign(
user.bot_id, user.uid, user.sign_switch, user.user_id
user.bot_id,
user.sr_uid,
user.sign_switch,
user.user_id,
)
)
if len(tasks) >= 1:

110
StarRailUID/starrailuid_user/__init__.py Normal file
View File

@ -0,0 +1,110 @@
from typing import List
from gsuid_core.sv import SV
from gsuid_core.bot import Bot
from gsuid_core.models import Event
from ..utils.api import get_sqla
# from .qrlogin import qrcode_login
# from .get_ck_help_msg import get_ck_help
from ..utils.message import send_diff_msg
from .draw_user_card import get_user_card
# from gsuid_core.segment import MessageSegment
# from .add_ck import deal_ck, get_ck_by_stoken, get_ck_by_all_stoken
sv_user_config = SV('sr用户管理', pm=2)
sv_user_add = SV('sr用户添加')
# sv_user_qrcode_login = SV('sr扫码登陆')
# sv_user_addck = SV('sr添加CK', area='DIRECT')
sv_user_info = SV('sr用户信息')
# sv_user_help = SV('sr绑定帮助')
# @sv_user_config.on_fullmatch(('sr刷新全部CK', 'sr刷新全部ck'))
# async def send_refresh_all_ck_msg(bot: Bot, ev: Event):
# await bot.logger.info('sr开始执行[刷新全部CK]')
# im = await get_ck_by_all_stoken(ev.bot_id)
# await bot.send(im)
#
#
# @sv_user_add.on_fullmatch(('sr刷新CK', 'sr刷新ck'))
# async def send_refresh_ck_msg(bot: Bot, ev: Event):
# await bot.logger.info('sr开始执行[刷新CK]')
# im = await get_ck_by_stoken(ev.bot_id, ev.user_id)
# await bot.send(im)
# @sv_user_qrcode_login.on_fullmatch(('sr扫码登陆', 'sr扫码登录'))
# async def send_qrcode_login(bot: Bot, ev: Event):
# await bot.logger.info('sr开始执行[扫码登陆]')
# im = await qrcode_login(bot, ev, ev.user_id)
# if not im:
# return
# im = await deal_ck(ev.bot_id, im, ev.user_id)
# await bot.send(im)
@sv_user_info.on_fullmatch(('sr绑定信息'))
async def send_bind_card(bot: Bot, ev: Event):
await bot.logger.info('sr开始执行[查询用户绑定状态]')
# im = await get_user_card(ev.bot_id, ev.user_id)
uid_list = await get_user_card(ev.bot_id, ev.user_id)
await bot.logger.info('sr[查询用户绑定状态]完成!等待图片发送中...')
await bot.send(uid_list)
# @sv_user_addck.on_prefix(('sr添加'))
# async def send_add_ck_msg(bot: Bot, ev: Event):
# im = await deal_ck(ev.bot_id, ev.text, ev.user_id)
# await bot.send(im)
@sv_user_info.on_command(('sr绑定uid', 'sr切换uid', 'sr删除uid', 'sr解绑uid'))
async def send_link_uid_msg(bot: Bot, ev: Event):
await bot.logger.info('sr开始执行[绑定/解绑用户信息]')
qid = ev.user_id
await bot.logger.info('sr[绑定/解绑]UserID: {}'.format(qid))
sqla = get_sqla(ev.bot_id)
sr_uid = ev.text.strip()
if sr_uid and not sr_uid.isdigit():
return await bot.send('你输入了错误的格式!')
if ev.command.startswith('sr绑定'):
data = await sqla.insert_bind_data(qid, sr_uid=sr_uid)
print(data)
return await send_diff_msg(
bot,
data,
{
0: f'绑定SR_UID{sr_uid}成功！',
-1: f'SR_UID{sr_uid}的位数不正确！',
-2: f'SR_UID{sr_uid}已经绑定过了！',
-3: '你输入了错误的格式!',
},
)
elif ev.command.startswith('sr切换'):
data = await sqla.switch_uid(qid, uid=sr_uid)
if isinstance(data, List):
return await bot.send(f'切换SR_UID{sr_uid}成功！')
else:
return await bot.send(f'尚未绑定该SR_UID{sr_uid}')
else:
data = await sqla.delete_bind_data(qid, sr_uid=sr_uid)
return await send_diff_msg(
bot,
data,
{
0: f'删除SR_UID{sr_uid}成功！',
-1: f'该SR_UID{sr_uid}不在已绑定列表中！',
},
)
# @sv_user_help.on_fullmatch(('ck帮助', '绑定帮助'))
# async def send_ck_help(bot: Bot, ev: Event):
# msg_list = await get_ck_help()
# await bot.send(MessageSegment.node(msg_list))

249
StarRailUID/starrailuid_user/add_ck.py Normal file
View File

@ -0,0 +1,249 @@
from pathlib import Path
from typing import Dict, List
from http.cookies import SimpleCookie
from ..utils.api import get_sqla
from ..utils.mys_api import mys_api
from ..utils.error_reply import UID_HINT
pic_path = Path(__file__).parent / 'pic'
id_list = [
'login_uid',
'login_uid_v2',
'account_mid_v2',
'account_mid',
'account_id',
'stuid',
'ltuid',
'ltmid',
'stmid',
'stmid_v2',
'ltmid_v2',
'stuid_v2',
'ltuid_v2',
]
sk_list = ['stoken', 'stoken_v2']
ck_list = ['cookie_token', 'cookie_token_v2']
lt_list = ['login_ticket', 'login_ticket_v2']
async def get_ck_by_all_stoken(bot_id: str):
sqla = get_sqla(bot_id)
uid_list: List = await sqla.get_all_uid_list()
uid_dict = {}
for uid in uid_list:
user_data = await sqla.select_user_data(uid)
if user_data:
uid_dict[uid] = user_data.user_id
im = await refresh_ck_by_uid_list(bot_id, uid_dict)
return im
async def get_ck_by_stoken(bot_id: str, user_id: str):
sqla = get_sqla(bot_id)
uid_list: List = await sqla.get_bind_uid_list(user_id)
uid_dict = {uid: user_id for uid in uid_list}
im = await refresh_ck_by_uid_list(bot_id, uid_dict)
return im
async def refresh_ck_by_uid_list(bot_id: str, uid_dict: Dict):
sqla = get_sqla(bot_id)
uid_num = len(uid_dict)
if uid_num == 0:
return '请先绑定一个UID噢~'
error_list = {}
skip_num = 0
error_num = 0
for uid in uid_dict:
stoken = await sqla.get_user_stoken(uid)
if stoken is None:
skip_num += 1
error_num += 1
continue
else:
qid = uid_dict[uid]
try:
mes = await _deal_ck(bot_id, stoken, qid)
except TypeError:
error_list[uid] = 'SK或CK已过期！'
error_num += 1
continue
ok_num = mes.count('成功')
if ok_num < 2:
error_list[uid] = '可能是SK已过期~'
error_num += 1
continue
s_im = f'执行完成~成功刷新CK{uid_num - error_num}个！跳过{skip_num}个!'
f_im = '\n'.join([f'UID{u}:{error_list[u]}' for u in error_list])
im = f'{s_im}\n{f_im}' if f_im else s_im
return im
async def deal_ck(bot_id: str, mes: str, user_id: str, mode: str = 'PIC'):
im = await _deal_ck(bot_id, mes, user_id)
if mode == 'PIC':
im = await _deal_ck_to_pic(im)
return im
async def _deal_ck_to_pic(im: str) -> bytes:
ok_num = im.count('成功')
if ok_num < 1:
status_pic = pic_path / 'ck_no.png'
elif ok_num < 2:
status_pic = pic_path / 'ck_ok.png'
else:
status_pic = pic_path / 'all_ok.png'
with open(status_pic, 'rb') as f:
img = f.read()
return img
async def get_account_id(simp_dict: SimpleCookie) -> str:
for _id in id_list:
if _id in simp_dict:
account_id = simp_dict[_id].value
break
else:
account_id = ''
return account_id
async def _deal_ck(bot_id: str, mes: str, user_id: str) -> str:
sqla = get_sqla(bot_id)
simp_dict = SimpleCookie(mes)
uid = await sqla.get_bind_uid(user_id)
sr_uid = await sqla.get_bind_sruid(user_id)
if uid is None and sr_uid is None:
if uid is None:
return UID_HINT
elif sr_uid is None:
return '请绑定星穹铁道UID...'
im_list = []
is_add_stoken = False
status = True
app_cookie, stoken = '', ''
account_id, cookie_token = '', ''
if status:
for sk in sk_list:
if sk in simp_dict:
account_id = await get_account_id(simp_dict)
if not account_id:
return '该CK字段出错, 缺少login_uid或stuid或ltuid字段!'
stoken = simp_dict[sk].value
if stoken.startswith('v2_'):
if 'mid' in simp_dict:
mid = simp_dict['mid'].value
app_cookie = (
f'stuid={account_id};stoken={stoken};mid={mid}'
)
else:
return 'v2类型SK必须携带mid...'
else:
app_cookie = f'stuid={account_id};stoken={stoken}'
cookie_token_data = await mys_api.get_cookie_token_by_stoken(
stoken, account_id, app_cookie
)
if isinstance(cookie_token_data, Dict):
cookie_token = cookie_token_data['cookie_token']
is_add_stoken = True
status = False
break
else:
return '返回值错误...'
if status:
for lt in lt_list:
if lt in simp_dict:
# 寻找stoken
login_ticket = simp_dict[lt].value
account_id = await get_account_id(simp_dict)
if not account_id:
return '该CK字段出错, 缺少login_uid或stuid或ltuid字段!'
stoken_data = await mys_api.get_stoken_by_login_ticket(
login_ticket, account_id
)
if isinstance(stoken_data, Dict):
stoken = stoken_data['list'][0]['token']
app_cookie = f'stuid={account_id};stoken={stoken}'
cookie_token_data = (
await mys_api.get_cookie_token_by_stoken(
stoken, account_id
)
)
if isinstance(cookie_token_data, Dict):
cookie_token = cookie_token_data['cookie_token']
is_add_stoken = True
status = False
break
if status:
for ck in ck_list:
if ck in simp_dict:
# 寻找uid
account_id = await get_account_id(simp_dict)
if not account_id:
return '该CK字段出错, 缺少login_uid或stuid或ltuid字段!'
cookie_token = simp_dict[ck].value
status = False
break
if status:
return (
'添加Cookies失败!Cookies中应该包含cookie_token或者login_ticket相关信息!'
'\n可以尝试退出米游社登陆重新登陆获取!'
)
account_cookie = f'account_id={account_id};cookie_token={cookie_token}'
try:
if sr_uid or (uid and int(uid[0]) < 6):
mys_data = await mys_api.get_mihoyo_bbs_info(
account_id, account_cookie
)
else:
mys_data = await mys_api.get_mihoyo_bbs_info(
account_id, account_cookie, True
)
# 剔除除了原神之外的其他游戏
if isinstance(mys_data, List):
for i in mys_data:
if i['game_id'] == 2:
uid = i['game_role_id']
elif i['game_id'] == 6:
sr_uid = i['game_role_id']
if uid and sr_uid:
break
else:
if not (uid or sr_uid):
return f'你的米游社账号{account_id}尚未绑定原神/星铁账号,请前往米游社操作！'
except Exception:
pass
if not uid:
return f'你的米游社账号{account_id}尚未绑定原神/星铁账号,请前往米游社操作！'
await sqla.refresh_cache(uid)
if is_add_stoken:
im_list.append(f'添加Stoken成功,stuid={account_id},stoken={stoken}')
await sqla.insert_user_data(
user_id, uid, sr_uid, account_cookie, app_cookie
)
im_list.append(
f'添加Cookies成功,account_id={account_id},cookie_token={cookie_token}'
)
im_list.append(
'Cookies和Stoken属于个人重要信息,如果你是在不知情的情况下添加,请马上修改米游社账户密码,保护个人隐私！'
)
im_list.append(
(
'如果需要【sr开启自动签到】和【sr开启推送】还需要在【群聊中】使用命令“绑定uid”绑定你的uid。'
'\n例如：绑定uid123456789。'
)
)
im_list.append('你可以使用命令【sr绑定信息】检查你的账号绑定情况！')
im = '\n'.join(im_list)
return im

114
StarRailUID/starrailuid_user/draw_user_card.py Normal file
View File

@ -0,0 +1,114 @@
# from pathlib import Path
from typing import List, Tuple, Optional
from PIL import Image
from ..utils.api import get_sqla
# from ..utils.image.convert import convert_img
# from ..utils.colors import sec_color, first_color
# from ..utils.fonts.genshin_fonts import gs_font_15, gs_font_30, gs_font_36
# from ..utils.image.image_tools import (
# get_color_bg,
# get_qq_avatar,
# draw_pic_with_ring,
# )
# TEXT_PATH = Path(__file__).parent / 'texture2d'
#
# status_off = Image.open(TEXT_PATH / 'status_off.png')
# status_on = Image.open(TEXT_PATH / 'status_on.png')
#
# EN_MAP = {'coin': '宝钱', 'resin': '体力', 'go': '派遣', 'transform': '质变仪'}
async def get_user_card(bot_id: str, user_id: str):
pass
sqla = get_sqla(bot_id)
uid_list: List = await sqla.get_bind_uid_list(user_id)
# w, h = 750, len(uid_list) * 750 + 470
#
# # 获取背景图片各项参数
# _id = str(user_id)
# if _id.startswith('http'):
# char_pic = await get_qq_avatar(avatar_url=_id)
# else:
# char_pic = await get_qq_avatar(qid=_id)
# char_pic = await draw_pic_with_ring(char_pic, 290)
#
# img = await get_color_bg(w, h)
# title = Image.open(TEXT_PATH / 'user_title.png')
# title.paste(char_pic, (241, 40), char_pic)
#
# title_draw = ImageDraw.Draw(title)
# title_draw.text(
# (375, 444), f'{bot_id} - {user_id}', first_color, gs_font_30, 'mm'
# )
# img.paste(title, (0, 0), title)
#
# for index, uid in enumerate(uid_list):
# user_card = Image.open(TEXT_PATH / 'user_bg.png')
# user_draw = ImageDraw.Draw(user_card)
# user_push_data = await sqla.select_push_data(uid)
# user_data = await sqla.select_user_data(uid)
# if user_data is None:
# user_data = GsUser(
# bot_id=bot_id,
# user_id=user_id,
# uid=uid,
# stoken=None,
# cookie=None,
# sign_switch='off',
# bbs_switch='off',
# push_switch='off',
# )
#
# user_draw.text(
# (375, 62),
# f'UID {uid}',
# first_color,
# font=gs_font_36,
# anchor='mm',
# )
#
# x, y = 331, 112
# paste_switch(user_card, user_data.cookie, (241, 128))
# paste_switch(user_card, user_data.stoken, (241 + x, 128))
# paste_switch(user_card, user_data.sign_switch, (241, 128 + y))
# paste_switch(user_card, user_data.bbs_switch, (241 + x, 128 + y))
# paste_switch(user_card, user_data.push_switch, (241, 128 + 2 * y))
# paste_switch(user_card, user_data.status, (241 + x,
# 128 + 2 * y), True)
#
# for _index, mode in enumerate(['coin', 'resin', 'go', 'transform']):
# paste_switch(
# user_card,
# getattr(user_push_data, f'{mode}_push'),
# (241 + _index % 2 * x, 128 + (_index // 2 + 3) * y),
# )
# if getattr(user_push_data, f'{mode}_push') != 'off':
# user_draw.text(
# (268 + _index % 2 * x, 168 + (_index // 2 + 3) * y),
# f'{getattr(user_push_data, f"{mode}_value")}',
# sec_color,
# font=gs_font_15,
# anchor='lm',
# )
# img.paste(user_card, (0, 500 + index * 690), user_card)
# return await convert_img(img)
return uid_list
def paste_switch(
card: Image.Image,
status: Optional[str],
pos: Tuple[int, int],
is_status: bool = False,
):
pass
# if is_status:
# pic = status_off if status else status_on
# else:
# pic = status_on if status != 'off' and status else status_off
# card.paste(pic, pos, pic)

43
StarRailUID/starrailuid_user/get_ck_help_msg.py Normal file
View File

@ -0,0 +1,43 @@
from typing import List
from gsuid_core.models import Message
CK_QRCODE_LOGIN = '''先发送【绑定uidxxx】绑定UID,
然后发送【扫码登陆】, 使用米游社APP扫码完成绑定, [或者]选择以下方法
'''
CK_CONSOLE = '''var cookie = document.cookie;
var Str_Num = cookie.indexOf('_MHYUUID=');
cookie = '添加 ' + cookie.substring(Str_Num);
var ask = confirm('Cookie:' + cookie + '\\n\\n按确认，然后粘贴发送给机器人');
if (ask == true) {
copy(cookie);
msg = cookie
} else {
msg = 'Cancel'
}
'''
CK_URL = '''1.复制上面全部代码，然后打开下面的网站
https://bbs.mihoyo.com/ys/obc/?bbs_presentation_style=no_header(国服)
https://www.hoyolab.com/home(国际服)
2.在页面上右键检查或者Ctrl+Shift+i
3.选择控制台(Console)，粘贴，回车，在弹出的窗口点确认(点完自动复制)
4.然后在和机器人的私聊窗口，粘贴发送即可
'''
SK_URL = '''如果想获取SK,操作方法和上面一致,网址更换为
http://user.mihoyo.com/(国服)
登陆后,进入控制台粘贴代码
然后在和机器人的私聊窗口，粘贴发送即可
'''
async def get_ck_help() -> List[Message]:
msg_list = []
msg_list.append(Message('text', '请先添加bot为好友'))
msg_list.append(Message('text', CK_QRCODE_LOGIN))
msg_list.append(Message('text', CK_CONSOLE))
msg_list.append(Message('text', CK_URL))
msg_list.append(Message('text', SK_URL))
return msg_list

160
StarRailUID/starrailuid_user/qrlogin.py Normal file
View File

@ -0,0 +1,160 @@
import io
import json
import base64
import asyncio
from http.cookies import SimpleCookie
from typing import Any, List, Tuple, Union, Literal
import qrcode
from gsuid_core.bot import Bot
from gsuid_core.models import Event
from gsuid_core.logger import logger
from qrcode.constants import ERROR_CORRECT_L
from gsuid_core.segment import MessageSegment
from ..utils.api import get_sqla
from ..utils.mys_api import mys_api
disnote = '''免责声明:您将通过扫码完成获取米游社sk以及ck。
本Bot将不会保存您的登录状态。
我方仅提供米游社查询及相关游戏内容服务
若您的账号封禁、被盗等处罚与我方无关。
害怕风险请勿扫码!
'''
def get_qrcode_base64(url):
qr = qrcode.QRCode(
version=1,
error_correction=ERROR_CORRECT_L,
box_size=10,
border=4,
)
qr.add_data(url)
qr.make(fit=True)
img = qr.make_image(fill_color='black', back_color='white')
img_byte = io.BytesIO()
img.save(img_byte, format='PNG') # type: ignore
img_byte = img_byte.getvalue()
return base64.b64encode(img_byte).decode()
async def refresh(
code_data: dict,
) -> Union[Tuple[Literal[False], None], Tuple[Literal[True], Any]]:
scanned = False
while True:
await asyncio.sleep(2)
status_data = await mys_api.check_qrcode(
code_data['app_id'], code_data['ticket'], code_data['device']
)
if isinstance(status_data, int):
logger.warning('二维码已过期')
return False, None
if status_data['stat'] == 'Scanned':
if not scanned:
logger.info('二维码已扫描')
scanned = True
continue
if status_data['stat'] == 'Confirmed':
logger.info('二维码已确认')
break
return True, json.loads(status_data['payload']['raw'])
async def qrcode_login(bot: Bot, ev: Event, user_id: str) -> str:
sqla = get_sqla(ev.bot_id)
async def send_msg(msg: str):
await bot.send(msg)
return ''
code_data = await mys_api.create_qrcode_url()
if isinstance(code_data, int):
return await send_msg('链接创建失败...')
try:
im = []
im.append(MessageSegment.text('请使用米游社扫描下方二维码登录：'))
im.append(
MessageSegment.image(
f'base64://{get_qrcode_base64(code_data["url"])}'
)
)
im.append(
MessageSegment.text(
'免责声明:您将通过扫码完成获取米游社sk以及ck。\n'
'本Bot将不会保存您的登录状态。\n'
'我方仅提供米游社查询及相关游戏内容服务,\n'
'若您的账号封禁、被盗等处罚与我方无关。\n'
'害怕风险请勿扫码~'
)
)
await bot.send(MessageSegment.node(im))
except Exception as e:
logger.error(e)
logger.warning(f'[扫码登录] {user_id} 图片发送失败')
status, game_token_data = await refresh(code_data)
if status:
assert game_token_data is not None # 骗过 pyright
logger.info('game_token获取成功')
cookie_token = await mys_api.get_cookie_token(**game_token_data)
stoken_data = await mys_api.get_stoken_by_game_token(
account_id=int(game_token_data['uid']),
game_token=game_token_data['token'],
)
if isinstance(stoken_data, int):
return await send_msg('获取SK失败...')
account_id = game_token_data['uid']
stoken = stoken_data['token']['token']
mid = stoken_data['user_info']['mid']
app_cookie = f'stuid={account_id};stoken={stoken};mid={mid}'
ck = await mys_api.get_cookie_token_by_stoken(
stoken, account_id, app_cookie
)
if isinstance(ck, int):
return await send_msg('获取CK失败...')
ck = ck['cookie_token']
cookie_check = f'account_id={account_id};cookie_token={ck}'
get_uid = await mys_api.get_mihoyo_bbs_info(account_id, cookie_check)
# 剔除除了原神之外的其他游戏
im = None
if isinstance(get_uid, List):
for i in get_uid:
if i['game_id'] == 2:
uid_check = i['game_role_id']
break
else:
im = f'你的米游社账号{account_id}尚未绑定原神账号，请前往米游社操作！'
return await send_msg(im)
else:
im = '请求失败, 请稍后再试...'
return await send_msg(im)
uid_bind = await sqla.get_bind_uid(user_id)
# 没有在gsuid绑定uid的情况
if not uid_bind:
logger.warning('game_token获取失败')
im = '你还没有绑定uid, 请输入[绑定uid123456]绑定你的uid, 再发送[扫码登录]进行绑定'
return await send_msg(im)
if isinstance(cookie_token, int):
return await send_msg('获取CK失败...')
# 比对gsuid数据库和扫码登陆获取到的uid
if str(uid_bind) == uid_check or str(uid_bind) == account_id:
return SimpleCookie(
{
'stoken_v2': stoken_data['token']['token'],
'stuid': stoken_data['user_info']['aid'],
'mid': stoken_data['user_info']['mid'],
'cookie_token': cookie_token['cookie_token'],
}
).output(header='', sep=';')
else:
logger.warning('game_token获取失败')
im = (
f'检测到扫码登录UID{uid_check}与绑定UID{uid_bind}不同, '
'gametoken获取失败, 请重新发送[扫码登录]进行登录！'
)
else:
logger.warning('game_token获取失败')
im = 'game_token获取失败: 二维码已过期'
return await send_msg(im)

9
StarRailUID/utils/api.py Normal file
View File

@ -0,0 +1,9 @@
from gsuid_core.utils.database.api import DBSqla
class SRDBSqla(DBSqla):
def __init__(self) -> None:
super().__init__(is_sr=True)
get_sqla = SRDBSqla().get_sqla

26
StarRailUID/utils/database.py
View File

@ -1,26 +0,0 @@
from typing import Dict
from sqlalchemy import event
from gsuid_core.data_store import get_res_path
from gsuid_core.utils.database.dal import SQLA
is_wal = False
active_sqla: Dict[str, SQLA] = {}
db_url = str(get_res_path().parent / 'GsData.db')
def get_sqla(bot_id) -> SQLA:
if bot_id not in active_sqla:
sqla = SQLA(db_url, bot_id)
active_sqla[bot_id] = sqla
sqla.create_all()
@event.listens_for(sqla.engine.sync_engine, 'connect')
def engine_connect(conn, branch):
if is_wal:
cursor = conn.cursor()
cursor.execute('PRAGMA journal_mode=WAL')
cursor.close()
return active_sqla[bot_id]

9
StarRailUID/utils/message.py Normal file
View File

@ -0,0 +1,9 @@
from typing import Any, Dict
from gsuid_core.bot import Bot
async def send_diff_msg(bot: Bot, code: Any, data: Dict):
for retcode in data:
if code == retcode:
return await bot.send(data[retcode])

118
StarRailUID/utils/mys_api.py
View File

@ -1,32 +1,22 @@
import copy
import random
from string import digits, ascii_letters
from typing import Dict, Union, Literal, Optional, cast
from gsuid_core.utils.api.mys import MysApi
from gsuid_core.utils.api.mys.models import MysSign, SignInfo
from gsuid_core.utils.api.mys.request import BaseMysApi
from gsuid_core.utils.api.mys.models import MysSign, SignInfo, SignList
from gsuid_core.utils.api.mys.tools import (
random_hex,
generate_os_ds,
get_web_ds_token,
)
from ..utils.database import get_sqla
from ..utils.api import get_sqla
from ..sruid_utils.api.mys.api import _API
from ....GenshinUID.GenshinUID.genshinuid_config.gs_config import gsconfig
mysVersion = '2.44.1'
_HEADER = {
'x-rpc-app_version': mysVersion,
'User-Agent': (
'Mozilla/5.0 (iPhone; CPU iPhone OS 13_2_3 like Mac OS X) '
f'AppleWebKit/605.1.15 (KHTML, like Gecko) miHoYoBBS/{mysVersion}'
),
'x-rpc-client_type': '5',
'Referer': 'https://webstatic.mihoyo.com/',
'Origin': 'https://webstatic.mihoyo.com',
}
class _MysApi(MysApi):
class _MysApi(BaseMysApi):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
@ -81,14 +71,58 @@ class _MysApi(MysApi):
sqla = get_sqla('TEMP')
return await sqla.get_user_stoken(uid)
async def get_sign_info(self, uid) -> Union[SignInfo, int]:
# server_id = RECOGNIZE_SERVER.get(str(uid)[0])
is_os = self.check_os(uid)
async def create_qrcode_url(self) -> Union[Dict, int]:
device_id: str = ''.join(random.choices(ascii_letters + digits, k=64))
app_id: str = '8'
data = await self._mys_request(
_API['CREATE_QRCODE'],
'POST',
header={},
data={'app_id': app_id, 'device': device_id},
)
if isinstance(data, Dict):
url: str = data['data']['url']
ticket = url.split('ticket=')[1]
return {
'app_id': app_id,
'ticket': ticket,
'device': device_id,
'url': url,
}
return data
async def get_sign_list(self, uid) -> Union[SignList, int]:
# is_os = self.check_os(uid)
is_os = False
if is_os:
params = {
'act_id': 'e202304121516551',
'lang': 'zh-cn',
}
else:
params = {
'act_id': 'e202304121516551',
'lang': 'zh-cn',
}
data = await self._mys_req_get(
'STAR_RAIL_SIGN_LIST_URL', is_os, params
)
if isinstance(data, Dict):
data = cast(SignList, data['data'])
return data
async def get_sign_info(self, uid) -> Union[SignInfo, int]:
# server_id = RECOGNIZE_SERVER.get(str(uid)[0])
# is_os = self.check_os(uid)
is_os = False
if is_os:
# TODO
params = {
'act_id': 'e202304121516551',
'lang': 'zh-cn',
'region': 'prod_gf_cn',
'uid': uid,
}
header = {
'DS': generate_os_ds(),
}
@ -96,6 +130,8 @@ class _MysApi(MysApi):
params = {
'act_id': 'e202304121516551',
'lang': 'zh-cn',
'region': 'prod_gf_cn',
'uid': uid,
}
header = {}
data = await self._mys_req_get(
@ -112,19 +148,14 @@ class _MysApi(MysApi):
if ck is None:
return -51
if int(str(uid)[0]) < 6:
HEADER = copy.deepcopy(_HEADER)
HEADER = copy.deepcopy(self._HEADER)
HEADER['Cookie'] = ck
HEADER['x-rpc-device_id'] = random_hex(32)
HEADER['x-rpc-app_version'] = '2.44.1'
HEADER['x-rpc-client_type'] = '5'
HEADER['X_Requested_With'] = 'com.mihoyo.hyperion'
HEADER['DS'] = get_web_ds_token(True)
HEADER['Referer'] = (
'https://webstatic.mihoyo.com/bbs/event/signin/hkrpg'
'/mys_sign?act_id=e202304121516551&bbs_auth_required=true'
'&bbs_presentation_style=fullscreen&utm_source=share'
'&utm_medium=bbs&utm_campaign=app'
)
HEADER['Referer'] = 'https://webstatic.mihoyo.com'
HEADER.update(header)
data = await self._mys_request(
url=_API['SIGN_URL'],
@ -132,6 +163,8 @@ class _MysApi(MysApi):
header=HEADER,
data={
'act_id': 'e202304121516551',
'region': 'prod_gf_cn',
'uid': uid,
'lang': 'zh-cn',
},
)
@ -141,5 +174,38 @@ class _MysApi(MysApi):
data = cast(MysSign, data['data'])
return data
async def _mys_req_get(
self,
url: str,
is_os: bool,
params: Dict,
header: Optional[Dict] = None,
) -> Union[Dict, int]:
if is_os:
_URL = _API[f'{url}_OS']
HEADER = copy.deepcopy(self._HEADER_OS)
use_proxy = True
else:
_URL = _API[url]
print(_URL)
HEADER = copy.deepcopy(self._HEADER)
use_proxy = False
if header:
HEADER.update(header)
if 'Cookie' not in HEADER and 'uid' in params:
ck = await self.get_ck(params['uid'])
if ck is None:
return -51
HEADER['Cookie'] = ck
data = await self._mys_request(
url=_URL,
method='GET',
header=HEADER,
params=params,
use_proxy=use_proxy,
)
return data
mys_api = _MysApi()

6
poetry.lock generated
View File

@ -914,14 +914,14 @@ files = [
[[package]]
name = "tomlkit"
version = "0.11.7"
version = "0.11.8"
description = "Style preserving TOML library"
category = "dev"
optional = false
python-versions = ">=3.7"
files = [
{file = "tomlkit-0.11.7-py3-none-any.whl", hash = "sha256:5325463a7da2ef0c6bbfefb62a3dc883aebe679984709aee32a317907d0a8d3c"},
{file = "tomlkit-0.11.7.tar.gz", hash = "sha256:f392ef70ad87a672f02519f99967d28a4d3047133e2d1df936511465fbb3791d"},
{file = "tomlkit-0.11.8-py3-none-any.whl", hash = "sha256:8c726c4c202bdb148667835f68d68780b9a003a9ec34167b6c673b38eff2a171"},
{file = "tomlkit-0.11.8.tar.gz", hash = "sha256:9330fc7faa1db67b541b28e62018c17d20be733177d290a13b24c62d1614e0c3"},
]
[[package]]