almalinux8安装tomcat9 从yum源直接安装tomcat9yum install tomcat tomcat-webapps tomcat-admin-webapps启动服务systemctl enable --now tomcat访问tomcat正常http://IP:8080/配置Manager App访问范围/usr/share/tomcat/webapps/manager/META-INF/context.xml原内容 <Context antiResourceLocking="false" privileged="true" > <CookieProcessor className="org.apache.tomcat.util.http.Rfc6265CookieProcessor" sameSiteCookies="strict" /> <Valve className="org.apache.catalina.valves.RemoteAddrValve" allow="127\.\d+\.\d+\.\d+|::1|0:0:0:0:0:0:0:1" /> <Manager sessionAttributeValueClassNameFilter="java\.lang\.(?:Boolean|Integer|Long|Number|String)|org\.apache\.catalina\.filters\.CsrfPreventionFilter\$LruCache(?:\$1)?|java\.util\.(?:Linked)?HashMap"/> </Context> 增加任意IP访问 \d+\.\d+\.\d+\.\d+ 修改后 <Context antiResourceLocking="false" privileged="true" > <CookieProcessor className="org.apache.tomcat.util.http.Rfc6265CookieProcessor" sameSiteCookies="strict" /> <Valve className="org.apache.catalina.valves.RemoteAddrValve" allow="127\.\d+\.\d+\.\d+|::1|0:0:0:0:0:0:0:1|\d+\.\d+\.\d+\.\d+" /> <Manager sessionAttributeValueClassNameFilter="java\.lang\.(?:Boolean|Integer|Long|Number|String)|org\.apache\.catalina\.filters\.CsrfPreventionFilter\$LruCache(?:\$1)?|java\.util\.(?:Linked)?HashMap"/> </Context> 配置Manager App用户名密码/usr/share/tomcat/conf/tomcat-users.xml在倒数第二行添加 <role rolename="manager-gui"/> <user username="admin" password="admin" roles="manager-gui"/>重启服务systemctl restart tomcat完毕

0基础上手python3编程，本地文件存活监控 架构图 判断文件是否存在；如果文件不存在，判断check.config是否为1，如果为1程序退出，如果不存在或为0，生成或写入1并发送企业微信告警；如果文件存在，判断check.config是否为0，如果为0程序退出，如果不存在或为1，生成或写入0并发送企业微信告警；预览 系统组成 由两个文件组成 check.py check.conf，其中 check.conf为自动生成代码 check.pyimport os,requests #企业微信机器人 def post_weixin(stats): url = 'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=用自己的' body = { "msgtype": "news", "news": { "articles": [ { "title": "文件监控", "description": stats, "url": "90apt.com", "picurl": "用自己的" } ] }} response = requests.post(url, json=body) print(response.text) print(response.status_code) if os.path.exists("D:\hello.txt") == False: #判断文件是否存在 f1 = open("D:\check.config", 'r') #读取文件 config = f1.read() if config == "1": f1.close() else: f1.close() f1 = open("D:\check.config", 'w+') f1.write("1") #写入文件 post_weixin("监控到文件丢失") f1.close() else: f1 = open("D:\check.config", 'r') config = f1.read() if config == "0": f1.close() else: f1 = open("D:\check.config", 'w+') f1.write("0") post_weixin("监控到文件已生成") f1.close()总结 简单

