mirror of
https://github.com/willmiao/ComfyUI-Lora-Manager.git
synced 2026-03-23 22:22:11 -03:00
feat: Add Lora recipes retrieval and filtering functionality
- Implemented a new API endpoint to fetch recipes associated with a specific Lora by its hash. - Enhanced the recipe scanning logic to support filtering by Lora hash and bypassing other filters. - Added a new method to retrieve a recipe by its ID with formatted metadata. - Created a new RecipeTab component to display recipes in the Lora modal. - Introduced session storage utilities for managing custom filter states. - Updated the UI to include a custom filter indicator and loading/error states for recipes. - Refactored existing recipe management logic to accommodate new features and improve maintainability.
This commit is contained in:
Will Miao
@@ -60,6 +60,9 @@ class RecipeRoutes:
|
||||
app.on_startup.append(routes._init_cache)
|
||||
|
||||
app.router.add_post('/api/recipes/save-from-widget', routes.save_recipe_from_widget)
|
||||
|
||||
# Add route to get recipes for a specific Lora
|
||||
app.router.add_get('/api/recipes/for-lora', routes.get_recipes_for_lora)
|
||||
|
||||
async def _init_cache(self, app):
|
||||
"""Initialize cache on startup"""
|
||||
@@ -98,6 +101,10 @@ class RecipeRoutes:
|
||||
base_models = request.query.get('base_models', None)
|
||||
tags = request.query.get('tags', None)
|
||||
|
||||
# New parameter: get LoRA hash filter
|
||||
lora_hash = request.query.get('lora_hash', None)
|
||||
bypass_filters = request.query.get('bypass_filters', 'false').lower() == 'true'
|
||||
|
||||
# Parse filter parameters
|
||||
filters = {}
|
||||
if base_models:
|
||||
@@ -113,14 +120,16 @@ class RecipeRoutes:
|
||||
'lora_model': search_lora_model
|
||||
}
|
||||
|
||||
# Get paginated data
|
||||
# Get paginated data with the new lora_hash parameter
|
||||
result = await self.recipe_scanner.get_paginated_data(
|
||||
page=page,
|
||||
page_size=page_size,
|
||||
sort_by=sort_by,
|
||||
search=search,
|
||||
filters=filters,
|
||||
search_options=search_options
|
||||
search_options=search_options,
|
||||
lora_hash=lora_hash,
|
||||
bypass_filters=bypass_filters
|
||||
)
|
||||
|
||||
# Format the response data with static URLs for file paths
|
||||
@@ -148,20 +157,14 @@ class RecipeRoutes:
|
||||
"""Get detailed information about a specific recipe"""
|
||||
try:
|
||||
recipe_id = request.match_info['recipe_id']
|
||||
|
||||
# Get all recipes from cache
|
||||
cache = await self.recipe_scanner.get_cached_data()
|
||||
|
||||
# Find the specific recipe
|
||||
recipe = next((r for r in cache.raw_data if str(r.get('id', '')) == recipe_id), None)
|
||||
# Use the new get_recipe_by_id method from recipe_scanner
|
||||
recipe = await self.recipe_scanner.get_recipe_by_id(recipe_id)
|
||||
|
||||
if not recipe:
|
||||
return web.json_response({"error": "Recipe not found"}, status=404)
|
||||
|
||||
# Format recipe data
|
||||
formatted_recipe = self._format_recipe_data(recipe)
|
||||
|
||||
return web.json_response(formatted_recipe)
|
||||
return web.json_response(recipe)
|
||||
except Exception as e:
|
||||
logger.error(f"Error retrieving recipe details: {e}", exc_info=True)
|
||||
return web.json_response({"error": str(e)}, status=500)
|
||||
@@ -1134,3 +1137,49 @@ class RecipeRoutes:
|
||||
except Exception as e:
|
||||
logger.error(f"Error reconnecting LoRA: {e}", exc_info=True)
|
||||
return web.json_response({"error": str(e)}, status=500)
|
||||
|
||||
async def get_recipes_for_lora(self, request: web.Request) -> web.Response:
|
||||
"""Get recipes that use a specific Lora"""
|
||||
try:
|
||||
lora_hash = request.query.get('hash')
|
||||
|
||||
# Hash is required
|
||||
if not lora_hash:
|
||||
return web.json_response({'success': False, 'error': 'Lora hash is required'}, status=400)
|
||||
|
||||
# Log the search parameters
|
||||
logger.info(f"Getting recipes for Lora by hash: {lora_hash}")
|
||||
|
||||
# Get all recipes from cache
|
||||
cache = await self.recipe_scanner.get_cached_data()
|
||||
|
||||
# Filter recipes that use this Lora by hash
|
||||
matching_recipes = []
|
||||
for recipe in cache.raw_data:
|
||||
# Check if any of the recipe's loras match this hash
|
||||
loras = recipe.get('loras', [])
|
||||
for lora in loras:
|
||||
if lora.get('hash', '').lower() == lora_hash.lower():
|
||||
matching_recipes.append(recipe)
|
||||
break # No need to check other loras in this recipe
|
||||
|
||||
# Process the recipes similar to get_paginated_data to ensure all needed data is available
|
||||
for recipe in matching_recipes:
|
||||
# Add inLibrary information for each lora
|
||||
if 'loras' in recipe:
|
||||
for lora in recipe['loras']:
|
||||
if 'hash' in lora and lora['hash']:
|
||||
lora['inLibrary'] = self.recipe_scanner._lora_scanner.has_lora_hash(lora['hash'].lower())
|
||||
lora['preview_url'] = self.recipe_scanner._lora_scanner.get_preview_url_by_hash(lora['hash'].lower())
|
||||
lora['localPath'] = self.recipe_scanner._lora_scanner.get_lora_path_by_hash(lora['hash'].lower())
|
||||
|
||||
# Ensure file_url is set (needed by frontend)
|
||||
if 'file_path' in recipe:
|
||||
recipe['file_url'] = self._format_recipe_file_url(recipe['file_path'])
|
||||
else:
|
||||
recipe['file_url'] = '/loras_static/images/no-preview.png'
|
||||
|
||||
return web.json_response({'success': True, 'recipes': matching_recipes})
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting recipes for Lora: {str(e)}")
|
||||
return web.json_response({'success': False, 'error': str(e)}, status=500)
|
||||
|
||||
@@ -330,7 +330,7 @@ class RecipeScanner:
|
||||
logger.error(f"Error getting base model for lora: {e}")
|
||||
return None
|
||||
|
||||
async def get_paginated_data(self, page: int, page_size: int, sort_by: str = 'date', search: str = None, filters: dict = None, search_options: dict = None):
|
||||
async def get_paginated_data(self, page: int, page_size: int, sort_by: str = 'date', search: str = None, filters: dict = None, search_options: dict = None, lora_hash: str = None, bypass_filters: bool = False):
|
||||
"""Get paginated and filtered recipe data
|
||||
|
||||
Args:
|
||||
@@ -340,69 +340,89 @@ class RecipeScanner:
|
||||
search: Search term
|
||||
filters: Dictionary of filters to apply
|
||||
search_options: Dictionary of search options to apply
|
||||
lora_hash: Optional SHA256 hash of a LoRA to filter recipes by
|
||||
bypass_filters: If True, ignore other filters when a lora_hash is provided
|
||||
"""
|
||||
cache = await self.get_cached_data()
|
||||
|
||||
# Get base dataset
|
||||
filtered_data = cache.sorted_by_date if sort_by == 'date' else cache.sorted_by_name
|
||||
|
||||
# Apply search filter
|
||||
if search:
|
||||
# Default search options if none provided
|
||||
if not search_options:
|
||||
search_options = {
|
||||
'title': True,
|
||||
'tags': True,
|
||||
'lora_name': True,
|
||||
'lora_model': True
|
||||
}
|
||||
# Special case: Filter by LoRA hash (takes precedence if bypass_filters is True)
|
||||
if lora_hash:
|
||||
# Filter recipes that contain this LoRA hash
|
||||
filtered_data = [
|
||||
item for item in filtered_data
|
||||
if 'loras' in item and any(
|
||||
lora.get('hash', '').lower() == lora_hash.lower()
|
||||
for lora in item['loras']
|
||||
)
|
||||
]
|
||||
|
||||
# Build the search predicate based on search options
|
||||
def matches_search(item):
|
||||
# Search in title if enabled
|
||||
if search_options.get('title', True):
|
||||
if fuzzy_match(str(item.get('title', '')), search):
|
||||
return True
|
||||
|
||||
# Search in tags if enabled
|
||||
if search_options.get('tags', True) and 'tags' in item:
|
||||
for tag in item['tags']:
|
||||
if fuzzy_match(tag, search):
|
||||
return True
|
||||
|
||||
# Search in lora file names if enabled
|
||||
if search_options.get('lora_name', True) and 'loras' in item:
|
||||
for lora in item['loras']:
|
||||
if fuzzy_match(str(lora.get('file_name', '')), search):
|
||||
return True
|
||||
|
||||
# Search in lora model names if enabled
|
||||
if search_options.get('lora_model', True) and 'loras' in item:
|
||||
for lora in item['loras']:
|
||||
if fuzzy_match(str(lora.get('modelName', '')), search):
|
||||
return True
|
||||
|
||||
# No match found
|
||||
return False
|
||||
|
||||
# Filter the data using the search predicate
|
||||
filtered_data = [item for item in filtered_data if matches_search(item)]
|
||||
if bypass_filters:
|
||||
# Skip other filters if bypass_filters is True
|
||||
pass
|
||||
# Otherwise continue with normal filtering after applying LoRA hash filter
|
||||
|
||||
# Apply additional filters
|
||||
if filters:
|
||||
# Filter by base model
|
||||
if 'base_model' in filters and filters['base_model']:
|
||||
filtered_data = [
|
||||
item for item in filtered_data
|
||||
if item.get('base_model', '') in filters['base_model']
|
||||
]
|
||||
# Skip further filtering if we're only filtering by LoRA hash with bypass enabled
|
||||
if not (lora_hash and bypass_filters):
|
||||
# Apply search filter
|
||||
if search:
|
||||
# Default search options if none provided
|
||||
if not search_options:
|
||||
search_options = {
|
||||
'title': True,
|
||||
'tags': True,
|
||||
'lora_name': True,
|
||||
'lora_model': True
|
||||
}
|
||||
|
||||
# Build the search predicate based on search options
|
||||
def matches_search(item):
|
||||
# Search in title if enabled
|
||||
if search_options.get('title', True):
|
||||
if fuzzy_match(str(item.get('title', '')), search):
|
||||
return True
|
||||
|
||||
# Search in tags if enabled
|
||||
if search_options.get('tags', True) and 'tags' in item:
|
||||
for tag in item['tags']:
|
||||
if fuzzy_match(tag, search):
|
||||
return True
|
||||
|
||||
# Search in lora file names if enabled
|
||||
if search_options.get('lora_name', True) and 'loras' in item:
|
||||
for lora in item['loras']:
|
||||
if fuzzy_match(str(lora.get('file_name', '')), search):
|
||||
return True
|
||||
|
||||
# Search in lora model names if enabled
|
||||
if search_options.get('lora_model', True) and 'loras' in item:
|
||||
for lora in item['loras']:
|
||||
if fuzzy_match(str(lora.get('modelName', '')), search):
|
||||
return True
|
||||
|
||||
# No match found
|
||||
return False
|
||||
|
||||
# Filter the data using the search predicate
|
||||
filtered_data = [item for item in filtered_data if matches_search(item)]
|
||||
|
||||
# Filter by tags
|
||||
if 'tags' in filters and filters['tags']:
|
||||
filtered_data = [
|
||||
item for item in filtered_data
|
||||
if any(tag in item.get('tags', []) for tag in filters['tags'])
|
||||
]
|
||||
# Apply additional filters
|
||||
if filters:
|
||||
# Filter by base model
|
||||
if 'base_model' in filters and filters['base_model']:
|
||||
filtered_data = [
|
||||
item for item in filtered_data
|
||||
if item.get('base_model', '') in filters['base_model']
|
||||
]
|
||||
|
||||
# Filter by tags
|
||||
if 'tags' in filters and filters['tags']:
|
||||
filtered_data = [
|
||||
item for item in filtered_data
|
||||
if any(tag in item.get('tags', []) for tag in filters['tags'])
|
||||
]
|
||||
|
||||
# Calculate pagination
|
||||
total_items = len(filtered_data)
|
||||
@@ -430,6 +450,74 @@ class RecipeScanner:
|
||||
}
|
||||
|
||||
return result
|
||||
|
||||
async def get_recipe_by_id(self, recipe_id: str) -> dict:
|
||||
"""Get a single recipe by ID with all metadata and formatted URLs
|
||||
|
||||
Args:
|
||||
recipe_id: The ID of the recipe to retrieve
|
||||
|
||||
Returns:
|
||||
Dict containing the recipe data or None if not found
|
||||
"""
|
||||
if not recipe_id:
|
||||
return None
|
||||
|
||||
# Get all recipes from cache
|
||||
cache = await self.get_cached_data()
|
||||
|
||||
# Find the recipe with the specified ID
|
||||
recipe = next((r for r in cache.raw_data if str(r.get('id', '')) == recipe_id), None)
|
||||
|
||||
if not recipe:
|
||||
return None
|
||||
|
||||
# Format the recipe with all needed information
|
||||
formatted_recipe = {**recipe} # Copy all fields
|
||||
|
||||
# Format file path to URL
|
||||
if 'file_path' in formatted_recipe:
|
||||
formatted_recipe['file_url'] = self._format_file_url(formatted_recipe['file_path'])
|
||||
|
||||
# Format dates for display
|
||||
for date_field in ['created_date', 'modified']:
|
||||
if date_field in formatted_recipe:
|
||||
formatted_recipe[f"{date_field}_formatted"] = self._format_timestamp(formatted_recipe[date_field])
|
||||
|
||||
# Add lora metadata
|
||||
if 'loras' in formatted_recipe:
|
||||
for lora in formatted_recipe['loras']:
|
||||
if 'hash' in lora and lora['hash']:
|
||||
lora_hash = lora['hash'].lower()
|
||||
lora['inLibrary'] = self._lora_scanner.has_lora_hash(lora_hash)
|
||||
lora['preview_url'] = self._lora_scanner.get_preview_url_by_hash(lora_hash)
|
||||
lora['localPath'] = self._lora_scanner.get_lora_path_by_hash(lora_hash)
|
||||
|
||||
return formatted_recipe
|
||||
|
||||
def _format_file_url(self, file_path: str) -> str:
|
||||
"""Format file path as URL for serving in web UI"""
|
||||
if not file_path:
|
||||
return '/loras_static/images/no-preview.png'
|
||||
|
||||
try:
|
||||
# Format file path as a URL that will work with static file serving
|
||||
recipes_dir = os.path.join(config.loras_roots[0], "recipes").replace(os.sep, '/')
|
||||
if file_path.replace(os.sep, '/').startswith(recipes_dir):
|
||||
relative_path = os.path.relpath(file_path, config.loras_roots[0]).replace(os.sep, '/')
|
||||
return f"/loras_static/root1/preview/{relative_path}"
|
||||
|
||||
# If not in recipes dir, try to create a valid URL from the file name
|
||||
file_name = os.path.basename(file_path)
|
||||
return f"/loras_static/root1/preview/recipes/{file_name}"
|
||||
except Exception as e:
|
||||
logger.error(f"Error formatting file URL: {e}")
|
||||
return '/loras_static/images/no-preview.png'
|
||||
|
||||
def _format_timestamp(self, timestamp: float) -> str:
|
||||
"""Format timestamp for display"""
|
||||
from datetime import datetime
|
||||
return datetime.fromtimestamp(timestamp).strftime('%Y-%m-%d %H:%M:%S')
|
||||
|
||||
async def update_recipe_metadata(self, recipe_id: str, metadata: dict) -> bool:
|
||||
"""Update recipe metadata (like title and tags) in both file system and cache
|
||||
|
||||
Reference in New Issue
Block a user