基于Python+Flask+Echarts的全国疫情监控系统源码+项目说明（疫情数据收集通过网络爬虫技术爬取）.zip

# -*- coding: utf-8 -*- import time,datetime, json, requests,pymysql import pandas as pd import traceback from selenium.webdriver import Chrome, ChromeOptions import sys # ----------------数据库连接、关闭------------------------ #连接数据库 def get_conn(): #建立连接 connect = pymysql.Connect( host='localhost', port=3306, user='root', passwd='123456', db='cov', charset='utf8' ) #获取游标 cursor = connect.cursor() return connect,cursor #关闭连接 def close_conn(connect,cursor): if connect: connect.close() if cursor: cursor.close() # ----------------爬取数据------------------------ # 抓取腾讯疫情国内每日实时详细各省市和中国每日历史数据 def get_tencent_data(): url1 = 'https://view.inews.qq.com/g2/getOnsInfo?name=disease_h5&callback=&_=%d'%int(time.time()*1000) url2 = 'https://view.inews.qq.com/g2/getOnsInfo?name=disease_other&callback=&_=%d'%int(time.time()*1000) headers = { 'user-agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/78.0.3904.70 Safari/537.36' } r1 = requests.get(url1, headers) r2 = requests.get(url2, headers) #json字符串转字典 res1 = json.loads(r1.text) res2 = json.loads(r2.text) data_all1 = json.loads(res1["data"]) data_all2 = json.loads(res2["data"]) #当日详细数据 details = [] update_time = data_all1["lastUpdateTime"] data_country = data_all1["areaTree"] data_province = data_country[0]["children"] for pro_infos in data_province: province = pro_infos["name"] for city_infos in pro_infos["children"]: city = city_infos["name"] confirm = city_infos["total"]["confirm"] confirm_add = city_infos["today"]["confirm"] nowConfirm = city_infos['total']['nowConfirm'] suspect = city_infos["total"]["suspect"] heal = city_infos["total"]["heal"] dead = city_infos["total"]["dead"] dead_rate = city_infos['total']['deadRate'] heal_rate = city_infos['total']['healRate'] details.append([update_time, province, city,nowConfirm, confirm, confirm_add, suspect,heal, dead,dead_rate,heal_rate]) #历史数据 history = {} for day_infos in data_all2["chinaDayList"]: ds = day_infos["y"]+"."+day_infos["date"] tup = time.strptime(ds, "%Y.%m.%d") # 匹配时间 ds = time.strftime("%Y-%m-%d", tup) #改变时间输入格式，不然插入数据库会报错，数据库是datatime格式 confirm = day_infos["confirm"] suspect = day_infos["suspect"] heal = day_infos["heal"] dead = day_infos["dead"] nowConfirm = day_infos["nowConfirm"] nowSevere = day_infos["nowSevere"] importedCase = day_infos["importedCase"] noInfect = day_infos["noInfect"] localConfirm = day_infos["localConfirm"] dead_rate = day_infos["deadRate"] heal_rate = day_infos["healRate"] history[ds] = {"confirm":confirm, "suspect":suspect, "heal":heal, "dead":dead, "importedCase": importedCase, "noInfect": noInfect, "localConfirm":localConfirm, "nowConfirm":nowConfirm, "nowSevere":nowSevere, "dead_rate":dead_rate, "heal_rate":heal_rate} for day_infos in data_all2["chinaDayAddList"]: ds = day_infos["y"]+"."+day_infos["date"] tup = time.strptime(ds, "%Y.%m.%d") # 匹配时间 ds = time.strftime("%Y-%m-%d", tup) #改变时间输入格式，不然插入数据库会报错，数据库是datatime格式 confirm = day_infos["confirm"] suspect = day_infos["suspect"] heal = day_infos["heal"] dead = day_infos["dead"] importedCase = day_infos["importedCase"] noInfect = day_infos["infect"] dead_rate = day_infos["deadRate"] heal_rate = day_infos["healRate"] localConfirm = day_infos["localConfirmadd"] history[ds].update({"confirm_add":confirm, "suspect_add":suspect, "heal_add":heal, "dead_add":dead, "importedCase_add": importedCase, "noInfect_add": noInfect, "localConfirm_add":localConfirm, "dead_rate_add":dead_rate, "heal_rate_add":heal_rate}) return history,details # 抓取各省从2020到2021的每日历史数据（无市区） def get_province_history_data(): headers = { "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/84.0.4147.135 Safari/537.36" } url = "http://111.231.75.86:8000/api/provinces/CHN/daily/" response = requests.get(url=url, headers=headers) res = json.loads(response.text) details = [] for infos in res: ds = datetime.datetime.strptime(str(infos["dateId"]), '%Y%m%d').strftime('%Y-%m-%d') province = infos["provinceName"] province_code = infos["provinceCode"] nowConfirm = infos["currentConfirmedCount"] nowConfirm_add = infos["currentConfirmedIncr"] confirm = infos["confirmedCount"] confirm_add = infos["confirmedIncr"] suspect = infos["suspectedCount"] suspect_add = infos["suspectedCountIncr"] heal = infos["curedCount"] dead = infos["deadCount"] heal_add = infos["curedIncr"] dead_add = infos["deadIncr"] nowSevere = infos["highDangerCount"] nowMidSevere = infos["midDangerCount"] details.append( [ds, province,province_code, confirm, confirm_add, nowConfirm, nowConfirm_add, suspect, suspect_add, heal, heal_add, dead, dead_add, nowSevere, nowMidSevere]) return details # 抓取本土风险划分数据 def get_localrisk_data(): url = 'https://view.inews.qq.com/g2/getOnsInfo?name=disease_other&callback=&_=%d' % int(time.time() * 1000) headers = { 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/78.0.3904.70 Safari/537.36' } r = requests.get(url, headers) res = json.loads(r.text) data_all = json.loads(res["data"]) locallist = [] for local in data_all["statisGradeCityDetail"]: ds = str(local["syear"]) + "/" + local["date"] tup = time.strptime(ds, "%Y/%m/%d") # 匹配时间 ds = time.strftime("%Y-%m-%d", tup) # 改变时间输入格式，不然插入数据库会报错，数据库是datatime格式 province = local["province"] city = local["city"] nowConfirm = local["nowConfirm"] confirm = local["confirm"] confirm_add = local["confirmAdd"] heal = local["heal"] dead = local["dead"] grade = local["grade"] locallist.append([ds, province,city,nowConfirm,confirm,confirm_add,heal,dead,grade]) return locallist #抓取全球各国以及美国各洲最新的数据 def get_global_country_latest_data(): url = 'https://view.inews.qq.com/g2/getOnsInfo?name=disease_foreign&callback=&_=%d' % int(time.time() * 1000) url2 = "https://api.inews.qq.com/newsqa/v1/automation/foreign/country/ranklist" headers = { 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/78.0.3904.70 Safari/537.36' } s = requests.session() s.keep_alive = False requests.DEFAULT_RETRIES = 5 # 各国各城市数据 details = [] america = [] #获取美国数据 r1 = requests.get(url, headers) res1 = json.loads(r1.text) data_all = json.loads(res1["data"]) # 获取全球数据 r2 = requests.post(url=url2, headers=headers) res2 = json.loads(r2.text) # print(res["data"]) for infos in res2["data"]: ds = infos["y"] + "." + infos["date"] country = infos["name"] continent = infos["continent"] nowConfirm = infos["nowConfirm"]