第24章：数据验证与清洗

地理空间数据在采集、转换和传输过程中，常常会出现几何无效、属性缺失、坐标越界、CRS 不一致等质量问题。如果不加以检查和修复，这些问题会导致空间分析结果错误甚至程序崩溃。本章将系统介绍如何使用 GeoPandas 和 Shapely 对矢量数据进行验证与清洗，帮助读者建立完整的数据质量控制流程。

24.1 数据验证与清洗概述

24.1.1 为什么需要数据验证

地理空间数据的质量直接影响分析结果的可靠性。常见的数据质量问题包括：

问题类别	典型表现	潜在影响
几何无效	自相交多边形、零面积几何	空间运算报错或结果异常
属性缺失	字段值为 NaN 或 None	统计分析偏差
坐标越界	经度超过 180°、纬度超过 90°	投影转换失败
CRS 错误	坐标参考系缺失或不匹配	空间叠加错位
重复数据	同一要素重复录入	面积、数量统计偏大
拓扑错误	相邻多边形存在缝隙或重叠	面积汇总不一致

24.1.2 数据验证的基本流程

一个完整的数据验证与清洗流程通常包括以下步骤：

import geopandas as gpd
import pandas as pd
from shapely.validation import make_valid

# 标准验证流程
# 1. 加载数据
gdf = gpd.read_file("data.shp")

# 2. 基本信息检查
print(f"记录数: {len(gdf)}")
print(f"几何类型: {gdf.geom_type.unique()}")
print(f"CRS: {gdf.crs}")

# 3. 几何有效性检查
print(f"无效几何数: {(~gdf.is_valid).sum()}")

# 4. 属性完整性检查
print(f"缺失值统计:\n{gdf.isnull().sum()}")

# 5. 修复与清洗
# 6. 导出清洗后的数据

24.2 几何有效性检查

24.2.1 使用 is_valid 检查几何有效性

GeoPandas 的 is_valid 属性基于 OGC 简单要素规范对每个几何对象进行有效性判断：

import geopandas as gpd
from shapely.geometry import Polygon

# 创建示例数据
valid_poly = Polygon([(0, 0), (1, 0), (1, 1), (0, 1)])
invalid_poly = Polygon([(0, 0), (1, 1), (1, 0), (0, 1)])  # 自相交蝴蝶结

gdf = gpd.GeoDataFrame(
    {"name": ["有效", "无效"]},
    geometry=[valid_poly, invalid_poly]
)

# 检查有效性
print(gdf.is_valid)
# 0     True
# 1    False
# dtype: bool

24.2.2 使用 is_empty 和 is_simple

from shapely.geometry import LineString, Point
from shapely import wkt

# is_empty: 检查是否为空几何
empty_geom = wkt.loads("GEOMETRYCOLLECTION EMPTY")
print(f"是否为空几何: {empty_geom.is_empty}")  # True

# is_simple: 检查线要素是否自交叉
simple_line = LineString([(0, 0), (1, 1), (2, 0)])
complex_line = LineString([(0, 0), (1, 1), (0, 1), (1, 0)])  # 自交叉

print(f"简单线: {simple_line.is_simple}")    # True
print(f"自交叉线: {complex_line.is_simple}")  # False

24.2.3 使用 is_ring 检查环

from shapely.geometry import LinearRing, LineString

# is_ring: 检查是否为闭合且不自交的环
ring = LinearRing([(0, 0), (1, 0), (1, 1), (0, 1)])
line = LineString([(0, 0), (1, 0), (1, 1)])

print(f"闭合环: {ring.is_ring}")    # True
print(f"非闭合线: {line.is_ring}")  # False

24.2.4 获取无效原因

Shapely 提供 explain_validity() 方法（或 validation.explain_validity()）来获取几何无效的具体原因：

from shapely.geometry import Polygon
from shapely.validation import explain_validity

invalid_poly = Polygon([(0, 0), (1, 1), (1, 0), (0, 1)])
reason = explain_validity(invalid_poly)
print(f"无效原因: {reason}")
# 无效原因: Self-intersection[0.5 0.5]