0基础上手python3编程，单线程监控摄像头巡检平台 架构图 读取配置文件；通过rtsp连接摄像头，保存一张截图；发送企业微信通知；记录log日志；预览 系统组成 rtsp.py rtsp.json代码 由于代码在windows平台运行，因此使用绝对路径C:\\jiankong\，文件放置在此目录下。rtsp.pyimport cv2,os,json,time,requests,codecs from func_timeout import func_set_timeout from multiprocessing import Process @func_set_timeout(5) def dayin(): video1 = "rtsp://"+rtspconfig['name']+":"+rtspconfig['password']+"@"+rtspconfig['ip']+rtsp_config[rtspconfig['brand']] cap = cv2.VideoCapture(video1) # 使用整数，此处打开的本地摄像头 while 1: ret, frame = cap.read() if ret == False: # 若没有帧返回，则重新刷新rtsp视频流 continue else: break; # cv2.imshow("capture",frame) cv2.imwrite(nowdir+"\\"+rtspconfig['ip']+".jpeg", frame) cap.release() return rtspconfig['hostname']+"保存成功" def get_config(): config = json.loads(open("C:\\jiankong\\rtsp.json", encoding='utf-8').read()) return config def post_weixin(stats): url = 'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=用自己的' body = { "msgtype": "news", "news": { "articles": [ { "title": "监控摄像头自动巡检", "description": stats, "url": "90apt.com", "picurl": "https://www.hikvision.com/content/dam/hikvision/cn/product/network-camera/fixed-ipc/%E7%BB%8F%E9%94%80%E7%B3%BB%E5%88%97/%E7%BB%8F%E9%94%80%E7%B3%BB%E5%88%97%E5%AF%BC%E8%88%AA%E7%9B%AE%E5%BD%95.jpg" } ] }} response = requests.post(url, json=body) print(response.text) print(response.status_code) total = 0 fail = 0 weixindata = "" rtsp_config = get_config() print(rtsp_config) #path1=os.path.abspath('.') path1=("C:\\jiankong\\") nowtime = time.strftime("%Y%m%d", time.localtime()) try: os.mkdir(path1 + "\\" + nowtime) except: None nowdir = (path1 + "\\" + nowtime) for rtspconfig in rtsp_config['rtsp']: total = total + 1 try: dayin() except: weixindata = weixindata + (rtspconfig['hostname']+" "+rtspconfig['ip']+" 网络或账号密码错误\n") fail = fail + 1 weixinpost = "总计巡检:"+str(total)+"台"+"，故障摄像头："+str(fail)+"台\n"+weixindata post_weixin(weixinpost) flog = codecs.open(nowdir+"\\"+nowtime+".log", 'w',encoding='utf-8') flog.write(weixinpost) flog.close() rtsp.json yushi指宇视品牌摄像头，haikang指海康威视{ "haikang" : ":554/h264/ch1/main/av_stream", "yushi" : ":554/video1", "rtsp" : [ {"ip" : "172.16.1.1","hostname" : "摄像头1","brand" : "yushi","name": "admin","password": "admin"}, {"ip" : "172.16.1.2","hostname" : "摄像头2","brand" : "yushi","name": "admin","password": "admin"}, {"ip" : "172.16.1.3","hostname" : "摄像头3","brand" : "haikang","name": "admin","password": "admin"} ] }总结 简单

python编译exe可执行文件 安装pyinstaller模块在终端运行pyinstaller -F -w 自己的py名字.py -F 打包为1个文件 -w 隐藏窗口模式

