updating the previous hardcoded hierarchy file to use the json provided during creation of datasnapshot

Author: ddelpiano

applications/sckanner/backend/sckanner/management/commands/update_knowledge_statements.py

@@ -55,13 +55,24 @@ class JsonData(BaseModel):
 class Command(BaseCommand):
     help = "Fetch and update knowledge statements from an external server"
 
+    def add_arguments(self, parser):
+        parser.add_argument(
+            '--a_b_via_c_json_url',
+            type=str,
+            help='URL to the A-B-via-C JSON file',
+        )
+
     def handle(self, *args, **kwargs):
         self.stdout.write("Starting the ingestion process...")
-
-        # Step 1: Fetch raw JSON from external source
-        raw_data_url = "https://raw.githubusercontent.com/smtifahim/SCKAN-Apps/master/sckan-explorer/json/a-b-via-c-2.json"
-        self.stdout.write(f"Fetching data from {raw_data_url}...")
-        response = requests.get(raw_data_url)
+        a_b_via_c_json_url = kwargs.get('a_b_via_c_json_url')
+
+        # Step 1: Fetch raw JSON from URL
+        if not a_b_via_c_json_url:
+            # Fallback to hardcoded URL (for backward compatibility)
+            a_b_via_c_json_url = "https://raw.githubusercontent.com/smtifahim/SCKAN-Apps/refs/heads/master/sckan-explorer/json/sckanner-data/hierarchy/sckanner-hierarchy.json"
+        
+        self.stdout.write(f"Fetching data from {a_b_via_c_json_url}...")
+        response = requests.get(a_b_via_c_json_url)
         response.raise_for_status()
         raw_data = JsonData(**response.json())
 
@@ -125,4 +136,4 @@ def update_database(self, detailed_data):
         # Insert new data
         KnowledgeStatement.objects.bulk_create(
             [KnowledgeStatement(data=entry) for entry in detailed_data]
-        )
+        )

applications/sckanner/backend/sckanner/migrations/data/composer_ingestion_script.py

@@ -123,15 +123,25 @@ def fetch_paginated_data(population_ids: list[str], stdout=None):
 	return detailed_data
 
 
-def get_statements(version="", stdout=None):
+def get_statements(version="", stdout=None, a_b_via_c_json_file_path=None):
     try:
-        # Step 1: Fetch raw JSON from external source
-        raw_data_url = "https://raw.githubusercontent.com/smtifahim/SCKAN-Apps/master/sckan-explorer/json/a-b-via-c-2.json"
-        if stdout:
-            stdout.write(f"Fetching data from {raw_data_url}...\n")
-        response = requests.get(raw_data_url)
-        response.raise_for_status()
-        raw_data = JsonData(**response.json())
+        # Step 1: Load raw JSON from file or external source
+        if a_b_via_c_json_file_path:
+            # Load from provided file path
+            if stdout:
+                stdout.write(f"Loading data from file: {a_b_via_c_json_file_path}...\n")
+            import json
+            with open(a_b_via_c_json_file_path, 'r') as f:
+                raw_json = json.load(f)
+            raw_data = JsonData(**raw_json)
+        else:
+            # Fallback to hardcoded URL (for backward compatibility)
+            raw_data_url = "https://raw.githubusercontent.com/smtifahim/SCKAN-Apps/refs/heads/master/sckan-explorer/json/sckanner-data/hierarchy/sckanner-hierarchy.json"
+            if stdout:
+                stdout.write(f"Fetching data from {raw_data_url}...\n")
+            response = requests.get(raw_data_url)
+            response.raise_for_status()
+            raw_data = JsonData(**response.json())
 
         # Step 2: Extract population IDs
         if stdout:
@@ -149,7 +159,7 @@ def get_statements(version="", stdout=None):
             # --- NOTE: ONLY FOR TESTING LOCALLY ---
 
         for population_id in batched(population_ids, KNOWLEDGE_STATEMENTS_BATCH_SIZE):
-            detailed_data.extend(fetch_paginated_data(list(population_id)))
+            detailed_data.extend(fetch_paginated_data(list(population_id), stdout))
 
         if stdout:
             stdout.write(f"Ingestion process completed successfully! Total statements: {len(detailed_data)}\n")

applications/sckanner/backend/sckanner/services/ingestion/connectivity_statement_adapter.py

@@ -40,7 +40,15 @@ def _parse_and_validate_statements(
 
         # Get statements from the uploaded module
         if hasattr(module, "get_statements") and callable(module.get_statements):
-            statements = module.get_statements(self.snapshot.version)
+            # Pass the a_b_via_c_json_file path if available
+            a_b_via_c_json_file_path = None
+            if self.snapshot.a_b_via_c_json_file:
+                a_b_via_c_json_file_path = self.snapshot.a_b_via_c_json_file.path
+            
+            statements = module.get_statements(
+                self.snapshot.version, 
+                a_b_via_c_json_file_path=a_b_via_c_json_file_path
+            )
             try:
                 # Validate the statements against the schema
                 current_path = os.path.dirname(os.path.abspath(__file__))