From 0ac09f25face6426ae45b382eb2d0b763036c247 Mon Sep 17 00:00:00 2001 From: Akimio521 Date: Sun, 25 Feb 2024 15:33:04 +0800 Subject: [PATCH] =?UTF-8?q?telegram=E8=87=AA=E5=AE=9A=E4=B9=89=E5=8F=82?= =?UTF-8?q?=E6=95=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- sample/notify.py | 43 ++++++++++++++++++++++++++++--------------- 1 file changed, 28 insertions(+), 15 deletions(-) diff --git a/sample/notify.py b/sample/notify.py index 98bedb73..2a6dd453 100644 --- a/sample/notify.py +++ b/sample/notify.py @@ -571,39 +571,52 @@ def wecom_bot(title: str, content: str, **kwargs) -> None: print("企业微信机器人推送失败！") -def telegram_bot(title: str, content: str) -> None: +def telegram_bot(title: str, content: str, **kwargs) -> None: """ 使用 telegram 机器人 推送消息。 """ - if not push_config.get("TG_BOT_TOKEN") or not push_config.get("TG_USER_ID"): - print("tg 服务的 bot_token 或者 user_id 未设置!!\n取消推送") + if not ((kwargs.get("TG_BOT_TOKEN") and kwargs.get("TG_USER_ID")) or (push_config.get("TG_BOT_TOKEN") and push_config.get("TG_USER_ID"))): + print("tg 服务的 TG_BOT_TOKEN 或者 TG_USER_ID 未设置!!\n取消推送") return print("tg 服务启动") + if kwargs.get("TG_BOT_TOKEN") and kwargs.get("TG_USER_ID"): + TG_BOT_TOKEN = kwargs.get("TG_BOT_TOKEN") + TG_USER_ID = kwargs.get("TG_USER_ID") + else: + TG_BOT_TOKEN = push_config.get("TG_BOT_TOKEN") + TG_USER_ID = push_config.get("TG_USER_ID") - if push_config.get("TG_API_HOST"): - url = f"{push_config.get('TG_API_HOST')}/bot{push_config.get('TG_BOT_TOKEN')}/sendMessage" + if kwargs.get("TG_API_HOST") or push_config.get("TG_API_HOST"): + TG_API_HOST = kwargs.get("TG_API_HOST", push_config.get("TG_API_HOST")) + url = f"{TG_API_HOST}/bot{TG_BOT_TOKEN}/sendMessage" else: url = ( - f"https://api.telegram.org/bot{push_config.get('TG_BOT_TOKEN')}/sendMessage" + f"https://api.telegram.org/bot{TG_BOT_TOKEN}/sendMessage" ) headers = {"Content-Type": "application/x-www-form-urlencoded"} payload = { - "chat_id": str(push_config.get("TG_USER_ID")), + "chat_id": str(TG_USER_ID), "text": f"{title}\n\n{content}", "disable_web_page_preview": "true", } proxies = None - if push_config.get("TG_PROXY_HOST") and push_config.get("TG_PROXY_PORT"): - if push_config.get("TG_PROXY_AUTH") is not None and "@" not in push_config.get( - "TG_PROXY_HOST" - ): - push_config["TG_PROXY_HOST"] = ( - push_config.get("TG_PROXY_AUTH") + if not ((kwargs.get("TG_PROXY_HOST") and kwargs.get("TG_PROXY_PORT")) or (push_config.get("TG_PROXY_HOST") and push_config.get("TG_PROXY_PORT"))): + if kwargs.get("TG_PROXY_HOST") and kwargs.get("TG_PROXY_PORT"): + TG_PROXY_HOST = kwargs.get("TG_PROXY_HOST") + TG_PROXY_PORT = kwargs.get("TG_PROXY_PORT") + else: + TG_PROXY_HOST = kwargs.get("TG_PROXY_HOST") + TG_PROXY_PORT = kwargs.get("TG_PROXY_PORT") + if kwargs.get("TG_PROXY_AUTH") or push_config.get("TG_PROXY_AUTH"): + TG_PROXY_AUTH = kwargs.get("TG_PROXY_AUTH", push_config.get("TG_PROXY_AUTH")) + if TG_PROXY_AUTH is not None and "@" not in TG_PROXY_HOST: + TG_PROXY_HOST = ( + TG_PROXY_AUTH + "@" - + push_config.get("TG_PROXY_HOST") + + TG_PROXY_HOST ) proxyStr = "http://{}:{}".format( - push_config.get("TG_PROXY_HOST"), push_config.get("TG_PROXY_PORT") + TG_PROXY_HOST, TG_PROXY_PORT ) proxies = {"http": proxyStr, "https": proxyStr} response = requests.post(