restructure the folder

app/.dockerignore

@@ -0,0 +1,10 @@
+__pycache__/
+.github/
+.venv/
+.vscode/
+config/config.toml
+config/cookies.json
+.env
+.dockerignore
+.gitignore
+Dockerfile

app/Dockerfile

@@ -0,0 +1,17 @@
+FROM python:alpine as builder
+
+RUN apk update && apk add  --no-cache tzdata ca-certificates
+ADD requirements.txt /tmp/
+RUN pip3 install --user -r /tmp/requirements.txt
+
+
+FROM python:alpine
+WORKDIR /smzdm_bot
+ENV TZ=Asia/Shanghai
+
+COPY --from=builder /root/.local /usr/local
+COPY --from=builder /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/
+COPY --from=builder /usr/share/zoneinfo /usr/share/zoneinfo
+COPY . /smzdm_bot
+
+CMD [ "python", "scheduler.py" ]

app/config/config_example.toml

@@ -0,0 +1,11 @@
+# Cookie
+USER_AGENT = ""
+ANDROID_COOKIE = ""
+SK = ""
+TOKEN = ""
+
+# Notification
+PUSH_PLUS_TOKEN = ""
+SC_KEY = ""
+TG_BOT_TOKEN = ""
+TG_USER_ID = ""

app/main.py

@@ -0,0 +1,119 @@
+import hashlib
+import os
+import random
+import sys
+import time
+from pathlib import Path
+
+import prettytable as pt
+import requests
+from loguru import logger
+
+from notify.notify import NotifyBot
+from utils.file_helper import TomlHelper
+
+CURRENT_PATH = Path(__file__).parent.resolve()
+CONFIG_PATH = Path(CURRENT_PATH, "config")
+
+
+class SmzdmBot(object):
+    KEY = "apr1$AwP!wRRT$gJ/q.X24poeBInlUJC"
+
+    def __init__(self, conf_kwargs: dict):
+        self.conf_kwargs = conf_kwargs
+        self.session = requests.Session()
+        self.start_timestamp = int(time.time())
+        self._set_header()
+
+    def _set_header(self):
+        request_key = f"{random.randint(10000000, 100000000) * 10000000000 + self.start_timestamp}"
+        headers = {
+            "user-agent": self.conf_kwargs.get("USER_AGENT"),
+            "request_key": request_key,
+            "cookie": self.conf_kwargs.get("ANDROID_COOKIE"),
+            "content-type": "application/x-www-form-urlencoded",
+        }
+        self.session.headers = headers
+
+    def _data(self):
+        time = self.start_timestamp * 1000
+        sk = self.conf_kwargs.get("SK")
+        token = self.conf_kwargs.get("TOKEN")
+        sign_str = f"f=android&sk={sk}&time={time}&token={token}&v=10.4.20&weixin=1&key={self.KEY}"
+        sign = self._str_to_md5(sign_str).upper()
+        data = {
+            "weixin": "1",
+            "captcha": "",
+            "f": "android",
+            "v": "10.4.20",
+            "sk": sk,
+            "sign": sign,
+            "touchstone_event": "",
+            "time": time,
+            "token": token,
+        }
+        return data
+
+    def _str_to_md5(self, m: str):
+        return hashlib.md5(m.encode()).hexdigest()
+
+    def checkin(self):
+        url = "https://user-api.smzdm.com/checkin"
+        data = self._data()
+        resp = self.session.post(url, data)
+        if resp.status_code == 200 and int(resp.json()["error_code"]) == 0:
+            resp_data = resp.json()["data"]
+            checkin_num = resp_data["daily_num"]
+            gold = resp_data["cgold"]
+            point = resp_data["cpoints"]
+            exp = resp_data["cexperience"]
+            rank = resp_data["rank"]
+            cards = resp_data["cards"]
+            tb = pt.PrettyTable()
+            tb.field_names = ["签到天数", "金币", "积分", "经验", "等级", "补签卡"]
+            tb.add_row([checkin_num, gold, point, exp, rank, cards])
+            logger.info(f"\n{tb}")
+            msg = f"""⭐签到成功{checkin_num}天
+            🏅金币{gold}
+            🏅积分{point}
+            🏅经验{exp}
+            🏅等级{rank}
+            🏅补签卡{cards}"""
+            return msg
+        else:
+            logger.error("Faile to sign in")
+            msg = "Fail to login in"
+            return msg
+
+
+def main():
+    conf_kwargs = {}
+
+    if Path.exists(Path(CONFIG_PATH, "config.toml")):
+        logger.info("Get configration from config.toml")
+        conf_kwargs = TomlHelper(Path(CONFIG_PATH, "config.toml")).read()
+    elif os.environ.get("ANDROID_COOKIE", None):
+        logger.info("Get configration from env")
+        conf_kwargs = {
+            "USER_AGENT": os.environ.get("USER_AGENT"),
+            "SK": os.environ.get("SK"),
+            "ANDROID_COOKIE": os.environ.get("ANDROID_COOKIE"),
+            "TOKEN": os.environ.get("TOKEN"),
+            "PUSH_PLUS_TOKEN": os.environ.get("PUSH_PLUS_TOKEN", None),
+            "SC_KEY": os.environ.get("SC_KEY", None),
+            "TG_BOT_TOKEN": os.environ.get("TG_BOT_TOKEN", None),
+            "TG_USER_ID": os.environ.get("TG_USER_ID", None),
+            "TG_BOT_API": os.environ.get("TG_BOT_API", None),
+        }
+    else:
+        logger.info("Please set cookies first")
+        sys.exit(1)
+    msg = SmzdmBot(conf_kwargs).checkin()
+    NotifyBot(content=msg, **conf_kwargs)
+    if msg == "Fail to login in":
+        logger.error("Fail the Github action job")
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()

