chg: add sync script for publications per researcher

Author: taheeraahmed

pyproject.toml

@@ -44,7 +44,8 @@ dependencies = [
   "pyyaml>=6.0.3",
   "deepdiff>=8.6.1",
   "xmldiff>=2.7.0",
-  "xmlschema>=4.3.1"
+  "xmlschema>=4.3.1",
+  "pandas>=2.3.3"
 ]
 description = ""
 license = "GPL-3.0+"
@@ -63,7 +64,7 @@ ubw_sync = "datasync.ubw:app"
 known_first_party = ["src"]
 
 [tool.deptry.per_rule_ignores]
-DEP002 = ["rich", "numpy", "setuptools"]
+DEP002 = ["rich", "numpy", "setuptools", "pandas"]
 
 [tool.hatch.metadata]
 allow-direct-references = true

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.py

@@ -189,7 +189,6 @@ def run(
         bucket=bucket,
         prefix=prefix,
         credentials=credentials,
-        region=region,
     )
 
     log.info(

src/datasync/nva/nva_get_publications_per_researcher.py

@@ -0,0 +1,138 @@
+# 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 .utils import (
+    setup_duckdb_s3_connection,
+    write_timestamp,
+)
+
+app = typer.Typer(help="Fetching publications from researchers at NINA")
+
+
+def _fetch_publications_for_researcher(
+    nva_id: str,
+    storage_s3_path: str,
+    storage_access_key: str,
+    storage_secret_key: str,
+    storage_bucket: str,
+    storage_prefix: str,
+    storage_region: str,
+) -> str:
+    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,
+        add_timestamp=False,
+    )
+    return nva_id
+
+
+@app.command()
+def get_pubs_per_researcher(
+    employees_parquet: str = typer.Option(
+        ...,
+        envvar="NINA_EMPLOYEES_PARQUET",
+        help="Path to employees parquet file",
+    ),
+    storage_s3_path: str = typer.Option(
+        ...,
+        envvar="NVA_S3_ENDPOINT_URL",
+        help="S3 endpoint URL for storage",
+    ),
+    storage_access_key: str = typer.Option(
+        ..., envvar="NVA_S3_ACCESS_KEY", help="S3 access key for storage"
+    ),
+    storage_secret_key: str = typer.Option(
+        ..., envvar="NVA_S3_SECRET_KEY", help="S3 secret key for storage"
+    ),
+    storage_bucket: str = typer.Option(
+        ..., envvar="NVA_S3_BUCKET", help="S3 bucket for storage"
+    ),
+    storage_prefix: str = typer.Option(
+        ..., envvar="NVA_S3_PREFIX", help="S3 prefix for storage"
+    ),
+    storage_region: str = typer.Option(
+        "us-east-1", envvar="NVA_S3_REGION", help="S3 region for storage"
+    ),
+):
+    log.debug("Fetching ALL publications from researchers at NINA")
+
+    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 employees with missing nva_id and log their names for debugging
+    missing_nva_id = employees["nva_id"].isna().sum()
+    if missing_nva_id > 0:
+        missing_names = (
+            employees[employees["nva_id"].isna()]["firstname"]
+            + " "
+            + employees[employees["nva_id"].isna()]["lastname"]
+        ).tolist()
+        log.warning(
+            "Employees missing NVA IDs",
+            missing_count=missing_nva_id,
+            total_employees=len(employees),
+            missing_names=missing_names,
+        )
+    con.close()
+    nva_ids = [str(nva_id) for nva_id in employees["nva_id"].dropna().unique()]
+
+    failed = []
+    for nva_id in nva_ids:
+        try:
+            _fetch_publications_for_researcher(
+                nva_id,
+                storage_s3_path,
+                storage_access_key,
+                storage_secret_key,
+                storage_bucket,
+                storage_prefix,
+                storage_region,
+            )
+        except Exception as e:
+            log.error(
+                "Failed to fetch publications for researcher",
+                nva_id=nva_id,
+                error=str(e),
+            )
+            failed.append(nva_id)
+
+    if failed:
+        log.warning(
+            "Some researchers failed to fetch", failed=failed, count=len(failed)
+        )
+
+    log.info(
+        "Fetched all publications for researchers",
+        total_researchers=len(nva_ids),
+        failed=len(failed),
+        succeeded=len(nva_ids) - len(failed),
+    )
+
+    con = setup_duckdb_s3_connection(
+        endpoint_url=storage_s3_path,
+        access_key=storage_access_key,
+        secret_key=storage_secret_key,
+        region=storage_region,
+    )
+    write_timestamp(con, storage_bucket, storage_prefix)
+    con.close()

