""" DWG 转换服务 REST API（ODA 方案） 使用 ODA File Converter 将 DWG 转换为 OBJ，再用 obj2gltf 转为 GLB 端点: POST /convert - 上传 DWG 文件并转换 GET /health - 健康检查 依赖: - ODA File Converter (需手动安装) - obj2gltf (npm install -g obj2gltf) """ import os import sys import uuid import json import subprocess import tempfile import shutil from pathlib import Path from flask import Flask, request, jsonify, send_file from werkzeug.utils import secure_filename app = Flask(__name__) # 配置 UPLOAD_FOLDER = '/app/uploads' OUTPUT_FOLDER = '/app/outputs' ALLOWED_EXTENSIONS = {'dwg', 'stl', 'obj'} MAX_CONTENT_LENGTH = 100 * 1024 * 1024 # 100MB # ODA File Converter 路径（Linux） ODA_CONVERTER = os.environ.get('ODA_CONVERTER_PATH', '/usr/bin/ODAFileConverter') app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER app.config['MAX_CONTENT_LENGTH'] = MAX_CONTENT_LENGTH def allowed_file(filename): """检查文件扩展名""" return '.' in filename and \ filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS def check_oda_available(): """检查 ODA File Converter 是否可用""" return os.path.exists(ODA_CONVERTER) or shutil.which('ODAFileConverter') def convert_dwg_to_glb(input_path: str, output_path: str) -> dict: """ 使用 ODA 将 DWG 转换为 GLB 流程: DWG -> DXF (ODA) -> 解析所有实体类型 -> OBJ -> GLB (obj2gltf) """ result = { 'success': False, 'error': None, 'stats': {} } temp_dir = tempfile.mkdtemp(prefix='dwg_convert_') try: input_dir = os.path.dirname(input_path) input_name = os.path.basename(input_path) # Step 1: DWG -> DXF (使用 ODA File Converter) oda_path = ODA_CONVERTER if os.path.exists(ODA_CONVERTER) else shutil.which('ODAFileConverter') if not oda_path: result['error'] = 'ODA File Converter 未安装' return result cmd = [ oda_path, input_dir, temp_dir, 'ACAD2018', 'DXF', '0', '1', input_name ] print(f'[ODA] 执行命令: {" ".join(cmd)}') proc = subprocess.run(cmd, capture_output=True, text=True, timeout=120) if proc.returncode != 0: result['error'] = f'ODA 转换失败: {proc.stderr or proc.stdout}' return result # 查找生成的 DXF 文件 dxf_files = list(Path(temp_dir).glob('*.dxf')) if not dxf_files: result['error'] = 'ODA 未生成 DXF 文件' return result dxf_path = str(dxf_files[0]) print(f'[ODA] DXF 文件生成: {dxf_path}') # Step 2: DXF -> OBJ (使用 ezdxf 提取所有 3D 实体) obj_path = os.path.join(temp_dir, 'model.obj') try: import ezdxf from ezdxf.addons import meshex import trimesh import numpy as np doc = ezdxf.readfile(dxf_path) msp = doc.modelspace() all_meshes = [] entity_stats = {} # 遍历所有实体 for entity in msp: etype = entity.dxftype() entity_stats[etype] = entity_stats.get(etype, 0) + 1 try: if etype == '3DFACE': pts = [entity.dxf.vtx0, entity.dxf.vtx1, entity.dxf.vtx2] if hasattr(entity.dxf, 'vtx3') and entity.dxf.vtx3 != entity.dxf.vtx2: pts.append(entity.dxf.vtx3) verts = [[p[0], p[1], p[2] if len(p) > 2 else 0] for p in pts] if len(pts) == 3: faces = [[0, 1, 2]] else: faces = [[0, 1, 2], [0, 2, 3]] mesh = trimesh.Trimesh(vertices=verts, faces=faces) all_meshes.append(mesh) elif etype == '3DSOLID': # 3D 实体 - 尝试使用 ezdxf 的 meshex 提取 try: from ezdxf.render import MeshBuilder builder = MeshBuilder() entity.render_mesh(builder) if builder.vertices: mesh = trimesh.Trimesh( vertices=np.array(builder.vertices), faces=np.array(builder.faces) if builder.faces else [] ) all_meshes.append(mesh) except Exception as e: print(f'[ODA] 3DSOLID 提取失败: {e}') elif etype == 'MESH': try: verts = list(entity.vertices) faces_data = list(entity.faces) if verts and faces_data: mesh = trimesh.Trimesh( vertices=np.array(verts), faces=np.array([f[:3] for f in faces_data if len(f) >= 3]) ) all_meshes.append(mesh) except Exception as e: print(f'[ODA] MESH 提取失败: {e}') elif etype == 'POLYFACE': # Polyface mesh try: from ezdxf.entities import Polyface if hasattr(entity, 'vertices') and hasattr(entity, 'faces'): verts = [list(v.dxf.location) for v in entity.vertices] faces_data = [list(f.indices()) for f in entity.faces] if verts and faces_data: mesh = trimesh.Trimesh( vertices=np.array(verts), faces=np.array([f[:3] for f in faces_data if len(f) >= 3]) ) all_meshes.append(mesh) except Exception as e: print(f'[ODA] POLYFACE 提取失败: {e}') elif etype in ('SURFACE', 'PLANESURFACE'): # 曲面实体 try: from ezdxf.render import MeshBuilder builder = MeshBuilder() entity.render_mesh(builder) if builder.vertices: mesh = trimesh.Trimesh( vertices=np.array(builder.vertices), faces=np.array(builder.faces) if builder.faces else [] ) all_meshes.append(mesh) except Exception as e: print(f'[ODA] SURFACE 提取失败: {e}') except Exception as e: print(f'[ODA] 实体 {etype} 处理异常: {e}') continue print(f'[ODA] 实体统计: {entity_stats}') print(f'[ODA] 成功提取 {len(all_meshes)} 个网格') if not all_meshes: # 检查是否有 3DSOLID 实体 if entity_stats.get('3DSOLID', 0) > 0: result['error'] = ( f'检测到 {entity_stats["3DSOLID"]} 个 3DSOLID 实体（布尔运算模型）。\n' '3DSOLID 格式暂不支持自动转换，请在 AutoCAD 中执行以下操作：\n' '1. 选择所有实体 → 输入命令 EXPORT → 选择 STL 或 OBJ 格式导出\n' '2. 或使用 FACETRES 命令调整精度后导出\n' '3. 然后上传导出的 STL/GLB 文件' ) else: result['error'] = f'DXF 文件中没有可转换的 3D 几何体。实体类型: {entity_stats}' return result # 合并所有网格 combined = trimesh.util.concatenate(all_meshes) combined.export(obj_path) print(f'[ODA] OBJ 文件生成: {obj_path}') result['stats']['mesh_count'] = len(all_meshes) result['stats']['vertex_count'] = len(combined.vertices) result['stats']['face_count'] = len(combined.faces) except ImportError as e: result['error'] = f'依赖库未安装: {e}' return result except Exception as e: result['error'] = f'DXF 解析失败: {str(e)}' import traceback print(traceback.format_exc()) return result # Step 3: OBJ -> GLB (使用 obj2gltf) obj2gltf_path = shutil.which('obj2gltf') if not obj2gltf_path: obj2gltf_path = 'npx' cmd = [obj2gltf_path, 'obj2gltf', '-i', obj_path, '-o', output_path, '-b'] else: cmd = [obj2gltf_path, '-i', obj_path, '-o', output_path, '-b'] print(f'[ODA] 执行 obj2gltf: {" ".join(cmd)}') proc = subprocess.run(cmd, capture_output=True, text=True, timeout=60) if proc.returncode != 0 or not os.path.exists(output_path): result['error'] = f'obj2gltf 转换失败: {proc.stderr or proc.stdout}' return result output_size = os.path.getsize(output_path) result['stats']['output_size_bytes'] = output_size result['stats']['output_size_mb'] = round(output_size / (1024 * 1024), 2) result['success'] = True print(f'[ODA] 转换成功: {output_path} ({result["stats"]["output_size_mb"]} MB)') except subprocess.TimeoutExpired: result['error'] = '转换超时' except Exception as e: result['error'] = f'转换过程出错: {str(e)}' import traceback print(traceback.format_exc()) finally: if os.path.exists(temp_dir): shutil.rmtree(temp_dir) return result @app.route('/health', methods=['GET']) def health_check(): """健康检查""" oda_available = check_oda_available() return jsonify({ 'status': 'healthy' if oda_available else 'degraded', 'service': 'dwg-converter-oda', 'version': '1.0.0', 'oda_available': oda_available, 'oda_path': ODA_CONVERTER if os.path.exists(ODA_CONVERTER) else shutil.which('ODAFileConverter') }) def convert_stl_obj_to_glb(input_path: str, output_path: str) -> dict: """ 将 STL/OBJ 文件转换为 GLB """ result = { 'success': False, 'error': None, 'stats': {} } try: file_ext = os.path.splitext(input_path)[1].lower() # 使用 obj2gltf 转换 obj2gltf_path = shutil.which('obj2gltf') if file_ext == '.stl': # STL -> OBJ -> GLB (使用 trimesh) import trimesh mesh = trimesh.load(input_path) temp_obj = input_path.replace('.stl', '.obj') mesh.export(temp_obj) if obj2gltf_path: cmd = [obj2gltf_path, '-i', temp_obj, '-o', output_path, '-b'] else: cmd = ['npx', 'obj2gltf', '-i', temp_obj, '-o', output_path, '-b'] proc = subprocess.run(cmd, capture_output=True, text=True, timeout=60) os.remove(temp_obj) elif file_ext == '.obj': # OBJ -> GLB if obj2gltf_path: cmd = [obj2gltf_path, '-i', input_path, '-o', output_path, '-b'] else: cmd = ['npx', 'obj2gltf', '-i', input_path, '-o', output_path, '-b'] proc = subprocess.run(cmd, capture_output=True, text=True, timeout=60) if os.path.exists(output_path): output_size = os.path.getsize(output_path) result['stats']['output_size_bytes'] = output_size result['stats']['output_size_mb'] = round(output_size / (1024 * 1024), 2) result['success'] = True print(f'[Converter] 转换成功: {output_path}') else: result['error'] = f'转换失败: {proc.stderr or proc.stdout}' except Exception as e: result['error'] = f'转换出错: {str(e)}' import traceback print(traceback.format_exc()) return result @app.route('/convert', methods=['POST']) def convert(): """上传 DWG/STL/OBJ 文件并转换为 GLB""" # 检查文件 if 'file' not in request.files: return jsonify({'error': '未提供文件'}), 400 file = request.files['file'] if file.filename == '': return jsonify({'error': '未选择文件'}), 400 if not allowed_file(file.filename): return jsonify({'error': '不支持的文件格式，支持 .dwg/.stl/.obj'}), 400 # 生成文件路径 task_id = str(uuid.uuid4()) filename = secure_filename(file.filename) input_path = os.path.join(UPLOAD_FOLDER, f'{task_id}_{filename}') output_path = os.path.join(OUTPUT_FOLDER, f'{task_id}.glb') # 保存上传文件 os.makedirs(UPLOAD_FOLDER, exist_ok=True) os.makedirs(OUTPUT_FOLDER, exist_ok=True) file.save(input_path) file_ext = os.path.splitext(filename)[1].lower() try: # 根据文件类型选择转换方法 if file_ext == '.dwg': # 检查 ODA 是否可用 if not check_oda_available(): return jsonify({ 'error': 'ODA File Converter 未安装' }), 503 result = convert_dwg_to_glb(input_path, output_path) else: # STL/OBJ 直接转换 result = convert_stl_obj_to_glb(input_path, output_path) if result['success']: return send_file( output_path, mimetype='model/gltf-binary', as_attachment=True, download_name='converted.glb' ) else: return jsonify({'error': result['error']}), 500 finally: # 清理输入文件 if os.path.exists(input_path): os.remove(input_path) if __name__ == '__main__': os.makedirs(UPLOAD_FOLDER, exist_ok=True) os.makedirs(OUTPUT_FOLDER, exist_ok=True) app.run(host='0.0.0.0', port=3001, debug=True)