SuiteForge
967fa0f59c
- test-linux: platform=linux, docker + python3 - test-windows: platform=windows, arcpy + python3 - 附示例 GeoJSON: 云南/昆明/大理/百色 Ref: SuiteHub/agent-profiles#22
120 lines
3.8 KiB
Python
120 lines
3.8 KiB
Python
#!/usr/bin/env python
|
|
"""
|
|
Step 2: 叠加分析 (arcpy)
|
|
判断各城市是否在省界内，输出分析结果 JSON
|
|
"""
|
|
import sys
|
|
import json
|
|
import os
|
|
import glob
|
|
import arcpy
|
|
|
|
arcpy.env.overwriteOutput = True
|
|
|
|
|
|
def main():
|
|
province_shp = os.environ.get("PARAM_PROVINCE_SHP", "")
|
|
cities_dir = os.environ.get("PARAM_CITIES_DIR", r"C:\temp\output\shp")
|
|
output_dir = os.environ.get("PARAM_OUTPUT_DIR", r"C:\temp\output\result")
|
|
|
|
if not province_shp or not os.path.isfile(province_shp):
|
|
print json.dumps({"error": "省界 SHP 文件不存在: " + str(province_shp), "success": False})
|
|
sys.exit(1)
|
|
|
|
if not os.path.isdir(output_dir):
|
|
os.makedirs(output_dir)
|
|
|
|
# 查找所有城市 SHP
|
|
city_shp_files = glob.glob(os.path.join(cities_dir, "*.shp"))
|
|
city_shp_files = [f for f in city_shp_files
|
|
if os.path.basename(f) != "province.shp"]
|
|
|
|
if not city_shp_files:
|
|
print json.dumps({"error": "未找到城市 SHP 文件", "success": False})
|
|
sys.exit(1)
|
|
|
|
# 创建省界图层
|
|
province_lyr = "province_layer"
|
|
arcpy.MakeFeatureLayer_management(province_shp, province_lyr)
|
|
|
|
results = []
|
|
desc = arcpy.Describe(province_shp)
|
|
sr = desc.spatialReference
|
|
|
|
for shp in city_shp_files:
|
|
city_name = os.path.splitext(os.path.basename(shp))[0]
|
|
print json.dumps({"msg": "分析城市: " + city_name})
|
|
|
|
# 创建城市图层
|
|
city_lyr = "city_layer_" + city_name
|
|
arcpy.MakeFeatureLayer_management(shp, city_lyr)
|
|
|
|
# 空间选择：城市中心点在省界内
|
|
# 先获取城市几何中心
|
|
city_centroid_shp = os.path.join(output_dir, city_name + "_centroid.shp")
|
|
arcpy.FeatureToPoint_management(shp, city_centroid_shp, "INSIDE")
|
|
|
|
centroid_lyr = "centroid_layer_" + city_name
|
|
arcpy.MakeFeatureLayer_management(city_centroid_shp, centroid_lyr)
|
|
|
|
# Select by location: 中心点在省界内
|
|
arcpy.SelectLayerByLocation_management(centroid_lyr, "WITHIN", province_lyr)
|
|
count = int(arcpy.GetCount_management(centroid_lyr).getOutput(0))
|
|
in_province = count > 0
|
|
|
|
# 获取属性
|
|
city_name_prop = city_name
|
|
city_code = ""
|
|
population = 0
|
|
try:
|
|
with arcpy.da.SearchCursor(shp, ["name", "code", "population"]) as cursor:
|
|
for row in cursor:
|
|
if row[0]:
|
|
city_name_prop = str(row[0])
|
|
if row[1]:
|
|
city_code = str(row[1])
|
|
if row[2]:
|
|
population = int(row[2])
|
|
break
|
|
except Exception:
|
|
pass
|
|
|
|
status = u"通过 \u2705" if in_province else u"不通过 \u274c"
|
|
|
|
results.append({
|
|
"city": city_name,
|
|
"city_code": city_code,
|
|
"city_name": city_name_prop,
|
|
"population": population,
|
|
"in_province": in_province,
|
|
"intersects": in_province,
|
|
"status": status
|
|
})
|
|
print json.dumps({"msg": city_name + ": " + status})
|
|
|
|
# 清理
|
|
arcpy.Delete_management(city_lyr)
|
|
arcpy.Delete_management(centroid_lyr)
|
|
arcpy.Delete_management(city_centroid_shp)
|
|
|
|
arcpy.Delete_management(province_lyr)
|
|
|
|
# 输出 JSON
|
|
result_json = os.path.join(output_dir, "overlay_results.json")
|
|
with open(result_json, "w") as f:
|
|
json.dump(results, f, indent=2)
|
|
|
|
summary = {
|
|
"success": True,
|
|
"total_cities": len(results),
|
|
"in_province": sum(1 for r in results if r["in_province"]),
|
|
"not_in_province": sum(1 for r in results if not r["in_province"]),
|
|
"results": results,
|
|
"result_json": result_json
|
|
}
|
|
print json.dumps(summary)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|