266 lines
8.9 KiB
Python
266 lines
8.9 KiB
Python
'''
|
|
@File : genshin.py
|
|
@Github : https://github.com/y1ndan/genshin-impact-helper
|
|
@Last modified by : y1ndan
|
|
@Last modified time : 2021-01-13 11:10:30
|
|
'''
|
|
import hashlib
|
|
import json
|
|
import random
|
|
import string
|
|
import time
|
|
import uuid
|
|
import os
|
|
|
|
import requests
|
|
from requests.exceptions import HTTPError
|
|
|
|
from settings import log, CONFIG
|
|
from notify import Notify
|
|
|
|
|
|
def hexdigest(text):
|
|
md5 = hashlib.md5()
|
|
md5.update(text.encode())
|
|
return md5.hexdigest()
|
|
|
|
|
|
class Base(object):
|
|
def __init__(self, cookies: str = None):
|
|
if not isinstance(cookies, str):
|
|
raise TypeError('%s want a %s but got %s' %
|
|
(self.__class__, type(__name__), type(cookies)))
|
|
self._cookie = cookies
|
|
|
|
def get_header(self):
|
|
header = {
|
|
'User-Agent': CONFIG.USER_AGENT,
|
|
'Referer': CONFIG.REFERER_URL,
|
|
'Accept-Encoding': 'gzip, deflate, br',
|
|
'Cookie': self._cookie
|
|
}
|
|
return header
|
|
|
|
@staticmethod
|
|
def to_python(json_str: str):
|
|
return json.loads(json_str)
|
|
|
|
@staticmethod
|
|
def to_json(obj):
|
|
return json.dumps(obj, indent=4, ensure_ascii=False)
|
|
|
|
|
|
class Roles(Base):
|
|
def get_awards(self):
|
|
response = dict()
|
|
try:
|
|
content = requests.Session().get(
|
|
CONFIG.AWARD_URL, headers=self.get_header()).text
|
|
response = self.to_python(content)
|
|
except json.JSONDecodeError as e:
|
|
log.error(e)
|
|
|
|
return response
|
|
|
|
def get_roles(self, max_attempt_number: int = 4):
|
|
log.info('准备获取账号信息...')
|
|
error = None
|
|
response = dict()
|
|
|
|
for i in range(1, max_attempt_number):
|
|
try:
|
|
content = requests.Session().get(
|
|
CONFIG.ROLE_URL, headers=self.get_header()).text
|
|
response = self.to_python(content)
|
|
except HTTPError as error:
|
|
log.error(
|
|
'HTTP error when get game roles, retry %s time(s)...' % i)
|
|
log.error('error is %s' % error)
|
|
continue
|
|
except KeyError as error:
|
|
log.error(
|
|
'Wrong response to get game roles, retry %s time(s)...'% i)
|
|
log.error('response is %s' % error)
|
|
continue
|
|
except Exception as error:
|
|
log.error('Unknown error %s, die' % error)
|
|
raise Exception(error)
|
|
error = None
|
|
break
|
|
|
|
if error:
|
|
log.error(
|
|
'Maximum retry times have been reached, error is %s ' % error)
|
|
raise Exception(error)
|
|
if response.get(
|
|
'retcode', 1) != 0 or response.get('data', None) is None:
|
|
raise Exception(response['message'])
|
|
|
|
log.info('账号信息获取完毕')
|
|
return response
|
|
|
|
|
|
class Sign(Base):
|
|
def __init__(self, cookies: str = None):
|
|
super(Sign, self).__init__(cookies)
|
|
self._region_list = []
|
|
self._region_name_list = []
|
|
self._uid_list = []
|
|
|
|
@staticmethod
|
|
def get_ds():
|
|
# v2.3.0-web @povsister & @journey-ad
|
|
n = 'h8w582wxwgqvahcdkpvdhbh2w9casgfl'
|
|
i = str(int(time.time()))
|
|
r = ''.join(random.sample(string.ascii_lowercase + string.digits, 6))
|
|
c = hexdigest('salt=' + n + '&t=' + i + '&r=' + r)
|
|
return '{},{},{}'.format(i, r, c)
|
|
|
|
def get_header(self):
|
|
header = super(Sign, self).get_header()
|
|
header.update({
|
|
'x-rpc-device_id':str(uuid.uuid3(
|
|
uuid.NAMESPACE_URL, self._cookie)).replace('-', '').upper(),
|
|
# 1: ios
|
|
# 2: android
|
|
# 4: pc web
|
|
# 5: mobile web
|
|
'x-rpc-client_type': '5',
|
|
'x-rpc-app_version': CONFIG.APP_VERSION,
|
|
'DS': self.get_ds(),
|
|
})
|
|
return header
|
|
|
|
def get_info(self):
|
|
user_game_roles = Roles(self._cookie).get_roles()
|
|
role_list = user_game_roles.get('data', {}).get('list', [])
|
|
|
|
# role list empty
|
|
if not role_list:
|
|
raise Exception(user_game_roles.get('message', 'Role list empty'))
|
|
|
|
log.info(f'当前账号绑定了 {len(role_list)} 个角色')
|
|
info_list = []
|
|
# cn_gf01: 天空岛
|
|
# cn_qd01: 世界树
|
|
self._region_list = [(i.get('region', 'NA')) for i in role_list]
|
|
self._region_name_list = [(i.get('region_name', 'NA'))
|
|
for i in role_list]
|
|
self._uid_list = [(i.get('game_uid', 'NA')) for i in role_list]
|
|
|
|
log.info('准备获取签到信息...')
|
|
for i in range(len(self._uid_list)):
|
|
info_url = CONFIG.INFO_URL.format(
|
|
self._region_list[i], CONFIG.ACT_ID, self._uid_list[i])
|
|
try:
|
|
content = requests.Session().get(
|
|
info_url, headers=self.get_header()).text
|
|
info_list.append(self.to_python(content))
|
|
except Exception as e:
|
|
raise Exception(e)
|
|
|
|
if not info_list:
|
|
raise Exception('User sign info list is empty')
|
|
log.info('签到信息获取完毕')
|
|
return info_list
|
|
|
|
def run(self):
|
|
info_list = self.get_info()
|
|
message_list = []
|
|
for i in range(len(info_list)):
|
|
today = info_list[i]['data']['today']
|
|
total_sign_day = info_list[i]['data']['total_sign_day']
|
|
awards = Roles(self._cookie).get_awards()['data']['awards']
|
|
uid = str(self._uid_list[i]).replace(
|
|
str(self._uid_list[i])[1:8], '******', 1)
|
|
|
|
log.info(f'准备为旅行者 {i + 1} 号签到...')
|
|
time.sleep(10)
|
|
messgae = {
|
|
'today': today,
|
|
'region_name': self._region_name_list[i],
|
|
'uid': uid,
|
|
'award_name': awards[total_sign_day]['name'],
|
|
'award_cnt': awards[total_sign_day]['cnt'],
|
|
'total_sign_day': total_sign_day,
|
|
'end': '',
|
|
}
|
|
if info_list[i]['data']['is_sign'] is True:
|
|
messgae['award_name'] = awards[total_sign_day - 1]['name']
|
|
messgae['award_cnt'] = awards[total_sign_day - 1]['cnt']
|
|
messgae['status'] = f'👀 旅行者 {i + 1} 号, 你已经签到过了哦'
|
|
message_list.append(self.message.format(**messgae))
|
|
continue
|
|
if info_list[i]['data']['first_bind'] is True:
|
|
messgae['status'] = f'💪 旅行者 {i + 1} 号, 请先前往米游社App手动签到一次'
|
|
message_list.append(self.message.format(**messgae))
|
|
continue
|
|
|
|
data = {
|
|
'act_id': CONFIG.ACT_ID,
|
|
'region': self._region_list[i],
|
|
'uid': self._uid_list[i]
|
|
}
|
|
|
|
try:
|
|
content = requests.Session().post(
|
|
CONFIG.SIGN_URL,
|
|
headers=self.get_header(),
|
|
data=json.dumps(data, ensure_ascii=False)).text
|
|
response = self.to_python(content)
|
|
except Exception as e:
|
|
raise Exception(e)
|
|
code = response.get('retcode', 99999)
|
|
# 0: success
|
|
# -5003: already signed in
|
|
if code != 0:
|
|
message_list.append(response)
|
|
continue
|
|
messgae['total_sign_day'] = total_sign_day + 1
|
|
messgae['status'] = response['message']
|
|
message_list.append(self.message.format(**messgae))
|
|
log.info('签到完毕')
|
|
|
|
return ''.join(message_list)
|
|
|
|
@property
|
|
def message(self):
|
|
return CONFIG.MESSGAE_TEMPLATE
|
|
|
|
|
|
if __name__ == '__main__':
|
|
log.info('任务开始')
|
|
notify = Notify()
|
|
msg_list = []
|
|
ret = success_num = fail_num = 0
|
|
# ============= miHoYo BBS COOKIE ============
|
|
# 此处填米游社的COOKIE
|
|
# 注: Github Actions用户请到Settings->Secrets里设置,Name=COOKIE,Value=<获取的值>
|
|
# 多个账号的COOKIE值之间用 # 号隔开,例如: 1#2#3#4
|
|
COOKIE = ''
|
|
|
|
if os.environ.get('COOKIE', '') != '':
|
|
COOKIE = os.environ['COOKIE']
|
|
|
|
cookie_list = COOKIE.split('#')
|
|
log.info(f'检测到共配置了 {len(cookie_list)} 个帐号')
|
|
for i in range(len(cookie_list)):
|
|
log.info(f'准备为 NO.{i + 1} 账号签到...')
|
|
try:
|
|
msg = f' NO.{i + 1} 账号:{Sign(cookie_list[i]).run()}'
|
|
msg_list.append(msg)
|
|
success_num = success_num + 1
|
|
except Exception as e:
|
|
msg = f' NO.{i + 1} 账号:\n {e}'
|
|
msg_list.append(msg)
|
|
fail_num = fail_num + 1
|
|
log.error(msg)
|
|
ret = -1
|
|
continue
|
|
notify.send(status=f'成功: {success_num} | 失败: {fail_num}', msg=msg_list)
|
|
if ret != 0:
|
|
log.error('异常退出')
|
|
exit(ret)
|
|
log.info('任务结束')
|
|
|