src/datasync/nva/nva_search_resources.py

@@ -59,6 +59,9 @@ def search_resources_api(
     storage_region: str = typer.Option(
         "us-east-1", "--storage-region", help="S3 region for storage"
     ),
+    add_timestamp: bool = typer.Option(
+        True, help="Whether to write last successful run timestamp to S3"
+    ),
 ):
     """
     Get resources from NVA API
@@ -128,7 +131,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,
@@ -142,11 +145,10 @@ def search_resources_api(
         bucket=storage_bucket,
         prefix=storage_prefix,
         credentials=credentials,
-        region=storage_region,
     )
 
-    log.info(f"Fetching resources from {base_url}")
-    pipeline.run(
+    log.debug(f"Fetching resources from {base_url}")
+    load_info = pipeline.run(
         nva_search_source(
             base_url=base_url,
             resource_name=resource_name,
@@ -156,14 +158,15 @@ def search_resources_api(
         loader_file_format="parquet",
     )
 
-    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,
-        )
+    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:
         apply_filter_transformation(
             con=con,
             bucket=storage_bucket,
@@ -172,11 +175,14 @@ def search_resources_api(
             check_additional_identifiers=True,
         )
 
-    write_timestamp(con, storage_bucket, storage_prefix)
+    if add_timestamp:
+        write_timestamp(con, storage_bucket, storage_prefix)
     con.close()
 
+    log.debug("NVA data sync completed")
     log.info(
-        f"Data available at: {storage_endpoint_url}/{storage_bucket}/{storage_prefix}/"
+        f"Data available at: {storage_endpoint_url}/{storage_bucket}/"
+        f"{storage_prefix.rstrip('/')}/"
         f"main/{resource_name}.parquet"
     )
 

src/datasync/nva/utils.py

@@ -26,7 +26,7 @@ def get_nva_resources(
     Yields:
         Resources from the NVA API
     """
-    log.info(f"Fetching resources with params: {search_params}")
+    log.debug(f"Fetching resources with params: {search_params}")
     yield from client.paginate(
         "search/resources",
         method="GET",
@@ -99,19 +99,16 @@ def create_pipeline(
     bucket: str,
     prefix: str,
     credentials: AwsCredentials,
-    region: str,
 ) -> dlt.Pipeline:
     """Create a DLT pipeline with filesystem destination."""
     return dlt.pipeline(
         pipeline_name=pipeline_name,
         destination=filesystem(
-            region_name=region,
-            bucket_url=f"s3://{bucket}/{prefix}",
+            bucket_url=f"s3://{bucket}/{prefix.rstrip('/')}",
             credentials=credentials,
             layout="{table_name}.{ext}",
         ),
         dataset_name="main",
-        progress="log",
     )
 
 
@@ -150,6 +147,7 @@ def write_timestamp(
 ) -> str:
     """Write last successful run timestamp to S3."""
     timestamp = datetime.datetime.now().isoformat()
+    prefix = prefix.rstrip("/")
     con.execute(f"""
         COPY (SELECT '{timestamp}' as last_successful_run)
         TO 's3://{bucket}/{prefix}/last_successful_run.parquet'