批量检查无效几何及其原因：

# 批量获取所有无效几何的原因
invalid_mask = ~gdf.is_valid
invalid_gdf = gdf[invalid_mask].copy()
invalid_gdf["无效原因"] = invalid_gdf.geometry.apply(
    lambda geom: explain_validity(geom)
)
print(invalid_gdf[["name", "无效原因"]])

24.3 无效几何的常见类型

24.3.1 自相交多边形

自相交（Self-intersection）是最常见的无效几何类型，多边形的边界与自身交叉形成”蝴蝶结”形状：

from shapely.geometry import Polygon
from shapely.validation import explain_validity

# 蝴蝶结形状 — 两条边交叉
bowtie = Polygon([(0, 0), (2, 2), (2, 0), (0, 2)])
print(f"有效: {bowtie.is_valid}")
print(f"原因: {explain_validity(bowtie)}")
# 原因: Self-intersection[1 1]

24.3.2 零面积几何

当多边形退化为线或点时，面积为零，虽然在部分场景下可能被视为有效，但通常需要过滤掉：

# 退化为线的多边形
degenerate = Polygon([(0, 0), (1, 0), (2, 0), (0, 0)])
print(f"面积: {degenerate.area}")          # 0.0
print(f"有效: {degenerate.is_valid}")      # False

24.3.3 重复顶点

连续的重复顶点不会导致几何无效，但可能影响计算效率：

from shapely.geometry import Polygon

# 包含重复顶点的多边形
poly_dup = Polygon([(0, 0), (1, 0), (1, 0), (1, 1), (0, 1)])
print(f"有效: {poly_dup.is_valid}")        # True
print(f"顶点数: {len(poly_dup.exterior.coords)}")  # 6（含闭合点）

# 使用 simplify(0) 去除重复顶点
poly_clean = poly_dup.simplify(0)
print(f"清理后顶点数: {len(poly_clean.exterior.coords)}")  # 5

24.3.4 未闭合的环

在某些数据源中，多边形的外环未正确闭合。Shapely 在构造时会自动闭合，但原始数据中的此类问题值得关注：

# Shapely 会自动闭合多边形
coords = [(0, 0), (1, 0), (1, 1), (0, 1)]  # 未显式闭合
poly = Polygon(coords)
print(f"首尾相同: {poly.exterior.coords[0] == poly.exterior.coords[-1]}")
# True — Shapely 自动添加闭合点

24.4 几何修复

24.4.1 使用 make_valid() 修复

make_valid() 是修复无效几何最推荐的方法，它遵循 OGC 标准，将无效几何转换为最接近的有效表示：

from shapely.geometry import Polygon
from shapely.validation import make_valid

# 自相交多边形
invalid_poly = Polygon([(0, 0), (2, 2), (2, 0), (0, 2)])
print(f"修复前有效: {invalid_poly.is_valid}")  # False

fixed = make_valid(invalid_poly)
print(f"修复后有效: {fixed.is_valid}")          # True
print(f"修复后类型: {fixed.geom_type}")         # MultiPolygon

在 GeoDataFrame 中批量修复：

import geopandas as gpd
from shapely.validation import make_valid

gdf = gpd.read_file("data.shp")

# 批量修复无效几何
invalid_count = (~gdf.is_valid).sum()
print(f"修复前无效几何数: {invalid_count}")

gdf["geometry"] = gdf.geometry.apply(
    lambda geom: make_valid(geom) if not geom.is_valid else geom
)

print(f"修复后无效几何数: {(~gdf.is_valid).sum()}")  # 0

24.4.2 使用 buffer(0) 修复

buffer(0) 是一种经典的修复技巧，它对几何做零缓冲区操作，副作用是可以修复部分无效几何：

from shapely.geometry import Polygon

invalid_poly = Polygon([(0, 0), (2, 2), (2, 0), (0, 2)])

