chg: add sync script for publications per researcher

Author: taheeraahmed

src/datasync/nva/__init__.py

@@ -2,7 +2,12 @@
 # this allows to attach the scripts to the app
 import typer
 
-from . import nva, nva_filter_resources, nva_search_resources
+from . import (
+    nva,
+    nva_filter_resources,
+    nva_get_publications_per_researcher,
+    nva_search_resources,
+)
 
 app = typer.Typer(help="Commands to handle NVA tasks")
 
@@ -15,3 +20,7 @@
 app.command(help="Fetch and filter NVA data with flexible search parameters")(
     nva_search_resources.search_resources_api
 )
+
+app.command(help="Fetch publications for researchers at NINA")(
+    nva_get_publications_per_researcher.get_pubs_per_researcher
+)

src/datasync/nva/nva_get_publications_per_researcher.py

@@ -0,0 +1,70 @@
+# get nva id's https://s3-ext-1.nina.no/dms/nina/employees.parquet
+# https://api.nva.unit.no/search/resources?contributor=https://api.nva.unit.no/cristin/person/408357
+# use search_api function in nva_search_resources.py
+import typer
+
+from ..settings import (
+    log,
+)
+from .nva_search_resources import search_resources_api
+from .settings import (
+    NINA_EMPLOYEES_PARQUET,
+    NVA_S3_ACCESS_KEY,
+    NVA_S3_BUCKET,
+    NVA_S3_ENDPOINT_URL,
+    NVA_S3_PREFIX,
+    NVA_S3_REGION,
+    NVA_S3_SECRET_KEY,
+)
+from .utils import (
+    setup_duckdb_s3_connection,
+)
+
+app = typer.Typer(help="Fetching publications from researchers at NINA")
+
+
+@app.command()
+def nva_pubs_per_researcher(
+    employees_parquet: str = NINA_EMPLOYEES_PARQUET,
+    storage_s3_path: str = NVA_S3_ENDPOINT_URL,
+    storage_access_key: str = NVA_S3_ACCESS_KEY,
+    storage_secret_key: str = NVA_S3_SECRET_KEY,
+    storage_bucket: str = NVA_S3_BUCKET,
+    storage_prefix: str = NVA_S3_PREFIX,
+    storage_region: str = NVA_S3_REGION,
+):
+    log.debug("Fetching ALL publications from researchers at NINA")
+
+    if not employees_parquet:
+        log.error(
+            "Path to employees parquet file is not set."
+            "Set the NINA_EMPLOYEES_PARQUET environment variable."
+        )
+        raise typer.Exit(code=1)
+
+    con = setup_duckdb_s3_connection(
+        endpoint_url=storage_s3_path,
+        access_key=storage_access_key,
+        secret_key=storage_secret_key,
+        region=storage_region,
+    )
+
+    employees = con.read_parquet(employees_parquet).to_df()
+    # get all nva_id from employees
+    nva_ids = employees["nva_id"].tolist()
+
+    for nva_id in nva_ids:
+        log.debug("Fetching publications for researcher", nva_id=nva_id)
+        search_resources_api(
+            resource_name=f"publications_{nva_id}",
+            filters=[f"contributor=https://api.nva.unit.no/cristin/person/{nva_id}"],
+            apply_filter=False,
+            storage_endpoint_url=storage_s3_path,
+            storage_access_key=storage_access_key,
+            storage_secret_key=storage_secret_key,
+            storage_bucket=storage_bucket,
+            storage_prefix=f"{storage_prefix}/{nva_id}/",
+            storage_region=storage_region,
+        )
+
+    log.info("Fetched all publications for researchers", total_researchers=len(nva_ids))

src/datasync/nva/nva_search_resources.py

@@ -126,7 +126,7 @@ def search_resources_api(
         log.error("Valid filters", valid_filters=VALID_PARAMS_NVA_API)
         raise typer.Exit(code=1)
 
-    log.info(f"Fetching NVA data with search parameters: {search_params}")
+    log.debug(f"Fetching NVA data with search parameters: {search_params}")
 
     credentials = create_s3_credentials(
         endpoint_url=storage_endpoint_url,
@@ -143,7 +143,7 @@ def search_resources_api(
         region=storage_region,
     )
 
-    log.info(f"Fetching resources from {base_url}")
+    log.debug(f"Fetching resources from {base_url}")
     load_info = pipeline.run(
         nva_search_source(
             base_url=base_url,
@@ -154,16 +154,15 @@ def search_resources_api(
         loader_file_format="parquet",
     )
 
-    log.info("Pipeline run completed", load_info=load_info)
+    log.debug("Pipeline run completed", load_info=load_info)
 
+    con = setup_duckdb_s3_connection(
+        endpoint_url=storage_endpoint_url,
+        access_key=storage_access_key,
+        secret_key=storage_secret_key,
+        region=storage_region,
+    )
     if apply_filter:
-        con = setup_duckdb_s3_connection(
-            endpoint_url=storage_endpoint_url,
-            access_key=storage_access_key,
-            secret_key=storage_secret_key,
-            region=storage_region,
-        )
-
         apply_filter_transformation(
             con=con,
             bucket=storage_bucket,
@@ -175,7 +174,7 @@ def search_resources_api(
     write_timestamp(con, storage_bucket, storage_prefix)
     con.close()
 
-    log.info("NVA data sync completed")
+    log.debug("NVA data sync completed")
     log.info(
         f"Data available at: {storage_endpoint_url}/{storage_bucket}/{storage_prefix}/"
         f"main/{resource_name}.parquet"

src/datasync/nva/settings.py

@@ -13,6 +13,9 @@
 NVA_S3_PREFIX = env("NVA_S3_PREFIX", default="")
 NVA_S3_REGION = env("NVA_S3_REGION", default="us-east-1")
 
+# Employees parquet file path
+NINA_EMPLOYEES_PARQUET = env("NINA_EMPLOYEES_PARQUET", default="")
+
 # Valid search parameters accepted by the NVA API search endpoint
 # Based on: https://swagger-ui.nva.unit.no/#/NVA%20Public%20Search%20API/searchResources
 VALID_PARAMS_NVA_API = {