Refactor metadata handling to use unified provider system

- Replaced direct usage of Civitai client with a fallback metadata provider across all recipe parsers.
- Updated metadata service to improve initialization and error handling.
- Enhanced download manager to utilize a downloader service for file operations.
- Improved recipe scanner to fetch model information through the new metadata provider.
- Updated utility functions to streamline image downloading and processing.
- Added comprehensive logging and error handling for better debugging and reliability.
- Introduced `get_default_metadata_provider()` for simplified access to the default provider.
- Ensured backward compatibility with existing APIs and workflows.
This commit is contained in:
Will Miao
2025-09-09 20:57:45 +08:00
parent 1ea468cfc4
commit 6fd74952b7
15 changed files with 350 additions and 178 deletions

View File

@@ -6,6 +6,7 @@ import logging
from typing import Dict, Any
from ..base import RecipeMetadataParser
from ..constants import GEN_PARAM_KEYS
from ...services.metadata_service import get_default_metadata_provider
logger = logging.getLogger(__name__)
@@ -30,6 +31,9 @@ class AutomaticMetadataParser(RecipeMetadataParser):
async def parse_metadata(self, user_comment: str, recipe_scanner=None, civitai_client=None) -> Dict[str, Any]:
"""Parse metadata from Automatic1111 format"""
try:
# Get metadata provider instead of using civitai_client directly
metadata_provider = await get_default_metadata_provider()
# Split on Negative prompt if it exists
if "Negative prompt:" in user_comment:
parts = user_comment.split('Negative prompt:', 1)
@@ -216,9 +220,9 @@ class AutomaticMetadataParser(RecipeMetadataParser):
}
# Get additional info from Civitai
if civitai_client:
if metadata_provider:
try:
civitai_info = await civitai_client.get_model_version_info(resource.get("modelVersionId"))
civitai_info = await metadata_provider.get_model_version_info(resource.get("modelVersionId"))
populated_entry = await self.populate_lora_from_civitai(
lora_entry,
civitai_info,
@@ -271,11 +275,11 @@ class AutomaticMetadataParser(RecipeMetadataParser):
}
# Try to get info from Civitai
if civitai_client:
if metadata_provider:
try:
if lora_hash:
# If we have hash, use it for lookup
civitai_info = await civitai_client.get_model_by_hash(lora_hash)
civitai_info = await metadata_provider.get_model_by_hash(lora_hash)
else:
civitai_info = None

View File

