修复仓库扫描 (#707)

来源mowerng

arknights_mower/solvers/cultivate_depot.py

@@ -1,166 +1,42 @@
-import hashlib
-import hmac
 import json
-import time
-from urllib import parse
 
 import requests
 
 from arknights_mower.utils import config
-from arknights_mower.utils.log import logger
 from arknights_mower.utils.path import get_path
-
-app_code = "4ca99fa6b56cc2ba"
-
-# 签到url
-sign_url = "https://zonai.skland.com/api/v1/game/attendance"
-# 绑定的角色url
-binding_url = "https://zonai.skland.com/api/v1/game/player/binding"
-# 验证码url
-login_code_url = "https://as.hypergryph.com/general/v1/send_phone_code"
-# 验证码登录
-token_phone_code_url = "https://as.hypergryph.com/user/auth/v2/token_by_phone_code"
-# 密码登录
-token_password_url = "https://as.hypergryph.com/user/auth/v1/token_by_phone_password"
-# 使用token获得认证代码
-grant_code_url = "https://as.hypergryph.com/user/oauth2/v2/grant"
-# 使用认证代码获得cred
-cred_code_url = "https://zonai.skland.com/api/v1/user/auth/generate_cred_by_code"
+from arknights_mower.utils.skland import (
+    get_binding_list,
+    get_cred_by_token,
+    get_sign_header,
+    header,
+    log,
+)
 
 
 class cultivate:
     def __init__(self):
         self.record_path = get_path("@app/tmp/cultivate.json")
-        self.header = {
-            "cred": "",
-            "User-Agent": "Skland/1.0.1 (com.hypergryph.skland; build:100001014; Android 31; ) Okhttp/4.11.0",
-            "Accept-Encoding": "gzip",
-            "Connection": "close",
-        }
-        self.header_login = {
-            "User-Agent": "Skland/1.0.1 (com.hypergryph.skland; build:100001014; Android 31; ) Okhttp/4.11.0",
-            "Accept-Encoding": "gzip",
-            "Connection": "close",
-        }
-
         self.reward = []
-
-        self.header_for_sign = {"platform": "", "timestamp": "", "dId": "", "vName": ""}
         self.sign_token = ""
         self.all_recorded = True
 
     def start(self):
         for item in config.conf.skland_info:
             if item.isCheck:
-                self.save_param(self.get_cred_by_token(self.log(item)))
-                for i in self.get_binding_list():
+                self.save_param(get_cred_by_token(log(item)))
+                for i in get_binding_list(self.sign_token):
                     if item.cultivate_select == i.get("isOfficial"):
                         body = {"gameId": 1, "uid": i.get("uid")}
                         ingame = f"https://zonai.skland.com/api/v1/game/cultivate/player?uid={i.get('uid')}"
                         resp = requests.get(
                             ingame,
-                            headers=self.get_sign_header(
-                                ingame, "get", body, self.header
+                            headers=get_sign_header(
+                                ingame, "get", body, self.sign_token
                             ),
                         ).json()
                         with open(self.record_path, "w", encoding="utf-8") as file:
                             json.dump(resp, file, ensure_ascii=False, indent=4)
 
     def save_param(self, cred_resp):
-        self.header["cred"] = cred_resp["cred"]
+        header["cred"] = cred_resp["cred"]
         self.sign_token = cred_resp["token"]
-
-    def log(self, account):
-        r = requests.post(
-            token_password_url,
-            json={"phone": account.account, "password": account.password},
-            headers=self.header_login,
-        ).json()
-        if r.get("status") != 0:
-            raise Exception(f'获得token失败：{r["msg"]}')
-        return r["data"]["token"]
-
-    def get_cred_by_token(self, token):
-        return self.get_cred(self.get_grant_code(token))
-
-    def get_grant_code(self, token):
-        response = requests.post(
-            grant_code_url,
-            json={"appCode": app_code, "token": token, "type": 0},
-            headers=self.header_login,
-        )
-        resp = response.json()
-        if response.status_code != 200:
-            raise Exception(f"获得认证代码失败：{resp}")
-        if resp.get("status") != 0:
-            raise Exception(f'获得认证代码失败：{resp["msg"]}')
-        return resp["data"]["code"]
-
-    def get_cred(self, grant):
-        resp = requests.post(
-            cred_code_url, json={"code": grant, "kind": 1}, headers=self.header_login
-        ).json()
-        if resp["code"] != 0:
-            raise Exception(f'获得cred失败：{resp["message"]}')
-        return resp["data"]
-
-    def get_binding_list(self):
-        v = []
-        resp = requests.get(
-            binding_url,
-            headers=self.get_sign_header(binding_url, "get", None, self.header),
-        ).json()
-
-        if resp["code"] != 0:
-            logger.warning(f"请求角色列表出现问题：{resp['message']}")
-            if resp.get("message") == "用户未登录":
-                logger.warning("用户登录可能失效了，请重新运行此程序！")
-                return []
-        for i in resp["data"]["list"]:
-            if i.get("appCode") != "arknights":
-                continue
-            v.extend(i.get("bindingList"))
-        return v
-
-    def get_sign_header(self, url: str, method, body, old_header):
-        h = json.loads(json.dumps(old_header))
-        p = parse.urlparse(url)
-        if method.lower() == "get":
-            h["sign"], header_ca = self.generate_signature(
-                self.sign_token, p.path, p.query
-            )
-        else:
-            h["sign"], header_ca = self.generate_signature(
-                self.sign_token, p.path, json.dumps(body)
-            )
-        for i in header_ca:
-            h[i] = header_ca[i]
-        return h
-
-    def generate_signature(self, token: str, path, body_or_query):
-        """
-        获得签名头
-        接口地址+方法为Get请求？用query否则用body+时间戳+ 请求头的四个重要参数（dId，platform，timestamp，vName）.toJSON()
-        将此字符串做HMAC加密，算法为SHA-256，密钥token为请求cred接口会返回的一个token值
-        再将加密后的字符串做MD5即得到sign
-        :param token: 拿cred时候的token
-        :param path: 请求路径（不包括网址）
-        :param body_or_query: 如果是GET，则是它的query。POST则为它的body
-        :return: 计算完毕的sign
-        """
-        # 总是说请勿修改设备时间，怕不是yj你的服务器有问题吧，所以这里特地-2
-
-        t = str(int(time.time()) - 2)
-        token = token.encode("utf-8")
-        header_ca = json.loads(json.dumps(self.header_for_sign))
-        header_ca["timestamp"] = t
-        header_ca_str = json.dumps(header_ca, separators=(",", ":"))
-        s = path + body_or_query + t + header_ca_str
-        hex_s = hmac.new(token, s.encode("utf-8"), hashlib.sha256).hexdigest()
-        md5 = (
-            hashlib.md5(hex_s.encode("utf-8"))
-            .hexdigest()
-            .encode("utf-8")
-            .decode("utf-8")
-        )
-        return md5, header_ca