From 65ec6aacb793dfda7ad38c744cb462ff21707143 Mon Sep 17 00:00:00 2001 From: Will Miao <13051207myq@gmail.com> Date: Mon, 4 Aug 2025 18:15:03 +0800 Subject: [PATCH] feat: add model moving endpoints for individual and bulk operations --- py/routes/base_model_routes.py | 81 +++++++++++++++++++++++++- py/routes/lora_routes.py | 103 --------------------------------- 2 files changed, 80 insertions(+), 104 deletions(-) diff --git a/py/routes/base_model_routes.py b/py/routes/base_model_routes.py index 96585c1e..f39a47d9 100644 --- a/py/routes/base_model_routes.py +++ b/py/routes/base_model_routes.py @@ -48,6 +48,8 @@ class BaseModelRoutes(ABC): app.router.add_post(f'/api/{prefix}/rename', self.rename_model) app.router.add_post(f'/api/{prefix}/bulk-delete', self.bulk_delete_models) app.router.add_post(f'/api/{prefix}/verify-duplicates', self.verify_duplicates) + app.router.add_post(f'/api/{prefix}/move_model', self.move_model) + app.router.add_post(f'/api/{prefix}/move_models_bulk', self.move_models_bulk) # Common query routes app.router.add_get(f'/api/{prefix}/top-tags', self.get_top_tags) @@ -616,4 +618,81 @@ class BaseModelRoutes(ABC): # This will be implemented by subclasses as they need CivitAI client access return web.json_response({ "error": "Not implemented in base class" - }, status=501) \ No newline at end of file + }, status=501) + + # Common model move handlers + async def move_model(self, request: web.Request) -> web.Response: + """Handle model move request""" + try: + data = await request.json() + file_path = data.get('file_path') + target_path = data.get('target_path') + if not file_path or not target_path: + return web.Response(text='File path and target path are required', status=400) + import os + source_dir = os.path.dirname(file_path) + if os.path.normpath(source_dir) == os.path.normpath(target_path): + logger.info(f"Source and target directories are the same: {source_dir}") + return web.json_response({'success': True, 'message': 'Source and target directories are the same'}) + file_name = os.path.basename(file_path) + target_file_path = os.path.join(target_path, file_name).replace(os.sep, '/') + if os.path.exists(target_file_path): + return web.json_response({ + 'success': False, + 'error': f"Target file already exists: {target_file_path}" + }, status=409) + success = await self.service.scanner.move_model(file_path, target_path) + if success: + return web.json_response({'success': True, 'new_file_path': target_file_path}) + else: + return web.Response(text='Failed to move model', status=500) + except Exception as e: + logger.error(f"Error moving model: {e}", exc_info=True) + return web.Response(text=str(e), status=500) + + async def move_models_bulk(self, request: web.Request) -> web.Response: + """Handle bulk model move request""" + try: + data = await request.json() + file_paths = data.get('file_paths', []) + target_path = data.get('target_path') + if not file_paths or not target_path: + return web.Response(text='File paths and target path are required', status=400) + results = [] + import os + for file_path in file_paths: + source_dir = os.path.dirname(file_path) + if os.path.normpath(source_dir) == os.path.normpath(target_path): + results.append({ + "path": file_path, + "success": True, + "message": "Source and target directories are the same" + }) + continue + file_name = os.path.basename(file_path) + target_file_path = os.path.join(target_path, file_name).replace(os.sep, '/') + if os.path.exists(target_file_path): + results.append({ + "path": file_path, + "success": False, + "message": f"Target file already exists: {target_file_path}" + }) + continue + success = await self.service.scanner.move_model(file_path, target_path) + results.append({ + "path": file_path, + "success": success, + "message": "Success" if success else "Failed to move model" + }) + success_count = sum(1 for r in results if r["success"]) + failure_count = len(results) - success_count + return web.json_response({ + 'success': True, + 'message': f'Moved {success_count} of {len(file_paths)} models', + 'results': results, + 'success_count': success_count, + 'failure_count': failure_count + }) + except Exception as e: + logger.error(f"Error moving models in bulk: {e}", exc_info=True) + return web.Response(text=str(e), status=500) \ No newline at end of file diff --git a/py/routes/lora_routes.py b/py/routes/lora_routes.py index 54f0494c..6f74db13 100644 --- a/py/routes/lora_routes.py +++ b/py/routes/lora_routes.py @@ -49,10 +49,6 @@ class LoraRoutes(BaseModelRoutes): app.router.add_get(f'/api/{prefix}/civitai-url', self.get_lora_civitai_url) app.router.add_get(f'/api/{prefix}/model-description', self.get_lora_model_description) - # LoRA-specific management routes - app.router.add_post(f'/api/{prefix}/move_model', self.move_model) - app.router.add_post(f'/api/{prefix}/move_models_bulk', self.move_models_bulk) - # CivitAI integration with LoRA-specific validation app.router.add_get(f'/api/{prefix}/civitai/versions/{{model_id}}', self.get_civitai_versions_lora) app.router.add_get(f'/api/{prefix}/civitai/model/version/{{modelVersionId}}', self.get_civitai_model_by_version) @@ -284,105 +280,6 @@ class LoraRoutes(BaseModelRoutes): "error": str(e) }, status=500) - # Model management methods - async def move_model(self, request: web.Request) -> web.Response: - """Handle model move request""" - try: - data = await request.json() - file_path = data.get('file_path') # full path of the model file - target_path = data.get('target_path') # folder path to move the model to - - if not file_path or not target_path: - return web.Response(text='File path and target path are required', status=400) - - # Check if source and destination are the same - import os - source_dir = os.path.dirname(file_path) - if os.path.normpath(source_dir) == os.path.normpath(target_path): - logger.info(f"Source and target directories are the same: {source_dir}") - return web.json_response({'success': True, 'message': 'Source and target directories are the same'}) - - # Check if target file already exists - file_name = os.path.basename(file_path) - target_file_path = os.path.join(target_path, file_name).replace(os.sep, '/') - - if os.path.exists(target_file_path): - return web.json_response({ - 'success': False, - 'error': f"Target file already exists: {target_file_path}" - }, status=409) # 409 Conflict - - # Call scanner to handle the move operation - success = await self.service.scanner.move_model(file_path, target_path) - - if success: - return web.json_response({'success': True, 'new_file_path': target_file_path}) - else: - return web.Response(text='Failed to move model', status=500) - - except Exception as e: - logger.error(f"Error moving model: {e}", exc_info=True) - return web.Response(text=str(e), status=500) - - async def move_models_bulk(self, request: web.Request) -> web.Response: - """Handle bulk model move request""" - try: - data = await request.json() - file_paths = data.get('file_paths', []) # list of full paths of the model files - target_path = data.get('target_path') # folder path to move the models to - - if not file_paths or not target_path: - return web.Response(text='File paths and target path are required', status=400) - - results = [] - import os - for file_path in file_paths: - # Check if source and destination are the same - source_dir = os.path.dirname(file_path) - if os.path.normpath(source_dir) == os.path.normpath(target_path): - results.append({ - "path": file_path, - "success": True, - "message": "Source and target directories are the same" - }) - continue - - # Check if target file already exists - file_name = os.path.basename(file_path) - target_file_path = os.path.join(target_path, file_name).replace(os.sep, '/') - - if os.path.exists(target_file_path): - results.append({ - "path": file_path, - "success": False, - "message": f"Target file already exists: {target_file_path}" - }) - continue - - # Try to move the model - success = await self.service.scanner.move_model(file_path, target_path) - results.append({ - "path": file_path, - "success": success, - "message": "Success" if success else "Failed to move model" - }) - - # Count successes and failures - success_count = sum(1 for r in results if r["success"]) - failure_count = len(results) - success_count - - return web.json_response({ - 'success': True, - 'message': f'Moved {success_count} of {len(file_paths)} models', - 'results': results, - 'success_count': success_count, - 'failure_count': failure_count - }) - - except Exception as e: - logger.error(f"Error moving models in bulk: {e}", exc_info=True) - return web.Response(text=str(e), status=500) - async def get_lora_model_description(self, request: web.Request) -> web.Response: """Get model description for a Lora model""" try: