这是百度迁徙系列的第五篇。
老规矩，如果想获取过去的数据的话，可以私聊或者电子邮件（chinshuuichi

简介

百度迁徙：百度迁徙

前几天瞅了眼百度迁徙，发现推出了实时拥堵指数，所以写了百度迁徙系列的第五篇博客。。。

效果展示

老实说，实时拥堵里面的所有数据在”拥堵详情“里都有，只不过展现的方式换了。所以我这里主要获取拥堵详情里的数据。要不然里面的类型实在是太多了。。。
拥堵详情大致分为五部分：
第一部分：实时拥堵指数与拥堵里程

第二部分：当日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我已经很贴心地整理为字典了，如下：

YDCityCode = {"北京":131,"成都":75,"东莞":119,"上海":289,"深圳":340,"苏州":224,"天津":332,"武汉":218,"西安":233,"郑州":268,"重庆":132,"保定":307,"佛山":138,"广州":257,"杭州":179,"合肥":127,"济南":288,"金华":333,"昆明":104,"临沂":234,"南京":315,"宁波":180,"青岛":236,"沈阳":58,"石家庄":150,"唐山":265,"潍坊":287,"温州":178,"无锡":317,"长沙":158,"沧州":149,"常州":348,"大连":167,"德州":372,"福州":300,"赣州":365,"贵阳":146,"哈尔滨":48,"邯郸":151,"呼和浩特":321,"惠州":301,"济宁":286,"嘉兴":334,"廊坊":191,"洛阳":153,"南昌":163,"南宁":261,"南通":161,"南阳":309,"泉州":134,"厦门":194,"绍兴":293,"台州":244,"太原":176,"乌鲁木齐":92,"新乡":152,"邢台":266,"徐州":316,"烟台":326,"盐城":223,"长春":53,"中山":187,"淄博":354,"潮州":201,"大理":111,"大同":355,"桂林":142,"海口":125,"衡水":208,"衡阳":159,"湖州":294,"淮安":162,"江门":302,"拉萨":100,"兰州":36,"乐山":79,"连云港":347,"柳州":305,"茂名":139,"绵阳":240,"南充":291,"秦皇岛":148,"清远":197,"三亚":121,"汕头":303,"韶关":137,"泰安":325,"西宁":66,"咸阳":323,"扬州":346,"阳泉":357,"宜宾":186,"银川":360,"云浮":258,"湛江":198,"张家口":264,"漳州":255,"肇庆":338,"镇江":160,"珠海":140,}

所以，在调用的时候只要将City和Code首先赋值，然后直接用以下代码就可以获取数据。
比如这里我要获取南京的数据，那么：

City = '南京'
Code = YDCityCode[City]yongdu_txt(City, Code) # 当日拥堵指数
his_yongdu_txt(City, Code) # 历史拥堵指数
zhengqu_yongdu_txt(City, Code) # 行政区拥堵排行
daolu_yongdu_txt(City, Code) # 道路拥堵排行

系列文章

【Python】基于Python的百度迁徙1——迁入、迁出数据（附代码）
【Python】基于Python的百度迁徙2——迁徙规模指数（附代码）
【Python】基于Python的百度迁徙3——城内出行强度（附代码）
【Python】基于Python的百度迁徙4——上班和休闲指数（附代码）
-----------------------分割线（以下是乞讨内容）-----------------------

【Python】基于Python的百度迁徙5——实时拥堵与拥堵指数（附代码）相关推荐

【邢不行|量化小讲堂系列46-实战篇】用Python验证A股名言：跳空必回补...吗？【附代码】
引言: 邢不行的系列帖子"量化小讲堂",通过实际案例教初学者使用python进行量化投资,了解行业研究方向,希望能对大家有帮助. [历史文章汇总]请点击此处 [必读文章]EOS期现 ...
图像处理/计算机视觉/python环境下/如何用四种不同滤波器处理噪声【附代码、亲测有效】
计算机视觉实操之图像处理一.问题描述二.效果图三.代码附录四.相关链接一.问题描述向图片中分别加入椒盐噪声.高斯噪声,使用四种不同的滤波器观察图片的处理效果(算术均值滤波.几何均值滤波 . ...
基于头脑风暴优化的BP神经网络（预测应用） - 附代码
基于头脑风暴优化的BP神经网络(预测应用) - 附代码文章目录基于头脑风暴优化的BP神经网络(预测应用) - 附代码 1.数据介绍 3.BSO优化BP神经网络 3.1 BP神经网络参数设置 3.2 ...
基于灰狼算法优化支持向量机的数据分类算法及其MATLAB实现-附代码
基于灰狼算法优化支持向量机的数据分类算法及其MATLAB实现-附代码文章目录基于灰狼算法优化支持向量机的数据分类算法及其MATLAB实现-附代码 1 GWO-SVM 模型 1.1 灰狼优化器GWO ...
Python 基于python+mysql浅谈redis缓存设计与数据库关联数据处理
基于python+mysql浅谈redis缓存设计与数据库关联数据处理 by:授客 QQ:1033553122 测试环境 redis-3.0.7 CentOS 6.5-x86_64 python 3 ...
Python 基于Python从mysql表读取千万数据实践
基于Python 从mysql表读取千万数据实践 by:授客 QQ:1033553122 场景: 有以下两个表,两者都有一个表字段,名为waybill_no,我们需要从tl_waybill_b ...
Python基于python实现的http+json协议接口自动化测试框架源码(实用改进版)
转载地址:https://www.cnblogs.com/clarke/p/5752527.html 1.写在前面抛转引玉,仅供参考 2.开发环境 win7 64位 JetBrains PyChar ...
Python 基于python实现ADSL宽带帐号，密码的获取及宽带拨号
基于python实现ADSL宽带帐号的获取及宽带拨号基本思想: 1.研究上网方式(实验环境为电信网线接入式ADSL,拨号方式PPPOE) 2.研究宽带帐号和密码生成规律(实验环境,宽带帐号为 ...
python基于Python的资产管理系统毕业设计-附源码201117
摘要现代企业管理越来越强调利用有形资产来提供优质服务的能力,即通过资产管理来确保有形资产物尽其用.安全运行,在希望的时间和地点提供需要的设备,同时尽可能地降低运行和维护成本.资产管理系统为企业提供 ...
python 爬虫实例-python爬虫实例，一小时上手爬取淘宝评论(附代码)
前言本文的文字及图片来源于网络,仅供学习.交流使用,不具有任何商业用途,版权归原作者所有,如有问题请及时联系我们以作处理. 1 明确目的通过访问天猫的网站,先搜索对应的商品,然后爬取它的评论数据. ...

【Python】基于Python的百度迁徙5——实时拥堵与拥堵指数（附代码）

目录

简介