Merge pull request #1 from y1ndan/master

Master

.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'
 

genshin.py

@@ -18,18 +18,42 @@ logging.basicConfig(
 
 class ConfMeta(type):
   @property
-  def index_url(self):
+  def ref_url(self):
-    return 'https://webstatic.mihoyo.com/bbs/event/signin-ys/index.html'
+    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) ' \
+    return 'Mozilla/5.0 (iPhone; CPU iPhone OS 14_0_1 like Mac OS X) Apple' \
-           'AppleWebKit/605.1.15 (KHTML, like Gecko) miHoYoBBS/%s' \
+    'WebKit/605.1.15 (KHTML, like Gecko) miHoYoBBS/{}'.format(self.app_version)
-           %(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):
+  def get_roles(self):
-    logging.info('Start getting user information ...')
+    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, " \
+        logging.error('HTTP error when get user game roles, ' \
-                    "retry %s time(s) ..." %(i))
+        'retry %s time(s) ...' %(i))
-        logging.error("error is %s" %(e))
+        logging.error('error is %s' %(e))
         errstr = str(e)
         continue
       except KeyError as e:
-        logging.error("Wrong response to get user game roles, " \
+        logging.error('Wrong response to get user game roles, ' \
-                    "retry %s time(s) ..." %(i))
+        'retry %s time(s) ...' %(i))
-        logging.error("response is %s" %(e))
+        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)
+    c = self.md5('salt=' + n + '&t='+ i + '&r=' + r)
-    return i + "," + r + "," + c
+    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):
+  def get_info(self):
-    # cn_gf01:  天空岛
+    roles = Roles(self._cookie).get_roles()
-    # 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))
     try:
-      jdict = json.loads(requests.Session().post(
+      rolesList = roles['data']['list']
-        self._url, headers = self.get_header(),
-        data = json.dumps(data, ensure_ascii=False)).text)
     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):
+def notify(sckey, status, message):
-  return json.dumps(
+  if sckey.startswith('SC'):
-    {
+    logging.info('准备推送通知...')
-      'result': result,
+    url = 'https://sc.ftqq.com/{}.send'.format(sckey)
-      'message': data
+    data = {'text': '原神签到小助手 签到{}'.format(status), 'desp': message}
-    },
-    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}
     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
-    
+    else:
-    return jdict
+      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]
+  cookie = secret[0]
-  sckey=secret[1]
+  sckey = secret[1]
-  jstr = Roles(cookie).get_rolesInfo()
+  seconds = random.randint(10, 300)
-  result = makeResult('Failed', jstr)
+  #seconds = random.randint(1, 3)
-  ret = -1
 
-  try:
+  logging.info('将在 {} 秒后开始任务...'.format(seconds))
-    rolesList = jstr['data']['list']
+  time.sleep(seconds)
-    logging.info('Your account has been bound %s role(s)' %(len(rolesList)))
 
-    for i in range(len(rolesList)):
+  Sign(cookie).run()
-      seconds = random.randint(10, 300)
-      logging.info('Sleep for %s seconds ...' %(seconds))
-      time.sleep(seconds)
 
-      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)