app/notify/notify.py

@@ -0,0 +1,87 @@
+import json
+from typing import Dict
+from urllib.parse import urljoin
+
+import requests
+from loguru import logger
+
+
+class NotifyBot(object):
+    def __init__(self, content, title="什么值得买签到", **kwargs: Dict) -> None:
+        self.content = content
+        self.title = title
+        self.kwargs = kwargs
+
+        self.push_plus()
+        self.server_chain()
+        self.tg_bot()
+
+    def push_plus(self, template="html"):
+        try:
+            if self.kwargs.get("PUSH_PLUS_TOKEN", None):
+                PUSH_PLUS_TOKEN = self.kwargs.get("PUSH_PLUS_TOKEN")
+            else:
+                logger.info("⚠️ PUSH_PLUS_TOKEN not set, skip PushPlus nofitication")
+                return
+
+            url = "https://www.pushplus.plus/send"
+            body = {
+                "token": PUSH_PLUS_TOKEN,
+                "title": self.title,
+                "content": self.content,
+                "template": template,
+            }
+            data = json.dumps(body).encode(encoding="utf-8")
+            headers = {"Content-Type": "application/json"}
+            resp = requests.post(url, data=data, headers=headers)
+            if resp.status_code == 200:
+                logger.info("✅ Push Plus notified")
+            return resp.json()
+        except Exception as e:
+            logger.error(e)
+
+    def server_chain(self):
+        try:
+            if self.kwargs.get("SC_KEY", None):
+                SC_KEY = self.kwargs.get("SC_KEY")
+            else:
+                logger.info("⚠️ SC_KEY not set, skip ServerChain notification")
+                return
+
+            url = f"http://sc.ftqq.com/{SC_KEY}.send"
+
+            data = {"text": self.title, "desp": self.content}
+            resp = requests.post(url, data=data)
+            if resp.status_code == 200:
+                logger.info("✅ Server Chain notified")
+            return resp.json()
+        except Exception as e:
+            logger.error(e)
+
+    def tg_bot(self):
+        try:
+            if self.kwargs.get("TG_BOT_TOKEN", None) and self.kwargs.get(
+                "TG_USER_ID", None
+            ):
+                TG_BOT_TOKEN = self.kwargs.get("TG_BOT_TOKEN")
+                TG_USER_ID = self.kwargs.get("TG_USER_ID")
+            else:
+                logger.info(
+                    "⚠️ TG_BOT_TOKEN & TG_USER_ID not set, skip TelegramBot notification"
+                )
+                return
+
+            TG_BOT_API = self.kwargs.get("TG_BOT_API", "https://api.telegram.org")
+            url = urljoin(TG_BOT_API, f"/bot{TG_BOT_TOKEN}/sendMessage")
+            headers = {"Content-Type": "application/x-www-form-urlencoded"}
+            payload = {
+                "chat_id": str(TG_USER_ID),
+                "text": f"{self.title}\n\n{self.content}",
+                "disable_web_page_preview": "true",
+            }
+            resp = requests.post(url=url, headers=headers, params=payload)
+            if resp.status_code == 200:
+                logger.info("✅ Telegram Bot notified")
+            return resp.json()
+        except Exception as e:
+            logger.error(e)

app/requirements.txt

@@ -0,0 +1,5 @@
+APScheduler==3.10.0
+loguru==0.6.0
+prettytable==3.6.0
+requests==2.28.2
+toml==0.10.2

app/scheduler.py

@@ -0,0 +1,18 @@
+import os
+from random import randint
+
+from apscheduler.schedulers.background import BlockingScheduler
+
+from main import main
+
+if __name__ == "__main__":
+    main()
+    SCH_HOUR = os.environ.get("SCH_HOUR", randint(0, 23))
+    SCH_MINUTE = os.environ.get("SCH_MINUTE", randint(0, 59))
+    scheduler = BlockingScheduler(timezone="Asia/Shanghai")
+    scheduler.add_job(main, "cron", hour=SCH_HOUR, minute=SCH_MINUTE)
+    print("Press Ctrl+{0} to exit".format("Break" if os.name == "nt" else "C"))
+    try:
+        scheduler.start()
+    except (KeyboardInterrupt, SystemExit):
+        pass

app/utils/file_helper.py

@@ -0,0 +1,28 @@
+import toml
+
+
+class TomlHelper:
+    def __init__(self, toml_filename):
+        self.t_dict = dict()
+        self.toml_file_path = toml_filename
+
+    def update(self, t_data):
+        self.t_dict.update(t_data)
+        return self.t_dict
+
+    def write(self, t_data):
+        with open(self.toml_file_path, "w", encoding="utf-8") as fs:
+            toml.dump(t_data, fs)
+
+    def read(self):
+        with open(self.toml_file_path, "r", encoding="utf-8") as fs:
+            t_data = toml.load(fs)
+        return t_data
+
+    def read_str(self, s_data):
+        t_data = toml.loads(s_data, _dict=dict)
+        return t_data
+
+    def read_dict(self, dict):
+        t_data = toml.dumps(dict)
+        return t_data