open-source/qinglong
1
0
Fork 0
mirror of https://github.com/whyour/qinglong.git synced 2025-07-08 20:52:29 +08:00
Code Issues Packages Projects Releases Wiki Activity

telegram自定义参数

Browse Source
This commit is contained in:
Akimio521 2024-02-25 15:33:04 +08:00
parent 1b5617baa0
commit 0ac09f25fa
1 changed files with 28 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
sample/notify.py
View File

@ -571,39 +571,52 @@ def wecom_bot(title: str, content: str, **kwargs) -> None:
print("企业微信机器人推送失败！") print("企业微信机器人推送失败！")
def telegram_bot(title: str, content: str) -> None: def telegram_bot(title: str, content: str, **kwargs) -> None:
""" """
使用 telegram 机器人 推送消息。 使用 telegram 机器人 推送消息。
""" """
if not push_config.get("TG_BOT_TOKEN") or not push_config.get("TG_USER_ID"): 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 服务的 bot_token 或者 user_id 未设置!!\n取消推送") print("tg 服务的 TG_BOT_TOKEN 或者 TG_USER_ID 未设置!!\n取消推送")
return return
print("tg 服务启动") 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"): if kwargs.get("TG_API_HOST") or push_config.get("TG_API_HOST"):
url = f"{push_config.get('TG_API_HOST')}/bot{push_config.get('TG_BOT_TOKEN')}/sendMessage" 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: else:
url = ( 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"} headers = {"Content-Type": "application/x-www-form-urlencoded"}
payload = { payload = {
"chat_id": str(push_config.get("TG_USER_ID")), "chat_id": str(TG_USER_ID),
"text": f"{title}\n\n{content}", "text": f"{title}\n\n{content}",
"disable_web_page_preview": "true", "disable_web_page_preview": "true",
} }
proxies = None proxies = None
if push_config.get("TG_PROXY_HOST") and push_config.get("TG_PROXY_PORT"): 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 push_config.get("TG_PROXY_AUTH") is not None and "@" not in push_config.get( if kwargs.get("TG_PROXY_HOST") and kwargs.get("TG_PROXY_PORT"):
"TG_PROXY_HOST" TG_PROXY_HOST = kwargs.get("TG_PROXY_HOST")
): TG_PROXY_PORT = kwargs.get("TG_PROXY_PORT")
push_config["TG_PROXY_HOST"] = ( else:
push_config.get("TG_PROXY_AUTH") 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( 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} proxies = {"http": proxyStr, "https": proxyStr}
response = requests.post( response = requests.post(