【Python】基于Python的百度迁徙5——实时拥堵与拥堵指数（附代码）
生活随笔
收集整理的這篇文章主要介紹了
【Python】基于Python的百度迁徙5——实时拥堵与拥堵指数（附代码）
小編覺得挺不錯的,現在分享給大家,幫大家做個參考.
這是百度遷徙系列的第五篇。
老規矩，如果想獲取過去的數據的話，可以私聊或者電子郵件（chinshuuichi
目錄
- 簡介
- 效果展示
- 代碼（當日24小時擁堵指數變化）
- 代碼（當日24小時擁堵指數變化-歷史）
- 代碼（實時擁堵行政區排行）
- 代碼（實時擁堵道路排行）
- 代碼調用
- 系列文章
簡介
百度遷徙 ：百度遷徙
前幾天瞅了眼百度遷徙，發現推出了實時擁堵指數，所以寫了百度遷徙系列的第五篇博客。。。
效果展示
老實說，實時擁堵里面的所有數據在”擁堵詳情“里都有，只不過展現的方式換了。所以我這里主要獲取擁堵詳情里的數據。要不然里面的類型實在是太多了。。。
擁堵詳情大致分為五部分：
第一部分：實時擁堵指數與擁堵里程
第二部分：當日24小時擁堵指數變化
這里包含了歷史的（說是歷史的其實目前來看是去年十月份均值。。）和實時的。
第三部分：近七天高峰擁堵指數變化
其實也就是第二部分在高峰期求平均罷了。。。
第四部分：實時擁堵行政區排行
第五部分：實時擁堵道路排行
這個和第四部分挺像的，不過第四部分列出了所有行政區，而第五部分只列出前十位道路
我整理了上面幾部分是數據，發現其實也就兩種數據類型，
第一種：全市擁堵指數及其附屬數據（里程、速度等）
第二種：行政區和道路擁堵指數及其附屬數據。
所以，我分兩個大模塊抓取數據，第一模塊市級的（包括當日和歷史），第二模塊微觀級別的（包括政區和道路）。所以要說是四個模塊也行。
存儲方式我用txt，因為數據更新太快了，txt方便實時續寫。
頁面數據里有的都可以抓到，甚至還可以抓到頁面數據里沒有的，比如街道的經緯度坐標、行政區劃代碼什么的。
爬取思路和前幾篇文章差不多。
正如上面所說，數據更新太快，所以這個代碼如果加個定時器是可以支持續寫的。續寫的邏輯是讀取你將要爬取的城市的信息，然后寫上你沒有爬取過的信息。
因此可以續寫到地老天荒。
定時器的話。。自己加。。
下面直接上代碼，四部分的代碼將以函數的形式表達，調用方式相似，具體調用方法看最后。
代碼（當日24小時擁堵指數變化）
# -*- coding: utf-8 -*- import requests import json import time import datetimedef yongdu_txt(City, Code):# 讀取數據try:with open(f"百度遷徙數據v3.1//擁堵指數（當日和歷史）//{Code}-{City}-擁堵指數（當日）.txt", "r") as f:old_data = f.readlines()print(old_data[-1])shangci_data = old_data[-1]shangci_data = str.split(shangci_data)shangci = shangci_data[2][-4:] + '-' + shangci_data[3][-2:] + '-' + shangci_data[4][-2:] + ',' + shangci_data[5][-2:] + ':' + shangci_data[6][-2:]k = 0 # 判斷器except:print('文件不存在')shangci = 'kkk'k = 1 # 判斷器# 寫入數據with open(f"百度遷徙數據v3.1//擁堵指數（當日和歷史）//{Code}-{City}-擁堵指數（當日）.txt", "a") as f:url=f'https://jiaotong.baidu.com/trafficindex/city/curve?cityCode={Code}&type=minute&his=0' # 實時print(f'{City}:{url}')try:response = requests.get(url, timeout=20) # #發出請求并json化處理except :print('正在準備重新啟動------------------------------------------\n')time.sleep(10)response = requests.get(url, timeout=20) # #發出請求并json化處理print('啟動中----------------------------------------------------\n')# 獲取時間信息today = datetime.datetime.now() # 今天yesterday = datetime.datetime.now() - datetime.timedelta(days=1) # 昨天today_str = today.strftime("%Y-%m-%d,%H:%M")yesterday_str = yesterday.strftime("%Y-%m-%d,%H:%M")# print(today)# print(yesterday)# print(today_str)# print(yesterday_str)day = yesterday_str # 預設日期為昨天if len(response.text) > 200:data_dict = json.loads(response.text) # 字典化data_list = data_dict['data']['list']for i in range(len(data_list)):data_list_dic = data_list[i]data_index = data_list_dic['index']data_speed = data_list_dic['speed']data_time = data_list_dic['time']if data_time == '00:00': # 到0點將日期切換為今天day = today_strdata_time = day[:11]+data_timeif k == 1 or shangci == data_time:k = 1 # 判斷器if shangci == data_time:print('正在續寫數據')else:f.write(f"City:{City} Code:{Code} Year:{data_time[:4]} Month:{data_time[5:7]} Day:{data_time[8:10]} Hour:{data_time[-5:-3]} Minute:{data_time[-2:]} Index:{data_index} Speed:{data_speed}\n") # 續寫數據寫入數據print('完成')代碼（當日24小時擁堵指數變化-歷史）
import requests import json import time import datetimedef his_yongdu_txt(City, Code):# 讀取數據try:with open(f"百度遷徙數據v3.1//擁堵指數（當日和歷史）//{Code}-{City}-擁堵指數（歷史）.txt", "r") as f:old_data = f.readlines()print(old_data[-1])shangci_data = old_data[-1]shangci_data = str.split(shangci_data)shangci = shangci_data[2][-4:] + '-' + shangci_data[3][-2:] + '-' + shangci_data[4][-2:] + ',' + shangci_data[5][-2:] + ':' + shangci_data[6][-2:]k = 0 # 判斷器except:print('文件不存在')shangci = 'kkk'k = 1 # 判斷器# 寫入數據with open(f"百度遷徙數據v3.1//擁堵指數（當日和歷史）//{Code}-{City}-擁堵指數（歷史）.txt", "a") as f:url=f'https://jiaotong.baidu.com/trafficindex/city/curve?cityCode={Code}&type=minute&his=1' # 歷史print(f'{City}:{url}')try:response = requests.get(url, timeout=20) # #發出請求并json化處理except :print('正在準備重新啟動------------------------------------------\n')time.sleep(10)response = requests.get(url, timeout=20) # #發出請求并json化處理print('啟動中----------------------------------------------------\n')# 獲取時間信息today = datetime.datetime.now() # 今天yesterday = datetime.datetime.now() - datetime.timedelta(days=1) # 昨天today_str = today.strftime("%Y-%m-%d,%H:%M")yesterday_str = yesterday.strftime("%Y-%m-%d,%H:%M")# print(today)# print(yesterday)# print(today_str)# print(yesterday_str)day = yesterday_str # 預設日期為昨天if len(response.text) > 200:data_dict = json.loads(response.text) # 字典化data_list = data_dict['data']['list']for i in range(len(data_list)):data_list_dic = data_list[i]data_index = data_list_dic['index']data_speed = data_list_dic['speed']data_time = data_list_dic['time']data_his_index = data_list_dic['hisIndex']data_his_speed = data_list_dic['hisSpeed']if data_time == '00:00': # 到0點將日期切換為今天day = today_strdata_time = day[:11]+data_timeif k == 1 or shangci == data_time:k = 1 # 判斷器if shangci == data_time:print('正在續寫數據')else:f.write(f"City:{City} Code:{Code} Year:{data_time[:4]} Month:{data_time[5:7]} Day:{data_time[8:10]} Hour:{data_time[-5:-3]} Minute:{data_time[-2:]} Index:{data_index} Speed:{data_speed} HisIndex:{data_his_index} HisSpeed:{data_his_speed}\n") # 續寫數據寫入數據if shangci[-5:] == '23:30':f.write(f"City:{City} Code:{Code} Year:{data_time[:4]} Month:{data_time[5:7]} Day:{data_time[8:10]} Hour:{data_time[-5:-3]} Minute:{data_time[-2:]} Index:{data_index} Speed:{data_speed} HisIndex:{data_his_index} HisSpeed:{data_his_speed}\n") # 續寫數據寫入數據print('完成')代碼（實時擁堵行政區排行）
import requests import json import time import datetimedef zhengqu_yongdu_txt(City, Code):# 讀取數據try:with open(f"{Code}-{City}-擁堵指數（政區）.txt", "r") as f:old_data = f.readlines()print(old_data[-1])except:print('文件不存在')# 寫入數據with open(f"{Code}-{City}-擁堵指數（政區）.txt", "a") as f:url=f'https://jiaotong.baidu.com/trafficindex/city/districtrank?cityCode={Code}&roadtype=0' # 實時print(f'{City}:{url}')try:response = requests.get(url, timeout=20) # #發出請求并json化處理except :print('正在準備重新啟動------------------------------------------\n')time.sleep(10)response = requests.get(url, timeout=20) # #發出請求并json化處理print('啟動中----------------------------------------------------\n')# 獲取時間信息today = datetime.datetime.now() # 今天yesterday = datetime.datetime.now() - datetime.timedelta(days=1) # 昨天today_str = today.strftime("%Y-%m-%d,%H:%M")yesterday_str = yesterday.strftime("%Y-%m-%d,%H:%M")# print(today)# print(yesterday)# print(today_str)# print(yesterday_str)day = yesterday_str # 預設日期為昨天if len(response.text) > 200:data_dict = json.loads(response.text) # 字典化data_list = data_dict['data']['list']for i in range(len(data_list)):data_list_dic = data_list[i]data_time = data_list_dic['timeHuman'] # 202110241425data_district_name = data_list_dic['district_name']data_district_code = data_list_dic['district_code']data_index = data_list_dic['index']data_index_level = data_list_dic['index_level']data_speed = data_list_dic['speed']data_length = data_list_dic['length']data_center = data_list_dic['center']f.write(f"City:{City} Code:{Code} Year:{data_time[:4]} Month:{data_time[4:6]} Day:{data_time[6:8]} Hour:{data_time[8:10]} Minute:{data_time[10:]} District:{data_district_name} DistrictCode:{data_district_code} Index:{data_index} IndexLevel:{data_index_level} Speed:{data_speed} Length:{data_length} Center:{data_center}\n") # 續寫數據寫入數據print('完成')代碼（實時擁堵道路排行）
import requests import json import time import datetimedef daolu_yongdu_txt(City, Code):# 讀取數據try:with open(f"{Code}-{City}-擁堵指數（道路）.txt", "r") as f:old_data = f.readlines()print(old_data[-1])except:print('文件不存在')# 寫入數據with open(f"{Code}-{City}-擁堵指數（道路）.txt", "a") as f:url=f'https://jiaotong.baidu.com/trafficindex/city/roadrank?cityCode={Code}&roadtype=0' # 實時print(f'{City}:{url}')try:response = requests.get(url, timeout=20) # #發出請求并json化處理except :print('正在準備重新啟動------------------------------------------\n')time.sleep(10)response = requests.get(url, timeout=20) # #發出請求并json化處理print('啟動中----------------------------------------------------\n')# 獲取時間信息today = datetime.datetime.now() # 今天yesterday = datetime.datetime.now() - datetime.timedelta(days=1) # 昨天today_str = today.strftime("%Y-%m-%d,%H:%M")yesterday_str = yesterday.strftime("%Y-%m-%d,%H:%M")# print(today)# print(yesterday)# print(today_str)# print(yesterday_str)day = yesterday_str # 預設日期為昨天if len(response.text) > 200:data_dict = json.loads(response.text) # 字典化data_list = data_dict['data']['list']for i in range(len(data_list)):data_list_dic = data_list[i]data_id = data_list_dic['id']data_time = data_list_dic['time'] # 202110241425data_district_type = data_list_dic['district_type']data_roadsegid = data_list_dic['roadsegid']data_speed = data_list_dic['speed']data_yongdu_length = data_list_dic['yongdu_length']data_road_type = data_list_dic['road_type']data_road_name = data_list_dic['roadname']data_index = data_list_dic['index']data_index_level = data_list_dic['index_level']data_length = data_list_dic['length']data_semantic = data_list_dic['semantic']data_links = data_list_dic['links']data_location = data_list_dic['location']infor_a = f"City:{City} Code:{Code} Year:{data_time[:4]} Month:{data_time[4:6]} Day:{data_time[6:8]} Hour:{data_time[8:10]} Minute:{data_time[10:]} "infor_b = f"DistrictType:{data_district_type} Roadsegid:{data_roadsegid} Speed:{data_speed} Yongdu_Length:{data_yongdu_length} RoadType:{data_road_type} RoadName:{data_road_name} "infor_c = f"Index:{data_index} IndexLevel:{data_index_level} Semantic:{data_semantic} Links:{data_links} Location:{data_location}\n"infor = infor_a + infor_b + infor_cf.write(infor) # 續寫數據寫入數據print('完成')代碼調用
可以發現，跳轉到指定城市的url必須輸入對應城市的code，因此調用上面任意一個函數的時候必須輸入城市名和對應的code。
也就是代碼里City和Code參數。
關于Code我已經很貼心地整理為字典了，如下：
所以，在調用的時候只要將City和Code首先賦值，然后直接用以下代碼就可以獲取數據。
比如這里我要獲取南京的數據，那么：
系列文章
【Python】基于Python的百度遷徙1——遷入、遷出數據（附代碼）
【Python】基于Python的百度遷徙2——遷徙規模指數（附代碼）
【Python】基于Python的百度遷徙3——城內出行強度（附代碼）
【Python】基于Python的百度遷徙4——上班和休閑指數（附代碼）
-----------------------分割線（以下是乞討內容）-----------------------
總結
以上是生活随笔為你收集整理的【Python】基于Python的百度迁徙5——实时拥堵与拥堵指数（附代码）的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： Spring系列（十三）:AOP相关知识
- 下一篇： python中ndarray对象实例化_