# 使用 buffer(0) 修复
fixed = invalid_poly.buffer(0)
print(f"修复后有效: {fixed.is_valid}")     # True
print(f"修复后类型: {fixed.geom_type}")    # MultiPolygon 或 Polygon

注意：buffer(0) 可能会丢弃某些小碎片几何，且修复结果不如 make_valid() 稳定，推荐优先使用 make_valid()。

24.4.3 两种方法的对比

特性	`make_valid()`	`buffer(0)`
标准化	遵循 OGC 标准	非标准技巧
适用范围	所有无效几何	部分无效几何
几何保留	尽量保留原始形状	可能丢弃碎片
性能	较快	较慢（涉及缓冲运算）
推荐度	⭐⭐⭐ 首选	⭐⭐ 备选

24.5 缺失值处理

24.5.1 检测 None 几何

几何列中的 None 值表示要素没有空间位置信息：

import geopandas as gpd
from shapely.geometry import Point

gdf = gpd.GeoDataFrame(
    {"name": ["A", "B", "C", "D"]},
    geometry=[Point(0, 0), None, Point(2, 2), None]
)

# 检测 None 几何
none_mask = gdf.geometry.is_empty | gdf.geometry.isna()
print(f"缺失几何数: {gdf.geometry.isna().sum()}")  # 2
print(gdf[gdf.geometry.isna()])

24.5.2 处理 None 几何

# 方法1: 删除缺失几何的行
gdf_clean = gdf.dropna(subset=["geometry"])
print(f"删除后记录数: {len(gdf_clean)}")  # 2

# 方法2: 用默认几何填充（谨慎使用）
from shapely.geometry import Point
default_point = Point(0, 0)
gdf_filled = gdf.copy()
gdf_filled.loc[gdf_filled.geometry.isna(), "geometry"] = default_point

24.5.3 处理属性缺失值

import geopandas as gpd
import numpy as np
from shapely.geometry import Point

gdf = gpd.GeoDataFrame({
    "name": ["北京", None, "广州", "深圳"],
    "population": [2154, 1893, np.nan, 1756],
    "area_km2": [16410, 6340, 7434, np.nan]
}, geometry=[Point(116.4, 39.9), Point(121.5, 31.2),
             Point(113.3, 23.1), Point(114.1, 22.5)])

# 查看缺失值
print("缺失值统计:")
print(gdf.isnull().sum())

# 删除含有任意缺失值的行
gdf_dropped = gdf.dropna()

# 用特定值填充
gdf_filled = gdf.fillna({"name": "未知", "population": 0, "area_km2": 0})

# 用均值填充数值列
gdf["population"] = gdf["population"].fillna(gdf["population"].mean())
print(gdf)

24.6 坐标范围检查

24.6.1 获取坐标边界

GeoPandas 的 bounds 属性可获取每个几何对象的边界框：

import geopandas as gpd
from shapely.geometry import Point

gdf = gpd.GeoDataFrame(
    {"name": ["正常", "经度越界", "纬度越界"]},
    geometry=[Point(116.4, 39.9), Point(200, 39.9), Point(116.4, 100)]
)

print(gdf.bounds)
#    minx   miny   maxx   maxy
# 0  116.4  39.9  116.4  39.9
# 1  200.0  39.9  200.0  39.9
# 2  116.4  100.0 116.4  100.0

24.6.2 验证经纬度范围

对于 WGS84（EPSG:4326）坐标系，经度范围应为 [-180, 180]，纬度范围应为 [-90, 90]：

def check_wgs84_bounds(gdf):
    """检查 WGS84 坐标是否在合法范围内"""
    bounds = gdf.bounds
    
    lon_valid = (bounds["minx"] >= -180) & (bounds["maxx"] <= 180)
    lat_valid = (bounds["miny"] >= -90) & (bounds["maxy"] <= 90)
    
    invalid_lon = gdf[~lon_valid]
    invalid_lat = gdf[~lat_valid]
    
    if len(invalid_lon) > 0:
        print(f"⚠ 经度越界的记录数: {len(invalid_lon)}")
    if len(invalid_lat) > 0:
        print(f"⚠ 纬度越界的记录数: {len(invalid_lat)}")
    
    all_valid = lon_valid & lat_valid
    return all_valid

