diff --git a/sample/notify.py b/sample/notify.py index 9e860d1f..bb5fa454 100644 --- a/sample/notify.py +++ b/sample/notify.py @@ -1,139 +1,218 @@ #!/usr/bin/env python3 # _*_ coding:utf-8 _*_ -import sys import os +import re +import sys + cur_path = os.path.abspath(os.path.dirname(__file__)) root_path = os.path.split(cur_path)[0] sys.path.append(root_path) -import requests -import json -import traceback -import time -import hmac -import hashlib + import base64 +import hashlib +import hmac +import json5 as json +import requests +import time import urllib.parse -from requests.adapters import HTTPAdapter -from urllib3.util import Retry -import re # 通知服务 -BARK = '' # bark服务,此参数如果以http或者https开头则判定为自建bark服务; secrets可填; -SCKEY = '' # Server酱的SCKEY; secrets可填 -TG_BOT_TOKEN = '' # tg机器人的TG_BOT_TOKEN; secrets可填 -TG_USER_ID = '' # tg机器人的TG_USER_ID; secrets可填 -TG_PROXY_IP = '' # tg机器人的TG_PROXY_IP; secrets可填 -TG_PROXY_PORT = '' # tg机器人的TG_PROXY_PORT; secrets可填 -DD_BOT_TOKEN = '' # 钉钉机器人的DD_BOT_TOKEN; secrets可填 -DD_BOT_SECRET = '' # 钉钉机器人的DD_BOT_SECRET; secrets可填 -QYWX_APP = '' # 企业微信应用的QYWX_APP; secrets可填 参考http://note.youdao.com/s/HMiudGkb +push_config = { + 'HITOKOTO': False, # 启用一言（随机句子） -notify_mode = [] + 'BARK': '', # bark服务,自行搜索; 此参数如果以http或者https开头则判定为自建bark服务 -# GitHub action运行需要填写对应的secrets -if "BARK" in os.environ and os.environ["BARK"]: - BARK = os.environ["BARK"] -if "SCKEY" in os.environ and os.environ["SCKEY"]: - SCKEY = os.environ["SCKEY"] -if "TG_BOT_TOKEN" in os.environ and os.environ["TG_BOT_TOKEN"] and "TG_USER_ID" in os.environ and os.environ["TG_USER_ID"]: - TG_BOT_TOKEN = os.environ["TG_BOT_TOKEN"] - TG_USER_ID = os.environ["TG_USER_ID"] -if "DD_BOT_TOKEN" in os.environ and os.environ["DD_BOT_TOKEN"] and "DD_BOT_SECRET" in os.environ and os.environ["DD_BOT_SECRET"]: - DD_BOT_TOKEN = os.environ["DD_BOT_TOKEN"] - DD_BOT_SECRET = os.environ["DD_BOT_SECRET"] -if "QYWX_APP" in os.environ and os.environ["QYWX_APP"]: - QYWX_APP = os.environ["QYWX_APP"] + 'SCKEY': '', # Server酱的SCKEY + + '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_IP': '', # tg 机器人的 TG_PROXY_IP + 'TG_PROXY_PORT': '', # tg 机器人的 TG_PROXY_PORT + + 'DD_BOT_ACCESS_TOKEN': '', # 钉钉机器人的 DD_BOT_ACCESS_TOKEN + 'DD_BOT_SECRET': '', # 钉钉机器人的 DD_BOT_SECRET + + 'QQ_MODE': '', # qq 机器人的 QQ_MODE + 'QQ_SKEY': '', # qq 机器人的 QQ_SKEY + + 'QYWX_APP': '', # 企业微信 + + 'PUSH_PLUS_TOKEN': '', # 微信推送 Plus+ + + 'GOBOT_URL': '', # go-cqhttp + # 推送到个人QQ: http://127.0.0.1/send_private_msg + # 群：http://127.0.0.1/send_group_msg + 'GOBOT_TOKEN': '', # go-cqhttp 的 access_token, 可不填 + 'GOBOT_QQ': '', # go-cqhttp的推送群或者用户 + # GOBOT_URL设置 /send_private_msg 填入 user_id=个人QQ + # /send_group_msg 填入 group_id=QQ群 + + +} +notify_function = [] + +# 读取配置文件中的变量 +CONFIG_PATH = os.getenv("NOTIFY_CONFIG_PATH") or "/ql/config/notify_config.json5" +if os.path.exists(CONFIG_PATH): + for k, v in dict(json.load(open(CONFIG_PATH, mode="r", encoding="utf-8"))).items(): + if k in push_config: + push_config[k] = v + +# GitHub action运行环境变量覆盖配置文件的变量 +for k in push_config: + if v := os.getenv(k): + push_config[k] = v -if BARK: - notify_mode.append('bark') - print("BARK 推送打开") -if SCKEY: - notify_mode.append('sc_key') - print("Server酱 推送打开") -if TG_BOT_TOKEN and TG_USER_ID: - notify_mode.append('telegram_bot') - print("Telegram 推送打开") -if DD_BOT_TOKEN and DD_BOT_SECRET: - notify_mode.append('dingding_bot') - print("钉钉机器人 推送打开") -if QYWX_APP: - notify_mode.append('qywxapp_bot') - print("企业微信应用 推送打开") def bark(title, content): print("\n") - if not BARK: - print("bark服务的bark_token未设置!!\n取消推送") + if not push_config.get('BARK'): + print("bark 服务的 bark_token 未设置!!\n取消推送") return - print("bark服务启动") - url = None - if BARK.startswith('http'): - url = f"""{BARK}/{title}/{content}""" + print("bark 服务启动") + + if push_config.get('BARK').startswith('http'): + url = f"""{push_config.get('BARK')}/{title}/{content}""" else: - url = f"""https://api.day.app/{BARK}/{title}/{content}""" + url = f"""https://api.day.app/{push_config.get('BARK')}/{title}/{content}""" response = requests.get(url).json() + if response['code'] == 200: - print('推送成功！') + print('bark 推送成功！') else: - print('推送失败！') + print('bark 推送失败！') + + +def go_cqhttp(title, content): + print("\n") + 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 serverJ(title, content): print("\n") - if not SCKEY: - print("server酱服务的SCKEY未设置!!\n取消推送") + if not push_config.get('SCKEY'): + print("server 酱服务的 SCKEY 未设置!!\n取消推送") return - print("serverJ服务启动") + print("serverJ 服务启动") + data = { "text": title, "desp": content.replace("\n", "\n\n") } - response = requests.post(f"https://sc.ftqq.com/{SCKEY}.send", data=data).json() - if response['errno'] == 0: - print('推送成功！') - else: - print('推送失败！') + response = requests.post(f"https://sct.ftqq.com/{push_config.get('SCKEY')}.send", data=data).json() + if response['errno'] == 0: + print('serverJ 推送成功！') + else: + print('serverJ 推送失败！') + + +# tg通知 def telegram_bot(title, content): print("\n") - bot_token = TG_BOT_TOKEN - user_id = TG_USER_ID - if not bot_token or not user_id: - print("tg服务的bot_token或者user_id未设置!!\n取消推送") + 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服务启动") - url=f"https://api.telegram.org/bot{TG_BOT_TOKEN}/sendMessage" + 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(TG_USER_ID), 'text': f'{title}\n\n{content}', 'disable_web_page_preview': 'true'} + payload = {'chat_id': str(push_config.get('TG_USER_ID')), 'text': f'{title}\n\n{content}', + 'disable_web_page_preview': 'true'} proxies = None - if TG_PROXY_IP and TG_PROXY_PORT: - proxyStr = "http://{}:{}".format(TG_PROXY_IP, TG_PROXY_PORT) + if push_config.get('TG_PROXY_IP') and push_config.get('TG_PROXY_PORT'): + proxyStr = "http://{}:{}".format(push_config.get('TG_PROXY_IP'), 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('推送成功！') + print('tg 推送成功！') else: - print('推送失败！') + print('tg 推送失败！') + def dingding_bot(title, content): - timestamp = str(round(time.time() * 1000)) # 时间戳 - secret_enc = DD_BOT_SECRET.encode('utf-8') - string_to_sign = '{}\n{}'.format(timestamp, DD_BOT_SECRET) + print("\n") + if not push_config.get('DD_BOT_SECRET') or not push_config.get('DD_BOT_ACCESS_TOKEN'): + print("钉钉机器人 服务的 DD_BOT_SECRET 或者 DD_BOT_ACCESS_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)) # 签名 - print('开始使用 钉钉机器人 推送消息...', end='') - url = f'https://oapi.dingtalk.com/robot/send?access_token={DD_BOT_TOKEN}×tamp={timestamp}&sign={sign}' + sign = urllib.parse.quote_plus(base64.b64encode(hmac_code)) + url = f'https://oapi.dingtalk.com/robot/send?access_token={push_config.get("DD_BOT_ACCESS_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('推送成功！') + print('钉钉机器人 推送成功！') else: - print('推送失败！') + print('钉钉机器人 推送失败！') + + +def coolpush_bot(title, content): + print("\n") + if not push_config.get('QQ_SKEY') or not push_config.get('QQ_MODE'): + print("qmsg 的 QQ_SKEY 或者 QQ_MODE 未设置!!\n取消推送") + return + print("qmsg 启动") + + url = f"https://qmsg.zendee.cn/{push_config.get('QQ_MODE')}/{push_config.get('QQ_SKEY')}" + payload = {'msg': f"{title}\n\n{content}".encode('utf-8')} + response = requests.post(url=url, params=payload).json() + + if response['code'] == 0: + print('qmsg 推送成功！') + else: + print('qmsg 推送失败！') + + +# push推送 +def pushplus_bot(title, content): + print("\n") + if not push_config.get('PUSH_PLUS_TOKEN'): + print("PUSHPLUS 服务的token未设置!!\n取消推送") + return + print("PUSHPLUS 服务启动") + + url = 'http://www.pushplus.plus/send' + data = { + "token": push_config.get('PUSH_PLUS_TOKEN'), + "title": title, + "content": content + } + 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: + print('PUSHPLUS 推送失败！') def qywxapp_bot(title, content): print("\n") @@ -228,46 +307,42 @@ def change_user_id(desp): else: return "@all" +def one(): + url = 'https://v1.hitokoto.cn/' + res = requests.get(url).json() + return res['hitokoto'] + ' ----' + res['from'] + + +if push_config.get('BARK'): + notify_function.append(bark) +if push_config.get('GOBOT_URL') and push_config.get('GOBOT_QQ'): + notify_function.append(go_cqhttp) +if push_config.get('SCKEY'): + notify_function.append(serverJ) +if push_config.get('TG_BOT_TOKEN') and push_config.get('TG_USER_ID'): + notify_function.append(telegram_bot) +if push_config.get('DD_BOT_ACCESS_TOKEN') and push_config.get('DD_BOT_SECRET'): + notify_function.append(dingding_bot) +if push_config.get('QQ_SKEY') and push_config.get('QQ_MODE'): + notify_function.append(coolpush_bot) +if push_config.get('PUSH_PLUS_TOKEN'): + notify_function.append(pushplus_bot) +if push_config.get('QYWX_AM'): + notify_function.append(wecom_app) + + def send(title, content): - """ - 使用 bark, telegram bot, dingding bot, serverJ 发送手机推送 - :param title: - :param content: - :return: - """ - for i in notify_mode: - if i == 'bark': - if BARK: - bark(title=title, content=content) - else: - print('未启用 bark') - continue - if i == 'sc_key': - if SCKEY: - serverJ(title=title, content=content) - else: - print('未启用 Server酱') - continue - elif i == 'dingding_bot': - if DD_BOT_TOKEN and DD_BOT_SECRET: - dingding_bot(title=title, content=content) - else: - print('未启用 钉钉机器人') - continue - elif i == 'telegram_bot': - if TG_BOT_TOKEN and TG_USER_ID: - telegram_bot(title=title, content=content) - else: - print('未启用 telegram机器人') - continue - elif i == 'qywxapp_bot': - if QYWX_APP: - qywxapp_bot(title=title, content=content) - else: - print('未启用 企业微信应用推送') - continue - else: - print('此类推送方式不存在') + hitokoto = push_config.get('HITOKOTO') + + text = one() if hitokoto else '' + content += '\n\n' + text + + for mode in notify_function: + try: + mode(title=title, content=content) + except requests.exceptions.RequestException as e: + print(f"网络请求失败： {str(e)}, {mode.__name__}") + def main(): send('title', 'content')