ArcPy如何批量处理安然产品数据？GIS自动化巡检方案（含：脚本源码）

wowwwai GIS研习社 · 工具流程与项目排障

ArcPy如何批量处理安然产品数据？GIS自动化巡检方案（含：脚本源码）这篇文章面向需要定期检查、清洗、汇总安然产品数据的 GIS 工程师，重点解决一个具体问题：当一个项目目录里有大量 Shapefile、File Geodatabase 要素类或导出的空间数据时，如何用 ArcPy 自动完成字段检查、坐标系检查、几何修复、重复值识别、面积长度统计和巡检报告输出。

这里的“安然产品数据”可以理解为某类业务系统或产品平台交付的 GIS 数据包，例如管线、站点、巡检点、行政区、设施面、道路中心线等图层。实际项目中，这类数据往往来自不同人员、不同软件版本、不同坐标系统和不同字段模板，人工逐个打开检查不仅慢，而且很容易漏掉问题。

引言：为什么要用 ArcPy 批量处理安然产品数据

在 GIS 数据交付和运维场景中，最常见的问题不是“不会打开数据”，而是数据量一多之后，人工检查很难保持一致性。比如同一个图层在 A 项目中字段叫 PIPE_ID，在 B 项目中叫 PipeId；有的要素没有坐标系，有的几何无效，有的线要素长度为 0，有的面要素面积异常。

如果你使用 ArcGIS Pro 或 ArcMap 生态，ArcPy 是最适合做这类批量巡检的工具之一。它可以直接访问 File Geodatabase、Shapefile、要素类、字段、空间参考和几何对象，并且可以把巡检结果写成 CSV 报告，方便后续交付、返工和质量复核。

本文给出一个可改造的 GIS 自动化巡检方案，重点覆盖以下任务：

• 批量遍历文件夹和地理数据库中的空间数据。
• 检查数据是否存在坐标系。
• 检查字段是否符合模板要求。
• 统计要素数量、空几何数量和重复编号。
• 执行几何修复并记录处理结果。
• 输出巡检报告，便于项目验收和问题追踪。

背景：安然产品数据批量巡检中常见的数据问题

很多 GIS 项目会把产品数据按区域、批次、图层类型拆成多个目录。表面上看，每个数据包都能在 ArcGIS Pro 中打开，但进入制图、分析、入库或 WebGIS 发布阶段后，问题才会集中暴露。

常见问题一：坐标系缺失或不一致

坐标系缺失会导致图层无法正确叠加；坐标系不一致则可能引起量算不准、空间查询偏移、服务发布后位置错乱等问题。尤其是经纬度坐标和投影坐标混用时，面积、长度统计结果会非常不可靠。

常见问题二：字段模板不统一

安然产品数据如果要进入统一数据库，字段名、字段类型、字段长度通常需要严格一致。字段缺失会影响业务系统读取，字段类型错误会导致入库失败，字段别名混乱则会影响制图和报表。

常见问题三：几何错误和空几何

无效几何、空几何、自相交面、长度为 0 的线、面积为 0 的面，都会影响空间分析。ArcPy 可以通过 RepairGeometry 修复一部分问题，但并不是所有异常都能自动恢复，因此必须把处理结果记录到报告中。

常见问题四：唯一编号重复

很多产品数据会使用设施编号、管线编号、图斑编号作为主键。如果唯一编号重复，后续关联属性表、更新业务状态、增量入库都会出错。批量检测重复值是自动化巡检中非常实用的一步。

原理：ArcPy 自动化巡检的核心思路

ArcPy 批量处理安然产品数据，本质上是把“人工打开图层检查”的动作拆成可重复执行的程序步骤。核心思路可以概括为四层：

1. 数据发现：递归遍历指定目录，找到 Shapefile 和 File Geodatabase 中的要素类。
2. 结构检查：读取字段列表、几何类型、空间参考和要素数量。
3. 质量检查：检查空几何、重复编号、缺失字段、坐标系异常和几何错误。
4. 结果输出：把每个图层的检查结果写入 CSV，形成可追踪的巡检报告。

这种方案的优势在于规则可以固化。只要项目数据模板稳定，后续每批安然产品数据交付时都可以运行同一个脚本，检查口径一致，报告格式一致，返工依据也更清楚。