RHEL8+Graylog5.1日志监控系统快速部署接入 本教程适用于RHEL8/Oracle linux8/ALMA Linux8操作系统，当前组件版本信息为：java-17-openjdk-17.0.7.0.7-3.0.1.el8.x86_64 opensearch-2.8.0-1.x86_64 mongodb-org-database-6.0.8-1.el8.x86_64 graylog-server-5.1.3-1.x86_64建议 装一台Alma8服务器，安装graylog，关闭selinux，保持系统更新。全部采用yum源安装，方便快速更新我防火墙仅开放了9000/TCP 1514/UDP 和 SSH端口官方安装教程 https://go2docs.graylog.org/5-1/downloading_and_installing_graylog/red_hat_installation.htm安装OpenJdk17yum install java-17-openjdk安装MongoDB 创建yum源文件/etc/yum.repos.d/mongodb-org.repo[mongodb-org-6.0] name=MongoDB Repository baseurl=https://repo.mongodb.org/yum/redhat/$releasever/mongodb-org/6.0/x86_64/ gpgcheck=1 enabled=1 gpgkey=https://www.mongodb.org/static/pgp/server-6.0.asc安装并启动sudo yum install -y mongodb-org sudo systemctl daemon-reload sudo systemctl enable mongod sudo systemctl start mongod sudo systemctl status mongod安装OpenSearchsudo curl -SL https://artifacts.opensearch.org/releases/bundle/opensearch/2.x/opensearch-2.x.repo -o /etc/yum.repos.d/opensearch-2.x.repo sudo yum install -y opensearch配置OpenSearch端口监听/etc/opensearch/opensearch.ymlcluster.name: graylog node.name: ${HOSTNAME} path.data: /var/lib/opensearch path.logs: /var/log/opensearch discovery.type: single-node network.host: 0.0.0.0 action.auto_create_index: false plugins.security.disabled: true修改内存占用/etc/opensearch/jvm.options-Xms6g -Xmx8g修改内核参数sudo sysctl -w vm.max_map_count=262144 sudo echo 'vm.max_map_count=262144' >> /etc/sysctl.conf启动程序sudo systemctl daemon-reload sudo systemctl enable opensearch sudo systemctl start opensearch sudo systemctl status opensearch安装Graylogsudo rpm -Uvh https://packages.graylog2.org/repo/packages/graylog-5.1-repository_latest.rpm sudo yum install graylog-server修改配置/etc/graylog/server/server.conf增加password_secret和root_password_sha2选项，不增加这两项不允许访问，配置http_bind_address，默认只监听127.0.0.1，修改为0.0.0.0取得root_password_sha2值echo -n "Enter Password: " && head -1 </dev/stdin | tr -d '\n' | sha256sum | cut -d" " -f1取得password_secret值< /dev/urandom tr -dc A-Z-a-z-0-9 | head -c${1:-96};echo;启动程序sudo systemctl daemon-reload sudo systemctl enable graylog-server.service sudo systemctl start graylog-server.service sudo systemctl --type=service --state=active | grep graylog登录系统 默认为http://IP:9000/用户名和密码均在server.conf中定义启动Syslog UDP监听器 注意：graylog无法监听1024以内端口，因此Syslog将使用1514UDP端口 graylog - system -inputsH3C交换机快速接入info-center loghost 172.16.0.1 port 1514浪潮BMC快速接入 群晖快速接入 预览 问题处理 时间不对，修改时区 server.confroot_timezone = Asia/Shanghai总结 为什么不用ELK？ 你研究ELK？ 等你研究出来马斯克都上火星了！ 人生苦短，我用Graylog！

0基础上手python3编程，通用Mysql数据库字段监控，企业微信告警 架构图 通过redis缓存mysql数据库字段进行对比，当发现内容变动时，进行企业微信告警预览 系统组成 由两部分组成 check.py和redis数据库，从mysql中读取指定字段存储到redis中代码import requests,redis import mysql.connector def post_weixin(data): url = 'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=用自己的' body = { "msgtype": "news", "news": { "articles": [ { "title": "卡号监控机器人", "description": data, "url": "90apt.com", "picurl": "https://www.zkteco.com/cn/uploads/image/20210301/3e1adaa2dce94812e658c5d42afc1525.png" } ] } } headers = {"Content-Type": "application/json"} response = requests.post(url, json=body) print(response.text) print(response.status_code) def sqlread(): mqdb = mysql.connector.connect( host="127.0.0.1", user="admin", passwd="password", database="mysqldata" ) mqcursor = mqdb.cursor() getconfig_sql = "SELECT number,card,name FROM userinfo where ifnull(name, '') <> ''" mqcursor.execute(getconfig_sql) mqconfig = mqcursor.fetchall() global cardchange cardchange = "" readredis = redis.Redis(connection_pool=redis.ConnectionPool(host="127.0.0.1", port="6379", password="password",decode_responses=True)) for i in mqconfig: if readredis.get(i[0]) == str(i[1]): None else: cardchange = cardchange + i[2]+readredis.get(i[0])+"变为"+str(i[1])+"\n" readredis.set(i[0], str(i[1])) sqlread() post_weixin(cardchange)总结 简单

RHEL8通过ODBC链接SQL Server数据库 安装完SQLServer数据库，开启数据库远程访问配置对IP的监听配置防火墙端口开放测试端口RHEL8系统安装ODBC驱动https://learn.microsoft.com/en-us/sql/connect/odbc/linux-mac/installing-the-microsoft-odbc-driver-for-sql-server?view=sql-server-ver16&tabs=redhat18-install%2Calpine17-install%2Cdebian8-install%2Credhat7-13-install%2Crhel7-offlinesudo su #Download appropriate package for the OS version #Choose only ONE of the following, corresponding to your OS version #RHEL 7 and Oracle Linux 7 curl https://packages.microsoft.com/config/rhel/7/prod.repo > /etc/yum.repos.d/mssql-release.repo #RHEL 8 and Oracle Linux 8 curl https://packages.microsoft.com/config/rhel/8/prod.repo > /etc/yum.repos.d/mssql-release.repo #RHEL 9 curl https://packages.microsoft.com/config/rhel/9.0/prod.repo > /etc/yum.repos.d/mssql-release.repo exit sudo yum remove unixODBC-utf16 unixODBC-utf16-devel #to avoid conflicts sudo ACCEPT_EULA=Y yum install -y msodbcsql18 # optional: for bcp and sqlcmd sudo ACCEPT_EULA=Y yum install -y mssql-tools18 echo 'export PATH="$PATH:/opt/mssql-tools18/bin"' >> ~/.bashrc source ~/.bashrc # optional: for unixODBC development headers sudo yum install -y unixODBC-devel配置/etc/odbc.inihttps://learn.microsoft.com/zh-cn/sql/connect/odbc/linux-mac/connection-string-keywords-and-data-source-names-dsns?view=sql-server-ver16[TEST] Driver = ODBC Driver 18 for SQL Server # Server = [protocol:]server[,port] Server = tcp:172.16.10.102,1433 TrustServerCertificate=YES # # Note: # Port isn't a valid keyword in the odbc.ini file # for the Microsoft ODBC driver on Linux or macOS 测试，连接成功isql -v TEST user passwd +---------------------------------------+ | Connected! | | | | sql-statement | | help [tablename] | | echo [string] | | quit | | | +---------------------------------------+ SQL>

0基础上手python3编程，批量自动备份交换机配置并进行企业微信通知 架构图 架构图懒得画了，通过telnet备份交换机配置并通知备份结果，展示失败列表；根据命令不同，可以自定义修改用于任意品牌交换机的配置备份。预览 系统组成 由两个文件组成 swtelnet.py config.json代码 由于代码在Windows平台编写，如果在linux平台运行，请把路径中的\换成/swtelnet.pyimport os.path,telnetlib,time,requests,json,codecs def get_config(): config = json.loads(open(path1+"\config.json", encoding='utf-8').read()) return config def gettelnet_config(): nowtime = time.strftime("%Y%m%d", time.localtime()) try: os.mkdir(path1 + "\\" +nowtime) except: None nowdir = (path1 + "\\" +nowtime) for host in config['ip']: try: tn = telnetlib.Telnet(host['host'], timeout=5) tn.write(config['user'].encode('ascii') + b'\n') time.sleep(3) tn.write(config['passwd'].encode('ascii') + b'\n') time.sleep(3) tn.write(b'screen-length disable\n') time.sleep(1) tn.write(b'dis cur\n') time.sleep(1) tn.write(b'undo screen-length disable\n') time.sleep(1) tn.write(b'quit\n') mac1 = tn.read_all() f1 = open(nowdir+"\\"+host['host']+".conf", 'wb') f1.write(mac1) f1.close() flog.write("finish " + host['hostname'] + " " + host['host']+"\n") except: flog.write("fail " + host['hostname'] + " " + host['host']+"\n") def count_sw(): lines = open(path1 + "\\backup.log", "r", encoding='utf-8').readlines() finish = 0 fail = 0 total = 0 failprint = "" for line in lines: line1 = line.split() total = total + 1 if line1[0] == "fail": failprint = failprint + str(line) fail = fail + 1 else: finish = finish + 1 return "交换机备份报告\n总计"+str(total)+" 完成"+str(finish)+" 失败"+str(fail)+"\n备份失败清单:\n"+failprint class network: def yiyan(): try: url = 'https://v1.hitokoto.cn/?c=d&c=k' response = requests.get(url) res = json.loads(response.text) text1 = res['hitokoto'] if res['from'] == None: text2 = "" else: text2 = res['from'] if res['from_who'] == None: text3 = "" else: text3 = res['from_who'] return text1 + " " + text2 + " " + text3 + "\n\n" except: return "一言API故障\n\n" def tianqi(): try: response2 = requests.get(config['weatherapi']) data1 = json.loads(response2.text) data2 = json.dumps(data1['now']) data2 = json.loads(data2) data3 = "环境温度" + data2['temp'] + " 体感温度" + data2['feelsLike'] + " 天气状况 " + data2[ 'text'] + "\n风向 " + data2['windDir'] + " 风力等级" + data2['windScale'] + " 风速" + data2[ 'windSpeed'] + " 湿度" + data2['humidity'] + " 能见度" + data2['vis'] + "公里\n\n" return data3 except: return "天气API故障\n\n" def post_weixin(stats): url = 'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=用自己的' body = { "msgtype": "news", "news": { "articles": [ { "title": "AI智慧运维", "description": network.tianqi()+network.yiyan()+stats, "url": "90apt.com", "picurl": "用自己的" } ] } } response = requests.post(url, json=body) print(response.text) print(response.status_code) path1=os.path.abspath('.') config = get_config() flog = codecs.open(path1+"\\backup.log", 'w',encoding='utf-8') gettelnet_config() flog.close() network.post_weixin(count_sw())config.json{ "user" : "用自己的账号", "passwd" : "用自己的密码", "weatherapi" : "https://devapi.qweather.com/用自己的API", "ip" : [ {"host" : "172.16.1.6","hostname" : "仓库6"}, {"host" : "172.16.1.5","hostname" : "仓库5"} ] }总结 简单

一种通过DHCP+DNS服务器对内网DHCP客户端进行定向柔性非侵入式劫持的方法 架构图 在我的环境中，由于没有实施严格的网络准入配置，且前期没有全面覆盖EDR，导致出现了一种神奇的现象，态势感知系统检测到某IP中毒，查找主机名资产中查不到，并且主机只在晚上上线，无法定位物理机的实际位置；如果直接断网，可能会引起客户投诉，那么有没有一种柔性的方法解决这个非技术问题呢？有的；在本方法中，可以通过DHCP绑定，固定目标客户端请求的IP，固定后单独设置DNS服务器地址，并且自行架设DNS服务器，将所有请求进行重定向；当目标客户端上网时，任意网站请求均会重定向至拦截页面，拦截页面为http服务器+自定义html页面，方便用户联系处理。预览 目标客户端访问任意网站，均出现拦截提醒。配置 配置DNS服务器https://90apt.com/4048配置DHCP服务器1、为目标保留指定IP2、为目标保留的IP配置独立DNS，目标再次DHCP续约时，DNS地址将被替换3、配置HTTP服务器，监听*域名，同时监听https端口总结 简单实用，用户抵触度低。

Python 实现DNS服务器(Python域名解析服务器) 版权声明：本文为CSDN博主「RobinTang」的原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接及本声明。原文链接：https://blog.csdn.net/trbbadboy/article/details/8093256''' Created on 2012-10-15 @author: RobinTang ''' import socketserver import struct # DNS Query class SinDNSQuery: def __init__(self, data): i = 1 self.name = '' while True: d = data[i] if d == 0: break; if d < 32: self.name = self.name + '.' else: self.name = self.name + chr(d) i = i + 1 self.querybytes = data[0:i + 1] (self.type, self.classify) = struct.unpack('>HH', data[i + 1:i + 5]) self.len = i + 5 def getbytes(self): return self.querybytes + struct.pack('>HH', self.type, self.classify) # DNS Answer RRS # this class is also can be use as Authority RRS or Additional RRS class SinDNSAnswer: def __init__(self, ip): self.name = 49164 self.type = 1 self.classify = 1 self.timetolive = 190 self.datalength = 4 self.ip = ip def getbytes(self): res = struct.pack('>HHHLH', self.name, self.type, self.classify, self.timetolive, self.datalength) s = self.ip.split('.') res = res + struct.pack('BBBB', int(s[0]), int(s[1]), int(s[2]), int(s[3])) return res # DNS frame # must initialized by a DNS query frame class SinDNSFrame: def __init__(self, data): (self.id, self.flags, self.quests, self.answers, self.author, self.addition) = struct.unpack('>HHHHHH', data[0:12]) self.query = SinDNSQuery(data[12:]) def getname(self): return self.query.name def setip(self, ip): self.answer = SinDNSAnswer(ip) self.answers = 1 self.flags = 33152 def getbytes(self): res = struct.pack('>HHHHHH', self.id, self.flags, self.quests, self.answers, self.author, self.addition) res = res + self.query.getbytes() if self.answers != 0: res = res + self.answer.getbytes() return res # A UDPHandler to handle DNS query class SinDNSUDPHandler(socketserver.BaseRequestHandler): def handle(self): data = self.request[0].strip() dns = SinDNSFrame(data) socket = self.request[1] namemap = SinDNSServer.namemap if(dns.query.type==1): # If this is query a A record, then response it name = dns.getname(); if namemap.__contains__(name): # If have record, response it dns.setip(namemap[name]) socket.sendto(dns.getbytes(), self.client_address) elif namemap.__contains__('*'): # Response default address dns.setip(namemap['*']) socket.sendto(dns.getbytes(), self.client_address) else: # ignore it socket.sendto(data, self.client_address) else: # If this is not query a A record, ignore it socket.sendto(data, self.client_address) # DNS Server # It only support A record query # user it, U can create a simple DNS server class SinDNSServer: def __init__(self, port=53): SinDNSServer.namemap = {} self.port = port def addname(self, name, ip): SinDNSServer.namemap[name] = ip def start(self): HOST, PORT = "0.0.0.0", self.port server = socketserver.UDPServer((HOST, PORT), SinDNSUDPHandler) server.serve_forever() # Now, test it if __name__ == "__main__": sev = SinDNSServer() sev.addname('www.aa.com', '192.168.0.1') # add a A record sev.addname('www.bb.com', '192.168.0.2') # add a A record sev.addname('*', '0.0.0.0') # default address sev.start() # start DNS server # Now, U can use "nslookup" command to test it # Such as "nslookup www.aa.com"

