免费通过 NS1 利用监控宝平台实现实时基于不同运营商的故障转移

1
import requests
2
# key为你的ns1 feed label；value为下载asn列表用到的url的一部分，一般情况下不需要修改
3
asn_dict = {
4
    'dianxin':'chinanet',
5
    'yidong':'cmcc',
6
    'liantong':'unicom'
7
    }
8

9
ns1_api_key = ''
10
source_id = ''
11
ns1_req_url = f'https://api.nsone.net/v1/feed/{source_id}'
12
headers = {
13
    "X-NSONE-Key":f"{ns1_api_key}"
14
}
15

16
for ns1_asn_label, gh_asn in asn_dict.items():
17
    url_to_down = f'https://raw.githubusercontent.com/xingpingcn/china-mainland-asn/main/asn_txt/{gh_asn}.txt'
18
    response = requests.get(url_to_down).text
19
    asn_list_from_gh = response.split('\n')[1].split(',')
20
    data = {
21
        f"{ns1_asn_label}":{"asn":asn_list_from_gh}
22
    }
23
    res = requests.post(ns1_req_url,headers=headers,json=data)
24
    print(res.status_code)

1
import requests
2

3
status = True
4

5
source_id = ''
6
label=''
7
ns1_api_key = ''
8
url = f'https://api.nsone.net/v1/feed/{source_id}'
9
headers = {
10
    "X-NSONE-Key":f"{ns1_api_key}"
11
}
12
data = {
13
    f"{label}":{"up":status}
14
}
15
res = requests.post(url,headers=headers,json=data)

1
import requests
2
import base64
3
import hashlib
4
import json
5

6
source_id = ''
7
label=''
8
status = True
9
ns1_api_key = ''
10

11
url_token = '' #回调token
12

13
def g_md5(content):
14
    md5hash = hashlib.md5(content)
15
    md5 = md5hash.hexdigest()
16
    return md5
17

18
def lambda_handler(event, context):
19
    json_data = json.loads(event['body'])
20
    verify = str(json_data['msg_id'])+str(json_data['task_id'])+str(json_data['fault_time'])+url_token
21
    md5 = g_md5(verify.encode('utf-8'))
22
    if md5==json_data['token']:
23
        if json_data['message_status'] == 1 or json_data['message_status'] == '1':
24
            status = False
25
        elif json_data['message_status'] == 2 or json_data['message_status'] == '2':
26
            status = True
27
        url = f'https://api.nsone.net/v1/feed/{source_id}'
28
        headers = {
29
            "X-NSONE-Key":f"{ns1_api_key}"
30
        }
31
        data = {
32
        f"{label}":{"up":status}
33
        }
34
        res = requests.post(url,headers=headers,json=data)
35
        print(res.text,f'[status] \'up\': {status}')

1
import json
2
import requests
3

4
source_id = ''
5
ns1_api_key = ''
6

7
def lambda_handler(event, context):
8
    # print(event,type(event))
9
    json_data = json.loads(event['body'])
10
    if json_data['isp']  == 'telecom':
11
        label = 'dianxin' # ns1的feed label，下同
12
    elif json_data['isp']  == 'unicom':
13
        label = 'liantong'
14
    elif json_data['isp']  == 'cmcc':
15
        label = 'yidong'
16
    if json_data['df_status'] == 'critical':
17
        status = False
18
    else:
19
        status = True
20
    url = f'https://api.nsone.net/v1/feed/{source_id}'
21
    headers = {
22
        "X-NSONE-Key":f"{ns1_api_key}"
23
    }
24
    data = {
25
    f"{label}":{"up":status}
26
    }
27
    res = requests.post(url,headers=headers,json=data)
28
    print(res.text,f'[status] {label} -> \'up\': {status}')