Merge branch 'dev' into 1030-resolve-0-value-document-type-in-nasa_science

Author: dhanur-sharma

CHANGELOG.md

@@ -51,6 +51,14 @@ For each PR made, an entry should be added to this changelog. It should contain
   - Changes:
     - Added `obj.document_type != 0` as a condition in the `get_document_type` method within the `CuratedURLAPISerializer`
 
+- 1014-add-logs-when-importing-urls-so-we-know-how-many-were-expected-how-many-succeeded-and-how-many-failed
+  - Description: When URLs of a given collection are imported into COSMOS, a Slack notification is sent. This notification includes the name of the collection imported,count of the existing curated URLs, total URLs count as per the server, URLs successfully imported from the server, delta URLs identified and delta URLs marked for deletion.
+  - Changes:
+    - The get_full_texts() function in sde_collections/sinequa_api.py is updated to yeild total_count along with rows.
+    - fetch_and_replace_full_text() function in sde_collections/tasks.py captures the total_server_count and triggers send_detailed_import_notification().
+    - Added a function send_detailed_import_notification() in sde_collections/utils/slack_utils.py to structure the notification to be sent.
+    - Updated the associated tests effected due to inclusion of this functionality.
+
 - 3228-bugfix-preserve-scroll-position--document-type-selection-behavior-on-individual-urls
   - Description: Upon selecting a document type on any individual URL, the page refreshes and returns to the top. This is not necessarily a bug but an inconvenience, especially when working at the bottom of the page. Fix the JS code.
   - Changes:

sde_collections/sinequa_api.py

@@ -257,7 +257,7 @@ def get_full_texts(
                 if total_count is None:
                     total_count = response.get("TotalRowCount", 0)
 
-                yield self._process_rows_to_records(rows)
+                yield (self._process_rows_to_records(rows))
 
                 current_offset += len(rows)
 
@@ -275,6 +275,35 @@ def get_full_texts(
                 print(f"Reducing batch size to {current_batch_size} and retrying...")
                 continue
 
+    def get_total_count(self, collection_config_folder: str, source: str = None) -> int:
+        """
+        Retrieves the total count of records for a given collection using Sinequa's TotalRowCount metadata.
+
+        Args:
+            collection_config_folder (str): The collection folder to query (e.g., "EARTHDATA", "CASEI").
+            source (str, optional): The source to query. If None, defaults to "scrapers" for dev servers
+                or "SDE" for other servers.
+
+        Returns:
+            int: The total number of records in the collection.
+        """
+        if not source:
+            source = self._get_source_name()
+
+        if (index := self.config.get("index")) is None:
+            raise ValueError(
+                f"Configuration error: Index not defined for server '{self.server_name}'. "
+                "Please update server configuration with the required index."
+            )
+
+        # Minimal query to get only metadata, no data retrieval
+        sql = f"SELECT * FROM {index} WHERE collection = '/{source}/{collection_config_folder}/' SKIP 0 COUNT 0"
+
+        response = self._execute_sql_query(sql)
+
+        # Extract TotalRowCount from metadata
+        return response.get("TotalRowCount", 0)
+
     @staticmethod
     def _process_full_text_response(batch_data: dict):
         if "Rows" not in batch_data or not isinstance(batch_data["Rows"], list):

sde_collections/tasks.py

@@ -14,8 +14,9 @@
     ReindexingStatusChoices,
     WorkflowStatusChoices,
 )
+from sde_collections.utils import slack_utils
 
-from .models.delta_url import DumpUrl
+from .models.delta_url import CuratedUrl, DeltaUrl, DumpUrl
 from .sinequa_api import Api
 from .utils.github_helper import GitHubHandler
 
@@ -173,8 +174,10 @@ def fetch_and_replace_full_text(collection_id, server_name):
     # Step 1: Delete existing DumpUrl entries
     deleted_count, _ = DumpUrl.objects.filter(collection=collection).delete()
     print(f"Deleted {deleted_count} old records.")
-
     try:
+        total_server_count = api.get_total_count(collection.config_folder)
+        print(f"Total records on the server: {total_server_count}")
+
         # Step 2: Process data in batches
         total_processed = 0
         for batch in api.get_full_texts(collection.config_folder):
@@ -193,9 +196,15 @@ def fetch_and_replace_full_text(collection_id, server_name):
             total_processed += len(batch)
             print(f"Processed batch of {len(batch)} records. Total: {total_processed}")
 
+        dump_count = DumpUrl.objects.filter(collection=collection).count()
+
         # Step 3: Migrate dump URLs to delta URLs
         collection.migrate_dump_to_delta()
 
+        curated_count = CuratedUrl.objects.filter(collection=collection).count()
+        delta_count = DeltaUrl.objects.filter(collection=collection).count()
+        marked_for_deletion_count = DeltaUrl.objects.filter(collection=collection, to_delete=True).count()
+
         # Step 4: Update statuses if needed
         collection.refresh_from_db()
 
@@ -215,6 +224,9 @@ def fetch_and_replace_full_text(collection_id, server_name):
             collection.reindexing_status = ReindexingStatusChoices.REINDEXING_READY_FOR_CURATION
             collection.save()
 
+        slack_utils.send_detailed_import_notification(
+            collection.name, total_server_count, curated_count, dump_count, delta_count, marked_for_deletion_count
+        )
         return f"Successfully processed {total_processed} records and updated the database."
 
     except Exception as e:

sde_collections/tests/test_import_fulltexts.py

@@ -30,9 +30,11 @@ def test_fetch_and_replace_full_text(disconnect_signals):
     ]
 
     def mock_generator():
