上海疫情数据及源代码实现

from lxml import etree import requests,re,shelve,pathlib def wash_region_data_item(item): ''' 将文章中提取的每个区上报的数据,格式进行清洗及统一,因为发现16个区上报的数据格式并不太一致 ''' #首先将返回的每个区内的每个上报数据，统一替换符号 if '居住于' in item: item=item.replace('和',',') else: item=item.replace('。',',').replace('、',',').replace('，',',') item=item.replace('\xa0','').strip() #然后再对item内，如果没有，,则程序加上这个符号，便于后面统一处理 if ',' not in item: item=item+',' return item def wash_region_data_list(region_data_list): #首先将列表内的每项字符串统一清洗格式化 region_data_list=[wash_region_data_item(item) for item in region_data_list] #然后再对数据列表连接起来，并重新使用,分割成新的列表 if '：' in ''.join(region_data_list): data_list=''.join(region_data_list).split('：') region_data_list=[data_list[0]]+data_list[1].split(',') elif '于' in ''.join(region_data_list): data_list=''.join(region_data_list).split('于') region_data_list=[data_list[0]]+data_list[1].split(',') else: region_data_list=[''.join(region_data_list)] #然后再去除新数据列表内不需要的列表项，因为使用的是pop，为了避免清洗遗漏，故使用了循环 need_wash_flag=True while need_wash_flag: for index,data in enumerate(region_data_list): count=0 #如果为空或包含消毒措施，则移除该列表项 if data == '' or '消毒措施' in data or '来源' in data: region_data_list.pop(index) count+=1 if count==0: need_wash_flag=False return region_data_list def fetch_date_data(top_title): #用于从文章顶部整体信息区域，提取当日日期、总确认、总无症状数据 top_title=top_title current_date=re.search('[0-9]{4,4}年[0-9]{1,2}月[0-9]{1,2}日',top_title).group(0) total_confirm_count=re.search('确诊病例([0-9]*)例',top_title).group(1) total_confirm_count=int(total_confirm_count) if total_confirm_count else 0 total_asymptomatic_count=re.search('感染者([0-9]*)例',top_title).group(1) total_asymptomatic_count=int(total_asymptomatic_count) if total_asymptomatic_count else 0 return (current_date,total_confirm_count,total_asymptomatic_count) def fetch_region_data(region_data_list): region_data_list=region_data_list #开始提取该区当日上报的数据，包括该区当日新增确认、新增无症状、新上报的社区名称列表 if '区' in region_data_list[0]: region_name=re.search('[,，](.*区)',region_data_list[0]).group(1) #有些区的数据就是不规范，有时候有区有时候没有，烦人 elif '青浦' in region_data_list[0]: region_name='青浦区' elif '奉贤' in region_data_list[0]: #专门为2022年3月28日的奉贤区数据进行适配 region_name='奉贤区' region_total_confirm=re.search('([0-9]*)例(本土)?(新冠肺炎)?确诊',region_data_list[0]) region_total_confirm=int(region_total_confirm.group(1)) if region_total_confirm else 0 region_total_asymptomatic=re.search('([0-9]*)例(本土)?无症状',region_data_list[0]) region_total_asymptomatic=int(region_total_asymptomatic.group(1)) if region_total_asymptomatic else 0 #只有该区有上报数据时，才记录，否则，为空 if len(region_data_list) >1: region_community_list=region_data_list[1:] else: region_community_list='' return(region_name,region_total_confirm,region_total_asymptomatic,region_community_list) def wash_item(item): item=item.xpath('span//text()') item=''.join(item) return item def fetch_region_data_before(html): #用于提取4月5日之前的数据 html=html regions_covid_data={} #初始化某天的数据字典 #获取当前整体数据，包括日期、总确认、总无症状 if len(html.xpath('//div[@id="ivs_content"]/p[1]/strong')): top_title=''.join(html.xpath('//div[@id="ivs_content"]/p')[0].xpath('strong//text()')) else: top_title=''.join(html.xpath('//div[@id="ivs_content"]/p')[0].xpath('span//text()')) date_datas=fetch_date_data(top_title) regions_covid_data['current_date']=date_datas[0] regions_covid_data['total_confirm_count']=date_datas[1] regions_covid_data['total_asymptomatic_count']=date_datas[2] #初始化对应各区的数据字典 regions_covid_data['region_coviods_data']={} #获取当日每个区的信息，包括区名称、总确诊、总无症状、上报社区列表 #先对文章内 的数据提取并按区分割 total_content_list=[wash_item(item) for item in html.xpath('//div[@id="ivs_content"]/p')] end_index=None #为了防止文章中出现各区重复内容，故进行去重判断 for index,item in enumerate(total_content_list[1:]): if '市卫健委' in item: end_index=index regions_content=total_content_list[1:end_index] if end_index else total_content_list[1:] split_index=[] for index,item in enumerate(regions_content): #只是判断月日是否在文本内，因为部分区里面的数据不一定包含年 if date_datas[0].replace('2022年','') in item: split_index.append(index) #切割后并按各区存储为列表 region_datas=[] for i in range(len(split_index)-1): region_datas.append(regions_content[split_index[i]:split_index[i+1]]) region_datas.append(regions_content[split_index[-1]:]) datas={} for region_data in region_datas: region_data_list=wash_region_data_list(region_data) fetch_datas=fetch_region_data(region_data_list) region_name=fetch_datas[0] datas[region_name]={} datas[region_name]['region_name']=region_name datas[region_name]['region_total_confirm']=fetch_datas[1] datas[region_name]['region_total_asymptomatic']=fetch_datas[2] datas[region_name]['region_community_list']=fetch_datas[3] #将各区的数据，存储到区数据字段内 regions_covid_data['region_coviods_data']=datas return regions_covid_data def fetch_region_data_after(html): html=html regions_covid_data={} #初始化某天的数据字典 #获取当前整体数据，包括日期、总确认、总无症状 top_title=html.xpath('//section[@data-id="106156"]//span/text()')[0] date_datas=fetch_date_data(top_title) regions_covid_data['current_date']=date_datas[0] regions_covid_data['total_confirm_count']=date_datas[1] regions_covid_data['total_asymptomatic_count']=date_datas[2] #初始化对应各区的数据字典 regions_covid_data['region_coviods_data']={} #然后再提取每个区的具体数据 region_datas=html.xpath('//section[@data-id="72469"]') or html.xpath('//section[@data-id="97598"]') datas={} for region_data in region_datas: #对每个区的数据列表进行清洗 region_data_list=region_data.xpath('section//text()') region_data_list=wash_region_data_list(region_data_list) fetch_datas=fetch_region_data(region_data_list) region_name=fetch_datas[0] datas[region_name]={} datas[region_name]['region_name']=region_name datas[region_name]['region_total_confirm']=fetch_datas[1] datas[region_name]['region_total_asymptomatic']=fetch_datas[2] datas[region_name]['region_community_list']=fetch_datas[3] #将各区的数据，存储到区