# 使用
valid_mask = check_wgs84_bounds(gdf)
gdf_valid = gdf[valid_mask]
print(f"有效记录数: {len(gdf_valid)}")

24.6.3 通用坐标范围检查

对于投影坐标系，可以根据投影的有效范围进行检查：

def check_bounds(gdf, xmin, ymin, xmax, ymax):
    """检查坐标是否在指定范围内"""
    bounds = gdf.bounds
    valid = (
        (bounds["minx"] >= xmin) &
        (bounds["miny"] >= ymin) &
        (bounds["maxx"] <= xmax) &
        (bounds["maxy"] <= ymax)
    )
    invalid_count = (~valid).sum()
    if invalid_count > 0:
        print(f"⚠ 超出范围的记录数: {invalid_count}")
    return valid

# 检查中国范围内的数据（大致范围）
valid_mask = check_bounds(gdf, xmin=73, ymin=3, xmax=136, ymax=54)

24.7 CRS 验证与修正

24.7.1 检查 CRS 是否存在

import geopandas as gpd

gdf = gpd.read_file("data.shp")

# 检查 CRS
if gdf.crs is None:
    print("⚠ 数据缺少坐标参考系 (CRS)")
else:
    print(f"CRS: {gdf.crs}")
    print(f"EPSG 代码: {gdf.crs.to_epsg()}")
    print(f"是否为地理坐标系: {gdf.crs.is_geographic}")
    print(f"是否为投影坐标系: {gdf.crs.is_projected}")

24.7.2 set_crs 与 to_crs 的区别

这是初学者常混淆的两个方法：

方法	作用	何时使用	是否改变坐标值
`set_crs()`	声明/指定 CRS	数据缺少 CRS 时	否
`to_crs()`	转换到新 CRS	需要重投影时	是

import geopandas as gpd
from shapely.geometry import Point

# 创建无 CRS 的数据
gdf = gpd.GeoDataFrame(
    {"name": ["北京"]},
    geometry=[Point(116.4, 39.9)]
)
print(f"初始 CRS: {gdf.crs}")  # None

# set_crs: 声明坐标系（不改变坐标值）
gdf = gdf.set_crs(epsg=4326)
print(f"声明后 CRS: {gdf.crs}")  # EPSG:4326

# to_crs: 投影转换（改变坐标值）
gdf_proj = gdf.to_crs(epsg=3857)
print(f"转换后坐标: {gdf_proj.geometry.iloc[0]}")

24.7.3 处理 CRS 不匹配

当需要合并来自不同坐标系的数据时，必须先统一 CRS：

import geopandas as gpd

gdf_4326 = gpd.read_file("data_wgs84.shp")    # EPSG:4326
gdf_3857 = gpd.read_file("data_webmerc.shp")  # EPSG:3857

# 检查 CRS 是否一致
if gdf_4326.crs != gdf_3857.crs:
    print("⚠ CRS 不一致，正在统一...")
    gdf_3857 = gdf_3857.to_crs(gdf_4326.crs)
    print(f"统一后 CRS: {gdf_3857.crs}")

# 现在可以安全地合并
import pandas as pd
gdf_merged = pd.concat([gdf_4326, gdf_3857], ignore_index=True)

24.7.4 纠正错误的 CRS

有时数据标注的 CRS 与实际坐标不符，需要先覆盖声明再做转换：

# 数据声明了 EPSG:3857 但坐标实际是 WGS84 经纬度
gdf_wrong = gpd.read_file("wrongly_labeled.shp")

# 强制覆盖为正确的 CRS（不改变坐标值）
gdf_fixed = gdf_wrong.set_crs(epsg=4326, allow_override=True)

# 如果需要，再转换到目标 CRS
gdf_result = gdf_fixed.to_crs(epsg=3857)

