genshin-impact-helper/genshin.py

'''
@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('任务结束')