基于ezdxf、geopandas、shapely的DWG转Shapefile问题求助

问题背景 #

本人是CAD与GIS领域新手，基于Python的ezdxf、geopandas和shapely开发DWG转Shapefile功能：通过ezdxf的odafc插件（依赖ODA File Converter）将DWG转DXF，遍历模型空间实体生成GeoDataFrame，修复破损图形后保存为Shapefile，但遇到以下问题：

• 无法获取全部点、线、面实体
• 无法提取全部文本，已提取文本在GeoDataFrame显示失真
• fix_lines函数生成多余线条，破坏Shapefile
• fix_polygons函数无作用，无法确认提取的多边形是否正确

问题解决方案 #

1. 无法获取全部实体的修复 #

当前代码仅处理LINE、LWPOLYLINE等少数实体类型，DWG中常见的ARC、ELLIPSE、块引用等未覆盖，需补充处理：

• ARC/ELLIPSE/SPLINE：用ezdxf的to_lwpolyline()方法离散为多段线后转换为LineString
• 块引用(INSERT)：递归遍历块定义内的实体，提取嵌套内容
• 隐藏/冻结图层：通过dxf.layers过滤仅处理可见图层

示例补充ARC处理代码：

elif entity.dxftype() == 'ARC':
    # 转成多段线离散，segments控制精度
    lwpoly = entity.to_lwpolyline(segments=32)
    points_list = lwpoly.get_points('xy')
    lines.append(LineString(points_list))

2. 文本提取失真与不全的修复 #

• 文本不全：补充处理ATTDEF（属性定义）、ATTRIB（块属性），块内文本需递归提取
• 显示失真：原sanitize_text函数过滤了所有非ASCII字符，导致中文/特殊字符丢失，修改为仅过滤控制字符：

def sanitize_text(text):
    """保留有效Unicode，过滤控制字符"""
    return re.sub(r'[\x00-\x1F\x7F]', '', text)

• 文本位置优化：MTEXT使用entity.get_text_location()获取实际文本显示位置，同时可提取文本高度、颜色等属性存入GeoDataFrame字段

3. fix_lines函数生成多余线条的修复 #

原函数强制给所有非闭合线条首尾连接，完全不符合需求。正确逻辑是修复拓扑错误（重复点、自相交），而非强制闭合：

def fix_lines(gdf_lines):
    fixed_geoms = []
    for geom in gdf_lines['geometry']:
        if geom.type == 'LineString':
            # 去重重复点
            coords = list(dict.fromkeys(geom.coords))
            if len(coords) >= 2:
                fixed = LineString(coords)
                # 修复自相交
                if not fixed.is_simple:
                    fixed = fixed.simplify(0.01, preserve_topology=True)
                fixed_geoms.append(fixed)
            else:
                fixed_geoms.append(None)
        elif geom.type == 'MultiLineString':
            fixed_parts = []
            for part in geom:
                coords = list(dict.fromkeys(part.coords))
                if len(coords) >=2:
                    fixed_part = LineString(coords)
                    if not fixed_part.is_simple:
                        fixed_part = fixed_part.simplify(0.01)
                    fixed_parts.append(fixed_part)
            fixed_geoms.append(MultiLineString(fixed_parts) if fixed_parts else None)
        else:
            fixed_geoms.append(geom)
    
    gdf_lines['geometry'] = fixed_geoms
    return gdf_lines.dropna(subset=['geometry']).reset_index(drop=True)

4. fix_polygons函数无作用的修复 #

shapely的Polygon本身要求必须闭合，问题出在多边形生成阶段而非修复阶段：

• 生成Polygon前，用浮点精度判断首尾点是否接近，而非直接用==对比
• 修复自相交、无效多边形：

先定义多边形创建工具函数：

def create_valid_polygon(points_list):
    """创建并修复有效多边形"""
    try:
        ring = LinearRing(points_list)
        if not ring.is_closed:
            ring = LinearRing(list(ring.coords) + [list(ring.coords)[0]])
        if not ring.is_valid:
            ring = ring.simplify(0.01, preserve_topology=True)
        polygon = Polygon(ring)
        if not polygon.is_valid:
            polygon = polygon.buffer(0)  # 缓冲0修复自相交
        return polygon
    except Exception as e:
        print(f"Failed to create polygon: {e}")
        return None

