#coding: utf-8 # +------------------------------------------------------------------- # | 宝塔Linux面板 # +------------------------------------------------------------------- # | Copyright (c) 2015-2020 宝塔软件(http://www.bt.cn) All rights reserved. # +------------------------------------------------------------------- # | Author: 沐落 # | Author: lx # | 消息通道邮箱模块 # +------------------------------------------------------------------- import os, sys import time panelPath = "/www/server/panel" os.chdir(panelPath) sys.path.insert(0,panelPath + "/class/") import public, json, requests from requests.packages import urllib3 # 关闭警告 urllib3.disable_warnings() import socket import requests.packages.urllib3.util.connection as urllib3_cn class wx_account_msg: __module_name = None __default_pl = "{}/data/default_msg_channel.pl".format(panelPath) conf_path = '{}/data/wx_account_msg.json'.format(panelPath) user_info = None def __init__(self): try: self.user_info = json.loads(public.ReadFile("{}/data/userInfo.json".format(public.get_panel_path()))) except: self.user_info=None self.__module_name = self.__class__.__name__.replace('_msg','') def get_version_info(self,get): """ 获取版本信息 """ data = {} data['ps'] = '宝塔微信公众号，用于接收面板消息推送' data['version'] = '1.0' data['date'] = '2022-08-15' data['author'] = '宝塔' data['help'] = 'http://www.bt.cn' return data def get_local_ip(self): '''获取内网IP''' import socket try: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.connect(('8.8.8.8', 80)) ip = s.getsockname()[0] return ip finally: s.close() return '127.0.0.1' def __get_default_channel(self): """ @获取默认消息通道 """ try: if public.readFile(self.__default_pl) == self.__module_name: return True except:pass return False def get_config(self, get): """ 微信公众号配置 """ if os.path.exists(self.conf_path): #60S内不重复加载 start_time=int(time.time()) if os.path.exists("data/wx_account_msg.lock"): lock_time=int(public.ReadFile("data/wx_account_msg.lock")) #大于60S重新加载 if start_time-lock_time>60: public.run_thread(self.get_web_info2) public.WriteFile("data/wx_account_msg.lock",str(start_time)) else: public.WriteFile("data/wx_account_msg.lock",str(start_time)) public.run_thread(self.get_web_info2) data=json.loads(public.ReadFile(self.conf_path)) data['default'] = self.__get_default_channel() return data else: public.run_thread(self.get_web_info2) return {"success":False,"res":"未获取到配置信息"} def set_config(self,get): """ @设置默认值 """ if 'default' in get and get['default']: public.writeFile(self.__default_pl, self.__module_name) return public.returnMsg(True, '设置成功') def get_web_info(self,get): if self.user_info is None: return public.returnMsg(False, '未获取到用户绑定的信息') url = "https://www.bt.cn/api/v2/user/wx_web/info" data = { "uid": self.user_info["uid"], "access_key": self.user_info["access_key"], "serverid":self.user_info["serverid"] } try: allowed_gai_family_lib=urllib3_cn.allowed_gai_family def allowed_gai_family(): family = socket.AF_INET return family urllib3_cn.allowed_gai_family = allowed_gai_family datas=requests.post(url=url, data=data,verify=False,timeout=10).json() urllib3_cn.allowed_gai_family=allowed_gai_family_lib if datas["success"]: public.WriteFile(self.conf_path,json.dumps(datas)) return public.returnMsg(True, datas) else: public.WriteFile(self.conf_path, json.dumps(datas)) return public.returnMsg(False, datas) except: public.WriteFile(self.conf_path, json.dumps({"success":False,"res":"链接云端失败,请检查网络"})) return public.returnMsg(False,"链接云端失败,请检查网络") def get_web_info2(self): if self.user_info is None: return public.returnMsg(False, '未获取到用户绑定的信息') url = "https://www.bt.cn/api/v2/user/wx_web/info" data = { "uid": self.user_info["uid"], "access_key": self.user_info["access_key"], "serverid":self.user_info["serverid"] } try: allowed_gai_family_lib=urllib3_cn.allowed_gai_family def allowed_gai_family(): family = socket.AF_INET return family urllib3_cn.allowed_gai_family = allowed_gai_family datas=requests.post(url=url, data=data,verify=False,timeout=5).json() urllib3_cn.allowed_gai_family=allowed_gai_family_lib if datas["success"]: public.WriteFile(self.conf_path,json.dumps(datas)) return public.returnMsg(True, datas) else: public.WriteFile(self.conf_path, json.dumps(datas)) return public.returnMsg(False, datas) except: public.WriteFile(self.conf_path, json.dumps({"success":False,"res":"链接云端失败"})) return public.returnMsg(False,"链接云端失败") def get_auth_url(self,get): if self.user_info is None: return public.returnMsg(False, '未获取到用户绑定的信息') url = "https://www.bt.cn/api/v2/user/wx_web/get_auth_url" data = { "uid": self.user_info["uid"], "access_key": self.user_info["access_key"], "serverid":self.user_info["serverid"] } try: allowed_gai_family_lib=urllib3_cn.allowed_gai_family def allowed_gai_family(): family = socket.AF_INET return family urllib3_cn.allowed_gai_family = allowed_gai_family datas=requests.post(url=url, data=data,verify=False,timeout=5).json() urllib3_cn.allowed_gai_family=allowed_gai_family_lib if datas["success"]: return public.returnMsg(True, datas) else: return public.returnMsg(False, datas) except: return public.returnMsg(False,"链接云端失败") def get_send_msg(self,msg): """ @name 处理md格式 """ try: import re if msg.find("####") >= 0: msg = msg.replace("####",">").replace("\n\n","\n").strip() s_list = msg.split('\n') if len(s_list) > 3: title = s_list[0].replace(" ","") s_list = s_list[3:] s_list.insert(0,title) msg = '\n'.join(s_list) reg = '(.+)' tmp = re.search(reg,msg) if tmp: tmp = tmp.groups()[0] msg = re.sub(reg,tmp,msg) except:pass return msg def send_msg(self,msg): """ 微信发送信息 @msg 消息正文 """ if self.user_info is None: return public.returnMsg(False,'未获取到用户信息') msg = self.get_send_msg(msg) url="https://www.bt.cn/api/v2/user/wx_web/send_template_msg" datassss = { "first": { "value": "堡塔主机告警", }, "keyword1": { "value": "内网IP " + self.get_local_ip() + "\n外网IP " + self.user_info["address"] + " \n服务器别名 " + public.GetConfigValue("title"), }, "keyword2": { "value": "堡塔主机告警", }, "keyword3": { "value": msg, }, "remark": { "value": "如有疑问，请联系宝塔客服", }, } data = { "uid": self.user_info["uid"], "access_key": self.user_info["access_key"], "data": datassss } headers = {'Content-Type': 'application/json'} try: allowed_gai_family_lib=urllib3_cn.allowed_gai_family def allowed_gai_family(): family = socket.AF_INET return family urllib3_cn.allowed_gai_family = allowed_gai_family x = requests.post(url=url, json=data, headers=headers,verify=False).json() urllib3_cn.allowed_gai_family=allowed_gai_family_lib if x['success']: return public.returnMsg(True,'发送成功') else: return public.returnMsg(False, '发送失败') except: return public.returnMsg(False,'微信消息发送失败。 --> {}'.format(public.get_error_info())) def push_data(self,data): return self.send_msg(data['msg']) def uninstall(self): if os.path.exists(self.conf_path): os.remove(self.conf_path)