chickliu/quark-auto-save
1
0
Fork 0
mirror of https://github.com/Cp0204/quark-auto-save.git synced 2026-01-15 17:00:43 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

跟进 qinglong 通知模块更新

Browse Source
This commit is contained in:
Cp0204 2024-04-14 00:14:21 +08:00
parent 56a071be76
commit da35ebfa33
1 changed files with 15 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
sendNotify.py
View File

@ -213,7 +213,7 @@ def feishu_bot(title: str, content: str) -> None:
data = {"msg_type": "text", "content": {"text": f"{title}\n\n{content}"}} data = {"msg_type": "text", "content": {"text": f"{title}\n\n{content}"}}
response = requests.post(url, data=json.dumps(data)).json() response = requests.post(url, data=json.dumps(data)).json()
if response.get("StatusCode") == 0: if response.get("StatusCode") == 0 or response.get("code") == 0:
print("飞书 推送成功！") print("飞书 推送成功！")
else: else:
print("飞书 推送失败！错误信息如下：\n", response) print("飞书 推送失败！错误信息如下：\n", response)
@ -749,13 +749,14 @@ def parse_headers(headers):
return parsed return parsed
def parse_string(input_string): def parse_string(input_string, value_format_fn=None):
matches = {} matches = {}
pattern = r"(\w+):\s*((?:(?!\n\w+:).)*)" pattern = r"(\w+):\s*((?:(?!\n\w+:).)*)"
regex = re.compile(pattern) regex = re.compile(pattern)
for match in regex.finditer(input_string): for match in regex.finditer(input_string):
key, value = match.group(1).strip(), match.group(2).strip() key, value = match.group(1).strip(), match.group(2).strip()
try: try:
value = value_format_fn(value) if value_format_fn else value
json_value = json.loads(value) json_value = json.loads(value)
matches[key] = json_value matches[key] = json_value
except: except:
@ -763,11 +764,11 @@ def parse_string(input_string):
return matches return matches
def parse_body(body, content_type): def parse_body(body, content_type, value_format_fn=None):
if not body or content_type == "text/plain": if not body or content_type == "text/plain":
return body return body
parsed = parse_string(body) parsed = parse_string(body, value_format_fn)
if content_type == "application/x-www-form-urlencoded": if content_type == "application/x-www-form-urlencoded":
data = urllib.parse.urlencode(parsed, doseq=True) data = urllib.parse.urlencode(parsed, doseq=True)
@ -780,18 +781,6 @@ def parse_body(body, content_type):
return parsed return parsed
def format_notify_content(url, body, title, content):
if "$title" not in url and "$title" not in body:
return {}
formatted_url = url.replace("$title", urllib.parse.quote_plus(title)).replace(
"$content", urllib.parse.quote_plus(content)
)
formatted_body = body.replace("$title", title).replace("$content", content)
return formatted_url, formatted_body
def custom_notify(title: str, content: str) -> None: def custom_notify(title: str, content: str) -> None:
""" """
通过 自定义通知 推送消息。 通过 自定义通知 推送消息。
@ -808,18 +797,21 @@ def custom_notify(title: str, content: str) -> None:
WEBHOOK_BODY = push_config.get("WEBHOOK_BODY") WEBHOOK_BODY = push_config.get("WEBHOOK_BODY")
WEBHOOK_HEADERS = push_config.get("WEBHOOK_HEADERS") WEBHOOK_HEADERS = push_config.get("WEBHOOK_HEADERS")
formatUrl, formatBody = format_notify_content( if "$title" not in WEBHOOK_URL and "$title" not in WEBHOOK_BODY:
WEBHOOK_URL, WEBHOOK_BODY, title, content
)
if not formatUrl and not formatBody:
print("请求头或者请求体中必须包含 $title 和 $content") print("请求头或者请求体中必须包含 $title 和 $content")
return return
headers = parse_headers(WEBHOOK_HEADERS) headers = parse_headers(WEBHOOK_HEADERS)
body = parse_body(formatBody, WEBHOOK_CONTENT_TYPE) body = parse_body(
WEBHOOK_BODY,
WEBHOOK_CONTENT_TYPE,
lambda v: v.replace("$title", title).replace("$content", content),
)
formatted_url = WEBHOOK_URL.replace(
"$title", urllib.parse.quote_plus(title)
).replace("$content", urllib.parse.quote_plus(content))
response = requests.request( response = requests.request(
method=WEBHOOK_METHOD, url=formatUrl, headers=headers, timeout=15, data=body method=WEBHOOK_METHOD, url=formatted_url, headers=headers, timeout=15, data=body
) )
if response.status_code == 200: if response.status_code == 200: