参考：【Python】基于Python的百度迁徙2——迁徙规模指数（附代码）-CSDN博客
记录于2024年8月，这篇是获取百度迁徙指数，之前我们都在讨论不同城市的迁徙比例关系，这篇我们来获取百度迁徙指数这个数据，代码整体逻辑没变，简化了一下步骤，保存结果为.xlsx，另外把代码拆分成三部分，城市尺度、省份尺度、全国尺度，为了降低读者对这个函数migration_index(FileTittle，classname，direction，CodeDict)的理解成本，需要用哪个部分直接使用即可，但是这里还是解释一下，migration_index(文件夹名称，城市级别（city/province/country），迁徙方向（in/out），城市编码字典）；
城市编码第一篇有说明，只需要新建一个.py的文件 文件名叫''ChineseAdminiDivisionsDict''，把城市编码复制进去即可，另外城市编码这个文件一定要和下面运行的这个.py在同一个目录下面，城市编码说明：
基于python的百度迁徙迁入、迁出数据分析（一）_百度迁徙数据怎么爬取-CSDN博客
完整代码#运行环境 Python 3.11（城市尺度版）
import requests # 导入请求模块
import json # 导入JSON解析模块
import time # 导入时间模块
import xlsxwriter # 导入Excel文件操作模块
from ChineseAdminiDivisionsDict import CitiesCode, ProvinceCode # 导入中国行政区划字典
def write_header(worksheet, cities):
worksheet.write(0, 0, '城市代码') # 写入城市代码标题
worksheet.write(0, 1, '城市') # 写入城市名称标题
cities_order = {} # 存放城市序号的字典
for idx, (city_name, code) in enumerate(cities.items(), start=1):
worksheet.write(idx, 0, str(code)) # 写入城市代码
worksheet.write(idx, 1, str(city_name)) # 写入城市名称
cities_order[str(city_name)] = idx # 记录城市序号
return cities_order
def fetch_migration_data(url):
try:
response = requests.get(url, timeout=2) # 发送HTTP请求
time.sleep(3) # 等待3秒以避免频繁访问
raw_data = response.text[4:-1] # 去除响应文本的首尾非数据字符
data_dict = json.loads(raw_data) # 解析JSON数据
if data_dict['errmsg'] == 'SUCCESS':
return data_dict['data']['list'] # 返回成功数据
else:
print('Error fetching data:', data_dict['errmsg']) # 打印错误信息
return None
except Exception as e:
print(f"An error occurred: {e}") # 打印异常信息
return None
def migration_index(file_title, class_name, direction, code_dict):
direction_name = '迁入' if direction == 'in' else '迁出' # 根据方向设置标题
file_path = f'{file_title} {direction_name}规模指数.xlsx'
workbook = xlsxwriter.Workbook(file_path) # 创建Excel工作簿
worksheet = workbook.add_worksheet('Sheet') # 添加工作表
cities_order = write_header(worksheet, code_dict) # 写入头部信息并获取城市序号字典
for area, code in code_dict.items():
url = f'http://huiyan.baidu.com/migration/historycurve.jsonp?dt={class_name}&id={code}&type=move_{direction}'
print(f'{area}: {url}') # 输出正在处理的城市及URL
data = fetch_migration_data(url) # 获取迁移数据
if data:
counter_date = 2 # 用于记录日期的位置
dates = sorted(data.keys()) # 排序日期
for date in dates:
index = data[date] # 获取某天的迁移规模指数
worksheet.write(0, counter_date, float(date)) # 写入日期
worksheet.write(cities_order[str(area)], counter_date, float(index)) # 写入迁移规模指数
counter_date += 1 # 移动到下一个日期列
workbook.close() # 关闭工作簿
if __name__ == "__main__":
migration_index('城市', 'city', 'in', CitiesCode) # 处理迁入数据
migration_index('城市', 'city', 'out', CitiesCode) # 处理迁出数据
print('全部完成') # 完成提示输出结果展示：
完整代码#运行环境 Python 3.11（省份尺度版）
import requests # 导入请求模块
import json # 导入JSON解析模块
import time # 导入时间模块
import xlsxwriter # 导入Excel文件操作模块
from ChineseAdminiDivisionsDict import CitiesCode, ProvinceCode # 导入中国行政区划字典
def write_header(worksheet, cities):
worksheet.write(0, 0, '城市代码') # 写入城市代码标题
worksheet.write(0, 1, '城市') # 写入城市名称标题
cities_order = {} # 存放城市序号的字典
for idx, (city_name, code) in enumerate(cities.items(), start=1):
worksheet.write(idx, 0, str(code)) # 写入城市代码
worksheet.write(idx, 1, str(city_name)) # 写入城市名称
cities_order[str(city_name)] = idx # 记录城市序号
return cities_order
def fetch_migration_data(url):
try:
response = requests.get(url, timeout=2) # 发送HTTP请求
time.sleep(3) # 等待3秒以避免频繁访问
raw_data = response.text[4:-1] # 去除响应文本的首尾非数据字符
data_dict = json.loads(raw_data) # 解析JSON数据
if data_dict['errmsg'] == 'SUCCESS':
return data_dict['data']['list'] # 返回成功数据
else:
print('Error fetching data:', data_dict['errmsg']) # 打印错误信息
return None
except Exception as e:
print(f"An error occurred: {e}") # 打印异常信息
return None
def migration_index(file_title, class_name, direction, code_dict):
direction_name = '迁入' if direction == 'in' else '迁出' # 根据方向设置标题
file_path = f'{file_title} {direction_name}规模指数.xlsx'
workbook = xlsxwriter.Workbook(file_path) # 创建Excel工作簿
worksheet = workbook.add_worksheet('Sheet') # 添加工作表
cities_order = write_header(worksheet, code_dict) # 写入头部信息并获取城市序号字典
for area, code in code_dict.items():
url = f'http://huiyan.baidu.com/migration/historycurve.jsonp?dt={class_name}&id={code}&type=move_{direction}'
print(f'{area}: {url}') # 输出正在处理的城市及URL
data = fetch_migration_data(url) # 获取迁移数据
if data:
counter_date = 2 # 用于记录日期的位置
dates = sorted(data.keys()) # 排序日期
for date in dates:
index = data[date] # 获取某天的迁移规模指数
worksheet.write(0, counter_date, float(date)) # 写入日期
worksheet.write(cities_order[str(area)], counter_date, float(index)) # 写入迁移规模指数
counter_date += 1 # 移动到下一个日期列
workbook.close() # 关闭工作簿
if __name__ == "__main__":
migration_index('省份', 'province', 'in', ProvinceCode) # 处理迁入数据
migration_index('省份', 'province', 'out', ProvinceCode) # 处理迁出数据
print('全部完成') # 完成提示
输出结果展示：
完整代码#运行环境 Python 3.11（全国尺度版）
import requests # 导入请求模块
import json # 导入JSON解析模块
import time # 导入时间模块
import xlsxwriter # 导入Excel文件操作模块
from ChineseAdminiDivisionsDict import CitiesCode, ProvinceCode # 导入中国行政区划字典
def write_header(worksheet, cities):
worksheet.write(0, 0, '城市代码') # 写入城市代码标题
worksheet.write(0, 1, '城市') # 写入城市名称标题
cities_order = {} # 存放城市序号的字典
for idx, (city_name, code) in enumerate(cities.items(), start=1):
worksheet.write(idx, 0, str(code)) # 写入城市代码
worksheet.write(idx, 1, str(city_name)) # 写入城市名称
cities_order[str(city_name)] = idx # 记录城市序号
return cities_order
def fetch_migration_data(url):
try:
response = requests.get(url, timeout=2) # 发送HTTP请求
time.sleep(3) # 等待3秒以避免频繁访问
raw_data = response.text[4:-1] # 去除响应文本的首尾非数据字符
data_dict = json.loads(raw_data) # 解析JSON数据
if data_dict['errmsg'] == 'SUCCESS':
return data_dict['data']['list'] # 返回成功数据
else:
print('Error fetching data:', data_dict['errmsg']) # 打印错误信息
return None
except Exception as e:
print(f"An error occurred: {e}") # 打印异常信息
return None
def migration_index(file_title, class_name, direction, code_dict):
direction_name = '迁入' if direction == 'in' else '迁出' # 根据方向设置标题
file_path = f'{file_title} {direction_name}规模指数.xlsx'
workbook = xlsxwriter.Workbook(file_path) # 创建Excel工作簿
worksheet = workbook.add_worksheet('Sheet') # 添加工作表
cities_order = write_header(worksheet, code_dict) # 写入头部信息并获取城市序号字典
for area, code in code_dict.items():
url = f'http://huiyan.baidu.com/migration/historycurve.jsonp?dt={class_name}&id={code}&type=move_{direction}'
print(f'{area}: {url}') # 输出正在处理的城市及URL
data = fetch_migration_data(url) # 获取迁移数据
if data:
counter_date = 2 # 用于记录日期的位置
dates = sorted(data.keys()) # 排序日期
for date in dates:
index = data[date] # 获取某天的迁移规模指数
worksheet.write(0, counter_date, float(date)) # 写入日期
worksheet.write(cities_order[str(area)], counter_date, float(index)) # 写入迁移规模指数
counter_date += 1 # 移动到下一个日期列
workbook.close() # 关闭工作簿
QUANGUO = {'全国': 0} # 全国编码
if __name__ == "__main__":
migration_index('全国', 'country', 'in', QUANGUO) # 处理全国迁入数据
migration_index('全国', 'country', 'out', QUANGUO) # 处理全国迁出数据
print('全部完成')输出结果展示：
三个尺度的版本都在这里了，需要哪个尺度的直接用即可，另外这个数据是获取的是百度迁徙数据库有的所有历史数据，数据范围(20190112, 20190309)，(20230104, 20230506)，(20240123, 20240807（也就是截止数据获取日期前一天）)；
文章仅用于分享个人学习成果与个人存档之用，分享知识，如有侵权，请联系作者进行删除。所有信息均基于作者的个人理解和经验，不代表任何官方立场或权威解读。