转换POLYLINE/LWPOLYLINE时调用该函数，替代直接创建Polygon：

elif entity.dxftype() == 'LWPOLYLINE':
    points_list = entity.get_points('xy')
    polygon = create_valid_polygon(points_list)
    if polygon:
        polygons.append(polygon)
    else:
        lines.append(LineString(points_list))

修改后的fix_polygons函数：

def fix_polygons(gdf_polygons):
    fixed_geoms = []
    for geom in gdf_polygons['geometry']:
        if geom.type == 'Polygon':
            if not geom.is_valid:
                geom = geom.buffer(0)
            exterior = geom.exterior
            if not exterior.is_closed:
                coords = list(exterior.coords)
                coords.append(coords[0])
                geom = Polygon(coords, [list(interior.coords) for interior in geom.interiors])
            fixed_geoms.append(geom)
        else:
            fixed_geoms.append(geom)
    gdf_polygons['geometry'] = fixed_geoms
    return gdf_polygons

无需依赖第三方软件的替代库 #

• dxfgrabber：纯Python读取DXF，无需ODA，但需先将DWG转DXF
• pyautocad：通过COM接口调用AutoCAD直接读取DWG，需安装AutoCAD，适合Windows环境
• GDAL/OGR：GDAL的Python绑定，支持直接读取DWG（编译时需加入DWG驱动），可直接转Shapefile，功能强大

修改后的完整代码 #

import geopandas as gpd
import re, copy
import matplotlib.pyplot as plt
from ezdxf.addons import odafc
from shapely.geometry import Polygon, LineString, Point, MultiLineString, LinearRing

def sanitize_text(text):
    """保留有效Unicode，过滤控制字符"""
    return re.sub(r'[\x00-\x1F\x7F]', '', text)

def create_valid_polygon(points_list):
    """创建并修复有效多边形"""
    try:
        ring = LinearRing(points_list)
        if not ring.is_closed:
            ring = LinearRing(list(ring.coords) + [list(ring.coords)[0]])
        if not ring.is_valid:
            ring = ring.simplify(0.01, preserve_topology=True)
        polygon = Polygon(ring)
        if not polygon.is_valid:
            polygon = polygon.buffer(0)
        return polygon
    except Exception as e:
        print(f"Failed to create polygon: {e}")
        return None

