Merge pull request #706 from pitangainnovare/fix/search-most-cited-sorting

Corrige perda de campo de ordenação por mais citados

Author: pitangainnovare

harvest/bronze_transform.py

@@ -5,11 +5,14 @@
 import json
 import logging
 
+from django.db.models import Exists, OuterRef
 from django.utils import timezone
-from etl.models import EtlPipelineConfig
+
+from etl.models import EtlItemProcess, EtlPipelineConfig
 from etl.services import enqueue_etl_item
 from search_gateway.client import get_opensearch_client
 
+from .models import TransformationScript, HarvestStatus, IndexStatus
 from .utils import source_hash
 
 logger = logging.getLogger(__name__)
@@ -214,8 +217,6 @@ def transform_after_indexing(instance, model_name):
     Returns:
         Dict com status e mensagem da operação
     """
-    from .models import TransformationScript
-
     harvest_model_key = model_name
     if model_name == "HarvestedSciELOData":
         type_data = getattr(instance, "type_data", None)
@@ -239,3 +240,37 @@ def transform_after_indexing(instance, model_name):
         return {"status": "error", "message": "Instância em identifiesr; não é possível transformar."}
 
     return transform_document(script, instance.identifier)
+
+
+def reconcile_missing_bronze_etl(document_model):
+    model_name = document_model.__name__
+
+    bronze_indices = set(
+        TransformationScript.objects.filter(
+            harvest_model=model_name,
+            is_active=True,
+        ).values_list("dest_index", flat=True)
+    )
+    if not bronze_indices:
+        logger.info("Reconciliação. Fase 2. Não há script para transformar raw em bronze para %s.", model_name)
+        return
+    
+    indexed_qs = document_model.objects.filter(
+        harvest_status=HarvestStatus.SUCCESS,
+        index_status=IndexStatus.SUCCESS,
+    ).exclude(raw_data={})
+
+    has_etl = EtlItemProcess.objects.filter(
+        source_index__in=bronze_indices,
+        external_id=OuterRef("identifier"),
+    )
+
+    missing_qs = indexed_qs.filter(~Exists(has_etl))
+    
+    logger.info("Reconciliação. Fase 2. %s: %d sem ETL", model_name, missing_qs.count())
+    for obj in missing_qs.iterator():
+        try:
+            transform_after_indexing(instance=obj, model_name=model_name)
+        except Exception as exc:
+            logger.warning("Reconcialiação. Fase 2. Falha ao criar ETL para documento do tipo %s (%s): %s",
+                            model_name, obj.identifier, exc)

harvest/indexing.py

@@ -7,36 +7,38 @@
 from django.conf import settings
 from django.utils import timezone
 
-from etl.models import EtlPipelineConfig
-from etl.services import enqueue_etl_item
 from search_gateway.client import get_opensearch_client
 
+from .bronze_transform import transform_after_indexing
 from .exception_logs import ExceptionContext
-from .models import HarvestStatus
+from .models import HarvestStatus, IndexStatus
 from .utils import source_hash
 
 logger = logging.getLogger(__name__)
 
 
-def _get_index_name(model_name=None, instance=None, type_data=None):
+def get_index_name(model_name=None, instance=None, type_data=None):
     if instance is not None and not model_name:
         model_name = instance.__class__.__name__
+
     if not model_name:
         return None
+
     if model_name == "HarvestedSciELOData":
         effective_type = type_data or getattr(instance, "type_data", None)
         index = {
             "dataset":  getattr(
-                settings, "OPENSEARCH_INDEX_RAW_SCIELO_DATA_DATASET", None
+                settings, "OS_INDEX_RAW_SCIELO_DATA_DATASET", None
             ),
             "dataverse": getattr(
-                settings, "OPENSEARCH_INDEX_RAW_SCIELO_DATA_DATAVERSE", None
+                settings, "OS_INDEX_RAW_SCIELO_DATA_DATAVERSE", None
             )
         }
         return index.get(effective_type, None)
+    
     return {
-        "HarvestedPreprint": getattr(settings, "OPENSEARCH_INDEX_RAW_PREPRINT", None),
-        "HarvestedBooks": getattr(settings, "OPENSEARCH_INDEX_RAW_BOOK", None),
+        "HarvestedPreprint": getattr(settings, "OS_INDEX_RAW_PREPRINT", None),
+        "HarvestedBooks": getattr(settings, "OS_INDEX_RAW_BOOK", None),
     }.get(model_name)
 
 
@@ -60,10 +62,24 @@ def index_harvested_raw_data(model, index_name=None, only_success=True, refresh=
     queryset = model.objects.all()
     if status_filter:
         queryset = queryset.filter(harvest_status__in=status_filter)
+
+    queryset = queryset.exclude(index_status=IndexStatus.SUCCESS)
+
     for obj in queryset.iterator():
-        index_name = _get_index_name(model_name=model.__name__, instance=obj)
+        index_name = get_index_name(model_name=model.__name__, instance=obj)
         index_harvested_instance(instance=obj, index_name=index_name, refresh=False)
 
+        if obj.index_status == IndexStatus.SUCCESS and obj.raw_data:
+            try:
+                transform_after_indexing(instance=obj, model_name=model.__name__)
+            except Exception as exc:
+                logger.warning(
+                    "Falha na transformação bronze %s (%s): %s",
+                    model.__name__,
+                    obj.identifier,
+                    exc,
+                )
+
 
 def index_harvested_instance(instance, index_name=None, refresh=False):
     """
