bugfix for batch retrieval

Author: chreman

server/preprocessing/other-scripts/base.R

@@ -156,37 +156,32 @@ get_papers <- function(query, params,
       has_custom_clustering_annotation <- unlist(lapply(metadata$subject_orig, function(x) grepl(paste0(cc, ":"), x, fixed=TRUE)))
       metadata <- metadata[has_custom_clustering_annotation,]
   }}
-  # don't deduplicate if params$deduplicate_base is set to FALSE
-  if (!is.null(params$deduplicate_base) && params$deduplicate_base != FALSE) {
-    # log to skip deduplication
-    blog$info(paste("vis_id:", .GlobalEnv$VIS_ID, "Deduplication skipped"))
-  } else {
-    while (nrow(metadata) - sum(metadata$is_duplicate) < limit && attr(res_raw, "numFound") > offset+120 && r < req_limit) {
-      offset <- offset+120
-      res_raw <- get_raw_data(limit,
-                              base_query,
-                              return_fields,
-                              sortby_string,
-                              filter,
-                              repo,
-                              coll,
-                              retry_opts,
-                              offset,
-                              non_public)
-      res <- bind_rows(res, res_raw$docs)
-      metadata <- etl(res, repo, non_public)
-      metadata <- unique(metadata, by = "id")
-      metadata <- sanitize_abstract(metadata)
-      metadata <- mark_duplicates(metadata)
-      metadata$has_dataset <- unlist(lapply(metadata$resulttype, function(x) "Dataset" %in% x))
-      # check if custom clustering annotation param is in metadata
-      if (!is.null(cc)) {
-        if (!(cc %in% names(fieldmapper))) {
-          has_custom_clustering_annotation <- unlist(lapply(metadata$subject_orig, function(x) grepl(paste0(cc, ":"), x, fixed=TRUE)))
-          metadata <- metadata[has_custom_clustering_annotation,]
-      }}
-      r <- r+1
-    }
+
+  while (nrow(metadata) - sum(metadata$is_duplicate) < limit && attr(res_raw, "numFound") > offset+120 && r < req_limit) {
+    offset <- offset+120
+    res_raw <- get_raw_data(limit,
+                            base_query,
+                            return_fields,
+                            sortby_string,
+                            filter,
+                            repo,
+                            coll,
+                            retry_opts,
+                            offset,
+                            non_public)
+    res <- bind_rows(res, res_raw$docs)
+    metadata <- etl(res, repo, non_public)
+    metadata <- unique(metadata, by = "id")
+    metadata <- sanitize_abstract(metadata)
+    metadata <- mark_duplicates(metadata)
+    metadata$has_dataset <- unlist(lapply(metadata$resulttype, function(x) "Dataset" %in% x))
+    # check if custom clustering annotation param is in metadata
+    if (!is.null(cc)) {
+      if (!(cc %in% names(fieldmapper))) {
+        has_custom_clustering_annotation <- unlist(lapply(metadata$subject_orig, function(x) grepl(paste0(cc, ":"), x, fixed=TRUE)))
+        metadata <- metadata[has_custom_clustering_annotation,]
+    }}
+    r <- r+1
   }
   # check if custom clustering annotation param is in metadata
   if (!is.null(cc)) {

server/workers/base/src/base.py

@@ -28,6 +28,7 @@
 from datetime import datetime
 import dateparser
 import sys
+from typing import Dict
 from common.rate_limiter import RateLimiter
 
 logger = logging.getLogger(__name__)
@@ -86,7 +87,7 @@ def execute_search(self, params):
             else:
                 metadata = pd.DataFrame(raw_metadata)
                 metadata = self.sanitize_metadata(metadata)
-                metadata = filter_duplicates(metadata, original_service)
+                metadata = filter_duplicates(metadata, original_service, params)
                 metadata = pd.concat(
                     [metadata, parse_annotations_for_all(metadata, "subject_orig")],
                     axis=1,
@@ -240,7 +241,11 @@ def handle_contentproviders(self, request_id, params):
 pattern_annotations = re.compile(r"([A-Za-z]+:[\w'\- ]+);?")
 
 
-def filter_duplicates(df, service):
+def filter_duplicates(df, service, params):
+    if logger.isEnabledFor(logging.DEBUG):
+        logger.debug(f"Filtering duplicates for service: {service}")
+        logger.debug(f"Initial number of records: {len(df)}")
+        _log_dataframe(df, params, "initial_records")
     df.drop_duplicates("id", inplace=True, keep="first")
     df["is_anchor"] = False
     df["doi_duplicate"] = False
@@ -303,6 +308,9 @@ def filter_duplicates(df, service):
         if c in filtered.columns:
             filtered.drop(c, axis=1, inplace=True)
 
+    if logger.isEnabledFor(logging.DEBUG):
+        logger.debug(f"Number of records after filtering: {len(filtered)}")
+        _log_dataframe(filtered, params, "filtered_records")
     return filtered
 
 
@@ -362,3 +370,22 @@ def sanitize_year(year_str):
         sanitized_year = year_str  # here we keep the original string
 
     return sanitized_year
+
+def _log_dataframe(df: pd.DataFrame, params: Dict[str, str], name: str, ):
+    vis_id = params.get('vis_id')
+    
+    columns_to_print = ['id', 'title', 'doi', 'merged_dois', 'paper_abstract', 'link', 'subject', 'subject_orig', 'oa_state']
+
+    available_columns = df.columns.tolist()
+    columns_to_print = [col for col in columns_to_print if col in available_columns]
+
+    transformed = df.copy().reindex(columns=columns_to_print)
+    
+    transformed = transformed.fillna(value='missing')
+    
+    # create folder
+    folder = f'./output/{vis_id}'
+    if not os.path.exists(folder):
+        os.makedirs(folder)
+    file_path = f"{folder}/{name}.csv"
+    transformed.to_csv(file_path, index=False)

server/workers/orcid/src/orcid_service.py

@@ -152,7 +152,7 @@ def _log_dataframe(self, df: pd.DataFrame, params: Dict[str, str], name: str, ):
 
     def request_base_metadata(self, dois: List[str], params: Dict[str, str]) -> pd.DataFrame:
         orcid = params.get('orcid')
-        batch_size = 15
+        batch_size = 20
         batches = [dois[i:i + batch_size] for i in range(0, len(dois), batch_size)]
         base_metadata = pd.DataFrame(dtype=object)
 
@@ -185,7 +185,6 @@ def request_base_metadata(self, dois: List[str], params: Dict[str, str]) -> pd.D
                     'vis_id': request_id, 
                     'limit': 360,
                     'list_size': 360,
-                    'deduplicate_base': 'false',
                     'exclude_date_filters': 'true',
                     'q_advanced_only': 'true'
                 },