新增推送到Telegram

env.docker.py.example

@@ -46,6 +46,12 @@ NOTIFICATION_VOICE_CODE_PHONE = 'your phone'  # 接受通知的手机号
 DINGTALK_ENABLED = 0
 DINGTALK_WEBHOOK = 'https://oapi.dingtalk.com/robot/send?access_token=your token'
 
+# Telegram消息推送
+# 打开链接关注Bot：https://t.me/notificationme_bot
+# 将Bot回复给你的专属链接复制到下方，如果没有回复则发送给Bot这条信息: /start
+TELEGRAM_ENABLED = 0
+TELEGRAM_BOT_API_URL = 'https://tgbot.lbyczf.com/sendMessage/:your_token'
+
 # 输出日志到文件 (Docker 中不建议修改此组配置项)
 OUT_PUT_LOG_TO_FILE_ENABLED = 1
 OUT_PUT_LOG_TO_FILE_PATH = '/config/12306.log'  # 日志目录

env.py.example

@@ -46,6 +46,12 @@ NOTIFICATION_VOICE_CODE_PHONE = 'your phone'  # 接受通知的手机号
 DINGTALK_ENABLED = 0
 DINGTALK_WEBHOOK = 'https://oapi.dingtalk.com/robot/send?access_token=your token'
 
+# Telegram消息推送
+# 打开链接关注Bot：https://t.me/notificationme_bot
+# 将Bot回复给你的专属链接复制到下方，如果没有回复则发送给Bot这条信息: /start
+TELEGRAM_ENABLED = 0
+TELEGRAM_BOT_API_URL = 'https://tgbot.lbyczf.com/sendMessage/:your_token'
+
 # 输出日志到文件
 OUT_PUT_LOG_TO_FILE_ENABLED = 0
 OUT_PUT_LOG_TO_FILE_PATH = 'runtime/12306.log'  # 日志目录

py12306/app.py

@@ -113,6 +113,10 @@ class App:
             CommonLog.add_quick_log(CommonLog.MESSAGE_TEST_SEND_DINGTALK).flush()
             Notification.dingtalk_webhook('测试发送信息')
 
+        if Config().TELEGRAM_ENABLED:  # Telegram通知
+            CommonLog.add_quick_log(CommonLog.MESSAGE_TEST_SEND_TELEGRAM).flush()
+            Notification.send_to_telegram('测试发送信息')
+
     @classmethod
     def run_check(cls):
         """

py12306/config.py

@@ -58,6 +58,10 @@ class Config:
     DINGTALK_ENABLED = 0
     DINGTALK_WEBHOOK = ''
 
+    # Telegram推送配置
+    TELEGRAM_ENABLED = 0
+    TELEGRAM_BOT_API_URL = ''
+
     # 邮箱配置
     EMAIL_ENABLED = 0
     EMAIL_SENDER = ''

py12306/helpers/notification.py

@@ -30,6 +30,11 @@ class Notification():
         self = cls()
         self.send_email_by_smtp(to, title, content)
 
+    @classmethod
+    def send_to_telegram(cls, content=''):
+        self = cls()
+        self.send_to_telegram_bot(content=content)
+
     def send_voice_code_of_yiyuan(self, phone, name='', content=''):
         """
         发送语音验证码
@@ -82,13 +87,29 @@ class Notification():
         except Exception as e:
             CommonLog.add_quick_log(CommonLog.MESSAGE_SEND_EMAIL_FAIL.format(e)).flush()
 
-    def send_dingtalk_by_webbook(self,content):
+    def send_dingtalk_by_webbook(self, content):
         from dingtalkchatbot.chatbot import DingtalkChatbot
         webhook = Config().DINGTALK_WEBHOOK
         dingtalk = DingtalkChatbot(webhook)
         dingtalk.send_text(msg=content, is_at_all=True)
         pass
 
