Passed
Branch master (6b64f1)
by Leon
02:15
created

notify_mtr_env.telegram_bot()   F

Complexity

Conditions 14

Size

Total Lines 56
Code Lines 43

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
cc 14
eloc 43
nop 2
dl 0
loc 56
rs 3.6
c 0
b 0
f 0

How to fix   Long Method    Complexity   

Long Method

Small methods make your code easier to understand, in particular if combined with a good name. Besides, if your method is small, finding a good name is usually much easier.

For example, if you find yourself adding comments to a method's body, this is usually a good sign to extract the commented part to a new method, and use the comment as a starting point when coming up with a good name for this new method.

Commonly applied refactorings include:

Complexity

Complex classes like notify_mtr_env.telegram_bot() often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes.

Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.

1
#!/usr/bin/env python3
2
# _*_ coding:utf-8 _*_
3
import base64
4
import hashlib
5
import hmac
6
import json
7
import os
8
import re
9
import threading
10
import time
11
import traceback
12
import urllib.parse
13
14
import requests
15
16
# 原先的 print 函数和主线程的锁
17
_print = print
18
mutex = threading.Lock()
19
20
21
# 定义新的 print 函数
22
def print(text, *args, **kw):
23
    """
24
    使输出有序进行,不出现多线程同一时间输出导致错乱的问题。
25
    """
26
    with mutex:
27
        _print(text, *args, **kw)
28
29
30
# 通知服务
31
# fmt: off
32
push_config = {
33
    'HITOKOTO': False,                  # 启用一言(随机句子)
34
35
    'BARK_PUSH': '',                    # bark IP 或设备码,例:https://api.day.app/DxHcxxxxxRxxxxxxcm
36
    'BARK_ARCHIVE': '',                 # bark 推送是否存档
37
    'BARK_GROUP': '',                   # bark 推送分组
38
    'BARK_SOUND': '',                   # bark 推送声音
39
40
    'CONSOLE': True,                    # 控制台输出
41
42
    'DD_BOT_SECRET': '',                # 钉钉机器人的 DD_BOT_SECRET
43
    'DD_BOT_TOKEN': '',                 # 钉钉机器人的 DD_BOT_TOKEN
44
45
    'FSKEY': '',                        # 飞书机器人的 FSKEY
46
47
    'GOBOT_URL': '',                    # go-cqhttp
48
                                        # 推送到个人QQ:http://127.0.0.1/send_private_msg
49
                                        # 群:http://127.0.0.1/send_group_msg
50
    'GOBOT_QQ': '',                     # go-cqhttp 的推送群或用户
51
                                        # GOBOT_URL 设置 /send_private_msg 时填入 user_id=个人QQ
52
                                        #               /send_group_msg   时填入 group_id=QQ群
53
    'GOBOT_TOKEN': '',                  # go-cqhttp 的 access_token
54
55
    'IGOT_PUSH_KEY': '',                # iGot 聚合推送的 IGOT_PUSH_KEY
56
57
    'PUSH_KEY': '',                     # server 酱的 PUSH_KEY,兼容旧版与 Turbo 版
58
59
    'PUSH_PLUS_TOKEN': '',              # push+ 微信推送的用户令牌
60
    'PUSH_PLUS_USER': '',               # push+ 微信推送的群组编码
61
62
    'QMSG_KEY': '',                     # qmsg 酱的 QMSG_KEY
63
    'QMSG_TYPE': '',                    # qmsg 酱的 QMSG_TYPE
64
65
    'QYWX_AM': '',                      # 企业微信应用
66
67
    'QYWX_KEY': '',                     # 企业微信机器人
68
69
    'TG_BOT_TOKEN': '',                 # tg 机器人的 TG_BOT_TOKEN,例:1407203283:AAG9rt-6RDaaX0HBLZQq0laNOh898iFYaRQ
70
    'TG_USER_ID': '',                   # tg 机器人的 TG_USER_ID,例:1434078534
71
    'TG_API_HOST': '',                  # tg 代理 api
72
    'TG_PROXY_AUTH': '',                # tg 代理认证参数
73
    'TG_PROXY_HOST': '',                # tg 机器人的 TG_PROXY_HOST
74
    'TG_PROXY_PORT': '',                # tg 机器人的 TG_PROXY_PORT
75
}
76
notify_function = []
77
# fmt: on
78
79
# 首先读取 面板变量 或者 github action 运行变量
80
for k in push_config:
81
    if os.getenv(k):