24.8 重复数据检测与去除

24.8.1 属性重复检测

import geopandas as gpd
from shapely.geometry import Point

gdf = gpd.GeoDataFrame({
    "name": ["北京", "上海", "北京", "广州"],
    "code": [110000, 310000, 110000, 440100]
}, geometry=[Point(116.4, 39.9), Point(121.5, 31.2),
             Point(116.4, 39.9), Point(113.3, 23.1)])

# 检测属性重复
dup_mask = gdf.duplicated(subset=["name", "code"], keep=False)
print(f"重复记录数: {dup_mask.sum()}")
print(gdf[dup_mask])

# 去除重复（保留第一条）
gdf_dedup = gdf.drop_duplicates(subset=["name", "code"], keep="first")
print(f"去重后记录数: {len(gdf_dedup)}")

24.8.2 几何重复检测

# 基于 WKT 表示检测几何重复
gdf["geom_wkt"] = gdf.geometry.apply(lambda g: g.wkt)
geom_dup_mask = gdf.duplicated(subset=["geom_wkt"], keep=False)
print(f"几何重复记录数: {geom_dup_mask.sum()}")

# 去除几何重复
gdf_geom_dedup = gdf.drop_duplicates(subset=["geom_wkt"], keep="first")
gdf_geom_dedup = gdf_geom_dedup.drop(columns=["geom_wkt"])

24.8.3 基于空间关系去重

对于坐标略有偏差但实际表示同一地物的情况，可以基于距离阈值去重：

import geopandas as gpd
from shapely.geometry import Point

gdf = gpd.GeoDataFrame({
    "name": ["点A", "点A近似", "点B"]
}, geometry=[Point(116.400, 39.900), Point(116.401, 39.901), Point(121.5, 31.2)],
   crs="EPSG:4326")

# 投影到平面坐标系以使用米为单位
gdf_proj = gdf.to_crs(epsg=3857)

def spatial_dedup(gdf, distance_threshold=100):
    """基于距离阈值的空间去重"""
    keep = [True] * len(gdf)
    for i in range(len(gdf)):
        if not keep[i]:
            continue
        for j in range(i + 1, len(gdf)):
            if not keep[j]:
                continue
            dist = gdf.geometry.iloc[i].distance(gdf.geometry.iloc[j])
            if dist < distance_threshold:
                keep[j] = False
    return gdf[keep]

gdf_dedup = spatial_dedup(gdf_proj, distance_threshold=200)
print(f"空间去重后: {len(gdf_dedup)} 条记录")

24.9 拓扑一致性检查

24.9.1 检测多边形之间的重叠

相邻多边形之间不应有重叠区域，可以通过两两求交来检测：

import geopandas as gpd
from shapely.geometry import Polygon
import itertools

polys = [
    Polygon([(0, 0), (2, 0), (2, 2), (0, 2)]),
    Polygon([(1, 0), (3, 0), (3, 2), (1, 2)]),  # 与第一个重叠
    Polygon([(4, 0), (6, 0), (6, 2), (4, 2)])
]

gdf = gpd.GeoDataFrame(
    {"id": [1, 2, 3]},
    geometry=polys
)

# 检测重叠
overlaps = []
for i, j in itertools.combinations(range(len(gdf)), 2):
    geom_i = gdf.geometry.iloc[i]
    geom_j = gdf.geometry.iloc[j]
    if geom_i.overlaps(geom_j):
        overlap_area = geom_i.intersection(geom_j).area
        overlaps.append({
            "要素1": gdf["id"].iloc[i],
            "要素2": gdf["id"].iloc[j],
            "重叠面积": round(overlap_area, 4)
        })

if overlaps:
    import pandas as pd
    print("检测到重叠:")
    print(pd.DataFrame(overlaps))
else:
    print("未检测到重叠")

24.9.2 检测多边形之间的缝隙

缝隙（Gap）是指相邻多边形之间本应无缝衔接但出现了空白区域：

