ECharts行政区划GeoJSON批量标准化转换合并工具

一、项目概述

在前端地图可视化开发场景中，ECharts对GeoJSON数据源具备严格格式要求。常规GIS平台导出的行政区划边界文件大多为独立单体Geometry结构，无法直接使用echarts.registerMap加载渲染，极易触发解析异常。同时原始文件普遍存在文件名冗余、缺少中心点属性、文件碎片化、未做轻量化压缩等工程问题，需要大量人工预处理。

为此搭建一套基于Python的自动化处理工具，实现原始边界JSON清洗、结构标准化、名称裁剪、几何中心点计算、文件压缩、批量合并一体化处理，输出开箱即用、适配ECharts的标准GeoJSON资源，减少人工干预，提升地图可视化项目开发效率。

二、现存数据痛点

1. 格式不兼容：原始数据仅为单纯Geometry对象，ECharts要求顶层必须为FeatureCollection结构，直接加载地图无法渲染；
2. 命名冗余：文件名称附带区县前缀（如鄞州区钟公庙街道），前端业务匹配时产生多余干扰；
3. 属性缺失：原始边界文件仅包含坐标，缺少区域中心点经纬度，无法支持弹窗、标记、自动定位等交互；
4. 文件体积偏大：格式化JSON包含大量换行与空格，前端请求加载耗时更长；
5. 文件碎片化：各镇街边界独立文件，无法一次性加载全域行政区划，需要手动拼接。

三、工具核心功能

1. 格式标准化：单体Geometry自动封装为ECharts标准FeatureCollection结构；
2. 名称自动清洗：正则剔除名称中XX区、XX县前缀，文件名与GeoJSON内部name字段同步更新；

   示例：鄞州区钟公庙街道.json → 钟公庙街道.geojson

3. 几何运算：基于Shapely自动计算区域质心，写入center_lng、center_lat属性；
4. 双模式输出：支持压缩单行（生产环境）/带缩进格式化（调试环境）自由切换；
5. 自动批量合并：所有单文件转换完成后，自动汇总生成全域合并GeoJSON；
6. 健壮容错：单文件解析失败不会中断整体任务，日志输出异常信息便于排查。

四、目录规范

./
├── origin/          # 放置原始边界json文件（程序运行前自行创建）
├── new/             # 程序自动生成，存放标准化单文件 + 合并总文件
└── geo_convert.py   # 主脚本

五、环境依赖

pip install shapely

六、完整Python源码

import os
import json
import re
from shapely.geometry import shape


def trim_district_name(name: str) -> str:
    """
    清洗行政区划名称，移除前缀【XX区、XX县】
    示例：鄞州区钟公庙街道 → 钟公庙街道
    """
    pattern = r'^.*?(区|县)'
    result = re.sub(pattern, '', name)
    return result.strip()


def process_json_to_echarts_geojson(src_file, out_dir, compress: bool = True):
    """
    原始几何JSON转换为ECharts标准GeoJSON
    :param src_file: 原始文件路径
    :param out_dir: 输出目录
    :param compress: 是否开启压缩单行输出
    :return: feature对象，合并使用
    """
    basename = os.path.basename(src_file)
    name_without_ext = os.path.splitext(basename)[0]
    simple_name = trim_district_name(name_without_ext)
    output_path = os.path.join(out_dir, f"{simple_name}.geojson")

    print(f"\n正在处理：【原始名称】{name_without_ext} → 【输出名称】{simple_name}")
    try:
        with open(src_file, "r", encoding="utf-8") as f:
            geom_data = json.load(f)

        geom = shape(geom_data)
        centroid = geom.centroid

        feature = {
            "type": "Feature",
            "properties": {
                "name": simple_name,
                "center_lng": round(centroid.x, 6),
                "center_lat": round(centroid.y, 6)
            },
            "geometry": geom_data
        }

        dump_kwargs = {"ensure_ascii": False}
        if compress:
            dump_kwargs["separators"] = (",", ":")
        else:
            dump_kwargs["indent"] = 2

        single_fc = {
            "type": "FeatureCollection",
            "features": [feature]
        }
        with open(output_path, "w", encoding="utf-8") as out_f:
            json.dump(single_fc, out_f, **dump_kwargs)

        print(f"✅ 单文件输出成功 → {output_path}")
        print(f"📍 中心点：{centroid.x:.6f}, {centroid.y:.6f}")
        return feature

    except Exception:
        import traceback
        print(f"❌ 处理失败 {basename}")
        traceback.print_exc()
        return None


def merge_geojson_files(target_dir, merge_filename="all_areas.geojson", compress: bool = True):
    """
    读取new目录所有geojson，合并为单一FeatureCollection
    """
    all_features = []
    print("\n==================== 开始合并GeoJSON文件 ====================")

    for filename in os.listdir(target_dir):
        if filename.lower().endswith(".geojson"):
            if filename == merge_filename:
                continue
            file_path = os.path.join(target_dir, filename)
            try:
                with open(file_path, "r", encoding="utf-8") as f:
                    data = json.load(f)
                if data.get("type") == "FeatureCollection" and "features" in data:
                    all_features.extend(data["features"])
                    print(f"已载入：{filename}")
            except Exception:
                import traceback
                print(f"⚠️ 读取文件失败 {filename}")
                traceback.print_exc()

    merged_collection = {
        "type": "FeatureCollection",
        "features": all_features
    }

    merge_output_path = os.path.join(target_dir, merge_filename)
    dump_kwargs = {"ensure_ascii": False}
    if compress:
        dump_kwargs["separators"] = (",", ":")
    else:
        dump_kwargs["indent"] = 2

    with open(merge_output_path, "w", encoding="utf-8") as out_f:
        json.dump(merged_collection, out_f, **dump_kwargs)

    print(f"\n🎉 合并完成！总共合并 {len(all_features)} 个区域要素")
    print(f"📦 合并文件路径：{merge_output_path}")
    return merge_output_path


if __name__ == "__main__":
    # ===================== 配置区 =====================
    SOURCE_DIR = "origin"
    TARGET_DIR = "new"
    COMPRESS_OUTPUT = True      # True生产压缩模式；False调试格式化模式
    MERGE_FILE_NAME = "all_areas.geojson"
    # ==================================================

    if not os.path.exists(TARGET_DIR):
        os.makedirs(TARGET_DIR)
        print(f"已创建输出目录：{TARGET_DIR}")

    if not os.path.isdir(SOURCE_DIR):
        print(f"错误：源目录【{SOURCE_DIR}】不存在！请创建origin文件夹放入原始json文件")
    else:
        file_count = 0
        for filename in os.listdir(SOURCE_DIR):
            if filename.lower().endswith(".json"):
                full_src = os.path.join(SOURCE_DIR, filename)
                process_json_to_echarts_geojson(full_src, TARGET_DIR, compress=COMPRESS_OUTPUT)
                file_count += 1

        print(f"\n===== 单文件转换结束，共处理 {file_count} 个JSON文件 =====")
        if file_count > 0:
            merge_geojson_files(TARGET_DIR, merge_filename=MERGE_FILE_NAME, compress=COMPRESS_OUTPUT)
        else:
            print("未找到任何可处理文件，跳过合并步骤")

七、使用步骤

1. 在脚本同级新建origin文件夹，放入所有原始边界JSON；
2. 根据需求修改头部配置参数；
3. 执行脚本；
4. 转换后的单文件以及合并后的all_areas.geojson输出至new目录。

八、ECharts前端调用示例

fetch("/new/all_areas.geojson")
.then(res => res.json())
.then(geoData => {
  echarts.registerMap("districtMap", geoData);
  const option = {
    series: [{
      type: "map",
      map: "districtMap",
      roam: true
    }]
  };
  myChart.setOption(option);
})

九、参数说明

十、拓展说明

如需适配XX市名称剔除需求，修改trim_district_name内正则表达式：

pattern = r'^.*?(区|县|市)'

推荐预览地址 乡镇行政区划GeoJSON下载工具

@漏刻有时