82
        v = os.getenv(k)
83
        push_config[k] = v
84
85
86
def bark(title: str, content: str) -> None:
87
    """
88
    使用 bark 推送消息。
89
    """
90
    if not push_config.get("BARK_PUSH"):
91
        print("bark 服务的 BARK_PUSH 未设置!!\n取消推送")
92
        return
93
    print("bark 服务启动")
94
95
    if push_config.get("BARK_PUSH").startswith("http"):
96
        url = f'{push_config.get("BARK_PUSH").rstrip("/")}/{urllib.parse.quote_plus(title)}/{urllib.parse.quote_plus(content)}'
97
    else:
98
        url = f'https://api.day.app/{push_config.get("BARK_PUSH")}/{urllib.parse.quote_plus(title)}/{urllib.parse.quote_plus(content)}'
99
100
    bark_params = {
101
        "BARK_ARCHIVE": "isArchive",
102
        "BARK_GROUP": "group",
103
        "BARK_SOUND": "sound",
104
    }
105
    params = ""
106
    for pair in filter(
107
        lambda pairs: pairs[0].startswith("BARK_")
108
        and pairs[0] != "BARK_PUSH"
109
        and pairs[1]
110
        and bark_params.get(pairs[0]),
111
        push_config.items(),
112
    ):
113
        params += f"{bark_params.get(pair[0])}={pair[1]}&"
114
    if params:
115
        url = url + "?" + params.rstrip("&")
116
117
    try:
118
        response = requests.get(url, timeout=15)
119
        try:
120
            datas = response.json()
121
            if datas.get("code") == 200:
122
                print("bark 推送成功!")
123
            elif datas.get("code") == 400:
124
                print("bark 推送失败!找不到 Key 对应的 DeviceToken。")
125
            else:
126
                print(f"bark 推送失败!响应数据:{datas}")
127
        except json.JSONDecodeError:
128
            print(f"推送返回值非 json 格式,请检查网址和账号是否填写正确。\n{response.text}")
129
    except requests.exceptions.RequestException:
130
        print(f"网络异常,请检查你的网络连接、推送服务器和代理配置。\n{traceback.format_exc()}")
131
    except Exception:
132
        print(f"其他错误信息:\n{traceback.format_exc()}")
133
134
135
def console(title: str, content: str) -> None:
136
    """
137
    使用 控制台 推送消息。
138
    """
139
    print(f"{title}\n\n{content}")
140
141
142
def dingding_bot(title: str, content: str) -> None:
143
    """
144
    使用 钉钉机器人 推送消息。
145
    """
146
    if not push_config.get("DD_BOT_SECRET") or not push_config.get("DD_BOT_TOKEN"):
147
        print("钉钉机器人 服务的 DD_BOT_SECRET 或者 DD_BOT_TOKEN 未设置!!\n取消推送")
148
        return
149
    print("钉钉机器人 服务启动")
150
151
    timestamp = str(round(time.time() * 1000))
152
    secret_enc = push_config.get("DD_BOT_SECRET").encode("utf-8")
153
    string_to_sign = "{}\n{}".format(timestamp, push_config.get("DD_BOT_SECRET"))
154
    string_to_sign_enc = string_to_sign.encode("utf-8")
155
    hmac_code = hmac.new(
156
        secret_enc, string_to_sign_enc, digestmod=hashlib.sha256
157
    ).digest()
158
    sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
159
    url = f'https://oapi.dingtalk.com/robot/send?access_token={push_config.get("DD_BOT_TOKEN")}&timestamp={timestamp}&sign={sign}'
160
    headers = {"Content-Type": "application/json;charset=utf-8"}
161
    data = {"msgtype": "text", "text": {"content": f"{title}\n\n{content}"}}
162
163
    try:
164
        response = requests.post(
165
            url=url, data=json.dumps(data), headers=headers, timeout=15
166
        )
167
        try:
168
            datas = response.json()
169
            if datas.get("errcode") == 0:
170
                print("钉钉机器人 推送成功!")
171
            else:
172
                print(f"钉钉机器人 推送失败!响应数据:{datas}")
173
        except json.JSONDecodeError:
174
            print(f"推送返回值非 json 格式,请检查网址和账号是否填写正确。\n{response.text}")
175
    except requests.exceptions.RequestException:
176
        print(f"网络异常,请检查你的网络连接、推送服务器和代理配置。\n{traceback.format_exc()}")
177
    except Exception:
178
        print(f"其他错误信息:\n{traceback.format_exc()}")
179
180
181
def feishu_bot(title: str, content: str) -> None:
182
    """
183
    使用 飞书机器人 推送消息。
184
    """
185
    if not push_config.get("FSKEY"):
186
        print("飞书 服务的 FSKEY 未设置!!\n取消推送")
187
        return
188
    print("飞书 服务启动")
189
190
    url = f'https://open.feishu.cn/open-apis/bot/v2/hook/{push_config.get("FSKEY")}'
191
    data = {"msg_type": "text", "content": {"text": f"{title}\n\n{content}"}}
192
    try:
193
        response = requests.post(url, data=json.dumps(data), timeout=15)
194
        try:
195
            datas = response.json()
196
            if datas.get("StatusCode") == 0:
197
                print("飞书 推送成功!")
198
            else:
199
                print(f"飞书 推送失败!响应数据:{datas}")
200
        except json.JSONDecodeError:
201
            print(f"推送返回值非 json 格式,请检查网址和账号是否填写正确。\n{response.text}")
202
    except requests.exceptions.RequestException:
203
        print(f"网络异常,请检查你的网络连接、推送服务器和代理配置。\n{traceback.format_exc()}")
204
    except Exception:
205
        print(f"其他错误信息:\n{traceback.format_exc()}")
206
207
208
def go_cqhttp(title: str, content: str) -> None:
209
    """
210
    使用 go_cqhttp 推送消息。
211
    """
212
    if not push_config.get("GOBOT_URL") or not push_config.get("GOBOT_QQ"):
213
        print("go-cqhttp 服务的 GOBOT_URL 或 GOBOT_QQ 未设置!!\n取消推送")
214
        return
215
    print("go-cqhttp 服务启动")
216
217
    url = f'{push_config.get("GOBOT_URL")}?access_token={push_config.get("GOBOT_TOKEN")}&{push_config.get("GOBOT_QQ")}&message=标题:{title}\n内容:{content}'
218
219
    try:
220
        response = requests.get(url, timeout=15)
221
        try:
222
            datas = response.json()
223
            if datas.get("status") == "ok":
224
                print("go-cqhttp 推送成功!")
225
            else:
226
                print(f"go-cqhttp 推送失败!响应数据:{datas}")
227
        except json.JSONDecodeError:
228
            print(f"推送返回值非 json 格式,请检查网址和账号是否填写正确。\n{response.text}")
229
    except requests.exceptions.RequestException:
230
        print(f"网络异常,请检查你的网络连接、推送服务器和代理配置。\n{traceback.format_exc()}")
231
    except Exception:
232
        print(f"其他错误信息:\n{traceback.format_exc()}")
233
234
235
def iGot(title: str, content: str) -> None:
236
    """
237
    使用 iGot 推送消息。
238
    """
239
    if not push_config.get("IGOT_PUSH_KEY"):
240
        print("iGot 服务的 IGOT_PUSH_KEY 未设置!!\n取消推送")
241
        return
242
    print("iGot 服务启动")
243
244
    url = f'https://push.hellyw.com/{push_config.get("IGOT_PUSH_KEY")}'
245
    data = {"title": title, "content": content}
246
    headers = {"Content-Type": "application/x-www-form-urlencoded"}
247
248
    try:
249
        response = requests.post(url, data=data, headers=headers, timeout=15)
250
        try:
251
            datas = response.json()
252
            if datas.get("ret") == 0:
253
                print("iGot 推送成功!")
254
            else:
255
                print(f'iGot 推送失败!错误信息:{datas.get("errMsg")}')
256
        except json.JSONDecodeError:
257
            print(f"推送返回值非 json 格式,请检查网址和账号是否填写正确。\n{response.text}")
258
    except requests.exceptions.RequestException:
259
        print(f"网络异常,请检查你的网络连接、推送服务器和代理配置。\n{traceback.format_exc()}")
260
    except Exception:
261
        print(f"其他错误信息:\n{traceback.format_exc()}")
262
263
264
def serverJ(title: str, content: str) -> None:
265
    """
266
    通过 serverJ 推送消息。
267
    """
268
    if not push_config.get("PUSH_KEY"):
269
        print("serverJ 服务的 PUSH_KEY 未设置!!\n取消推送")
270
        return
271
    print("serverJ 服务启动")
272
273
    data = {"text": title, "desp": content.replace("\n", "\n\n")}
274
    if push_config.get("PUSH_KEY").index("SCT") != -1:
275
        url = f'https://sctapi.ftqq.com/{push_config.get("PUSH_KEY")}.send'
276
    else:
277
        url = f'https://sc.ftqq.com/${push_config.get("PUSH_KEY")}.send'
278
279
    try:
280
        response = requests.post(url, data=data, timeout=15)
281
        try:
282
            datas = response.json()
283
            if datas.get("errno") == 0 or datas.get("code") == 0:
284
                print("serverJ 推送成功!")
285
            elif datas.get("code") == 40001:
286
                print("serverJ 推送失败!PUSH_KEY 错误。")
287
            else:
288
                print(f'serverJ 推送失败!错误码:{datas.get("message")}')
289
        except json.JSONDecodeError:
290
            print(f"推送返回值非 json 格式,请检查网址和账号是否填写正确。\n{response.text}")
291
    except requests.exceptions.RequestException:
292
        print(f"网络异常,请检查你的网络连接、推送服务器和代理配置。\n{traceback.format_exc()}")
293
    except Exception:
294
        print(f"其他错误信息:\n{traceback.format_exc()}")
295
296
297
def pushplus_bot(title: str, content: str) -> None:
298
    """
299
    通过 push+ 推送消息。
300
    """
301
    if not push_config.get("PUSH_PLUS_TOKEN"):
302
        print("PUSHPLUS 服务的 PUSH_PLUS_TOKEN 未设置!!\n取消推送")
303
        return
304
    print("PUSHPLUS 服务启动")
305
306
    url = "http://www.pushplus.plus/send"
307
    data = {
308
        "token": push_config.get("PUSH_PLUS_TOKEN"),
309
        "title": title,
310
        "content": content,
311
        "topic": push_config.get("PUSH_PLUS_USER"),
312
    }
313
    body = json.dumps(data).encode(encoding="utf-8")
314
    headers = {"Content-Type": "application/json"}
315
316
    try:
317
        response = requests.post(url=url, data=body, headers=headers, timeout=15)
318
        try:
319
            datas = response.json()
320
            if datas.get("code") == 200:
321
                print("PUSHPLUS 推送成功!")
322
            elif datas.get("code") == 600:
323
                url2 = "http://pushplus.hxtrip.com/send"
324
                headers["Accept"] = "application/json"
325
                response2 = requests.post(
326
                    url=url2, data=body, headers=headers, timeout=15
327
                )
328
                try:
329
                    datas2 = response2.json()
330
                    if datas2.get("code") == 200:
331
                        print("PUSHPLUS(hxtrip) 推送成功!")
332
                    elif datas2.get("code") == 600:
333
                        print("PUSHPLUS 推送失败!PUSH_PLUS_TOKEN 错误。")
334
                    else:
335
                        print(f"PUSHPLUS(hxtrip) 推送失败!响应数据:{datas2}")
336
                except json.JSONDecodeError:
337
                    print(f"推送返回值非 json 格式,请检查网址和账号是否填写正确。\n{response2.text}")
338
            else:
339
                print(f"PUSHPLUS 推送失败!响应数据:{datas}")
