gis-actions/suites/test-windows/scripts/arcpy_convert.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
Step 1: GeoJSON → SHP (arcpy)
从内置 data/ 目录读取标准 GeoJSON，手动解析后构建 Shapefile。

背景（GitHub #3）：arcpy.JSONToFeatures_conversion 仅支持 ArcGIS 内部 JSON 格式，
不兼容标准 GeoJSON（RFC 7946），需自行解析坐标并构建要素。
"""
import sys
import json
import os
import arcpy

arcpy.env.overwriteOutput = True

SR_WGS84 = arcpy.SpatialReference(4326)


def get_data_dir():
    """返回内置 data/ 目录路径（相对脚本位置）"""
    script_dir = os.path.dirname(os.path.abspath(__file__))
    return os.path.normpath(os.path.join(script_dir, "..", "data"))


def coords_to_arcpy_polygon(coordinates):
    """
    将 GeoJSON Polygon 坐标数组转为 arcpy Polygon 对象。

    GeoJSON Polygon.coordinates = [outer_ring, hole1, hole2, ...]
    每个 ring = [[x1,y1], [x2,y2], ...]（首尾闭合）
    """
    rings = arcpy.Array()
    for ring in coordinates:
        pts = arcpy.Array()
        for pt in ring:
            # GeoJSON 坐标顺序是 [经度, 纬度] = [x, y]
            pts.add(arcpy.Point(pt[0], pt[1]))
        rings.add(pts)

    return arcpy.Polygon(rings, SR_WGS84)


def process_geojson(filepath, shp_path):
    """读取单个 GeoJSON 文件，写入 Shapefile"""
    with open(filepath, "r") as f:
        geojson = json.load(f)

    fc_type = geojson.get("type", "")
    features = geojson.get("features", [])

    if fc_type != "FeatureCollection" or not features:
        raise ValueError(u"GeoJSON 应为 FeatureCollection，包含至少一个 Feature")

    # 从第一个要素探查属性字段
    sample_props = features[0].get("properties", {})
    field_defs = []
    for key, val in sample_props.items():
        if isinstance(val, int):
            field_defs.append((key, "LONG"))
        elif isinstance(val, float):
            field_defs.append((key, "DOUBLE"))
        else:
            field_defs.append((key, "TEXT"))

    # 创建 Shapefile（面要素）
    out_path, out_name = os.path.split(shp_path)
    arcpy.CreateFeatureclass_management(
        out_path, out_name, "POLYGON",
        spatial_reference=SR_WGS84
    )

    # 添加属性字段（Shapefile 字段名最多 10 字符，这里都满足）
    for fname, ftype in field_defs:
        arcpy.AddField_management(shp_path, fname, ftype)

    # 写入要素
    field_names = ["SHAPE@"] + [f[0] for f in field_defs]
    with arcpy.da.InsertCursor(shp_path, field_names) as cursor:
        for feat in features:
            geom = feat.get("geometry", {})
            props = feat.get("properties", {})

            geom_type = geom.get("type")
            coords = geom.get("coordinates", [])

            if geom_type == "Polygon":
                shape = coords_to_arcpy_polygon(coords)
            elif geom_type == "MultiPolygon":
                # 多个 Polygon 合并为一个 MultiPolygon
                parts = arcpy.Array()
                for polygon_coords in coords:
                    for ring in polygon_coords:
                        pts = arcpy.Array()
                        for pt in ring:
                            pts.add(arcpy.Point(pt[0], pt[1]))
                        parts.add(pts)
                shape = arcpy.Polygon(parts, SR_WGS84)
            else:
                print json.dumps({"warn": u"跳过不支持的几何类型: " + str(geom_type)})
                continue

            row = [shape]
            for fname, _ in field_defs:
                val = props.get(fname)
                if val is None:
                    val = 0 if _ in ("LONG", "DOUBLE") else ""
                row.append(val)
            cursor.insertRow(tuple(row))


def main():
    output_dir = os.environ.get("PARAM_OUTPUT_DIR", r"C:\temp\output\shp")

    if not os.path.isdir(output_dir):
        os.makedirs(output_dir)

    data_dir = get_data_dir()

    # 数据文件映射：输出名 → 文件名
    geo_files = {
        "province": "yunnan.geojson",
        "kunming":  "kunming.geojson",
        "dali":     "dali.geojson",
        "baise":    "baise.geojson",
    }

    converted = []
    for name, filename in geo_files.items():
        filepath = os.path.join(data_dir, filename)
        if not os.path.isfile(filepath):
            print json.dumps({"warn": u"内置数据文件不存在，跳过: " + filepath})
            continue

        print json.dumps({"msg": u"读取内置数据: " + filename})
        shp_path = os.path.join(output_dir, name + ".shp")

        try:
            process_geojson(filepath, shp_path)
            converted.append(name)
            print json.dumps({"msg": u"已转换: " + filename + u" \u2192 " + shp_path})
        except Exception as e:
            print json.dumps({"error": u"转换失败 " + filename + u": " + str(e), "success": False})
            sys.exit(1)

    if not converted:
        print json.dumps({"error": u"未成功转换任何 GeoJSON 数据", "success": False})
        sys.exit(1)

    result = {
        "success": True,
        "output_dir": output_dir,
        "converted": converted
    }
    print json.dumps(result)


if __name__ == "__main__":
    main()