@@ -5,6 +5,7 @@ import logging
from typing import Dict, Any, Union
from ..base import RecipeMetadataParser
from ..constants import GEN_PARAM_KEYS
from ...services.metadata_service import get_default_metadata_provider
logger = logging.getLogger(__name__)
@@ -36,12 +37,15 @@ class CivitaiApiMetadataParser(RecipeMetadataParser):
Args:
metadata: The metadata from the image (dict)
recipe_scanner: Optional recipe scanner service
civitai_client: Optional Civitai API client
civitai_client: Optional Civitai API client (deprecated, use metadata_provider instead)
Returns:
Dict containing parsed recipe data
"""
try:
# Get metadata provider instead of using civitai_client directly
metadata_provider = await get_default_metadata_provider()
# Initialize result structure
result = {
'base_model': None,
@@ -85,9 +89,9 @@ class CivitaiApiMetadataParser(RecipeMetadataParser):
# Extract base model information - directly if available
if "baseModel" in metadata:
result["base_model"] = metadata["baseModel"]
elif "Model hash" in metadata and civitai_client:
elif "Model hash" in metadata and metadata_provider:
model_hash = metadata["Model hash"]
model_info = await civitai_client.get_model_by_hash(model_hash)
model_info = await metadata_provider.get_model_by_hash(model_hash)
if model_info:
result["base_model"] = model_info.get("baseModel", "")
elif "Model" in metadata and isinstance(metadata.get("resources"), list):
@@ -95,8 +99,8 @@ class CivitaiApiMetadataParser(RecipeMetadataParser):
for resource in metadata.get("resources", []):
if resource.get("type") == "model" and resource.get("name") == metadata.get("Model"):
# This is likely the checkpoint model
if civitai_client and resource.get("hash"):
model_info = await civitai_client.get_model_by_hash(resource.get("hash"))
if metadata_provider and resource.get("hash"):
model_info = await metadata_provider.get_model_by_hash(resource.get("hash"))
if model_info:
result["base_model"] = model_info.get("baseModel", "")
@@ -138,9 +142,9 @@ class CivitaiApiMetadataParser(RecipeMetadataParser):
}
# Try to get info from Civitai if hash is available
if lora_entry['hash'] and civitai_client:
if lora_entry['hash'] and metadata_provider:
try:
civitai_info = await civitai_client.get_model_by_hash(lora_hash)
civitai_info = await metadata_provider.get_model_by_hash(lora_hash)
populated_entry = await self.populate_lora_from_civitai(
lora_entry,
@@ -194,10 +198,10 @@ class CivitaiApiMetadataParser(RecipeMetadataParser):
}
# Try to get info from Civitai if modelVersionId is available
if version_id and civitai_client:
if version_id and metadata_provider:
try:
# Use get_model_version_info instead of get_model_version
civitai_info, error = await civitai_client.get_model_version_info(version_id)
civitai_info, error = await metadata_provider.get_model_version_info(version_id)
if error:
logger.warning(f"Error getting model version info: {error}")
@@ -259,11 +263,11 @@ class CivitaiApiMetadataParser(RecipeMetadataParser):
'isDeleted': False
}
# If we have a version ID and civitai client, try to get more info
if version_id and civitai_client:
# If we have a version ID and metadata provider, try to get more info
if version_id and metadata_provider:
try:
# Use get_model_version_info with the version ID
civitai_info, error = await civitai_client.get_model_version_info(version_id)
civitai_info, error = await metadata_provider.get_model_version_info(version_id)
if error:
logger.warning(f"Error getting model version info: {error}")
@@ -316,9 +320,9 @@ class CivitaiApiMetadataParser(RecipeMetadataParser):
}
# Try to get info from Civitai if hash is available
if lora_entry['hash'] and civitai_client:
if lora_entry['hash'] and metadata_provider:
try:
civitai_info = await civitai_client.get_model_by_hash(lora_hash)
civitai_info = await metadata_provider.get_model_by_hash(lora_hash)
populated_entry = await self.populate_lora_from_civitai(
lora_entry,

View File

@@ -6,6 +6,7 @@ import logging
from typing import Dict, Any
from ..base import RecipeMetadataParser
from ..constants import GEN_PARAM_KEYS
from ...services.metadata_service import get_default_metadata_provider
logger = logging.getLogger(__name__)
@@ -26,6 +27,9 @@ class ComfyMetadataParser(RecipeMetadataParser):
async def parse_metadata(self, user_comment: str, recipe_scanner=None, civitai_client=None) -> Dict[str, Any]:
"""Parse metadata from Civitai ComfyUI metadata format"""
try:
# Get metadata provider instead of using civitai_client directly
metadata_provider = await get_default_metadata_provider()
data = json.loads(user_comment)
loras = []
@@ -73,10 +77,10 @@ class ComfyMetadataParser(RecipeMetadataParser):
'isDeleted': False
}
# Get additional info from Civitai if client is available
if civitai_client:
# Get additional info from Civitai if metadata provider is available
if metadata_provider:
try:
civitai_info_tuple = await civitai_client.get_model_version_info(model_version_id)
civitai_info_tuple = await metadata_provider.get_model_version_info(model_version_id)
# Populate lora entry with Civitai info
populated_entry = await self.populate_lora_from_civitai(
lora_entry,
@@ -116,9 +120,9 @@ class ComfyMetadataParser(RecipeMetadataParser):
}
# Get additional checkpoint info from Civitai
if civitai_client:
if metadata_provider:
try:
civitai_info_tuple = await civitai_client.get_model_version_info(checkpoint_version_id)
civitai_info_tuple = await metadata_provider.get_model_version_info(checkpoint_version_id)
civitai_info, _ = civitai_info_tuple if isinstance(civitai_info_tuple, tuple) else (civitai_info_tuple, None)
# Populate checkpoint with Civitai info
checkpoint = await self.populate_checkpoint_from_civitai(checkpoint, civitai_info)

View File

@@ -5,6 +5,7 @@ import logging
from typing import Dict, Any
from ..base import RecipeMetadataParser
from ..constants import GEN_PARAM_KEYS
from ...services.metadata_service import get_default_metadata_provider
logger = logging.getLogger(__name__)
@@ -18,8 +19,11 @@ class MetaFormatParser(RecipeMetadataParser):
return re.search(self.METADATA_MARKER, user_comment, re.IGNORECASE | re.DOTALL) is not None
async def parse_metadata(self, user_comment: str, recipe_scanner=None, civitai_client=None) -> Dict[str, Any]:
"""Parse metadata from images with meta format metadata"""
"""Parse metadata from images with meta format metadata (Lora_N Model hash format)"""
try:
# Get metadata provider instead of using civitai_client directly
metadata_provider = await get_default_metadata_provider()
# Extract prompt and negative prompt
parts = user_comment.split('Negative prompt:', 1)
prompt = parts[0].strip()
@@ -122,9 +126,9 @@ class MetaFormatParser(RecipeMetadataParser):
}
# Get info from Civitai by hash if available
if civitai_client and hash_value:
if metadata_provider and hash_value:
try:
civitai_info = await civitai_client.get_model_by_hash(hash_value)
civitai_info = await metadata_provider.get_model_by_hash(hash_value)
# Populate lora entry with Civitai info
populated_entry = await self.populate_lora_from_civitai(
lora_entry,

View File

@@ -7,6 +7,7 @@ from typing import Dict, Any
from ...config import config
from ..base import RecipeMetadataParser
from ..constants import GEN_PARAM_KEYS
from ...services.metadata_service import get_default_metadata_provider
logger = logging.getLogger(__name__)
@@ -23,6 +24,9 @@ class RecipeFormatParser(RecipeMetadataParser):
async def parse_metadata(self, user_comment: str, recipe_scanner=None, civitai_client=None) -> Dict[str, Any]:
"""Parse metadata from images with dedicated recipe metadata format"""
try:
# Get metadata provider instead of using civitai_client directly
metadata_provider = await get_default_metadata_provider()
# Extract recipe metadata from user comment
try:
# Look for recipe metadata section
@@ -71,9 +75,9 @@ class RecipeFormatParser(RecipeMetadataParser):
lora_entry['localPath'] = None
# Try to get additional info from Civitai if we have a model version ID
if lora.get('modelVersionId') and civitai_client:
if lora.get('modelVersionId') and metadata_provider:
try:
civitai_info_tuple = await civitai_client.get_model_version_info(lora['modelVersionId'])
civitai_info_tuple = await metadata_provider.get_model_version_info(lora['modelVersionId'])
# Populate lora entry with Civitai info
populated_entry = await self.populate_lora_from_civitai(
lora_entry,