340
        except json.JSONDecodeError:
341
            print(f"推送返回值非 json 格式,请检查网址和账号是否填写正确。\n{response.text}")
342
    except requests.exceptions.RequestException:
343
        print(f"网络异常,请检查你的网络连接、推送服务器和代理配置。\n{traceback.format_exc()}")
344
    except Exception:
345
        print(f"其他错误信息:\n{traceback.format_exc()}")
346
347
348
def qmsg_bot(title: str, content: str) -> None:
349
    """
350
    使用 qmsg 推送消息。
351
    """
352
    if not push_config.get("QMSG_KEY") or not push_config.get("QMSG_TYPE"):
353
        print("qmsg 的 QMSG_KEY 或者 QMSG_TYPE 未设置!!\n取消推送")
354
        return
355
    print("qmsg 服务启动")
356
357
    url = f'https://qmsg.zendee.cn/{push_config.get("QMSG_TYPE")}/{push_config.get("QMSG_KEY")}'
358
    payload = {"msg": f'{title}\n\n{content.replace("----", "-")}'.encode("utf-8")}
359
360
    try:
361
        response = requests.post(url=url, params=payload, timeout=15)
362
        try:
363
            datas = response.json()
364
            if response.get("code") == 0:
365
                print("qmsg 推送成功!")
366
            else:
367
                print(f'qmsg 推送失败!错误信息:{datas.get("reason")}')
368
        except json.JSONDecodeError:
369
            print(f"推送返回值非 json 格式,请检查网址和账号是否填写正确。\n{response.text}")
370
    except requests.exceptions.RequestException:
371
        print(f"网络异常,请检查你的网络连接、推送服务器和代理配置。\n{traceback.format_exc()}")
372
    except Exception:
373
        print(f"其他错误信息:\n{traceback.format_exc()}")
374
375
376
def wecom_app(title: str, content: str) -> None:
377
    """
378
    通过 企业微信 APP 推送消息。
379
    """
380
    if not push_config.get("QYWX_AM"):
381
        print("QYWX_AM 未设置!!\n取消推送")
382
        return
383
    QYWX_AM_AY = re.split(",", push_config.get("QYWX_AM"))
384
    if 4 < len(QYWX_AM_AY) > 5:
385
        print("QYWX_AM 设置错误!!\n取消推送")
386
        return
387
    print("企业微信 APP 服务启动")
388
389
    corpid = QYWX_AM_AY[0]
390
    corpsecret = QYWX_AM_AY[1]
391
    touser = QYWX_AM_AY[2]
392
    agentid = QYWX_AM_AY[3]
393
    try:
394
        media_id = QYWX_AM_AY[4]
395
    except IndexError:
396
        media_id = ""
397
    wx = WeCom(corpid, corpsecret, agentid)
398
    # 如果没有配置 media_id 默认就以 text 方式发送
399
    try:
400
        if not media_id:
401
            message = title + "\n\n" + content
402
            datas = wx.send_text(message, touser)
403
        else:
404
            datas = wx.send_mpnews(title, content, media_id, touser)
405
        if datas == "ok":
406
            print("企业微信推送成功!")
407
        else:
408
            print(f"企业微信推送失败!错误信息:{datas}")
409
    except requests.exceptions.RequestException:
410
        print(f"网络异常,请检查你的网络连接、推送服务器和代理配置。\n{traceback.format_exc()}")
411
    except Exception:
412
        print(f"其他错误信息:\n{traceback.format_exc()}")
413
414
415
class WeCom:
416
    def __init__(self, corpid, corpsecret, agentid):
417
        self.CORPID = corpid
418
        self.CORPSECRET = corpsecret
419
        self.AGENTID = agentid
420
421
    def get_access_token(self):
422
        url = "https://qyapi.weixin.qq.com/cgi-bin/gettoken"
423
        values = {
424
            "corpid": self.CORPID,
425
            "corpsecret": self.CORPSECRET,
426
        }
427
        req = requests.post(url, params=values, timeout=15)
428
        datas = json.loads(req.text)
429
        return datas.get("access_token")
430
431
    def send_text(self, message, touser="@all"):
432
        send_url = (
433
            "https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token="
434
            + self.get_access_token()
435
        )
436
        send_values = {
437
            "touser": touser,
438
            "msgtype": "text",
439
            "agentid": self.AGENTID,
440
            "text": {"content": message},
441
            "safe": "0",
442
        }
443
        send_msges = bytes(json.dumps(send_values), "utf-8")
444
        response = requests.post(send_url, send_msges, timeout=15)
445
        try:
446
            datas = response.json()
447
            return datas.get("errmsg")
448
        except json.JSONDecodeError:
449
            print(f"推送返回值非 json 格式,请检查网址和账号是否填写正确。\n{response.text}")
450
            return response.text
451
452
    def send_mpnews(self, title, message, media_id, touser="@all"):
453
        send_url = (
454
            "https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token="
455
            + self.get_access_token()
456
        )
457
        send_values = {
458
            "touser": touser,
459
            "msgtype": "mpnews",
460
            "agentid": self.AGENTID,
461
            "mpnews": {
462
                "articles": [
463
                    {
464
                        "title": title,
465
                        "thumb_media_id": media_id,
466
                        "author": "Author",
467
                        "content_source_url": "",
468
                        "content": message.replace("\n", "<br/>"),
469
                        "digest": message,
470
                    }
471
                ]
472
            },
473
        }
474
        send_msges = bytes(json.dumps(send_values), "utf-8")
475
        response = requests.post(send_url, send_msges, timeout=15)
476
        try:
477
            datas = response.json()
478
            return datas.get("errmsg")
479
        except json.JSONDecodeError:
480
            print(f"推送返回值非 json 格式,请检查网址和账号是否填写正确。\n{response.text}")
481
            return response.text
482
483
484
def wecom_bot(title: str, content: str) -> None:
485
    """
486
    通过 企业微信机器人 推送消息。
487
    """
488
    if not push_config.get("QYWX_KEY"):
489
        print("企业微信机器人 服务的 QYWX_KEY 未设置!!\n取消推送")
490
        return
491
    print("企业微信机器人服务启动")
492
493
    url = f"https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key={push_config.get('QYWX_KEY')}"
494
    headers = {"Content-Type": "application/json;charset=utf-8"}
495
    data = {"msgtype": "text", "text": {"content": f"{title}\n\n{content}"}}
496
497
    try:
498
        response = requests.post(
499
            url=url, data=json.dumps(data), headers=headers, timeout=15
500
        )
501
        try:
502
            datas = response.json()
503
            if datas.get("errcode") == 0:
504
                print("企业微信机器人 推送成功!")
505
            else:
506
                print(f"企业微信机器人 推送失败!响应数据:{datas}")
507
        except json.JSONDecodeError:
508
            print(f"推送返回值非 json 格式,请检查网址和账号是否填写正确。\n{response.text}")
509
    except requests.exceptions.RequestException:
510
        print(f"网络异常,请检查你的网络连接、推送服务器和代理配置。\n{traceback.format_exc()}")
511
    except Exception:
512
        print(f"其他错误信息:\n{traceback.format_exc()}")
513
514
515
def telegram_bot(title: str, content: str) -> None:
516
    """
517
    使用 telegram 机器人 推送消息。
518
    """
519
    if not push_config.get("TG_BOT_TOKEN") or not push_config.get("TG_USER_ID"):
520
        print("tg 服务的 bot_token 或者 user_id 未设置!!\n取消推送")
521
        return
522
    print("tg 服务启动")
523
524
    if push_config.get("TG_API_HOST"):
525
        url = f"https://{push_config.get('TG_API_HOST')}/bot{push_config.get('TG_BOT_TOKEN')}/sendMessage"
526
    else:
527
        url = (
528
            f"https://api.telegram.org/bot{push_config.get('TG_BOT_TOKEN')}/sendMessage"
529
        )
530
    headers = {"Content-Type": "application/x-www-form-urlencoded"}
531
    payload = {
532
        "chat_id": str(push_config.get("TG_USER_ID")),
533
        "text": f"{title}\n\n{content}",
534
        "disable_web_page_preview": "true",
535
    }
