Update Solr Query to identify Complex Objects and Relationships

importer-solr.py

@@ -1,6 +1,7 @@
 import sys, requests, json, argparse
 import logging
 from typing import List, Any, Dict
+from utils import is_complex_solr
 
 
 
@@ -26,13 +27,15 @@
                     help='Print the full Solr query response')
 parser.add_argument('--verbose', '-v', action='store_true',
                     help='Print PIDs and data for works with errors')
+parser.add_argument('--list-complex', '-c', action='store_true',
+                    help='Print the complex object relationships')
 
 def build_solr_query_url(importer_no: int) -> tuple[str, Dict[str, str]]:
     """Create Solr query URL for a given importer"""
     base_url = "https://solr-od2.library.oregonstate.edu/solr/prod/select?"
     params = {
         'q': f'bulkrax_identifier_sim:{importer_no}*',
-        'fl': 'id,member_of_collection_ids_ssim,member_of_collections_ssim,file_set_ids_ssim,thumbnail_path_ss,suppressed_bsi,workflow_state_name_ssim,visibility_ssi',
+        'fl': 'id,member_of_collection_ids_ssim,member_of_collections_ssim,file_set_ids_ssim,thumbnail_path_ss,suppressed_bsi,workflow_state_name_ssim,visibility_ssi,has_model_ssim,resource_type_label_ssim,member_ids_ssim',
         'rows': '1000'
     }
     return base_url, params
@@ -55,11 +58,21 @@ def analyze_works(docs: List[Dict]) -> tuple[List[str], List[Any], List[str]]:
 
     for work in docs:
         work_id = work['id']
-
-        # Check for file set value
-        if 'file_set_ids_ssim' not in work:
-            no_file_set.append(work_id)
-
+        is_complex = is_complex_solr(work)
+
+        # Checks that only apply to non-complex objects
+        if not is_complex:
+            # Check for file set value
+            if 'file_set_ids_ssim' not in work:
+                no_file_set.append(work_id)
+            # Check thumbnail path format
+            if 'thumbnail_path_ss' in work:
+                thumbnail = work['thumbnail_path_ss']
+                if not (thumbnail.startswith('/downloads/') and '?file=thumbnail' in thumbnail):
+                    bad_thumbnail.append(f"{work_id} (thumbnail: {thumbnail})")
+            else:
+                bad_thumbnail.append(f"{work_id} (no thumbnail)")
+
         # Check for collection value
         if 'member_of_collection_ids_ssim' in work:
             coll_id = work['member_of_collection_ids_ssim']
@@ -68,14 +81,6 @@ def analyze_works(docs: List[Dict]) -> tuple[List[str], List[Any], List[str]]:
         else:
             no_coll_id.append(work_id)
 
-        # Check thumbnail path format
-        if 'thumbnail_path_ss' in work:
-            thumbnail = work['thumbnail_path_ss']
-            if not (thumbnail.startswith('/downloads/') and '?file=thumbnail' in thumbnail):
-                bad_thumbnail.append(f"{work_id} (thumbnail: {thumbnail})")
-        else:
-            bad_thumbnail.append(f"{work_id} (no thumbnail)")
-
         # Check suppressed status
         if work.get('suppressed_bsi') == True:
             suppressed_works.append(work_id)
@@ -165,7 +170,7 @@ def log_visibility_status(bad_visibility: List[str], total_works: int, verbose:
         logger.info(f"All {total_works} works have correct visibility")
 
 def main():
-    """Run file"""
+    """Run status checks for an importer and log the summary"""
     args = parser.parse_args()
 
     logger.info(f"Querying Solr for importer {args.importer_no}")
@@ -199,6 +204,20 @@ def main():
     log_workflow_status(bad_workflow, num_found, args.verbose)
     log_visibility_status(bad_visibility, num_found, args.verbose)
 
+    # Print the complex relationships
+    if args.list_complex:
+        relationships = {
+            work["id"]: work.get("member_ids_ssim", [])
+            for work in docs
+            if is_complex_solr(work)
+        }
+
+        print("Complex object relationships:")
+        for parent_id, child_ids in relationships.items():
+            print(f"{parent_id}:")
+            for child_id in child_ids:
+                print(f"  {child_id}")
+
     # Print response if requested
     if args.print_response:
         logger.info("Full Solr query response:")

utils.py

@@ -1,8 +1,11 @@
 import re
+import sys, requests
 
 COMPLEX_MODEL_VALUE = "generic"
 COMPLEX_RESOURCE_TYPE_VALUE = "http://purl.org/dc/dcmitype/Collection"
 FILESET_TYPE_VALUE = "fileset"
+COMPLEX_SOLR_MODEL_VALUE = "Generic"
+COMPLEX_SOLR_RESOURCE_TYPE_VALUE = "Complex Object"
 
 def base_header(header: str) -> str:
     """Return the base header without _X for an enumerated header"""
@@ -16,6 +19,12 @@ def is_complex(row) -> bool:
     resource = str(row.get("resource_type")).lower()
     return model == COMPLEX_MODEL_VALUE.lower() and resource == COMPLEX_RESOURCE_TYPE_VALUE.lower()
 
+def is_complex_solr(work) -> bool:
+    """Return True if work is complex, False otherwise"""
+    model = work.get("has_model_ssim", [])
+    resource_type = work.get("resource_type_label_ssim", [])
+    return COMPLEX_SOLR_MODEL_VALUE in model and COMPLEX_SOLR_RESOURCE_TYPE_VALUE in resource_type
+
 def is_fileset(row) -> bool:
     """Return True if row is a file set, False otherwise"""
     if "model" not in row: