From 68f887140331cf71ba71d4fb43a319cffbba7865 Mon Sep 17 00:00:00 2001
From: Will Miao <13051207myq@gmail.com>
Date: Wed, 10 Sep 2025 11:20:58 +0800
Subject: [PATCH] feat(metadata): add source tracking for SQLite metadata and
 implement Civitai API metadata validation

---
 py/services/model_metadata_provider.py |  6 ++--
 py/utils/routes_common.py              | 42 ++++++++++++++++++--------
 2 files changed, 33 insertions(+), 15 deletions(-)

diff --git a/py/services/model_metadata_provider.py b/py/services/model_metadata_provider.py
index 2a30df11..9f54a2e7 100644
--- a/py/services/model_metadata_provider.py
+++ b/py/services/model_metadata_provider.py
@@ -127,7 +127,8 @@ class SQLiteModelMetadataProvider(ModelMetadataProvider):
                     'model': {
                         'name': model_row['name'],
                         'type': model_type,
-                    }
+                    },
+                    'source': 'archive_db'
                 }
                 # Update with any additional data
                 version_entry.update(version_data)
@@ -266,7 +267,8 @@ class SQLiteModelMetadataProvider(ModelMetadataProvider):
                 "creator": {
                     "username": model_row['username'] or model_data.get("creator", {}).get("username"),
                     "image": model_data.get("creator", {}).get("image")
-                }
+                },
+                "source": "archive_db"
             }
             
             # Add any additional fields from version data
diff --git a/py/utils/routes_common.py b/py/utils/routes_common.py
index eba348ea..a82dbd2f 100644
--- a/py/utils/routes_common.py
+++ b/py/utils/routes_common.py
@@ -37,6 +37,18 @@ class ModelRouteUtils:
         local_metadata['from_civitai'] = False
         await MetadataManager.save_metadata(metadata_path, local_metadata)
 
+    @staticmethod
+    def is_civitai_api_metadata(meta: dict) -> bool:
+        """
+        Determine if the given civitai metadata is from the civitai API.
+        Returns True if both 'files' and 'images' exist and are non-empty.
+        """
+        if not isinstance(meta, dict):
+            return False
+        files = meta.get('files')
+        images = meta.get('images')
+        return bool(files) and bool(images)
+
     @staticmethod
     async def update_model_metadata(metadata_path: str, local_metadata: Dict, 
                                   civitai_metadata: Dict, metadata_provider=None) -> None:
@@ -44,21 +56,25 @@ class ModelRouteUtils:
         # Save existing trainedWords and customImages if they exist
         existing_civitai = local_metadata.get('civitai') or {}  # Use empty dict if None
 
-        # Create a new civitai metadata by updating existing with new
-        merged_civitai = existing_civitai.copy()
-        merged_civitai.update(civitai_metadata)
+        # Check if we should skip the update to avoid overwriting richer data
+        if civitai_metadata.get('source') == 'archive_db' and ModelRouteUtils.is_civitai_api_metadata(existing_civitai):
+            logger.info(f"Skip civitai update for {local_metadata.get('model_name', '')}: {existing_civitai.get('name', '')}")
+        else:
+            # Create a new civitai metadata by updating existing with new
+            merged_civitai = existing_civitai.copy()
+            merged_civitai.update(civitai_metadata)
 
-        # Special handling for trainedWords - ensure we don't lose any existing trained words
-        if 'trainedWords' in existing_civitai:
-            existing_trained_words = existing_civitai.get('trainedWords', [])
-            new_trained_words = civitai_metadata.get('trainedWords', [])
-            # Use a set to combine words without duplicates, then convert back to list
-            merged_trained_words = list(set(existing_trained_words + new_trained_words))
-            merged_civitai['trainedWords'] = merged_trained_words
+            # Special handling for trainedWords - ensure we don't lose any existing trained words
+            if 'trainedWords' in existing_civitai:
+                existing_trained_words = existing_civitai.get('trainedWords', [])
+                new_trained_words = civitai_metadata.get('trainedWords', [])
+                # Use a set to combine words without duplicates, then convert back to list
+                merged_trained_words = list(set(existing_trained_words + new_trained_words))
+                merged_civitai['trainedWords'] = merged_trained_words
 
-        # Update local metadata with merged civitai data
-        local_metadata['civitai'] = merged_civitai
-        local_metadata['from_civitai'] = True
+            # Update local metadata with merged civitai data
+            local_metadata['civitai'] = merged_civitai
+            local_metadata['from_civitai'] = True
         
         # Update model name if available
         if 'model' in civitai_metadata: