diff --git a/config/config.ini b/config/config.ini index be6585b..4bc6961 100644 --- a/config/config.ini +++ b/config/config.ini @@ -27,6 +27,8 @@ ts录制完成后自动增加生成m4a格式 = 否 钉钉推送接口链接 = 微信推送接口链接 = bark推送接口链接 = +bark推送中断级别 = active +bark推送铃声 = 钉钉通知@对象(填手机号) = tgapi令牌 = tg聊天id(个人或者群组id) = diff --git a/main.py b/main.py index 17c5124..ebcc3b1 100644 --- a/main.py +++ b/main.py @@ -4,7 +4,7 @@ Author: Hmily GitHub: https://github.com/ihmily Date: 2023-07-17 23:52:05 -Update: 2024-09-19 01:50:12 +Update: 2024-09-24 20:49:12 Copyright (c) 2023-2024 by Hmily, All Rights Reserved. Function: Record live stream video. """ @@ -637,12 +637,12 @@ def push_message(content: str) -> Union[str, list]: if '邮箱' in live_status_push: push_pts.append('邮箱') email_message(mail_host, mail_password, from_email, to_email, "直播间状态更新通知", content) - if 'TG' in live_status_push or 'tg' in live_status_push: + if 'TG' in live_status_push.upper(): push_pts.append('TG') tg_bot(tg_chat_id, tg_token, content) - if 'BARK' in live_status_push or 'bark' in live_status_push: + if 'BARK' in live_status_push.upper(): push_pts.append('BARK') - bark(bark_api_url, content) + bark(bark_msg_api, title="直播录制通知", content=content, level=bark_msg_level, sound=bark_msg_ring) push_pts = '、'.join(push_pts) if len(push_pts) > 0 else [] return push_pts @@ -1679,7 +1679,9 @@ while True: live_status_push = read_config_value(config, '推送配置', '直播状态通知(可选微信|钉钉|tg|邮箱|bark或者都填)', "") dingtalk_api_url = read_config_value(config, '推送配置', '钉钉推送接口链接', "") xizhi_api_url = read_config_value(config, '推送配置', '微信推送接口链接', "") - bark_api_url = read_config_value(config, '推送配置', 'bark推送接口链接', "") + bark_msg_api = read_config_value(config, '推送配置', 'bark推送接口链接', "") + bark_msg_level = read_config_value(config, '推送配置', 'bark推送中断级别', "active") + bark_msg_ring = read_config_value(config, '推送配置', 'bark推送铃声', "bell") dingtalk_phone_num = read_config_value(config, '推送配置', '钉钉通知@对象(填手机号)', "") tg_token = read_config_value(config, '推送配置', 'tgapi令牌', "") tg_chat_id = read_config_value(config, '推送配置', 'tg聊天id(个人或者群组id)', "") @@ -1917,4 +1919,4 @@ while True: first_run = False - time.sleep(3) + time.sleep(3) \ No newline at end of file diff --git a/msg_push.py b/msg_push.py index 2201e85..ac9dd16 100644 --- a/msg_push.py +++ b/msg_push.py @@ -4,8 +4,8 @@ Author: Hmily GitHub: https://github.com/ihmily Date: 2023-09-03 19:18:36 -Update: 2024-09-18 12:30:12 -Copyright (c) 2023 by Hmily, All Rights Reserved. +Update: 2024-09-24 20:43:12 +Copyright (c) 2023-2024 by Hmily, All Rights Reserved. """ from typing import Dict, Any, Optional import json @@ -15,8 +15,6 @@ import smtplib from email.header import Header from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText -from urllib.parse import urljoin -import urllib.parse no_proxy_handler = urllib.request.ProxyHandler({}) opener = urllib.request.build_opener(no_proxy_handler) @@ -60,8 +58,7 @@ def xizhi(url: str, content: str) -> Dict[str, Any]: @trace_error_decorator def email_message(mail_host: str, mail_pass: str, from_email: str, to_email: str, title: str, content: str) -> Dict[ - str, Any]: - + str, Any]: message = MIMEMultipart() message['From'] = "{}".format(from_email) message['Subject'] = Header(title, 'utf-8') @@ -96,11 +93,26 @@ def tg_bot(chat_id: int, token: str, content: str) -> Dict[str, Any]: json_data = json.loads(json_str) return json_data + @trace_error_decorator -def bark(url: str, content: str) -> Dict[str, Any]: - json_data = {"body": content} +def bark(api: str, title: str = "message", content: str = 'test', level: str = "active", + badge: int = 1, autoCopy: int = 1, sound: str = "", icon: str = "", group: str = "", + isArchive: int = 1, url: str = "") -> Dict[str, Any]: + json_data = { + "title": title, + "body": content, + "level": level, + "badge": badge, + "autoCopy": autoCopy, + "sound": sound, + "icon": icon, + "group": group, + "isArchive": isArchive, + "url": url + } + data = json.dumps(json_data).encode('utf-8') - req = urllib.request.Request(url, data = data, headers=headers) + req = urllib.request.Request(api, data=data, headers=headers) response = opener.open(req, timeout=10) json_str = response.read().decode("utf-8") json_data = json.loads(json_str) @@ -108,6 +120,7 @@ def bark(url: str, content: str) -> Dict[str, Any]: if __name__ == '__main__': + send_title = '直播通知' # 标题 send_content = '张三 开播了!' # 推送内容 # 钉钉推送通知 @@ -118,7 +131,7 @@ if __name__ == '__main__': # 微信推送通知 # 替换成自己的单点推送接口,获取地址:https://xz.qqoq.net/#/admin/one # 当然也可以使用其他平台API 如server酱 使用方法一样 - xizhi_api = 'https://xizhi.qqoq.net/XZa5a4a224987c88ab828acbcc0aa4c853.send' + xizhi_api = 'https://xizhi.qqoq.net/xxxxxxxxx.send' # xizhi(xizhi_api, send_content) # telegram推送通知 @@ -128,5 +141,5 @@ if __name__ == '__main__': # email_message("", "", "", "", "", "") - bark_url= 'https://xxx.xxx.com/key/' - # bark(bark_url, send_content) + bark_url = 'https://xxx.xxx.com/key/' + # bark(bark_url, send_title, send_content)