0基础上手python编程，企业微信值班提醒助手 架构图 通过数据库查询并推送值班人员信息，其中allday字段代表24小时值班，24小时值班时8点和17点均进行推送，不填写此字段时只在17点推送。预览 值班人员提醒人员未维护提醒数据库维护，按此格式维护renyuan表zhibanbiao表系统组成 zhibantouch.py 初始化数据库zhiban.py 主程序初始化数据库 首先手动创建库和用户，通过zhibantouch.py初始化，生成表import mysql.connector mqdb = mysql.connector.connect( host="127.0.0.1", user="zhiban", passwd="passwd", database="zhiban" ) mqcursor = mqdb.cursor() mqcursor.execute("CREATE TABLE renyuan (id INT AUTO_INCREMENT PRIMARY KEY,name VARCHAR(255), phone VARCHAR(255))charset=utf8mb4") mqcursor.execute("CREATE TABLE zhibanbiao (id INT AUTO_INCREMENT PRIMARY KEY, time VARCHAR(255), name VARCHAR(255), allday VARCHAR(255))charset=utf8mb4") mqdb.commit() mqcursor.close()初始化后的数据库监控主程序通过定时任务运行zhiban.py主程序定时任务，每天的8:01和17:01执行脚本1 8,17 * * * python3 /root/zhibancheck/zhiban.py 程序代码import requests,time import mysql.connector def check_user(): nowtime = time.strftime("%Y-%m-%d", time.localtime()) mqdb = mysql.connector.connect( host="127.0.0.1", user="zhiban", passwd="passwd", database="zhiban" ) mqcursor = mqdb.cursor() mqcursor.execute("SELECT * FROM zhibanbiao WHERE time = '%s'" % (nowtime)) if mqcursor.fetchall() == []: return None else: mqcursor.execute( "select renyuan.name,renyuan.phone,zhibanbiao.allday from zhibanbiao inner join renyuan on zhibanbiao.name=renyuan.name where zhibanbiao.time='%s'" % ( nowtime)) info = mqcursor.fetchall() return info def post_weixin(data): url = 'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=用你自己的' body = { "msgtype": "news", "news": { "articles": [ { "title": "值班信息", "description": data, "url": "90apt.com", "picurl": "用你自己的图片" } ] } } headers = {"Content-Type": "application/json"} response = requests.post(url, json=body) print(response.text) print(response.status_code) checkuser = check_user() if time.strftime("%H", time.localtime()) == "8": if checkuser == None: post_weixin("没有维护值班人员信息") elif checkuser[0][2] == "1": post_weixin("今日值班人员 "+checkuser[0][0]+"\n电话："+checkuser[0][1]) else: None elif time.strftime("%H", time.localtime()) == "17": if checkuser == None: post_weixin("没有维护值班人员信息") elif checkuser[0][2] == "1": None else: post_weixin("今晚值班人员 " + checkuser[0][0] + "\n电话：" + checkuser[0][1]) else: print("禁止推送时间")总结 简单

