sladro
90ba65e484
- 新增XCAF转换器，直接从STP到GLB保留完整装配层级 - 修复HTTP API与CLI参数不一致问题 - 修复静默吞没错误的问题，遵循快速失败原则 - 清理旧文件，整理测试文件到tests目录 - 添加URL下载支持，可直接转换远程STP文件 - 更新文档，准确描述XCAF装配保留功能 技术改进： - 使用STEPCAFControl_Reader读取带装配信息的STEP文件 - 通过RWGltf_CafWriter直接导出GLB，无需STL中间格式 - 支持CPU多线程并行三角化 - HTTP API和CLI使用完全一致的转换参数 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
179 lines
5.9 KiB
Python
179 lines
5.9 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
网络下载工具模块
|
||
提供URL验证和文件下载功能
|
||
"""
|
||
|
||
import os
|
||
import tempfile
|
||
import urllib.request
|
||
import urllib.parse
|
||
from typing import Callable, Optional, Tuple
|
||
from urllib.error import URLError, HTTPError
|
||
|
||
|
||
def is_url(path: str) -> bool:
|
||
"""
|
||
判断输入是否为URL
|
||
|
||
Args:
|
||
path: 输入路径或URL
|
||
|
||
Returns:
|
||
bool: 是否为有效的HTTP/HTTPS URL
|
||
"""
|
||
if not path:
|
||
return False
|
||
|
||
try:
|
||
parsed = urllib.parse.urlparse(path)
|
||
return parsed.scheme.lower() in ['http', 'https'] and bool(parsed.netloc)
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
def validate_url(url: str) -> Tuple[bool, str]:
|
||
"""
|
||
验证URL的有效性和可访问性
|
||
|
||
Args:
|
||
url: 要验证的URL
|
||
|
||
Returns:
|
||
Tuple[bool, str]: (是否有效, 错误信息)
|
||
"""
|
||
if not is_url(url):
|
||
return False, "输入不是有效的HTTP/HTTPS URL"
|
||
|
||
try:
|
||
# 发送HEAD请求检查URL可访问性
|
||
request = urllib.request.Request(url, method='HEAD')
|
||
with urllib.request.urlopen(request, timeout=10) as response:
|
||
content_type = response.headers.get('content-type', '').lower()
|
||
|
||
# 检查是否可能是STP文件
|
||
if 'application/octet-stream' not in content_type and 'application/step' not in content_type:
|
||
# 如果没有明确的内容类型，通过URL扩展名判断
|
||
if not url.lower().endswith(('.stp', '.step')):
|
||
return False, "URL指向的文件可能不是STP格式"
|
||
|
||
return True, ""
|
||
|
||
except HTTPError as e:
|
||
return False, f"HTTP错误 {e.code}: {e.reason}"
|
||
except URLError as e:
|
||
return False, f"URL错误: {e.reason}"
|
||
except Exception as e:
|
||
return False, f"验证失败: {str(e)}"
|
||
|
||
|
||
def download_file(url: str, progress_callback: Optional[Callable[[int, str], None]] = None) -> str:
|
||
"""
|
||
下载文件到临时目录
|
||
|
||
Args:
|
||
url: 要下载的文件URL
|
||
progress_callback: 进度回调函数
|
||
|
||
Returns:
|
||
str: 下载文件的本地路径
|
||
|
||
Raises:
|
||
Exception: 下载失败时抛出异常
|
||
"""
|
||
def update_progress(progress: int, message: str) -> None:
|
||
"""更新进度"""
|
||
if progress_callback:
|
||
progress_callback(progress, message)
|
||
|
||
update_progress(0, f"开始下载文件: {url}")
|
||
|
||
# 验证URL
|
||
is_valid, error_msg = validate_url(url)
|
||
if not is_valid:
|
||
raise ValueError(f"URL验证失败: {error_msg}")
|
||
|
||
try:
|
||
# 获取文件名和扩展名
|
||
parsed_url = urllib.parse.urlparse(url)
|
||
filename = os.path.basename(parsed_url.path)
|
||
|
||
# 如果URL没有文件名，生成一个默认名
|
||
if not filename or '.' not in filename:
|
||
filename = "downloaded_model.stp"
|
||
|
||
# 确保文件扩展名是STP格式
|
||
if not filename.lower().endswith(('.stp', '.step')):
|
||
base_name = os.path.splitext(filename)[0]
|
||
filename = f"{base_name}.stp"
|
||
|
||
update_progress(10, f"准备下载文件: {filename}")
|
||
|
||
# 创建临时文件
|
||
temp_dir = tempfile.gettempdir()
|
||
temp_path = os.path.join(temp_dir, f"stp2glb_{os.getpid()}_{filename}")
|
||
|
||
update_progress(15, f"临时文件路径: {temp_path}")
|
||
|
||
# 下载文件
|
||
request = urllib.request.Request(url)
|
||
|
||
with urllib.request.urlopen(request) as response:
|
||
total_size = int(response.headers.get('content-length', 0))
|
||
downloaded = 0
|
||
|
||
with open(temp_path, 'wb') as f:
|
||
while True:
|
||
chunk = response.read(8192) # 8KB chunks
|
||
if not chunk:
|
||
break
|
||
|
||
f.write(chunk)
|
||
downloaded += len(chunk)
|
||
|
||
# 计算并更新进度
|
||
if total_size > 0:
|
||
progress = 15 + int((downloaded / total_size) * 70) # 15-85%的进度范围
|
||
update_progress(progress, f"下载进度: {downloaded / 1024 / 1024:.1f}MB / {total_size / 1024 / 1024:.1f}MB")
|
||
else:
|
||
# 如果不知道总大小，显示已下载的字节数
|
||
update_progress(50, f"已下载: {downloaded / 1024 / 1024:.1f}MB")
|
||
|
||
# 验证下载的文件
|
||
if not os.path.exists(temp_path) or os.path.getsize(temp_path) == 0:
|
||
raise ValueError("下载的文件为空或不存在")
|
||
|
||
file_size_mb = os.path.getsize(temp_path) / 1024 / 1024
|
||
update_progress(90, f"文件下载完成: {file_size_mb:.1f}MB")
|
||
|
||
return temp_path
|
||
|
||
except HTTPError as e:
|
||
raise Exception(f"HTTP下载错误 {e.code}: {e.reason}")
|
||
except URLError as e:
|
||
raise Exception(f"网络错误: {e.reason}")
|
||
except Exception as e:
|
||
# 清理可能创建的临时文件
|
||
if 'temp_path' in locals() and os.path.exists(temp_path):
|
||
try:
|
||
os.remove(temp_path)
|
||
except Exception as cleanup_error:
|
||
# 记录清理失败但继续抛出原始异常
|
||
print(f"警告: 无法清理临时文件 {temp_path}: {cleanup_error}")
|
||
raise Exception(f"下载失败: {str(e)}")
|
||
|
||
|
||
def cleanup_downloaded_file(file_path: str) -> None:
|
||
"""
|
||
清理下载的临时文件
|
||
|
||
Args:
|
||
file_path: 要清理的文件路径
|
||
"""
|
||
if file_path and os.path.exists(file_path):
|
||
try:
|
||
# 只清理临时目录中的文件，避免误删用户文件
|
||
if tempfile.gettempdir() in os.path.abspath(file_path):
|
||
os.remove(file_path)
|
||
except Exception as e:
|
||
raise RuntimeError(f"无法清理下载的临时文件 {file_path}: {str(e)}") |