疫情风险等级爬虫小记

近日看到国务院印发的《全国不同风险地区企事业单位复工复产疫情防控措施指南》中提到了低风险，中风险和高风险地区，搜索发现只能通过“国务院客户端微信小程序”以区/县为单位进行查询，想了解一下全国各地的形式需要挨个查询，没有一个可以纵览全局的地图，心血来潮想要自己画一个。

数据源

网上常见的查询入口都是微信小程序，但是微信小程序相对封闭，咱也不属于数据获取相对不方便，想找个普通的网站服务。经过一番搜索发现国家政务服务平台也提供了一个区/县疫情风险等级查询入口，开干。

点击进入，虎躯一震，查询一个公开的疫情风险信息竟然还要先注册账户，政府网站的账户不出以外的要把身份信息扒干净。算了，麻利填完，查一个试试。

先查个武汉

再查个北京

接口分析

初步判断应该只需要一个地区的编码或者名字就可以查询了，上 F12 一看果然是，post方法需要xzqhdm和Xzqhmc两个参数（BizType应该是固定的查询类别，先不管）。

function query(city,cityName){
            $.ajax({
                url: '/fwmh/healthCode/front/healthQuery.do',
                data: {BizType: "0117", Xzqhdm: city, Xzqhmc: cityName},
                type: "post",
                crossDomain: true,
               ...
            })
}

翻一翻select控件看看参数要怎么填，应该就是身份证开头的那几位行政区划代码了，名称的格式是省/市/区（不过最后发现了名称参数没啥用，只要代码对了就行）。

行政区划代码可以找现成的，也可以从这个网站上爬下来，为了保证数据一致避免不必要的麻烦，也从网站上爬吧。找到相关查询行政区划代码的脚本，发现是通过一定的层级结构查询的，以000000作为pid可以查到所以省，然后再以省的代码查到相应的市，以此类推。

$(function () {
    $.ajax({
        url: '/fwmh/healthCode/HealthCodeAreaBypParentCode.do',
        type: "post",
        crossDomain: true,
        data: {code: "000000"},
        ...
    });
});

编写爬虫

接口和参数都清楚了，开始写爬虫。导入requests包，定义两个查询接口的地址。查询需要登录，因此还要从浏览器中找到cookies，转换成键值对的格式作为后续请求接口的参数。

import requests

risk_url = "http://gjzwfw.www.gov.cn/fwmh/healthCode/front/healthQuery.do"
area_url = "http://gjzwfw.www.gov.cn/fwmh/healthCode/HealthCodeAreaBypParentCode.do"

raw_cookies = 'xxxx'

def cookies_convert(raw_cookies):
    cookies = {}
    for line in raw_cookies.split(';'):
        key, value = line.split('=', 1)  # 1代表只分一次，得到两个数据
        cookies[key] = value
    return cookies

cookies = cookies_convert(raw_cookies)

查询行政区划代码并保存在’area_map.txt’文件中。area_query负责单次查询的处理，因为网络等各种问题，很可能返回无法正确解析的内容，因此要加一个异常处理，无法解析就重试一次。当然也要限制一下重试的次数，避免掉进死循环。

因为有省/市/区三级，所以获取所有区划代码的get_area_map函数嵌套了好几层，其实也可以通过写成递归的形式简化代码，不过这里就先将就一下吧。

用time.sleep设置一个随机的查询间隔，避免被封。

每查一个就要保存一下，不然遇到网络问题程序崩溃就GG了。实践中还发现这个系统的数据有点问题，有一个压根不存在的行政区划代码，相应的名称为空。为此加一行判断，代码或名称为空的就跳过。

area_map_file = 'area_map.txt'

def area_query(area_code, retry=0):
    if retry == 5:
        print('Retry too many times! Area code: ', area_code)
        return None
    area_resp = requests.post(area_url, {'code': area_code}, cookies=cookies)
    try:
        data = json.loads(area_resp.text)['result']
    except json.decoder.JSONDecodeError:
        data = area_query(area_code, retry+1)
    return data

def get_area_map():
    area_map = {}
    with open(area_map_file, 'w') as f:
        area_code = '000000'
        prov_data = area_query(area_code)
        for povince in prov_data:
            prov_name = povince['region_name']
            prov_code = povince['region_code']
            city_data = area_query(prov_code)
            for city in city_data:
                city_name = city['region_name']
                city_code = city['region_code']
                # 加一个延时，避免查询频率过高被封
                time.sleep(random.random())
                country_data = area_query(city_code)
                for country in country_data:
                    country_name = country['region_name']
                    country_code = country['region_code']
                    # 代码或名称为空时跳过
                    if not (country_name and country_code):
                        continue
                    area_map[country_code] = prov_name + \
                        '/'+city_name+'/'+country_name
                    print(country_code, area_map[country_code])
                    f.write(country_code+'\t'+area_map[country_code]+'\n')
    return area_map

根据行政区划代码查询风险等级的代码也是类似的。既然这么类似，后续也可以考虑从两个函数中抽象出一个带重试的API访问函数。不过总共也才两个函数，也先不优化了，ε=ε=ε=┏(゜ロ゜;)┛

def query_risk(city_code, city_name, retry=0):
    if retry == 5:
        print('Retry too many times! Area code: ', city_code)
        return None
    city_data = {'BizType': '0117',
                 'Xzqhdm': city_code,
                 'Xzqhmc': city_name}
    time.sleep(random.random())             
    risk_resp = requests.post(risk_url, city_data, cookies=cookies)
    try:
        data = json.loads(risk_resp.text)['xzqhWarnLv']['fxdj']
    except:
        print('Retry:', city_code, city_name)
        data = query_risk(city_code,city_name, retry+1)
    return data

函数都定义完了，开始给我爬。

1. 先判断保存行政区划代码的文件是否存在，不存在就重新爬一下
2. 读取行政区划代码用于查询风险等级
3. 读取已经查询过的数据，避免重复查询，如果没有历史数据则写入表头
4. 查询并保存各地区的风险水平，有些地方没有提供数据，用-1代替

# 查询行政区划代码和并称并保存
if not os.path.exists(area_map_file):
    print('Area code file does not exist, uncomment the line below to get it.')
    # area_map = get_area_map()

# 读取行政区划代码用于查询风险等级
with open(area_map_file, 'r') as f:
    areas = f.readlines()

# 读取已经查询过的数据，避免重复查询
with open('risk.csv', 'r+', encoding='utf-8') as csvfile:
    reader = csv.DictReader(csvfile)
    try:
        exist_code = set([row['code'] for row in reader])
    except:
        # 第一次要写入表头
        writer = csv.writer(csvfile)
        writer.writerow(["code","name","risk"])
        exist_code = set()

with open('risk.csv', 'a+', newline='', encoding='utf-8') as csvfile:
    writer = csv.writer(csvfile)
    for i, line in enumerate(areas):
        code, name = line.split()
        # 跳过查询过的地区
        if code in exist_code: continue
        risk = query_risk(code, name)
        # 使用 -1 表示未知
        if risk == 'null': risk = -1
        print(str(i)+'/'+str(len(areas)),code,name,risk)
        writer.writerow([code,name,risk])

结果

最终爬到2912个区/县的风险等级，其中24个未知，7个为中风险，剩下的全是低风险。

全国形式一片大好，画出来的地图也就是一只绿公鸡，没啥必要，就TJ了，ε=ε=ε=┏(゜ロ゜;)┛

不过还是简单调研了一下，地图绘制相关的包，可以用下面两种方案，下回可以画点其他的。

• matplotlib + basemap
• pyechart.Map