0基础上手python、PHP编程，ActiveMQ监控、报警、查询系统 架构图 通过定时任务采集解析MQ XML数据存储到MYSQL数据库中，当前MQ积累值超过100时，说明消费异常，通过企业微信报警，MQ.php可查询历史记录。预览 告警页面查询页面，显示最后1000行数据系统组成 MQchecktouch.py 初始化数据库MQcheck.py 监控主程序MQ.php 历史记录查询程序初始化数据库 首先手动创建库和用户，通过MQchecktouch.py初始化，生成表import mysql.connector mqdb = mysql.connector.connect( host="127.0.0.1", user="mquser", passwd="mqpasswd", database="mq" ) mqcursor = mqdb.cursor() mqcursor.execute("CREATE TABLE mqdata (id INT AUTO_INCREMENT PRIMARY KEY,time VARCHAR(255), name VARCHAR(255), number VARCHAR(255))") mqcursor.execute("CREATE TABLE configkey (name VARCHAR(255), config VARCHAR(255))") insert_sql = "INSERT INTO configkey (name, config) VALUES ('config','1')" mqcursor.execute(insert_sql) mqdb.commit() mqcursor.close()初始化后的数据库监控主程序 通过定时任务运行主程序import requests,time import xml.etree.cElementTree as ET import mysql.connector def get_mqxml(): mqreq = requests.post(url='http://你的MQ地址:8161/admin/xml/queues.jsp', auth=('admin', 'admin')) mqxml = ET.fromstring(mqreq.content.decode()) mqname_list = [] mqnumbers_list = [] for queue in mqxml.iter('queue'): mqname_list.append(queue.get("name")) mqnumbers_list.append(queue.find('stats').get("size")) return mqname_list,mqnumbers_list def post_weixin(num,mqnumber): if num == 0: data = "6啊，MQ恢复了，当前累计值：" else: data = "G了，MQ当前累计值：" url = 'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=你自己的企业微信机器人' body = { "msgtype": "news", "news": { "articles": [ { "title": "MQ监控机器人", "description": data+str(mqnumber), "url": "90apt.com", "picurl": "你自己的图片" } ] } } headers = {"Content-Type": "application/json"} response = requests.post(url, json=body) print(response.text) print(response.status_code) def save_mysql(): total = 0 mqdb = mysql.connector.connect( host="127.0.0.1", user="mquser", passwd="mqpasswd", database="mq" ) mqcursor = mqdb.cursor() nowtime = time.asctime() for ele in range(0, len(mqnumbers_list)): sql = "INSERT INTO mqdata (time ,name, number) VALUES (%s, %s, %s)" val = (nowtime, mqname_list[ele], mqnumbers_list[ele]) mqcursor.execute(sql, val) total = total + int(mqnumbers_list[ele]) mqdb.commit() print(total) if total > 10: getconfig_sql = "SELECT * FROM configkey WHERE name ='config'" mqcursor.execute(getconfig_sql) mqconfig = mqcursor.fetchall()[0][1] if mqconfig == "0": updateconfig_sql = "UPDATE configkey SET config = '1' WHERE name = 'config'" mqcursor.execute(updateconfig_sql) mqdb.commit() mqcursor.close() post_weixin(1,total) else: None else: getconfig_sql = "SELECT * FROM configkey WHERE name ='config'" mqcursor.execute(getconfig_sql) mqconfig = mqcursor.fetchall()[0][1] if mqconfig == "1": updateconfig_sql = "UPDATE configkey SET config = '0' WHERE name = 'config'" mqcursor.execute(updateconfig_sql) post_weixin(0, total) mqdb.commit() mqcursor.close() else: None mqname_list = get_mqxml()[0] mqnumbers_list = get_mqxml()[1] save_mysql() 历史记录查询程序<?php $con=mysqli_connect("localhost","mquser","mqpasswd","mq"); // 检测连接 if (mysqli_connect_errno()) { echo "连接失败: " . mysqli_connect_error(); } $result = mysqli_query($con,"SELECT * FROM mqdata order by time desc limit 1,1000"); while($row = mysqli_fetch_array($result)) { echo $row['time'] . " " . $row['name'] . " " . $row['number']; echo "<br>"; } $conn = null; ?>