from shapely.ops import unary_union
from shapely.geometry import box

# 计算所有多边形的并集
union_geom = unary_union(gdf.geometry)

# 定义研究区域的边界框
study_area = box(*gdf.total_bounds)

# 缝隙 = 研究区域 - 并集
gaps = study_area.difference(union_geom)

if not gaps.is_empty:
    print(f"检测到缝隙，面积: {gaps.area:.4f}")
else:
    print("未检测到缝隙")

24.9.3 使用空间索引加速拓扑检查

对于大数据集，使用空间索引可显著提高检测效率：

import geopandas as gpd

def detect_overlaps_fast(gdf):
    """使用空间索引快速检测重叠"""
    sindex = gdf.sindex
    overlaps = []
    
    for idx, row in gdf.iterrows():
        possible_matches_idx = list(sindex.intersection(row.geometry.bounds))
        for match_idx in possible_matches_idx:
            if match_idx <= idx:
                continue  # 避免重复比较
            other = gdf.geometry.iloc[match_idx]
            if row.geometry.overlaps(other):
                overlaps.append((idx, match_idx))
    
    return overlaps

# overlapping_pairs = detect_overlaps_fast(gdf)

24.10 数据清洗工作流

24.10.1 完整的清洗管线

以下是一个将前述所有检查步骤整合在一起的完整数据清洗流程：

import geopandas as gpd
import pandas as pd
from shapely.validation import make_valid, explain_validity

def validate_and_clean(filepath, target_crs="EPSG:4326"):
    """
    完整的地理数据验证与清洗管线

    参数:
        filepath: 输入文件路径
        target_crs: 目标坐标参考系

    返回:
        清洗后的 GeoDataFrame 及清洗报告
    """
    report = {}

    # === 步骤 1: 加载数据 ===
    gdf = gpd.read_file(filepath)
    report["原始记录数"] = len(gdf)
    print(f"[1/7] 加载数据: {len(gdf)} 条记录")

    # === 步骤 2: CRS 检查与修正 ===
    if gdf.crs is None:
        print("[2/7] ⚠ 缺少 CRS，设置为目标 CRS")
        gdf = gdf.set_crs(target_crs)
    elif gdf.crs.to_epsg() != int(target_crs.split(":")[1]):
        print(f"[2/7] 转换 CRS: {gdf.crs} → {target_crs}")
        gdf = gdf.to_crs(target_crs)
    else:
        print(f"[2/7] CRS 正确: {gdf.crs}")

    # === 步骤 3: 删除 None 几何 ===
    none_count = gdf.geometry.isna().sum()
    if none_count > 0:
        gdf = gdf.dropna(subset=["geometry"])
        print(f"[3/7] 删除空几何: {none_count} 条")
    else:
        print("[3/7] 无空几何")
    report["空几何数"] = none_count

    # === 步骤 4: 修复无效几何 ===
    invalid_mask = ~gdf.is_valid
    invalid_count = invalid_mask.sum()
    if invalid_count > 0:
        gdf.loc[invalid_mask, "geometry"] = gdf.loc[invalid_mask, "geometry"].apply(
            make_valid
        )
        print(f"[4/7] 修复无效几何: {invalid_count} 个")
    else:
        print("[4/7] 所有几何有效")
    report["无效几何数"] = invalid_count

    # === 步骤 5: 坐标范围检查 ===
    if target_crs == "EPSG:4326":
        bounds = gdf.bounds
        out_of_range = ~(
            (bounds["minx"] >= -180) & (bounds["maxx"] <= 180) &
            (bounds["miny"] >= -90) & (bounds["maxy"] <= 90)
        )
        oor_count = out_of_range.sum()
        if oor_count > 0:
            gdf = gdf[~out_of_range]
            print(f"[5/7] 删除越界坐标: {oor_count} 条")
        else:
            print("[5/7] 坐标范围正常")
        report["坐标越界数"] = oor_count
    else:
        print("[5/7] 跳过坐标范围检查（非 WGS84）")
        report["坐标越界数"] = "N/A"

    # === 步骤 6: 去除重复记录 ===
    gdf["_geom_wkt"] = gdf.geometry.apply(lambda g: g.wkt)
    dup_count = gdf.duplicated(subset=["_geom_wkt"]).sum()
    if dup_count > 0:
        gdf = gdf.drop_duplicates(subset=["_geom_wkt"], keep="first")
        print(f"[6/7] 去除重复记录: {dup_count} 条")
    else:
        print("[6/7] 无重复记录")
    gdf = gdf.drop(columns=["_geom_wkt"])
    report["重复记录数"] = dup_count

    # === 步骤 7: 属性缺失值处理 ===
    null_counts = gdf.drop(columns=["geometry"]).isnull().sum()
    total_nulls = null_counts.sum()
    if total_nulls > 0:
        print(f"[7/7] 属性缺失值:")
        for col, cnt in null_counts[null_counts > 0].items():
            print(f"       {col}: {cnt} 个缺失")
    else:
        print("[7/7] 无属性缺失值")
    report["属性缺失值总数"] = total_nulls

    # === 汇总 ===
    report["清洗后记录数"] = len(gdf)
    gdf = gdf.reset_index(drop=True)

    print("\n=== 清洗报告 ===")
    for k, v in report.items():
        print(f"  {k}: {v}")

    return gdf, report

