📖 目录

• 一、场景概述
• 二、数据准备
• 三、转换方法
• 四、完整代码
• 五、PowerBI成果图

---

一、场景概述

需求：地图展现与下钻功能

• 目标：将行政区划地理数据转换为Power BI可识别的格式
• 技术路线：GeoJSON → WKT(MULTIPOLYGON)转换
• 最终产出：包含WKT几何信息的结构化数据表

---

二、数据准备

2.1 数据来源

• 阿里云DataV行政区划选择器：
  • 地址：https://datav.aliyun.com/portal/school/atlas/area_selector
  • 格式：GeoJSON格式行政区划数据
  • 层级：支持省、市、区县三级下钻

2.2 需求内容

• 以省为例子，需求福建省与下级市区JSON文件

2.3 数据预处理

• 文件筛选：排除带括号的文件（示例文件）
• 目录结构：省级、市级GeoJSON文件分别存放
• 编码格式：UTF-8编码确保中文字符正确处理

---

三、转换方法

3.1 核心转换流程

GeoJSON文件 → 读取解析 → 几何对象转换 → WKT格式 → 坐标简化 → 数据表输出

3.2 代码模块解析

3.2.1 坐标精度处理

def round_coordinates_in_string(wkt_string):
    """
    功能：对WKT字符串中的坐标值进行四舍五入
    输入：原始的WKT字符串
    输出：坐标精度为3位小数的WKT字符串
    原理：使用正则表达式匹配并处理所有浮点数坐标
    """

3.2.2 GeoJSON转WKT核心函数

def get_WKT(path):
    """
    功能：将GeoJSON文件转换为WKT格式
    处理步骤：
    1. 读取并解析GeoJSON文件
    2. 使用Shapely库的shape()将GeoJSON几何转换为几何对象
    3. 对几何对象进行简化（tolerance=0.01）
    4. 提取属性信息：名称、adcode、中心点坐标
    5. 转换为WKT字符串并存储到全局列表
    """

3.3 数据处理流程

步骤1：初始化变量

wkt_strings = []    # 存储WKT几何字符串
cityName = []       # 存储城市/区域名称
centroid = []       # 存储中心点坐标
adcode = []         # 存储行政区划代码
prov = []           # 存储省份名称（从文件名提取）

步骤2：批量处理文件

# 获取当前目录下所有非示例GeoJSON文件
filtered_files = list_files_with_flag(directory_path, flag='N')

# 遍历处理每个文件
for file in filtered_files:
    get_WKT(file)  # 执行GeoJSON到WKT的转换

步骤3：构建数据表

df_prov_city = pd.DataFrame(columns=['Prov', 'City', 'centroid', 'adcode', 'wkt'])
for i in range(len(wkt_strings)):
    df_prov_city.loc[i] = [prov[i], cityName[i], centroid[i], adcode[i],
                          round_coordinates_in_string(wkt_strings[i])]

步骤4：数据输出

# 输出到Excel文件
df_prov_city.to_excel('./Prov_city.xlsx', index=False)

3.4 WKT格式说明

3.4.1 MULTIPOLYGON结构

MULTIPOLYGON (
  ((经度1 纬度1, 经度2 纬度2, ..., 经度n 纬度n, 经度1 纬度1)),  # 多边形1
  ((经度1 纬度1, 经度2 纬度2, ..., 经度m 纬度m, 经度1 纬度1))   # 多边形2
)

3.4.2 坐标简化策略

• 目的：减少数据量，提高Power BI渲染性能
• 方法：Shapely的simplify()函数，tolerance=0.01
• 效果：在保持形状基本特征的前提下减少坐标点数

---

四、完整代码

#%%
## 
import re
import json
import os
from shapely.geometry import shape
import pandas as pd

def round_coordinates_in_string(wkt_string):
    """
    功能：对WKT字符串中的坐标值进行四舍五入处理
    输入：原始的WKT字符串
    输出：坐标精度为3位小数的WKT字符串
    """
    # 正则表达式匹配坐标值
    pattern = r"(-?\d+\.\d+)"
    matches = re.findall(pattern, wkt_string)

    # 对匹配到的坐标值进行四舍五入处理
    rounded_matches = [round(float(match), 3) for match in matches]

    # 将处理后的坐标值替换回原字符串
    def replace(match):
        return str(rounded_matches.pop(0))

    rounded_wkt_string = re.sub(pattern, replace, wkt_string)

    return rounded_wkt_string

def extract_prefix(filename):
    """
    功能：从文件名中提取省份/城市名称
    规则：提取括号前的内容或去除文件扩展名
    """
    # 定义正则表达式模式，匹配括号前的内容或者去除文件扩展名
    pattern = r'^(.*?)(?:\s*\(|\.[^.]+$)'

    # 使用正则表达式搜索文件名
    match = re.search(pattern, filename)

    if match:
        # 如果找到匹配项，则返回匹配的内容（括号前的文本）
        return match.group(1).strip()
    else:
        # 如果没有找到匹配项，返回去掉扩展名的文件名
        return re.sub(r'\.[^.]+$', '', filename)

def get_WKT(path):
    """
    功能：将GeoJSON文件转换为WKT格式
    处理步骤：
    1. 读取并解析GeoJSON文件
    2. 将GeoJSON几何转换为几何对象
    3. 简化几何对象
    4. 提取属性信息并存储
    """
    with open(path, 'r', encoding='utf-8') as f:
        geojson_data = json.load(f)
    features = geojson_data['features']

    # 遍历每个几何对象
    for feature in features:
        # 构建 WKT 字符串
        geometry = shape(feature['geometry'])
        cityName.append(feature['properties'].get('name'))
        centroid.append(feature['properties'].get('centroid'))
        adcode.append(feature['properties'].get('adcode'))
        prov.append(extract_prefix(path))
        
        # 简化几何对象（可选参数 tolerance 控制简化程度）
        simplified_geometry = geometry.simplify(tolerance=0.01)

        # 将简化后的几何对象转换为 WKT 字符串
        wkt_string = simplified_geometry.wkt

        # 添加到列表中
        wkt_strings.append(wkt_string)
    print(f'{path}文件已处理')

def list_files_with_flag(directory, char_to_exclude=')', flag='N'):
    """
    功能：根据标志筛选文件
    逻辑：
    - flag='N'：排除包含指定字符的文件
    - flag='Y'：只包含包含指定字符的文件
    """
    # 获取指定目录下的所有文件和文件夹名称
    files_and_dirs = os.listdir(directory)

    # 根据 flag 的值决定是否包含或排除特定字符的文件
    if flag.upper() == 'N':
        # 如果 flag 是 N，选择不包含特定字符的文件
        files = [f for f in files_and_dirs if os.path.isfile(os.path.join(directory, f)) and char_to_exclude not in f]
    elif flag.upper() == 'Y':
        # 如果 flag 是 Y，选择包含特定字符的文件
        files = [f for f in files_and_dirs if os.path.isfile(os.path.join(directory, f)) and char_to_exclude in f]
    else:
        # 如果 flag 不是 Y 或 N，默认选择不包含特定字符的文件
        files = [f for f in files_and_dirs if os.path.isfile(os.path.join(directory, f)) and char_to_exclude not in f]

    return files

# 初始化全局变量
wkt_strings = []
cityName = []
centroid = []
adcode = []
prov = []

# 使用函数并打印结果
directory_path = os.getcwd()
flag_value = 'N'  # 设置为 'Y' 或 'N' 来控制筛选行为

filtered_files = list_files_with_flag(directory_path, flag=flag_value)
print("筛选到的文件列表：")
print(filtered_files)

# 批量处理所有文件
for i in filtered_files:
    get_WKT(i)


# 创建省-市级别数据表
df = pd.DataFrame(columns=['Prov','City', 'centroid','adcode','wkt'])

# 逐行处理 WKT 字符串并添加到 DataFrame 中
for i in range(len(wkt_strings)):
    df.loc[i] = [prov[i],cityName[i],centroid[i],adcode[i],round_coordinates_in_string(wkt_strings[i])]
    print(f"已处理：{prov[i]}-{cityName[i]}，WKT长度：{len(round_coordinates_in_string(wkt_strings[i]))}")

# 保存省-市级别数据
df.to_excel(r'./Prov_city.xlsx', index=False)

---

五、PowerBI成果图

5.1 BI样式

---

转载请注明出处，欢迎交流讨论