add: first commit

.gitignore

@@ -0,0 +1,2 @@
+venv
+.idea

Dockerfile

@@ -0,0 +1,6 @@
+FROM python:3.11-alpine
+WORKDIR /app
+COPY requirements.txt /
+RUN pip install -r /requirements.txt  -i https://pypi.tuna.tsinghua.edu.cn/simple
+COPY . /app
+CMD ["python","main.py"]

MsgSender.py

@@ -0,0 +1,134 @@
+import requests
+import json
+import time
+import hmac
+import hashlib
+import base64
+import urllib.parse
+from loguru import logger
+
+from email.header import Header
+from email.mime.text import MIMEText
+from email.mime.multipart import MIMEMultipart
+from email.utils import parseaddr, formataddr
+import smtplib
+
+
+class TelegramSender(object):
+    def __init__(self, config):
+        self.bot_token = config.get('bot_token')
+        self.chat_id = config.get('chat_id')
+
+    def send_msg(self, msg):
+        if not self.bot_token or not self.chat_id:
+            return False
+
+        url = "https://api.telegram.org/bot" + self.bot_token + "/sendMessage"
+        data = {
+            "chat_id": self.chat_id,
+            "text": msg
+        }
+
+        req = requests.session()
+        try:
+            resp = req.post(url, data=data)
+            result = json.loads(resp.text)['ok']
+        except Exception as e:
+            logger.error('Telegram消息发送失败, 错误%s' % e)
+            return False
+        return result
+
+
+class DingDingSender(object):
+    def __init__(self, config):
+        self.ding_secret = config.get('ding_secret')
+        self.dingding_base_url = config.get('dingding_base_url')
+
+    def send_msg(self, msg):
+        if not self.ding_secret or not self.dingding_base_url:
+            return False
+
+        timestamp = str(round(time.time() * 1000))
+        secret_enc = self.ding_secret.encode('utf-8')
+        string_to_sign = '{}\n{}'.format(timestamp, self.ding_secret)
+        string_to_sign_enc = string_to_sign.encode('utf-8')
+        hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
+        sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
+
+        dingding_url = self.dingding_base_url + "&timestamp=" + timestamp + "&sign=" + sign
+        headers = {'Content-Type': 'application/json;charset=utf-8'}
+        body = {
+            "msgtype": "markdown",
+            "markdown": {
+                "title": "提醒",
+                "text": msg
+            },
+            "at": {
+                "isAtAll": "true"
+            }
+        }
+
+        req = requests.Session()
+        try:
+            resp = req.post(dingding_url, headers=headers, json=body)
+            errcode = json.loads(resp.text)['errcode']
+            result = True if errcode == 0 else False
+        except Exception as e:
+            logger.error('钉钉消息发送失败, 错误%s' % e)
+            return False
+        return result
+
+
+class MailSender(object):
+    def __init__(self, config):
+        self.from_add = config.get('from_add')
+        self.to_add = config.get('to_add')
+        self.auth = config.get('auth')
+        self.smtp_server = config.get('smtp_server')
+        self.smtp_port = config.get('smtp_port')
+
+    @staticmethod
+    def format_add(s):
+        name, addr = parseaddr(s)
+        return formataddr((Header(name, 'utf-8').encode(), addr))
+
+    def send_msg(self, message):
+        if not self.from_add or not self.to_add or not self.auth or not self.smtp_server or not self.smtp_port:
+            return False
+
+        msg = MIMEMultipart('alternative')
+        msg.attach(MIMEText(message, "plain", "utf-8"))
+        msg['From'] = self.format_add("aliyun-sign自动签到 <%s>" % self.from_add)
+        msg['To'] = self.format_add("云盘用户 <%s>" % self.to_add)
+        msg['Subject'] = Header("阿里云盘自动签到", 'utf-8').encode()
+
+        server = smtplib.SMTP_SSL(self.smtp_server, self.smtp_port)
+        server.login(self.from_add, self.auth)
+        server.sendmail(self.from_add, [self.to_add], msg.as_string())
+        server.quit()
+        return True
+
+
+class MessageSender(object):
+
+    def __init__(self):
+        self.channel_handlers = {}
+
+    def register_channel(self, channel_name, handler):
+        self.channel_handlers[channel_name] = handler
+
+    def unregister_channel(self, channel_name):
+        self.channel_handlers.pop(channel_name, None)
+
+    def channel_handle(self, channel_name):
+        return self.channel_handlers.get(channel_name, None)
+
+    def send(self, msg):
+        for channel_name in self.channel_handlers:
+            channel = self.channel_handle(channel_name)
+            if not channel or not hasattr(channel, "send_msg"):
+                continue
+
+            result = channel.send_msg(msg)
+            logger.info('渠道: %s 消息发送结果: %s' % (channel_name, "成功" if result else "失败"))
+

README.md

@@ -1,3 +1,20 @@
-# Aliyunsign
+# 阿里云盘自动签到脚本
 
-阿里云盘自动签到脚本
+### 获取 refresh_token
+
+1. 网页登录阿里云盘官网`https://www.aliyundrive.com/drive`
+2. 按F12, 进入开发者模式, 在菜单栏点击`Application`, 找到`Local Storage`下的域名`https://www.aliyundrive.com`，点击`token`选项，复制`refresh_token`字段的值
+
+### 配置环境变量
+
+编辑config文件
+
+- env_time: 脚本运行的时间
+- refresh_token_list: 阿里云盘refresh_token, 多个refresh_token以逗号分开, 如`"refresh_token_list": "token1,token2"`
+- dingding: 钉钉机器人
+- telegram: telegram机器人
+- mail: 邮件通知
+
+### 启动
+
+`docker-compose  up -d `