536
    proxies = None
537
    if push_config.get("TG_PROXY_HOST") and push_config.get("TG_PROXY_PORT"):
538
        if push_config.get("TG_PROXY_AUTH") is not None and "@" not in push_config.get(
539
            "TG_PROXY_HOST"
540
        ):
541
            push_config["TG_PROXY_HOST"] = (
542
                push_config.get("TG_PROXY_AUTH")
543
                + "@"
544
                + push_config.get("TG_PROXY_HOST")
545
            )
546
        proxyStr = "http://{}:{}".format(
547
            push_config.get("TG_PROXY_HOST"), push_config.get("TG_PROXY_PORT")
548
        )
549
        proxies = {"http": proxyStr, "https": proxyStr}
550
551
    try:
552
        response = requests.post(
553
            url=url, headers=headers, params=payload, proxies=proxies, timeout=15
554
        )
555
        try:
556
            datas = response.json()
557
            if datas.get("ok") == True:
558
                print("tg 推送成功!")
559
            elif datas.get("error_code") == 400:
560
                print("tg 推送失败!请主动给 bot 发送一条消息并检查接收用户 TG_USER_ID 是否正确。")
561
            elif datas.get("error_code") == 401:
562
                print("tg 推送失败!TG_BOT_TOKEN 填写错误。")
563
            else:
564
                print(f"tg 推送失败!响应数据:{datas}")
565
        except json.JSONDecodeError:
566
            print(f"推送返回值非 json 格式,请检查网址和账号是否填写正确。\n{response.text}")
567
    except requests.exceptions.RequestException:
568
        print(f"网络异常,请检查你的网络连接、推送服务器和代理配置。\n{traceback.format_exc()}")
569
    except Exception:
570
        print(f"其他错误信息:\n{traceback.format_exc()}")
571
572
573
def one() -> str:
574
    """
575
    获取一条一言。
576
    :return:
577
    """
578
    try:
579
        url = "https://v1.hitokoto.cn/"
580
        res = requests.get(url).json()
581
        return res["hitokoto"] + "    ----" + res["from"]
582
    except requests.exceptions.ConnectionError:
583
        return ""
584
585
586
if push_config.get("BARK_PUSH"):
587
    notify_function.append(bark)
588
if push_config.get("CONSOLE"):
589
    notify_function.append(console)
590
if push_config.get("DD_BOT_TOKEN") and push_config.get("DD_BOT_SECRET"):
591
    notify_function.append(dingding_bot)
592
if push_config.get("FSKEY"):
593
    notify_function.append(feishu_bot)
594
if push_config.get("GOBOT_URL") and push_config.get("GOBOT_QQ"):
595
    notify_function.append(go_cqhttp)
596
if push_config.get("IGOT_PUSH_KEY"):
597
    notify_function.append(iGot)
598
if push_config.get("PUSH_KEY"):
599
    notify_function.append(serverJ)
600
if push_config.get("PUSH_PLUS_TOKEN"):
601
    notify_function.append(pushplus_bot)
602
if push_config.get("QMSG_KEY") and push_config.get("QMSG_TYPE"):
603
    notify_function.append(qmsg_bot)
604
if push_config.get("QYWX_AM"):
605
    notify_function.append(wecom_app)
606
if push_config.get("QYWX_KEY"):
607
    notify_function.append(wecom_bot)
608
if push_config.get("TG_BOT_TOKEN") and push_config.get("TG_USER_ID"):
609
    notify_function.append(telegram_bot)
610
611
612
def send(title: str, content: str) -> None:
613
    if not content:
614
        print(f"{title} 推送内容为空!")
615
        return
616
617
    hitokoto = push_config.get("HITOKOTO")
618
619
    text = one() if hitokoto else ""
620
    content += "\n\n" + text
621
622
    ts = [
623
        threading.Thread(target=mode, args=(title, content), name=mode.__name__)
624
        for mode in notify_function
625
    ]
626
    [t.start() for t in ts]
627
    [t.join() for t in ts]
628
629
630
def main():
631
    send("title", "content")
632
633
634
if __name__ == "__main__":
635
    main()
636