open-source/qinglong
1
0
Fork 0
mirror of https://github.com/whyour/qinglong.git synced 2025-05-24 15:43:24 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1 from Cp0204/develop

Browse Source 
改进自定义通知方法，修正自定义 body 错误
This commit is contained in:
秋澪Akimio 2024-04-09 11:25:59 +08:00 committed by GitHub
commit ec87bbadfd
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 11 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
sample/notify.py
View File

@ -113,14 +113,10 @@ push_config = {
'WEBHOOK_METHOD': '', # 自定义通知 请求方法 'WEBHOOK_METHOD': '', # 自定义通知 请求方法
'WEBHOOK_CONTENT_TYPE': '' # 自定义通知 content-type 'WEBHOOK_CONTENT_TYPE': '' # 自定义通知 content-type
} }
push_config_keys = tuple(push_config.keys()) # 保存push_config的键
notify_function = []
# fmt: on # fmt: on
# 首先读取 面板变量 或者 github action 运行变量 # 首先读取 面板变量 或者 github action 运行变量
for k in push_config_keys: for k in push_config:
if os.getenv(k): if os.getenv(k):
v = os.getenv(k) v = os.getenv(k)
push_config[k] = v push_config[k] = v
@ -771,10 +767,10 @@ def parse_body(body, content_type):
if not body or content_type == "text/plain": if not body or content_type == "text/plain":
return body return body
parsed = parse_string(input_string) parsed = parse_string(body)
if content_type == "application/x-www-form-urlencoded": if content_type == "application/x-www-form-urlencoded":
data = urlencode(parsed, doseq=True) data = urllib.parse.urlencode(parsed, doseq=True)
return data return data
if content_type == "application/json": if content_type == "application/json":
@ -843,6 +839,7 @@ def one() -> str:
def add_notify_function(): def add_notify_function():
notify_function = []
if push_config.get("BARK_PUSH"): if push_config.get("BARK_PUSH"):
notify_function.append(bark) notify_function.append(bark)
if push_config.get("CONSOLE"): if push_config.get("CONSOLE"):
@ -900,17 +897,16 @@ def add_notify_function():
if not notify_function: if not notify_function:
print(f"无推送渠道，请检查通知变量是否正确") print(f"无推送渠道，请检查通知变量是否正确")
return notify_function
def send(title: str, content: str, ignore_default_config: bool = False, **kwargs): def send(title: str, content: str, ignore_default_config: bool = False, **kwargs):
if ignore_default_config and kwargs:
global push_config
push_config = {} # 清空从环境变量获取的配置
if kwargs: if kwargs:
for k in kwargs: global push_config
if k in push_config_keys: if ignore_default_config:
push_config[k] = kwargs.get(k) push_config = kwargs # 清空从环境变量获取的配置
else:
push_config.update(kwargs)
if not content: if not content:
print(f"{title} 推送内容为空！") print(f"{title} 推送内容为空！")
@ -926,7 +922,7 @@ def send(title: str, content: str, ignore_default_config: bool = False, **kwargs
hitokoto = push_config.get("HITOKOTO") hitokoto = push_config.get("HITOKOTO")
content += "\n\n" + one() if hitokoto else "" content += "\n\n" + one() if hitokoto else ""
add_notify_function() notify_function = add_notify_function()
ts = [ ts = [
threading.Thread(target=mode, args=(title, content), name=mode.__name__) threading.Thread(target=mode, args=(title, content), name=mode.__name__)
for mode in notify_function for mode in notify_function