diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index 211031d..27d9925 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -6,13 +6,13 @@ on: workflow_dispatch: env: - # 是否允许从y1ndan/genshin-impact-helper同步代码 + # auto merge from y1ndan/genshin-impact-helper, default: false ALLOW_MERGE: 'false' jobs: build: runs-on: ubuntu-latest - if: github.ref == 'refs/heads/master' + # if: github.ref == 'refs/heads/master' steps: - name: Checkout master @@ -21,7 +21,7 @@ jobs: fetch-depth: 0 ref: master - - name: Merge From y1ndan/genshin-impact-helper + - name: Auto merge if: ${{ env.ALLOW_MERGE != 'false' }} run: | git config --global user.name github-actions @@ -37,6 +37,7 @@ jobs: - name: Run sign run: | + python -m pip install --upgrade pip pip install -r requirements.txt echo "${{ secrets.COOKIE }}" | tr '#' "\n" | sed 's/$/&#${{ secrets.SCKEY }}/g' | xargs -I {} sh -c 'echo "{}" | python3 ./genshin.py' diff --git a/genshin.py b/genshin.py index 94af84b..3a147d9 100755 --- a/genshin.py +++ b/genshin.py @@ -18,18 +18,42 @@ logging.basicConfig( class ConfMeta(type): @property - def index_url(self): - return 'https://webstatic.mihoyo.com/bbs/event/signin-ys/index.html' + 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.1.0' + return '2.3.0' @property def ua(self): - return 'Mozilla/5.0 (iPhone; CPU iPhone OS 14_0_1 like Mac OS X) ' \ - 'AppleWebKit/605.1.15 (KHTML, like Gecko) miHoYoBBS/%s' \ - %(self.app_version) + 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): @@ -39,99 +63,85 @@ class Conf(metaclass=ConfMeta): class Roles(object): def __init__(self, cookie:str=None): if type(cookie) is not str: - raise TypeError("%s want a %s but got %s" %( + raise TypeError('%s want a %s but got %s' %( self.__class__, type(__name__), type(cookie))) - self._cookie = cookie - self._url = "https://api-takumi.mihoyo.com/binding/api/" \ - "getUserGameRolesByCookie?game_biz=%s" %('hk4e_cn') def get_header(self): - actid = 'e202009291139501' - ref = "%s?bbs_auth_required=%s&act_id=%s&utm_source=%s" \ - "&utm_medium=%s&utm_campaign=%s" %( - Conf.index_url, 'true', actid, 'bbs', 'mys', 'icon') - return { 'User-Agent': Conf.ua, - 'Referer': ref, + 'Referer': Conf.ref_url, 'Accept-Encoding': 'gzip, deflate, br', 'Cookie': self._cookie } - def get_roles(self): + def get_awards(self): try: jdict = json.loads( requests.Session().get( - self._url, headers = self.get_header()).text) + Conf.award_url, headers = self.get_header()).text) except Exception as e: logging.error(e) - raise HTTPError return jdict - def get_rolesInfo(self): - logging.info('Start getting user information ...') + def get_roles(self): + logging.info('准备获取账号信息...') errstr = None for i in range(1, 4): try: - self._rolesInfo = self.get_roles() + 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)) + 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)) + 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)) + logging.error('Unknown error %s, die' %(e)) errstr = str(e) raise else: break try: - self._rolesInfo + jdict except AttributeError: raise Exception(errstr) - - return self._rolesInfo + + 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" %( + raise TypeError('%s want a %s but got %s' %( self.__class__, type(__name__), type(cookie))) - self._cookie = cookie - self._url = 'https://api-takumi.mihoyo.com/event/bbs_sign_reward/sign' - # Provided by Steesha def md5(self, text): md5 = hashlib.md5() md5.update(text.encode()) - return (md5.hexdigest()) + return md5.hexdigest() def get_DS(self): - n = self.md5(Conf.app_version) + # n = self.md5(2.1.0) # v2.1.0 @Steesha + # n = 'cx2y9z9a29tfqvr1qsq6c7yz99b5jsqt' # v2.2.0 @Womsxd + 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) - return i + "," + r + "," + c + c = self.md5('salt=' + n + '&t='+ i + '&r=' + r) + return '{},{},{}'.format(i, r, c) def get_header(self): - actid = 'e202009291139501' - ref = "%s?bbs_auth_required=%s&act_id=%s&utm_source=%s" \ - "&utm_medium=%s&utm_campaign=%s" %( - Conf.index_url, 'true', actid, 'bbs', 'mys', 'icon') - return { 'x-rpc-device_id': str(uuid.uuid3( uuid.NAMESPACE_URL, self._cookie)).replace('-','').upper(), @@ -142,110 +152,134 @@ class Sign(object): 'x-rpc-client_type': '5', 'Accept-Encoding': 'gzip, deflate, br', 'User-Agent': Conf.ua, - 'Referer': ref, + 'Referer': Conf.ref_url, 'x-rpc-app_version': Conf.app_version, 'DS': self.get_DS(), 'Cookie': self._cookie } - def run(self): - # cn_gf01: 天空岛 - # cn_qd01: 世界树 - self._region = rolesList[i]['region'] - self._region_name = rolesList[i]['region_name'] - self._uid = rolesList[i]['game_uid'] - - data = { - 'act_id': 'e202009291139501', - 'region': self._region, - 'uid': self._uid - } - - logging.info('Start signing in the NO.%s role which UID is %s in %s ...' %( - i+1, str(self._uid).replace(str(self._uid)[3:6],'***',1), self._region_name)) + def get_info(self): + roles = Roles(self._cookie).get_roles() try: - jdict = json.loads(requests.Session().post( - self._url, headers = self.get_header(), - data = json.dumps(data, ensure_ascii=False)).text) + rolesList = roles['data']['list'] except Exception as e: - raise + message = roles['message'] + notify(sckey, '失败', message) + exit(-1) + else: + logging.info('当前账号绑定了 {} 个角色'.format(len(rolesList))) + infoList = [] + # 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] - return jdict + logging.info('准备获取签到信息...') + for i in range(len(self._uidList)): + info_url = Conf.info_url.format(self._regionList[i], + Conf.act_id, self._uidList[i]) + try: + infoList.append(json.loads(requests.Session().get( + info_url, headers = self.get_header()).text)) + except Exception as e: + logging.error(e) + + return infoList + + 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] + } + + logging.info('准备为旅行者 {} 号签到...' \ + '\nRegion: {}\nUID: {}'.format(i + 1, self._regionNameList[i], uid)) + try: + jdict = json.loads(requests.Session().post( + Conf.sign_url, headers = self.get_header(), + data = json.dumps(data, ensure_ascii=False)).text) + except Exception as e: + raise + else: + code = jdict['retcode'] + # 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'], '')) + else: + status = '失败' + messageList = jdict + + return notify(sckey, status, messageList) + + def message(self): + return ''' + {:#^30} + 🔅[{}]{} + 今日奖励: {} × {} + 本月累签: {} 天 + 签到结果: {} + {:#^30} + ''' -def makeResult(result:str, data=None): - return json.dumps( - { - 'result': result, - 'message': data - }, - sort_keys=False, indent=2, ensure_ascii=False - ) - -def notify(key, massage): - if key != '': - logging.info('正在推送通知...') - url = 'https://sc.ftqq.com/%s.send' %(key) - data = {'text':'原神签到小助手', 'desp':massage} +def notify(sckey, status, message): + if sckey.startswith('SC'): + logging.info('准备推送通知...') + url = 'https://sc.ftqq.com/{}.send'.format(sckey) + data = {'text': '原神签到小助手 签到{}'.format(status), 'desp': message} try: jdict = json.loads( requests.Session().post(url, data = data).text) - errmsg = jdict['errmsg'] - if errmsg == 'success': - logging.info('推送通知成功') - else: - logging.info('推送通知失败') - logging.error(jdict) except Exception as e: logging.error(e) raise HTTPError - - return jdict + else: + errmsg = jdict['errmsg'] + if errmsg == 'success': + logging.info('推送成功') + else: + logging.error('{}: {}'.format('推送失败', jdict)) else: - logging.info('未配置SCKEY,正在跳过推送...') + logging.info('未配置SCKEY,正在跳过推送') + + logging.info('签到{}: {}'.format(status, message)) + return logging.info('任务结束') -if __name__ == "__main__": +if __name__ == '__main__': secret = input().strip().split('#') secret.append('') - cookie=secret[0] - sckey=secret[1] - jstr = Roles(cookie).get_rolesInfo() - result = makeResult('Failed', jstr) - ret = -1 + cookie = secret[0] + sckey = secret[1] + seconds = random.randint(10, 300) + #seconds = random.randint(1, 3) - try: - rolesList = jstr['data']['list'] - logging.info('Your account has been bound %s role(s)' %(len(rolesList))) + logging.info('将在 {} 秒后开始任务...'.format(seconds)) + time.sleep(seconds) - for i in range(len(rolesList)): - seconds = random.randint(10, 300) - logging.info('Sleep for %s seconds ...' %(seconds)) - time.sleep(seconds) + Sign(cookie).run() - try: - jdict = Sign(cookie).run() - jstr = json.dumps(jdict, ensure_ascii=False) - code = jdict['retcode'] - except Exception as e: - jstr = str(e) - - try: - code - except NameError: - code = -1 - - # 0: success - # -5003: already signed in - if code in [0, -5003]: - result = makeResult('Success', jstr) - ret = 0 - - logging.info(result) - - except Exception as e: - logging.info(result) - - notify(sckey, result) - logging.info('签到完成!') - exit(ret)