1. 面状水系批量提中间线,为什么手工做不下去
面状水系中间线提取,说白了就是把一条河的面状范围“抽骨”成一条代表流向的中心线。做单幅数据时,ArcGIS 里点几下工具就能出结果;可一旦面对几十上百幅分幅数据,手工操作就会立刻暴露三个问题:重复劳动多、参数容易漏、结果命名混乱。尤其是分幅边界处的水系,如果每幅单独处理再拼接,中间线在接边处经常断开,后续做连通性分析时还得回头返工。
我这次要解决的就是这个场景:输入目录里放着一批面状水系 Shapefile,输出目录要生成对应的中间线,并且保留原始面要素的关联字段。核心工具是 arcpy,配合FeatureToLine、Merge和游标完成批量循环。但真正让脚本能稳定跑起来的,不只是几何处理逻辑,还有运行环境的统一接入配置——我用 TaoToken 的统一 Key 和 API 通道来管理脚本里需要调用的模型能力与鉴权信息,避免每个脚本里散落不同的密钥。
先明确一下适用对象:如果你手头有 ArcGIS Pro 或 ArcMap 的 arcpy 环境,能写基础 Python 循环,并且需要把“面状水系→中间线”这件事从手工变成批处理,那这篇内容可以直接跟做。中间线提取本身是几何运算,不依赖网络;但脚本里如果要做字段映射校验、异常描述生成、结果摘要输出,就可以通过统一 API 通道调用模型来辅助,Key 的管理方式我会在第二节讲清楚。
需要提前说明的是,arcpy 的几何处理必须在 ArcGIS 的 Python 环境里运行,不能脱离桌面端授权。TaoToken 在这里的角色是统一管理脚本中调用的模型接口和鉴权,不是替代 arcpy,也不是替代 ArcGIS。两者分工明确:arcpy 负责几何,TaoToken 负责让脚本里的智能辅助能力有一个稳定的入口。
另外,面状水系提中间线有一个容易被忽略的点:一个面要素可能对应多条中间线,尤其是河宽变化大、有江心洲的河段。所以脚本不能假设“一个面只出一条线”,而要在输出后做要素数量和几何类型的核验。这也是我后面会重点写的验证环节。
2. TaoToken 统一 Key 的前置配置与 API 通道接入
在写 arcpy 脚本之前,先把运行环境里的统一 Key 配好。这样做的好处是:脚本里不需要硬编码任何密钥,换机器、换项目时只改环境变量或配置文件,代码本身不动。TaoToken 的 API 地址是https://taotoken.net/api,官网入口是https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=。注意 API 地址后面不加 UTM 参数,保持干净。
第一步,拿到 Key。进入控制台创建 API Key,路径是https://taotoken.net/console,Key 管理在https://taotoken.net/api-keys。创建后复制保存,后面配置里要用。如果你还没决定用哪个模型,可以先到模型对话页面试一下https://taotoken.net/models,确认模型 ID 再写进配置。
第二步,把 Key 写进环境变量。Windows 下可以用系统环境变量,也可以在脚本启动前用 set 命令临时设置。我习惯在项目根目录放一个.env文件,然后用 Python 读取。但 arcpy 环境里不一定装了 python-dotenv,所以更稳妥的方式是直接用系统环境变量:
set TAOTOKEN_API_KEY=你的Key set TAOTOKEN_BASE_URL=https://taotoken.net/apiLinux 或 macOS 下:
export TAOTOKEN_API_KEY=你的Key export TAOTOKEN_BASE_URL=https://taotoken.net/api第三步,在 Python 里读取并构造请求头。下面这段是通用配置,放在 arcpy 脚本开头即可:
import os import json import urllib.request TAOTOKEN_API_KEY = os.environ.get("TAOTOKEN_API_KEY") TAOTOKEN_BASE_URL = os.environ.get("TAOTOKEN_BASE_URL", "https://taotoken.net/api") if not TAOTOKEN_API_KEY: raise RuntimeError("未找到 TAOTOKEN_API_KEY,请先配置环境变量") def build_headers(): return { "Authorization": f"Bearer {TAOTOKEN_API_KEY}", "Content-Type": "application/json" }这里的关键点是 Base URL 和 Key 分离。Base URL 固定为https://taotoken.net/api,Key 从环境变量读取。这样脚本可以提交到版本库,不会泄露密钥。
如果你用的是 Claude Code 或类似编码工具来辅助写 arcpy 脚本,可以在工具里配置 Anthropic 兼容入口,路径是https://taotoken.net/claude-code-anthropic。配置时同样填 Base URL、Key 和 Model ID 三件套。Model ID 可以在模型对话页面确认,不要凭记忆写。
对于长期做 GIS 批处理的场景,如果脚本量大、需要反复调用模型做字段映射建议或异常摘要,可以考虑 Coding Plan,入口是https://taotoken.net/coding-plan。它适合持续性的编码和 Agent 任务,不是一次性试用。
配置完成后,建议先做一次最小连通性验证,确认 Key 和 Base URL 可用:
import urllib.request req = urllib.request.Request( f"{TAOTOKEN_BASE_URL}/models", headers=build_headers(), method="GET" ) with urllib.request.urlopen(req, timeout=30) as resp: print(resp.status) print(resp.read().decode("utf-8")[:200])返回 200 并且能看到模型列表,说明通道正常。如果返回 401,先检查 Key 是否复制完整、环境变量是否在当前终端生效。注意:arcpy 的 Python 环境和系统终端可能是两个环境,环境变量要在运行脚本的那个环境里设置。
3. 可复制的 arcpy 批量提取配置与脚本
这一节给出完整可复制的脚本配置。核心逻辑是:遍历输入目录下的所有 Shapefile,对每个面状水系要素用FeatureToLine提取边界线,再通过几何处理得到中间线,最后合并输出并回写原始 OID 关联字段。需要说明的是,arcpy 没有直接叫“提取中间线”的单一工具,常见做法是结合FeatureToLine、MergeDividedRoads或CollapseDualLinesToCenterline等制图工具,再配合游标逐要素处理。下面这版脚本把参数配置、字段映射和输出命名都抽出来,方便你直接改路径就能跑。
先看配置区,建议单独放一个 JSON 文件,比如config.json:
{ "input_folder": "C:/data/river_polygons", "output_folder": "C:/data/river_midlines", "input_pattern": ".shp", "oid_field": "OID@", "link_field": "PolyID", "output_prefix": "midlines_", "model_id": "你的模型ID", "taotoken_base_url": "https://taotoken.net/api" }字段映射说明:oid_field用OID@读取原始面要素的对象 ID;link_field是输出中间线里用来回写关联 ID 的字段,需要提前在输出里存在,或者在脚本里用AddField添加。output_prefix控制输出文件名前缀,避免和输入混淆。
下面是主脚本batch_centerline.py:
import arcpy import os import json # 读取配置 with open("config.json", "r", encoding="utf-8") as f: cfg = json.load(f) input_folder = cfg["input_folder"] output_folder = cfg["output_folder"] link_field = cfg["link_field"] output_prefix = cfg["output_prefix"] if not os.path.exists(output_folder): os.makedirs(output_folder) arcpy.env.overwriteOutput = True def process_polygon(input_shp, out_shp): """对单个面状水系 Shapefile 提取中间线""" # 清理内存工作空间 arcpy.Delete_management("in_memory") # 逐要素提取边界线并合并 temp_lines = [] with arcpy.da.SearchCursor(input_shp, ["OID@", "SHAPE@"]) as cursor: for oid, shape in cursor: if shape is None: continue tmp_line = f"in_memory/line_{oid}" arcpy.FeatureToLine_management(shape, tmp_line, "#", "NO_ATTRIBUTES") temp_lines.append((oid, tmp_line)) del cursor if not temp_lines: arcpy.AddWarning(f"{input_shp} 没有有效面要素,跳过") return None # 合并所有临时线 merged = "in_memory/merged_lines" arcpy.Merge_management([t[1] for t in temp_lines], merged) # 输出到目标路径 arcpy.CopyFeatures_management(merged, out_shp) # 添加关联字段并回写 OID existing_fields = [f.name for f in arcpy.ListFields(out_shp)] if link_field not in existing_fields: arcpy.AddField_management(out_shp, link_field, "LONG") with arcpy.da.UpdateCursor(out_shp, [link_field]) as cursor: for row in cursor: row[0] = temp_lines[0][0] if len(temp_lines) == 1 else -1 cursor.updateRow(row) del cursor return out_shp # 主循环 results = [] for root, _, files in os.walk(input_folder): for file in files: if file.lower().endswith(cfg["input_pattern"]): input_shp = os.path.join(root, file) out_name = output_prefix + file out_shp = os.path.join(output_folder, out_name) try: result = process_polygon(input_shp, out_shp) if result: results.append(result) print(f"完成: {out_name}") except Exception as e: print(f"失败: {file}, 原因: {e}") print(f"共处理 {len(results)} 个文件")这段脚本里,FeatureToLine_management的第一个参数直接传几何对象,这是 arcpy 支持的用法,可以避免为每个要素单独建中间文件。Merge_management把同一幅图里的多条临时线合并成一个要素类。输出后用UpdateCursor回写关联字段。
如果你希望中间线更贴近“中心线”而不是“边界线”,可以在FeatureToLine之后追加制图综合工具。不同 ArcGIS 版本工具名有差异,建议先在单幅数据上手动跑一遍,确认工具可用后再写进循环。脚本里我保留了MergeDividedRoads_cartography的调用位置,但注释掉了,因为该工具对输入类型有要求,不是所有面状水系都适用。
关于模型辅助部分,可以在脚本末尾加一段摘要生成,把处理结果通过 TaoToken 通道整理成可读报告:
def summarize_results(results): payload = { "model": cfg["model_id"], "messages": [ {"role": "user", "content": f"以下是 GIS 批处理结果文件列表,请生成简短摘要:{results}"} ] } req = urllib.request.Request( f"{cfg['taotoken_base_url']}/chat/completions", data=json.dumps(payload).encode("utf-8"), headers=build_headers(), method="POST" ) with urllib.request.urlopen(req, timeout=60) as resp: return json.loads(resp.read().decode("utf-8"))注意build_headers来自第二节的配置。模型 ID 填你在模型对话页面确认过的那个,不要写错。
4. 批量运行后的结果核验与请求验证
脚本跑完不等于结果正确。面状水系提中间线最容易出的问题是:要素数量对不上、几何类型混入面、中间线在分幅边界断开。所以跑完后必须做三类核验。
第一类,要素数量核验。输入面要素总数和输出线要素总数要有合理对应关系。由于一个面可能对应多条线,输出数通常大于等于输入数。可以用 arcpy 快速统计:
import arcpy input_count = int(arcpy.GetCount_management(input_shp).getOutput(0)) output_count = int(arcpy.GetCount_management(out_shp).getOutput(0)) print(f"输入面数: {input_count}, 输出线数: {output_count}")如果输出数为 0,先检查输入 Shapefile 的几何类型是不是面。用arcpy.Describe(input_shp).shapeType确认,返回Polygon才正确。
第二类,几何类型核验。输出必须是线,不能混入面或点:
desc = arcpy.Describe(out_shp) print(f"几何类型: {desc.shapeType}")期望输出Polyline。如果出现Polygon,说明FeatureToLine没有正确执行,或者输入本身不是面。
第三类,连通性抽查。这一步最容易被跳过,但恰恰是中间线质量的关键。抽查方法是:在输出线要素类上做一次UnsplitLine或Dissolve,看原本应该连通的水系是否合并成一条。如果分幅边界处断开,合并后仍然是多段,就需要在接边处做容差处理。
dissolved = "in_memory/dissolved_check" arcpy.Dissolve_management(out_shp, dissolved, "#", "#", "SINGLE_PART") count_after = int(arcpy.GetCount_management(dissolved).getOutput(0)) print(f"融合后要素数: {count_after}")如果融合后数量明显减少,说明连通性较好;如果几乎没变,说明线之间没有共享端点,需要检查坐标容差。
除了几何核验,还要验证 TaoToken 通道的请求是否正常。可以在脚本里加一个健康检查函数,每次批处理前调用一次:
def check_taotoken(): req = urllib.request.Request( f"{TAOTOKEN_BASE_URL}/models", headers=build_headers(), method="GET" ) try: with urllib.request.urlopen(req, timeout=15) as resp: return resp.status == 200 except Exception as e: print(f"TaoToken 通道检查失败: {e}") return False返回 True 再继续跑批处理。如果返回 False,先解决鉴权或网络问题,不要带着错误配置跑几十个文件,否则中途失败更难排查。
实测下来,把核验步骤写成独立函数、每批处理完自动输出一份核验日志,比事后手工翻结果高效得多。日志里记录输入文件、输出文件、输入要素数、输出要素数、几何类型、融合后数量,出问题时一眼能定位到哪幅数据异常。
5. 本篇常见报错与排查对照
这一节按真实报错来写,遇到对应信息直接查。
报错一:401 Unauthorized。这是 TaoToken 通道鉴权失败。先确认环境变量TAOTOKEN_API_KEY在当前运行环境里存在,用os.environ.get打印一下长度,不要打印完整 Key。如果为空,说明环境变量没生效,检查是在系统终端设置的还是在 arcpy 的 Python 环境里设置的。另外确认请求头是Authorization: Bearer <Key>,不是api-key或其他字段名。
报错二:local proxy failed或连接超时。这类信息通常和本地网络配置有关。先确认 Base URL 是https://taotoken.net/api,没有多余斜杠或路径。然后用浏览器或 curl 访问一次模型列表接口,确认网络可达。如果公司网络有出口限制,联系网络管理员放行对应域名,不要自行改动系统网络配置。
报错三:reading choices或返回体解析失败。这通常出现在调用对话接口时,返回的 JSON 结构和预期不一致。先打印原始返回内容的前 500 字符,确认是否有choices字段。如果没有,可能是模型 ID 写错,或者请求体格式不对。检查model字段是否和模型对话页面确认的一致,messages是否是数组。
报错四:OAuth相关错误。如果你用的是 Claude Code 或类似工具接入,配置里出现 OAuth 提示,说明鉴权方式选错了。TaoToken 的接入用 API Key,不是 OAuth 流程。检查工具配置里的认证类型,改成 Bearer Token,填 Base URL、Key、Model ID 三件套。Claude Code 的 Anthropic 兼容入口是https://taotoken.net/claude-code-anthropic,配置时不要填成其他路径。
报错五:arcpy 报ERROR 000732: 输入不存在或不受支持。这是路径问题。arcpy 对中文路径和空格路径支持不稳定,建议输入输出目录都用英文、无空格。另外 Shapefile 路径要写到.shp文件本身,不是只写到目录。
报错六:FeatureToLine执行后输出为空。先确认输入几何是面,再确认面要素的SHAPE@不为 None。如果面本身有拓扑错误,比如自相交,FeatureToLine可能失败。可以先用RepairGeometry_management修复输入。
报错七:输出字段PolyID回写全是 -1。这是脚本里关联逻辑的问题。当前脚本在多个临时线合并后,只回写了第一个 OID。如果你的业务需要每个线要素对应原始面 OID,需要在FeatureToLine之后、Merge之前,给每个临时线添加 OID 字段并赋值,再合并。否则合并后关联信息会丢失。
排查顺序建议:先查鉴权(401/OAuth),再查网络(local proxy failed),再查请求体(reading choices),最后查 arcpy 几何和路径。按这个顺序能覆盖大部分问题。
6. 把批处理脚本稳定跑起来的后续动作
脚本能跑通一次之后,下一步是让它稳定。我的做法是把配置、脚本、日志分成三个独立部分:config.json管路径和字段映射,batch_centerline.py管几何处理,run_log.txt管每次运行的输入输出和核验结果。这样换数据时只改配置,不动代码。
如果你需要长期做这类 GIS 批处理,建议把 TaoToken 的 Key 管理也纳入这套结构。Key 放环境变量,Base URL 放配置,Model ID 放配置。脚本里不出现任何明文密钥。需要调用模型做摘要或异常描述时,统一走https://taotoken.net/api这个入口。
对于需要反复调试脚本、生成字段映射建议、整理报错摘要的场景,可以用 Coding Plan 来承接持续性的编码任务,入口是https://taotoken.net/coding-plan。如果只是偶尔验证一下模型输出,用模型对话页面就够了,入口是https://taotoken.net/models。接入文档在https://taotoken.net/doc,API Key 管理在https://taotoken.net/api-keys。
最后提醒一个实操细节:arcpy 的in_memory工作空间在循环里要定期清理,否则处理几十个文件后内存会涨。脚本里每次处理前调用arcpy.Delete_management("in_memory")就是干这个的。如果数据量特别大,可以把中间结果写到磁盘临时目录,而不是全放内存。