-        yield mock_batch
+        yield (mock_batch)
 
-    with patch("sde_collections.sinequa_api.Api.get_full_texts") as mock_get_full_texts:
+    with patch("sde_collections.sinequa_api.Api.get_full_texts") as mock_get_full_texts, patch(
+        "sde_collections.sinequa_api.Api.get_total_count", return_value=2
+    ), patch("sde_collections.utils.slack_utils.send_detailed_import_notification"):
         mock_get_full_texts.return_value = mock_generator()
 
         fetch_and_replace_full_text(collection.id, "lrm_dev")
@@ -59,9 +61,11 @@ def mock_batch_generator():
         total_records = 20000
 
         for start in range(0, total_records, batch_size):
-            yield create_batch(start, min(batch_size, total_records - start))
+            yield (create_batch(start, min(batch_size, total_records - start)))
 
-    with patch("sde_collections.sinequa_api.Api.get_full_texts") as mock_get_full_texts:
+    with patch("sde_collections.sinequa_api.Api.get_full_texts") as mock_get_full_texts, patch(
+        "sde_collections.sinequa_api.Api.get_total_count", return_value=20000
+    ), patch("sde_collections.utils.slack_utils.send_detailed_import_notification"):
         mock_get_full_texts.return_value = mock_batch_generator()
 
         # Execute the task

sde_collections/tests/test_sinequa_api.py

@@ -170,7 +170,6 @@ def test_get_full_texts_pagination(self, mock_execute_sql, api_instance):
 
         # Collect all batches from the iterator
         batches = list(api_instance.get_full_texts("test_folder"))
-
         assert len(batches) == 2  # Should have two batches
         assert len(batches[0]) == 2  # First batch has 2 records
         assert len(batches[1]) == 1  # Second batch has 1 record

sde_collections/tests/test_workflow_status_triggers.py

@@ -108,9 +108,10 @@ def setUp(self):
             {"url": "http://example.com/2", "title": "Title 2", "full_text": "Content 2"},
         ]
 
+    @patch("sde_collections.utils.slack_utils.send_detailed_import_notification")
     @patch("sde_collections.tasks.Api")
     @patch("sde_collections.models.collection.GitHubHandler")
-    def test_full_text_import_workflow(self, MockGitHub, MockApi):
+    def test_full_text_import_workflow(self, MockGitHub, MockApi, MockSlackNotification):
         """Test the full process of importing full text data"""
         # Setup mock GitHub handler with proper XML content
         mock_github = Mock()
@@ -412,7 +413,7 @@ def test_full_text_import_workflow(self, MockGitHub, MockApi):
 
         # Setup mock API
         mock_api = Mock()
-        mock_api.get_full_texts.return_value = [self.api_response]
+        mock_api.get_full_texts.return_value = iter([self.api_response])
         MockApi.return_value = mock_api
 
         # Setup initial workflow state

sde_collections/utils/slack_utils.py

@@ -59,6 +59,25 @@ def format_slack_message(name, details, collection_id):
     return message_template.format(name=linked_name)
 
 
+def send_detailed_import_notification(
+    collection_name, total_server_count, curated_count, dump_count, delta_count, marked_for_deletion_count
+):
+    message = (
+        f"'{collection_name}' brought into COSMOS.\n"
+        f"Prior Curated: {curated_count}\n"
+        f"Server Count: {total_server_count}\n"
+        f"URLs Imported: {dump_count}\n"
+        f"New Deltas: {delta_count}\n"
+        f"Marked For Deletion: {marked_for_deletion_count}\n"
+    )
+
+    webhook_url = settings.SLACK_WEBHOOK_URL
+    payload = {"text": message}
+    response = requests.post(webhook_url, json=payload)
+    if response.status_code != 200:
+        print(f"Error sending Slack message: {response.text}")
+
+
 def send_slack_message(message):
     webhook_url = settings.SLACK_WEBHOOK_URL
     payload = {"text": message}