config.json

@@ -0,0 +1,19 @@
+{
+  "env_time": 9,
+  "refresh_token_list": "",
+  "dingding": {
+    "ding_secret": "",
+    "dingding_base_url": ""
+  },
+  "telegram": {
+    "bot_token": "",
+    "chat_id": ""
+  },
+  "mail": {
+    "from_add": "",
+    "to_add": "",
+    "auth": "",
+    "smtp_server": "smtp.qq.com",
+    "smtp_port": 465
+  }
+}

docker-compose.yaml

@@ -0,0 +1,9 @@
+version: "3.9"
+services:
+  aliyun-sign:
+    build: .
+    container_name: "aliyun-sign"
+    restart: "always"
+    volumes:
+      - "/etc/localtime:/etc/localtime"
+      - "./config.json:/app/config.json"

main.py

@@ -0,0 +1,108 @@
+import requests
+import json
+import schedule
+import datetime
+import time
+from loguru import logger
+from MsgSender import MessageSender, DingDingSender, TelegramSender, MailSender
+
+
+def aliyundrive_sign(token):
+    update_token_url = 'https://auth.aliyundrive.com/v2/account/token'
+    headers = {
+        'Content-Type': 'application/json'
+    }
+    data = {
+        'grant_type': 'refresh_token',
+        'refresh_token': token
+    }
+
+    req = requests.Session()
+    try:
+        resp = req.post(update_token_url, json=data, headers=headers)
+        result = json.loads(resp.text)
+        access_token = result['access_token']
+        user_name = result['user_name']
+    except Exception as e:
+        log_and_send('', '获取access token请求失败, 错误%s' % e)
+        return
+
+    signin_url = 'https://member.aliyundrive.com/v1/activity/sign_in_list'
+    headers = {
+        'Content-Type': 'application/json',
+        'Authorization': 'Bearer ' + access_token
+    }
+    params = {'_rx-s': 'mobile'}
+
+    try:
+        resp = req.post(signin_url, headers=headers, params=params, json={'isReward': False})
+        result = json.loads(resp.text)
+    except Exception as e:
+        log_and_send(user_name, '签到请求失败, 错误%s' % e)
+        return
+
+    if result.get('code') == 'AccessTokenInvalid':
+        log_and_send(user_name, 'access token 无效')
+        return
+
+    if not result.get('success'):
+        log_and_send(user_name, '阿里云盘签到失败, 错误信息:\n %s' % result)
+        return
+
+    try:
+        signin_count = result['result']['signInCount']
+        reward_url = 'https://member.aliyundrive.com/v1/activity/sign_in_reward'
+        resp = req.post(reward_url, params=params, headers=headers, json={'signInDay': signin_count})
+        result = json.loads(resp.text)
+    except Exception as e:
+        log_and_send(user_name, '阿里云盘签到成功, 领奖请求失败, 错误%s' % e)
+        return
+
+    if not result.get('success'):
+        log_and_send(user_name, '阿里云盘签到成功, 领奖失败, 错误信息:\n %s' % result)
+        return
+
+    log_and_send(user_name, '阿里云盘签到成功，奖品: %s' % result['result']['name'], False)
+
+
+def log_and_send(user_name, msg, is_error=True):
+    msg = '用户%s: %s' % (user_name, msg) if user_name else msg
+    logger.error(msg) if is_error else logger.info(msg)
+    sender.send(msg)
+
+
+def init_sender():
+    msg_sender = MessageSender()
+
+    dingding = "dingding"
+    msg_sender.register_channel(dingding, DingDingSender(config[dingding]))
+
+    telegram = "telegram"
+    msg_sender.register_channel(telegram, TelegramSender(config[telegram]))
+
+    mail = "mail"
+    msg_sender.register_channel(mail, MailSender(config[mail]))
+
+    return msg_sender
+
+
+if __name__ == '__main__':
+    with open('config.json', 'r', encoding='utf-8') as f:
+        config = json.load(f)
+
+    env_time = config['env_time']
+    refresh_token_list = config['refresh_token_list'].split(',')
+    sender = init_sender()
+
+    start_time = datetime.datetime.now().replace(hour=env_time, minute=0, second=0, microsecond=0)
+    for refresh_token in refresh_token_list:
+        schedule_time = start_time.strftime('%H:%M')
+        start_time += datetime.timedelta(seconds=60)
+        logger.info("refresh_token: %s, schedule_time: %s" % (refresh_token, schedule_time))
+        schedule.every().day.at(schedule_time).do(aliyundrive_sign, refresh_token)
+
+    logger.info("应用启动成功")
+    sender.send("阿里云盘自动签到应用启动成功")
+    while True:
+        schedule.run_pending()
+        time.sleep(1)

requirements.txt

@@ -0,0 +1,7 @@
+certifi==2023.11.17
+charset-normalizer==3.3.2
+idna==3.6
+loguru==0.7.2
+requests==2.31.0
+schedule==1.2.1
+urllib3==2.1.0