diff --git a/EMS邮惠中心.py b/EMS邮惠中心.py new file mode 100644 index 0000000..143d190 --- /dev/null +++ b/EMS邮惠中心.py @@ -0,0 +1,295 @@ +""" +作者: Mist (由临渊模板改造) +日期: 2025/08/13 +name: code版_EMS邮惠中心 +入口: EMS邮政快递小程序 +功能: 签到、查询连续签到、查询积分 +变量: emsyhzx_data (openId) 多个账号用换行或#分割 +定时: 一天一次 +cron: 0 9 * * * +------------更新日志------------ +2025/8/13 V1.0 基于临渊模板V1.1进行适配改造 +""" + +import json +import random +import time +import requests +import os +import traceback +import ssl + +MULTI_ACCOUNT_SPLIT = ["\n", "#"] # 分隔符列表 +MULTI_ACCOUNT_PROXY = False # 是否使用多账号代理，默认不使用，True则使用多账号代理 +NOTIFY = os.getenv("LY_NOTIFY") or False # 是否推送日志，默认不推送，True则推送 + +class AutoTask: + def __init__(self, script_name): + """ + 初始化自动任务类 + :param script_name: 脚本名称，用于日志显示 + """ + self.script_name = script_name + self.log_msgs = [] + self.proxy_url = os.getenv("PROXY_API_URL") # 代理api，返回一条txt文本，内容为代理ip:端口 + self.wx_appid = "wx52872495fb375c4b" # EMS小程序id + self.host = "ump.ems.com.cn" + self.token = "" + self.user_id = "" + self.nickname = "" + self.user_agent = "Mozilla/5.0 (iPhone; CPU iPhone OS 17_4_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148 MicroMessenger/8.0.48(0x18003030) NetType/WIFI Language/zh_CN" + + def log(self, msg, level="info"): + """ + 记录并打印日志 + """ + # 获取当前时间 + current_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) + # 格式化日志消息 + formatted_msg = f"[{current_time}] [{level.upper()}] {msg}" + print(formatted_msg) + self.log_msgs.append(msg) # 保存原始消息用于通知 + + def get_proxy(self): + """ + 获取代理 + :return: 代理 + """ + if not self.proxy_url: + self.log("[获取代理] 没有找到环境变量PROXY_API_URL，不使用代理", level="warning") + return None + try: + url = self.proxy_url + response = requests.get(url, timeout=5) + response.raise_for_status() + proxy = response.text + self.log(f"[获取代理] {proxy}") + return proxy + except Exception as e: + self.log(f"[获取代理] 获取代理失败: {e}", level="error") + return None + + def check_env(self): + """ + 检查环境变量 + :return: 环境变量迭代器 + """ + try: + # 从环境变量获取 + env_data = os.getenv("emsyhzx_data") + if not env_data: + self.log("[检查环境变量] 没有找到环境变量emsyhzx_data，请检查环境变量", level="error") + return None + + # 自动检测分隔符 + split_char = None + for sep in MULTI_ACCOUNT_SPLIT: + if sep in env_data: + split_char = sep + break + + accounts = [env_data] if not split_char else env_data.split(split_char) + + for account in accounts: + if account.strip(): + yield account.strip() + except Exception as e: + self.log(f"[检查环境变量] 发生错误: {str(e)}\n{traceback.format_exc()}", level="error") + raise + + def login(self, session, open_id): + """ + 使用openId登录 + :param session: session + :param open_id: 用户的openId + :return: 登录结果 + """ + try: + url = f"https://{self.host}/memberCenterApiV2/member/findByOpenIdAppId" + payload = { + "appId": self.wx_appid, + "openId": open_id, + "source": "JD" + } + response = session.post(url, json=payload, timeout=15) + response_json = response.json() + if response_json.get('code') == "000000": + self.token = response_json['info']['token'] + self.user_id = response_json['info']['memberId'] + self.nickname = f"用户_{str(self.user_id)[-4:]}" # 构造一个昵称 + session.headers['MC-TOKEN'] = self.token + self.log(f"[{self.nickname}] 登录成功") + return True + else: + self.log(f"[登录] 失败，错误信息: {response_json.get('msg', '未知错误')}", level="error") + return False + except requests.RequestException as e: + self.log(f"[登录] 发生网络错误: {str(e)}\n{traceback.format_exc()}", level="error") + return False + except Exception as e: + self.log(f"[登录] 发生错误: {str(e)}\n{traceback.format_exc()}", level="error") + return False + + def sign_in(self, session, open_id): + """ + 签到 + :param session: session + :param open_id: 用户的openId + """ + try: + url = f"https://{self.host}/activCenterApi/signActivInfo/sign" + payload = { + "userId": self.user_id, + "appId": self.wx_appid, + "openId": open_id, + "activId": "c0c4a0a3ef8145f49f2e294741a3cd62" + } + response = session.post(url, json=payload, timeout=15) + response_json = response.json() + if response_json.get('code') == "000000": + prize_size = response_json.get('info', [{}])[0].get('prizeSize', '未知') + self.log(f"[{self.nickname}] 签到成功，获得[{prize_size}]积分") + return True + else: + self.log(f"[{self.nickname}] 签到失败: {response_json.get('msg', '未知错误')}", level="warning") + return False + except Exception as e: + self.log(f"[{self.nickname}] 签到时发生错误: {str(e)}\n{traceback.format_exc()}", level="error") + return False + + def check_cumulative_sign_in(self, session, open_id): + """ + 查询连续签到天数 + :param session: session + :param open_id: 用户的openId + """ + try: + url = f"https://{self.host}/activCenterApi/signActivInfo/querySignDetail" + payload = { + "userId": self.user_id, + "appId": self.wx_appid, + "openId": open_id, + "activId": "d191dce0740849b1b7377e83c00475d6", + } + response = session.post(url, json=payload, timeout=15) + response_json = response.json() + if response_json.get('code') == '000000': + sign_day = response_json.get('info', {}).get('signDay', '未知') + self.log(f"[{self.nickname}] 当前累计签到: [{sign_day}]天") + return True + else: + self.log(f"[{self.nickname}] 查询累计签到失败: {response_json.get('msg', '未知错误')}", level="error") + return False + except Exception as e: + self.log(f"[{self.nickname}] 查询累计签到时发生错误: {str(e)}\n{traceback.format_exc()}", level="error") + return False + + def get_user_points(self, session): + """ + 获取用户积分 + :param session: session + """ + try: + url = f"https://{self.host}/memberCenterApiV2/golds/memberGoldsInfo" + response = session.post(url, json={}, timeout=15) + response_json = response.json() + if response_json.get('code') == '000000': + points = response_json.get('info', {}).get('availableGoldsTotal', '未知') + self.log(f"[{self.nickname}] 查询成功，当前总积分: [{points}]") + return True + else: + self.log(f"[{self.nickname}] 查询积分失败: {response_json.get('msg', '未知错误')}", level="error") + return False + except Exception as e: + self.log(f"[{self.nickname}] 查询积分时发生错误: {str(e)}\n{traceback.format_exc()}", level="error") + return False + + def run(self): + """ + 运行任务 + """ + try: + self.log(f"【{self.script_name}】开始执行任务") + accounts = self.check_env() + if not accounts: + return + + for index, open_id in enumerate(accounts, 1): + # 清理账号信息 + self.nickname = "" + self.token = "" + self.user_id = "" + self.log("") + self.log(f"------ 【账号{index}】开始执行任务 ------") + + session = requests.Session() + headers = { + "User-Agent": self.user_agent, + "Content-Type": 'application/json', + "authority": self.host + } + session.headers.update(headers) + + if MULTI_ACCOUNT_PROXY: + proxy = self.get_proxy() + if proxy: + session.proxies.update({"http": f"http://{proxy}", "https": f"http://{proxy}"}) + + # 1. 登录 + if not self.login(session, open_id): + self.log(f"账号{index} 登录失败，跳过该账号", level="error") + session.close() + continue + + time.sleep(random.randint(2, 4)) + + # 2. 签到 + self.sign_in(session, open_id) + time.sleep(random.randint(2, 4)) + + # 3. 查询连续签到 + self.check_cumulative_sign_in(session, open_id) + time.sleep(random.randint(2, 4)) + + # 4. 查询积分 + self.get_user_points(session) + + self.log(f"------ 【账号{index}】执行任务完成 ------") + # 清理session + session.close() + + except Exception as e: + self.log(f"【{self.script_name}】执行过程中发生错误: {str(e)}\n{traceback.format_exc()}", level="error") + finally: + if NOTIFY: + # 如果notify模块不存在，从远程下载至本地 + if not os.path.exists("notify.py"): + try: + url = "https://raw.githubusercontent.com/whyour/qinglong/refs/heads/develop/sample/notify.py" + response = requests.get(url) + response.raise_for_status() + with open("notify.py", "w", encoding="utf-8") as f: + f.write(response.text) + self.log("下载notify.py成功") + import notify + except Exception as e: + self.log(f"下载notify.py失败: {e}", level="error") + else: + import notify + + # 任务结束后推送日志 + try: + title = f"{self.script_name} 运行日志" + header = "作者：Mist (由临渊模板改造)\n" + content = header + "\n" +"\n".join(self.log_msgs) + notify.send(title, content) + self.log("日志已通过notify发送") + except NameError: + self.log("notify模块未成功导入，无法发送通知", level="warning") + except Exception as e: + self.log(f"通过notify发送日志时发生错误: {e}", level="error") + + +if __name__ == "__main__": + auto_task = AutoTask("EMS邮惠中心") + auto_task.run() \ No newline at end of file