def convert_dwg_to_shp(dwg):
    try:
        polygons = []
        lines = []
        points = []
        annotations = []

        dxf = odafc.readfile(dwg)
        msp = dxf.modelspace()

        # 递归处理块内实体
        def process_block(block):
            for entity in block:
                process_entity(entity)
        
        def process_entity(entity):
            try:
                if entity.dxftype() == 'LINE':
                    points_list = [entity.dxf.start, entity.dxf.end]
                    lines.append(LineString(points_list))
                elif entity.dxftype() == 'LWPOLYLINE':
                    points_list = entity.get_points('xy')
                    polygon = create_valid_polygon(points_list)
                    if polygon:
                        polygons.append(polygon)
                    else:
                        lines.append(LineString(points_list))
                elif entity.dxftype() == 'POLYLINE':
                    points_list = [point[:2] for point in entity.points()]
                    polygon = create_valid_polygon(points_list)
                    if polygon:
                        polygons.append(polygon)
                    else:
                        lines.append(LineString(points_list))
                elif entity.dxftype() == 'POINT':
                    points.append(Point(entity.dxf.location))
                elif entity.dxftype() == 'CIRCLE':
                    center = entity.dxf.center
                    radius = entity.dxf.radius
                    circle = Point(center).buffer(radius)
                    polygons.append(circle)
                elif entity.dxftype() == 'ARC':
                    lwpoly = entity.to_lwpolyline(segments=32)
                    points_list = lwpoly.get_points('xy')
                    lines.append(LineString(points_list))
                elif entity.dxftype() == 'TEXT':
                    text_content = sanitize_text(entity.dxf.text)
                    position = entity.dxf.insert
                    annotations.append({
                        'text': text_content,
                        'height': entity.dxf.height,
                        'geometry': Point(position[0], position[1])
                    })
                elif entity.dxftype() == 'MTEXT':
                    text_content = sanitize_text(entity.get_text())
                    position = entity.get_text_location()
                    annotations.append({
                        'text': text_content,
                        'height': entity.dxf.char_height,
                        'geometry': Point(position[0], position[1])
                    })
                elif entity.dxftype() == 'INSERT':
                    block = dxf.blocks.get(entity.dxf.name)
                    if block:
                        process_block(block)
                elif entity.dxftype() == 'ATTRIB':
                    text_content = sanitize_text(entity.dxf.text)
                    position = entity.dxf.insert
                    annotations.append({
                        'text': text_content,
                        'height': entity.dxf.height,
                        'geometry': Point(position[0], position[1])
                    })

            except Exception as e:
                print(f"Error processing entity {entity.dxftype()}: {e}")

        # 遍历模型空间实体
        for entity in msp:
            process_entity(entity)

        # 封装GeoDataFrame创建逻辑
        def create_gdf(geom_list, crs=4326):
            if not geom_list:
                return gpd.GeoDataFrame(columns=['geometry'], crs=crs)
            gdf = gpd.GeoDataFrame({'geometry': geom_list}, crs=crs)
            return gdf.set_geometry('geometry')

        geo_dataframe_lines = create_gdf(lines)
        geo_dataframe_points = create_gdf(points)
        geo_dataframe_polygons = create_gdf(polygons)

        annotations_gdf = gpd.GeoDataFrame(annotations, crs=4326) if annotations else gpd.GeoDataFrame(columns=['text', 'height', 'geometry'], crs=4326)

        # 分图预览
        fig, axes = plt.subplots(2, 2, figsize=(12, 10))
        geo_dataframe_lines.plot(ax=axes[0,0], aspect=1, color='blue', title='Lines')
        geo_dataframe_points.plot(ax=axes[0,1], aspect=1, color='red', marker='o', title='Points')
        geo_dataframe_polygons.plot(ax=axes[1,0], aspect=1, facecolor='none', edgecolor="green", title='Polygons')
        annotations_gdf.plot(ax=axes[1,1], aspect=1, color='black', marker='*', title='Annotations')
        plt.tight_layout()
        plt.show()

        return geo_dataframe_lines, geo_dataframe_points, geo_dataframe_polygons, annotations_gdf

    except Exception as e:
        print(f"An error occurred: {e}")
        return None, None, None, None
    
def fix_lines(gdf_lines):
    fixed_geoms = []
    for geom in gdf_lines['geometry']:
        if geom.type == 'LineString':
            coords = list(dict.fromkeys(geom.coords))
            if len(coords) >= 2:
                fixed = LineString(coords)
                if not fixed.is_simple:
                    fixed = fixed.simplify(0.01, preserve_topology=True)
                fixed_geoms.append(fixed)
            else:
                fixed_geoms.append(None)
        elif geom.type == 'MultiLineString':
            fixed_parts = []
            for part in geom:
                coords = list(dict.fromkeys(part.coords))
                if len(coords) >=2:
                    fixed_part = LineString(coords)
                    if not fixed_part.is_simple:
                        fixed_part = fixed_part.simplify(0.01)
                    fixed_parts.append(fixed_part)
            fixed_geoms.append(MultiLineString(fixed_parts) if fixed_parts else None)
        else:
            fixed_geoms.append(geom)
    
    gdf_lines['geometry'] = fixed_geoms
    return gdf_lines.dropna(subset=['geometry']).reset_index(drop=True)

def fix_polygons(gdf_polygons):
    fixed_geoms = []
    for geom in gdf_polygons['geometry']:
        if geom.type == 'Polygon':
            if not geom.is_valid:
                geom = geom.buffer(0)
            exterior = geom.exterior
            if not exterior.is_closed:
                coords = list(exterior.coords)
                coords.append(coords[0])
                geom = Polygon(coords, [list(interior.coords) for interior in geom.interiors])
            fixed_geoms.append(geom)
        else:
            fixed_geoms.append(geom)
    gdf_polygons['geometry'] = fixed_geoms
    return gdf_polygons

内容的提问来源于stack exchange，提问作者Rahul Roy Verma