24.10.2 使用清洗管线

# 执行清洗
# gdf_clean, report = validate_and_clean("raw_data.shp")

# 导出清洗后的数据
# gdf_clean.to_file("clean_data.gpkg", driver="GPKG")

24.10.3 生成数据质量报告

def generate_quality_report(gdf, output_path="quality_report.csv"):
    """生成数据质量报告"""
    report_rows = []
    
    # 几何统计
    report_rows.append({"检查项": "总记录数", "结果": len(gdf)})
    report_rows.append({"检查项": "几何类型", "结果": str(gdf.geom_type.unique())})
    report_rows.append({"检查项": "无效几何数", "结果": (~gdf.is_valid).sum()})
    report_rows.append({"检查项": "空几何数", "结果": gdf.geometry.isna().sum()})
    report_rows.append({"检查项": "CRS", "结果": str(gdf.crs)})
    
    # 属性统计
    for col in gdf.columns:
        if col != "geometry":
            null_count = gdf[col].isnull().sum()
            report_rows.append({
                "检查项": f"字段 '{col}' 缺失数",
                "结果": null_count
            })
    
    report_df = pd.DataFrame(report_rows)
    report_df.to_csv(output_path, index=False, encoding="utf-8-sig")
    print(f"质量报告已保存至: {output_path}")
    return report_df

24.11 本章小结

本章系统介绍了使用 GeoPandas 进行地理空间数据验证与清洗的核心方法。下表汇总了主要的检查与修复手段：

检查项目	检查方法	修复方法
几何有效性	`is_valid`、`explain_validity()`	`make_valid()`、`buffer(0)`
空几何	`geometry.isna()`、`is_empty`	`dropna(subset=["geometry"])`
几何简单性	`is_simple`、`is_ring`	`simplify(0)`
坐标越界	`bounds` 属性比较	过滤或裁剪
CRS 问题	`gdf.crs` 属性检查	`set_crs()` / `to_crs()`
属性缺失	`isnull().sum()`	`fillna()` / `dropna()`
重复数据	`duplicated()`	`drop_duplicates()`
拓扑一致性	`overlaps()`、`difference()`	手动编辑或自动修复

关键要点：

先验证后分析：在进行任何空间分析之前，务必完成数据验证。
优先使用 make_valid()：修复无效几何时，优先选择 make_valid() 而非 buffer(0)。
区分 set_crs 和 to_crs：前者只是声明坐标系，后者会实际转换坐标值。
建立标准化流程：将验证与清洗步骤封装为可复用的函数或管线，确保数据处理的一致性。
保留清洗日志：记录每个步骤的处理结果，便于追溯数据质量问题。