From 13e0ff55d4ddca94a1418cfb8d677b3e6bdd243d Mon Sep 17 00:00:00 2001 From: Henglie <90338155+Henglie@users.noreply.github.com> Date: Mon, 9 Nov 2026 01:11:32 +0800 Subject: [PATCH] fix(notify): bound notification requests and thread waits (#3129) Add a 10s timeout to every requests call in sample/notify.py so a non-responsive push service cannot block a task indefinitely. Mark notification threads as daemon and join them with a bounded timeout, recording channels that do not finish as failures. Add a regression test covering missing timeouts and send() returning after a hung channel. --- sample/notify.py | 108 +++++++++++++++++++++-------- test/back/notify-timeouts.test.cjs | 69 ++++++++++++++++++ 2 files changed, 147 insertions(+), 30 deletions(-) create mode 100644 test/back/notify-timeouts.test.cjs diff --git a/sample/notify.py b/sample/notify.py index c969df2b..16f314fb 100644 --- a/sample/notify.py +++ b/sample/notify.py @@ -18,6 +18,11 @@ from email.utils import formataddr import requests +# 单次 HTTP 请求的默认超时（秒），避免推送服务无响应时永久阻塞 +REQUEST_TIMEOUT = 10 +# 等待推送线程结束的最长时间（秒），超时后跳过并继续退出 +NOTIFY_JOIN_TIMEOUT = 30 + # 原先的 print 函数和主线程的锁 _print = print mutex = threading.Lock() @@ -199,7 +204,7 @@ def bark(title: str, content: str) -> Optional[bool]: data[bark_params.get(pair[0])] = pair[1] headers = {"Content-Type": "application/json;charset=utf-8"} response = requests.post( - url=url, data=json.dumps(data), headers=headers, timeout=15 + url=url, data=json.dumps(data), headers=headers, timeout=REQUEST_TIMEOUT ).json() if response["code"] == 200: @@ -241,7 +246,7 @@ def dingding_bot(title: str, content: str) -> Optional[bool]: 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 + url=url, data=json.dumps(data), headers=headers, timeout=REQUEST_TIMEOUT ).json() if not response["errcode"]: @@ -278,7 +283,9 @@ def feishu_bot(title: str, content: str) -> Optional[bool]: data["timestamp"] = timestamp data["sign"] = sign - response = requests.post(url, data=json.dumps(data)).json() + response = requests.post( + url, data=json.dumps(data), timeout=REQUEST_TIMEOUT + ).json() if response.get("StatusCode") == 0 or response.get("code") == 0: audit_results.append(True) @@ -299,7 +306,7 @@ def go_cqhttp(title: str, content: str) -> Optional[bool]: 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() + response = requests.get(url, timeout=REQUEST_TIMEOUT).json() if response["status"] == "ok": audit_results.append(True) @@ -325,7 +332,7 @@ def gotify(title: str, content: str) -> Optional[bool]: "message": content, "priority": push_config.get("GOTIFY_PRIORITY"), } - response = requests.post(url, data=data).json() + response = requests.post(url, data=data, timeout=REQUEST_TIMEOUT).json() if response.get("id"): audit_results.append(True) @@ -348,7 +355,9 @@ def iGot(title: str, content: str) -> Optional[bool]: 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() + response = requests.post( + url, data=data, headers=headers, timeout=REQUEST_TIMEOUT + ).json() if response["ret"] == 0: audit_results.append(True) @@ -377,7 +386,7 @@ def serverJ(title: str, content: str) -> Optional[bool]: else: url = f'https://sctapi.ftqq.com/{push_config.get("PUSH_KEY")}.send' - response = requests.post(url, data=data).json() + response = requests.post(url, data=data, timeout=REQUEST_TIMEOUT).json() if response.get("errno") == 0 or response.get("code") == 0: audit_results.append(True) @@ -406,7 +415,7 @@ def pushdeer(title: str, content: str) -> Optional[bool]: if push_config.get("DEER_URL"): url = push_config.get("DEER_URL") - response = requests.post(url, data=data).json() + response = requests.post(url, data=data, timeout=REQUEST_TIMEOUT).json() if len(response.get("content").get("result")) > 0: audit_results.append(True) @@ -427,7 +436,7 @@ def chat(title: str, content: str) -> Optional[bool]: print("chat 服务启动") data = "payload=" + json.dumps({"text": title + "\n" + content}) url = push_config.get("CHAT_URL") + push_config.get("CHAT_TOKEN") - response = requests.post(url, data=data) + response = requests.post(url, data=data, timeout=REQUEST_TIMEOUT) if response.status_code == 200: audit_results.append(True) @@ -461,7 +470,9 @@ def pushplus_bot(title: str, content: str) -> Optional[bool]: } body = json.dumps(data).encode(encoding="utf-8") headers = {"Content-Type": "application/json"} - response = requests.post(url=url, data=body, headers=headers).json() + response = requests.post( + url=url, data=body, headers=headers, timeout=REQUEST_TIMEOUT + ).json() code = response["code"] if code == 200: @@ -477,7 +488,9 @@ def pushplus_bot(title: str, content: str) -> Optional[bool]: else: url_old = "http://pushplus.hxtrip.com/send" headers["Accept"] = "application/json" - response = requests.post(url=url_old, data=body, headers=headers).json() + response = requests.post( + url=url_old, data=body, headers=headers, timeout=REQUEST_TIMEOUT + ).json() if response["code"] == 200: audit_results.append(True) @@ -513,7 +526,9 @@ def weplus_bot(title: str, content: str) -> Optional[bool]: } body = json.dumps(data).encode(encoding="utf-8") headers = {"Content-Type": "application/json"} - response = requests.post(url=url, data=body, headers=headers).json() + response = requests.post( + url=url, data=body, headers=headers, timeout=REQUEST_TIMEOUT + ).json() if response["code"] == 200: audit_results.append(True) @@ -535,7 +550,9 @@ def qmsg_bot(title: str, content: str) -> Optional[bool]: 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() + response = requests.post( + url=url, params=payload, timeout=REQUEST_TIMEOUT + ).json() if response["code"] == 0: audit_results.append(True) @@ -599,7 +616,7 @@ class WeCom: "corpid": self.CORPID, "corpsecret": self.CORPSECRET, } - req = requests.post(url, params=values) + req = requests.post(url, params=values, timeout=REQUEST_TIMEOUT) data = json.loads(req.text) return data["access_token"] @@ -615,7 +632,9 @@ class WeCom: "safe": "0", } send_msges = bytes(json.dumps(send_values), "utf-8") - respone = requests.post(send_url, send_msges) + respone = requests.post( + send_url, send_msges, timeout=REQUEST_TIMEOUT + ) respone = respone.json() return respone["errmsg"] @@ -641,7 +660,9 @@ class WeCom: }, } send_msges = bytes(json.dumps(send_values), "utf-8") - respone = requests.post(send_url, send_msges) + respone = requests.post( + send_url, send_msges, timeout=REQUEST_TIMEOUT + ) respone = respone.json() return respone["errmsg"] @@ -663,7 +684,7 @@ def wecom_bot(title: str, content: str) -> Optional[bool]: 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 + url=url, data=json.dumps(data), headers=headers, timeout=REQUEST_TIMEOUT ).json() if response["errcode"] == 0: @@ -711,7 +732,11 @@ def telegram_bot(title: str, content: str) -> Optional[bool]: ) proxies = {"http": proxyStr, "https": proxyStr} response = requests.post( - url=url, headers=headers, params=payload, proxies=proxies + url=url, + headers=headers, + params=payload, + proxies=proxies, + timeout=REQUEST_TIMEOUT, ).json() if response["ok"]: @@ -752,7 +777,9 @@ def aibotk(title: str, content: str) -> Optional[bool]: } body = json.dumps(data).encode(encoding="utf-8") headers = {"Content-Type": "application/json"} - response = requests.post(url=url, data=body, headers=headers).json() + response = requests.post( + url=url, data=body, headers=headers, timeout=REQUEST_TIMEOUT + ).json() print(response) if response["code"] == 0: audit_results.append(True) @@ -836,7 +863,7 @@ def pushme(title: str, content: str) -> Optional[bool]: "date": push_config.get("date") if push_config.get("date") else "", "type": push_config.get("type") if push_config.get("type") else "", } - response = requests.post(url, data=data) + response = requests.post(url, data=data, timeout=REQUEST_TIMEOUT) if response.status_code == 200 and response.text == "success": audit_results.append(True) @@ -883,7 +910,9 @@ def chronocat(title: str, content: str) -> Optional[bool]: } ], } - response = requests.post(url, headers=headers, data=json.dumps(data)) + response = requests.post( + url, headers=headers, data=json.dumps(data), timeout=REQUEST_TIMEOUT + ) if response.status_code == 200: if chat_type == 1: audit_results.append(True) @@ -936,7 +965,9 @@ def ntfy(title: str, content: str) -> Optional[bool]: headers['Actions'] = encode_rfc2047(push_config.get("NTFY_ACTIONS")) url = push_config.get("NTFY_URL") + "/" + push_config.get("NTFY_TOPIC") - response = requests.post(url, data=data, headers=headers) + response = requests.post( + url, data=data, headers=headers, timeout=REQUEST_TIMEOUT + ) if response.status_code == 200: # 使用 response.status_code 进行检查 audit_results.append(True) print("Ntfy 推送成功！") @@ -995,7 +1026,9 @@ def wxpusher_bot(title: str, content: str) -> Optional[bool]: } headers = {"Content-Type": "application/json"} - response = requests.post(url=url, json=data, headers=headers).json() + response = requests.post( + url=url, json=data, headers=headers, timeout=REQUEST_TIMEOUT + ).json() if response.get("code") == 1000: audit_results.append(True) @@ -1046,7 +1079,9 @@ def wxpusher_spt(title: str, content: str) -> Optional[bool]: data["sptList"] = spts headers = {"Content-Type": "application/json"} - response = requests.post(url=url, json=data, headers=headers).json() + response = requests.post( + url=url, json=data, headers=headers, timeout=REQUEST_TIMEOUT + ).json() if response.get("code") == 1000: audit_results.append(True) @@ -1080,7 +1115,9 @@ def wpush(title: str, content: str) -> Optional[bool]: data["topic_code"] = push_config.get("WPUSH_TOPIC_CODE") headers = {"Content-Type": "application/json"} - response = requests.post(url=url, json=data, headers=headers, timeout=15).json() + response = requests.post( + url=url, json=data, headers=headers, timeout=REQUEST_TIMEOUT + ).json() if response.get("code") == 0: audit_results.append(True) @@ -1121,7 +1158,9 @@ def openilink(title: str, content: str) -> Optional[bool]: if push_config.get("OPENILINK_CONTEXT_TOKEN"): data["context_token"] = push_config.get("OPENILINK_CONTEXT_TOKEN") - response = requests.post(url=url, json=data, headers=headers).json() + response = requests.post( + url=url, json=data, headers=headers, timeout=REQUEST_TIMEOUT + ).json() if response.get("ok"): audit_results.append(True) @@ -1215,7 +1254,7 @@ def custom_notify(title: str, content: str) -> Optional[bool]: "$title", urllib.parse.quote_plus(title) ).replace("$content", urllib.parse.quote_plus(content)) response = requests.request( - method=WEBHOOK_METHOD, url=formatted_url, headers=headers, timeout=15, data=body + method=WEBHOOK_METHOD, url=formatted_url, headers=headers, timeout=REQUEST_TIMEOUT, data=body ) if response.status_code == 200: @@ -1233,7 +1272,7 @@ def one() -> str: :return: """ url = "https://v1.hitokoto.cn/" - res = requests.get(url).json() + res = requests.get(url, timeout=REQUEST_TIMEOUT).json() return res["hitokoto"] + " ----" + res["from"] @@ -1374,11 +1413,20 @@ def send(title: str, content: str, ignore_default_config: bool = False, **kwargs notify_function = add_notify_function() results = [] ts = [ - threading.Thread(target=audited_notification, args=(mode, title, content, results), name=mode.__name__) + threading.Thread( + target=audited_notification, + args=(mode, title, content, results), + name=mode.__name__, + daemon=True, + ) for mode in notify_function ] [t.start() for t in ts] - [t.join() for t in ts] + for t in ts: + t.join(timeout=NOTIFY_JOIN_TIMEOUT) + if t.is_alive(): + results.append((t.name, False)) + print(f"{t.name} 推送超时（>{NOTIFY_JOIN_TIMEOUT}s），已跳过，任务继续结束") report_notifications(results, title, content) diff --git a/test/back/notify-timeouts.test.cjs b/test/back/notify-timeouts.test.cjs new file mode 100644 index 00000000..b74bb2fc --- /dev/null +++ b/test/back/notify-timeouts.test.cjs @@ -0,0 +1,69 @@ +const test = require('node:test'), + assert = require('node:assert/strict'), + fs = require('node:fs'), + path = require('node:path'), + { spawnSync } = require('node:child_process'); + +const notifyFile = path.resolve( + process.env.NOTIFY_PY_PATH || 'sample/notify.py', +); +const source = fs.readFileSync(notifyFile, 'utf8'); + +function httpCalls(text) { + const calls = []; + const pattern = /requests\.(?:get|post|request)\s*\(/g; + let match; + while ((match = pattern.exec(text))) { + let depth = 0; + let end = match.index + match[0].length - 1; + for (; end < text.length; end++) { + if (text[end] === '(') depth++; + else if (text[end] === ')' && --depth === 0) break; + } + calls.push(text.slice(match.index, end + 1)); + } + return calls; +} + +test('sample/notify.py sends every HTTP request with a timeout', () => { + const calls = httpCalls(source); + assert.ok( + calls.length >= 27, + `expected at least 27 HTTP calls, got ${calls.length}`, + ); + const missing = calls.filter((call) => !/\btimeout\s*=/.test(call)); + assert.deepEqual(missing, []); +}); + +test('notify.send returns after a bounded wait when a channel hangs', () => { + const script = ` +import os, sys, time, types +requests = types.ModuleType("requests") +requests.get = requests.post = requests.request = lambda *a, **k: None +sys.modules["requests"] = requests +sys.path.insert(0, ${JSON.stringify(path.dirname(notifyFile))}) +os.environ.pop("QL_DIR", None) +import notify +notify.NOTIFY_JOIN_TIMEOUT = 1 +notify.push_config = {key: "" for key in notify.push_config} +notify.push_config["HITOKOTO"] = False +notify.push_config["CONSOLE"] = True +def hang(*args, **kwargs): + time.sleep(60) +notify.console = hang +start = time.monotonic() +notify.send("title", "content") +elapsed = time.monotonic() - start +assert elapsed < 5, elapsed +print("elapsed=%.2f" % elapsed) +`; + const python = process.platform === 'win32' ? 'python' : 'python3'; + const result = spawnSync(python, ['-'], { + input: script, + encoding: 'utf8', + timeout: 15000, + env: { ...process.env, QL_DIR: '' }, + }); + assert.equal(result.status, 0, result.stdout + result.stderr); + assert.match(result.stdout, /elapsed=/); +});