主要功能是递归查询官网相关接口是否可用，对于不可用的接口，自动进行飞书告警，部分代码调用了其他已实现模块，直接复制下来，不能直接运行，请自行删减
import requests
from bs4 import BeautifulSoup
from urllib.parse import urljoin
import logging as log
from feishu import send_message
from config import get_id_by_name
header = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/86.0.4240.198 Safari/537.36",
"X-Requested-With": "XMLHttpRequest"
}
base_url_list = [
("https://www.test.com/file/UKEY1.pdf", "测试报错"),
("https://www.test.com/file/UKEY2.pdf", "测试报错2")
]
url_list = []
error_url_list = []
def get_href(url, text):
"""
递归检查接口
:param url:接口地址
:param text:接口名称
:return:
"""
req = requests.get(url, headers=header)
log.info(f'{url}，返回：{req.status_code}')
if req.status_code != 200:
error_url_list.append(f"{text}报错{req.status_code}，{url}")
soup = BeautifulSoup(req.content, "lxml") # 格式化
for a in soup.findAll('a'): # 查找所有a元素
log.debug(a)
if a.get('href'): # 存在没有链接的元素
new_url = urljoin(url, a['href']) # 拼接url
if new_url not in url_list: # 避免重复
url_list.append(new_url)
get_href(new_url, f'{text}-{a.getText()}') # 递归查找
else:
log.debug(f"重复的地址:{new_url}")
else:
log.debug(f"没有链接地址:{a}")
def format_list_info(info_list):
"""
格式化成飞书机器人消息所需格式
:param info_list:
:return:
"""
if not info_list:
return [[{"tag": "text", "text": "所有接口正常"}]]
result = [[{"tag": "at", "user_id": get_id_by_name("测试人")}]]
for temp in info_list:
result.append([{"tag": "text", "text": temp}])
return result
def main():
# 初始化接口列表
for temp_url, temp_text in base_url_list:
url_list.append(temp_url)
# 接口检查
for temp_url, temp_text in base_url_list:
get_href(temp_url, temp_text)
# 异常接口展示
for temp in error_url_list:
log.error(temp)
# 飞书告警提示
send_message()
if __name__ == '__main__':
import mylog
main()
