当前位置：首页 > news >正文

基于python的百度迁徙迁入、迁出数据分析（七）

news 来源：原创 2024/9/20 3:02:37

参考：【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_orderdef 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 Noneexcept Exception as e:print(f"An error occurred: {e}")  # 打印异常信息return Nonedef 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}')  # 输出正在处理的城市及URLdata = 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_orderdef 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 Noneexcept Exception as e:print(f"An error occurred: {e}")  # 打印异常信息return Nonedef 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}')  # 输出正在处理的城市及URLdata = 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_orderdef 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 Noneexcept Exception as e:print(f"An error occurred: {e}")  # 打印异常信息return Nonedef 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}')  # 输出正在处理的城市及URLdata = 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('全部完成')

输出结果展示：