feat: 实现文件重命名管理器、CAD软件配置及初始API端点。
This commit is contained in:
parent
a0066a6fe9
commit
a5bfd778a3
199
app/api/v1/files.py
Normal file
199
app/api/v1/files.py
Normal file
@ -0,0 +1,199 @@
|
||||
"""
|
||||
文件管理API路由
|
||||
提供CAD文件列表和下载功能
|
||||
"""
|
||||
from fastapi import APIRouter, HTTPException
|
||||
from fastapi.responses import FileResponse, StreamingResponse
|
||||
from typing import List
|
||||
from pathlib import Path
|
||||
import os
|
||||
import zipfile
|
||||
import io
|
||||
from app.config import settings, software_config
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
# CAD文件扩展名映射
|
||||
CAD_EXTENSIONS = {
|
||||
'creo': ['.prt', '.asm', '.drw'], # Creo文件，包括.prt.1等版本文件
|
||||
'pdms': ['.rvm', '.dri'], # PDMS文件
|
||||
'revit': ['.rvt', '.rfa', '.rte'] # Revit文件
|
||||
}
|
||||
|
||||
|
||||
def get_cad_files_path() -> Path:
|
||||
"""获取CAD文件存储路径"""
|
||||
cad_path = software_config.get_cad_files_path()
|
||||
path = Path(cad_path)
|
||||
|
||||
if not path.exists():
|
||||
raise HTTPException(status_code=404, detail=f"CAD文件路径不存在: {cad_path}")
|
||||
|
||||
return path
|
||||
|
||||
|
||||
def is_cad_file(filename: str) -> bool:
|
||||
"""判断文件是否为CAD文件（包括版本号后缀）"""
|
||||
filename_lower = filename.lower()
|
||||
|
||||
# 检查所有CAD扩展名
|
||||
for software, extensions in CAD_EXTENSIONS.items():
|
||||
for ext in extensions:
|
||||
# 检查标准扩展名
|
||||
if filename_lower.endswith(ext):
|
||||
return True
|
||||
# 检查带版本号的扩展名（如 .prt.1, .prt.2 等）
|
||||
if ext in filename_lower and '.' in filename_lower.split(ext)[-1]:
|
||||
# 验证版本号部分是否为数字
|
||||
version_part = filename_lower.split(ext)[-1]
|
||||
if version_part.startswith('.') and version_part[1:].isdigit():
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
|
||||
def scan_cad_files(base_path: Path) -> List[dict]:
|
||||
"""扫描目录下的所有CAD文件"""
|
||||
cad_files = []
|
||||
|
||||
try:
|
||||
for root, dirs, files in os.walk(base_path):
|
||||
for file in files:
|
||||
if is_cad_file(file):
|
||||
file_path = Path(root) / file
|
||||
relative_path = file_path.relative_to(base_path)
|
||||
|
||||
# 获取文件信息
|
||||
stat = file_path.stat()
|
||||
|
||||
cad_files.append({
|
||||
'filename': file,
|
||||
'relative_path': str(relative_path).replace('\\', '/'),
|
||||
'absolute_path': str(file_path),
|
||||
'size': stat.st_size,
|
||||
'modified_time': stat.st_mtime,
|
||||
'extension': get_file_extension(file)
|
||||
})
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=500, detail=f"扫描文件失败: {str(e)}")
|
||||
|
||||
return cad_files
|
||||
|
||||
|
||||
def get_file_extension(filename: str) -> str:
|
||||
"""获取文件扩展名（包括版本号）"""
|
||||
filename_lower = filename.lower()
|
||||
|
||||
# 检查是否有版本号
|
||||
for software, extensions in CAD_EXTENSIONS.items():
|
||||
for ext in extensions:
|
||||
if ext in filename_lower:
|
||||
idx = filename_lower.find(ext)
|
||||
return filename[idx:]
|
||||
|
||||
# 如果没有匹配，返回标准扩展名
|
||||
return Path(filename).suffix
|
||||
|
||||
|
||||
@router.get("/files/list")
|
||||
async def list_cad_files():
|
||||
"""
|
||||
获取CAD文件列表
|
||||
返回所有Creo、PDMS、Revit格式的文件
|
||||
"""
|
||||
base_path = get_cad_files_path()
|
||||
files = scan_cad_files(base_path)
|
||||
|
||||
return {
|
||||
'success': True,
|
||||
'base_path': str(base_path),
|
||||
'total_count': len(files),
|
||||
'files': files
|
||||
}
|
||||
|
||||
|
||||
@router.get("/files/download/{file_path:path}")
|
||||
async def download_file(file_path: str):
|
||||
"""
|
||||
下载单个文件
|
||||
|
||||
Args:
|
||||
file_path: 文件的相对路径
|
||||
"""
|
||||
base_path = get_cad_files_path()
|
||||
full_path = base_path / file_path
|
||||
|
||||
# 安全检查：确保文件在允许的目录内
|
||||
try:
|
||||
full_path = full_path.resolve()
|
||||
base_path = base_path.resolve()
|
||||
|
||||
if not str(full_path).startswith(str(base_path)):
|
||||
raise HTTPException(status_code=403, detail="访问被拒绝")
|
||||
except Exception:
|
||||
raise HTTPException(status_code=400, detail="无效的文件路径")
|
||||
|
||||
# 检查文件是否存在
|
||||
if not full_path.exists():
|
||||
raise HTTPException(status_code=404, detail="文件不存在")
|
||||
|
||||
if not full_path.is_file():
|
||||
raise HTTPException(status_code=400, detail="不是有效的文件")
|
||||
|
||||
# 检查是否为CAD文件
|
||||
if not is_cad_file(full_path.name):
|
||||
raise HTTPException(status_code=403, detail="只能下载CAD文件")
|
||||
|
||||
return FileResponse(
|
||||
path=str(full_path),
|
||||
filename=full_path.name,
|
||||
media_type='application/octet-stream'
|
||||
)
|
||||
|
||||
|
||||
@router.post("/files/download/batch")
|
||||
async def download_batch_files(file_paths: List[str]):
|
||||
"""
|
||||
批量下载文件（打包为ZIP）
|
||||
|
||||
Args:
|
||||
file_paths: 文件相对路径列表
|
||||
"""
|
||||
if not file_paths:
|
||||
raise HTTPException(status_code=400, detail="文件列表不能为空")
|
||||
|
||||
base_path = get_cad_files_path()
|
||||
|
||||
# 创建内存中的ZIP文件
|
||||
zip_buffer = io.BytesIO()
|
||||
|
||||
with zipfile.ZipFile(zip_buffer, 'w', zipfile.ZIP_DEFLATED) as zip_file:
|
||||
for file_path in file_paths:
|
||||
full_path = base_path / file_path
|
||||
|
||||
# 安全检查
|
||||
try:
|
||||
full_path = full_path.resolve()
|
||||
base_path_resolved = base_path.resolve()
|
||||
|
||||
if not str(full_path).startswith(str(base_path_resolved)):
|
||||
continue # 跳过不安全的路径
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
# 检查文件是否存在且为CAD文件
|
||||
if full_path.exists() and full_path.is_file() and is_cad_file(full_path.name):
|
||||
# 使用相对路径作为ZIP内的路径
|
||||
arcname = file_path.replace('\\', '/')
|
||||
zip_file.write(str(full_path), arcname=arcname)
|
||||
|
||||
# 重置缓冲区位置
|
||||
zip_buffer.seek(0)
|
||||
|
||||
return StreamingResponse(
|
||||
zip_buffer,
|
||||
media_type='application/zip',
|
||||
headers={
|
||||
'Content-Disposition': 'attachment; filename="cad_files.zip"'
|
||||
}
|
||||
)
|
||||
@ -25,6 +25,14 @@ class WSMessageType:
|
||||
GET_LOG_STATS = "get_log_stats"
|
||||
CLEANUP_LOGS = "cleanup_logs"
|
||||
GET_OPERATION_TYPES = "get_operation_types"
|
||||
# 文件管理相关消息类型
|
||||
GET_FILE_LIST = "get_file_list"
|
||||
DOWNLOAD_FILE = "download_file"
|
||||
DOWNLOAD_BATCH = "download_batch"
|
||||
# 文件重命名相关消息类型
|
||||
GET_RENAME_STRATEGIES = "get_rename_strategies"
|
||||
PREVIEW_RENAME = "preview_rename"
|
||||
RENAME_FILES = "rename_files"
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
@ -116,6 +124,12 @@ async def handle_client_message(message: dict, client_id: str, user_id: str):
|
||||
- get_log_stats: 获取日志统计信息
|
||||
- cleanup_logs: 清理过期日志
|
||||
- get_operation_types: 获取操作类型列表
|
||||
- get_file_list: 获取CAD文件列表
|
||||
- download_file: 获取单个文件下载URL
|
||||
- download_batch: 获取批量文件下载URL
|
||||
- get_rename_strategies: 获取可用的重命名策略列表
|
||||
- preview_rename: 预览重命名结果
|
||||
- rename_files: 执行批量重命名
|
||||
"""
|
||||
from app.core.software_manager import software_manager
|
||||
from app.core.log_manager import log_manager
|
||||
@ -499,6 +513,255 @@ async def handle_client_message(message: dict, client_id: str, user_id: str):
|
||||
"message": f"获取操作类型失败: {str(e)}",
|
||||
"timestamp": websocket_manager._get_timestamp()
|
||||
}, client_id)
|
||||
|
||||
elif message_type == WSMessageType.GET_FILE_LIST:
|
||||
# 获取CAD文件列表
|
||||
try:
|
||||
from app.api.v1.files import get_cad_files_path, scan_cad_files
|
||||
|
||||
base_path = get_cad_files_path()
|
||||
files = scan_cad_files(base_path)
|
||||
|
||||
await websocket_manager.send_personal_message({
|
||||
"type": MessageType.INFO,
|
||||
"message": "获取文件列表成功",
|
||||
"data": {
|
||||
"base_path": str(base_path),
|
||||
"total_count": len(files),
|
||||
"files": files
|
||||
},
|
||||
"timestamp": websocket_manager._get_timestamp()
|
||||
}, client_id)
|
||||
|
||||
except Exception as e:
|
||||
await websocket_manager.send_personal_message({
|
||||
"type": MessageType.ERROR,
|
||||
"message": f"获取文件列表失败: {str(e)}",
|
||||
"timestamp": websocket_manager._get_timestamp()
|
||||
}, client_id)
|
||||
|
||||
elif message_type == WSMessageType.DOWNLOAD_FILE:
|
||||
# 获取单个文件下载URL
|
||||
file_path = message.get("file_path")
|
||||
if not file_path:
|
||||
await websocket_manager.send_personal_message({
|
||||
"type": MessageType.ERROR,
|
||||
"message": "缺少参数: file_path",
|
||||
"timestamp": websocket_manager._get_timestamp()
|
||||
}, client_id)
|
||||
return
|
||||
|
||||
try:
|
||||
from app.api.v1.files import get_cad_files_path, is_cad_file
|
||||
from pathlib import Path
|
||||
|
||||
base_path = get_cad_files_path()
|
||||
full_path = base_path / file_path
|
||||
|
||||
# 安全检查
|
||||
full_path = full_path.resolve()
|
||||
base_path = base_path.resolve()
|
||||
|
||||
if not str(full_path).startswith(str(base_path)):
|
||||
raise ValueError("访问被拒绝")
|
||||
|
||||
if not full_path.exists():
|
||||
raise FileNotFoundError("文件不存在")
|
||||
|
||||
if not is_cad_file(full_path.name):
|
||||
raise ValueError("只能下载CAD文件")
|
||||
|
||||
# 返回下载URL（使用HTTP接口）
|
||||
download_url = f"/api/v1/files/download/{file_path}"
|
||||
|
||||
await websocket_manager.send_personal_message({
|
||||
"type": MessageType.INFO,
|
||||
"message": "文件下载链接已生成",
|
||||
"data": {
|
||||
"file_path": file_path,
|
||||
"filename": full_path.name,
|
||||
"download_url": download_url,
|
||||
"file_size": full_path.stat().st_size
|
||||
},
|
||||
"timestamp": websocket_manager._get_timestamp()
|
||||
}, client_id)
|
||||
|
||||
except Exception as e:
|
||||
await websocket_manager.send_personal_message({
|
||||
"type": MessageType.ERROR,
|
||||
"message": f"生成下载链接失败: {str(e)}",
|
||||
"timestamp": websocket_manager._get_timestamp()
|
||||
}, client_id)
|
||||
|
||||
elif message_type == WSMessageType.DOWNLOAD_BATCH:
|
||||
# 获取批量文件下载URL
|
||||
file_paths = message.get("file_paths")
|
||||
if not file_paths or not isinstance(file_paths, list):
|
||||
await websocket_manager.send_personal_message({
|
||||
"type": MessageType.ERROR,
|
||||
"message": "缺少参数: file_paths (必须是数组)",
|
||||
"timestamp": websocket_manager._get_timestamp()
|
||||
}, client_id)
|
||||
return
|
||||
|
||||
try:
|
||||
# 返回批量下载URL（使用HTTP接口）
|
||||
download_url = "/api/v1/files/download/batch"
|
||||
|
||||
await websocket_manager.send_personal_message({
|
||||
"type": MessageType.INFO,
|
||||
"message": "批量下载链接已生成",
|
||||
"data": {
|
||||
"file_count": len(file_paths),
|
||||
"file_paths": file_paths,
|
||||
"download_url": download_url,
|
||||
"method": "POST",
|
||||
"note": "请使用POST方法，将file_paths作为JSON数组发送到此URL"
|
||||
},
|
||||
"timestamp": websocket_manager._get_timestamp()
|
||||
}, client_id)
|
||||
|
||||
except Exception as e:
|
||||
await websocket_manager.send_personal_message({
|
||||
"type": MessageType.ERROR,
|
||||
"message": f"生成批量下载链接失败: {str(e)}",
|
||||
"timestamp": websocket_manager._get_timestamp()
|
||||
}, client_id)
|
||||
|
||||
elif message_type == WSMessageType.GET_RENAME_STRATEGIES:
|
||||
# 获取可用的重命名策略列表
|
||||
try:
|
||||
from app.core.rename_manager import rename_manager
|
||||
|
||||
strategies = rename_manager.get_available_strategies()
|
||||
|
||||
await websocket_manager.send_personal_message({
|
||||
"type": MessageType.INFO,
|
||||
"message": "获取重命名策略成功",
|
||||
"data": {
|
||||
"strategies": strategies,
|
||||
"total_count": len(strategies)
|
||||
},
|
||||
"timestamp": websocket_manager._get_timestamp()
|
||||
}, client_id)
|
||||
|
||||
except Exception as e:
|
||||
await websocket_manager.send_personal_message({
|
||||
"type": MessageType.ERROR,
|
||||
"message": f"获取重命名策略失败: {str(e)}",
|
||||
"timestamp": websocket_manager._get_timestamp()
|
||||
}, client_id)
|
||||
|
||||
elif message_type == WSMessageType.PREVIEW_RENAME:
|
||||
# 预览重命名结果
|
||||
file_paths = message.get("file_paths")
|
||||
strategy_name = message.get("strategy")
|
||||
params = message.get("params", {})
|
||||
|
||||
if not file_paths or not isinstance(file_paths, list):
|
||||
await websocket_manager.send_personal_message({
|
||||
"type": MessageType.ERROR,
|
||||
"message": "缺少参数: file_paths (必须是数组)",
|
||||
"timestamp": websocket_manager._get_timestamp()
|
||||
}, client_id)
|
||||
return
|
||||
|
||||
if not strategy_name:
|
||||
await websocket_manager.send_personal_message({
|
||||
"type": MessageType.ERROR,
|
||||
"message": "缺少参数: strategy",
|
||||
"timestamp": websocket_manager._get_timestamp()
|
||||
}, client_id)
|
||||
return
|
||||
|
||||
try:
|
||||
from app.core.rename_manager import rename_manager
|
||||
from pathlib import Path
|
||||
|
||||
# 提取文件名
|
||||
filenames = [Path(fp).name for fp in file_paths]
|
||||
|
||||
# 预览重命名
|
||||
preview_results = rename_manager.preview_rename(
|
||||
filenames=filenames,
|
||||
strategy_name=strategy_name,
|
||||
params=params
|
||||
)
|
||||
|
||||
await websocket_manager.send_personal_message({
|
||||
"type": MessageType.INFO,
|
||||
"message": "重命名预览生成成功",
|
||||
"data": {
|
||||
"strategy": strategy_name,
|
||||
"params": params,
|
||||
"preview": preview_results,
|
||||
"total_count": len(preview_results)
|
||||
},
|
||||
"timestamp": websocket_manager._get_timestamp()
|
||||
}, client_id)
|
||||
|
||||
except Exception as e:
|
||||
await websocket_manager.send_personal_message({
|
||||
"type": MessageType.ERROR,
|
||||
"message": f"预览重命名失败: {str(e)}",
|
||||
"timestamp": websocket_manager._get_timestamp()
|
||||
}, client_id)
|
||||
|
||||
elif message_type == WSMessageType.RENAME_FILES:
|
||||
# 执行批量重命名
|
||||
file_paths = message.get("file_paths")
|
||||
strategy_name = message.get("strategy")
|
||||
params = message.get("params", {})
|
||||
|
||||
if not file_paths or not isinstance(file_paths, list):
|
||||
await websocket_manager.send_personal_message({
|
||||
"type": MessageType.ERROR,
|
||||
"message": "缺少参数: file_paths (必须是数组)",
|
||||
"timestamp": websocket_manager._get_timestamp()
|
||||
}, client_id)
|
||||
return
|
||||
|
||||
if not strategy_name:
|
||||
await websocket_manager.send_personal_message({
|
||||
"type": MessageType.ERROR,
|
||||
"message": "缺少参数: strategy",
|
||||
"timestamp": websocket_manager._get_timestamp()
|
||||
}, client_id)
|
||||
return
|
||||
|
||||
try:
|
||||
from app.core.rename_manager import rename_manager
|
||||
from app.api.v1.files import get_cad_files_path
|
||||
|
||||
base_path = get_cad_files_path()
|
||||
|
||||
# 执行重命名
|
||||
results = rename_manager.execute_rename(
|
||||
base_path=base_path,
|
||||
file_paths=file_paths,
|
||||
strategy_name=strategy_name,
|
||||
params=params
|
||||
)
|
||||
|
||||
await websocket_manager.send_personal_message({
|
||||
"type": MessageType.INFO,
|
||||
"message": f"批量重命名完成: 成功 {results['success_count']} 个, 失败 {results['failed_count']} 个",
|
||||
"data": {
|
||||
"strategy": strategy_name,
|
||||
"params": params,
|
||||
"success_count": results["success_count"],
|
||||
"failed_count": results["failed_count"],
|
||||
"results": results["results"]
|
||||
},
|
||||
"timestamp": websocket_manager._get_timestamp()
|
||||
}, client_id)
|
||||
|
||||
except Exception as e:
|
||||
await websocket_manager.send_personal_message({
|
||||
"type": MessageType.ERROR,
|
||||
"message": f"批量重命名失败: {str(e)}",
|
||||
"timestamp": websocket_manager._get_timestamp()
|
||||
}, client_id)
|
||||
|
||||
else:
|
||||
# 未知消息类型
|
||||
|
||||
@ -35,6 +35,9 @@ class Settings(BaseSettings):
|
||||
software_config_path: str = str(BASE_DIR / "configs" / "software_config.yaml")
|
||||
users_config_path: str = str(BASE_DIR / "configs" / "users.json")
|
||||
|
||||
# CAD文件存储路径
|
||||
cad_files_path: str = r"C:\Users\Public\Documents"
|
||||
|
||||
|
||||
# CORS配置
|
||||
cors_origins: List[str] = ["*"]
|
||||
@ -82,6 +85,15 @@ class SoftwareConfig:
|
||||
def validate_software_exists(self, software_id: str) -> bool:
|
||||
"""验证软件是否存在于配置中"""
|
||||
return software_id in self.get_software_list()
|
||||
|
||||
def get_cad_files_path(self) -> str:
|
||||
"""获取CAD文件存储路径"""
|
||||
if not self._config:
|
||||
self.load_config()
|
||||
|
||||
file_storage = self._config.get('file_storage', {})
|
||||
return file_storage.get('cad_files_path', r'C:\Users\Public\Documents')
|
||||
|
||||
|
||||
|
||||
# 创建全局配置实例
|
||||
|
||||
366
app/core/rename_manager.py
Normal file
366
app/core/rename_manager.py
Normal file
@ -0,0 +1,366 @@
|
||||
"""
|
||||
文件重命名管理器
|
||||
提供批量重命名策略和执行功能
|
||||
"""
|
||||
from pathlib import Path
|
||||
from typing import List, Dict, Optional, Callable
|
||||
from datetime import datetime
|
||||
import re
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class RenameStrategy:
|
||||
"""重命名策略基类"""
|
||||
|
||||
def __init__(self, name: str, description: str):
|
||||
self.name = name
|
||||
self.description = description
|
||||
|
||||
def apply(self, filename: str, index: int, params: dict) -> str:
|
||||
"""
|
||||
应用重命名策略
|
||||
|
||||
Args:
|
||||
filename: 原文件名
|
||||
index: 文件索引（从0开始）
|
||||
params: 策略参数
|
||||
|
||||
Returns:
|
||||
新文件名
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
class AddPrefixStrategy(RenameStrategy):
|
||||
"""添加前缀策略"""
|
||||
|
||||
def __init__(self):
|
||||
super().__init__(
|
||||
name="add_prefix",
|
||||
description="在文件名前添加前缀"
|
||||
)
|
||||
|
||||
def apply(self, filename: str, index: int, params: dict) -> str:
|
||||
prefix = params.get("prefix", "")
|
||||
return f"{prefix}{filename}"
|
||||
|
||||
|
||||
class AddSuffixStrategy(RenameStrategy):
|
||||
"""添加后缀策略（在扩展名前）"""
|
||||
|
||||
def __init__(self):
|
||||
super().__init__(
|
||||
name="add_suffix",
|
||||
description="在扩展名前添加后缀"
|
||||
)
|
||||
|
||||
def apply(self, filename: str, index: int, params: dict) -> str:
|
||||
suffix = params.get("suffix", "")
|
||||
# 分离文件名和扩展名（包括版本号）
|
||||
name_part, ext_part = self._split_name_ext(filename)
|
||||
return f"{name_part}{suffix}{ext_part}"
|
||||
|
||||
@staticmethod
|
||||
def _split_name_ext(filename: str) -> tuple:
|
||||
"""分离文件名和扩展名（保留版本号）"""
|
||||
# 处理如 file.prt.1 这样的文件
|
||||
parts = filename.split('.')
|
||||
if len(parts) > 1:
|
||||
# 检查最后一部分是否为数字（版本号）
|
||||
if parts[-1].isdigit() and len(parts) > 2:
|
||||
# 有版本号：file.prt.1 -> file, .prt.1
|
||||
name = parts[0]
|
||||
ext = '.' + '.'.join(parts[1:])
|
||||
else:
|
||||
# 无版本号：file.prt -> file, .prt
|
||||
name = '.'.join(parts[:-1])
|
||||
ext = '.' + parts[-1]
|
||||
else:
|
||||
name = filename
|
||||
ext = ''
|
||||
return name, ext
|
||||
|
||||
|
||||
class SequenceStrategy(RenameStrategy):
|
||||
"""序号重命名策略"""
|
||||
|
||||
def __init__(self):
|
||||
super().__init__(
|
||||
name="sequence",
|
||||
description="使用序号重命名文件"
|
||||
)
|
||||
|
||||
def apply(self, filename: str, index: int, params: dict) -> str:
|
||||
base_name = params.get("base_name", "file")
|
||||
start_number = params.get("start_number", 1)
|
||||
digits = params.get("digits", 3)
|
||||
separator = params.get("separator", "_")
|
||||
|
||||
# 保留扩展名
|
||||
_, ext_part = AddSuffixStrategy._split_name_ext(filename)
|
||||
|
||||
# 生成序号
|
||||
number = start_number + index
|
||||
number_str = str(number).zfill(digits)
|
||||
|
||||
return f"{base_name}{separator}{number_str}{ext_part}"
|
||||
|
||||
|
||||
class ReplaceTextStrategy(RenameStrategy):
|
||||
"""文本替换策略"""
|
||||
|
||||
def __init__(self):
|
||||
super().__init__(
|
||||
name="replace_text",
|
||||
description="替换文件名中的文本"
|
||||
)
|
||||
|
||||
def apply(self, filename: str, index: int, params: dict) -> str:
|
||||
search_text = params.get("search", "")
|
||||
replace_text = params.get("replace", "")
|
||||
case_sensitive = params.get("case_sensitive", True)
|
||||
|
||||
if not search_text:
|
||||
return filename
|
||||
|
||||
if case_sensitive:
|
||||
return filename.replace(search_text, replace_text)
|
||||
else:
|
||||
# 不区分大小写的替换
|
||||
pattern = re.compile(re.escape(search_text), re.IGNORECASE)
|
||||
return pattern.sub(replace_text, filename)
|
||||
|
||||
|
||||
class AddDateTimeStrategy(RenameStrategy):
|
||||
"""添加日期时间策略"""
|
||||
|
||||
def __init__(self):
|
||||
super().__init__(
|
||||
name="add_datetime",
|
||||
description="添加日期时间戳"
|
||||
)
|
||||
|
||||
def apply(self, filename: str, index: int, params: dict) -> str:
|
||||
format_str = params.get("format", "%Y%m%d_%H%M%S")
|
||||
position = params.get("position", "suffix") # prefix or suffix
|
||||
separator = params.get("separator", "_")
|
||||
|
||||
timestamp = datetime.now().strftime(format_str)
|
||||
name_part, ext_part = AddSuffixStrategy._split_name_ext(filename)
|
||||
|
||||
if position == "prefix":
|
||||
return f"{timestamp}{separator}{name_part}{ext_part}"
|
||||
else:
|
||||
return f"{name_part}{separator}{timestamp}{ext_part}"
|
||||
|
||||
|
||||
class ChangeCaseStrategy(RenameStrategy):
|
||||
"""大小写转换策略"""
|
||||
|
||||
def __init__(self):
|
||||
super().__init__(
|
||||
name="change_case",
|
||||
description="转换文件名大小写"
|
||||
)
|
||||
|
||||
def apply(self, filename: str, index: int, params: dict) -> str:
|
||||
case_type = params.get("case_type", "lower") # lower, upper, title
|
||||
|
||||
name_part, ext_part = AddSuffixStrategy._split_name_ext(filename)
|
||||
|
||||
if case_type == "lower":
|
||||
name_part = name_part.lower()
|
||||
elif case_type == "upper":
|
||||
name_part = name_part.upper()
|
||||
elif case_type == "title":
|
||||
name_part = name_part.title()
|
||||
|
||||
return f"{name_part}{ext_part}"
|
||||
|
||||
|
||||
class RenameManager:
|
||||
"""重命名管理器"""
|
||||
|
||||
def __init__(self):
|
||||
self.strategies: Dict[str, RenameStrategy] = {}
|
||||
self._register_default_strategies()
|
||||
|
||||
def _register_default_strategies(self):
|
||||
"""注册默认策略"""
|
||||
strategies = [
|
||||
AddPrefixStrategy(),
|
||||
AddSuffixStrategy(),
|
||||
SequenceStrategy(),
|
||||
ReplaceTextStrategy(),
|
||||
AddDateTimeStrategy(),
|
||||
ChangeCaseStrategy()
|
||||
]
|
||||
|
||||
for strategy in strategies:
|
||||
self.strategies[strategy.name] = strategy
|
||||
|
||||
def get_available_strategies(self) -> List[Dict[str, str]]:
|
||||
"""获取可用的重命名策略列表"""
|
||||
return [
|
||||
{
|
||||
"name": strategy.name,
|
||||
"description": strategy.description
|
||||
}
|
||||
for strategy in self.strategies.values()
|
||||
]
|
||||
|
||||
def preview_rename(
|
||||
self,
|
||||
filenames: List[str],
|
||||
strategy_name: str,
|
||||
params: dict
|
||||
) -> List[Dict[str, str]]:
|
||||
"""
|
||||
预览重命名结果
|
||||
|
||||
Args:
|
||||
filenames: 文件名列表
|
||||
strategy_name: 策略名称
|
||||
params: 策略参数
|
||||
|
||||
Returns:
|
||||
预览结果列表，包含原文件名和新文件名
|
||||
"""
|
||||
if strategy_name not in self.strategies:
|
||||
raise ValueError(f"未知的重命名策略: {strategy_name}")
|
||||
|
||||
strategy = self.strategies[strategy_name]
|
||||
results = []
|
||||
|
||||
for index, filename in enumerate(filenames):
|
||||
try:
|
||||
new_name = strategy.apply(filename, index, params)
|
||||
results.append({
|
||||
"original": filename,
|
||||
"new": new_name,
|
||||
"success": True
|
||||
})
|
||||
except Exception as e:
|
||||
results.append({
|
||||
"original": filename,
|
||||
"new": filename,
|
||||
"success": False,
|
||||
"error": str(e)
|
||||
})
|
||||
|
||||
return results
|
||||
|
||||
def execute_rename(
|
||||
self,
|
||||
base_path: Path,
|
||||
file_paths: List[str],
|
||||
strategy_name: str,
|
||||
params: dict
|
||||
) -> Dict[str, any]:
|
||||
"""
|
||||
执行批量重命名
|
||||
|
||||
Args:
|
||||
base_path: 基础路径
|
||||
file_paths: 相对文件路径列表
|
||||
strategy_name: 策略名称
|
||||
params: 策略参数
|
||||
|
||||
Returns:
|
||||
执行结果
|
||||
"""
|
||||
if strategy_name not in self.strategies:
|
||||
raise ValueError(f"未知的重命名策略: {strategy_name}")
|
||||
|
||||
strategy = self.strategies[strategy_name]
|
||||
results = {
|
||||
"success_count": 0,
|
||||
"failed_count": 0,
|
||||
"results": []
|
||||
}
|
||||
|
||||
for index, file_path in enumerate(file_paths):
|
||||
try:
|
||||
# 获取完整路径
|
||||
full_path = base_path / file_path
|
||||
|
||||
# 安全检查
|
||||
full_path = full_path.resolve()
|
||||
base_path_resolved = base_path.resolve()
|
||||
|
||||
if not str(full_path).startswith(str(base_path_resolved)):
|
||||
raise ValueError("访问被拒绝")
|
||||
|
||||
if not full_path.exists():
|
||||
raise FileNotFoundError("文件不存在")
|
||||
|
||||
# 生成新文件名
|
||||
original_filename = full_path.name
|
||||
new_filename = strategy.apply(original_filename, index, params)
|
||||
|
||||
# 处理文件名冲突
|
||||
new_path = full_path.parent / new_filename
|
||||
new_path = self._resolve_conflict(new_path)
|
||||
|
||||
# 执行重命名
|
||||
full_path.rename(new_path)
|
||||
|
||||
results["results"].append({
|
||||
"original_path": file_path,
|
||||
"original_name": original_filename,
|
||||
"new_name": new_path.name,
|
||||
"new_path": str(new_path.relative_to(base_path_resolved)).replace('\\', '/'),
|
||||
"success": True
|
||||
})
|
||||
results["success_count"] += 1
|
||||
|
||||
logger.info(f"重命名成功: {original_filename} -> {new_path.name}")
|
||||
|
||||
except Exception as e:
|
||||
results["results"].append({
|
||||
"original_path": file_path,
|
||||
"original_name": Path(file_path).name,
|
||||
"success": False,
|
||||
"error": str(e)
|
||||
})
|
||||
results["failed_count"] += 1
|
||||
|
||||
logger.error(f"重命名失败: {file_path}, 错误: {str(e)}")
|
||||
|
||||
return results
|
||||
|
||||
def _resolve_conflict(self, path: Path) -> Path:
|
||||
"""
|
||||
解决文件名冲突
|
||||
|
||||
Args:
|
||||
path: 目标路径
|
||||
|
||||
Returns:
|
||||
解决冲突后的路径
|
||||
"""
|
||||
if not path.exists():
|
||||
return path
|
||||
|
||||
# 分离文件名和扩展名
|
||||
name_part, ext_part = AddSuffixStrategy._split_name_ext(path.name)
|
||||
parent = path.parent
|
||||
|
||||
# 添加数字后缀
|
||||
counter = 1
|
||||
while True:
|
||||
new_name = f"{name_part}_{counter}{ext_part}"
|
||||
new_path = parent / new_name
|
||||
if not new_path.exists():
|
||||
return new_path
|
||||
counter += 1
|
||||
|
||||
# 防止无限循环
|
||||
if counter > 1000:
|
||||
raise ValueError("无法解决文件名冲突")
|
||||
|
||||
|
||||
# 全局实例
|
||||
rename_manager = RenameManager()
|
||||
@ -3,7 +3,7 @@ FastAPI主入口文件
|
||||
"""
|
||||
from fastapi import FastAPI
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from app.api.v1 import websocket
|
||||
from app.api.v1 import websocket, files
|
||||
from app.core.websocket_manager import websocket_manager
|
||||
from app.core.software_manager import software_manager
|
||||
from app.core.log_manager import log_manager
|
||||
@ -41,6 +41,8 @@ app.add_middleware(
|
||||
|
||||
# 注册API路由
|
||||
app.include_router(websocket.router, prefix="/api/v1/ws", tags=["WebSocket"])
|
||||
app.include_router(files.router, prefix="/api/v1", tags=["Files"])
|
||||
|
||||
|
||||
|
||||
@app.get("/")
|
||||
|
||||
@ -1,3 +1,7 @@
|
||||
# 文件存储路径配置
|
||||
file_storage:
|
||||
cad_files_path: "C:\\Users\\Public\\Documents"
|
||||
|
||||
software:
|
||||
creo:
|
||||
name: "PTC Creo"
|
||||
|
||||
1007
websocket-file-api-docs.md
Normal file
1007
websocket-file-api-docs.md
Normal file
File diff suppressed because it is too large
Load Diff
Loading…
Reference in New Issue
Block a user