一个系统只有经常维护,才能保证各个功能的正常。这么久过去了,一些功能失效。有的是调用的接口更新了。花了几天时间,再次调试完成。同时优化了一些细节。
原先的IP地址自动更新上游接口失效了。重新适配了一个新的接口
def get_new_ip2(opera, city):
"""
新版本
获取新的监控IP地址
opera:联通,电信,移动
city:节点
"""
ip_dict = {}
ip_arr = []
print(opera, city)
#加载省份字典
pro_dict = load_json('./data/cityinfo2.json')
#获取IP
url = 'https://ip.bczs.net/city/{}'.format(pro_dict[city]['code'])
print(url)
resp = requests.get(url, verify=False)
resp.encoding = 'utf-8'
p_list = etree.HTML(resp.text).xpath('//table/tbody/tr')
skip_num = random.randint(0,int(len(p_list)/2))
i = 0
for p in p_list:
i = i + 1
if i < skip_num:
continue
# 提取目标值
ip_s = p.xpath("./td[1]/a/text()")[0]
ip_e = p.xpath("./td[2]/text()")[0]
loc = p.xpath("./td[4]/text()")[0]
if opera in loc:
(exit_code, aip) = _ping_check_alive(ip_s, ip_e)
if exit_code:
return aip
return False
发现了一个新的工具nexttrace,可以生成可视化图
一键测试脚本get_nexttrace.sh,安装测试并返回结果
#!/bin/bash
#节点回程路由测试
WorkDir='/tmp/.nexttrace'
ETC_DIR=`cat ../setting.py | grep 'ETC' | awk -F"'" '{print $2}'`
declare -A dic
dic=([CT]="电信" [CU]="联通" [CM]="移动")
check_nexttrace() {
if [ ! -e "${WorkDir}/nexttrace" ]; then
mkdir -p ${WorkDir} && cd ${WorkDir}
wget --no-check-certificate -qO nexttrace.zip http://dl.itbunan.xyz/nexttrace.zip && unzip nexttrace.zip > /dev/null 2>&1 && chmod +x nexttrace
cd - > /dev/null
fi
}
get_traceroute_detail() {
isp=$1
#电信节点
counts=`cat .${ETC_DIR}/smokeping.json | jq ".smokeping.${isp}.znodes|length" `
for((i=0;i<$counts;i=i+1))
do
#取六条记录
if [[ i -gt 2 ]];then
break;
fi
#节点
title=${dic[$1]}"/"`cat .${ETC_DIR}/smokeping.json | jq ".smokeping.${isp}.znodes[$i].title"|sed 's/\"//g'`
ip=`cat .${ETC_DIR}/smokeping.json | jq ".smokeping.${isp}.znodes[$i].host"|sed 's/\"//g'`
result=`cd ${WorkDir} && ./nexttrace ${ip} | grep -v NextTrace | grep -v LeoMoeAPI |grep -v traceroute`
#输出结果
printf "%s%s\n" "$title" "/路由追踪"
echo "跳数 路由信息"
printf "%s\n" "${result}"
done;
}
run_get_traceroute() {
#定义(CT:电信, CU:联通, CM:移动)
for i in 'CT' 'CU' 'CM'
do
get_traceroute_detail $i
done
}
check_nexttrace
run_get_traceroute
#清理文件
rm -rf ${WorkDir}
调整可视化显示的细节
def load_json_to_html_for_zreport():
"""
加载测试结果,并转化为html
"""
re_list = []
re_dict = load_json('{}/sreport.json'.format(ETC_DIR))
if "result" in re_dict.keys():
for key, item in re_dict['result'].items():
if item:
if key == '路由追踪':
h_str = ''
for k, v in item.items():
t_str = ''
for i in v[:-1]:
t_str = t_str + i +"\n"
urls = re.findall(r'https?://[^\s<>"]+|www\.[^\s<>"]+', v[-1])
h_str += '{}-在线报告{}'.format(k, urls[0], t_str)
html_str = '{}:{}'.format(key, h_str)
elif key == 'Geekbench跑分':
url = item.pop('Full Test')
table_str = json2html.convert(json = item)
html_str = '{}-完整报告{}'.format(key, url, table_str)
else:
#table = json2html.convert(json = item, table_attributes='cellspacing=\"0\"')
table_str = json2html.convert(json = item)
html_str = '{}:{}'.format(key, table_str)
re_list.append(html_str)
else:
html_str = '没有结果,请登录服务器运行测试脚本:python3 sreport.py'
re_list.append(html_str)
return re_dict['node'], re_dict['update_time'], re_list