From 5966c7f1bde8fe57fee008a01d019d7eb95f1377 Mon Sep 17 00:00:00 2001 From: LuckyHunter Date: Sun, 18 Dec 2022 13:32:18 +0800 Subject: [PATCH] Introduce loguru --- README.md | 4 +++ main.py | 67 ++++++++++++++++++++++++------------------------ notify/notify.py | 61 ++++++++++++++++++++++--------------------- requirements.txt | 3 ++- scheduler.py | 4 +-- 5 files changed, 73 insertions(+), 66 deletions(-) diff --git a/README.md b/README.md index d605adb..3fb0b3e 100644 --- a/README.md +++ b/README.md @@ -76,3 +76,7 @@ SCH_MINUTE= - 使用 Chrome 浏览器访问[什么值得买官网](https://www.smzdm.com/), 登录账号 - 打开开发者工具 (Windows 快捷键`F12`, MacOS 快捷键`option + command + i`) - 选择 Network, 刷新页面, 选择第一个`www.smzdm.com`, 找到`Requests Headers`里的`Cookie` + +### 3.2 连续签到后突然失败 + +Cookies 的有效期暂为止，测试反馈的结果是签到 90 天左右后开始签到失败，此时需要从电脑浏览器端从新签到一次，并更新 cookies diff --git a/main.py b/main.py index 994a679..a95be97 100644 --- a/main.py +++ b/main.py @@ -2,33 +2,35 @@ import json import os import sys from pathlib import Path -from pprint import pp, pprint 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') +CONFIG_PATH = Path(CURRENT_PATH, "config") class SMZDM_Bot(object): DEFAULT_HEADERS = { - 'Accept': '*/*', - 'Accept-Encoding': 'gzip, deflate, br', - 'Accept-Language': 'zh-CN,zh;q=0.9', - 'Connection': 'keep-alive', - 'Host': 'zhiyou.smzdm.com', - 'Referer': 'https://www.smzdm.com/', - 'Sec-Fetch-Dest': 'script', - 'Sec-Fetch-Mode': 'no-cors', - 'Sec-Fetch-Site': 'same-site', - 'User-Agent': ('Mozilla/5.0 (Windows NT 10.0; Win64; x64) ' - 'AppleWebKit/537.36 (KHTML, like Gecko) ' - 'Chrome/106.0.0.0 Safari/537.36 Edg/106.0.1370.42'), + "Accept": "*/*", + "Accept-Encoding": "gzip, deflate, br", + "Accept-Language": "zh-CN,zh;q=0.9", + "Connection": "keep-alive", + "Host": "zhiyou.smzdm.com", + "Referer": "https://www.smzdm.com/", + "Sec-Fetch-Dest": "script", + "Sec-Fetch-Mode": "no-cors", + "Sec-Fetch-Site": "same-site", + "User-Agent": ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/108.0.0.0 Safari/537.36 Edg/108.0.1462.54" + ), } def __init__(self): @@ -39,12 +41,12 @@ class SMZDM_Bot(object): self.session.cookies.update(cookies) def set_cookies(self, cookies): - self.session.headers['Cookie'] = cookies + self.session.headers["Cookie"] = cookies def checkin(self): - url = 'https://zhiyou.smzdm.com/user/checkin/jsonp_checkin' + url = "https://zhiyou.smzdm.com/user/checkin/jsonp_checkin" resp = self.session.get(url) - if resp.status_code == 200: + if resp.status_code == 200 and resp.json()["error_code"] == 0: resp_data = resp.json()["data"] checkin_num = resp_data["checkin_num"] days_of_week = resp_data["continue_checkin_days"] @@ -54,20 +56,19 @@ class SMZDM_Bot(object): rank = resp_data["rank"] cards = resp_data["cards"] tb = pt.PrettyTable() - tb.field_names = ["签到天数", "星期", "金币", "积分", "经验", "等级", "补签卡"] - tb.add_row([checkin_num, days_of_week, - gold, point, exp, rank, cards]) - pprint(tb) - msg = f'''⭐签到成功{checkin_num}天 + tb.field_names = ["签到天数", "连续签到", "金币", "积分", "经验", "等级", "补签卡"] + tb.add_row([checkin_num, days_of_week, gold, point, exp, rank, cards]) + logger.info(f"\n{tb}") + msg = f"""⭐签到成功{checkin_num}天 🏅金币{gold} 🏅积分{point} 🏅经验{exp} 🏅等级{rank} - 🏅补签卡{cards}''' + 🏅补签卡{cards}""" return msg else: - pprint("Faile to sign in") - sys.exit(1) + logger.error("Faile to sign in") + msg = "签到失败,请从浏览器手动签到一次,并更新cookies" def main(): @@ -75,13 +76,12 @@ def main(): conf_kwargs = {} if Path.exists(Path(CONFIG_PATH, "config.toml")): - pprint("Get configration from config.toml") + logger.info("Get configration from config.toml") conf_kwargs = TomlHelper(Path(CONFIG_PATH, "config.toml")).read() - SMZDM_COOKIE = conf_kwargs.get( - "SMZDM_COOKIE").encode('UTF-8').decode('latin-1') + SMZDM_COOKIE = conf_kwargs.get("SMZDM_COOKIE").encode("UTF-8").decode("latin-1") smzdm_bot.set_cookies(SMZDM_COOKIE) elif os.environ.get("SMZDM_COOKIE", None): - pprint("Get configration from env") + logger.info("Get configration from env") conf_kwargs = { "SMZDM_COOKIE": os.environ.get("SMZDM_COOKIE"), "PUSH_PLUS_TOKEN": os.environ.get("PUSH_PLUS_TOKEN", None), @@ -90,11 +90,10 @@ def main(): "TG_USER_ID": os.environ.get("TG_USER_ID", None), "TG_BOT_API": os.environ.get("TG_BOT_API", None), } - SMZDM_COOKIE = conf_kwargs.get( - "SMZDM_COOKIE").encode('UTF-8').decode('latin-1') + SMZDM_COOKIE = conf_kwargs.get("SMZDM_COOKIE").encode("UTF-8").decode("latin-1") smzdm_bot.set_cookies(SMZDM_COOKIE) elif Path.exists(Path(CONFIG_PATH, "cookies.json")): - pprint("Load cookis from cookies.json") + logger.info("Load cookis from cookies.json") with open(Path(CONFIG_PATH, "cookies.json", "r")) as f: cookies = json.load(f) smzdm_cookies = {} @@ -102,11 +101,11 @@ def main(): smzdm_cookies.update({cookie["name"]: cookie["value"]}) smzdm_bot.update_cookies(smzdm_cookies) else: - pprint("Fail to get SMZDM_COOKIE, exit") + logger.info("Fail to get SMZDM_COOKIE, exit") sys.exit(1) msg = smzdm_bot.checkin() NotifyBot(content=msg, **conf_kwargs) -if __name__ == '__main__': +if __name__ == "__main__": main() diff --git a/notify/notify.py b/notify/notify.py index 8d935ff..60c29b3 100644 --- a/notify/notify.py +++ b/notify/notify.py @@ -1,13 +1,12 @@ import json -from pprint import pprint 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 @@ -17,68 +16,72 @@ class NotifyBot(object): self.server_chain() self.tg_bot() - def push_plus(self, template='html'): + 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: - pprint("⚠️ PUSH_PLUS_TOKEN not set, skip PushPlus nofitication") + logger.info("⚠️ PUSH_PLUS_TOKEN not set, skip PushPlus nofitication") return - url = 'https://www.pushplus.plus/send' + url = "https://www.pushplus.plus/send" body = { - 'token': PUSH_PLUS_TOKEN, - 'title': self.title, - 'content': self.content, - 'template': template + "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'} + 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: - pprint("✅ Push Plus notified") + logger.info("✅ Push Plus notified") return resp.json() except Exception as e: - pprint(e) + logger.error(e) def server_chain(self): try: if self.kwargs.get("SC_KEY", None): SC_KEY = self.kwargs.get("SC_KEY") else: - pprint("⚠️ SC_KEY not set, skip ServerChain notification") + logger.info("⚠️ SC_KEY not set, skip ServerChain notification") return - url = f'http://sc.ftqq.com/{SC_KEY}.send' + url = f"http://sc.ftqq.com/{SC_KEY}.send" - data = {'text': self.title, 'desp': self.content} + data = {"text": self.title, "desp": self.content} resp = requests.post(url, data=data) if resp.status_code == 200: - pprint("✅ Server Chain notified") + logger.info("✅ Server Chain notified") return resp.json() except Exception as e: - pprint(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): + 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: - pprint( - "⚠️ TG_BOT_TOKEN & TG_USER_ID not set, skip TelegramBot notification") + 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") + 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'} + 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: - pprint("✅ Telegram Bot notified") + logger.info("✅ Telegram Bot notified") return resp.json() except Exception as e: - pprint(e) + logger.error(e) diff --git a/requirements.txt b/requirements.txt index 35206b4..ea7d3d8 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,5 @@ -APScheduler==3.9.1 +APScheduler==3.9.1.post1 +loguru==0.6.0 prettytable==3.5.0 requests==2.28.1 toml==0.10.2 diff --git a/scheduler.py b/scheduler.py index 5e05abc..785e346 100644 --- a/scheduler.py +++ b/scheduler.py @@ -10,8 +10,8 @@ if __name__ == "__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')) + 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):