本文共 4247 字，大约阅读时间需要 14 分钟。

如何使用Python脚本监控网站状态并通过钉钉机器人发送报警

在实际开发中，需要对网站的状态进行持续监控，确保服务的稳定性。本文将介绍一个基于Python的脚本，能够自动检测网站的状态码，并通过钉钉机器人发送相关报警信息。

1. 脚本概述

本脚本主要包含以下几个部分：

• 数据读取：从hosts.txt文件中读取需要监控的URL列表。
• 状态码检测：通过多线程方式向目标网站发起请求，获取其状态码。
• 状态码判断：对获取到的状态码进行合法性判断，非法状态码将被记录并触发报警。
• 报警处理：通过钉钉机器人发送报警信息，包括URL地址、状态码以及相关评论内容。

2. 代码结构

脚本的核心逻辑主要集中在getStatusCode函数中，该函数负责获取目标网站的状态码，并将结果返回给主线程处理。为了提高效率，脚本采用了线程池的方式，能够同时对多个URL进行状态码检测。

3. 实施步骤

3.1 文件准备

在项目启动前，需要在当前工作目录下创建一个名为hosts.txt的文件。文件内容格式如下：

https://www.baidu.com,百度首页
https://www.trc.com,泰然城首页
https://www.jd.com,京东商城

每一行的格式为：<URL>,<评论内容>

3.2 钉钉机器人配置

在脚本中，钉钉机器人的webhook地址需要手动配置。默认配置如下：

ding_url = 'https://oapi.dingtalk.com/robot/send?access_token=e0bef403aded94c230953384353bc411a7fba57389ebd59bc0e63cc602ec175f'
HEADERS = {
    "Content-Type": "application/json ;charset=utf-8"
}

如需更换钉钉机器人的地址，请在对应位置进行修改。

3.3 脚本运行流程

4. 代码实现

以下是脚本的核心代码实现：

import re
import requests
import json
from threading import Thread, Lock
from concurrent.futures import ThreadPoolExecutor
list1 = []
list2 = []
code_list = [200, 301, 302, 401]
class MyThread(Thread):
    def __init__(self, func, args=()):
        super(MyThread, self).__init__()
        self.func = func
        self.args = args
    def run(self):
        self.result = self.func(*self.args)
    def get_result(self):
        try:
            return self.result
        except Exception:
            return None
def get_url():
    try:
        with open('hosts.txt', 'r', encoding='utf-8') as f:
            data = f.readlines()
        return data
    except Exception:
        return False
def verdictUrl():
    url_list = []
    comment_list = []
    get_url_res = get_url()
    if get_url_res:
        for data in get_url_res:
            url = data.split(',')[0]
            comment = data.split(',')[-1]
            try:
                res = re.search(r'http\w{0,1}://(\w+\.){2}\w+.*', url).group()
                url_list.append(res)
                comment_list.append(comment)
            except Exception:
                print(f'URL:{url} 有误')
        return (url_list, comment_list)
    else:
        print('文件不存在......')
def getStatusCode(url, comment):
    global list1, list2
    try:
        res = requests.head(url)
        if res.status_code in code_list:
            lock.acquire()
            list1.append(res.status_code)
            lock.release()
        except requests.exceptions.ConnectionError:
            status = 0
            lock.acquire()
            list2.append(status)
            lock.release()
        else:
            status = res.status_code
        finally:
            return {'url': url, 'StatusCode': status, 'comment': comment}
    return None
def sendDingDing(bc):
    ding_url = 'https://oapi.dingtalk.com/robot/send?access_token=e0bef403aded94c230953384353bc411a7fba57389ebd59bc0e63cc602ec175f'
    HEADERS = {
        "Content-Type": "application/json ;charset=utf-8"
    }
    bc = bc.result()
    url = bc['url']
    status = bc['StatusCode']
    comment = bc['comment']
    string_textMsg = {
        'msgtype': 'text',
        'text': {
            'content': f'URL地址：{url}\n' 
                     f'URL名称：{comment}\n' 
                     f'状态码：{status}\n'
        }
    }
    string_textMsg = json.dumps(string_textMsg)
    res = requests.post(ding_url, data=string_textMsg, headers=HEADERS)
    return res
if __name__ == '__main__':
    lock = Lock()
    pool = ThreadPoolExecutor(4)
    url, comment = verdictUrl()
    res = zip(url, comment)
    li = []
    for i in res:
        for j in range(4):
            t = MyThread(getStatusCode, (i[0], i[1]))
            li.append(t)
            t.start()
        for t in li:
            t.join()
        if len(list1) > 3 or len(list2) > 3:
            pool.submit(getStatusCode, i[0], i[1]).add_done_callback(sendDingDing)

5. 使用说明

通过以上步骤，您可以轻松实现对多个网站的状态码监控，并通过钉钉机器人接收相关报警信息。