建议把 ArcPy 巡检脚本放在项目数据目录之外，不要直接在原始数据上做不可逆操作。几何修复、字段新增、投影转换等步骤最好先在副本中测试。

步骤：ArcPy 批量处理安然产品数据脚本源码

下面给出一个完整示例脚本。它适合在 ArcGIS Pro 的 Python 环境中运行。你可以把它保存为 arcpy_product_inspection.py，然后修改输入目录、报告输出路径、必需字段和唯一编号字段。

步骤一：准备运行环境

• 建议使用 ArcGIS Pro 自带的 Python 环境运行。
• 确保已安装 ArcGIS Pro，并且许可可用。
• 把待巡检的安然产品数据放到同一个根目录下。
• 如果数据是压缩包，先解压后再运行脚本。
• 不要在正在编辑的数据上直接执行几何修复。

步骤二：配置检查规则

在脚本开头配置三个关键参数：

• ROOT_DIR：需要批量处理的数据根目录。
• REPORT_CSV：巡检报告输出路径。
• REQUIRED_FIELDS：按图层类型配置必需字段。

如果你的安然产品数据字段模板不同，只需要改 REQUIRED_FIELDS 和 UNIQUE_ID_FIELD 即可。

步骤三：运行完整脚本

import arcpy
import os
import csv
from collections import Counter

arcpy.env.overwriteOutput = True

ROOT_DIR = r"D:GIS_ProjectAnran_Product_Data"
REPORT_CSV = r"D:GIS_Projectinspection_report.csv"

REQUIRED_FIELDS = {
    "Point": ["OBJECTID", "NAME", "CODE"],
    "Polyline": ["OBJECTID", "NAME", "CODE"],
    "Polygon": ["OBJECTID", "NAME", "CODE"]
}

UNIQUE_ID_FIELD = "CODE"


def list_feature_classes(root_dir):
    result = []

    for dirpath, dirnames, filenames in os.walk(root_dir):
        for filename in filenames:
            if filename.lower().endswith(".shp"):
                result.append(os.path.join(dirpath, filename))

        for dirname in dirnames:
            if dirname.lower().endswith(".gdb"):
                gdb_path = os.path.join(dirpath, dirname)
                arcpy.env.workspace = gdb_path

                for fc in arcpy.ListFeatureClasses() or []:
                    result.append(os.path.join(gdb_path, fc))

                for dataset in arcpy.ListDatasets(feature_type="feature") or []:
                    arcpy.env.workspace = os.path.join(gdb_path, dataset)
                    for fc in arcpy.ListFeatureClasses() or []:
                        result.append(os.path.join(gdb_path, dataset, fc))

                arcpy.env.workspace = None

    return result


def get_field_names(feature_class):
    return [field.name for field in arcpy.ListFields(feature_class)]


def count_null_geometry(feature_class):
    count = 0
    with arcpy.da.SearchCursor(feature_class, ["SHAPE@"]) as cursor:
        for row in cursor:
            if row[0] is None:
                count += 1
    return count


def count_duplicate_values(feature_class, field_name):
    fields = get_field_names(feature_class)
    if field_name not in fields:
        return -1

    values = []
    with arcpy.da.SearchCursor(feature_class, [field_name]) as cursor:
        for row in cursor:
            value = row[0]
            if value not in [None, ""]:
                values.append(value)

    counter = Counter(values)
    duplicate_count = sum(1 for value, freq in counter.items() if freq > 1)
    return duplicate_count


def check_required_fields(feature_class, shape_type):
    existing_fields = get_field_names(feature_class)
    required = REQUIRED_FIELDS.get(shape_type, [])
    missing = [field for field in required if field not in existing_fields]
    return missing


def repair_geometry_safe(feature_class):
    try:
        arcpy.management.RepairGeometry(feature_class)
        return "已执行"
    except Exception as ex:
        return "失败：" + str(ex)


