This commit is contained in:
zhipeng wang 2021-01-05 15:38:58 +08:00
parent bcbd35e770
commit ff7d8d6fd2
6 changed files with 383 additions and 260 deletions

View File

@ -6,8 +6,9 @@ on:
workflow_dispatch:
env:
# auto merge from y1ndan/genshin-impact-helper, default: false
# auto merge from PomeloWang/genshin-impact-helper, default: false
ALLOW_MERGE: 'false'
RUN_ENV: 'prod'
jobs:
build:
@ -26,7 +27,7 @@ jobs:
run: |
git config --global user.name github-actions
git config --global user.email github-actions@github.com
git remote add upstream https://github.com/y1ndan/genshin-impact-helper
git remote add upstream https://github.com/PomeloWang/genshin-impact-helper
git pull upstream master --allow-unrelated-histories
git push origin master

56
.gitignore vendored Normal file
View File

@ -0,0 +1,56 @@
*.py[cod]
# C extensions
*.so
# Packages
*.egg
*.egg-info
dist
build
eggs
parts
bin
var
sdist
develop-eggs
.installed.cfg
lib
lib64
__pycache__
# Installer logs
pip-log.txt
# Unit tmp / coverage reports
.coverage
.tox
nosetests.xml
.pytest_cache
# Translations
*.mo
# Mr Developer
.mr.developer.cfg
.project
.pydevproject
# temp file
.DS_Store
*.pkl
# venv
.venv/
# Cookiecutter
output/
# vscode
.vscode
# notebooks
notebooks/
# idea
.idea

1
.python-version Normal file
View File

@ -0,0 +1 @@
genshin-impact-helper

View File

@ -142,6 +142,49 @@ Error: Process completed with exit code 255.
</details>
## 🔨开发
如果需要重构或增加额外功能参考以下数据
```python
roles = Roles(cookie).get_roles()
roles = {
'retcode': 0,
'message': 'OK',
'data': {
'list': [
{
'game_biz': 'hk4e_cn',
'region': 'cn_gf01',
'game_uid': '111111111',
'nickname': '酸柚子',
'level': 48,
'is_chosen': False,
'region_name': '天空岛',
'is_official': True
}
]
}
}
```
```python
infos = Sign(cookie).get_info()
infos = [
{
'retcode': 0,
'message': 'OK',
'data': {
'total_sign_day': 5,
'today': '2021-01-05',
'is_sign': True,
'first_bind': False,
'is_sub': False,
'month_first': False
}
}
]
```
## 🔔订阅
若开启订阅推送,无论成功与否,都会收到微信通知。

404
genshin.py Executable file → Normal file
View File

@ -1,240 +1,218 @@
#!/usr/bin/env python3
import hashlib
import json
import random
import string
import time
import uuid
import requests
import json
import uuid
import logging
import time
import random
import hashlib
import string
from requests.exceptions import *
logging.basicConfig(
level = logging.INFO,
format = '%(asctime)s %(levelname)s %(message)s',
datefmt = '%Y-%m-%dT%H:%M:%S')
from settings import *
class ConfMeta(type):
@property
def ref_url(self):
return 'https://webstatic.mihoyo.com/bbs/event/signin-ys/index.html?' \
'bbs_auth_required={}&act_id={}&utm_source={}&utm_medium={}&' \
'utm_campaign={}'.format('true', self.act_id, 'bbs', 'mys', 'icon')
@property
def award_url(self):
return 'https://api-takumi.mihoyo.com/event/bbs_sign_reward/home?' \
'act_id={}'.format(self.act_id)
@property
def role_url(self):
return 'https://api-takumi.mihoyo.com/binding/api/' \
'getUserGameRolesByCookie?game_biz={}'.format('hk4e_cn')
@property
def info_url(self):
return 'https://api-takumi.mihoyo.com/event/bbs_sign_reward/info?' \
'region={}&act_id={}&uid={}'
@property
def sign_url(self):
return 'https://api-takumi.mihoyo.com/event/bbs_sign_reward/sign'
@property
def app_version(self):
return '2.3.0'
@property
def ua(self):
return 'Mozilla/5.0 (iPhone; CPU iPhone OS 14_0_1 like Mac OS X) Apple' \
'WebKit/605.1.15 (KHTML, like Gecko) miHoYoBBS/{}'.format(self.app_version)
@property
def act_id(self):
return 'e202009291139501'
class Conf(metaclass=ConfMeta):
pass
class Roles(object):
def __init__(self, cookie:str=None):
if type(cookie) is not str:
raise TypeError('%s want a %s but got %s' %(
self.__class__, type(__name__), type(cookie)))
self._cookie = cookie
def get_header(self):
return {
'User-Agent': Conf.ua,
'Referer': Conf.ref_url,
'Accept-Encoding': 'gzip, deflate, br',
'Cookie': self._cookie
}
def get_awards(self):
try:
jdict = json.loads(
requests.Session().get(
Conf.award_url, headers = self.get_header()).text)
except Exception as e:
logging.error(e)
return jdict
def get_roles(self):
logging.info('准备获取账号信息...')
errstr = None
for i in range(1, 4):
try:
jdict = json.loads(requests.Session().get(
Conf.role_url, headers = self.get_header()).text)
except HTTPError as e:
logging.error('HTTP error when get user game roles, ' \
'retry %s time(s) ...' %(i))
logging.error('error is %s' %(e))
errstr = str(e)
continue
except KeyError as e:
logging.error('Wrong response to get user game roles, ' \
'retry %s time(s) ...' %(i))
logging.error('response is %s' %(e))
errstr = str(e)
continue
except Exception as e:
logging.error('Unknown error %s, die' %(e))
errstr = str(e)
raise
else:
break
try:
jdict
except AttributeError:
raise Exception(errstr)
return jdict
class Sign(object):
def __init__(self, cookie:str=None):
if type(cookie) is not str:
raise TypeError('%s want a %s but got %s' %(
self.__class__, type(__name__), type(cookie)))
self._cookie = cookie
def md5(self, text):
def hexdigest(text):
md5 = hashlib.md5()
md5.update(text.encode())
return md5.hexdigest()
def get_DS(self):
# n = self.md5(2.1.0) # v2.1.0 @Steesha
# n = 'cx2y9z9a29tfqvr1qsq6c7yz99b5jsqt' # v2.2.0 @Womsxd
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)
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 user game roles, retry %s time(s) ...' % i)
log.error('error is %s' % error)
continue
except KeyError as error:
log.error('Wrong response to get user 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 error
error = None
break
if error:
log.error('Maximum retry times have been reached, error is %s ' % error)
raise error
if response.get('retcode', 1) != 0 or response.get('data', None) is None:
log.error(response)
exit(-1)
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():
n = 'h8w582wxwgqvahcdkpvdhbh2w9casgfl' # v2.3.0 web @povsister & @journey-ad
i = str(int(time.time()))
r = ''.join(random.sample(string.ascii_lowercase + string.digits, 6))
c = self.md5('salt=' + n + '&t='+ i + '&r=' + r)
c = hexdigest('salt=' + n + '&t=' + i + '&r=' + r)
return '{},{},{}'.format(i, r, c)
def get_header(self):
return {
'x-rpc-device_id': str(uuid.uuid3(
uuid.NAMESPACE_URL, self._cookie)).replace('-','').upper(),
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',
'Accept-Encoding': 'gzip, deflate, br',
'User-Agent': Conf.ua,
'Referer': Conf.ref_url,
'x-rpc-app_version': Conf.app_version,
'DS': self.get_DS(),
'Cookie': self._cookie
}
'x-rpc-app_version': CONFIG.APP_VERSION,
'DS': self.get_ds(),
})
return header
def get_info(self):
roles = Roles(self._cookie).get_roles()
try:
rolesList = roles['data']['list']
except Exception as e:
message = roles['message']
notify(sckey, '失败', message)
user_game_roles = Roles(self._cookie).get_roles()
role_list = user_game_roles.get('data', {}).get('list', [])
# role list empty
if not role_list:
message = user_game_roles.get('message', 'role list empty')
notify(sc_secret, '失败', message)
exit(-1)
else:
logging.info('当前账号绑定了 {} 个角色'.format(len(rolesList)))
infoList = []
log.info('当前账号绑定了 {} 个角色'.format(len(role_list)))
info_list = []
# cn_gf01: 天空岛
# cn_qd01: 世界树
self._regionList = [(i.get('region', 'NA')) for i in rolesList]
self._regionNameList = [(i.get('region_name', 'NA')) for i in rolesList]
self._uidList = [(i.get('game_uid', 'NA')) for i in rolesList]
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]
logging.info('准备获取签到信息...')
for i in range(len(self._uidList)):
info_url = Conf.info_url.format(self._regionList[i],
Conf.act_id, self._uidList[i])
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:
infoList.append(json.loads(requests.Session().get(
info_url, headers = self.get_header()).text))
content = requests.Session().get(info_url, headers=self.get_header()).text
info_list.append(self.to_python(content))
except Exception as e:
logging.error(e)
log.error(e)
return infoList
if not info_list:
log.error("user sign info list is empty, exit...")
exit(-1)
return info_list
def run(self):
logging.info('任务开始')
messageList = []
infoList = self.get_info()
for i in range(len(infoList)):
if infoList[i]['data']['is_sign'] is True:
#if infoList[i]['data']['is_sign'] is False:
message = '旅行者 {} 号,你已经签到过了'.format(i + 1)
notify(sckey, '成功', message)
elif infoList[i]['data']['first_bind'] is True:
message = '旅行者 {} 号,请先前往米游社绑定账号'.format(i + 1)
notify(sckey, '失败', message)
exit(-1)
else:
today = infoList[i]['data']['today']
totalSignDay = infoList[i]['data']['total_sign_day']
award = Roles(self._cookie).get_awards()['data']['awards'][totalSignDay - 1]
uid = str(self._uidList[i]).replace(
str(self._uidList[i])[3:6], '***', 1)
data = {
'act_id': Conf.act_id,
'region': self._regionList[i],
'uid': self._uidList[i]
log.info('任务开始')
status = "成功"
messages = {
'success_message': [],
'failed_message': [],
'already_signed_in': []
}
logging.info('准备为旅行者 {} 号签到...' \
'\nRegion: {}\nUID: {}'.format(i + 1, self._regionNameList[i], uid))
info_list = self.get_info()
for i in range(len(info_list)):
# 已经签到, 处理下一个用户
if info_list[i]['data']['is_sign'] is True:
message = '旅行者 {} 号, 你已经签到过了'.format(i + 1)
messages['already_signed_in'] = messages.get('already_signed_in', []).append(message)
continue
if info_list[i]['data']['first_bind'] is True:
message = '旅行者 {} 号,请先前往米游社绑定账号'.format(i + 1)
messages['failed_message'] = messages.get('failed_message', []).append(message)
exit(-1)
today = info_list[i]['data']['today']
total_sign_day = info_list[i]['data']['total_sign_day']
award = Roles(self._cookie).get_awards()['data']['awards'][total_sign_day - 1]
uid = str(self._uid_list[i]).replace(str(self._uid_list[i])[3:6], '***', 1)
data = {
'act_id': CONFIG.ACT_ID,
'region': self._region_list[i],
'uid': self._uid_list[i]
}
log.info('准备为旅行者 {} 号签到... {}'.format(i + 1, self.to_json({
'Region': self._region_name_list[i],
'UID': uid
})))
try:
jdict = json.loads(requests.Session().post(
Conf.sign_url, headers = self.get_header(),
data = json.dumps(data, ensure_ascii=False)).text)
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
else:
code = jdict['retcode']
raise e
code = response.get('retcode', 99999)
# 0: success
# -5003: already signed in
if code == 0:
status = '成功'
messageList.append(self.message().format(today,
self._regionNameList[i], uid, award['name'], award['cnt'],
totalSignDay, jdict['message'], ''))
message = self.message.format(
today,
self._region_name_list[i],
uid,
award['name'],
award['cnt'],
total_sign_day,
response['message'],
''
)
messages['success_message'] = messages.get('success_message', []).append(message)
else:
status = '失败'
messageList = jdict
messages['failed_message'] = messages.get('failed_message', []).append(response)
return notify(sckey, status, messageList)
if messages.get('failed_message', []):
status = "失败"
return notify(sc_secret, status, messages)
@property
def message(self):
return '''
{:#^30}
@ -246,40 +224,38 @@ class Sign(object):
'''
def notify(sckey, status, message):
if sckey.startswith('SC'):
logging.info('准备推送通知...')
url = 'https://sc.ftqq.com/{}.send'.format(sckey)
def notify(secret: str, status: str, message):
if secret.startswith('SC'):
log.info('准备推送通知...')
url = 'https://sc.ftqq.com/{}.send'.format(secret)
data = {'text': '原神签到小助手 签到{}'.format(status), 'desp': message}
try:
jdict = json.loads(
requests.Session().post(url, data = data).text)
response = Sign.to_python(requests.Session().post(url, data=data).text)
except Exception as e:
logging.error(e)
log.error(e)
raise HTTPError
else:
errmsg = jdict['errmsg']
errmsg = response['errmsg']
if errmsg == 'success':
logging.info('推送成功')
log.info('推送成功')
else:
logging.error('{}: {}'.format('推送失败', jdict))
log.error('{}: {}'.format('推送失败', response))
else:
logging.info('未配置SCKEY,正在跳过推送')
logging.info('签到{}: {}'.format(status, message))
return logging.info('任务结束')
log.info('未配置SCKEY,正在跳过推送')
if isinstance(message, list) or isinstance(message, dict):
message = Sign.to_json(message)
log.info('签到{}: {}'.format(status, message))
return log.info('任务结束')
if __name__ == '__main__':
secret = input().strip().split('#')
secret.append('')
cookie = secret[0]
sckey = secret[1]
seconds = random.randint(10, 300)
#seconds = random.randint(1, 3)
sc_secret = secret[1]
logging.info('将在 {} 秒后开始任务...'.format(seconds))
seconds = random.randint(10, 300)
log.info('将在 {} 秒后开始任务...'.format(seconds))
time.sleep(seconds)
Sign(cookie).run()

46
settings.py Normal file
View File

@ -0,0 +1,46 @@
# settings
import logging
import os
__all__ = ['log', 'CONFIG']
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s %(levelname)s %(message)s',
datefmt='%Y-%m-%dT%H:%M:%S')
log = logger = logging
class _Config:
ACT_ID = 'e202009291139501'
APP_VERSION = '2.3.0'
REFERER_URL = 'https://webstatic.mihoyo.com/bbs/event/signin-ys/index.html?' \
'bbs_auth_required={}&act_id={}&utm_source={}&utm_medium={}&' \
'utm_campaign={}'.format('true', ACT_ID, 'bbs', 'mys', 'icon')
AWARD_URL = 'https://api-takumi.mihoyo.com/event/bbs_sign_reward/home?act_id={}'.format(ACT_ID)
ROLE_URL = 'https://api-takumi.mihoyo.com/binding/api/getUserGameRolesByCookie?game_biz={}'.format('hk4e_cn')
INFO_URL = 'https://api-takumi.mihoyo.com/event/bbs_sign_reward/info?region={}&act_id={}&uid={}'
SIGN_URL = 'https://api-takumi.mihoyo.com/event/bbs_sign_reward/sign'
USER_AGENT = 'Mozilla/5.0 (iPhone; CPU iPhone OS 14_0_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) ' \
'miHoYoBBS/{}'.format(APP_VERSION)
class ProductionConfig(_Config):
LOG_LEVEL = logging.INFO
class DevelopmentConfig(_Config):
LOG_LEVEL = logging.DEBUG
RUN_ENV = os.environ.get('RUN_ENV', 'dev')
if RUN_ENV == 'dev':
CONFIG = DevelopmentConfig()
else:
CONFIG = ProductionConfig()
log.basicConfig(level=CONFIG.LOG_LEVEL)