From 4acfa156f828c5629bd2ec73b88878b21ac29264 Mon Sep 17 00:00:00 2001 From: gaoyuan Date: Thu, 18 Jan 2024 11:30:45 +0800 Subject: [PATCH] 11 --- README.md | 28 +++ iMTShop.py | 308 ++++++++++++++++++++++++++++++ notify.py | 546 +++++++++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 882 insertions(+) create mode 100644 README.md create mode 100644 iMTShop.py create mode 100644 notify.py diff --git a/README.md b/README.md new file mode 100644 index 0000000..40ee895 --- /dev/null +++ b/README.md @@ -0,0 +1,28 @@ +# imaotai +i茅台自动预约 + +## 抓包 +所有参数是ios逆向出来的 +``` + MTTokenD是茅台预约参数，多个请换行，格式 + '省份,城市,经度,维度,设备id,token,MT-Token-Wap(抓包小茅运) + Mt_Version是app版本号如 1.3.6 必填（填最新版本，目前好像没法获取） + +``` +## 部署 +使用青龙面板部署 +建议每天早上9点之后执行。 +目前支持的茅台类型 +``` + +'10213': '贵州茅台酒（癸卯兔年）', +'2478': '贵州茅台酒（珍品）' +'10214': '贵州茅台酒（癸卯兔年) + +``` +## 使用图 +只支持目前的推送公众号plustoken +![file](https://zhaoxincheng.com/wp-content/uploads/2023/05/2023052406101438.png) +[![](http://zhaoxincheng.com/wp-content/uploads/2021/11/2021110910052362.jpeg)](http://zhaoxincheng.com/wp-content/uploads/2021/11/2021110910052362.jpeg) + +此脚本仅用于学习交流，请勿用于非法途径 diff --git a/iMTShop.py b/iMTShop.py new file mode 100644 index 0000000..67fb14e --- /dev/null +++ b/iMTShop.py @@ -0,0 +1,308 @@ +import datetime +import os +import random +import time + +import requests +import base64 +import json + + +from notify import send + +# 微信公众号@爱上羊毛侠 二次开发 + +# 微信公众号@小白技术社首发 + +# 青龙面板加入环境变量Mt_Version和MTTokenD +# MTTokenD是茅台预约参数，多个请换行，格式'省份,城市,经度,维度,设备id,token,MT-Token-Wap(抓包小茅运)' +# Mt_Version是app版本号如 1.3.6 必填（填最新版本，目前好像没法获取） +# MT-Token-Wap参数是小茅运的领奖励，不需要的话MTTokenD格式改成 省份,城市,经度,维度,设备id,token,'' + +p_c_map = {} +mt_r = 'clips_OlU6TmFRag5rCXwbNAQ/Tz1SKlN8THcecBp/' +# res_map = {'10213': '贵州茅台酒（癸卯兔年）', '2476': '贵州茅台酒（壬寅虎年）', '2478': '贵州茅台酒（珍品）', +# '10214': '贵州茅台酒（癸卯兔年）x2'} +# 下面定义的是申请哪几个，想申请全部的话把上面注释删掉，把下面的注释掉 +res_map = {'10941': '贵州茅台酒（甲辰龙年）', + '10942': '贵州茅台酒（甲辰龙年）x2'} + + +def mt_add(itemId, shopId, sessionId, userId, token, Device_ID): + MT_K = f'{int(time.time() * 1000)}' + r = requests.get( + f'http://82.157.10.108:8086/get_mtv?DeviceID={Device_ID}&MTk={MT_K}&version={mt_version}&key=yaohuo') + headers = {'User-Agent': 'iPhone 14', + 'MT-Token': token, + 'MT-Network-Type': 'WIFI', 'MT-User-Tag': '0', + 'MT-R': mt_r, 'MT-Lat': '', 'MT-K': MT_K, + 'MT-Lng': '', 'MT-Info': '028e7f96f6369cafe1d105579c5b9377', 'MT-APP-Version': mt_version, + 'MT-Request-ID': f'{int(time.time() * 1000)}', 'Accept-Language': 'zh-Hans-CN;q=1', + 'MT-Device-ID': Device_ID, 'MT-V': r.text, + 'MT-Bundle-ID': 'com.moutai.mall', + 'mt-lng': lng, + 'mt-lat': lat} + d = {"itemInfoList": [{"count": 1, "itemId": str(itemId)}], "sessionId": sessionId, "userId": str(userId), + "shopId": str(shopId)} + r = requests.get('http://82.157.10.108:8086/get_actParam?key=yaohuo&actParam=' + base64.b64encode( + json.dumps(d).replace(' ', '').encode('utf8')).decode()) + d['actParam'] = r.text + json_data = d + response = requests.post('https://app.moutai519.com.cn/xhr/front/mall/reservation/add', headers=headers, + json=json_data) + code = response.json().get('code', 0) + if code == 2000: + return response.json().get('data', {}).get('successDesc', "未知") + return '申购失败:' + response.json().get('message', "未知原因") + + +def tongzhi(ss): + user_list = os.getenv('mtec_user', '').split(',') + for user in user_list: + url = 'http://wxpusher.zjiecode.com/api/send/message/?appToken=&content={}&uid={}'.format( + ss, user) + r = requests.get(url) + print(r.text) + + +def get_session_id(device_id, token): + headers = { + 'mt-device-id': device_id, + 'mt-user-tag': '0', + 'accept': '*/*', + 'mt-network-type': 'WIFI', + 'mt-token': token, + 'mt-bundle-id': 'com.moutai.mall', + 'accept-language': 'zh-Hans-CN;q=1', + 'mt-request-id': f'{int(time.time() * 1000)}', + 'mt-app-version': mt_version, + 'user-agent': 'iPhone 14', + 'mt-r': mt_r, + 'mt-lng': lng, + 'mt-lat': lat + } + + response = requests.get('https://static.moutai519.com.cn/mt-backend/xhr/front/mall/index/session/get/' + time_keys, + headers=headers) + sessionId = response.json().get('data', {}).get('sessionId') + itemList = response.json().get('data', {}).get('itemList', []) + itemCodes = [item.get('itemCode') for item in itemList] + return sessionId, itemCodes + + +def get_shop_item(sessionId, itemId, device_id, token, province, city): + headers = { + 'mt-device-id': device_id, + 'mt-user-tag': '0', + 'mt-lat': '', + 'accept': '*/*', + 'mt-network-type': 'WIFI', + 'mt-token': token, + 'mt-bundle-id': 'com.moutai.mall', + 'accept-language': 'zh-Hans-CN;q=1', + 'mt-request-id': f'{int(time.time() * 1000)}', + 'mt-r': mt_r, + 'mt-app-version': mt_version, + 'user-agent': 'iPhone 14', + 'mt-lng': lng, + 'mt-lat': lat + } + + response = requests.get( + 'https://static.moutai519.com.cn/mt-backend/xhr/front/mall/shop/list/slim/v3/' + str( + sessionId) + '/' + province + '/' + str(itemId) + '/' + time_keys, + headers=headers) + data = response.json().get('data', {}) + shops = data.get('shops', []) + shop_id_ = p_c_map[province][city] + for shop in shops: + if not shop.get('shopId') in shop_id_: + continue + if itemId in str(shop): + print("itemId:",itemId,"shopId:",shop.get('shopId')) + return shop.get('shopId') + + +def get_user_id(token, Device_ID): + headers = { + 'MT-User-Tag': '0', + 'Accept': '*/*', + 'MT-Network-Type': 'WIFI', + 'MT-Token': token, + 'MT-Bundle-ID': 'com.moutai.mall', + 'Accept-Language': 'zh-Hans-CN;q=1, en-CN;q=0.9', + 'MT-Request-ID': f'{int(time.time() * 1000)}', + 'MT-APP-Version': mt_version, + 'User-Agent': 'iOS;16.0.1;Apple;iPhone 14 ProMax', + 'MT-R': mt_r, + 'MT-Device-ID': Device_ID, + 'mt-lng': lng, + 'mt-lat': lat + } + + response = requests.get( + 'https://app.moutai519.com.cn/xhr/front/user/info', headers=headers) + userName = response.json().get('data', {}).get('userName') + userId = response.json().get('data', {}).get('userId') + mobile = response.json().get('data', {}).get('mobile') + return userName, userId, mobile + + +def getUserEnergyAward(device_id, ck): + """ + 领取耐力 + :return: + """ + + cookies = { + 'MT-Device-ID-Wap': device_id, + 'MT-Token-Wap': ck, + 'YX_SUPPORT_WEBP': '1', + } + + headers = { + 'X-Requested-With': 'XMLHttpRequest', + 'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 15_2_1 like Mac OS X)', + 'Referer': 'https://h5.moutai519.com.cn/gux/game/main?appConfig=2_1_2', + 'Client-User-Agent': 'iOS;15.0.1;Apple;iPhone 12 ProMax', + 'MT-R': mt_r, + 'Origin': 'https://h5.moutai519.com.cn', + 'MT-APP-Version': mt_version, + 'MT-Request-ID': f'{int(time.time() * 1000)}', + 'Accept-Language': 'zh-CN,zh-Hans;q=0.9', + 'MT-Device-ID': device_id, + 'Accept': 'application/json, text/javascript, */*; q=0.01', + 'mt-lng': lng, + 'mt-lat': lat + } + response = requests.post('https://h5.moutai519.com.cn/game/isolationPage/getUserEnergyAward', cookies=cookies, + headers=headers, json={}) + return response.json().get('message') if '无法领取奖励' in response.text else "领取奖励成功" + + +def get_map(): + global p_c_map + url = 'https://static.moutai519.com.cn/mt-backend/xhr/front/mall/resource/get' + headers = { + 'X-Requested-With': 'XMLHttpRequest', + 'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 15_0_1 like Mac OS X)', + 'Referer': 'https://h5.moutai519.com.cn/gux/game/main?appConfig=2_1_2', + 'Client-User-Agent': 'iOS;16.0.1;Apple;iPhone 14 ProMax', + 'MT-R': mt_r, + 'Origin': 'https://h5.moutai519.com.cn', + 'MT-APP-Version': mt_version, + 'MT-Request-ID': f'{int(time.time() * 1000)}{random.randint(1111111, 999999999)}{int(time.time() * 1000)}', + 'Accept-Language': 'zh-CN,zh-Hans;q=1', + 'MT-Device-ID': f'{int(time.time() * 1000)}{random.randint(1111111, 999999999)}{int(time.time() * 1000)}', + 'Accept': 'application/json, text/javascript, */*; q=0.01', + 'mt-lng': lng, + 'mt-lat': lat + } + res = requests.get(url, headers=headers, ) + mtshops = res.json().get('data', {}).get('mtshops_pc', {}) + urls = mtshops.get('url') + r = requests.get(urls) + for k, v in dict(r.json()).items(): + provinceName = v.get('provinceName') + cityName = v.get('cityName') + if not p_c_map.get(provinceName): + p_c_map[provinceName] = {} + if not p_c_map[provinceName].get(cityName, None): + p_c_map[provinceName][cityName] = [k] + else: + p_c_map[provinceName][cityName].append(k) + return p_c_map + + +def login(phone, vCode, Device_ID): + """ + + :param phone: 手机号 + :param vCode: 验证码 + :param Device_ID: 设备id + :return: + """ + MT_K = f'{int(time.time() * 1000)}' + r = requests.get( + f'http://82.157.10.108:8086/get_mtv?DeviceID={Device_ID}&MTk={MT_K}&version={mt_version}&key=yaohuo') + headers = { + 'MT-Device-ID': Device_ID, + 'MT-User-Tag': '0', + 'Accept': '*/*', + 'MT-Network-Type': 'WIFI', + 'MT-Token': '', + 'MT-K': MT_K, + 'MT-Bundle-ID': 'com.moutai.mall', + 'MT-V': r.text, + 'User-Agent': 'iOS;16.0.1;Apple;iPhone 14 ProMax', + 'Accept-Language': 'zh-Hans-CN;q=1', + 'MT-Request-ID': f'{int(time.time() * 1000)}18342', + 'MT-R': mt_r, + 'MT-APP-Version': mt_version, + } + + json_data = { + 'ydToken': '', + 'mobile': f'{phone}', + 'vCode': f'{vCode}', + 'ydLogId': '', + } + + response = requests.post('https://app.moutai519.com.cn/xhr/front/user/register/login', headers=headers, + json=json_data) + data = response.json().get('data', {}) + token = data.get('token') + cookie = data.get('cookie') # MT-Token-Wap + print(Device_ID, token, cookie) + return Device_ID, token, cookie + + +if __name__ == '__main__': + mt_tokens = os.getenv("MTTokenD") + mt_version = os.getenv("Mt_Version") + if not mt_tokens: + print('MTToken is null') + exit() + if not mt_version: + print('版本号为空 is null') + exit() + mt_token_list = mt_tokens.split('&') + s = "-------------------总共" + \ + str(int(len(mt_token_list))) + \ + "个用户-------------------"+'\n' + userCount = 0 + if len(mt_token_list) > 0: + for mt_token in mt_token_list: + userCount += 1 + print("mt_token:",mt_token) + province, city, lng, lat, device_id, token, ck = mt_token.split( + ',') + time_keys = str( + int(time.mktime(datetime.date.today().timetuple())) * 1000) + get_map() + + try: + sessionId, itemCodes = get_session_id(device_id, token) + userName, user_id, mobile = get_user_id(token, device_id) + if not user_id: + s += "第"+str(userCount)+"个用户token失效，请重新登录"+'\n' + continue + s += "第"+str(userCount)+"个用户----------------"+userName + '_' + \ + mobile + "开始任务" + "----------------"+'\n' + for itemCode in itemCodes: + name = res_map.get(str(itemCode)) + if name: + shop_id = get_shop_item( + sessionId, itemCode, device_id, token, province, city) + res = mt_add(itemCode, str(shop_id), sessionId, + user_id, token, device_id) + s += itemCode + \ + '_' + name + '---------------' + res + '\n' + if not ck: + r = getUserEnergyAward(device_id, ck) + s += userName + '_' + mobile + '---------------' + \ + "小茅运:" + r + '\n' + s += userName + '_' + mobile + "正常结束任务"+'\n \n' + except Exception as e: + s += userName + '_' + mobile + "异常信息"+e + send("i茅台申购+小茅运", s) diff --git a/notify.py b/notify.py new file mode 100644 index 0000000..4e7efa7 --- /dev/null +++ b/notify.py @@ -0,0 +1,546 @@ +#!/usr/bin/env python3 +# _*_ coding:utf-8 _*_ +import base64 +import hashlib +import hmac +import json +import os +import re +import threading +import time +import urllib.parse + +import requests + +# 原先的 print 函数和主线程的锁 +_print = print +mutex = threading.Lock() + + +# 定义新的 print 函数 +def print(text, *args, **kw): + """ + 使输出有序进行，不出现多线程同一时间输出导致错乱的问题。 + """ + with mutex: + _print(text, *args, **kw) + + +# 通知服务 +# fmt: off +push_config = { + 'HITOKOTO': False, # 启用一言（随机句子） + + 'BARK_PUSH': '', # bark IP 或设备码，例：https://api.day.app/DxHcxxxxxRxxxxxxcm/ + 'BARK_ARCHIVE': '', # bark 推送是否存档 + 'BARK_GROUP': '', # bark 推送分组 + 'BARK_SOUND': '', # bark 推送声音 + + 'CONSOLE': True, # 控制台输出 + + 'DD_BOT_SECRET': '', # 钉钉机器人的 DD_BOT_SECRET + 'DD_BOT_TOKEN': '', # 钉钉机器人的 DD_BOT_TOKEN + + 'FSKEY': '', # 飞书机器人的 FSKEY + + 'GOBOT_URL': '', # go-cqhttp + # 推送到个人QQ：http://127.0.0.1/send_private_msg + # 群：http://127.0.0.1/send_group_msg + 'GOBOT_QQ': '', # go-cqhttp 的推送群或用户 + # GOBOT_URL 设置 /send_private_msg 时填入 user_id=个人QQ + # /send_group_msg 时填入 group_id=QQ群 + 'GOBOT_TOKEN': '', # go-cqhttp 的 access_token + + 'GOTIFY_URL': '', # gotify地址,如https://push.example.de:8080 + 'GOTIFY_TOKEN': '', # gotify的消息应用token + 'GOTIFY_PRIORITY': 0, # 推送消息优先级,默认为0 + + 'IGOT_PUSH_KEY': '', # iGot 聚合推送的 IGOT_PUSH_KEY + + 'PUSH_KEY': '', # server 酱的 PUSH_KEY，兼容旧版与 Turbo 版 + + 'PUSH_PLUS_TOKEN': '', # push+ 微信推送的用户令牌 + 'PUSH_PLUS_USER': '', # push+ 微信推送的群组编码 + + 'QMSG_KEY': '', # qmsg 酱的 QMSG_KEY + 'QMSG_TYPE': '', # qmsg 酱的 QMSG_TYPE + + 'QYWX_AM': '', # 企业微信应用 + + 'QYWX_KEY': '', # 企业微信机器人 + + 'TG_BOT_TOKEN': '', # tg 机器人的 TG_BOT_TOKEN，例：1407203283:AAG9rt-6RDaaX0HBLZQq0laNOh898iFYaRQ + 'TG_USER_ID': '', # tg 机器人的 TG_USER_ID，例：1434078534 + 'TG_API_HOST': '', # tg 代理 api + 'TG_PROXY_AUTH': '', # tg 代理认证参数 + 'TG_PROXY_HOST': '', # tg 机器人的 TG_PROXY_HOST + 'TG_PROXY_PORT': '', # tg 机器人的 TG_PROXY_PORT +} +notify_function = [] +# fmt: on + +# 首先读取 面板变量 或者 github action 运行变量 +for k in push_config: + if os.getenv(k): + v = os.getenv(k) + push_config[k] = v + + +def bark(title: str, content: str) -> None: + """ + 使用 bark 推送消息。 + """ + if not push_config.get("BARK_PUSH"): + print("bark 服务的 BARK_PUSH 未设置!!\n取消推送") + return + print("bark 服务启动") + + if push_config.get("BARK_PUSH").startswith("http"): + url = f'{push_config.get("BARK_PUSH")}/{urllib.parse.quote_plus(title)}/{urllib.parse.quote_plus(content)}' + else: + url = f'https://api.day.app/{push_config.get("BARK_PUSH")}/{urllib.parse.quote_plus(title)}/{urllib.parse.quote_plus(content)}' + + bark_params = { + "BARK_ARCHIVE": "isArchive", + "BARK_GROUP": "group", + "BARK_SOUND": "sound", + } + params = "" + for pair in filter( + lambda pairs: pairs[0].startswith("BARK_") + and pairs[0] != "BARK_PUSH" + and pairs[1] + and bark_params.get(pairs[0]), + push_config.items(), + ): + params += f"{bark_params.get(pair[0])}={pair[1]}&" + if params: + url = url + "?" + params.rstrip("&") + response = requests.get(url).json() + + if response["code"] == 200: + print("bark 推送成功！") + else: + print("bark 推送失败！") + + +def console(title: str, content: str) -> None: + """ + 使用 控制台 推送消息。 + """ + print(f"{title}\n\n{content}") + + +def dingding_bot(title: str, content: str) -> None: + """ + 使用 钉钉机器人 推送消息。 + """ + if not push_config.get("DD_BOT_SECRET") or not push_config.get("DD_BOT_TOKEN"): + print("钉钉机器人 服务的 DD_BOT_SECRET 或者 DD_BOT_TOKEN 未设置!!\n取消推送") + return + print("钉钉机器人 服务启动") + + timestamp = str(round(time.time() * 1000)) + secret_enc = push_config.get("DD_BOT_SECRET").encode("utf-8") + string_to_sign = "{}\n{}".format( + timestamp, push_config.get("DD_BOT_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)) + url = f'https://oapi.dingtalk.com/robot/send?access_token={push_config.get("DD_BOT_TOKEN")}×tamp={timestamp}&sign={sign}' + headers = {"Content-Type": "application/json;charset=utf-8"} + data = {"msgtype": "text", "text": {"content": f"{title}\n\n{content}"}} + response = requests.post( + url=url, data=json.dumps(data), headers=headers, timeout=15 + ).json() + + if not response["errcode"]: + print("钉钉机器人 推送成功！") + else: + print("钉钉机器人 推送失败！") + + +def feishu_bot(title: str, content: str) -> None: + """ + 使用 飞书机器人 推送消息。 + """ + if not push_config.get("FSKEY"): + print("飞书 服务的 FSKEY 未设置!!\n取消推送") + return + print("飞书 服务启动") + + url = f'https://open.feishu.cn/open-apis/bot/v2/hook/{push_config.get("FSKEY")}' + data = {"msg_type": "text", "content": {"text": f"{title}\n\n{content}"}} + response = requests.post(url, data=json.dumps(data)).json() + + if response.get("StatusCode") == 0: + print("飞书 推送成功！") + else: + print("飞书 推送失败！错误信息如下：\n", response) + + +def go_cqhttp(title: str, content: str) -> None: + """ + 使用 go_cqhttp 推送消息。 + """ + if not push_config.get("GOBOT_URL") or not push_config.get("GOBOT_QQ"): + print("go-cqhttp 服务的 GOBOT_URL 或 GOBOT_QQ 未设置!!\n取消推送") + return + print("go-cqhttp 服务启动") + + url = f'{push_config.get("GOBOT_URL")}?access_token={push_config.get("GOBOT_TOKEN")}&{push_config.get("GOBOT_QQ")}&message=标题:{title}\n内容:{content}' + response = requests.get(url).json() + + if response["status"] == "ok": + print("go-cqhttp 推送成功！") + else: + print("go-cqhttp 推送失败！") + + +def gotify(title: str, content: str) -> None: + """ + 使用 gotify 推送消息。 + """ + if not push_config.get("GOTIFY_URL") or not push_config.get("GOTIFY_TOKEN"): + print("gotify 服务的 GOTIFY_URL 或 GOTIFY_TOKEN 未设置!!\n取消推送") + return + print("gotify 服务启动") + + url = f'{push_config.get("GOTIFY_URL")}/message?token={push_config.get("GOTIFY_TOKEN")}' + data = {"title": title, "message": content, + "priority": push_config.get("GOTIFY_PRIORITY")} + response = requests.post(url, data=data).json() + + if response.get("id"): + print("gotify 推送成功！") + else: + print("gotify 推送失败！") + + +def iGot(title: str, content: str) -> None: + """ + 使用 iGot 推送消息。 + """ + if not push_config.get("IGOT_PUSH_KEY"): + print("iGot 服务的 IGOT_PUSH_KEY 未设置!!\n取消推送") + return + print("iGot 服务启动") + + url = f'https://push.hellyw.com/{push_config.get("IGOT_PUSH_KEY")}' + data = {"title": title, "content": content} + headers = {"Content-Type": "application/x-www-form-urlencoded"} + response = requests.post(url, data=data, headers=headers).json() + + if response["ret"] == 0: + print("iGot 推送成功！") + else: + print(f'iGot 推送失败！{response["errMsg"]}') + + +def serverJ(title: str, content: str) -> None: + """ + 通过 serverJ 推送消息。 + """ + if not push_config.get("PUSH_KEY"): + print("serverJ 服务的 PUSH_KEY 未设置!!\n取消推送") + return + print("serverJ 服务启动") + + data = {"text": title, "desp": content.replace("\n", "\n\n")} + if push_config.get("PUSH_KEY").index("SCT") != -1: + url = f'https://sctapi.ftqq.com/{push_config.get("PUSH_KEY")}.send' + else: + url = f'https://sc.ftqq.com/${push_config.get("PUSH_KEY")}.send' + response = requests.post(url, data=data).json() + + if response.get("errno") == 0 or response.get("code") == 0: + print("serverJ 推送成功！") + else: + print(f'serverJ 推送失败！错误码：{response["message"]}') + + +def pushplus_bot(title: str, content: str) -> None: + """ + 通过 push+ 推送消息。 + """ + if not push_config.get("PUSH_PLUS_TOKEN"): + print("PUSHPLUS 服务的 PUSH_PLUS_TOKEN 未设置!!\n取消推送") + return + print("PUSHPLUS 服务启动") + + url = "http://www.pushplus.plus/send" + data = { + "token": push_config.get("PUSH_PLUS_TOKEN"), + "title": title, + "content": content, + "topic": push_config.get("PUSH_PLUS_USER"), + } + body = json.dumps(data).encode(encoding="utf-8") + headers = {"Content-Type": "application/json"} + response = requests.post(url=url, data=body, headers=headers).json() + + if response["code"] == 200: + print("PUSHPLUS 推送成功！") + + else: + + url_old = "http://pushplus.hxtrip.com/send" + headers["Accept"] = "application/json" + response = requests.post( + url=url_old, data=body, headers=headers).json() + + if response["code"] == 200: + print("PUSHPLUS(hxtrip) 推送成功！") + + else: + print("PUSHPLUS 推送失败！") + + +def qmsg_bot(title: str, content: str) -> None: + """ + 使用 qmsg 推送消息。 + """ + if not push_config.get("QMSG_KEY") or not push_config.get("QMSG_TYPE"): + print("qmsg 的 QMSG_KEY 或者 QMSG_TYPE 未设置!!\n取消推送") + return + print("qmsg 服务启动") + + url = f'https://qmsg.zendee.cn/{push_config.get("QMSG_TYPE")}/{push_config.get("QMSG_KEY")}' + payload = { + "msg": f'{title}\n\n{content.replace("----", "-")}'.encode("utf-8")} + response = requests.post(url=url, params=payload).json() + + if response["code"] == 0: + print("qmsg 推送成功！") + else: + print(f'qmsg 推送失败！{response["reason"]}') + + +def wecom_app(title: str, content: str) -> None: + """ + 通过 企业微信 APP 推送消息。 + """ + if not push_config.get("QYWX_AM"): + print("QYWX_AM 未设置!!\n取消推送") + return + QYWX_AM_AY = re.split(",", push_config.get("QYWX_AM")) + if 4 < len(QYWX_AM_AY) > 5: + print("QYWX_AM 设置错误!!\n取消推送") + return + print("企业微信 APP 服务启动") + + corpid = QYWX_AM_AY[0] + corpsecret = QYWX_AM_AY[1] + touser = QYWX_AM_AY[2] + agentid = QYWX_AM_AY[3] + try: + media_id = QYWX_AM_AY[4] + except IndexError: + media_id = "" + wx = WeCom(corpid, corpsecret, agentid) + # 如果没有配置 media_id 默认就以 text 方式发送 + if not media_id: + message = title + "\n\n" + content + response = wx.send_text(message, touser) + else: + response = wx.send_mpnews(title, content, media_id, touser) + + if response == "ok": + print("企业微信推送成功！") + else: + print("企业微信推送失败！错误信息如下：\n", response) + + +class WeCom: + def __init__(self, corpid, corpsecret, agentid): + self.CORPID = corpid + self.CORPSECRET = corpsecret + self.AGENTID = agentid + + def get_access_token(self): + url = "https://qyapi.weixin.qq.com/cgi-bin/gettoken" + values = { + "corpid": self.CORPID, + "corpsecret": self.CORPSECRET, + } + req = requests.post(url, params=values) + data = json.loads(req.text) + return data["access_token"] + + def send_text(self, message, touser="@all"): + send_url = ( + "https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token=" + + self.get_access_token() + ) + send_values = { + "touser": touser, + "msgtype": "text", + "agentid": self.AGENTID, + "text": {"content": message}, + "safe": "0", + } + send_msges = bytes(json.dumps(send_values), "utf-8") + respone = requests.post(send_url, send_msges) + respone = respone.json() + return respone["errmsg"] + + def send_mpnews(self, title, message, media_id, touser="@all"): + send_url = ( + "https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token=" + + self.get_access_token() + ) + send_values = { + "touser": touser, + "msgtype": "mpnews", + "agentid": self.AGENTID, + "mpnews": { + "articles": [ + { + "title": title, + "thumb_media_id": media_id, + "author": "Author", + "content_source_url": "", + "content": message.replace("\n", "
"), + "digest": message, + } + ] + }, + } + send_msges = bytes(json.dumps(send_values), "utf-8") + respone = requests.post(send_url, send_msges) + respone = respone.json() + return respone["errmsg"] + + +def wecom_bot(title: str, content: str) -> None: + """ + 通过 企业微信机器人 推送消息。 + """ + if not push_config.get("QYWX_KEY"): + print("企业微信机器人 服务的 QYWX_KEY 未设置!!\n取消推送") + return + print("企业微信机器人服务启动") + + url = f"https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key={push_config.get('QYWX_KEY')}" + headers = {"Content-Type": "application/json;charset=utf-8"} + data = {"msgtype": "text", "text": {"content": f"{title}\n\n{content}"}} + response = requests.post( + url=url, data=json.dumps(data), headers=headers, timeout=15 + ).json() + + if response["errcode"] == 0: + print("企业微信机器人推送成功！") + else: + print("企业微信机器人推送失败！") + + +def telegram_bot(title: str, content: str) -> None: + """ + 使用 telegram 机器人 推送消息。 + """ + if not push_config.get("TG_BOT_TOKEN") or not push_config.get("TG_USER_ID"): + print("tg 服务的 bot_token 或者 user_id 未设置!!\n取消推送") + return + print("tg 服务启动") + + if push_config.get("TG_API_HOST"): + url = f"https://{push_config.get('TG_API_HOST')}/bot{push_config.get('TG_BOT_TOKEN')}/sendMessage" + else: + url = ( + f"https://api.telegram.org/bot{push_config.get('TG_BOT_TOKEN')}/sendMessage" + ) + headers = {"Content-Type": "application/x-www-form-urlencoded"} + payload = { + "chat_id": str(push_config.get("TG_USER_ID")), + "text": f"{title}\n\n{content}", + "disable_web_page_preview": "true", + } + proxies = None + if push_config.get("TG_PROXY_HOST") and push_config.get("TG_PROXY_PORT"): + if push_config.get("TG_PROXY_AUTH") is not None and "@" not in push_config.get( + "TG_PROXY_HOST" + ): + push_config["TG_PROXY_HOST"] = ( + push_config.get("TG_PROXY_AUTH") + + "@" + + push_config.get("TG_PROXY_HOST") + ) + proxyStr = "http://{}:{}".format( + push_config.get("TG_PROXY_HOST"), push_config.get("TG_PROXY_PORT") + ) + proxies = {"http": proxyStr, "https": proxyStr} + response = requests.post( + url=url, headers=headers, params=payload, proxies=proxies + ).json() + + if response["ok"]: + print("tg 推送成功！") + else: + print("tg 推送失败！") + + +def one() -> str: + """ + 获取一条一言。 + :return: + """ + url = "https://v1.hitokoto.cn/" + res = requests.get(url).json() + return res["hitokoto"] + " ----" + res["from"] + + +if push_config.get("BARK_PUSH"): + notify_function.append(bark) +if push_config.get("CONSOLE"): + notify_function.append(console) +if push_config.get("DD_BOT_TOKEN") and push_config.get("DD_BOT_SECRET"): + notify_function.append(dingding_bot) +if push_config.get("FSKEY"): + notify_function.append(feishu_bot) +if push_config.get("GOBOT_URL") and push_config.get("GOBOT_QQ"): + notify_function.append(go_cqhttp) +if push_config.get("GOTIFY_URL") and push_config.get("GOTIFY_TOKEN"): + notify_function.append(gotify) +if push_config.get("IGOT_PUSH_KEY"): + notify_function.append(iGot) +if push_config.get("PUSH_KEY"): + notify_function.append(serverJ) +if push_config.get("PUSH_PLUS_TOKEN"): + notify_function.append(pushplus_bot) +if push_config.get("QMSG_KEY") and push_config.get("QMSG_TYPE"): + notify_function.append(qmsg_bot) +if push_config.get("QYWX_AM"): + notify_function.append(wecom_app) +if push_config.get("QYWX_KEY"): + notify_function.append(wecom_bot) +if push_config.get("TG_BOT_TOKEN") and push_config.get("TG_USER_ID"): + notify_function.append(telegram_bot) + + +def send(title: str, content: str) -> None: + if not content: + print(f"{title} 推送内容为空！") + return + + hitokoto = push_config.get("HITOKOTO") + + text = one() if hitokoto else "" + content += "\n\n" + text + + ts = [ + threading.Thread(target=mode, args=( + title, content), name=mode.__name__) + for mode in notify_function + ] + [t.start() for t in ts] + [t.join() for t in ts] + + +def main(): + send("title", "content") + + +if __name__ == "__main__": + main()