def inspect_feature_class(feature_class):
    desc = arcpy.Describe(feature_class)

    name = desc.name
    path = feature_class
    shape_type = desc.shapeType

    try:
        feature_count = int(arcpy.management.GetCount(feature_class)[0])
    except Exception:
        feature_count = -1

    spatial_ref = desc.spatialReference
    if spatial_ref and spatial_ref.name:
        spatial_ref_name = spatial_ref.name
    else:
        spatial_ref_name = "Unknown"

    missing_fields = check_required_fields(feature_class, shape_type)
    null_geometry_count = count_null_geometry(feature_class)
    duplicate_id_count = count_duplicate_values(feature_class, UNIQUE_ID_FIELD)
    repair_result = repair_geometry_safe(feature_class)

    return {
        "name": name,
        "path": path,
        "shape_type": shape_type,
        "feature_count": feature_count,
        "spatial_reference": spatial_ref_name,
        "missing_fields": ";".join(missing_fields) if missing_fields else "",
        "null_geometry_count": null_geometry_count,
        "duplicate_id_count": duplicate_id_count,
        "repair_result": repair_result
    }


def write_report(rows, report_csv):
    fieldnames = [
        "name",
        "path",
        "shape_type",
        "feature_count",
        "spatial_reference",
        "missing_fields",
        "null_geometry_count",
        "duplicate_id_count",
        "repair_result"
    ]

    with open(report_csv, "w", newline="", encoding="utf-8-sig") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        for row in rows:
            writer.writerow(row)


def main():
    feature_classes = list_feature_classes(ROOT_DIR)

    rows = []
    for fc in feature_classes:
        print("正在巡检：", fc)
        try:
            rows.append(inspect_feature_class(fc))
        except Exception as ex:
            rows.append({
                "name": os.path.basename(fc),
                "path": fc,
                "shape_type": "ERROR",
                "feature_count": -1,
                "spatial_reference": "",
                "missing_fields": "",
                "null_geometry_count": -1,
                "duplicate_id_count": -1,
                "repair_result": "巡检失败：" + str(ex)
            })

    write_report(rows, REPORT_CSV)
    print("巡检完成，报告输出到：", REPORT_CSV)


if __name__ == "__main__":
    main()

步骤四：查看巡检报告

脚本运行完成后，会生成一个 CSV 文件。建议重点查看以下列：

• spatial_reference：如果显示 Unknown，说明坐标系缺失，需要确认原始坐标系后再定义投影。
• missing_fields：如果有字段名，说明该图层不符合字段模板。
• null_geometry_count：如果大于 0，需要检查是否存在空几何要素。
• duplicate_id_count：如果大于 0，需要检查唯一编号是否重复。
• repair_result：如果显示失败，需要查看具体错误原因。

步骤五：根据报告处理问题数据

巡检脚本的目标不是替代所有人工判断，而是快速定位问题。比如坐标系为 Unknown 时，不能随便使用投影转换工具，应该先确认数据真实坐标系。如果字段缺失，也要判断是源数据错误，还是不同图层确实使用不同模板。

对于需要进一步自动处理的项目，可以在这个脚本基础上扩展：

• 按坐标系规则自动筛选异常图层。
• 把错误图层复制到单独的问题目录。
• 自动输出每个图层的范围、面积、长度统计。
• 把 CSV 报告写入 Excel 模板。
• 把巡检结果入库到 PostGIS 或企业级地理数据库。

常见坑：ArcPy 批量处理安然产品数据时容易踩的问题

常见坑一：把 Define Projection 当成 Project 使用

Define Projection 是“定义坐标系”，用于数据本身没有坐标系标记但你知道它真实坐标系的情况。Project 是“投影转换”，用于把一个已知坐标系的数据转换到另一个坐标系。两者不能混用。

如果安然产品数据位置偏移，先检查数据坐标值范围和空间参考，不要直接批量定义投影。

常见坑二：字段大小写和别名混淆

ArcPy 的字段检查通常使用字段名，而不是字段别名。很多人在 ArcGIS Pro 表格界面看到的是字段别名，脚本读取的是字段真实名称。配置 REQUIRED_FIELDS 时，应以字段真实名称为准。

常见坑三：Shapefile 字段名长度限制

Shapefile 字段名存在长度限制，长字段名会被截断。如果安然产品数据从地理数据库导出为 Shapefile，字段名可能已经发生变化。此时字段模板检查可能出现大量缺失字段，需要先确认数据格式是否符合交付要求。

常见坑四：RepairGeometry 不等于万能修复

RepairGeometry 可以修复部分几何问题，但不能保证恢复业务正确性。例如面要素自相交被修复后，空间形态可能仍然不符合业务规则。因此，几何修复后仍建议进行抽样检查。

常见坑五：在原始数据上直接批量修改

自动化脚本效率很高，也意味着错误会被快速放大。正式批量处理前，建议先复制一份数据，在副本上运行脚本，确认巡检结果和修复逻辑没有问题后再进入正式流程。

方法比较：人工检查、ArcGIS 工具箱和 ArcPy 自动化巡检

如果只是处理一两个图层，使用 ArcGIS Pro 工具箱即可。如果需要每周或每批次处理安然产品数据，ArcPy 自动化巡检更适合，因为它可以把检查规则、处理步骤和报告输出统一起来。

检查清单：运行 ArcPy 自动化巡检前后要确认什么

运行前检查

• 是否备份了原始安然产品数据。
• 是否确认数据格式，包括 Shapefile、File Geodatabase 或企业级地理数据库。
• 是否确认字段模板和唯一编号字段。
• 是否确认坐标系检查规则。
• 是否在 ArcGIS Pro Python 环境中运行脚本。
• 是否有目标目录的读写权限。

运行后检查

• 巡检报告是否正常生成。
• 是否存在 Unknown 坐标系。
• 是否存在缺失字段。
• 是否存在重复编号。
• 是否存在空几何或异常要素数量。
• 几何修复是否成功执行。
• 问题图层是否已经单独记录并安排返工。

FAQ：ArcPy 批量处理安然产品数据常见问题

ArcPy 批量处理安然产品数据必须安装 ArcGIS Pro 吗？

通常需要。ArcPy 是 Esri ArcGIS 体系中的 Python 站点包，一般随 ArcGIS Pro 或 ArcGIS Desktop 安装。本文脚本建议在 ArcGIS Pro 自带 Python 环境中运行，这样路径、许可和依赖更稳定。

脚本能不能直接处理多个 File Geodatabase？

可以。示例脚本会递归遍历根目录，识别 .gdb 目录，并读取其中的要素类。如果地理数据库中存在要素数据集，脚本也会尝试继续遍历。

为什么巡检报告里坐标系显示 Unknown？

这通常说明数据没有写入空间参考信息。此时不要直接投影转换，应先确认数据真实坐标系，再使用定义投影工具补充空间参考。否则可能造成整体位置错误。

ArcPy 修复几何后还需要人工检查吗？

需要。几何修复可以解决部分技术性错误，但不能判断业务形态是否正确。例如一条管线被错误切断，几何本身可能合法，但业务上仍然是错误数据。

如何把巡检报告输出为 Excel？

本文为了减少依赖，使用 CSV 输出。CSV 可以直接用 Excel 打开。如果需要生成带样式的 Excel，可以在 ArcGIS Pro Python 环境中安装或使用支持 Excel 写入的库，也可以先输出 CSV，再由项目模板统一整理。

唯一编号字段不叫 CODE 怎么办？

修改脚本中的 UNIQUE_ID_FIELD 即可。例如你的设施编号字段叫 FAC_ID，就把 UNIQUE_ID_FIELD = "CODE" 改为 UNIQUE_ID_FIELD = "FAC_ID"。

安然产品数据字段模板很多，怎么配置更合理？

可以按图层名或几何类型分别配置字段模板。示例脚本为了便于理解，按 Point、Polyline、Polygon 配置。实际项目中建议按图层类型配置，例如管线层、阀门层、站点层、行政区层分别设置必需字段。

结论：把安然产品数据巡检规则固化成可复用脚本

ArcPy 批量处理安然产品数据的价值，不只是节省打开图层的时间，更重要的是把数据质量检查变成可重复、可追踪、可交付的流程。对于 GIS 自动化巡检来说，坐标系、字段、几何、数量和唯一编号是最基础也最值得优先固化的检查项。

本文提供的脚本可以作为项目模板使用。你可以先用它完成基础巡检，再根据实际业务继续扩展字段标准、空间范围检查、拓扑规则、图层命名规则和入库前校验。这样，每一批安然产品数据进入制图、分析、发布或数据库之前，都能先经过统一的质量闸口。