+    def send_to_telegram_bot(self, content):
+        bot_api_url = Config().TELEGRAM_BOT_API_URL
+        if not bot_api_url:
+            return False
+        data = {
+            'text': content
+        }
+        response = self.session.request(url=bot_api_url, method='POST', data=data)
+        result = response.json().get('result')
+        response_status = result.get('statusCode')
+        if response_status == 200:
+            CommonLog.add_quick_log(CommonLog.MESSAGE_SEND_TELEGRAM_SUCCESS).flush()
+        else:
+            response_error_message = result.get('description')
+            CommonLog.add_quick_log(CommonLog.MESSAGE_SEND_TELEGRAM_FAIL.format(response_error_message)).flush()
+
 
 if __name__ == '__main__':
     name = '张三4'

py12306/log/common_log.py

@@ -24,6 +24,7 @@ class CommonLog(BaseLog):
     MESSAGE_TEST_SEND_VOICE_CODE = '正在测试发送语音验证码...'
     MESSAGE_TEST_SEND_EMAIL = '正在测试发送邮件...'
     MESSAGE_TEST_SEND_DINGTALK = '正在测试发送钉钉消息...'
+    MESSAGE_TEST_SEND_TELEGRAM = '正在测试推送到Telegram...'
 
     MESSAGE_CONFIG_FILE_DID_CHANGED = '配置文件已修改，正在重新加载中\n'
     MESSAGE_API_RESPONSE_CAN_NOT_BE_HANDLE = '接口返回错误'
@@ -31,6 +32,9 @@ class CommonLog(BaseLog):
     MESSAGE_SEND_EMAIL_SUCCESS = '邮件发送成功，请检查收件箱'
     MESSAGE_SEND_EMAIL_FAIL = '邮件发送失败，请手动检查配置，错误原因 {}'
 
+    MESSAGE_SEND_TELEGRAM_SUCCESS = 'Telegram推送成功'
+    MESSAGE_SEND_TELEGRAM_FAIL = 'Telegram推送失败，错误原因 {}'
+
     MESSAGE_OUTPUT_TO_FILE_IS_UN_ENABLE = '请先打开配置：输出到文件'
 
     MESSAGE_GET_RESPONSE_FROM_FREE_AUTO_CODE = '从免费打码获取结果失败'
@@ -72,6 +76,7 @@ class CommonLog(BaseLog):
             '语音验证码: {}'.format(get_true_false_text(Config().NOTIFICATION_BY_VOICE_CODE, enable, disable)))
         self.add_quick_log('邮件通知: {}'.format(get_true_false_text(Config().EMAIL_ENABLED, enable, disable)))
         self.add_quick_log('钉钉通知: {}'.format(get_true_false_text(Config().DINGTALK_ENABLED, enable, disable)))
+        self.add_quick_log('Telegram通知: {}'.format(get_true_false_text(Config().TELEGRAM_ENABLED, enable, disable)))
         self.add_quick_log('查询间隔: {} 秒'.format(Config().QUERY_INTERVAL))
         self.add_quick_log('用户心跳检测间隔: {} 秒'.format(Config().USER_HEARTBEAT_INTERVAL))
         self.add_quick_log('WEB 管理页面: {}'.format(get_true_false_text(Config().WEB_ENABLE, enable, disable)))

py12306/order/order.py

@@ -84,6 +84,9 @@ class Order:
                                     OrderLog.MESSAGE_ORDER_SUCCESS_NOTIFICATION_OF_EMAIL_CONTENT.format(self.order_id))
         if Config().DINGTALK_ENABLED:  # 钉钉通知
             Notification.dingtalk_webhook(OrderLog.MESSAGE_ORDER_SUCCESS_NOTIFICATION_OF_EMAIL_CONTENT.format(self.order_id))
+        if Config().TELEGRAM_ENABLED:   # Telegram推送
+            Notification.send_to_telegram(
+                OrderLog.MESSAGE_ORDER_SUCCESS_NOTIFICATION_OF_EMAIL_CONTENT.format(self.order_id))
         while sustain_time:  # TODO 后面直接查询有没有待支付的订单就可以
             num += 1
             if Config().NOTIFICATION_BY_VOICE_CODE:  # 语音通知