思路：
在騰訊疫情數據網站F12解析網站結構，使用Python爬取當日疫情數據和歷史疫情數據，分別存儲到details和history兩個mysql表。
①此方法用於爬取每日詳細疫情數據
1 import requests 2 import json 3 import time 4 def get_details(): 5 url = 'https://view.inews.qq.com/g2/getOnsInfo?name=disease_h5&callback=jQuery34102848205531413024_1584924641755&_=1584924641756' 6 headers ={ 7 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/70.0.3538.25 Safari/537.36 Core/1.70.3741.400 QQBrowser/10.5.3863.400' 8 } 9 res = requests.get(url,headers=headers) 10 #輸出全部信息 11 # print(res.text) 12 response_data = json.loads(res.text.replace('jQuery34102848205531413024_1584924641755(','')[:-1]) 13 #輸出這個字典的鍵值 dict_keys(['ret', 'data'])ret是響應值，0代表請求成功，data里是我們需要的數據 14 # print(response_data.keys()) 15 """上面已經轉化過一次字典，然后獲取里面的data，因為data是字符串，所以需要再次轉化字典 16 print(json.loads(reponse_data['data']).keys()) 17 結果： 18 dict_keys(['lastUpdateTime', 'chinaTotal', 'chinaAdd', 'isShowAdd', 'showAddSwitch', 19 'areaTree', 'chinaDayList', 'chinaDayAddList', 'dailyNewAddHistory', 'dailyHistory', 20 'wuhanDayList', 'articleList']) 21 lastUpdateTime是最新更新時間，chinaTotal是全國疫情總數，chinaAdd是全國新增數據， 22 isShowAdd代表是否展示新增數據，showAddSwitch是顯示哪些數據，areaTree中有全國疫情數據 23 """ 24 areaTree_data = json.loads(response_data['data'])['areaTree'] 25 temp=json.loads(response_data['data']) 26 # print(temp.keys()) 27 # print(areaTree_data[0].keys()) 28 """ 29 獲取上一級字典里的areaTree 30 然后查看里面中國鍵值 31 print(areaTree_data[0].keys()) 32 dict_keys(['name', 'today', 'total', 'children']) 33 name代表國家名稱，today代表今日數據，total代表總數,children里有全國各地數據，我們需要獲取全國各地數據，查看children數據 34 print(areaTree_data[0]['children']) 35 這里面是 36 name是地區名稱，today是今日數據，total是總數，children是市級數據， 37 我們通過這個接口可以獲取每個地區的總數據。我們遍歷這個列表，取出name，這個是省級的數據，還需要獲取市級數據， 38 需要取出name，children（市級數據）下的name、total(歷史總數)下的confirm、heal、dead，today(今日數據)下的confirm（增加數）， 39 這些就是我們需要的數據 40 """ 41 # print(areaTree_data[0]['children']) 42 # for province_data in areaTree_data[0]['children']: 43 # print(province_data) 44 45 ds= temp['lastUpdateTime'] 46 details=[] 47 for pro_infos in areaTree_data[0]['children']: 48 province_name = pro_infos['name'] # 省名 49 for city_infos in pro_infos['children']: 50 city_name = city_infos['name'] # 市名 51 confirm = city_infos['total']['confirm']#歷史總數 52 confirm_add = city_infos['today']['confirm']#今日增加數 53 heal = city_infos['total']['heal']#治愈 54 dead = city_infos['total']['dead']#死亡 55 # print(ds,province_name,city_name,confirm,confirm_add,heal,dead) 56 details.append([ds,province_name,city_name,confirm,confirm_add,heal,dead]) 57 return details
單獨測試方法：
1 # d=get_details() 2 # print(d)
②此方法用於爬取歷史詳細數據
1 import requests 2 import json 3 import time 4 def get_history(): 5 url = 'https://view.inews.qq.com/g2/getOnsInfo?name=disease_other&callback=jQuery341026745307075030955_1584946267054&_=1584946267055' 6 headers ={ 7 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/70.0.3538.25 Safari/537.36 Core/1.70.3741.400 QQBrowser/10.5.3863.400' 8 } 9 res = requests.get(url,headers=headers) 10 # print(res.text) 11 response_data = json.loads(res.text.replace('jQuery341026745307075030955_1584946267054(','')[:-1]) 12 # print(response_data) 13 data = json.loads(response_data['data']) 14 # print(data.keys()) 15 chinaDayList = data['chinaDayList']#歷史記錄 16 chinaDayAddList = data['chinaDayAddList']#歷史新增記錄 17 history = {} 18 for i in chinaDayList: 19 ds = '2021.' + i['date']#時間 20 tup = time.strptime(ds,'%Y.%m.%d') 21 ds = time.strftime('%Y-%m-%d',tup)#改變時間格式，插入數據庫 22 confirm = i['confirm'] 23 suspect = i['suspect'] 24 heal = i['heal'] 25 dead = i['dead'] 26 history[ds] = {'confirm':confirm,'suspect':suspect,'heal':heal,'dead':dead} 27 for i in chinaDayAddList: 28 ds = '2021.' + i['date']#時間 29 tup = time.strptime(ds,'%Y.%m.%d') 30 ds = time.strftime('%Y-%m-%d',tup)#改變時間格式，插入數據庫 31 confirm_add = i['confirm'] 32 suspect_add = i['suspect'] 33 heal_add = i['heal'] 34 dead_add = i['dead'] 35 history[ds].update({'confirm_add':confirm_add,'suspect_add':suspect_add,'heal_add':heal_add,'dead_add':dead_add}) 36 return history
單獨測試此方法：
1 # h=get_history() 2 # print(h)
③此方法用於數據庫的連接與關閉：
1 import time 2 import pymysql 3 import traceback 4 def get_conn(): 5 """ 6 :return: 連接，游標 7 """ 8 # 創建連接 9 conn = pymysql.connect(host="127.0.0.1", 10 user="root", 11 password="000429", 12 db="mydb", 13 charset="utf8") 14 # 創建游標 15 cursor = conn.cursor() # 執行完畢返回的結果集默認以元組顯示 16 return conn, cursor 17 def close_conn(conn, cursor): 18 if cursor: 19 cursor.close() 20 if conn: 21 conn.close()
④此方法用於更新並插入每日詳細數據到數據庫表：
1 def update_details(): 2 """ 3 更新 details 表 4 :return: 5 """ 6 cursor = None 7 conn = None 8 try: 9 li = get_details() 10 conn, cursor = get_conn() 11 sql = "insert into details(update_time,province,city,confirm,confirm_add,heal,dead) values(%s,%s,%s,%s,%s,%s,%s)" 12 sql_query = 'select %s=(select update_time from details order by id desc limit 1)' #對比當前最大時間戳 13 cursor.execute(sql_query,li[0][0]) 14 if not cursor.fetchone()[0]: 15 print(f"{time.asctime()}開始更新最新數據") 16 for item in li: 17 cursor.execute(sql, item) 18 conn.commit() # 提交事務 update delete insert操作 19 print(f"{time.asctime()}更新最新數據完畢") 20 else: 21 print(f"{time.asctime()}已是最新數據！") 22 except: 23 traceback.print_exc() 24 finally: 25 close_conn(conn, cursor)
單獨測試能否插入數據到details表：
1 update_details()
⑤此方法用於插入歷史數據到history表
1 def insert_history(): 2 """ 3 插入歷史數據 4 :return: 5 """ 6 cursor = None 7 conn = None 8 try: 9 dic = get_history() 10 print(f"{time.asctime()}開始插入歷史數據") 11 conn, cursor = get_conn() 12 sql = "insert into history values(%s,%s,%s,%s,%s,%s,%s,%s,%s)" 13 for k, v in dic.items(): 14 # item 格式 {'2021-01-13': {'confirm': 41, 'suspect': 0, 'heal': 0, 'dead': 1} 15 cursor.execute(sql, [k, v.get("confirm"), v.get("confirm_add"), v.get("suspect"), 16 v.get("suspect_add"), v.get("heal"), v.get("heal_add"), 17 v.get("dead"), v.get("dead_add")]) 18 19 conn.commit() # 提交事務 update delete insert操作 20 print(f"{time.asctime()}插入歷史數據完畢") 21 except: 22 traceback.print_exc() 23 finally: 24 close_conn(conn, cursor)
單獨測試能否插入數據到history表：
1 # insert_history()
⑥此方法用於根據時間來更新歷史數據表的內容：
1 def update_history(): 2 """ 3 更新歷史數據 4 :return: 5 """ 6 cursor = None 7 conn = None 8 try: 9 dic = get_history() 10 print(f"{time.asctime()}開始更新歷史數據") 11 conn, cursor = get_conn() 12 sql = "insert into history values(%s,%s,%s,%s,%s,%s,%s,%s,%s)" 13 sql_query = "select confirm from history where ds=%s" 14 for k, v in dic.items(): 15 # item 格式 {'2020-01-13': {'confirm': 41, 'suspect': 0, 'heal': 0, 'dead': 1} 16 if not cursor.execute(sql_query, k): 17 cursor.execute(sql, [k, v.get("confirm"), v.get("confirm_add"), v.get("suspect"), 18 v.get("suspect_add"), v.get("heal"), v.get("heal_add"), 19 v.get("dead"), v.get("dead_add")]) 20 conn.commit() # 提交事務 update delete insert操作 21 print(f"{time.asctime()}歷史數據更新完畢") 22 except: 23 traceback.print_exc() 24 finally: 25 close_conn(conn, cursor)
單獨測試更新歷史數據表的方法：
1 # update_history()
最后是兩個數據表的詳細建立代碼（也可以使用mysql可視化工具直接建立）：
1 create table history( 2 ds datetime not null comment '日期', 3 confirm int(11) default null comment '累計確診', 4 confirm_add int(11) default null comment '當日新增確診', 5 suspect int(11) default null comment '剩余疑似', 6 suspect_add int(11) default null comment '當日新增疑似', 7 heal int(11) default null comment '累計治愈', 8 heal_add int(11) default null comment '當日新增治愈', 9 dead int(11) default null comment '累計死亡', 10 dead_add int(11) default null comment '當日新增死亡', 11 primary key(ds) using btree 12 )engine=InnoDB DEFAULT charset=utf8mb4; 13 create table details( 14 id int(11) not null auto_increment, 15 update_time datetime default null comment '數據最后更新時間', 16 province varchar(50) default null comment '省', 17 city varchar(50) default null comment '市', 18 confirm int(11) default null comment '累計確診', 19 confirm_add int(11) default null comment '新增確診', 20 heal int(11) default null comment '累計治愈', 21 dead int(11) default null comment '累計死亡', 22 primary key(id) 23 )engine=InnoDB default charset=utf8mb4;
Tomorrow the birds will sing.