✨ 新增(dailycheckin/aliyun): 添加阿里云盘签到功能模块

dailycheckin/aliyun/main.py

@@ -0,0 +1,78 @@
+import json
+import os
+
+import requests
+import urllib3
+
+from dailycheckin import CheckIn
+
+urllib3.disable_warnings()
+
+
+class AliYun(CheckIn):
+    name = "AliYun"
+
+    def __init__(self, check_item: dict):
+        self.check_item = check_item
+
+    def update_token(self, refresh_token):
+        url = "https://auth.aliyundrive.com/v2/account/token"
+        data = {"grant_type": "refresh_token", "refresh_token": refresh_token}
+        response = requests.post(url=url, json=data).json()
+        access_token = response["access_token"]
+        return access_token
+
+    def sign(self, access_token):
+        url = "https://member.aliyundrive.com/v1/activity/sign_in_list"
+        headers = {"Authorization": access_token, "Content-Type": "application/json"}
+        result = requests.post(url=url, headers=headers, json={}).json()
+        sign_days = result["result"]["signInCount"]
+        data = {"signInDay": sign_days}
+        url_reward = "https://member.aliyundrive.com/v1/activity/sign_in_reward"
+        requests.post(url=url_reward, headers=headers, data=json.dumps(data))
+        if "success" in result:
+            print("签到成功")
+            for i, j in enumerate(result["result"]["signInLogs"]):
+                if j["status"] == "miss":
+                    day_json = result["result"]["signInLogs"][i - 1]
+                    if not day_json["isReward"]:
+                        msg = [
+                            {
+                                "name": "阿里云盘",
+                                "value": "签到成功，今日未获得奖励",
+                            }
+                        ]
+                    else:
+                        msg = [
+                            {
+                                "name": "累计签到",
+                                "value": result["result"]["signInCount"],
+                            },
+                            {
+                                "name": "阿里云盘",
+                                "value": "获得奖励：{}{}".format(
+                                    day_json["reward"]["name"],
+                                    day_json["reward"]["description"],
+                                ),
+                            },
+                        ]
+
+                    return msg
+
+    def main(self):
+        refresh_token = self.check_item.get("refresh_token")
+        access_token = self.update_token(refresh_token)
+        msg = self.sign(access_token)
+        msg = "\n".join([f"{one.get('name')}: {one.get('value')}" for one in msg])
+        return msg
+
+
+if __name__ == "__main__":
+    with open(
+        os.path.join(os.path.dirname(os.path.dirname(__file__)), "config.json"),
+        encoding="utf-8",
+    ) as f:
+        datas = json.loads(f.read())
+    _check_item = datas.get("ALIYUN", [])[0]
+    # _check_item = {"refresh_token": "599dcf6504364c6ca362eddf62e3d5c6"}
+    print(AliYun(check_item=_check_item).main())