@@ -74,10 +90,12 @@ def index_harvested_instance(instance, index_name=None, refresh=False):
         log_model=instance.harvest_error_log.model,
         fk_field=_get_error_log_fk_field(instance),
     )
+
     client = get_opensearch_client()
     if client is None:
         logger.warning("OpenSearch client não configurado.")
         return
+    
     if not index_name:
         logger.warning(
             f"Index name não configurado para {instance.__class__.__name__} ({instance.identifier})."
@@ -99,13 +117,8 @@ def index_harvested_instance(instance, index_name=None, refresh=False):
             },
             refresh=False,
         )
-        if EtlPipelineConfig.objects.select_for_source(index_name, instance.raw_data):
-            enqueue_etl_item(
-                source_index=index_name,
-                external_id=instance.identifier,
-                source_payload=instance.raw_data,
-            )
         instance.mark_as_indexed(index_name=index_name)
+
     except Exception as exc:
         logger.warning(
             f"Falha ao indexar em {index_name} {instance.__class__.__name__} ({instance.identifier}): {exc}"
@@ -119,14 +132,16 @@ def index_harvested_instance(instance, index_name=None, refresh=False):
 
 
 def delete_harvested_document(model_name, identifier, refresh=False):
-    index_name = _get_index_name(model_name=model_name)
+    index_name = get_index_name(model_name=model_name)
     if not index_name:
         logger.warning(f"Index name não configurado para {model_name}.")
         return
+
     client = get_opensearch_client()
     if client is None:
         logger.warning("OpenSearch client não configurado.")
         return
+
     try:
         logging.info(f"Removendo documento {identifier} do indice {index_name}")
         client.delete(index=index_name, id=identifier, refresh=refresh)

harvest/management/commands/create_raw_indices.py

@@ -46,14 +46,14 @@ def handle(self, *args, **options):
 
         indices = {
             "HarvestedPreprint": getattr(
-                settings, "OPENSEARCH_INDEX_RAW_PREPRINT", None
+                settings, "OS_INDEX_RAW_PREPRINT", None
             ),
-            "HarvestedBooks": getattr(settings, "OPENSEARCH_INDEX_RAW_BOOK", None),
+            "HarvestedBooks": getattr(settings, "OS_INDEX_RAW_BOOK", None),
             "HarvestedSciELOData(dataset)": getattr(
-                settings, "OPENSEARCH_INDEX_RAW_SCIELO_DATA_DATASET", None
+                settings, "OS_INDEX_RAW_SCIELO_DATA_DATASET", None
             ),
             "HarvestedSciELOData(dataverse)": getattr(
-                settings, "OPENSEARCH_INDEX_RAW_SCIELO_DATA_DATAVERSE", None
+                settings, "OS_INDEX_RAW_SCIELO_DATA_DATAVERSE", None
             ),
         }
         force = options["force"]

harvest/signals.py

@@ -5,11 +5,11 @@
 
 from .bronze_transform import transform_after_indexing
 from .indexing import (
-    _get_index_name,
+    get_index_name,
     delete_harvested_document,
     index_harvested_instance,
 )
-from .models import HarvestedBooks, HarvestedPreprint, HarvestedSciELOData
+from .models import HarvestedBooks, HarvestedPreprint, HarvestedSciELOData, IndexStatus
 
 logger = logging.getLogger(__name__)
 
@@ -18,6 +18,7 @@ def _store_previous_raw_data(instance):
     if not instance.pk:
         instance._raw_data_before = None
         return
+
     instance._raw_data_before = (
         instance.__class__.objects.filter(pk=instance.pk)
         .values_list("raw_data", flat=True)
@@ -34,27 +35,36 @@ def _should_index_raw_data(instance, created, update_fields):
                 instance.identifier,
             )
         return False
+
     if created:
         return True
+
     if update_fields is not None:
         return "raw_data" in update_fields
+
     if hasattr(instance, "_raw_data_before"):
-        return instance._raw_data_before != instance.raw_data
+        if instance._raw_data_before != instance.raw_data:
+            return True
+    
+    if hasattr(instance, "index_status") and instance.index_status in (IndexStatus.PENDING, IndexStatus.FAILED):
+        return True
+
     return False
 
 
 def _index_if_raw_data_saved(instance, created, update_fields):
     if not _should_index_raw_data(instance, created, update_fields):
         return
-    index_name = _get_index_name(
+
+    index_name = get_index_name(
         model_name=instance.__class__.__name__,
         instance=instance,
     )
     index_harvested_instance(instance=instance, index_name=index_name)
 
     model_name = instance.__class__.__name__
     try:
-        if instance.raw_data:
+        if instance.index_status == IndexStatus.SUCCESS and instance.raw_data:
             transform_after_indexing(instance=instance, model_name=model_name)
     except Exception as exc:
         logger.warning(
@@ -92,7 +102,7 @@ def track_books_raw_data(sender, instance, **kwargs):
 
 @receiver(post_delete, sender=HarvestedBooks)
 def delete_books_on_delete(sender, instance, **kwargs):
-    index_name = _get_index_name(model_name=instance.__class__.__name__)
+    index_name = get_index_name(model_name=instance.__class__.__name__)
     if not index_name:
         return
     delete_harvested_document(

harvest/tasks.py

@@ -18,8 +18,9 @@
 )
 from harvest.harvests.harvest_data import harvest_data, harvest_single_scielo_data
 from harvest.harvests.harvest_preprint import harvest_preprint
-from harvest.indexing import index_harvested_instance
+from harvest.indexing import index_harvested_instance, index_harvested_raw_data
 
+from .bronze_transform import reconcile_missing_bronze_etl
 from .models import (
     HarvestedBooks,
     HarvestedPreprint,
@@ -36,15 +37,21 @@
 
 
 @celery_app.task(name="Harvest data preprint")
-def harvest_preprint_in_endpoint_oai_pmh(username, user_id=None, reprocess=None, url=None, verify=True):
-    latest_preprint = None
+def harvest_preprint_in_endpoint_oai_pmh(username, user_id=None, reprocess=None, from_date=None, url=None, verify=True):
     user = User.objects.get(username=username)
+    
     url = ENDPOINT_PREPRINT
-    if not reprocess:
+
+    if from_date is None and not reprocess:
         latest_preprint = HarvestedPreprint.get_latest_preprint()
-    from_date = latest_preprint.datestamp.date().__str__() if latest_preprint else None
-    logging.info(f"Coleta a partir da data {from_date}")
-    recs = service_oai_pmh_scythe(url=url, from_date=from_date, verify=verify)
+        from_date = latest_preprint.datestamp.date().__str__() if latest_preprint else None
+    
+    if from_date:
+        logging.info(f"Coleta a partir da data {from_date}")
+    else:
+        logging.info("Coletando todos os registros")
+
+    recs = service_oai_pmh_scythe(url=url, from_date=from_date, verify=verify)    
     harvest_preprint(recs=recs, user=user)
 
 
@@ -164,7 +171,6 @@ def retry_failed_preprints_oai_pmh(username, user_id=None, url=None, verify=True
         harvest_preprint(recs=[rec], user=user)
 
 
-
 @celery_app.task(name="Retry failed books")
 def retry_failed_books(username, user_id=None, db_name="scielobooks_1a", headers=None):
     failed_books = HarvestedBooks.objects.filter(
@@ -235,3 +241,22 @@ def apply_global_metrics_upload_to_silver(
         harvest_index=harvest_index,
         silver_index=silver_index,
     )
+
+
+@celery_app.task(name="Reconcile harvest pipeline")
+def reconcile_harvest_pipeline(document_type, user_id=None):
+    DOCUMENT_TYPE_TO_DOCUMENT_MODEL = {
+        "book": HarvestedBooks,
+        "data": HarvestedSciELOData,
+        "preprint": HarvestedPreprint,
+    }
+
+    document_model = DOCUMENT_TYPE_TO_DOCUMENT_MODEL.get(document_type)
+    if not document_model:
+        logging.error("Reconciliação não foi executada. document_type %s inválido. "
+                      "Valores possíveis são: book, data, preprint.",
+                      document_type)
+        return
+
+    index_harvested_raw_data(document_model)
+    reconcile_missing_bronze_etl(document_model)

harvest/wagtail_hooks.py

@@ -47,7 +47,7 @@ class HarvestedPreprintViewSet(SnippetViewSet):
     add_to_admin_menu = False
     add_view_class = CommonControlFieldCreateView
     edit_view_class = CommonControlFieldEditView
-    list_display = ("identifier", "harvest_status", "datestamp", "created")
+    list_display = ("identifier", "harvest_status", "index_status", "datestamp", "created")
     search_fields = ("identifier", "source_url")
     list_filter = ("harvest_status", "index_status")
     ordering = ("-created",)
@@ -60,7 +60,7 @@ class HarvestedSciELODataViewSet(SnippetViewSet):
     add_to_admin_menu = False
     add_view_class = CommonControlFieldCreateView
     edit_view_class = CommonControlFieldEditView
-    list_display = ("identifier", "harvest_status", "datestamp", "created")
+    list_display = ("identifier", "harvest_status", "index_status", "datestamp", "created")
     search_fields = ("identifier", "source_url")
     list_filter = ("harvest_status", "index_status", "type_data")
     ordering = ("-created",)
@@ -73,7 +73,7 @@ class HarvestedBooksViewSet(SnippetViewSet):
     add_to_admin_menu = False
     add_view_class = CommonControlFieldCreateView
     edit_view_class = CommonControlFieldEditView
-    list_display = ("identifier", "type_data", "harvest_status", "datestamp", "created")
+    list_display = ("identifier", "type_data", "harvest_status", "index_status", "datestamp", "created")
     search_fields = ("identifier", "source_url")
     list_filter = ("harvest_status", "index_status", "type_data")
     ordering = ("-created",)