405 lines
15 KiB
Python
405 lines
15 KiB
Python
"""
|
||
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)
|