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

参考：【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('全部完成')

输出结果展示：